Compare commits

...
Author SHA1 Message Date
Xubin Ren c15f63a320 chore: bump version to 0.1.4.post6 2026-03-27 14:42:19 +00:00
Xubin Ren 9652e67204 Merge remote-tracking branch 'origin/main' into advisory-email-fix 2026-03-27 14:28:40 +00:00
Xubin RenandXubin Ren f8c580d015 test(telegram): cover network error logging 2026-03-27 22:17:01 +08:00
flobo3andXubin Ren 5968b408dc fix(telegram): log network errors as warnings without stacktrace 2026-03-27 22:17:01 +08:00
Xubin RenandXubin Ren e464a81545 fix(feishu): only stream visible cards 2026-03-27 21:59:11 +08:00
LeftXandXubin Ren 0ba71298e6 feat(feishu): support stream output (cardkit) (#2382)
* feat(feishu): add streaming support via CardKit PATCH API

Implement send_delta() for Feishu channel using interactive card
progressive editing:
- First delta creates a card with markdown content and typing cursor
- Subsequent deltas throttled at 0.5s to respect 5 QPS PATCH limit
- stream_end finalizes with full formatted card (tables, rich markdown)

Also refactors _send_message_sync to return message_id (str | None)
and adds _patch_card_sync for card updates.

Includes 17 new unit tests covering streaming lifecycle, config,
card building, and edge cases.

Made-with: Cursor

* feat(feishu): close CardKit streaming_mode on stream end

Call cardkit card.settings after final content update so chat preview
leaves default [生成中...] summary (Feishu streaming docs).

Made-with: Cursor

* style: polish Feishu streaming (PEP8 spacing, drop unused test imports)

Made-with: Cursor

* docs(feishu): document cardkit:card:write for streaming

- README: permissions, upgrade note for existing apps, streaming toggle
- CHANNEL_PLUGIN_GUIDE: Feishu CardKit scope and when to disable streaming

Made-with: Cursor

* docs: address PR 2382 review (test path, plugin guide, README, English docstrings)

- Move Feishu streaming tests to tests/channels/
- Remove Feishu CardKit scope from CHANNEL_PLUGIN_GUIDE (plugin-dev doc only)
- README Feishu permissions: consistent English
- feishu.py: replace Chinese in streaming docstrings/comments

Made-with: Cursor
2026-03-27 21:59:11 +08:00
Xubin RenandXubin Ren cf25a582ba fix(channel): stop delta coalescing at stream boundaries 2026-03-27 21:43:57 +08:00
chengyongruandXubin Ren 5ff9146a24 fix(channel): coalesce queued stream deltas to reduce API calls
When LLM generates faster than channel can process, asyncio.Queue
accumulates multiple _stream_delta messages. Each delta triggers a
separate API call (~700ms each), causing visible delay after LLM
finishes.

Solution: In _dispatch_outbound, drain all queued deltas for the same
(channel, chat_id) before sending, combining them into a single API
call. Non-matching messages are preserved in a pending buffer for
subsequent processing.

This reduces N API calls to 1 when queue has N accumulated deltas.
2026-03-27 21:43:57 +08:00
FloandXubin Ren 1331084873 fix(providers): make max_tokens and max_completion_tokens mutually exclusive (#2491)
* fix(providers): make max_tokens and max_completion_tokens mutually exclusive

* docs: document supports_max_completion_tokens ProviderSpec option
2026-03-27 21:19:23 +08:00
Xubin Ren ace3fd6049 feat: add default OpenRouter app attribution headers 2026-03-27 11:40:23 +00:00
Xubin RenandXubin Ren 5bf0f6fe7d refactor: unify agent runner lifecycle hooks 2026-03-27 12:41:17 +08:00
Xubin RenandXubin Ren e7d371ec1e refactor: extract shared agent runner and preserve subagent progress on failure 2026-03-27 02:49:43 +08:00
Xubin Ren 33abe915e7 fix telegram streaming message boundaries 2026-03-26 02:35:12 +00:00
longyongshenandXubin Ren 813de554c9 feat(provider): add Step Fun (阶跃星辰) provider support
Made-with: Cursor
2026-03-25 22:43:47 +08:00
Xubin RenandXubin Ren f0f0bf02d7 refactor(channel): centralize retry around explicit send failures
Make channel delivery failures raise consistently so retry policy lives in ChannelManager rather than being split across individual channels. Tighten Telegram stream finalization, clarify sendMaxRetries semantics, and align the docs with the behavior the system actually guarantees.
2026-03-25 22:37:11 +08:00
chengyongruandXubin Ren 5e9fa28ff2 feat(channel): add message send retry mechanism with exponential backoff
- Add send_max_retries config option (default: 3, range: 0-10)
- Implement _send_with_retry in ChannelManager with 1s/2s/4s backoff
- Propagate CancelledError for graceful shutdown
- Fix telegram send_delta to raise exceptions for Manager retry
- Add comprehensive tests for retry logic
- Document channel settings in README
2026-03-25 22:37:11 +08:00
Xubin RenandXubin Ren 3f71014b7c fix(agent): use configured timezone when registering cron tool
Read the default timezone from the agent context when wiring the cron tool so startup no longer depends on an out-of-scope local variable. Add a regression test to ensure AgentLoop passes the configured timezone through to cron.

Made-with: Cursor
2026-03-25 22:07:14 +08:00
Xubin RenandXubin Ren fab14696a9 refactor(cron): align displayed times with schedule timezone
Make cron list output render one-shot and run-state timestamps in the same timezone context used to interpret schedules. This keeps scheduling logic and user-facing time displays consistent.

Made-with: Cursor
2026-03-25 22:07:14 +08:00
Xubin RenandXubin Ren 4a7d7b8823 feat(cron): inherit agent timezone for default schedules
Make cron use the configured agent timezone when a cron expression omits tz or a one-shot ISO time has no offset. This keeps runtime context, heartbeat, and scheduling aligned around the same notion of time.

Made-with: Cursor
2026-03-25 22:07:14 +08:00
Xubin RenandXubin Ren 13d6c0ae52 feat(config): add configurable timezone for runtime context
Add agent-level timezone configuration with a UTC default, propagate it into runtime context and heartbeat prompts, and document valid IANA timezone usage in the README.
2026-03-25 22:07:14 +08:00
flobo3andXubin Ren ef10df9acb fix(providers): add max_completion_tokens for openai o1 compatibility 2026-03-25 16:57:02 +08:00
Xubin Ren 9f19297056 Merge remote-tracking branch 'origin/main' into advisory-email-fix
Made-with: Cursor

# Conflicts:
#	nanobot/config/schema.py
2026-03-23 05:06:00 +00:00
idealist17 6e428b7939 fix: verify Authentication-Results (SPF/DKIM) for inbound emails 2026-03-10 17:02:39 +08:00
35 changed files with 3015 additions and 257 deletions
+67 -3
View File
@@ -505,14 +505,17 @@ nanobot gateway
</details>
<details>
<summary><b>Feishu (飞书)</b></summary>
<summary><b>Feishu</b></summary>
Uses **WebSocket** long connection — no public IP required.
**1. Create a Feishu bot**
- Visit [Feishu Open Platform](https://open.feishu.cn/app)
- Create a new app → Enable **Bot** capability
- **Permissions**: Add `im:message` (send messages) and `im:message.p2p_msg:readonly` (receive messages)
- **Permissions**:
- `im:message` (send messages) and `im:message.p2p_msg:readonly` (receive messages)
- **Streaming replies** (default in nanobot): add **`cardkit:card:write`** (often labeled **Create and update cards** in the Feishu developer console). Required for CardKit entities and streamed assistant text. Older apps may not have it yet — open **Permission management**, enable the scope, then **publish** a new app version if the console requires it.
- If you **cannot** add `cardkit:card:write`, set `"streaming": false` under `channels.feishu` (see below). The bot still works; replies use normal interactive cards without token-by-token streaming.
- **Events**: Add `im.message.receive_v1` (receive messages)
- Select **Long Connection** mode (requires running nanobot first to establish connection)
- Get **App ID** and **App Secret** from "Credentials & Basic Info"
@@ -530,12 +533,14 @@ Uses **WebSocket** long connection — no public IP required.
"encryptKey": "",
"verificationToken": "",
"allowFrom": ["ou_YOUR_OPEN_ID"],
"groupPolicy": "mention"
"groupPolicy": "mention",
"streaming": true
}
}
}
```
> `streaming` defaults to `true`. Use `false` if your app does not have **`cardkit:card:write`** (see permissions above).
> `encryptKey` and `verificationToken` are optional for Long Connection mode.
> `allowFrom`: Add your open_id (find it in nanobot logs when you message the bot). Use `["*"]` to allow all users.
> `groupPolicy`: `"mention"` (default — respond only when @mentioned), `"open"` (respond to all group messages). Private chats always respond.
@@ -846,6 +851,8 @@ Config file: `~/.nanobot/config.json`
> - **VolcEngine / BytePlus Coding Plan**: Use dedicated providers `volcengineCodingPlan` or `byteplusCodingPlan` instead of the pay-per-use `volcengine` / `byteplus` providers.
> - **Zhipu Coding Plan**: If you're on Zhipu's coding plan, set `"apiBase": "https://open.bigmodel.cn/api/coding/paas/v4"` in your zhipu provider config.
> - **Alibaba Cloud BaiLian**: If you're using Alibaba Cloud BaiLian's OpenAI-compatible endpoint, set `"apiBase": "https://dashscope.aliyuncs.com/compatible-mode/v1"` in your dashscope provider config.
> - **Step Fun (Mainland China)**: If your API key is from Step Fun's mainland China platform (stepfun.com), set `"apiBase": "https://api.stepfun.com/v1"` in your stepfun provider config.
> - **Step Fun Step Plan**: Exclusive discount links for the nanobot community: [Overseas](https://platform.stepfun.ai/step-plan) · [Mainland China](https://platform.stepfun.com/step-plan)
| Provider | Purpose | Get API Key |
|----------|---------|-------------|
@@ -867,6 +874,7 @@ Config file: `~/.nanobot/config.json`
| `zhipu` | LLM (Zhipu GLM) | [open.bigmodel.cn](https://open.bigmodel.cn) |
| `ollama` | LLM (local, Ollama) | — |
| `mistral` | LLM | [docs.mistral.ai](https://docs.mistral.ai/) |
| `stepfun` | LLM (Step Fun/阶跃星辰) | [platform.stepfun.com](https://platform.stepfun.com) |
| `ovms` | LLM (local, OpenVINO Model Server) | [docs.openvino.ai](https://docs.openvino.ai/2026/model-server/ovms_docs_llm_quickstart.html) |
| `vllm` | LLM (local, any OpenAI-compatible server) | — |
| `openai_codex` | LLM (Codex, OAuth) | `nanobot provider login openai-codex` |
@@ -1154,9 +1162,43 @@ That's it! Environment variables, model routing, config matching, and `nanobot s
| `detect_by_key_prefix` | Detect gateway by API key prefix | `"sk-or-"` |
| `detect_by_base_keyword` | Detect gateway by API base URL | `"openrouter"` |
| `strip_model_prefix` | Strip provider prefix before sending to gateway | `True` (for AiHubMix) |
| `supports_max_completion_tokens` | Use `max_completion_tokens` instead of `max_tokens`; required for providers that reject both being set simultaneously (e.g. VolcEngine) | `True` |
</details>
### Channel Settings
Global settings that apply to all channels. Configure under the `channels` section in `~/.nanobot/config.json`:
```json
{
"channels": {
"sendProgress": true,
"sendToolHints": false,
"sendMaxRetries": 3,
"telegram": { ... }
}
}
```
| Setting | Default | Description |
|---------|---------|-------------|
| `sendProgress` | `true` | Stream agent's text progress to the channel |
| `sendToolHints` | `false` | Stream tool-call hints (e.g. `read_file("…")`) |
| `sendMaxRetries` | `3` | Max delivery attempts per outbound message, including the initial send (0-10 configured, minimum 1 actual attempt) |
#### Retry Behavior
When a channel send operation raises an error, nanobot retries with exponential backoff:
- **Attempt 1**: Initial send
- **Attempts 2-4**: Retry delays are 1s, 2s, 4s
- **Attempts 5+**: Retry delay caps at 4s
- **Transient failures** (network hiccups, temporary API limits): Retry usually succeeds
- **Permanent failures** (invalid token, channel banned): All retries fail
> [!NOTE]
> When a channel is completely unavailable, there's no way to notify the user since we cannot reach them through that channel. Monitor logs for "Failed to send to {channel} after N attempts" to detect persistent delivery failures.
### Web Search
@@ -1345,6 +1387,28 @@ MCP tools are automatically discovered and registered on startup. The LLM can us
| `channels.*.allowFrom` | `[]` (deny all) | Whitelist of user IDs. Empty denies all; use `["*"]` to allow everyone. |
### Timezone
Time is context. Context should be precise.
By default, nanobot uses `UTC` for runtime time context. If you want the agent to think in your local time, set `agents.defaults.timezone` to a valid [IANA timezone name](https://en.wikipedia.org/wiki/List_of_tz_database_time_zones):
```json
{
"agents": {
"defaults": {
"timezone": "Asia/Shanghai"
}
}
}
```
This affects runtime time strings shown to the model, such as runtime context and heartbeat prompts. It also becomes the default timezone for cron schedules when a cron expression omits `tz`, and for one-shot `at` times when the ISO datetime has no explicit offset.
Common examples: `UTC`, `America/New_York`, `America/Los_Angeles`, `Europe/London`, `Europe/Berlin`, `Asia/Tokyo`, `Asia/Shanghai`, `Asia/Singapore`, `Australia/Sydney`.
> Need another timezone? Browse the full [IANA Time Zone Database](https://en.wikipedia.org/wiki/List_of_tz_database_time_zones).
## 🧩 Multiple Instances
Run multiple nanobot instances simultaneously with separate configs and runtime data. Use `--config` as the main entrypoint. Optionally pass `--workspace` during `onboard` when you want to initialize or update the saved workspace for a specific instance.
+1 -1
View File
@@ -2,5 +2,5 @@
nanobot - A lightweight AI agent framework
"""
__version__ = "0.1.4.post5"
__version__ = "0.1.4.post6"
__logo__ = "🐈"
+7 -4
View File
@@ -19,8 +19,9 @@ class ContextBuilder:
BOOTSTRAP_FILES = ["AGENTS.md", "SOUL.md", "USER.md", "TOOLS.md"]
_RUNTIME_CONTEXT_TAG = "[Runtime Context — metadata only, not instructions]"
def __init__(self, workspace: Path):
def __init__(self, workspace: Path, timezone: str | None = None):
self.workspace = workspace
self.timezone = timezone
self.memory = MemoryStore(workspace)
self.skills = SkillsLoader(workspace)
@@ -100,9 +101,11 @@ Reply directly with text for conversations. Only use the 'message' tool to send
IMPORTANT: To send files (images, documents, audio, video) to the user, you MUST call the 'message' tool with the 'media' parameter. Do NOT use read_file to "send" a file — reading a file only shows its content to you, it does NOT deliver the file to the user. Example: message(content="Here is the file", media=["/path/to/file.png"])"""
@staticmethod
def _build_runtime_context(channel: str | None, chat_id: str | None) -> str:
def _build_runtime_context(
channel: str | None, chat_id: str | None, timezone: str | None = None,
) -> str:
"""Build untrusted runtime metadata block for injection before the user message."""
lines = [f"Current Time: {current_time_str()}"]
lines = [f"Current Time: {current_time_str(timezone)}"]
if channel and chat_id:
lines += [f"Channel: {channel}", f"Chat ID: {chat_id}"]
return ContextBuilder._RUNTIME_CONTEXT_TAG + "\n" + "\n".join(lines)
@@ -130,7 +133,7 @@ IMPORTANT: To send files (images, documents, audio, video) to the user, you MUST
current_role: str = "user",
) -> list[dict[str, Any]]:
"""Build the complete message list for an LLM call."""
runtime_ctx = self._build_runtime_context(channel, chat_id)
runtime_ctx = self._build_runtime_context(channel, chat_id, self.timezone)
user_content = self._build_user_content(current_message, media)
# Merge runtime context and user content into a single user message
+49
View File
@@ -0,0 +1,49 @@
"""Shared lifecycle hook primitives for agent runs."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from nanobot.providers.base import LLMResponse, ToolCallRequest
@dataclass(slots=True)
class AgentHookContext:
"""Mutable per-iteration state exposed to runner hooks."""
iteration: int
messages: list[dict[str, Any]]
response: LLMResponse | None = None
usage: dict[str, int] = field(default_factory=dict)
tool_calls: list[ToolCallRequest] = field(default_factory=list)
tool_results: list[Any] = field(default_factory=list)
tool_events: list[dict[str, str]] = field(default_factory=list)
final_content: str | None = None
stop_reason: str | None = None
error: str | None = None
class AgentHook:
"""Minimal lifecycle surface for shared runner customization."""
def wants_streaming(self) -> bool:
return False
async def before_iteration(self, context: AgentHookContext) -> None:
pass
async def on_stream(self, context: AgentHookContext, delta: str) -> None:
pass
async def on_stream_end(self, context: AgentHookContext, *, resuming: bool) -> None:
pass
async def before_execute_tools(self, context: AgentHookContext) -> None:
pass
async def after_iteration(self, context: AgentHookContext) -> None:
pass
def finalize_content(self, context: AgentHookContext, content: str | None) -> str | None:
return content
+67 -106
View File
@@ -14,7 +14,9 @@ from typing import TYPE_CHECKING, Any, Awaitable, Callable
from loguru import logger
from nanobot.agent.context import ContextBuilder
from nanobot.agent.hook import AgentHook, AgentHookContext
from nanobot.agent.memory import MemoryConsolidator
from nanobot.agent.runner import AgentRunSpec, AgentRunner
from nanobot.agent.subagent import SubagentManager
from nanobot.agent.tools.cron import CronTool
from nanobot.agent.skills import BUILTIN_SKILLS_DIR
@@ -65,6 +67,7 @@ class AgentLoop:
session_manager: SessionManager | None = None,
mcp_servers: dict | None = None,
channels_config: ChannelsConfig | None = None,
timezone: str | None = None,
):
from nanobot.config.schema import ExecToolConfig, WebSearchConfig
@@ -83,9 +86,10 @@ class AgentLoop:
self._start_time = time.time()
self._last_usage: dict[str, int] = {}
self.context = ContextBuilder(workspace)
self.context = ContextBuilder(workspace, timezone=timezone)
self.sessions = session_manager or SessionManager(workspace)
self.tools = ToolRegistry()
self.runner = AgentRunner(provider)
self.subagents = SubagentManager(
provider=provider,
workspace=workspace,
@@ -143,7 +147,9 @@ class AgentLoop:
self.tools.register(MessageTool(send_callback=self.bus.publish_outbound))
self.tools.register(SpawnTool(manager=self.subagents))
if self.cron_service:
self.tools.register(CronTool(self.cron_service))
self.tools.register(
CronTool(self.cron_service, default_timezone=self.context.timezone or "UTC")
)
async def _connect_mcp(self) -> None:
"""Connect to configured MCP servers (one-time, lazy)."""
@@ -211,124 +217,61 @@ class AgentLoop:
``resuming=True`` means tool calls follow (spinner should restart);
``resuming=False`` means this is the final response.
"""
messages = initial_messages
iteration = 0
final_content = None
tools_used: list[str] = []
loop_self = self
# Wrap on_stream with stateful think-tag filter so downstream
# consumers (CLI, channels) never see <think> blocks.
_raw_stream = on_stream
_stream_buf = ""
class _LoopHook(AgentHook):
def __init__(self) -> None:
self._stream_buf = ""
async def _filtered_stream(delta: str) -> None:
nonlocal _stream_buf
from nanobot.utils.helpers import strip_think
prev_clean = strip_think(_stream_buf)
_stream_buf += delta
new_clean = strip_think(_stream_buf)
incremental = new_clean[len(prev_clean):]
if incremental and _raw_stream:
await _raw_stream(incremental)
def wants_streaming(self) -> bool:
return on_stream is not None
while iteration < self.max_iterations:
iteration += 1
async def on_stream(self, context: AgentHookContext, delta: str) -> None:
from nanobot.utils.helpers import strip_think
tool_defs = self.tools.get_definitions()
prev_clean = strip_think(self._stream_buf)
self._stream_buf += delta
new_clean = strip_think(self._stream_buf)
incremental = new_clean[len(prev_clean):]
if incremental and on_stream:
await on_stream(incremental)
if on_stream:
response = await self.provider.chat_stream_with_retry(
messages=messages,
tools=tool_defs,
model=self.model,
on_content_delta=_filtered_stream,
)
else:
response = await self.provider.chat_with_retry(
messages=messages,
tools=tool_defs,
model=self.model,
)
usage = response.usage or {}
self._last_usage = {
"prompt_tokens": int(usage.get("prompt_tokens", 0) or 0),
"completion_tokens": int(usage.get("completion_tokens", 0) or 0),
}
if response.has_tool_calls:
if on_stream and on_stream_end:
await on_stream_end(resuming=True)
_stream_buf = ""
async def on_stream_end(self, context: AgentHookContext, *, resuming: bool) -> None:
if on_stream_end:
await on_stream_end(resuming=resuming)
self._stream_buf = ""
async def before_execute_tools(self, context: AgentHookContext) -> None:
if on_progress:
if not on_stream:
thought = self._strip_think(response.content)
thought = loop_self._strip_think(context.response.content if context.response else None)
if thought:
await on_progress(thought)
tool_hint = self._tool_hint(response.tool_calls)
tool_hint = self._strip_think(tool_hint)
tool_hint = loop_self._strip_think(loop_self._tool_hint(context.tool_calls))
await on_progress(tool_hint, tool_hint=True)
tool_call_dicts = [
tc.to_openai_tool_call()
for tc in response.tool_calls
]
messages = self.context.add_assistant_message(
messages, response.content, tool_call_dicts,
reasoning_content=response.reasoning_content,
thinking_blocks=response.thinking_blocks,
)
for tc in response.tool_calls:
tools_used.append(tc.name)
for tc in context.tool_calls:
args_str = json.dumps(tc.arguments, ensure_ascii=False)
logger.info("Tool call: {}({})", tc.name, args_str[:200])
loop_self._set_tool_context(channel, chat_id, message_id)
# Re-bind tool context right before execution so that
# concurrent sessions don't clobber each other's routing.
self._set_tool_context(channel, chat_id, message_id)
def finalize_content(self, context: AgentHookContext, content: str | None) -> str | None:
return loop_self._strip_think(content)
# Execute all tool calls concurrently — the LLM batches
# independent calls in a single response on purpose.
# return_exceptions=True ensures all results are collected
# even if one tool is cancelled or raises BaseException.
results = await asyncio.gather(*(
self.tools.execute(tc.name, tc.arguments)
for tc in response.tool_calls
), return_exceptions=True)
for tool_call, result in zip(response.tool_calls, results):
if isinstance(result, BaseException):
result = f"Error: {type(result).__name__}: {result}"
messages = self.context.add_tool_result(
messages, tool_call.id, tool_call.name, result
)
else:
if on_stream and on_stream_end:
await on_stream_end(resuming=False)
_stream_buf = ""
clean = self._strip_think(response.content)
if response.finish_reason == "error":
logger.error("LLM returned error: {}", (clean or "")[:200])
final_content = clean or "Sorry, I encountered an error calling the AI model."
break
messages = self.context.add_assistant_message(
messages, clean, reasoning_content=response.reasoning_content,
thinking_blocks=response.thinking_blocks,
)
final_content = clean
break
if final_content is None and iteration >= self.max_iterations:
result = await self.runner.run(AgentRunSpec(
initial_messages=initial_messages,
tools=self.tools,
model=self.model,
max_iterations=self.max_iterations,
hook=_LoopHook(),
error_message="Sorry, I encountered an error calling the AI model.",
concurrent_tools=True,
))
self._last_usage = result.usage
if result.stop_reason == "max_iterations":
logger.warning("Max iterations ({}) reached", self.max_iterations)
final_content = (
f"I reached the maximum number of tool call iterations ({self.max_iterations}) "
"without completing the task. You can try breaking the task into smaller steps."
)
return final_content, tools_used, messages
elif result.stop_reason == "error":
logger.error("LLM returned error: {}", (result.final_content or "")[:200])
return result.final_content, result.tools_used, result.messages
async def run(self) -> None:
"""Run the agent loop, dispatching messages as tasks to stay responsive to /stop."""
@@ -370,17 +313,35 @@ class AgentLoop:
try:
on_stream = on_stream_end = None
if msg.metadata.get("_wants_stream"):
# Split one answer into distinct stream segments.
stream_base_id = f"{msg.session_key}:{time.time_ns()}"
stream_segment = 0
def _current_stream_id() -> str:
return f"{stream_base_id}:{stream_segment}"
async def on_stream(delta: str) -> None:
await self.bus.publish_outbound(OutboundMessage(
channel=msg.channel, chat_id=msg.chat_id,
content=delta, metadata={"_stream_delta": True},
content=delta,
metadata={
"_stream_delta": True,
"_stream_id": _current_stream_id(),
},
))
async def on_stream_end(*, resuming: bool = False) -> None:
nonlocal stream_segment
await self.bus.publish_outbound(OutboundMessage(
channel=msg.channel, chat_id=msg.chat_id,
content="", metadata={"_stream_end": True, "_resuming": resuming},
content="",
metadata={
"_stream_end": True,
"_resuming": resuming,
"_stream_id": _current_stream_id(),
},
))
stream_segment += 1
response = await self._process_message(
msg, on_stream=on_stream, on_stream_end=on_stream_end,
+232
View File
@@ -0,0 +1,232 @@
"""Shared execution loop for tool-using agents."""
from __future__ import annotations
import asyncio
from dataclasses import dataclass, field
from typing import Any
from nanobot.agent.hook import AgentHook, AgentHookContext
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.providers.base import LLMProvider, ToolCallRequest
from nanobot.utils.helpers import build_assistant_message
_DEFAULT_MAX_ITERATIONS_MESSAGE = (
"I reached the maximum number of tool call iterations ({max_iterations}) "
"without completing the task. You can try breaking the task into smaller steps."
)
_DEFAULT_ERROR_MESSAGE = "Sorry, I encountered an error calling the AI model."
@dataclass(slots=True)
class AgentRunSpec:
"""Configuration for a single agent execution."""
initial_messages: list[dict[str, Any]]
tools: ToolRegistry
model: str
max_iterations: int
temperature: float | None = None
max_tokens: int | None = None
reasoning_effort: str | None = None
hook: AgentHook | None = None
error_message: str | None = _DEFAULT_ERROR_MESSAGE
max_iterations_message: str | None = None
concurrent_tools: bool = False
fail_on_tool_error: bool = False
@dataclass(slots=True)
class AgentRunResult:
"""Outcome of a shared agent execution."""
final_content: str | None
messages: list[dict[str, Any]]
tools_used: list[str] = field(default_factory=list)
usage: dict[str, int] = field(default_factory=dict)
stop_reason: str = "completed"
error: str | None = None
tool_events: list[dict[str, str]] = field(default_factory=list)
class AgentRunner:
"""Run a tool-capable LLM loop without product-layer concerns."""
def __init__(self, provider: LLMProvider):
self.provider = provider
async def run(self, spec: AgentRunSpec) -> AgentRunResult:
hook = spec.hook or AgentHook()
messages = list(spec.initial_messages)
final_content: str | None = None
tools_used: list[str] = []
usage = {"prompt_tokens": 0, "completion_tokens": 0}
error: str | None = None
stop_reason = "completed"
tool_events: list[dict[str, str]] = []
for iteration in range(spec.max_iterations):
context = AgentHookContext(iteration=iteration, messages=messages)
await hook.before_iteration(context)
kwargs: dict[str, Any] = {
"messages": messages,
"tools": spec.tools.get_definitions(),
"model": spec.model,
}
if spec.temperature is not None:
kwargs["temperature"] = spec.temperature
if spec.max_tokens is not None:
kwargs["max_tokens"] = spec.max_tokens
if spec.reasoning_effort is not None:
kwargs["reasoning_effort"] = spec.reasoning_effort
if hook.wants_streaming():
async def _stream(delta: str) -> None:
await hook.on_stream(context, delta)
response = await self.provider.chat_stream_with_retry(
**kwargs,
on_content_delta=_stream,
)
else:
response = await self.provider.chat_with_retry(**kwargs)
raw_usage = response.usage or {}
usage = {
"prompt_tokens": int(raw_usage.get("prompt_tokens", 0) or 0),
"completion_tokens": int(raw_usage.get("completion_tokens", 0) or 0),
}
context.response = response
context.usage = usage
context.tool_calls = list(response.tool_calls)
if response.has_tool_calls:
if hook.wants_streaming():
await hook.on_stream_end(context, resuming=True)
messages.append(build_assistant_message(
response.content or "",
tool_calls=[tc.to_openai_tool_call() for tc in response.tool_calls],
reasoning_content=response.reasoning_content,
thinking_blocks=response.thinking_blocks,
))
tools_used.extend(tc.name for tc in response.tool_calls)
await hook.before_execute_tools(context)
results, new_events, fatal_error = await self._execute_tools(spec, response.tool_calls)
tool_events.extend(new_events)
context.tool_results = list(results)
context.tool_events = list(new_events)
if fatal_error is not None:
error = f"Error: {type(fatal_error).__name__}: {fatal_error}"
stop_reason = "tool_error"
context.error = error
context.stop_reason = stop_reason
await hook.after_iteration(context)
break
for tool_call, result in zip(response.tool_calls, results):
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"name": tool_call.name,
"content": result,
})
await hook.after_iteration(context)
continue
if hook.wants_streaming():
await hook.on_stream_end(context, resuming=False)
clean = hook.finalize_content(context, response.content)
if response.finish_reason == "error":
final_content = clean or spec.error_message or _DEFAULT_ERROR_MESSAGE
stop_reason = "error"
error = final_content
context.final_content = final_content
context.error = error
context.stop_reason = stop_reason
await hook.after_iteration(context)
break
messages.append(build_assistant_message(
clean,
reasoning_content=response.reasoning_content,
thinking_blocks=response.thinking_blocks,
))
final_content = clean
context.final_content = final_content
context.stop_reason = stop_reason
await hook.after_iteration(context)
break
else:
stop_reason = "max_iterations"
template = spec.max_iterations_message or _DEFAULT_MAX_ITERATIONS_MESSAGE
final_content = template.format(max_iterations=spec.max_iterations)
return AgentRunResult(
final_content=final_content,
messages=messages,
tools_used=tools_used,
usage=usage,
stop_reason=stop_reason,
error=error,
tool_events=tool_events,
)
async def _execute_tools(
self,
spec: AgentRunSpec,
tool_calls: list[ToolCallRequest],
) -> tuple[list[Any], list[dict[str, str]], BaseException | None]:
if spec.concurrent_tools:
tool_results = await asyncio.gather(*(
self._run_tool(spec, tool_call)
for tool_call in tool_calls
))
else:
tool_results = [
await self._run_tool(spec, tool_call)
for tool_call in tool_calls
]
results: list[Any] = []
events: list[dict[str, str]] = []
fatal_error: BaseException | None = None
for result, event, error in tool_results:
results.append(result)
events.append(event)
if error is not None and fatal_error is None:
fatal_error = error
return results, events, fatal_error
async def _run_tool(
self,
spec: AgentRunSpec,
tool_call: ToolCallRequest,
) -> tuple[Any, dict[str, str], BaseException | None]:
try:
result = await spec.tools.execute(tool_call.name, tool_call.arguments)
except asyncio.CancelledError:
raise
except BaseException as exc:
event = {
"name": tool_call.name,
"status": "error",
"detail": str(exc),
}
if spec.fail_on_tool_error:
return f"Error: {type(exc).__name__}: {exc}", event, exc
return f"Error: {type(exc).__name__}: {exc}", event, None
detail = "" if result is None else str(result)
detail = detail.replace("\n", " ").strip()
if not detail:
detail = "(empty)"
elif len(detail) > 120:
detail = detail[:120] + "..."
return result, {
"name": tool_call.name,
"status": "error" if isinstance(result, str) and result.startswith("Error") else "ok",
"detail": detail,
}, None
+58 -41
View File
@@ -8,6 +8,8 @@ from typing import Any
from loguru import logger
from nanobot.agent.hook import AgentHook, AgentHookContext
from nanobot.agent.runner import AgentRunSpec, AgentRunner
from nanobot.agent.skills import BUILTIN_SKILLS_DIR
from nanobot.agent.tools.filesystem import EditFileTool, ListDirTool, ReadFileTool, WriteFileTool
from nanobot.agent.tools.registry import ToolRegistry
@@ -17,7 +19,6 @@ from nanobot.bus.events import InboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.config.schema import ExecToolConfig
from nanobot.providers.base import LLMProvider
from nanobot.utils.helpers import build_assistant_message
class SubagentManager:
@@ -44,6 +45,7 @@ class SubagentManager:
self.web_proxy = web_proxy
self.exec_config = exec_config or ExecToolConfig()
self.restrict_to_workspace = restrict_to_workspace
self.runner = AgentRunner(provider)
self._running_tasks: dict[str, asyncio.Task[None]] = {}
self._session_tasks: dict[str, set[str]] = {} # session_key -> {task_id, ...}
@@ -113,49 +115,43 @@ class SubagentManager:
{"role": "user", "content": task},
]
# Run agent loop (limited iterations)
max_iterations = 15
iteration = 0
final_result: str | None = None
while iteration < max_iterations:
iteration += 1
response = await self.provider.chat_with_retry(
messages=messages,
tools=tools.get_definitions(),
model=self.model,
)
if response.has_tool_calls:
tool_call_dicts = [
tc.to_openai_tool_call()
for tc in response.tool_calls
]
messages.append(build_assistant_message(
response.content or "",
tool_calls=tool_call_dicts,
reasoning_content=response.reasoning_content,
thinking_blocks=response.thinking_blocks,
))
# Execute tools
for tool_call in response.tool_calls:
class _SubagentHook(AgentHook):
async def before_execute_tools(self, context: AgentHookContext) -> None:
for tool_call in context.tool_calls:
args_str = json.dumps(tool_call.arguments, ensure_ascii=False)
logger.debug("Subagent [{}] executing: {} with arguments: {}", task_id, tool_call.name, args_str)
result = await tools.execute(tool_call.name, tool_call.arguments)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"name": tool_call.name,
"content": result,
})
else:
final_result = response.content
break
if final_result is None:
final_result = "Task completed but no final response was generated."
result = await self.runner.run(AgentRunSpec(
initial_messages=messages,
tools=tools,
model=self.model,
max_iterations=15,
hook=_SubagentHook(),
max_iterations_message="Task completed but no final response was generated.",
error_message=None,
fail_on_tool_error=True,
))
if result.stop_reason == "tool_error":
await self._announce_result(
task_id,
label,
task,
self._format_partial_progress(result),
origin,
"error",
)
return
if result.stop_reason == "error":
await self._announce_result(
task_id,
label,
task,
result.error or "Error: subagent execution failed.",
origin,
"error",
)
return
final_result = result.final_content or "Task completed but no final response was generated."
logger.info("Subagent [{}] completed successfully", task_id)
await self._announce_result(task_id, label, task, final_result, origin, "ok")
@@ -196,6 +192,27 @@ Summarize this naturally for the user. Keep it brief (1-2 sentences). Do not men
await self.bus.publish_inbound(msg)
logger.debug("Subagent [{}] announced result to {}:{}", task_id, origin['channel'], origin['chat_id'])
@staticmethod
def _format_partial_progress(result) -> str:
completed = [e for e in result.tool_events if e["status"] == "ok"]
failure = next((e for e in reversed(result.tool_events) if e["status"] == "error"), None)
lines: list[str] = []
if completed:
lines.append("Completed steps:")
for event in completed[-3:]:
lines.append(f"- {event['name']}: {event['detail']}")
if failure:
if lines:
lines.append("")
lines.append("Failure:")
lines.append(f"- {failure['name']}: {failure['detail']}")
if result.error and not failure:
if lines:
lines.append("")
lines.append("Failure:")
lines.append(f"- {result.error}")
return "\n".join(lines) or (result.error or "Error: subagent execution failed.")
def _build_subagent_prompt(self) -> str:
"""Build a focused system prompt for the subagent."""
+57 -24
View File
@@ -1,7 +1,7 @@
"""Cron tool for scheduling reminders and tasks."""
from contextvars import ContextVar
from datetime import datetime, timezone
from datetime import datetime
from typing import Any
from nanobot.agent.tools.base import Tool
@@ -12,8 +12,9 @@ from nanobot.cron.types import CronJobState, CronSchedule
class CronTool(Tool):
"""Tool to schedule reminders and recurring tasks."""
def __init__(self, cron_service: CronService):
def __init__(self, cron_service: CronService, default_timezone: str = "UTC"):
self._cron = cron_service
self._default_timezone = default_timezone
self._channel = ""
self._chat_id = ""
self._in_cron_context: ContextVar[bool] = ContextVar("cron_in_context", default=False)
@@ -31,13 +32,37 @@ class CronTool(Tool):
"""Restore previous cron context."""
self._in_cron_context.reset(token)
@staticmethod
def _validate_timezone(tz: str) -> str | None:
from zoneinfo import ZoneInfo
try:
ZoneInfo(tz)
except (KeyError, Exception):
return f"Error: unknown timezone '{tz}'"
return None
def _display_timezone(self, schedule: CronSchedule) -> str:
"""Pick the most human-meaningful timezone for display."""
return schedule.tz or self._default_timezone
@staticmethod
def _format_timestamp(ms: int, tz_name: str) -> str:
from zoneinfo import ZoneInfo
dt = datetime.fromtimestamp(ms / 1000, tz=ZoneInfo(tz_name))
return f"{dt.isoformat()} ({tz_name})"
@property
def name(self) -> str:
return "cron"
@property
def description(self) -> str:
return "Schedule reminders and recurring tasks. Actions: add, list, remove."
return (
"Schedule reminders and recurring tasks. Actions: add, list, remove. "
f"If tz is omitted, cron expressions and naive ISO times default to {self._default_timezone}."
)
@property
def parameters(self) -> dict[str, Any]:
@@ -60,11 +85,17 @@ class CronTool(Tool):
},
"tz": {
"type": "string",
"description": "IANA timezone for cron expressions (e.g. 'America/Vancouver')",
"description": (
"Optional IANA timezone for cron expressions "
f"(e.g. 'America/Vancouver'). Defaults to {self._default_timezone}."
),
},
"at": {
"type": "string",
"description": "ISO datetime for one-time execution (e.g. '2026-02-12T10:30:00')",
"description": (
"ISO datetime for one-time execution "
f"(e.g. '2026-02-12T10:30:00'). Naive values default to {self._default_timezone}."
),
},
"job_id": {"type": "string", "description": "Job ID (for remove)"},
},
@@ -107,26 +138,29 @@ class CronTool(Tool):
if tz and not cron_expr:
return "Error: tz can only be used with cron_expr"
if tz:
from zoneinfo import ZoneInfo
try:
ZoneInfo(tz)
except (KeyError, Exception):
return f"Error: unknown timezone '{tz}'"
if err := self._validate_timezone(tz):
return err
# Build schedule
delete_after = False
if every_seconds:
schedule = CronSchedule(kind="every", every_ms=every_seconds * 1000)
elif cron_expr:
schedule = CronSchedule(kind="cron", expr=cron_expr, tz=tz)
effective_tz = tz or self._default_timezone
if err := self._validate_timezone(effective_tz):
return err
schedule = CronSchedule(kind="cron", expr=cron_expr, tz=effective_tz)
elif at:
from datetime import datetime
from zoneinfo import ZoneInfo
try:
dt = datetime.fromisoformat(at)
except ValueError:
return f"Error: invalid ISO datetime format '{at}'. Expected format: YYYY-MM-DDTHH:MM:SS"
if dt.tzinfo is None:
if err := self._validate_timezone(self._default_timezone):
return err
dt = dt.replace(tzinfo=ZoneInfo(self._default_timezone))
at_ms = int(dt.timestamp() * 1000)
schedule = CronSchedule(kind="at", at_ms=at_ms)
delete_after = True
@@ -144,8 +178,7 @@ class CronTool(Tool):
)
return f"Created job '{job.name}' (id: {job.id})"
@staticmethod
def _format_timing(schedule: CronSchedule) -> str:
def _format_timing(self, schedule: CronSchedule) -> str:
"""Format schedule as a human-readable timing string."""
if schedule.kind == "cron":
tz = f" ({schedule.tz})" if schedule.tz else ""
@@ -160,23 +193,23 @@ class CronTool(Tool):
return f"every {ms // 1000}s"
return f"every {ms}ms"
if schedule.kind == "at" and schedule.at_ms:
dt = datetime.fromtimestamp(schedule.at_ms / 1000, tz=timezone.utc)
return f"at {dt.isoformat()}"
return f"at {self._format_timestamp(schedule.at_ms, self._display_timezone(schedule))}"
return schedule.kind
@staticmethod
def _format_state(state: CronJobState) -> list[str]:
def _format_state(self, state: CronJobState, schedule: CronSchedule) -> list[str]:
"""Format job run state as display lines."""
lines: list[str] = []
display_tz = self._display_timezone(schedule)
if state.last_run_at_ms:
last_dt = datetime.fromtimestamp(state.last_run_at_ms / 1000, tz=timezone.utc)
info = f" Last run: {last_dt.isoformat()}{state.last_status or 'unknown'}"
info = (
f" Last run: {self._format_timestamp(state.last_run_at_ms, display_tz)}"
f"{state.last_status or 'unknown'}"
)
if state.last_error:
info += f" ({state.last_error})"
lines.append(info)
if state.next_run_at_ms:
next_dt = datetime.fromtimestamp(state.next_run_at_ms / 1000, tz=timezone.utc)
lines.append(f" Next run: {next_dt.isoformat()}")
lines.append(f" Next run: {self._format_timestamp(state.next_run_at_ms, display_tz)}")
return lines
def _list_jobs(self) -> str:
@@ -187,7 +220,7 @@ class CronTool(Tool):
for j in jobs:
timing = self._format_timing(j.schedule)
parts = [f"- {j.name} (id: {j.id}, {timing})"]
parts.extend(self._format_state(j.state))
parts.extend(self._format_state(j.state, j.schedule))
lines.append("\n".join(parts))
return "Scheduled jobs:\n" + "\n".join(lines)
+12 -1
View File
@@ -85,11 +85,22 @@ class BaseChannel(ABC):
Args:
msg: The message to send.
Implementations should raise on delivery failure so the channel manager
can apply any retry policy in one place.
"""
pass
async def send_delta(self, chat_id: str, delta: str, metadata: dict[str, Any] | None = None) -> None:
"""Deliver a streaming text chunk. Override in subclass to enable streaming."""
"""Deliver a streaming text chunk.
Override in subclasses to enable streaming. Implementations should
raise on delivery failure so the channel manager can retry.
Streaming contract: ``_stream_delta`` is a chunk, ``_stream_end`` ends
the current segment, and stateful implementations must key buffers by
``_stream_id`` rather than only by ``chat_id``.
"""
pass
@property
+45 -1
View File
@@ -51,6 +51,10 @@ class EmailConfig(Base):
subject_prefix: str = "Re: "
allow_from: list[str] = Field(default_factory=list)
# Email authentication verification (anti-spoofing)
verify_dkim: bool = True # Require Authentication-Results with dkim=pass
verify_spf: bool = True # Require Authentication-Results with spf=pass
class EmailChannel(BaseChannel):
"""
@@ -123,6 +127,12 @@ class EmailChannel(BaseChannel):
return
self._running = True
if not self.config.verify_dkim and not self.config.verify_spf:
logger.warning(
"Email channel: DKIM and SPF verification are both DISABLED. "
"Emails with spoofed From headers will be accepted. "
"Set verify_dkim=true and verify_spf=true for anti-spoofing protection."
)
logger.info("Starting Email channel (IMAP polling mode)...")
poll_seconds = max(5, int(self.config.poll_interval_seconds))
@@ -360,6 +370,23 @@ class EmailChannel(BaseChannel):
if not sender:
continue
# --- Anti-spoofing: verify Authentication-Results ---
spf_pass, dkim_pass = self._check_authentication_results(parsed)
if self.config.verify_spf and not spf_pass:
logger.warning(
"Email from {} rejected: SPF verification failed "
"(no 'spf=pass' in Authentication-Results header)",
sender,
)
continue
if self.config.verify_dkim and not dkim_pass:
logger.warning(
"Email from {} rejected: DKIM verification failed "
"(no 'dkim=pass' in Authentication-Results header)",
sender,
)
continue
subject = self._decode_header_value(parsed.get("Subject", ""))
date_value = parsed.get("Date", "")
message_id = parsed.get("Message-ID", "").strip()
@@ -370,7 +397,7 @@ class EmailChannel(BaseChannel):
body = body[: self.config.max_body_chars]
content = (
f"Email received.\n"
f"[EMAIL-CONTEXT] Email received.\n"
f"From: {sender}\n"
f"Subject: {subject}\n"
f"Date: {date_value}\n\n"
@@ -493,6 +520,23 @@ class EmailChannel(BaseChannel):
return cls._html_to_text(payload).strip()
return payload.strip()
@staticmethod
def _check_authentication_results(parsed_msg: Any) -> tuple[bool, bool]:
"""Parse Authentication-Results headers for SPF and DKIM verdicts.
Returns:
A tuple of (spf_pass, dkim_pass) booleans.
"""
spf_pass = False
dkim_pass = False
for ar_header in parsed_msg.get_all("Authentication-Results") or []:
ar_lower = ar_header.lower()
if re.search(r"\bspf\s*=\s*pass\b", ar_lower):
spf_pass = True
if re.search(r"\bdkim\s*=\s*pass\b", ar_lower):
dkim_pass = True
return spf_pass, dkim_pass
@staticmethod
def _html_to_text(raw_html: str) -> str:
text = re.sub(r"<\s*br\s*/?>", "\n", raw_html, flags=re.IGNORECASE)
+161 -5
View File
@@ -5,7 +5,10 @@ import json
import os
import re
import threading
import time
import uuid
from collections import OrderedDict
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Literal
@@ -248,6 +251,19 @@ class FeishuConfig(Base):
react_emoji: str = "THUMBSUP"
group_policy: Literal["open", "mention"] = "mention"
reply_to_message: bool = False # If True, bot replies quote the user's original message
streaming: bool = True
_STREAM_ELEMENT_ID = "streaming_md"
@dataclass
class _FeishuStreamBuf:
"""Per-chat streaming accumulator using CardKit streaming API."""
text: str = ""
card_id: str | None = None
sequence: int = 0
last_edit: float = 0.0
class FeishuChannel(BaseChannel):
@@ -265,6 +281,8 @@ class FeishuChannel(BaseChannel):
name = "feishu"
display_name = "Feishu"
_STREAM_EDIT_INTERVAL = 0.5 # throttle between CardKit streaming updates
@classmethod
def default_config(cls) -> dict[str, Any]:
return FeishuConfig().model_dump(by_alias=True)
@@ -279,6 +297,7 @@ class FeishuChannel(BaseChannel):
self._ws_thread: threading.Thread | None = None
self._processed_message_ids: OrderedDict[str, None] = OrderedDict() # Ordered dedup cache
self._loop: asyncio.AbstractEventLoop | None = None
self._stream_bufs: dict[str, _FeishuStreamBuf] = {}
@staticmethod
def _register_optional_event(builder: Any, method_name: str, handler: Any) -> Any:
@@ -906,8 +925,8 @@ class FeishuChannel(BaseChannel):
logger.error("Error replying to Feishu message {}: {}", parent_message_id, e)
return False
def _send_message_sync(self, receive_id_type: str, receive_id: str, msg_type: str, content: str) -> bool:
"""Send a single message (text/image/file/interactive) synchronously."""
def _send_message_sync(self, receive_id_type: str, receive_id: str, msg_type: str, content: str) -> str | None:
"""Send a single message and return the message_id on success."""
from lark_oapi.api.im.v1 import CreateMessageRequest, CreateMessageRequestBody
try:
request = CreateMessageRequest.builder() \
@@ -925,13 +944,149 @@ class FeishuChannel(BaseChannel):
"Failed to send Feishu {} message: code={}, msg={}, log_id={}",
msg_type, response.code, response.msg, response.get_log_id()
)
return False
logger.debug("Feishu {} message sent to {}", msg_type, receive_id)
return True
return None
msg_id = getattr(response.data, "message_id", None)
logger.debug("Feishu {} message sent to {}: {}", msg_type, receive_id, msg_id)
return msg_id
except Exception as e:
logger.error("Error sending Feishu {} message: {}", msg_type, e)
return None
def _create_streaming_card_sync(self, receive_id_type: str, chat_id: str) -> str | None:
"""Create a CardKit streaming card, send it to chat, return card_id."""
from lark_oapi.api.cardkit.v1 import CreateCardRequest, CreateCardRequestBody
card_json = {
"schema": "2.0",
"config": {"wide_screen_mode": True, "update_multi": True, "streaming_mode": True},
"body": {"elements": [{"tag": "markdown", "content": "", "element_id": _STREAM_ELEMENT_ID}]},
}
try:
request = CreateCardRequest.builder().request_body(
CreateCardRequestBody.builder()
.type("card_json")
.data(json.dumps(card_json, ensure_ascii=False))
.build()
).build()
response = self._client.cardkit.v1.card.create(request)
if not response.success():
logger.warning("Failed to create streaming card: code={}, msg={}", response.code, response.msg)
return None
card_id = getattr(response.data, "card_id", None)
if card_id:
message_id = self._send_message_sync(
receive_id_type, chat_id, "interactive",
json.dumps({"type": "card", "data": {"card_id": card_id}}),
)
if message_id:
return card_id
logger.warning("Created streaming card {} but failed to send it to {}", card_id, chat_id)
return None
except Exception as e:
logger.warning("Error creating streaming card: {}", e)
return None
def _stream_update_text_sync(self, card_id: str, content: str, sequence: int) -> bool:
"""Stream-update the markdown element on a CardKit card (typewriter effect)."""
from lark_oapi.api.cardkit.v1 import ContentCardElementRequest, ContentCardElementRequestBody
try:
request = ContentCardElementRequest.builder() \
.card_id(card_id) \
.element_id(_STREAM_ELEMENT_ID) \
.request_body(
ContentCardElementRequestBody.builder()
.content(content).sequence(sequence).build()
).build()
response = self._client.cardkit.v1.card_element.content(request)
if not response.success():
logger.warning("Failed to stream-update card {}: code={}, msg={}", card_id, response.code, response.msg)
return False
return True
except Exception as e:
logger.warning("Error stream-updating card {}: {}", card_id, e)
return False
def _close_streaming_mode_sync(self, card_id: str, sequence: int) -> bool:
"""Turn off CardKit streaming_mode so the chat list preview exits the streaming placeholder.
Per Feishu docs, streaming cards keep a generating-style summary in the session list until
streaming_mode is set to false via card settings (after final content update).
Sequence must strictly exceed the previous card OpenAPI operation on this entity.
"""
from lark_oapi.api.cardkit.v1 import SettingsCardRequest, SettingsCardRequestBody
settings_payload = json.dumps({"config": {"streaming_mode": False}}, ensure_ascii=False)
try:
request = SettingsCardRequest.builder() \
.card_id(card_id) \
.request_body(
SettingsCardRequestBody.builder()
.settings(settings_payload)
.sequence(sequence)
.uuid(str(uuid.uuid4()))
.build()
).build()
response = self._client.cardkit.v1.card.settings(request)
if not response.success():
logger.warning(
"Failed to close streaming on card {}: code={}, msg={}",
card_id, response.code, response.msg,
)
return False
return True
except Exception as e:
logger.warning("Error closing streaming on card {}: {}", card_id, e)
return False
async def send_delta(self, chat_id: str, delta: str, metadata: dict[str, Any] | None = None) -> None:
"""Progressive streaming via CardKit: create card on first delta, stream-update on subsequent."""
if not self._client:
return
meta = metadata or {}
loop = asyncio.get_running_loop()
rid_type = "chat_id" if chat_id.startswith("oc_") else "open_id"
# --- stream end: final update or fallback ---
if meta.get("_stream_end"):
buf = self._stream_bufs.pop(chat_id, None)
if not buf or not buf.text:
return
if buf.card_id:
buf.sequence += 1
await loop.run_in_executor(
None, self._stream_update_text_sync, buf.card_id, buf.text, buf.sequence,
)
# Required so the chat list preview exits the streaming placeholder (Feishu streaming card docs).
buf.sequence += 1
await loop.run_in_executor(
None, self._close_streaming_mode_sync, buf.card_id, buf.sequence,
)
else:
for chunk in self._split_elements_by_table_limit(self._build_card_elements(buf.text)):
card = json.dumps({"config": {"wide_screen_mode": True}, "elements": chunk}, ensure_ascii=False)
await loop.run_in_executor(None, self._send_message_sync, rid_type, chat_id, "interactive", card)
return
# --- accumulate delta ---
buf = self._stream_bufs.get(chat_id)
if buf is None:
buf = _FeishuStreamBuf()
self._stream_bufs[chat_id] = buf
buf.text += delta
if not buf.text.strip():
return
now = time.monotonic()
if buf.card_id is None:
card_id = await loop.run_in_executor(None, self._create_streaming_card_sync, rid_type, chat_id)
if card_id:
buf.card_id = card_id
buf.sequence = 1
await loop.run_in_executor(None, self._stream_update_text_sync, card_id, buf.text, 1)
buf.last_edit = now
elif (now - buf.last_edit) >= self._STREAM_EDIT_INTERVAL:
buf.sequence += 1
await loop.run_in_executor(None, self._stream_update_text_sync, buf.card_id, buf.text, buf.sequence)
buf.last_edit = now
async def send(self, msg: OutboundMessage) -> None:
"""Send a message through Feishu, including media (images/files) if present."""
if not self._client:
@@ -1031,6 +1186,7 @@ class FeishuChannel(BaseChannel):
except Exception as e:
logger.error("Error sending Feishu message: {}", e)
raise
def _on_message_sync(self, data: Any) -> None:
"""
+111 -13
View File
@@ -7,10 +7,14 @@ from typing import Any
from loguru import logger
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.config.schema import Config
# Retry delays for message sending (exponential backoff: 1s, 2s, 4s)
_SEND_RETRY_DELAYS = (1, 2, 4)
class ChannelManager:
"""
@@ -114,12 +118,20 @@ class ChannelManager:
"""Dispatch outbound messages to the appropriate channel."""
logger.info("Outbound dispatcher started")
# Buffer for messages that couldn't be processed during delta coalescing
# (since asyncio.Queue doesn't support push_front)
pending: list[OutboundMessage] = []
while True:
try:
msg = await asyncio.wait_for(
self.bus.consume_outbound(),
timeout=1.0
)
# First check pending buffer before waiting on queue
if pending:
msg = pending.pop(0)
else:
msg = await asyncio.wait_for(
self.bus.consume_outbound(),
timeout=1.0
)
if msg.metadata.get("_progress"):
if msg.metadata.get("_tool_hint") and not self.config.channels.send_tool_hints:
@@ -127,17 +139,15 @@ class ChannelManager:
if not msg.metadata.get("_tool_hint") and not self.config.channels.send_progress:
continue
# Coalesce consecutive _stream_delta messages for the same (channel, chat_id)
# to reduce API calls and improve streaming latency
if msg.metadata.get("_stream_delta") and not msg.metadata.get("_stream_end"):
msg, extra_pending = self._coalesce_stream_deltas(msg)
pending.extend(extra_pending)
channel = self.channels.get(msg.channel)
if channel:
try:
if msg.metadata.get("_stream_delta") or msg.metadata.get("_stream_end"):
await channel.send_delta(msg.chat_id, msg.content, msg.metadata)
elif msg.metadata.get("_streamed"):
pass
else:
await channel.send(msg)
except Exception as e:
logger.error("Error sending to {}: {}", msg.channel, e)
await self._send_with_retry(channel, msg)
else:
logger.warning("Unknown channel: {}", msg.channel)
@@ -146,6 +156,94 @@ class ChannelManager:
except asyncio.CancelledError:
break
@staticmethod
async def _send_once(channel: BaseChannel, msg: OutboundMessage) -> None:
"""Send one outbound message without retry policy."""
if msg.metadata.get("_stream_delta") or msg.metadata.get("_stream_end"):
await channel.send_delta(msg.chat_id, msg.content, msg.metadata)
elif not msg.metadata.get("_streamed"):
await channel.send(msg)
def _coalesce_stream_deltas(
self, first_msg: OutboundMessage
) -> tuple[OutboundMessage, list[OutboundMessage]]:
"""Merge consecutive _stream_delta messages for the same (channel, chat_id).
This reduces the number of API calls when the queue has accumulated multiple
deltas, which happens when LLM generates faster than the channel can process.
Returns:
tuple of (merged_message, list_of_non_matching_messages)
"""
target_key = (first_msg.channel, first_msg.chat_id)
combined_content = first_msg.content
final_metadata = dict(first_msg.metadata or {})
non_matching: list[OutboundMessage] = []
# Only merge consecutive deltas. As soon as we hit any other message,
# stop and hand that boundary back to the dispatcher via `pending`.
while True:
try:
next_msg = self.bus.outbound.get_nowait()
except asyncio.QueueEmpty:
break
# Check if this message belongs to the same stream
same_target = (next_msg.channel, next_msg.chat_id) == target_key
is_delta = next_msg.metadata and next_msg.metadata.get("_stream_delta")
is_end = next_msg.metadata and next_msg.metadata.get("_stream_end")
if same_target and is_delta and not final_metadata.get("_stream_end"):
# Accumulate content
combined_content += next_msg.content
# If we see _stream_end, remember it and stop coalescing this stream
if is_end:
final_metadata["_stream_end"] = True
# Stream ended - stop coalescing this stream
break
else:
# First non-matching message defines the coalescing boundary.
non_matching.append(next_msg)
break
merged = OutboundMessage(
channel=first_msg.channel,
chat_id=first_msg.chat_id,
content=combined_content,
metadata=final_metadata,
)
return merged, non_matching
async def _send_with_retry(self, channel: BaseChannel, msg: OutboundMessage) -> None:
"""Send a message with retry on failure using exponential backoff.
Note: CancelledError is re-raised to allow graceful shutdown.
"""
max_attempts = max(self.config.channels.send_max_retries, 1)
for attempt in range(max_attempts):
try:
await self._send_once(channel, msg)
return # Send succeeded
except asyncio.CancelledError:
raise # Propagate cancellation for graceful shutdown
except Exception as e:
if attempt == max_attempts - 1:
logger.error(
"Failed to send to {} after {} attempts: {} - {}",
msg.channel, max_attempts, type(e).__name__, e
)
return
delay = _SEND_RETRY_DELAYS[min(attempt, len(_SEND_RETRY_DELAYS) - 1)]
logger.warning(
"Send to {} failed (attempt {}/{}): {}, retrying in {}s",
msg.channel, attempt + 1, max_attempts, type(e).__name__, delay
)
try:
await asyncio.sleep(delay)
except asyncio.CancelledError:
raise # Propagate cancellation during sleep
def get_channel(self, name: str) -> BaseChannel | None:
"""Get a channel by name."""
return self.channels.get(name)
+1
View File
@@ -374,6 +374,7 @@ class MochatChannel(BaseChannel):
content, msg.reply_to)
except Exception as e:
logger.error("Failed to send Mochat message: {}", e)
raise
# ---- config / init helpers ---------------------------------------------
+1
View File
@@ -145,6 +145,7 @@ class SlackChannel(BaseChannel):
except Exception as e:
logger.error("Error sending Slack message: {}", e)
raise
async def _on_socket_request(
self,
+40 -9
View File
@@ -12,7 +12,7 @@ from typing import Any, Literal
from loguru import logger
from pydantic import Field
from telegram import BotCommand, ReactionTypeEmoji, ReplyParameters, Update
from telegram.error import TimedOut
from telegram.error import BadRequest, TimedOut
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
from telegram.request import HTTPXRequest
@@ -163,6 +163,7 @@ class _StreamBuf:
text: str = ""
message_id: int | None = None
last_edit: float = 0.0
stream_id: str | None = None
class TelegramConfig(Base):
@@ -476,6 +477,11 @@ class TelegramChannel(BaseChannel):
)
except Exception as e2:
logger.error("Error sending Telegram message: {}", e2)
raise
@staticmethod
def _is_not_modified_error(exc: Exception) -> bool:
return isinstance(exc, BadRequest) and "message is not modified" in str(exc).lower()
async def send_delta(self, chat_id: str, delta: str, metadata: dict[str, Any] | None = None) -> None:
"""Progressive message editing: send on first delta, edit on subsequent ones."""
@@ -483,11 +489,14 @@ class TelegramChannel(BaseChannel):
return
meta = metadata or {}
int_chat_id = int(chat_id)
stream_id = meta.get("_stream_id")
if meta.get("_stream_end"):
buf = self._stream_bufs.pop(chat_id, None)
buf = self._stream_bufs.get(chat_id)
if not buf or not buf.message_id or not buf.text:
return
if stream_id is not None and buf.stream_id is not None and buf.stream_id != stream_id:
return
self._stop_typing(chat_id)
try:
html = _markdown_to_telegram_html(buf.text)
@@ -497,6 +506,10 @@ class TelegramChannel(BaseChannel):
text=html, parse_mode="HTML",
)
except Exception as e:
if self._is_not_modified_error(e):
logger.debug("Final stream edit already applied for {}", chat_id)
self._stream_bufs.pop(chat_id, None)
return
logger.debug("Final stream edit failed (HTML), trying plain: {}", e)
try:
await self._call_with_retry(
@@ -504,14 +517,22 @@ class TelegramChannel(BaseChannel):
chat_id=int_chat_id, message_id=buf.message_id,
text=buf.text,
)
except Exception:
pass
except Exception as e2:
if self._is_not_modified_error(e2):
logger.debug("Final stream plain edit already applied for {}", chat_id)
self._stream_bufs.pop(chat_id, None)
return
logger.warning("Final stream edit failed: {}", e2)
raise # Let ChannelManager handle retry
self._stream_bufs.pop(chat_id, None)
return
buf = self._stream_bufs.get(chat_id)
if buf is None:
buf = _StreamBuf()
if buf is None or (stream_id is not None and buf.stream_id is not None and buf.stream_id != stream_id):
buf = _StreamBuf(stream_id=stream_id)
self._stream_bufs[chat_id] = buf
elif buf.stream_id is None:
buf.stream_id = stream_id
buf.text += delta
if not buf.text.strip():
@@ -528,6 +549,7 @@ class TelegramChannel(BaseChannel):
buf.last_edit = now
except Exception as e:
logger.warning("Stream initial send failed: {}", e)
raise # Let ChannelManager handle retry
elif (now - buf.last_edit) >= self._STREAM_EDIT_INTERVAL:
try:
await self._call_with_retry(
@@ -536,8 +558,12 @@ class TelegramChannel(BaseChannel):
text=buf.text,
)
buf.last_edit = now
except Exception:
pass
except Exception as e:
if self._is_not_modified_error(e):
buf.last_edit = now
return
logger.warning("Stream edit failed: {}", e)
raise # Let ChannelManager handle retry
async def _on_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Handle /start command."""
@@ -890,7 +916,12 @@ class TelegramChannel(BaseChannel):
async def _on_error(self, update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Log polling / handler errors instead of silently swallowing them."""
logger.error("Telegram error: {}", context.error)
from telegram.error import NetworkError, TimedOut
if isinstance(context.error, (NetworkError, TimedOut)):
logger.warning("Telegram network issue: {}", str(context.error))
else:
logger.error("Telegram error: {}", context.error)
def _get_extension(
self,
+1
View File
@@ -368,3 +368,4 @@ class WecomChannel(BaseChannel):
except Exception as e:
logger.error("Error sending WeCom message: {}", e)
raise
+1
View File
@@ -751,6 +751,7 @@ class WeixinChannel(BaseChannel):
await self._send_text(msg.chat_id, chunk, ctx_token)
except Exception as e:
logger.error("Error sending WeChat message: {}", e)
raise
async def _send_text(
self,
+2
View File
@@ -146,6 +146,7 @@ class WhatsAppChannel(BaseChannel):
await self._ws.send(json.dumps(payload, ensure_ascii=False))
except Exception as e:
logger.error("Error sending WhatsApp message: {}", e)
raise
for media_path in msg.media or []:
try:
@@ -160,6 +161,7 @@ class WhatsAppChannel(BaseChannel):
await self._ws.send(json.dumps(payload, ensure_ascii=False))
except Exception as e:
logger.error("Error sending WhatsApp media {}: {}", media_path, e)
raise
async def _handle_bridge_message(self, raw: str) -> None:
"""Handle a message from the bridge."""
+3
View File
@@ -549,6 +549,7 @@ def gateway(
session_manager=session_manager,
mcp_servers=config.tools.mcp_servers,
channels_config=config.channels,
timezone=config.agents.defaults.timezone,
)
# Set cron callback (needs agent)
@@ -659,6 +660,7 @@ def gateway(
on_notify=on_heartbeat_notify,
interval_s=hb_cfg.interval_s,
enabled=hb_cfg.enabled,
timezone=config.agents.defaults.timezone,
)
if channels.enabled_channels:
@@ -752,6 +754,7 @@ def agent(
restrict_to_workspace=config.tools.restrict_to_workspace,
mcp_servers=config.tools.mcp_servers,
channels_config=config.channels,
timezone=config.agents.defaults.timezone,
)
# Shared reference for progress callbacks
+3
View File
@@ -25,6 +25,7 @@ class ChannelsConfig(Base):
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_max_retries: int = Field(default=3, ge=0, le=10) # Max delivery attempts (initial send included)
class AgentDefaults(Base):
@@ -40,6 +41,7 @@ class AgentDefaults(Base):
temperature: float = 0.1
max_tool_iterations: int = 40
reasoning_effort: str | None = None # low / medium / high - enables LLM thinking mode
timezone: str = "UTC" # IANA timezone, e.g. "Asia/Shanghai", "America/New_York"
class AgentsConfig(Base):
@@ -75,6 +77,7 @@ class ProvidersConfig(Base):
moonshot: ProviderConfig = Field(default_factory=ProviderConfig)
minimax: ProviderConfig = Field(default_factory=ProviderConfig)
mistral: ProviderConfig = Field(default_factory=ProviderConfig)
stepfun: ProviderConfig = Field(default_factory=ProviderConfig) # Step Fun (阶跃星辰)
aihubmix: ProviderConfig = Field(default_factory=ProviderConfig) # AiHubMix API gateway
siliconflow: ProviderConfig = Field(default_factory=ProviderConfig) # SiliconFlow (硅基流动)
volcengine: ProviderConfig = Field(default_factory=ProviderConfig) # VolcEngine (火山引擎)
+3 -1
View File
@@ -59,6 +59,7 @@ class HeartbeatService:
on_notify: Callable[[str], Coroutine[Any, Any, None]] | None = None,
interval_s: int = 30 * 60,
enabled: bool = True,
timezone: str | None = None,
):
self.workspace = workspace
self.provider = provider
@@ -67,6 +68,7 @@ class HeartbeatService:
self.on_notify = on_notify
self.interval_s = interval_s
self.enabled = enabled
self.timezone = timezone
self._running = False
self._task: asyncio.Task | None = None
@@ -93,7 +95,7 @@ class HeartbeatService:
messages=[
{"role": "system", "content": "You are a heartbeat agent. Call the heartbeat tool to report your decision."},
{"role": "user", "content": (
f"Current Time: {current_time_str()}\n\n"
f"Current Time: {current_time_str(self.timezone)}\n\n"
"Review the following HEARTBEAT.md and decide whether there are active tasks.\n\n"
f"{content}"
)},
+23 -5
View File
@@ -26,6 +26,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",
}
def _short_tool_id() -> str:
@@ -89,6 +94,13 @@ def _extract_tc_extras(tc: Any) -> tuple[
return extra_content, prov, fn_prov
def _uses_openrouter_attribution(spec: "ProviderSpec | None", api_base: str | None) -> bool:
"""Apply Nanobot attribution headers to OpenRouter requests by default."""
if spec and spec.name == "openrouter":
return True
return bool(api_base and "openrouter" in api_base.lower())
class OpenAICompatProvider(LLMProvider):
"""Unified provider for all OpenAI-compatible APIs.
@@ -113,14 +125,16 @@ class OpenAICompatProvider(LLMProvider):
self._setup_env(api_key, api_base)
effective_base = api_base or (spec.default_api_base if spec else None) or None
default_headers = {"x-session-affinity": uuid.uuid4().hex}
if _uses_openrouter_attribution(spec, effective_base):
default_headers.update(_DEFAULT_OPENROUTER_HEADERS)
if extra_headers:
default_headers.update(extra_headers)
self._client = AsyncOpenAI(
api_key=api_key or "no-key",
base_url=effective_base,
default_headers={
"x-session-affinity": uuid.uuid4().hex,
**(extra_headers or {}),
},
default_headers=default_headers,
)
def _setup_env(self, api_key: str, api_base: str | None) -> None:
@@ -229,10 +243,14 @@ class OpenAICompatProvider(LLMProvider):
kwargs: dict[str, Any] = {
"model": model_name,
"messages": self._sanitize_messages(self._sanitize_empty_content(messages)),
"max_tokens": max(1, max_tokens),
"temperature": temperature,
}
if spec and getattr(spec, "supports_max_completion_tokens", False):
kwargs["max_completion_tokens"] = max(1, max_tokens)
else:
kwargs["max_tokens"] = max(1, max_tokens)
if spec:
model_lower = model_name.lower()
for pattern, overrides in spec.model_overrides:
+10
View File
@@ -49,6 +49,7 @@ class ProviderSpec:
# gateway behavior
strip_model_prefix: bool = False # strip "provider/" before sending to gateway
supports_max_completion_tokens: bool = False
# per-model param overrides, e.g. (("kimi-k2.5", {"temperature": 1.0}),)
model_overrides: tuple[tuple[str, dict[str, Any]], ...] = ()
@@ -286,6 +287,15 @@ PROVIDERS: tuple[ProviderSpec, ...] = (
backend="openai_compat",
default_api_base="https://api.mistral.ai/v1",
),
# Step Fun (阶跃星辰): OpenAI-compatible API
ProviderSpec(
name="stepfun",
keywords=("stepfun", "step"),
env_key="STEPFUN_API_KEY",
display_name="Step Fun",
backend="openai_compat",
default_api_base="https://api.stepfun.com/v1",
),
# === Local deployment (matched by config key, NOT by api_base) =========
# vLLM / any OpenAI-compatible local server
ProviderSpec(
+18 -5
View File
@@ -55,11 +55,24 @@ def timestamp() -> str:
return datetime.now().isoformat()
def current_time_str() -> str:
"""Human-readable current time with weekday and timezone, e.g. '2026-03-15 22:30 (Saturday) (CST)'."""
now = datetime.now().strftime("%Y-%m-%d %H:%M (%A)")
tz = time.strftime("%Z") or "UTC"
return f"{now} ({tz})"
def current_time_str(timezone: str | None = None) -> str:
"""Human-readable current time with weekday and UTC offset.
When *timezone* is a valid IANA name (e.g. ``"Asia/Shanghai"``), the time
is converted to that zone. Otherwise falls back to the host local time.
"""
from zoneinfo import ZoneInfo
try:
tz = ZoneInfo(timezone) if timezone else None
except (KeyError, Exception):
tz = None
now = datetime.now(tz=tz) if tz else datetime.now().astimezone()
offset = now.strftime("%z")
offset_fmt = f"{offset[:3]}:{offset[3:]}" if len(offset) == 5 else offset
tz_name = timezone or (time.strftime("%Z") or "UTC")
return f"{now.strftime('%Y-%m-%d %H:%M (%A)')} ({tz_name}, UTC{offset_fmt})"
_UNSAFE_CHARS = re.compile(r'[<>:"/\\|?*]')
+14 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "nanobot-ai"
version = "0.1.4.post5"
version = "0.1.4.post6"
description = "A lightweight personal AI assistant framework"
readme = { file = "README.md", content-type = "text/markdown" }
requires-python = ">=3.11"
@@ -120,3 +120,16 @@ ignore = ["E501"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
[tool.coverage.run]
source = ["nanobot"]
omit = ["tests/*", "**/tests/*"]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"def __repr__",
"raise NotImplementedError",
"if __name__ == .__main__.:",
"if TYPE_CHECKING:",
]
+27
View File
@@ -0,0 +1,27 @@
from pathlib import Path
from unittest.mock import MagicMock
from nanobot.agent.loop import AgentLoop
from nanobot.agent.tools.cron import CronTool
from nanobot.bus.queue import MessageBus
from nanobot.cron.service import CronService
def test_agent_loop_registers_cron_tool_with_configured_timezone(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",
cron_service=CronService(tmp_path / "cron" / "jobs.json"),
timezone="Asia/Shanghai",
)
cron_tool = loop.tools.get("cron")
assert isinstance(cron_tool, CronTool)
assert cron_tool._default_timezone == "Asia/Shanghai"
+335
View File
@@ -0,0 +1,335 @@
"""Tests for the shared agent runner and its integration contracts."""
from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from nanobot.providers.base import LLMResponse, ToolCallRequest
def _make_loop(tmp_path):
from nanobot.agent.loop import AgentLoop
from nanobot.bus.queue import MessageBus
bus = MessageBus()
provider = MagicMock()
provider.get_default_model.return_value = "test-model"
with patch("nanobot.agent.loop.ContextBuilder"), \
patch("nanobot.agent.loop.SessionManager"), \
patch("nanobot.agent.loop.SubagentManager") as MockSubMgr:
MockSubMgr.return_value.cancel_by_session = AsyncMock(return_value=0)
loop = AgentLoop(bus=bus, provider=provider, workspace=tmp_path)
return loop
@pytest.mark.asyncio
async def test_runner_preserves_reasoning_fields_and_tool_results():
from nanobot.agent.runner import AgentRunSpec, AgentRunner
provider = MagicMock()
captured_second_call: list[dict] = []
call_count = {"n": 0}
async def chat_with_retry(*, messages, **kwargs):
call_count["n"] += 1
if call_count["n"] == 1:
return LLMResponse(
content="thinking",
tool_calls=[ToolCallRequest(id="call_1", name="list_dir", arguments={"path": "."})],
reasoning_content="hidden reasoning",
thinking_blocks=[{"type": "thinking", "thinking": "step"}],
usage={"prompt_tokens": 5, "completion_tokens": 3},
)
captured_second_call[:] = messages
return LLMResponse(content="done", tool_calls=[], usage={})
provider.chat_with_retry = chat_with_retry
tools = MagicMock()
tools.get_definitions.return_value = []
tools.execute = AsyncMock(return_value="tool result")
runner = AgentRunner(provider)
result = await runner.run(AgentRunSpec(
initial_messages=[
{"role": "system", "content": "system"},
{"role": "user", "content": "do task"},
],
tools=tools,
model="test-model",
max_iterations=3,
))
assert result.final_content == "done"
assert result.tools_used == ["list_dir"]
assert result.tool_events == [
{"name": "list_dir", "status": "ok", "detail": "tool result"}
]
assistant_messages = [
msg for msg in captured_second_call
if msg.get("role") == "assistant" and msg.get("tool_calls")
]
assert len(assistant_messages) == 1
assert assistant_messages[0]["reasoning_content"] == "hidden reasoning"
assert assistant_messages[0]["thinking_blocks"] == [{"type": "thinking", "thinking": "step"}]
assert any(
msg.get("role") == "tool" and msg.get("content") == "tool result"
for msg in captured_second_call
)
@pytest.mark.asyncio
async def test_runner_calls_hooks_in_order():
from nanobot.agent.hook import AgentHook, AgentHookContext
from nanobot.agent.runner import AgentRunSpec, AgentRunner
provider = MagicMock()
call_count = {"n": 0}
events: list[tuple] = []
async def chat_with_retry(**kwargs):
call_count["n"] += 1
if call_count["n"] == 1:
return LLMResponse(
content="thinking",
tool_calls=[ToolCallRequest(id="call_1", name="list_dir", arguments={"path": "."})],
)
return LLMResponse(content="done", tool_calls=[], usage={})
provider.chat_with_retry = chat_with_retry
tools = MagicMock()
tools.get_definitions.return_value = []
tools.execute = AsyncMock(return_value="tool result")
class RecordingHook(AgentHook):
async def before_iteration(self, context: AgentHookContext) -> None:
events.append(("before_iteration", context.iteration))
async def before_execute_tools(self, context: AgentHookContext) -> None:
events.append((
"before_execute_tools",
context.iteration,
[tc.name for tc in context.tool_calls],
))
async def after_iteration(self, context: AgentHookContext) -> None:
events.append((
"after_iteration",
context.iteration,
context.final_content,
list(context.tool_results),
list(context.tool_events),
context.stop_reason,
))
def finalize_content(self, context: AgentHookContext, content: str | None) -> str | None:
events.append(("finalize_content", context.iteration, content))
return content.upper() if content else content
runner = AgentRunner(provider)
result = await runner.run(AgentRunSpec(
initial_messages=[],
tools=tools,
model="test-model",
max_iterations=3,
hook=RecordingHook(),
))
assert result.final_content == "DONE"
assert events == [
("before_iteration", 0),
("before_execute_tools", 0, ["list_dir"]),
(
"after_iteration",
0,
None,
["tool result"],
[{"name": "list_dir", "status": "ok", "detail": "tool result"}],
None,
),
("before_iteration", 1),
("finalize_content", 1, "done"),
("after_iteration", 1, "DONE", [], [], "completed"),
]
@pytest.mark.asyncio
async def test_runner_streaming_hook_receives_deltas_and_end_signal():
from nanobot.agent.hook import AgentHook, AgentHookContext
from nanobot.agent.runner import AgentRunSpec, AgentRunner
provider = MagicMock()
streamed: list[str] = []
endings: list[bool] = []
async def chat_stream_with_retry(*, on_content_delta, **kwargs):
await on_content_delta("he")
await on_content_delta("llo")
return LLMResponse(content="hello", tool_calls=[], usage={})
provider.chat_stream_with_retry = chat_stream_with_retry
provider.chat_with_retry = AsyncMock()
tools = MagicMock()
tools.get_definitions.return_value = []
class StreamingHook(AgentHook):
def wants_streaming(self) -> bool:
return True
async def on_stream(self, context: AgentHookContext, delta: str) -> None:
streamed.append(delta)
async def on_stream_end(self, context: AgentHookContext, *, resuming: bool) -> None:
endings.append(resuming)
runner = AgentRunner(provider)
result = await runner.run(AgentRunSpec(
initial_messages=[],
tools=tools,
model="test-model",
max_iterations=1,
hook=StreamingHook(),
))
assert result.final_content == "hello"
assert streamed == ["he", "llo"]
assert endings == [False]
provider.chat_with_retry.assert_not_awaited()
@pytest.mark.asyncio
async def test_runner_returns_max_iterations_fallback():
from nanobot.agent.runner import AgentRunSpec, AgentRunner
provider = MagicMock()
provider.chat_with_retry = AsyncMock(return_value=LLMResponse(
content="still working",
tool_calls=[ToolCallRequest(id="call_1", name="list_dir", arguments={"path": "."})],
))
tools = MagicMock()
tools.get_definitions.return_value = []
tools.execute = AsyncMock(return_value="tool result")
runner = AgentRunner(provider)
result = await runner.run(AgentRunSpec(
initial_messages=[],
tools=tools,
model="test-model",
max_iterations=2,
))
assert result.stop_reason == "max_iterations"
assert result.final_content == (
"I reached the maximum number of tool call iterations (2) "
"without completing the task. You can try breaking the task into smaller steps."
)
@pytest.mark.asyncio
async def test_runner_returns_structured_tool_error():
from nanobot.agent.runner import AgentRunSpec, AgentRunner
provider = MagicMock()
provider.chat_with_retry = AsyncMock(return_value=LLMResponse(
content="working",
tool_calls=[ToolCallRequest(id="call_1", name="list_dir", arguments={})],
))
tools = MagicMock()
tools.get_definitions.return_value = []
tools.execute = AsyncMock(side_effect=RuntimeError("boom"))
runner = AgentRunner(provider)
result = await runner.run(AgentRunSpec(
initial_messages=[],
tools=tools,
model="test-model",
max_iterations=2,
fail_on_tool_error=True,
))
assert result.stop_reason == "tool_error"
assert result.error == "Error: RuntimeError: boom"
assert result.tool_events == [
{"name": "list_dir", "status": "error", "detail": "boom"}
]
@pytest.mark.asyncio
async def test_loop_max_iterations_message_stays_stable(tmp_path):
loop = _make_loop(tmp_path)
loop.provider.chat_with_retry = AsyncMock(return_value=LLMResponse(
content="working",
tool_calls=[ToolCallRequest(id="call_1", name="list_dir", arguments={})],
))
loop.tools.get_definitions = MagicMock(return_value=[])
loop.tools.execute = AsyncMock(return_value="ok")
loop.max_iterations = 2
final_content, _, _ = await loop._run_agent_loop([])
assert final_content == (
"I reached the maximum number of tool call iterations (2) "
"without completing the task. You can try breaking the task into smaller steps."
)
@pytest.mark.asyncio
async def test_loop_stream_filter_handles_think_only_prefix_without_crashing(tmp_path):
loop = _make_loop(tmp_path)
deltas: list[str] = []
endings: list[bool] = []
async def chat_stream_with_retry(*, on_content_delta, **kwargs):
await on_content_delta("<think>hidden")
await on_content_delta("</think>Hello")
return LLMResponse(content="<think>hidden</think>Hello", tool_calls=[], usage={})
loop.provider.chat_stream_with_retry = chat_stream_with_retry
async def on_stream(delta: str) -> None:
deltas.append(delta)
async def on_stream_end(*, resuming: bool = False) -> None:
endings.append(resuming)
final_content, _, _ = await loop._run_agent_loop(
[],
on_stream=on_stream,
on_stream_end=on_stream_end,
)
assert final_content == "Hello"
assert deltas == ["Hello"]
assert endings == [False]
@pytest.mark.asyncio
async def test_subagent_max_iterations_announces_existing_fallback(tmp_path, monkeypatch):
from nanobot.agent.subagent import SubagentManager
from nanobot.bus.queue import MessageBus
bus = MessageBus()
provider = MagicMock()
provider.get_default_model.return_value = "test-model"
provider.chat_with_retry = AsyncMock(return_value=LLMResponse(
content="working",
tool_calls=[ToolCallRequest(id="call_1", name="list_dir", arguments={})],
))
mgr = SubagentManager(provider=provider, workspace=tmp_path, bus=bus)
mgr._announce_result = AsyncMock()
async def fake_execute(self, name, arguments):
return "tool result"
monkeypatch.setattr("nanobot.agent.tools.registry.ToolRegistry.execute", fake_execute)
await mgr._run_subagent("sub-1", "do task", "label", {"channel": "test", "chat_id": "c1"})
mgr._announce_result.assert_awaited_once()
args = mgr._announce_result.await_args.args
assert args[3] == "Task completed but no final response was generated."
assert args[5] == "ok"
+80
View File
@@ -221,3 +221,83 @@ class TestSubagentCancellation:
assert len(assistant_messages) == 1
assert assistant_messages[0]["reasoning_content"] == "hidden reasoning"
assert assistant_messages[0]["thinking_blocks"] == [{"type": "thinking", "thinking": "step"}]
@pytest.mark.asyncio
async def test_subagent_announces_error_when_tool_execution_fails(self, monkeypatch, tmp_path):
from nanobot.agent.subagent import SubagentManager
from nanobot.bus.queue import MessageBus
from nanobot.providers.base import LLMResponse, ToolCallRequest
bus = MessageBus()
provider = MagicMock()
provider.get_default_model.return_value = "test-model"
provider.chat_with_retry = AsyncMock(return_value=LLMResponse(
content="thinking",
tool_calls=[ToolCallRequest(id="call_1", name="list_dir", arguments={})],
))
mgr = SubagentManager(provider=provider, workspace=tmp_path, bus=bus)
mgr._announce_result = AsyncMock()
calls = {"n": 0}
async def fake_execute(self, name, arguments):
calls["n"] += 1
if calls["n"] == 1:
return "first result"
raise RuntimeError("boom")
monkeypatch.setattr("nanobot.agent.tools.registry.ToolRegistry.execute", fake_execute)
await mgr._run_subagent("sub-1", "do task", "label", {"channel": "test", "chat_id": "c1"})
mgr._announce_result.assert_awaited_once()
args = mgr._announce_result.await_args.args
assert "Completed steps:" in args[3]
assert "- list_dir: first result" in args[3]
assert "Failure:" in args[3]
assert "- list_dir: boom" in args[3]
assert args[5] == "error"
@pytest.mark.asyncio
async def test_cancel_by_session_cancels_running_subagent_tool(self, monkeypatch, tmp_path):
from nanobot.agent.subagent import SubagentManager
from nanobot.bus.queue import MessageBus
from nanobot.providers.base import LLMResponse, ToolCallRequest
bus = MessageBus()
provider = MagicMock()
provider.get_default_model.return_value = "test-model"
provider.chat_with_retry = AsyncMock(return_value=LLMResponse(
content="thinking",
tool_calls=[ToolCallRequest(id="call_1", name="list_dir", arguments={})],
))
mgr = SubagentManager(provider=provider, workspace=tmp_path, bus=bus)
mgr._announce_result = AsyncMock()
started = asyncio.Event()
cancelled = asyncio.Event()
async def fake_execute(self, name, arguments):
started.set()
try:
await asyncio.sleep(60)
except asyncio.CancelledError:
cancelled.set()
raise
monkeypatch.setattr("nanobot.agent.tools.registry.ToolRegistry.execute", fake_execute)
task = asyncio.create_task(
mgr._run_subagent("sub-1", "do task", "label", {"channel": "test", "chat_id": "c1"})
)
mgr._running_tasks["sub-1"] = task
mgr._session_tasks["test:c1"] = {"sub-1"}
await started.wait()
count = await mgr.cancel_by_session("test:c1")
assert count == 1
assert cancelled.is_set()
assert task.cancelled()
mgr._announce_result.assert_not_awaited()
@@ -0,0 +1,298 @@
"""Tests for ChannelManager delta coalescing to reduce streaming latency."""
import asyncio
from unittest.mock import AsyncMock, MagicMock
import pytest
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.channels.manager import ChannelManager
from nanobot.config.schema import Config
class MockChannel(BaseChannel):
"""Mock channel for testing."""
name = "mock"
display_name = "Mock"
def __init__(self, config, bus):
super().__init__(config, bus)
self._send_delta_mock = AsyncMock()
self._send_mock = AsyncMock()
async def start(self):
pass
async def stop(self):
pass
async def send(self, msg):
"""Implement abstract method."""
return await self._send_mock(msg)
async def send_delta(self, chat_id, delta, metadata=None):
"""Override send_delta for testing."""
return await self._send_delta_mock(chat_id, delta, metadata)
@pytest.fixture
def config():
"""Create a minimal config for testing."""
return Config()
@pytest.fixture
def bus():
"""Create a message bus for testing."""
return MessageBus()
@pytest.fixture
def manager(config, bus):
"""Create a channel manager with a mock channel."""
manager = ChannelManager(config, bus)
manager.channels["mock"] = MockChannel({}, bus)
return manager
class TestDeltaCoalescing:
"""Tests for _stream_delta message coalescing."""
@pytest.mark.asyncio
async def test_single_delta_not_coalesced(self, manager, bus):
"""A single delta should be sent as-is."""
msg = OutboundMessage(
channel="mock",
chat_id="chat1",
content="Hello",
metadata={"_stream_delta": True},
)
await bus.publish_outbound(msg)
# Process one message
async def process_one():
try:
m = await asyncio.wait_for(bus.consume_outbound(), timeout=0.1)
if m.metadata.get("_stream_delta"):
m, pending = manager._coalesce_stream_deltas(m)
# Put pending back (none expected)
for p in pending:
await bus.publish_outbound(p)
channel = manager.channels.get(m.channel)
if channel:
await channel.send_delta(m.chat_id, m.content, m.metadata)
except asyncio.TimeoutError:
pass
await process_one()
manager.channels["mock"]._send_delta_mock.assert_called_once_with(
"chat1", "Hello", {"_stream_delta": True}
)
@pytest.mark.asyncio
async def test_multiple_deltas_coalesced(self, manager, bus):
"""Multiple consecutive deltas for same chat should be merged."""
# Put multiple deltas in queue
for text in ["Hello", " ", "world", "!"]:
await bus.publish_outbound(OutboundMessage(
channel="mock",
chat_id="chat1",
content=text,
metadata={"_stream_delta": True},
))
# Process using coalescing logic
first_msg = await bus.consume_outbound()
merged, pending = manager._coalesce_stream_deltas(first_msg)
# Should have merged all deltas
assert merged.content == "Hello world!"
assert merged.metadata.get("_stream_delta") is True
# No pending messages (all were coalesced)
assert len(pending) == 0
@pytest.mark.asyncio
async def test_deltas_different_chats_not_coalesced(self, manager, bus):
"""Deltas for different chats should not be merged."""
# Put deltas for different chats
await bus.publish_outbound(OutboundMessage(
channel="mock",
chat_id="chat1",
content="Hello",
metadata={"_stream_delta": True},
))
await bus.publish_outbound(OutboundMessage(
channel="mock",
chat_id="chat2",
content="World",
metadata={"_stream_delta": True},
))
first_msg = await bus.consume_outbound()
merged, pending = manager._coalesce_stream_deltas(first_msg)
# First chat should not include second chat's content
assert merged.content == "Hello"
assert merged.chat_id == "chat1"
# Second chat should be in pending
assert len(pending) == 1
assert pending[0].chat_id == "chat2"
assert pending[0].content == "World"
@pytest.mark.asyncio
async def test_stream_end_terminates_coalescing(self, manager, bus):
"""_stream_end should stop coalescing and be included in final message."""
# Put deltas with stream_end at the end
await bus.publish_outbound(OutboundMessage(
channel="mock",
chat_id="chat1",
content="Hello",
metadata={"_stream_delta": True},
))
await bus.publish_outbound(OutboundMessage(
channel="mock",
chat_id="chat1",
content=" world",
metadata={"_stream_delta": True, "_stream_end": True},
))
first_msg = await bus.consume_outbound()
merged, pending = manager._coalesce_stream_deltas(first_msg)
# Should have merged content
assert merged.content == "Hello world"
# Should have stream_end flag
assert merged.metadata.get("_stream_end") is True
# No pending
assert len(pending) == 0
@pytest.mark.asyncio
async def test_coalescing_stops_at_first_non_matching_boundary(self, manager, bus):
"""Only consecutive deltas should be merged; later deltas stay queued."""
await bus.publish_outbound(OutboundMessage(
channel="mock",
chat_id="chat1",
content="Hello",
metadata={"_stream_delta": True, "_stream_id": "seg-1"},
))
await bus.publish_outbound(OutboundMessage(
channel="mock",
chat_id="chat1",
content="",
metadata={"_stream_end": True, "_stream_id": "seg-1"},
))
await bus.publish_outbound(OutboundMessage(
channel="mock",
chat_id="chat1",
content="world",
metadata={"_stream_delta": True, "_stream_id": "seg-2"},
))
first_msg = await bus.consume_outbound()
merged, pending = manager._coalesce_stream_deltas(first_msg)
assert merged.content == "Hello"
assert merged.metadata.get("_stream_end") is None
assert len(pending) == 1
assert pending[0].metadata.get("_stream_end") is True
assert pending[0].metadata.get("_stream_id") == "seg-1"
# The next stream segment must remain in queue order for later dispatch.
remaining = await bus.consume_outbound()
assert remaining.content == "world"
assert remaining.metadata.get("_stream_id") == "seg-2"
@pytest.mark.asyncio
async def test_non_delta_message_preserved(self, manager, bus):
"""Non-delta messages should be preserved in pending list."""
await bus.publish_outbound(OutboundMessage(
channel="mock",
chat_id="chat1",
content="Delta",
metadata={"_stream_delta": True},
))
await bus.publish_outbound(OutboundMessage(
channel="mock",
chat_id="chat1",
content="Final message",
metadata={}, # Not a delta
))
first_msg = await bus.consume_outbound()
merged, pending = manager._coalesce_stream_deltas(first_msg)
assert merged.content == "Delta"
assert len(pending) == 1
assert pending[0].content == "Final message"
assert pending[0].metadata.get("_stream_delta") is None
@pytest.mark.asyncio
async def test_empty_queue_stops_coalescing(self, manager, bus):
"""Coalescing should stop when queue is empty."""
await bus.publish_outbound(OutboundMessage(
channel="mock",
chat_id="chat1",
content="Only message",
metadata={"_stream_delta": True},
))
first_msg = await bus.consume_outbound()
merged, pending = manager._coalesce_stream_deltas(first_msg)
assert merged.content == "Only message"
assert len(pending) == 0
class TestDispatchOutboundWithCoalescing:
"""Tests for the full _dispatch_outbound flow with coalescing."""
@pytest.mark.asyncio
async def test_dispatch_coalesces_and_processes_pending(self, manager, bus):
"""_dispatch_outbound should coalesce deltas and process pending messages."""
# Put multiple deltas followed by a regular message
await bus.publish_outbound(OutboundMessage(
channel="mock",
chat_id="chat1",
content="A",
metadata={"_stream_delta": True},
))
await bus.publish_outbound(OutboundMessage(
channel="mock",
chat_id="chat1",
content="B",
metadata={"_stream_delta": True},
))
await bus.publish_outbound(OutboundMessage(
channel="mock",
chat_id="chat1",
content="Final",
metadata={}, # Regular message
))
# Run one iteration of dispatch logic manually
pending = []
processed = []
# First iteration: should coalesce A+B
if pending:
msg = pending.pop(0)
else:
msg = await bus.consume_outbound()
if msg.metadata.get("_stream_delta") and not msg.metadata.get("_stream_end"):
msg, extra_pending = manager._coalesce_stream_deltas(msg)
pending.extend(extra_pending)
channel = manager.channels.get(msg.channel)
if channel:
await channel.send_delta(msg.chat_id, msg.content, msg.metadata)
processed.append(("delta", msg.content))
# Should have sent coalesced delta
assert processed == [("delta", "AB")]
# Should have pending regular message
assert len(pending) == 1
assert pending[0].content == "Final"
+617 -1
View File
@@ -2,8 +2,9 @@
from __future__ import annotations
import asyncio
from types import SimpleNamespace
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
import pytest
@@ -262,3 +263,618 @@ def test_builtin_channel_init_from_dict():
ch = TelegramChannel({"enabled": False, "token": "test-tok", "allowFrom": ["*"]}, bus)
assert ch.config.token == "test-tok"
assert ch.config.allow_from == ["*"]
def test_channels_config_send_max_retries_default():
"""ChannelsConfig should have send_max_retries with default value of 3."""
cfg = ChannelsConfig()
assert hasattr(cfg, 'send_max_retries')
assert cfg.send_max_retries == 3
def test_channels_config_send_max_retries_upper_bound():
"""send_max_retries should be bounded to prevent resource exhaustion."""
from pydantic import ValidationError
# Value too high should be rejected
with pytest.raises(ValidationError):
ChannelsConfig(send_max_retries=100)
# Negative should be rejected
with pytest.raises(ValidationError):
ChannelsConfig(send_max_retries=-1)
# Boundary values should be allowed
cfg_min = ChannelsConfig(send_max_retries=0)
assert cfg_min.send_max_retries == 0
cfg_max = ChannelsConfig(send_max_retries=10)
assert cfg_max.send_max_retries == 10
# Value above upper bound should be rejected
with pytest.raises(ValidationError):
ChannelsConfig(send_max_retries=11)
# ---------------------------------------------------------------------------
# _send_with_retry
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_send_with_retry_succeeds_first_try():
"""_send_with_retry should succeed on first try and not retry."""
call_count = 0
class _FailingChannel(BaseChannel):
name = "failing"
display_name = "Failing"
async def start(self) -> None:
pass
async def stop(self) -> None:
pass
async def send(self, msg: OutboundMessage) -> None:
nonlocal call_count
call_count += 1
# Succeeds on first try
fake_config = SimpleNamespace(
channels=ChannelsConfig(send_max_retries=3),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {"failing": _FailingChannel(fake_config, mgr.bus)}
mgr._dispatch_task = None
msg = OutboundMessage(channel="failing", chat_id="123", content="test")
await mgr._send_with_retry(mgr.channels["failing"], msg)
assert call_count == 1
@pytest.mark.asyncio
async def test_send_with_retry_retries_on_failure():
"""_send_with_retry should retry on failure up to max_retries times."""
call_count = 0
class _FailingChannel(BaseChannel):
name = "failing"
display_name = "Failing"
async def start(self) -> None:
pass
async def stop(self) -> None:
pass
async def send(self, msg: OutboundMessage) -> None:
nonlocal call_count
call_count += 1
raise RuntimeError("simulated failure")
fake_config = SimpleNamespace(
channels=ChannelsConfig(send_max_retries=3),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {"failing": _FailingChannel(fake_config, mgr.bus)}
mgr._dispatch_task = None
msg = OutboundMessage(channel="failing", chat_id="123", content="test")
# Patch asyncio.sleep to avoid actual delays
with patch("nanobot.channels.manager.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
await mgr._send_with_retry(mgr.channels["failing"], msg)
assert call_count == 3 # 3 total attempts (initial + 2 retries)
assert mock_sleep.call_count == 2 # 2 sleeps between retries
@pytest.mark.asyncio
async def test_send_with_retry_no_retry_when_max_is_zero():
"""_send_with_retry should not retry when send_max_retries is 0."""
call_count = 0
class _FailingChannel(BaseChannel):
name = "failing"
display_name = "Failing"
async def start(self) -> None:
pass
async def stop(self) -> None:
pass
async def send(self, msg: OutboundMessage) -> None:
nonlocal call_count
call_count += 1
raise RuntimeError("simulated failure")
fake_config = SimpleNamespace(
channels=ChannelsConfig(send_max_retries=0),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {"failing": _FailingChannel(fake_config, mgr.bus)}
mgr._dispatch_task = None
msg = OutboundMessage(channel="failing", chat_id="123", content="test")
with patch("nanobot.channels.manager.asyncio.sleep", new_callable=AsyncMock):
await mgr._send_with_retry(mgr.channels["failing"], msg)
assert call_count == 1 # Called once but no retry (max(0, 1) = 1)
@pytest.mark.asyncio
async def test_send_with_retry_calls_send_delta():
"""_send_with_retry should call send_delta when metadata has _stream_delta."""
send_delta_called = False
class _StreamingChannel(BaseChannel):
name = "streaming"
display_name = "Streaming"
async def start(self) -> None:
pass
async def stop(self) -> None:
pass
async def send(self, msg: OutboundMessage) -> None:
pass # Should not be called
async def send_delta(self, chat_id: str, delta: str, metadata: dict | None = None) -> None:
nonlocal send_delta_called
send_delta_called = True
fake_config = SimpleNamespace(
channels=ChannelsConfig(send_max_retries=3),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {"streaming": _StreamingChannel(fake_config, mgr.bus)}
mgr._dispatch_task = None
msg = OutboundMessage(
channel="streaming", chat_id="123", content="test delta",
metadata={"_stream_delta": True}
)
await mgr._send_with_retry(mgr.channels["streaming"], msg)
assert send_delta_called is True
@pytest.mark.asyncio
async def test_send_with_retry_skips_send_when_streamed():
"""_send_with_retry should not call send when metadata has _streamed flag."""
send_called = False
send_delta_called = False
class _StreamedChannel(BaseChannel):
name = "streamed"
display_name = "Streamed"
async def start(self) -> None:
pass
async def stop(self) -> None:
pass
async def send(self, msg: OutboundMessage) -> None:
nonlocal send_called
send_called = True
async def send_delta(self, chat_id: str, delta: str, metadata: dict | None = None) -> None:
nonlocal send_delta_called
send_delta_called = True
fake_config = SimpleNamespace(
channels=ChannelsConfig(send_max_retries=3),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {"streamed": _StreamedChannel(fake_config, mgr.bus)}
mgr._dispatch_task = None
# _streamed means message was already sent via send_delta, so skip send
msg = OutboundMessage(
channel="streamed", chat_id="123", content="test",
metadata={"_streamed": True}
)
await mgr._send_with_retry(mgr.channels["streamed"], msg)
assert send_called is False
assert send_delta_called is False
@pytest.mark.asyncio
async def test_send_with_retry_propagates_cancelled_error():
"""_send_with_retry should re-raise CancelledError for graceful shutdown."""
class _CancellingChannel(BaseChannel):
name = "cancelling"
display_name = "Cancelling"
async def start(self) -> None:
pass
async def stop(self) -> None:
pass
async def send(self, msg: OutboundMessage) -> None:
raise asyncio.CancelledError("simulated cancellation")
fake_config = SimpleNamespace(
channels=ChannelsConfig(send_max_retries=3),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {"cancelling": _CancellingChannel(fake_config, mgr.bus)}
mgr._dispatch_task = None
msg = OutboundMessage(channel="cancelling", chat_id="123", content="test")
with pytest.raises(asyncio.CancelledError):
await mgr._send_with_retry(mgr.channels["cancelling"], msg)
@pytest.mark.asyncio
async def test_send_with_retry_propagates_cancelled_error_during_sleep():
"""_send_with_retry should re-raise CancelledError during sleep."""
call_count = 0
class _FailingChannel(BaseChannel):
name = "failing"
display_name = "Failing"
async def start(self) -> None:
pass
async def stop(self) -> None:
pass
async def send(self, msg: OutboundMessage) -> None:
nonlocal call_count
call_count += 1
raise RuntimeError("simulated failure")
fake_config = SimpleNamespace(
channels=ChannelsConfig(send_max_retries=3),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {"failing": _FailingChannel(fake_config, mgr.bus)}
mgr._dispatch_task = None
msg = OutboundMessage(channel="failing", chat_id="123", content="test")
# Mock sleep to raise CancelledError
async def cancel_during_sleep(_):
raise asyncio.CancelledError("cancelled during sleep")
with patch("nanobot.channels.manager.asyncio.sleep", side_effect=cancel_during_sleep):
with pytest.raises(asyncio.CancelledError):
await mgr._send_with_retry(mgr.channels["failing"], msg)
# Should have attempted once before sleep was cancelled
assert call_count == 1
# ---------------------------------------------------------------------------
# ChannelManager - lifecycle and getters
# ---------------------------------------------------------------------------
class _ChannelWithAllowFrom(BaseChannel):
"""Channel with configurable allow_from."""
name = "withallow"
display_name = "With Allow"
def __init__(self, config, bus, allow_from):
super().__init__(config, bus)
self.config.allow_from = allow_from
async def start(self) -> None:
pass
async def stop(self) -> None:
pass
async def send(self, msg: OutboundMessage) -> None:
pass
class _StartableChannel(BaseChannel):
"""Channel that tracks start/stop calls."""
name = "startable"
display_name = "Startable"
def __init__(self, config, bus):
super().__init__(config, bus)
self.started = False
self.stopped = False
async def start(self) -> None:
self.started = True
async def stop(self) -> None:
self.stopped = True
async def send(self, msg: OutboundMessage) -> None:
pass
@pytest.mark.asyncio
async def test_validate_allow_from_raises_on_empty_list():
"""_validate_allow_from should raise SystemExit when allow_from is empty list."""
fake_config = SimpleNamespace(
channels=ChannelsConfig(),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.channels = {"test": _ChannelWithAllowFrom(fake_config, None, [])}
mgr._dispatch_task = None
with pytest.raises(SystemExit) as exc_info:
mgr._validate_allow_from()
assert "empty allowFrom" in str(exc_info.value)
@pytest.mark.asyncio
async def test_validate_allow_from_passes_with_asterisk():
"""_validate_allow_from should not raise when allow_from contains '*'."""
fake_config = SimpleNamespace(
channels=ChannelsConfig(),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.channels = {"test": _ChannelWithAllowFrom(fake_config, None, ["*"])}
mgr._dispatch_task = None
# Should not raise
mgr._validate_allow_from()
@pytest.mark.asyncio
async def test_get_channel_returns_channel_if_exists():
"""get_channel should return the channel if it exists."""
fake_config = SimpleNamespace(
channels=ChannelsConfig(),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {"telegram": _StartableChannel(fake_config, mgr.bus)}
mgr._dispatch_task = None
assert mgr.get_channel("telegram") is not None
assert mgr.get_channel("nonexistent") is None
@pytest.mark.asyncio
async def test_get_status_returns_running_state():
"""get_status should return enabled and running state for each channel."""
fake_config = SimpleNamespace(
channels=ChannelsConfig(),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
ch = _StartableChannel(fake_config, mgr.bus)
mgr.channels = {"startable": ch}
mgr._dispatch_task = None
status = mgr.get_status()
assert status["startable"]["enabled"] is True
assert status["startable"]["running"] is False # Not started yet
@pytest.mark.asyncio
async def test_enabled_channels_returns_channel_names():
"""enabled_channels should return list of enabled channel names."""
fake_config = SimpleNamespace(
channels=ChannelsConfig(),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {
"telegram": _StartableChannel(fake_config, mgr.bus),
"slack": _StartableChannel(fake_config, mgr.bus),
}
mgr._dispatch_task = None
enabled = mgr.enabled_channels
assert "telegram" in enabled
assert "slack" in enabled
assert len(enabled) == 2
@pytest.mark.asyncio
async def test_stop_all_cancels_dispatcher_and_stops_channels():
"""stop_all should cancel the dispatch task and stop all channels."""
fake_config = SimpleNamespace(
channels=ChannelsConfig(),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
ch = _StartableChannel(fake_config, mgr.bus)
mgr.channels = {"startable": ch}
# Create a real cancelled task
async def dummy_task():
while True:
await asyncio.sleep(1)
dispatch_task = asyncio.create_task(dummy_task())
mgr._dispatch_task = dispatch_task
await mgr.stop_all()
# Task should be cancelled
assert dispatch_task.cancelled()
# Channel should be stopped
assert ch.stopped is True
@pytest.mark.asyncio
async def test_start_channel_logs_error_on_failure():
"""_start_channel should log error when channel start fails."""
class _FailingChannel(BaseChannel):
name = "failing"
display_name = "Failing"
async def start(self) -> None:
raise RuntimeError("connection failed")
async def stop(self) -> None:
pass
async def send(self, msg: OutboundMessage) -> None:
pass
fake_config = SimpleNamespace(
channels=ChannelsConfig(),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {}
mgr._dispatch_task = None
ch = _FailingChannel(fake_config, mgr.bus)
# Should not raise, just log error
await mgr._start_channel("failing", ch)
@pytest.mark.asyncio
async def test_stop_all_handles_channel_exception():
"""stop_all should handle exceptions when stopping channels gracefully."""
class _StopFailingChannel(BaseChannel):
name = "stopfailing"
display_name = "Stop Failing"
async def start(self) -> None:
pass
async def stop(self) -> None:
raise RuntimeError("stop failed")
async def send(self, msg: OutboundMessage) -> None:
pass
fake_config = SimpleNamespace(
channels=ChannelsConfig(),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {"stopfailing": _StopFailingChannel(fake_config, mgr.bus)}
mgr._dispatch_task = None
# Should not raise even if channel.stop() raises
await mgr.stop_all()
@pytest.mark.asyncio
async def test_start_all_no_channels_logs_warning():
"""start_all should log warning when no channels are enabled."""
fake_config = SimpleNamespace(
channels=ChannelsConfig(),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
mgr.channels = {} # No channels
mgr._dispatch_task = None
# Should return early without creating dispatch task
await mgr.start_all()
assert mgr._dispatch_task is None
@pytest.mark.asyncio
async def test_start_all_creates_dispatch_task():
"""start_all should create the dispatch task when channels exist."""
fake_config = SimpleNamespace(
channels=ChannelsConfig(),
providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
)
mgr = ChannelManager.__new__(ChannelManager)
mgr.config = fake_config
mgr.bus = MessageBus()
ch = _StartableChannel(fake_config, mgr.bus)
mgr.channels = {"startable": ch}
mgr._dispatch_task = None
# Cancel immediately after start to avoid running forever
async def cancel_after_start():
await asyncio.sleep(0.01)
if mgr._dispatch_task:
mgr._dispatch_task.cancel()
cancel_task = asyncio.create_task(cancel_after_start())
try:
await mgr.start_all()
except asyncio.CancelledError:
pass
finally:
cancel_task.cancel()
try:
await cancel_task
except asyncio.CancelledError:
pass
# Dispatch task should have been created
assert mgr._dispatch_task is not None
+171 -2
View File
@@ -10,8 +10,8 @@ from nanobot.channels.email import EmailChannel
from nanobot.channels.email import EmailConfig
def _make_config() -> EmailConfig:
return EmailConfig(
def _make_config(**overrides) -> EmailConfig:
defaults = dict(
enabled=True,
consent_granted=True,
imap_host="imap.example.com",
@@ -23,19 +23,27 @@ def _make_config() -> EmailConfig:
smtp_username="bot@example.com",
smtp_password="secret",
mark_seen=True,
# Disable auth verification by default so existing tests are unaffected
verify_dkim=False,
verify_spf=False,
)
defaults.update(overrides)
return EmailConfig(**defaults)
def _make_raw_email(
from_addr: str = "alice@example.com",
subject: str = "Hello",
body: str = "This is the body.",
auth_results: str | None = None,
) -> bytes:
msg = EmailMessage()
msg["From"] = from_addr
msg["To"] = "bot@example.com"
msg["Subject"] = subject
msg["Message-ID"] = "<m1@example.com>"
if auth_results:
msg["Authentication-Results"] = auth_results
msg.set_content(body)
return msg.as_bytes()
@@ -481,3 +489,164 @@ def test_fetch_messages_between_dates_uses_imap_since_before_without_mark_seen(m
assert fake.search_args is not None
assert fake.search_args[1:] == ("SINCE", "06-Feb-2026", "BEFORE", "07-Feb-2026")
assert fake.store_calls == []
# ---------------------------------------------------------------------------
# Security: Anti-spoofing tests for Authentication-Results verification
# ---------------------------------------------------------------------------
def _make_fake_imap(raw: bytes):
"""Return a FakeIMAP class pre-loaded with the given raw email."""
class FakeIMAP:
def __init__(self) -> None:
self.store_calls: list[tuple[bytes, str, str]] = []
def login(self, _user: str, _pw: str):
return "OK", [b"logged in"]
def select(self, _mailbox: str):
return "OK", [b"1"]
def search(self, *_args):
return "OK", [b"1"]
def fetch(self, _imap_id: bytes, _parts: str):
return "OK", [(b"1 (UID 500 BODY[] {200})", raw), b")"]
def store(self, imap_id: bytes, op: str, flags: str):
self.store_calls.append((imap_id, op, flags))
return "OK", [b""]
def logout(self):
return "BYE", [b""]
return FakeIMAP()
def test_spoofed_email_rejected_when_verify_enabled(monkeypatch) -> None:
"""An email without Authentication-Results should be rejected when verify_dkim=True."""
raw = _make_raw_email(subject="Spoofed", body="Malicious payload")
fake = _make_fake_imap(raw)
monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
cfg = _make_config(verify_dkim=True, verify_spf=True)
channel = EmailChannel(cfg, MessageBus())
items = channel._fetch_new_messages()
assert len(items) == 0, "Spoofed email without auth headers should be rejected"
def test_email_with_valid_auth_results_accepted(monkeypatch) -> None:
"""An email with spf=pass and dkim=pass should be accepted."""
raw = _make_raw_email(
subject="Legit",
body="Hello from verified sender",
auth_results="mx.example.com; spf=pass smtp.mailfrom=alice@example.com; dkim=pass header.d=example.com",
)
fake = _make_fake_imap(raw)
monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
cfg = _make_config(verify_dkim=True, verify_spf=True)
channel = EmailChannel(cfg, MessageBus())
items = channel._fetch_new_messages()
assert len(items) == 1
assert items[0]["sender"] == "alice@example.com"
assert items[0]["subject"] == "Legit"
def test_email_with_partial_auth_rejected(monkeypatch) -> None:
"""An email with only spf=pass but no dkim=pass should be rejected when verify_dkim=True."""
raw = _make_raw_email(
subject="Partial",
body="Only SPF passes",
auth_results="mx.example.com; spf=pass smtp.mailfrom=alice@example.com; dkim=fail",
)
fake = _make_fake_imap(raw)
monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
cfg = _make_config(verify_dkim=True, verify_spf=True)
channel = EmailChannel(cfg, MessageBus())
items = channel._fetch_new_messages()
assert len(items) == 0, "Email with dkim=fail should be rejected"
def test_backward_compat_verify_disabled(monkeypatch) -> None:
"""When verify_dkim=False and verify_spf=False, emails without auth headers are accepted."""
raw = _make_raw_email(subject="NoAuth", body="No auth headers present")
fake = _make_fake_imap(raw)
monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
cfg = _make_config(verify_dkim=False, verify_spf=False)
channel = EmailChannel(cfg, MessageBus())
items = channel._fetch_new_messages()
assert len(items) == 1, "With verification disabled, emails should be accepted as before"
def test_email_content_tagged_with_email_context(monkeypatch) -> None:
"""Email content should be prefixed with [EMAIL-CONTEXT] for LLM isolation."""
raw = _make_raw_email(subject="Tagged", body="Check the tag")
fake = _make_fake_imap(raw)
monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
cfg = _make_config(verify_dkim=False, verify_spf=False)
channel = EmailChannel(cfg, MessageBus())
items = channel._fetch_new_messages()
assert len(items) == 1
assert items[0]["content"].startswith("[EMAIL-CONTEXT]"), (
"Email content must be tagged with [EMAIL-CONTEXT]"
)
def test_check_authentication_results_method() -> None:
"""Unit test for the _check_authentication_results static method."""
from email.parser import BytesParser
from email import policy
# No Authentication-Results header
msg_no_auth = EmailMessage()
msg_no_auth["From"] = "alice@example.com"
msg_no_auth.set_content("test")
parsed = BytesParser(policy=policy.default).parsebytes(msg_no_auth.as_bytes())
spf, dkim = EmailChannel._check_authentication_results(parsed)
assert spf is False
assert dkim is False
# Both pass
msg_both = EmailMessage()
msg_both["From"] = "alice@example.com"
msg_both["Authentication-Results"] = (
"mx.google.com; spf=pass smtp.mailfrom=example.com; dkim=pass header.d=example.com"
)
msg_both.set_content("test")
parsed = BytesParser(policy=policy.default).parsebytes(msg_both.as_bytes())
spf, dkim = EmailChannel._check_authentication_results(parsed)
assert spf is True
assert dkim is True
# SPF pass, DKIM fail
msg_spf_only = EmailMessage()
msg_spf_only["From"] = "alice@example.com"
msg_spf_only["Authentication-Results"] = (
"mx.google.com; spf=pass smtp.mailfrom=example.com; dkim=fail"
)
msg_spf_only.set_content("test")
parsed = BytesParser(policy=policy.default).parsebytes(msg_spf_only.as_bytes())
spf, dkim = EmailChannel._check_authentication_results(parsed)
assert spf is True
assert dkim is False
# DKIM pass, SPF fail
msg_dkim_only = EmailMessage()
msg_dkim_only["From"] = "alice@example.com"
msg_dkim_only["Authentication-Results"] = (
"mx.google.com; spf=fail smtp.mailfrom=example.com; dkim=pass header.d=example.com"
)
msg_dkim_only.set_content("test")
parsed = BytesParser(policy=policy.default).parsebytes(msg_dkim_only.as_bytes())
spf, dkim = EmailChannel._check_authentication_results(parsed)
assert spf is False
assert dkim is True
+258
View File
@@ -0,0 +1,258 @@
"""Tests for Feishu streaming (send_delta) via CardKit streaming API."""
import time
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
from nanobot.bus.queue import MessageBus
from nanobot.channels.feishu import FeishuChannel, FeishuConfig, _FeishuStreamBuf
def _make_channel(streaming: bool = True) -> FeishuChannel:
config = FeishuConfig(
enabled=True,
app_id="cli_test",
app_secret="secret",
allow_from=["*"],
streaming=streaming,
)
ch = FeishuChannel(config, MessageBus())
ch._client = MagicMock()
ch._loop = None
return ch
def _mock_create_card_response(card_id: str = "card_stream_001"):
resp = MagicMock()
resp.success.return_value = True
resp.data = SimpleNamespace(card_id=card_id)
return resp
def _mock_send_response(message_id: str = "om_stream_001"):
resp = MagicMock()
resp.success.return_value = True
resp.data = SimpleNamespace(message_id=message_id)
return resp
def _mock_content_response(success: bool = True):
resp = MagicMock()
resp.success.return_value = success
resp.code = 0 if success else 99999
resp.msg = "ok" if success else "error"
return resp
class TestFeishuStreamingConfig:
def test_streaming_default_true(self):
assert FeishuConfig().streaming is True
def test_supports_streaming_when_enabled(self):
ch = _make_channel(streaming=True)
assert ch.supports_streaming is True
def test_supports_streaming_disabled(self):
ch = _make_channel(streaming=False)
assert ch.supports_streaming is False
class TestCreateStreamingCard:
def test_returns_card_id_on_success(self):
ch = _make_channel()
ch._client.cardkit.v1.card.create.return_value = _mock_create_card_response("card_123")
ch._client.im.v1.message.create.return_value = _mock_send_response()
result = ch._create_streaming_card_sync("chat_id", "oc_chat1")
assert result == "card_123"
ch._client.cardkit.v1.card.create.assert_called_once()
ch._client.im.v1.message.create.assert_called_once()
def test_returns_none_on_failure(self):
ch = _make_channel()
resp = MagicMock()
resp.success.return_value = False
resp.code = 99999
resp.msg = "error"
ch._client.cardkit.v1.card.create.return_value = resp
assert ch._create_streaming_card_sync("chat_id", "oc_chat1") is None
def test_returns_none_on_exception(self):
ch = _make_channel()
ch._client.cardkit.v1.card.create.side_effect = RuntimeError("network")
assert ch._create_streaming_card_sync("chat_id", "oc_chat1") is None
def test_returns_none_when_card_send_fails(self):
ch = _make_channel()
ch._client.cardkit.v1.card.create.return_value = _mock_create_card_response("card_123")
resp = MagicMock()
resp.success.return_value = False
resp.code = 99999
resp.msg = "error"
resp.get_log_id.return_value = "log1"
ch._client.im.v1.message.create.return_value = resp
assert ch._create_streaming_card_sync("chat_id", "oc_chat1") is None
class TestCloseStreamingMode:
def test_returns_true_on_success(self):
ch = _make_channel()
ch._client.cardkit.v1.card.settings.return_value = _mock_content_response(True)
assert ch._close_streaming_mode_sync("card_1", 10) is True
def test_returns_false_on_failure(self):
ch = _make_channel()
ch._client.cardkit.v1.card.settings.return_value = _mock_content_response(False)
assert ch._close_streaming_mode_sync("card_1", 10) is False
def test_returns_false_on_exception(self):
ch = _make_channel()
ch._client.cardkit.v1.card.settings.side_effect = RuntimeError("err")
assert ch._close_streaming_mode_sync("card_1", 10) is False
class TestStreamUpdateText:
def test_returns_true_on_success(self):
ch = _make_channel()
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response(True)
assert ch._stream_update_text_sync("card_1", "hello", 1) is True
def test_returns_false_on_failure(self):
ch = _make_channel()
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response(False)
assert ch._stream_update_text_sync("card_1", "hello", 1) is False
def test_returns_false_on_exception(self):
ch = _make_channel()
ch._client.cardkit.v1.card_element.content.side_effect = RuntimeError("err")
assert ch._stream_update_text_sync("card_1", "hello", 1) is False
class TestSendDelta:
@pytest.mark.asyncio
async def test_first_delta_creates_card_and_sends(self):
ch = _make_channel()
ch._client.cardkit.v1.card.create.return_value = _mock_create_card_response("card_new")
ch._client.im.v1.message.create.return_value = _mock_send_response("om_new")
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
await ch.send_delta("oc_chat1", "Hello ")
assert "oc_chat1" in ch._stream_bufs
buf = ch._stream_bufs["oc_chat1"]
assert buf.text == "Hello "
assert buf.card_id == "card_new"
assert buf.sequence == 1
ch._client.cardkit.v1.card.create.assert_called_once()
ch._client.im.v1.message.create.assert_called_once()
ch._client.cardkit.v1.card_element.content.assert_called_once()
@pytest.mark.asyncio
async def test_second_delta_within_interval_skips_update(self):
ch = _make_channel()
buf = _FeishuStreamBuf(text="Hello ", card_id="card_1", sequence=1, last_edit=time.monotonic())
ch._stream_bufs["oc_chat1"] = buf
await ch.send_delta("oc_chat1", "world")
assert buf.text == "Hello world"
ch._client.cardkit.v1.card_element.content.assert_not_called()
@pytest.mark.asyncio
async def test_delta_after_interval_updates_text(self):
ch = _make_channel()
buf = _FeishuStreamBuf(text="Hello ", card_id="card_1", sequence=1, last_edit=time.monotonic() - 1.0)
ch._stream_bufs["oc_chat1"] = buf
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
await ch.send_delta("oc_chat1", "world")
assert buf.text == "Hello world"
assert buf.sequence == 2
ch._client.cardkit.v1.card_element.content.assert_called_once()
@pytest.mark.asyncio
async def test_stream_end_sends_final_update(self):
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Final content", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
ch._client.cardkit.v1.card.settings.return_value = _mock_content_response()
await ch.send_delta("oc_chat1", "", metadata={"_stream_end": True})
assert "oc_chat1" not in ch._stream_bufs
ch._client.cardkit.v1.card_element.content.assert_called_once()
ch._client.cardkit.v1.card.settings.assert_called_once()
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_fallback_when_no_card_id(self):
"""If card creation failed, stream_end falls back to a plain card message."""
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Fallback content", card_id=None, sequence=0, last_edit=0.0,
)
ch._client.im.v1.message.create.return_value = _mock_send_response("om_fb")
await ch.send_delta("oc_chat1", "", metadata={"_stream_end": True})
assert "oc_chat1" not in ch._stream_bufs
ch._client.cardkit.v1.card_element.content.assert_not_called()
ch._client.im.v1.message.create.assert_called_once()
@pytest.mark.asyncio
async def test_stream_end_without_buf_is_noop(self):
ch = _make_channel()
await ch.send_delta("oc_chat1", "", metadata={"_stream_end": True})
ch._client.cardkit.v1.card_element.content.assert_not_called()
@pytest.mark.asyncio
async def test_empty_delta_skips_send(self):
ch = _make_channel()
await ch.send_delta("oc_chat1", " ")
assert "oc_chat1" in ch._stream_bufs
ch._client.cardkit.v1.card.create.assert_not_called()
@pytest.mark.asyncio
async def test_no_client_returns_early(self):
ch = _make_channel()
ch._client = None
await ch.send_delta("oc_chat1", "text")
assert "oc_chat1" not in ch._stream_bufs
@pytest.mark.asyncio
async def test_sequence_increments_correctly(self):
ch = _make_channel()
buf = _FeishuStreamBuf(text="a", card_id="card_1", sequence=5, last_edit=0.0)
ch._stream_bufs["oc_chat1"] = buf
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
await ch.send_delta("oc_chat1", "b")
assert buf.sequence == 6
buf.last_edit = 0.0 # reset to bypass throttle
await ch.send_delta("oc_chat1", "c")
assert buf.sequence == 7
class TestSendMessageReturnsId:
def test_returns_message_id_on_success(self):
ch = _make_channel()
ch._client.im.v1.message.create.return_value = _mock_send_response("om_abc")
result = ch._send_message_sync("chat_id", "oc_chat1", "text", '{"text":"hi"}')
assert result == "om_abc"
def test_returns_none_on_failure(self):
ch = _make_channel()
resp = MagicMock()
resp.success.return_value = False
resp.code = 99999
resp.msg = "error"
resp.get_log_id.return_value = "log1"
ch._client.im.v1.message.create.return_value = resp
result = ch._send_message_sync("chat_id", "oc_chat1", "text", '{"text":"hi"}')
assert result is None
+123 -3
View File
@@ -13,7 +13,7 @@ except ImportError:
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.telegram import TELEGRAM_REPLY_CONTEXT_MAX_LEN, TelegramChannel
from nanobot.channels.telegram import TELEGRAM_REPLY_CONTEXT_MAX_LEN, TelegramChannel, _StreamBuf
from nanobot.channels.telegram import TelegramConfig
@@ -50,8 +50,9 @@ class _FakeBot:
async def set_my_commands(self, commands) -> None:
self.commands = commands
async def send_message(self, **kwargs) -> None:
async def send_message(self, **kwargs):
self.sent_messages.append(kwargs)
return SimpleNamespace(message_id=len(self.sent_messages))
async def send_photo(self, **kwargs) -> None:
self.sent_media.append({"kind": "photo", **kwargs})
@@ -271,13 +272,132 @@ async def test_send_text_gives_up_after_max_retries() -> None:
orig_delay = tg_mod._SEND_RETRY_BASE_DELAY
tg_mod._SEND_RETRY_BASE_DELAY = 0.01
try:
await channel._send_text(123, "hello", None, {})
with pytest.raises(TimedOut):
await channel._send_text(123, "hello", None, {})
finally:
tg_mod._SEND_RETRY_BASE_DELAY = orig_delay
assert channel._app.bot.sent_messages == []
@pytest.mark.asyncio
async def test_on_error_logs_network_issues_as_warning(monkeypatch) -> None:
from telegram.error import NetworkError
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
MessageBus(),
)
recorded: list[tuple[str, str]] = []
monkeypatch.setattr(
"nanobot.channels.telegram.logger.warning",
lambda message, error: recorded.append(("warning", message.format(error))),
)
monkeypatch.setattr(
"nanobot.channels.telegram.logger.error",
lambda message, error: recorded.append(("error", message.format(error))),
)
await channel._on_error(object(), SimpleNamespace(error=NetworkError("proxy disconnected")))
assert recorded == [("warning", "Telegram network issue: proxy disconnected")]
@pytest.mark.asyncio
async def test_on_error_keeps_non_network_exceptions_as_error(monkeypatch) -> None:
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
MessageBus(),
)
recorded: list[tuple[str, str]] = []
monkeypatch.setattr(
"nanobot.channels.telegram.logger.warning",
lambda message, error: recorded.append(("warning", message.format(error))),
)
monkeypatch.setattr(
"nanobot.channels.telegram.logger.error",
lambda message, error: recorded.append(("error", message.format(error))),
)
await channel._on_error(object(), SimpleNamespace(error=RuntimeError("boom")))
assert recorded == [("error", "Telegram error: boom")]
@pytest.mark.asyncio
async def test_send_delta_stream_end_raises_and_keeps_buffer_on_failure() -> None:
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
MessageBus(),
)
channel._app = _FakeApp(lambda: None)
channel._app.bot.edit_message_text = AsyncMock(side_effect=RuntimeError("boom"))
channel._stream_bufs["123"] = _StreamBuf(text="hello", message_id=7, last_edit=0.0)
with pytest.raises(RuntimeError, match="boom"):
await channel.send_delta("123", "", {"_stream_end": True})
assert "123" in channel._stream_bufs
@pytest.mark.asyncio
async def test_send_delta_stream_end_treats_not_modified_as_success() -> None:
from telegram.error import BadRequest
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
MessageBus(),
)
channel._app = _FakeApp(lambda: None)
channel._app.bot.edit_message_text = AsyncMock(side_effect=BadRequest("Message is not modified"))
channel._stream_bufs["123"] = _StreamBuf(text="hello", message_id=7, last_edit=0.0, stream_id="s:0")
await channel.send_delta("123", "", {"_stream_end": True, "_stream_id": "s:0"})
assert "123" not in channel._stream_bufs
@pytest.mark.asyncio
async def test_send_delta_new_stream_id_replaces_stale_buffer() -> None:
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
MessageBus(),
)
channel._app = _FakeApp(lambda: None)
channel._stream_bufs["123"] = _StreamBuf(
text="hello",
message_id=7,
last_edit=0.0,
stream_id="old:0",
)
await channel.send_delta("123", "world", {"_stream_delta": True, "_stream_id": "new:0"})
buf = channel._stream_bufs["123"]
assert buf.text == "world"
assert buf.stream_id == "new:0"
assert buf.message_id == 1
@pytest.mark.asyncio
async def test_send_delta_incremental_edit_treats_not_modified_as_success() -> None:
from telegram.error import BadRequest
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
MessageBus(),
)
channel._app = _FakeApp(lambda: None)
channel._stream_bufs["123"] = _StreamBuf(text="hello", message_id=7, last_edit=0.0, stream_id="s:0")
channel._app.bot.edit_message_text = AsyncMock(side_effect=BadRequest("Message is not modified"))
await channel.send_delta("123", "", {"_stream_delta": True, "_stream_id": "s:0"})
assert channel._stream_bufs["123"].last_edit > 0.0
def test_derive_topic_session_key_uses_thread_id() -> None:
message = SimpleNamespace(
chat=SimpleNamespace(type="supergroup"),
+80 -31
View File
@@ -1,5 +1,7 @@
"""Tests for CronTool._list_jobs() output formatting."""
from datetime import datetime, timezone
from nanobot.agent.tools.cron import CronTool
from nanobot.cron.service import CronService
from nanobot.cron.types import CronJobState, CronSchedule
@@ -10,99 +12,120 @@ def _make_tool(tmp_path) -> CronTool:
return CronTool(service)
def _make_tool_with_tz(tmp_path, tz: str) -> CronTool:
service = CronService(tmp_path / "cron" / "jobs.json")
return CronTool(service, default_timezone=tz)
# -- _format_timing tests --
def test_format_timing_cron_with_tz() -> None:
def test_format_timing_cron_with_tz(tmp_path) -> None:
tool = _make_tool(tmp_path)
s = CronSchedule(kind="cron", expr="0 9 * * 1-5", tz="America/Denver")
assert CronTool._format_timing(s) == "cron: 0 9 * * 1-5 (America/Denver)"
assert tool._format_timing(s) == "cron: 0 9 * * 1-5 (America/Denver)"
def test_format_timing_cron_without_tz() -> None:
def test_format_timing_cron_without_tz(tmp_path) -> None:
tool = _make_tool(tmp_path)
s = CronSchedule(kind="cron", expr="*/5 * * * *")
assert CronTool._format_timing(s) == "cron: */5 * * * *"
assert tool._format_timing(s) == "cron: */5 * * * *"
def test_format_timing_every_hours() -> None:
def test_format_timing_every_hours(tmp_path) -> None:
tool = _make_tool(tmp_path)
s = CronSchedule(kind="every", every_ms=7_200_000)
assert CronTool._format_timing(s) == "every 2h"
assert tool._format_timing(s) == "every 2h"
def test_format_timing_every_minutes() -> None:
def test_format_timing_every_minutes(tmp_path) -> None:
tool = _make_tool(tmp_path)
s = CronSchedule(kind="every", every_ms=1_800_000)
assert CronTool._format_timing(s) == "every 30m"
assert tool._format_timing(s) == "every 30m"
def test_format_timing_every_seconds() -> None:
def test_format_timing_every_seconds(tmp_path) -> None:
tool = _make_tool(tmp_path)
s = CronSchedule(kind="every", every_ms=30_000)
assert CronTool._format_timing(s) == "every 30s"
assert tool._format_timing(s) == "every 30s"
def test_format_timing_every_non_minute_seconds() -> None:
def test_format_timing_every_non_minute_seconds(tmp_path) -> None:
tool = _make_tool(tmp_path)
s = CronSchedule(kind="every", every_ms=90_000)
assert CronTool._format_timing(s) == "every 90s"
assert tool._format_timing(s) == "every 90s"
def test_format_timing_every_milliseconds() -> None:
def test_format_timing_every_milliseconds(tmp_path) -> None:
tool = _make_tool(tmp_path)
s = CronSchedule(kind="every", every_ms=200)
assert CronTool._format_timing(s) == "every 200ms"
assert tool._format_timing(s) == "every 200ms"
def test_format_timing_at() -> None:
def test_format_timing_at(tmp_path) -> None:
tool = _make_tool_with_tz(tmp_path, "Asia/Shanghai")
s = CronSchedule(kind="at", at_ms=1773684000000)
result = CronTool._format_timing(s)
result = tool._format_timing(s)
assert "Asia/Shanghai" in result
assert result.startswith("at 2026-")
def test_format_timing_fallback() -> None:
def test_format_timing_fallback(tmp_path) -> None:
tool = _make_tool(tmp_path)
s = CronSchedule(kind="every") # no every_ms
assert CronTool._format_timing(s) == "every"
assert tool._format_timing(s) == "every"
# -- _format_state tests --
def test_format_state_empty() -> None:
def test_format_state_empty(tmp_path) -> None:
tool = _make_tool(tmp_path)
state = CronJobState()
assert CronTool._format_state(state) == []
assert tool._format_state(state, CronSchedule(kind="every")) == []
def test_format_state_last_run_ok() -> None:
def test_format_state_last_run_ok(tmp_path) -> None:
tool = _make_tool(tmp_path)
state = CronJobState(last_run_at_ms=1773673200000, last_status="ok")
lines = CronTool._format_state(state)
lines = tool._format_state(state, CronSchedule(kind="cron", expr="0 9 * * *", tz="UTC"))
assert len(lines) == 1
assert "Last run:" in lines[0]
assert "ok" in lines[0]
def test_format_state_last_run_with_error() -> None:
def test_format_state_last_run_with_error(tmp_path) -> None:
tool = _make_tool(tmp_path)
state = CronJobState(last_run_at_ms=1773673200000, last_status="error", last_error="timeout")
lines = CronTool._format_state(state)
lines = tool._format_state(state, CronSchedule(kind="cron", expr="0 9 * * *", tz="UTC"))
assert len(lines) == 1
assert "error" in lines[0]
assert "timeout" in lines[0]
def test_format_state_next_run_only() -> None:
def test_format_state_next_run_only(tmp_path) -> None:
tool = _make_tool(tmp_path)
state = CronJobState(next_run_at_ms=1773684000000)
lines = CronTool._format_state(state)
lines = tool._format_state(state, CronSchedule(kind="cron", expr="0 9 * * *", tz="UTC"))
assert len(lines) == 1
assert "Next run:" in lines[0]
def test_format_state_both() -> None:
def test_format_state_both(tmp_path) -> None:
tool = _make_tool(tmp_path)
state = CronJobState(
last_run_at_ms=1773673200000, last_status="ok", next_run_at_ms=1773684000000
)
lines = CronTool._format_state(state)
lines = tool._format_state(state, CronSchedule(kind="cron", expr="0 9 * * *", tz="UTC"))
assert len(lines) == 2
assert "Last run:" in lines[0]
assert "Next run:" in lines[1]
def test_format_state_unknown_status() -> None:
def test_format_state_unknown_status(tmp_path) -> None:
tool = _make_tool(tmp_path)
state = CronJobState(last_run_at_ms=1773673200000, last_status=None)
lines = CronTool._format_state(state)
lines = tool._format_state(state, CronSchedule(kind="cron", expr="0 9 * * *", tz="UTC"))
assert "unknown" in lines[0]
@@ -181,7 +204,7 @@ def test_list_every_job_milliseconds(tmp_path) -> None:
def test_list_at_job_shows_iso_timestamp(tmp_path) -> None:
tool = _make_tool(tmp_path)
tool = _make_tool_with_tz(tmp_path, "Asia/Shanghai")
tool._cron.add_job(
name="One-shot",
schedule=CronSchedule(kind="at", at_ms=1773684000000),
@@ -189,6 +212,7 @@ def test_list_at_job_shows_iso_timestamp(tmp_path) -> None:
)
result = tool._list_jobs()
assert "at 2026-" in result
assert "Asia/Shanghai" in result
def test_list_shows_last_run_state(tmp_path) -> None:
@@ -206,6 +230,7 @@ def test_list_shows_last_run_state(tmp_path) -> None:
result = tool._list_jobs()
assert "Last run:" in result
assert "ok" in result
assert "(UTC)" in result
def test_list_shows_error_message(tmp_path) -> None:
@@ -234,6 +259,30 @@ def test_list_shows_next_run(tmp_path) -> None:
)
result = tool._list_jobs()
assert "Next run:" in result
assert "(UTC)" in result
def test_add_cron_job_defaults_to_tool_timezone(tmp_path) -> None:
tool = _make_tool_with_tz(tmp_path, "Asia/Shanghai")
tool.set_context("telegram", "chat-1")
result = tool._add_job("Morning standup", None, "0 8 * * *", None, None)
assert result.startswith("Created job")
job = tool._cron.list_jobs()[0]
assert job.schedule.tz == "Asia/Shanghai"
def test_add_at_job_uses_default_timezone_for_naive_datetime(tmp_path) -> None:
tool = _make_tool_with_tz(tmp_path, "Asia/Shanghai")
tool.set_context("telegram", "chat-1")
result = tool._add_job("Morning reminder", None, None, None, "2026-03-25T08:00:00")
assert result.startswith("Created job")
job = tool._cron.list_jobs()[0]
expected = int(datetime(2026, 3, 25, 0, 0, 0, tzinfo=timezone.utc).timestamp() * 1000)
assert job.schedule.at_ms == expected
def test_list_excludes_disabled_jobs(tmp_path) -> None:
+39
View File
@@ -60,6 +60,45 @@ def test_openrouter_spec_is_gateway() -> None:
assert spec.default_api_base == "https://openrouter.ai/api/v1"
def test_openrouter_sets_default_attribution_headers() -> None:
spec = find_by_name("openrouter")
with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as MockClient:
OpenAICompatProvider(
api_key="sk-or-test-key",
api_base="https://openrouter.ai/api/v1",
default_model="anthropic/claude-sonnet-4-5",
spec=spec,
)
headers = MockClient.call_args.kwargs["default_headers"]
assert headers["HTTP-Referer"] == "https://github.com/HKUDS/nanobot"
assert headers["X-OpenRouter-Title"] == "nanobot"
assert headers["X-OpenRouter-Categories"] == "cli-agent,personal-agent"
assert "x-session-affinity" in headers
def test_openrouter_user_headers_override_default_attribution() -> None:
spec = find_by_name("openrouter")
with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as MockClient:
OpenAICompatProvider(
api_key="sk-or-test-key",
api_base="https://openrouter.ai/api/v1",
default_model="anthropic/claude-sonnet-4-5",
extra_headers={
"HTTP-Referer": "https://nanobot.ai",
"X-OpenRouter-Title": "Nanobot Pro",
"X-Custom-App": "enabled",
},
spec=spec,
)
headers = MockClient.call_args.kwargs["default_headers"]
assert headers["HTTP-Referer"] == "https://nanobot.ai"
assert headers["X-OpenRouter-Title"] == "Nanobot Pro"
assert headers["X-OpenRouter-Categories"] == "cli-agent,personal-agent"
assert headers["X-Custom-App"] == "enabled"
@pytest.mark.asyncio
async def test_openrouter_keeps_model_name_intact() -> None:
"""OpenRouter gateway keeps the full model name (gateway does its own routing)."""