智能体 | Nanobot 核心模式的业务逻辑

async def cmd_help(ctx: CommandContext) -> OutboundMessage:
    """Return available slash commands."""
    return OutboundMessage(
        channel=ctx.msg.channel,
        chat_id=ctx.msg.chat_id,
        content=build_help_text(),
        metadata={**dict(ctx.msg.metadata or {}), "render_as": "text"},
    )

def register_builtin_commands(router: CommandRouter) -> None:
    """Register the default set of slash commands."""
    router.priority("/stop", cmd_stop)
    router.priority("/restart", cmd_restart)
    router.priority("/status", cmd_status)
    router.exact("/new", cmd_new)
    router.exact("/status", cmd_status)
    router.exact("/dream", cmd_dream)
    router.exact("/dream-log", cmd_dream_log)
    router.prefix("/dream-log ", cmd_dream_log)
    router.exact("/dream-restore", cmd_dream_restore)
    router.prefix("/dream-restore ", cmd_dream_restore)
    router.exact("/help", cmd_help)

时间轴 →
UserA 发送 msg1 ──────────────────────→ [Lock A 获取] ──→ [Gate] ──→ 处理中...
UserA 发送 msg2 ──→ [Lock A 等待...] ──────────────────────────────────→ 处理完msg1后获取
UserB 发送 msg1 ──────────────────────→ [Lock B 获取] ──→ [Gate] ──→ 处理中...
UserC 发送 msg1 ──────────────────────→ [Lock C 获取] ──→ [Gate] ──→ 处理中...
UserD 发送 msg1 ──────────────────────→ [Lock D 获取] ──→ [Gate 等待...] ──→ 有空位后处理
                  ↑                      ↑               ↑
             每个会话独立锁          默认最多3个并发

self._session_locks: dict[str, asyncio.Lock] = {}
# NANOBOT_MAX_CONCURRENT_REQUESTS: <=0 means unlimited; default 3.
_max = int(os.environ.get("NANOBOT_MAX_CONCURRENT_REQUESTS", "3"))
self._concurrency_gate: asyncio.Semaphore | None = (
    asyncio.Semaphore(_max) if _max > 0 else None)

# _dispatch 中的并发控制
async def _dispatch(self, msg: InboundMessage) -> None:
    lock = self._session_locks.setdefault(msg.session_key, asyncio.Lock())
    gate = self._concurrency_gate or nullcontext()
    
    async with lock, gate:  # 同时获取两个！
        # lock: 同一会话的消息排队等待
        # gate: 最多 N 个请求并行（跨所有会话）
        response = await self._process_message(...)

正常流程:
  User:"帮我分析这个文件"
    ↓
  LLM → read_file("data.csv")  ← 中断！进程崩溃
    ↓
[保存断点到 Session.metadata]
    {"assistant_message":{...},"completed_tool_results":[...],"pending_tool_calls":[...]}

下次请求:
  User:"继续" (或任何新消息)
    ↓
  _restore_runtime_checkpoint(session)  ← 检测到未完成的断点
    ↓
  取出 pending_tool_calls 中的工具，并在 session 中标记为错误
  恢复 assistant_message + completed_tool_results 到历史
    ↓
  LLM 收到上下文后可以重新决策

def _set_runtime_checkpoint(self, session: Session, payload: dict) -> None:
    """保存当前进行中的状态（由 runner.checkpoint_callback 调用）"""
    session.metadata[self._RUNTIME_CHECKPOINT_KEY] = payload
    self.sessions.save(session)  # 持久化！

def_restore_runtime_checkpoint(self, session: Session) -> bool:
    """恢复中断的回合"""
    checkpoint = session.metadata.get("runtime_checkpoint")
    ifnotisinstance(checkpoint, dict):
        returnFalse# 无断点，正常处理
    
    # 1. 恢复助手消息
    restored_messages.append(checkpoint["assistant_message"])
    
    # 2. 恢复已完成的工具结果
    for msg in checkpoint["completed_tool_results"]:
        restored_messages.append(msg)
    
    # 3. 在对话中，将未完成的工具标记为错误
    for tc in checkpoint["pending_tool_calls"]:
        restored_messages.append({
            "role": "tool",
            "content": "Error: Task interrupted before this tool finished.",
            ...
        })
    
    # 4. 去重后写入历史
    session.messages.extend(restored_messages[overlap:])
    
    self._clear_runtime_checkpoint(session)  # 清除断点
    returnTrue

启动时刻:
  AgentLoop.__init__()
  → self._mcp_connected = False
  → self._mcp_servers = config  # 仅存储配置，不连接

第一条消息到达:
  _process_message() 
  → _connect_mcp()  ← 首次调用
  → 检查: notself._mcp_connected? → 连接
  → AsyncExitStack 管理 MCP 服务器生命周期

后续消息:
  → _connect_mcp() 
  → 检查: self._mcp_connected == True? → 跳过（直接返回）

关闭时:
  close_mcp() → aclose() 清理所有 MCP 连接

async def_connect_mcp(self) -> None:
    """延迟连接 MCP 服务器（首次消息触发）"""
    # 三重保护防止重复连接
    ifself._mcp_connected orself._mcp_connecting ornotself._mcp_servers:
        return
    
    self._mcp_connecting = True# 正在连接标志（防并发）
    try:
        self._mcp_stack = AsyncExitStack()  # 自动资源管理
        awaitself._mcp_stack.__aenter__()
        await connect_mcp_servers(self._mcp_servers, self.tools, self._mcp_stack)
        self._mcp_connected = True# 连接成功标志
    except BaseException as e:
        logger.error("MCP 连接失败（下次消息会重试）: {}", e)
        # 清理失败的连接，不设置 _connected，下次重试
    finally:
        self._mcp_connecting = False# 无论成败都释放连接锁

asyncdefclose_mcp(self) -> None:
    """优雅关闭：先等待后台任务完成，再关闭 MCP"""
    ifself._background_tasks:
        await asyncio.gather(*self._background_tasks, return_exceptions=True)
    ifself._mcp_stack:
        awaitself._mcp_stack.aclose()  # AsyncExitStack 自动关闭所有 MCP

# 主循环 run() 中：启动时立即尝试连接（预热）
asyncdefrun(self):
    self._running = True
    awaitself._connect_mcp()  # 启动时连接（可选，非阻塞后续运行）
    ...

# process_direct() 中：直接调用也触发连接
asyncdefprocess_direct(self, ...):
    awaitself._connect_mcp()  # CLI 直接模式也需要
    ...

# _process_message() 中：通过 _run_agent_loop 间接依赖
# （runner 可能调用 MCP 工具，但连接已在入口处保证）