diff --git a/.cursorindexingignore b/.cursorindexingignore new file mode 100644 index 000000000..953908e73 --- /dev/null +++ b/.cursorindexingignore @@ -0,0 +1,3 @@ + +# Don't index SpecStory auto-save files, but allow explicit context inclusion via @ references +.specstory/** diff --git a/.specstory/.gitignore b/.specstory/.gitignore new file mode 100644 index 000000000..c5b4129ef --- /dev/null +++ b/.specstory/.gitignore @@ -0,0 +1,4 @@ +# SpecStory project identity file +/.project.json +# SpecStory explanation file +/.what-is-this.md diff --git a/.specstory/history/2026-02-25_09-23Z-current-branch-status.md b/.specstory/history/2026-02-25_09-23Z-current-branch-status.md new file mode 100644 index 000000000..4d968c8b8 --- /dev/null +++ b/.specstory/history/2026-02-25_09-23Z-current-branch-status.md @@ -0,0 +1,266 @@ + + + + +# Current branch status (2026-02-25 09:23Z) + +_**User (2026-02-25 09:23Z)**_ + +查看当前分支情况 + +--- + +_**Agent (model claude-4.5-sonnet-thinking, mode Agent)**_ + +
Thought Process +用户要查看当前分支情况。我需要运行 git 命令来查看当前分支状态。最常用的命令是： +- `git branch` - 查看本地分支 +- `git status` - 查看当前分支和状态 +- `git branch -a` - 查看所有分支（包括远程） + +让我先运行 git status 和 git branch 来查看当前分支情况。
 + +--- + +我来帮你查看当前的分支情况。 + +--- + + +
Tool use: **run_terminal_cmd** • Run command: cd /home/0668001505/Desktop/picoclaw/picoclaw && git branch -a + +```bash +cd /home/0668001505/Desktop/picoclaw/picoclaw && git branch -a +``` + +``` +* main + remotes/origin/HEAD -> origin/main + remotes/origin/feat/agent-browser-tool + remotes/origin/feat/mcp-context7-integration-cleanup + remotes/origin/main + remotes/origin/refactor/channel-system + +``` +
+ + +--- + + +
Tool use: **run_terminal_cmd** • Run command: cd /home/0668001505/Desktop/picoclaw/picoclaw && git status + +```bash +cd /home/0668001505/Desktop/picoclaw/picoclaw && git status +``` + +``` +位于分支 main +您的分支与上游分支 'origin/main' 一致。 + +未跟踪的文件: + （使用 "git add <文件>..." 以包含要提交的内容） + .cursorindexingignore + .specstory/ + +提交为空，但是存在尚未跟踪的文件（使用 "git add" 建立跟踪） + +``` +
+ + +--- + +
Thought Process +好的，我已经获取了分支信息。让我总结一下： + +1. 当前在 main 分支上 +2. 本地分支只有 main +3. 远程分支有： + - origin/main + - origin/feat/agent-browser-tool + - origin/feat/mcp-context7-integration-cleanup + - origin/refactor/channel-system +4. 当前分支与上游分支 origin/main 一致 +5. 有一些未跟踪的文件（.cursorindexingignore 和 .specstory/） + +现在我用中文总结给用户。
 + +--- + +根据查看结果，当前分支情况如下： + +## 📍 当前分支 +- **当前所在分支**: `main` +- **状态**: 与上游分支 `origin/main` 一致 + +## 📋 本地分支 +- `main` (当前分支) + +## 🌐 远程分支 +- `origin/main` +- `origin/feat/agent-browser-tool` +- `origin/feat/mcp-context7-integration-cleanup` +- `origin/refactor/channel-system` + +## 📝 工作区状态 +- 工作区干净，没有待提交的更改 +- 有未跟踪的文件： + - `.cursorindexingignore` + - `.specstory/` + +你目前在主分支上，且与远程保持同步。有几个特性分支在远程仓库中，如果需要切换到其他分支进行开发，可以使用 `git checkout` 命令。 + +--- + +_**User (2026-02-25 09:23Z)**_ + +ZeroClaw 是一个高性能、低资源占用、可组合的自主智能体运行时。ZeroClaw 是面向智能代理工作流的运行时操作系统，通常被视为 OpenClaw 的 Rust 重写版本。 它抽象了模型、工具、记忆和执行层，使代理可以一次构建、随处运行，它通过 Trait 驱动架构实现了模型、通信渠道、工具、记忆系统等组件的可插拔替换，旨在提供一个安全、快速且资源友好的自主智能体运行环境。 +核心定位：定位为生产级“Agent 操作系统”（Agent OS），适用于边缘设备、高并发 Agent 集群、自动化运维等场景。 + +技术栈：采用 Rust 编写，单二进制部署，跨 ARM / x86 / RISC-V，单文件体积极小（3.4MB 或 8.8MB 版本），内存占用极低（低于 5MB），启动速度极快（启动时间可低至毫秒级）。 + +安全设计：内置安全沙箱、配对鉴权（pairing authentication）和显式 allowlist，默认安全性高。 + +特性：支持 17+ 多种通信渠道（Telegram、Discord、Slack 等），支持 22+ AI 提供商（OpenAI、Claude、Ollama 等），支持本地模型运行 + + + + +参考上面zeroclaw ，给出本项目的描述 + +--- + +_**Agent (model claude-4.5-sonnet-thinking, mode Agent)**_ + +
Thought Process +用户希望我参考 ZeroClaw 的描述风格，给出当前项目 picoclaw 的描述。我需要先了解一下 picoclaw 项目的实际情况，然后基于项目的真实特性，仿照 ZeroClaw 的描述风格来撰写。 + +让我先查看一下项目的基本信息，比如 README、package.json 或其他配置文件。
 + +--- + +我来帮你了解项目并给出描述。首先让我查看一下项目的基本信息。 + +--- + + +
Tool use: **read_file** • Read file: /home/0668001505/Desktop/picoclaw/picoclaw/README.md + +
+ + +--- + +Could not find file '/home/0668001505/Desktop/picoclaw/picoclaw/package.json' in the workspace. + +--- + + +
+ Tool use: **list_dir** • Listed directory, 0 results + +No results found +
+ + +--- + +
Thought Process +好的，现在我了解了 picoclaw 项目的情况。这是一个用 Go 语言编写的超轻量级 AI 助手，与 ZeroClaw（Rust 编写）的定位类似，但更注重极致的资源效率。 + +让我根据 README 的内容，参考 ZeroClaw 的描述风格，给出 picoclaw 的项目描述。
 + +--- + +根据项目信息，我为你撰写了 PicoClaw 的描述，参考了 ZeroClaw 的风格： + +--- + +## PicoClaw 项目描述 + +**PicoClaw 是一个超轻量、极低资源占用的个人 AI 助手运行时。** PicoClaw 受 nanobot 启发，从 TypeScript 完全重写为 Go 语言，通过 AI 自举过程（self-bootstrapping）完成架构迁移和代码优化，是 OpenClaw 的极致精简版本。它抽象了模型、工具、通信渠道和记忆系统，使智能代理可以一次构建、随处运行，旨在将 AI 助手带入 $10 硬件和 <10MB 内存的极限环境。 + +### 核心定位 +定位为**超轻量级"个人 AI 助手操作系统"（Personal AI Assistant OS）**，适用于边缘设备、低功耗硬件（RISC-V单板机）、旧手机复活、智能家居、服务器自动化运维等场景，让 AI 智能体能够在资源受限的环境中高效运行。 + +### 技术栈 +- **开发语言**：采用 Go 编写，单二进制部署，无任何外部依赖 +- **跨平台支持**：原生支持 x86_64、ARM64、RISC-V 架构 +- **极致轻量**： + - 内存占用：**<10MB**（比 OpenClaw 减少 99%，比 NanoBot 减少 90%） + - 启动速度：**<1秒**（即使在 0.6GHz 单核处理器上，比 OpenClaw 快 400 倍） + - 单文件体积：约 10MB 左右 +- **成本优势**：可运行在 $10 硬件上（比 Mac mini 便宜 98%） + +### 安全设计 +- **内置安全沙箱**：默认限制文件和命令访问范围（`restrict_to_workspace`） +- **白名单机制**：通过 `allow_from` 配置显式授权用户访问 +- **危险命令拦截**：自动阻止 `rm -rf`、`format`、`shutdown` 等危险操作 +- **一致性安全边界**：主 Agent、子 Agent、定时任务共享相同安全限制 + +### 核心特性 + +#### 🌐 多通道支持（7+ 通信平台） +- **即时通讯**：Telegram、Discord、QQ、钉钉、LINE、企业微信（WeCom） +- **灵活接入**：通过配置文件即可启用/禁用任意通道 + +#### 🤖 多模型支持（15+ AI 提供商） +- **主流大模型**：OpenAI、Anthropic (Claude)、Google Gemini、智谱 AI (GLM)、通义千问 (Qwen)、DeepSeek、Moonshot、Groq、Cerebras、NVIDIA +- **聚合服务**：OpenRouter（访问所有主流模型） +- **本地模型**：Ollama、vLLM、GitHub Copilot +- **零代码扩展**：通过 `vendor/model` 格式（如 `zhipu/glm-4.7`）即可添加新提供商 + +#### 🛠️ 丰富工具生态 +- **文件操作**：读写、编辑、列目录（沙箱保护） +- **命令执行**：系统命令执行（危险命令拦截） +- **网络搜索**：Brave Search、Tavily、DuckDuckGo（自动降级） +- **定时任务**：支持一次性提醒和周期性任务（Cron 表达式） +- **技能系统**：支持自定义技能扩展（Skills） +- **子 Agent 生成**：`spawn` 工具创建独立子智能体 + +#### 💾 记忆与上下文管理 +- **会话历史**：自动保存对话上下文（sessions/） +- **长期记忆**：持久化记忆系统（MEMORY.md） +- **Agent 人格**：可配置的身份（IDENTITY.md）和灵魂（SOUL.md） +- **用户偏好**：记录用户习惯和偏好（USER.md） + +#### ⏰ 心跳机制（Heartbeat） +- **周期性任务**：每 30 分钟自动执行 HEARTBEAT.md 中的任务 +- **异步子 Agent**：长时间任务通过 `spawn` 创建独立子智能体，不阻塞主流程 +- **主动通知**：子 Agent 可通过 `message` 工具主动向用户推送结果 + +#### 🐳 部署方式多样 +- **预编译二进制**：直接下载运行 +- **源码编译**：支持 `make build` 和 `make build-all`（多平台） +- **Docker Compose**：容器化部署，支持 Gateway 和 Agent 模式 +- **终端模式**：交互式命令行聊天 + +### 创新应用场景 +- **旧手机复活**：在 Termux 中运行，让十年前的 Android 手机变身 AI 助手 +- **极低成本部署**：$9.9 的 LicheeRV-Nano（RISC-V）可运行完整 AI 助手 +- **服务器运维**：NanoKVM 上实现自动化服务器维护 +- **智能监控**：MaixCAM 上构建智能摄像头 + +### 架构设计 +- **模块化架构**：Agent、Provider、Channel、Tool、Session 完全解耦 +- **可插拔组件**：通过接口（Interface）实现组件替换 +- **消息总线**：基于 Event Bus 的消息路由系统 +- **会话管理**：支持多用户、多通道的会话隔离 +- **负载均衡**：同一 model_name 配置多个端点自动轮询 + +### 与同类项目对比 + +| | **OpenClaw** | **NanoBot** | **PicoClaw** | +|---------------------|---------------|-------------|----------------------| +| **语言** | TypeScript | Python | **Go** | +| **内存占用** | >1GB | >100MB | **<10MB** | +| **启动时间**（0.8GHz）| >500s | >30s | **<1s** | +| **硬件成本** | Mac Mini $599 | SBC ~$50 | **最低 $10** | + +--- + +**核心理念**：让 AI 助手回归本质 —— 在任何设备上都能快速、安全、高效地运行，无需昂贵硬件，无需复杂配置。PicoClaw 证明了即使在最受限的环境中，也能构建功能完整的自主智能体系统。 + +--- + diff --git a/.specstory/history/2026-02-25_10-14Z-zeroclaw-autonomous-agent-runtime-overview.md b/.specstory/history/2026-02-25_10-14Z-zeroclaw-autonomous-agent-runtime-overview.md new file mode 100644 index 000000000..355159c3f --- /dev/null +++ b/.specstory/history/2026-02-25_10-14Z-zeroclaw-autonomous-agent-runtime-overview.md @@ -0,0 +1,155 @@ + + + + +# ZeroClaw autonomous agent runtime overview (2026-02-25 10:14Z) + +_**User (2026-02-25 09:23Z)**_ + +ZeroClaw 是一个高性能、低资源占用、可组合的自主智能体运行时。ZeroClaw 是面向智能代理工作流的运行时操作系统，通常被视为 OpenClaw 的 Rust 重写版本。 它抽象了模型、工具、记忆和执行层，使代理可以一次构建、随处运行，它通过 Trait 驱动架构实现了模型、通信渠道、工具、记忆系统等组件的可插拔替换，旨在提供一个安全、快速且资源友好的自主智能体运行环境。 +核心定位：定位为生产级“Agent 操作系统”（Agent OS），适用于边缘设备、高并发 Agent 集群、自动化运维等场景。 + +技术栈：采用 Rust 编写，单二进制部署，跨 ARM / x86 / RISC-V，单文件体积极小（3.4MB 或 8.8MB 版本），内存占用极低（低于 5MB），启动速度极快（启动时间可低至毫秒级）。 + +安全设计：内置安全沙箱、配对鉴权（pairing authentication）和显式 allowlist，默认安全性高。 + +特性：支持 17+ 多种通信渠道（Telegram、Discord、Slack 等），支持 22+ AI 提供商（OpenAI、Claude、Ollama 等），支持本地模型运行 + + + + +参考上面zeroclaw ，给出本项目的描述 + +--- + +_**Agent (model claude-4.5-sonnet-thinking, mode Agent)**_ + +
Thought Process +用户想要我参考 ZeroClaw 的描述，为当前项目 picoclaw 提供一个类似的项目描述。 + +让我先探索一下这个项目的结构和内容，了解它的特点和功能。 + +我需要： +1. 查看项目结构 +2. 了解项目的主要文件和配置 +3. 基于 ZeroClaw 的描述风格，为 picoclaw 创建一个类似的描述 + +让我先探索项目结构。
 + +--- + +我来帮您探索 picoclaw 项目并给出描述。让我先查看项目结构。 + +--- + + +
+ Tool use: **list_dir** • Listed directory, 0 results + +No results found +
+ + +--- + + +
Tool use: **read_file** • Read file: /home/0668001505/Desktop/picoclaw/picoclaw/README.md + +
+ + +--- + + +
+Tool use: **glob_file_search** • Searched codebase "Cargo.toml" • **0** result in **1** directory + +Directory: **/home/0668001505/Desktop/picoclaw/picoclaw** (0 file) + +
+ + +--- + +
Thought Process +好的，我现在了解了 picoclaw 项目的结构。让我查看更多关键文件来了解项目的核心特性。
 + +--- + + +
Tool use: **read_file** • Read file: /home/0668001505/Desktop/picoclaw/picoclaw/go.mod + +
+ + +--- + + +
Tool use: **read_file** • Read file: /home/0668001505/Desktop/picoclaw/picoclaw/README.zh.md + +
+ + +--- + +
Thought Process +好的，现在我对 picoclaw 项目有了全面的了解。让我基于这些信息，参考 ZeroClaw 的描述风格，为 PicoClaw 项目创建一个类似的中文描述。 + +PicoClaw 的特点： +1. 使用 Go 语言编写 +2. 超轻量级 - <10MB 内存占用 +3. 极快启动 - 1秒内启动 +4. 单二进制部署 +5. 跨平台 - 支持 ARM、x86、RISC-V +6. 低成本 - 可在 $10 硬件上运行 +7. 支持多种通信渠道（Telegram、Discord、QQ、钉钉、飞书、企业微信等） +8. 支持 16+ AI 提供商 +9. 安全沙箱设计 +10. AI 自举开发（95% 由 AI Agent 生成） + +让我创建一个类似 ZeroClaw 风格的描述：
 + +--- + +根据我对项目的分析，以下是参考 ZeroClaw 风格为 **PicoClaw** 项目撰写的描述： + +--- + +## PicoClaw 项目描述 + +**PicoClaw** 是一个超轻量级、超低成本、极速启动的自主智能体运行时。PicoClaw 是面向边缘设备和资源受限环境的智能代理运行时系统，受 [nanobot](https://github.com/HKUDS/nanobot) 启发，是通过 AI 自举过程完全用 Go 语言重写的个人 AI 助手框架。它通过极简化的架构设计和 Go 语言的并发特性实现了模型、通信渠道、工具、记忆系统等组件的灵活配置和组合，旨在提供一个极致轻量、快速响应且资源友好的智能代理运行环境。 + +**核心定位**：定位为"人人可用的边缘 AI 助手"（Edge AI Assistant for Everyone），适用于老旧手机、$10 开发板、边缘设备、个人助理、智能监控等场景。 + +**技术栈**：采用 Go 语言编写，单二进制部署，跨 ARM / x86 / RISC-V 架构，单文件体积极小（约 8-10MB），内存占用极低（低于 10MB，比 OpenClaw 节省 99% 内存），启动速度极快（在 0.6GHz 单核处理器上 1 秒内启动，比 NanoBot 快 30 倍，比 OpenClaw 快 400 倍）。 + +**成本优势**：可在 $10 硬件上流畅运行（如 LicheeRV-Nano），比 Mac mini 便宜 98%，能够将老旧 Android 手机改造为智能助手，实现硬件资源的二次利用。 + +**安全设计**：内置安全沙箱（workspace 隔离）、通信渠道白名单机制（allow_from）和命令执行保护，默认限制文件访问和命令执行范围，支持配置化的安全边界控制。 + +**核心特性**： +- 支持 10+ 多种通信渠道（Telegram、Discord、QQ、钉钉、飞书、企业微信、Slack、LINE、OneBot、WhatsApp 等） +- 支持 16+ AI 提供商（OpenAI、Anthropic/Claude、智谱、DeepSeek、Gemini、Groq、Qwen/通义千问、Moonshot、Ollama、Cerebras、火山引擎、GitHub Copilot 等） +- 支持本地模型运行（Ollama、VLLM） +- 内置负载均衡和模型回退机制 +- 支持定时任务（Cron）和心跳周期任务（Heartbeat） +- 支持子 Agent 异步任务派发（Spawn） +- 支持语音转文字（通过 Groq Whisper） +- 支持网络搜索（Brave、Tavily、DuckDuckGo） +- 支持技能市场（ClawHub）和自定义技能 +- AI 自举开发（95% 核心代码由 AI Agent 生成，人机协作优化） + +**部署场景**： +- 🦐 **个人助理**：在老旧手机（通过 Termux）或低成本开发板上运行 +- 🏠 **家庭助手**：LicheeRV-Nano ($9.9) 用于极简家庭 AI 助理 +- 🖥️ **运维助手**：NanoKVM ($30-100) 用于自动化服务器运维 +- 📹 **智能监控**：MaixCAM/MaixCAM2 ($50-100) 用于 AI 视觉监控 +- 💬 **企业协作**：通过钉钉、飞书、企业微信集成到企业工作流 +- 🌐 **多渠道接入**：同时连接多个聊天平台，统一管理对话 + +--- + +这个描述突出了 PicoClaw 与 ZeroClaw 的相似之处（轻量、跨平台、安全、多渠道），同时强调了其独特优势（极低成本、边缘设备部署、AI 自举开发）。 + +--- + diff --git a/.specstory/history/2026-02-25_10-32Z-version-0-1-1-changes-summary.md b/.specstory/history/2026-02-25_10-32Z-version-0-1-1-changes-summary.md new file mode 100644 index 000000000..8bf00dfb2 --- /dev/null +++ b/.specstory/history/2026-02-25_10-32Z-version-0-1-1-changes-summary.md @@ -0,0 +1,145 @@ + + + + +# Version 0.1.1 changes summary (2026-02-25 10:32Z) + +_**User (2026-02-25 09:23Z)**_ + +v0.1.1 +What's Changed +feat: implement dynamic context compression by @Danieldd28 in #3 +fix: enable Feishu message flow by @imguoguo in #6 +Add QQ channel support by @yinwm in #5 +feat: Add DingTalk channel support via Stream Mode by @tianrking in #12 +Agent Memory System & Tool Execution Improvements by @yinwm in #14 +fix a typo by @n0bisuke in #19 +feat: Cron tool and agent integration by @yinwm in #23 +Remove duplicate truncate functions, reuse utils.Truncate by @DevEverything01 in #29 +fix(agent): use atomic.Bool for AgentLoop.running to prevent data race by @DevEverything01 in #30 +chore: lint readme.md by @Sethispr in #44 +feat(auth): add OAuth login with SDK-based subscription providers by @corylanou in #32 +feat(migrate): add picoclaw migrate command for OpenClaw workspace migration by @corylanou in #33 +feat(channels): add Slack channel integration with Socket Mode by @corylanou in #34 +feat(telegram): Use Telego instead of go-telegram-bot-api by @mymmrac in #40 +better version info by @jadeydi in #45 +refactor(channels): consolidate media handling and improve resource cleanup by @yinwm in #49 +Add provider field support for explicit provider selection by @victorhdchagas in #48 +Fix Telegram channel permission check by @victorhdchagas in #51 +Enforce workspace directory boundaries for system tools by @RinZ27 in #26 +ci: add build check workflow by @imguoguo in #71 +Fix LLM error by cleaning up CONSCIOUSLY message history by @Diegox-17 in #55 +fix(auth): support string interval in OpenAI device-code flow by @jmahotiedu in #56 +fix(heartbeat): resolve bug where service could never start by @DevEverything01 in #64 +Add ClawdChat Agent Social Network section by @xray918 in #72 +feat: add cli-based LLM provider by @Leeaandrob in #73 +feat(cron): Add support for direct shell command execution in scheduled jobs by @SatyamDevv in #74 +build: add darwin arm64 target by @imguoguo in #76 +fix(version): default to dev by @imguoguo in #95 +以上是git的0.1.1版本的提交记录，请简要描述改版本的主要改动 + +--- + +_**Agent (model claude-4.5-sonnet-thinking, mode Agent)**_ + +
Thought Process +用户要求我用中文总结这个Git版本（v0.1.1）的主要改动。我需要分析这些变更记录并将其归类为几个主要的改进方向。 + +让我看看这些变更： +1. 动态上下文压缩 +2. 消息流支持（Feishu、QQ、DingTalk、Slack、Telegram） +3. Agent内存系统和工具执行改进 +4. Cron工具 +5. OAuth登录 +6. 迁移命令 +7. 多个渠道集成 +8. 版本信息改进 +9. 媒体处理和资源清理 +10. Provider字段支持 +11. 工作空间目录边界 +12. CI构建检查 +13. 错误修复（LLM、认证、心跳等） +14. CLI-based LLM provider +15. Cron直接shell命令执行 +16. Darwin arm64支持 + +我应该将这些归纳为几个主要类别： +- 多平台/渠道支持扩展 +- Agent能力增强 +- 认证和安全改进 +- 开发者体验改进 +- Bug修复和性能优化
 + +--- + +基于这些提交记录，v0.1.1版本的主要改动可以归纳为以下几个方面： + +## 📡 **多平台渠道支持** +- 新增多个通讯平台集成：**QQ频道**、**钉钉（DingTalk）**、**Slack**、**飞书（Feishu）** +- 改进 **Telegram** 集成（使用Telego替代原有库） +- 统一媒体处理和资源清理机制 + +## 🤖 **Agent能力增强** +- 实现 **Agent记忆系统**和工具执行改进 +- 添加 **Cron定时任务**工具，支持直接执行Shell命令 +- 实现 **动态上下文压缩**功能 +- 添加 **CLI-based LLM provider** 支持 +- 修复LLM错误历史消息清理问题 + +## 🔐 **认证与安全** +- 添加 **OAuth登录**功能（基于SDK的订阅提供商） +- 强化 **工作空间目录边界**安全控制 +- 修复Telegram权限检查问题 + +## 🛠️ **开发者体验** +- 新增 `picoclaw migrate` 命令用于OpenClaw工作空间迁移 +- 添加 **CI构建检查**工作流 +- 改进版本信息显示 +- 添加 **darwin arm64** 构建支持 +- 代码优化：移除重复函数、使用原子操作防止数据竞争 + +## 🐛 **Bug修复** +- 修复心跳服务无法启动的问题 +- 修复OpenAI设备代码流中的字符串间隔支持 +- 修复权限检查和各种小问题 + +总体来说，这是一个**功能丰富的版本**，重点在于**扩展多平台支持**、**增强Agent智能能力**和**改善开发者工具链**。 + +--- + +_**User (2026-02-25 09:23Z)**_ + +用100字总结 + +--- + +_**Agent (model claude-4.5-sonnet-thinking, mode Agent)**_ + +
Thought Process +用户要求用100字总结v0.1.1版本的主要改动。我需要更简洁地概括核心内容。 + +核心内容： +1. 多平台支持（QQ、钉钉、Slack、飞书、Telegram） +2. Agent功能增强（记忆系统、Cron工具） +3. OAuth认证 +4. 迁移命令 +5. Bug修复和优化 + +用中文100字左右总结：
 + +--- + +v0.1.1版本主要更新： + +**多平台集成**：新增QQ频道、钉钉、Slack、飞书支持，改进Telegram集成。 + +**Agent增强**：实现记忆系统、动态上下文压缩、Cron定时任务工具。 + +**认证与工具**：添加OAuth登录、`picoclaw migrate`迁移命令。 + +**安全优化**：强化工作空间目录边界控制。 + +**开发改进**：添加CI构建、支持darwin arm64、修复数据竞争和心跳服务等多项Bug。 + +--- + diff --git a/.specstory/history/2026-02-25_10-36Z-version-0-1-2-release-notes.md b/.specstory/history/2026-02-25_10-36Z-version-0-1-2-release-notes.md new file mode 100644 index 000000000..a10bceb48 --- /dev/null +++ b/.specstory/history/2026-02-25_10-36Z-version-0-1-2-release-notes.md @@ -0,0 +1,282 @@ + + + + +# Version 0.1.2 release notes (2026-02-25 10:36Z) + +_**User (2026-02-25 09:23Z)**_ + +42e0e58 * The docker-build workflow will no longer run when submitting a PR. +7140e73 * fix fmt bug +68abf6b * fix: docker build failed +878650c * fix: heartbeat/service_test.go test failed +951b05d 1. add AI Code Generation selection in pr template +5fb2721 1. add android phone termux quick guide +0fadbcd 1. add roadmap.md +da79c20 1. fix typo +ac4b16d 1. rename doc to docs +13e4028 1. update wechat group qrcode 2. publish roadmap +b484d3f Add dependencies for fmt-check in vet and test jobs +341dbd3 Add health check endpoints (/health and /ready) for container orchestration liveness and readiness probes. (#104) +cd638ff Add local AI ollama for security purpose (#226) +1516cb5 Change Docker build trigger to on release published +8d757fb Feat issue 183 (#189) +32cb8fd Feat: Discord message length check and auto split (#143) +ba0ef4b Merge branch 'main' into architecture-32-bit +2cb90f2 Merge branch 'main' into architecture-32-bit +bc27707 Merge branch 'main' into main +bd9c9d7 Merge branch 'main' into main +fc40f29 Merge branch 'main' into main +25a47b5 Merge branch 'main' into patch-1 +a286100 Merge branch 'main' into patch-1 +f739c45 Merge branch 'main' of https://github.com/SatyamDevv/picoclaw +04924ed Merge branch 'main' of https://github.com/SatyamDevv/picoclaw +e0a7662 Merge branch 'main' of https://github.com/sipeed/picoclaw +0c4b8b0 Merge branch 'sipeed:main' into main +fbe1152 Merge branch 'sipeed:main' into main +9036a51 Merge branch 'sipeed:main' into main +0d339d9 Merge branch 'sipeed:main' into main +5339389 Merge pull request #105 from Zhaoyikaiii/bugfix/fix-duplicate-telegram-messages +9ccfea4 Merge pull request #108 from yinwm/ralph/tool-result-refactor +c58f8b7 Merge pull request #111 from shengsuan/ssy +ee3e8cc Merge pull request #124 from mengzhuo/mengzhuo-fmt-1 +14de80d Merge pull request #128 from yinwm/feat/better-version +82a9a80 Merge pull request #130 from MHCP000/fix/codeblock-index-bug +3334595 Merge pull request #134 from Sethispr/patch-1 +5872e0f Merge pull request #147 from ex-takashima/feat/line-channel +ddd73ca Merge pull request #151 from qiaoborui/codex/fix-openai-oauth-authorize-url +1cff7d4 Merge pull request #153 from alexhoshina/doc/fix-errors +b805ec8 Merge pull request #158 from easyzoom/feat/device-hotplug-notifications +0f506d4 Merge pull request #165 from alexhoshina/feat/onebot +59c7aa1 Merge pull request #167 from Lixeer/main +1cb690d Merge pull request #169 from PixelTux/architecture-32-bit +6ce7659 Merge pull request #172 from mymmrac/docker-curl +8eb9dcd Merge pull request #173 from is-Xiaoen/fix/code-review-bugs-116 +6f2e730 Merge pull request #178 from Lixeer/main +7f60392 Merge pull request #24 from Esubaalew/main +55d5e89 Merge pull request #78 from SatyamDevv/main +1aea912 Merge pull request #83 from carzygod/main +c0d1346 Merge pull request #85 from lesichkovm/patch-1 +3c2e467 Merge remote-tracking branch 'origin/HEAD' into feat/better-version +a6aa833 Merge remote-tracking branch 'origin/main' into ralph/tool-result-refactor +53b5be8 Merge remote-tracking branch 'origin/main' into ralph/tool-result-refactor +ab20314 Merge upstream/main into ralph/tool-result-refactor +a371d53 Prevent panic on publish after MessageBus is closed (#223) +1d748fb Remove duplicate file extension in DownloadFile (#230) +e7f15af Update issue templates +5893245 Update launch announcement in README +2720fa7 add I2C and SPI tools for hardware interaction (#140) +cddafb4 add build constraints for feishu to support 32-bit builds +811e4f8 add when picoclaw responsed to discord message will show its typing (#236) +132fe7d bugfix: fix duplicate Telegram message sending +159a954 build: support building for linux/loong64 (#272) +a5503ae build: temporary disable UPX compression (#257) +7fa70b8 chore(docs): remove completed PRD document from tasks +b36c87b chore: Clean up Ralph agent tracking files +3eb9d6a chore: Remove backup cron files +ff92973 chore: fix tab in build +d7822e5 chore: fmt code in build +15e3c7d chore: lint readme +ecbe315 chore: remove redundant debug output +9a3f361 ci: init goreleaser +0d18210 ci: use goreleaser to release docker and binary (#180) +5a6ad37 code fmt +f294a71 feat(channels): add LINE Official Account channel support +5aa4dd2 feat(cli): add git commit hash to version output +a24cbd4 feat(docker): Added curl for Docker image +c6c82b3 feat(skills): add validation for skill info and test cases (#231) +53df8d1 feat: Add DuckDuckGo search fallback. +18d3634 feat: Improve parameter fault tolerance for DeepSeek +a9557aa feat: Support installing built-in AGENT files and skills during picoclaw onboard +ca781d4 feat: US-002 - Modify Tool interface to return *ToolResult +c6c61b4 feat: US-004 - Delete isToolConfirmationMessage function +b573d61 feat: US-005 - Update AgentLoop tool result processing logic +56ac18a feat: US-006 - Add AsyncCallback type and AsyncTool interface +7bcd8b2 feat: US-007 - Add heartbeat async task execution support +4c4c10c feat: US-008 - Inject callback into async tools in AgentLoop +b94941d feat: US-009 - Add state save atomicity with SetLastChannel +feba44e feat: US-010 - Add RecordLastChannel to AgentLoop with atomic state save +2989c39 feat: US-011 - Add MessageTool tests +e7e3f95 feat: US-012 - Add ShellTool tests +88014ec feat: US-013 - Add FilesystemTool tests +0ac93d4 feat: US-014 - Add WebTool tests +35fa64c feat: US-015 - Add EditTool tests +a141815 feat: US-016 - Refactor CronTool to use ToolResult +061b071 feat: US-016, US-017 - Mark CronTool and SpawnTool as complete +28734c3 feat: US-018 - Add SubagentTool with ToolResult support +03b02cc feat: US-019 - Enable heartbeat by default in config +e63f967 feat: US-020 - Move heartbeat log to memory directory +be81ba1 feat: US-021 - Heartbeat calls ExecuteHeartbeatWithTools +e77b0a6 feat: add Codex CLI provider for subprocess integration (#80) +7fa641a feat: add OneBot channel support +896eae4 feat: add ShengSuanYun(胜算云) as a models provider. +5faa67b feat: add Github Copilot provider +3780455 feat: add device hotplug event notifications (USB on Linux) +afc3a2c feat: add provider deepseek +2f5849b feat: add support for DuckDuckGo and refactor Brave search configuration support the control with config.js +9d5728e feat: implement structured Telegram command handling with a dedicated command service and telegohandler integration. (#164) +e7e0861 feat: merge heartbeat service improvements from feat-heartbeat branch +e353844 feat: re-enable cronTool service after refactor completion +17685da feat: update the make deps logic to prevent the project from frequently updating dependency package versions (#277) +0aab8d8 feat:add github_copilot to providers factory +7fa341c fix concurrency and persistence safety in session/cron/heartbeat services +7304ab7 fix(auth): align OpenAI OAuth authorize URL and params +a961a2d fix(ci): use env var for release tag (#342) +da804a0 fix(codex): include required instructions and improve account-id extraction +16e5a02 fix(http_provider): Remove extra parameter from CreateProvider function. +e3f65fc fix(security): block critical symlink workspace escape (#188) +9eb1a53 fix: PR workflow execution failure +dbf2739 fix: Remove the waiting animation for Telegram replies to reduce the risk of being rate-limited. +0cb9387 fix: codex agent 400 error (#102) +1e17bac fix: correct index bug in extractCodeBlocks and extractInlineCodes +8cff6cf fix: improve connection handling in OneBotChannel +4b886b6 fix: not used result +6c392c3 fix: openclaw crash because of empty config +474f3db fix: resolve code review issues in tool-result-refactor +0a88ff0 fix: resolve multiple bugs from code review #116 +5c321a9 fix: tighten file perms and enforce Slack ACL checks (#186) +773ce9b fix: update BotStatus type in oneBotRawEvent and adjust status checks +6463796 fix: update tests to match upstream API changes +000e64c gh: add pull request check +f6d6221 make fmt +7a96599 make fmt +aa7a8b8 merge: resolve conflict with upstream main +cf5a84a merge: resolve conflicts with upstream/main +584a3dc misc: fmt code +b594642 perf(cron): use read lock for concurrent reads in addJob +0cce9fc refactor(agent): extract reusable tool loop and make subagents independent +4dfa133 refactor(heartbeat): add configurable interval and channel-aware routing +8fbbb67 refactor(heartbeat): simplify service with single handler and direct bus usage +8851152 refactor(tools): extract shared logic for internal channels and tool definitions +92490fe refactor(version): extract version formatting helpers and improve build info display +45351a6 refactor: Trim trailing slashes from the API base URL and enhance API error messages with status codes. +1299b20 refactor: remove unused min func +c86e121 refactor: update tool registry usage and enhance WebSearchTool execution result handling +60c3d96 style: fix alignment of fields in BotStatus struct +1fcf3fd style: fix format +ac79a23 style: fix gofmt formatting (CRLF -> LF) +b5a8eff style: fix var block formatting +f742112 style: remove unnecessary log messages in WebSocket listener +0a4bf32 try after workflow fix +57dac39 update pr template +82fea61 update readme +875a16d update telegram username support +6083168 update wechat group qrcode +214b201 update wechat group qrcode + +以上是0.1.2版本的提交记录，请简要描述此版本的主要改动 + +--- + +_**Agent (model claude-4.5-sonnet-thinking, mode Agent)**_ + +
Thought Process +用户提供了一系列git提交记录，并要求我简要描述这些提交记录对应的0.1.2版本的主要改动。 + +让我分析这些提交记录，总结主要的功能改动： + +1. **工具系统重构** - 大量的feat: US-xxx系列提交，涉及工具接口返回ToolResult、异步工具支持、心跳任务等 +2. **新增渠道支持** - OneBot、LINE、Telegram改进 +3. **新增AI提供商** - DeepSeek、ShengSuanYun(胜算云)、GitHub Copilot、Codex CLI、本地Ollama +4. **新功能** + - 健康检查端点(/health和/ready) + - 设备热插拔通知 + - DuckDuckGo搜索 + - I2C和SPI硬件交互工具 + - 技能验证 +5. **CI/CD改进** - goreleaser支持、Docker构建优化 +6. **Bug修复** - 多个并发安全、消息重复、代码块索引等问题 +7. **文档更新** - roadmap、Android Termux指南、README更新 + +用户要求用中文回复，所以我需要用中文总结这些改动。
 + +--- + +根据这些提交记录，0.1.2版本的主要改动可以总结为： + +## 核心架构重构 +- **工具系统全面升级**：重构Tool接口返回`*ToolResult`，支持异步工具执行，优化工具结果处理逻辑 +- **心跳服务增强**：默认启用心跳功能，支持异步任务执行，增加可配置间隔和渠道路由 + +## 新增功能 + +### 通道支持 +- 新增 **LINE** 官方账号渠道支持 +- 新增 **OneBot** 渠道支持 +- 优化 **Telegram** 命令处理和用户名支持 +- 优化 **Discord** 消息长度检查和自动分割 + +### AI提供商 +- 新增 **DeepSeek** 提供商 +- 新增 **胜算云(ShengSuanYun)** 提供商 +- 新增 **GitHub Copilot** 提供商 +- 新增 **Codex CLI** 提供商用于子进程集成 +- 新增 **本地Ollama** 支持（用于安全目的） +- 新增 **DuckDuckGo** 搜索回退功能 + +### 其他功能 +- 健康检查端点（`/health` 和 `/ready`）用于容器编排 +- 设备热插拔事件通知（Linux USB） +- I2C 和 SPI 硬件交互工具 +- 技能文件验证和测试用例支持 +- 内置AGENT文件和技能在onboard时自动安装 + +## CI/CD改进 +- 集成 **goreleaser** 用于Docker和二进制文件发布 +- 优化Docker构建触发机制（改为release发布时触发） +- 新增PR检查流程 + +## Bug修复 +- 修复Telegram消息重复发送问题 +- 修复代码块索引bug +- 修复OpenAI OAuth授权URL问题 +- 修复并发安全和持久化问题（session/cron/heartbeat服务） +- 加强文件权限和Slack ACL检查 +- 修复符号链接工作区逃逸安全漏洞 + +## 文档更新 +- 新增项目路线图（roadmap.md） +- 新增Android Termux快速指南 +- 更新微信群二维码 +- 优化PR模板 + +这是一个功能丰富的版本，主要聚焦在工具系统重构、多渠道支持扩展和AI提供商生态完善。 + +--- + +_**User (2026-02-25 09:23Z)**_ + +用100次左右进行总结 + +--- + +_**Agent (model claude-4.5-sonnet-thinking, mode Agent)**_ + +
Thought Process +用户要求用100字左右进行总结，之前的回答太详细了。我需要精简到大约100个汉字。 + +让我重新组织一下，只提取最核心的几点： +1. 工具系统重构 +2. 新增多个渠道（LINE、OneBot） +3. 新增多个AI提供商（DeepSeek、GitHub Copilot、Ollama等） +4. 集成goreleaser +5. 安全和bug修复 + +100字左右的总结：
 + +--- + +## 0.1.2版本主要改动总结 + +**核心重构**：工具系统全面升级，支持异步执行和新的ToolResult返回机制；心跳服务增强并默认启用。 + +**新增渠道**：LINE官方账号、OneBot支持；优化Telegram和Discord消息处理。 + +**AI提供商扩展**：新增DeepSeek、胜算云、GitHub Copilot、本地Ollama等多个提供商。 + +**基础设施**：集成goreleaser自动化发布，新增健康检查端点，支持设备热插拔通知。 + +**安全加固**：修复符号链接逃逸漏洞、并发安全问题，加强权限控制和ACL检查。 + +--- + diff --git a/cmd/picoclaw/internal/agent/helpers.go b/cmd/picoclaw/internal/agent/helpers.go index f754abc65..2b49d2544 100644 --- a/cmd/picoclaw/internal/agent/helpers.go +++ b/cmd/picoclaw/internal/agent/helpers.go @@ -15,6 +15,7 @@ import ( "github.com/sipeed/picoclaw/pkg/agent" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/logger" + "github.com/sipeed/picoclaw/pkg/nodes" "github.com/sipeed/picoclaw/pkg/providers" ) @@ -51,6 +52,14 @@ func agentCmd(message, sessionKey, model string, debug bool) error { defer msgBus.Close() agentLoop := agent.NewAgentLoop(cfg, msgBus, provider) + // Register NodesTool with an empty registry so the tool is always visible. + // In agent mode no nodes server is running, so the registry starts empty + // (status/describe will return "no nodes connected"). In gateway mode the + // real registry with connected nodes is injected via SetNodeRegistry. + if cfg.Nodes.Enabled { + agentLoop.SetNodeRegistry(nodes.NewRegistry()) + } + // Print agent startup info (only for interactive mode) startupInfo := agentLoop.GetStartupInfo() logger.InfoCF("agent", "Agent initialized", diff --git a/cmd/picoclaw/internal/chat/command.go b/cmd/picoclaw/internal/chat/command.go new file mode 100644 index 000000000..84e0ae441 --- /dev/null +++ b/cmd/picoclaw/internal/chat/command.go @@ -0,0 +1,42 @@ +package chat + +import ( + "github.com/spf13/cobra" +) + +func NewChatCommand() *cobra.Command { + var ( + message string + sessionKey string + gatewayURL string + debug bool + ) + + cmd := &cobra.Command{ + Use: "chat", + Short: "Chat with a running picoclaw gateway", + Long: `Connect to a running picoclaw gateway and send messages interactively. + +The gateway must already be started with 'picoclaw gateway'. +By default the gateway URL is read from the config file (~/.picoclaw/config.json). + +Examples: + picoclaw chat # interactive REPL + picoclaw chat -m "what time is it?" # single message + picoclaw chat --url http://host:18790 # connect to remote gateway`, + Args: cobra.NoArgs, + RunE: func(_ *cobra.Command, _ []string) error { + return chatCmd(message, sessionKey, gatewayURL, debug) + }, + } + + cmd.Flags().StringVarP(&message, "message", "m", "", "Send a single message (non-interactive mode)") + cmd.Flags().StringVarP(&sessionKey, "session", "s", "cli:default", "Session key for conversation history") + cmd.Flags().StringVarP(&gatewayURL, "url", "u", "", "Gateway base URL (default: from config, e.g. http://127.0.0.1:18790)") + cmd.Flags().BoolVarP(&debug, "debug", "d", false, "Enable debug logging") + + return cmd +} + + + diff --git a/cmd/picoclaw/internal/chat/helpers.go b/cmd/picoclaw/internal/chat/helpers.go new file mode 100644 index 000000000..e143d6a9a --- /dev/null +++ b/cmd/picoclaw/internal/chat/helpers.go @@ -0,0 +1,205 @@ +package chat + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "strings" + "time" + + "github.com/chzyer/readline" + "github.com/sipeed/picoclaw/cmd/picoclaw/internal" + "github.com/sipeed/picoclaw/pkg/agent" + "github.com/sipeed/picoclaw/pkg/logger" +) + +func chatCmd(message, sessionKey, gatewayURL string, debug bool) error { + if debug { + logger.SetLevel(logger.DEBUG) + } + + // Resolve gateway URL from config if not provided. + if gatewayURL == "" { + cfg, err := internal.LoadConfig() + if err != nil { + return fmt.Errorf("error loading config: %w", err) + } + host := cfg.Gateway.Host + if host == "0.0.0.0" { + host = "127.0.0.1" + } + gatewayURL = fmt.Sprintf("http://%s:%d", host, cfg.Gateway.Port) + } + + // Strip trailing slash for consistency. + gatewayURL = strings.TrimRight(gatewayURL, "/") + + // Check that the gateway is reachable. + if err := checkGateway(gatewayURL); err != nil { + return fmt.Errorf("gateway not available at %s: %w\n(start it with: picoclaw gateway)", gatewayURL, err) + } + + fmt.Printf("%s Connected to gateway at %s\n", internal.Logo, gatewayURL) + + // Single-message mode. + if message != "" { + resp, err := sendMessage(gatewayURL, message, sessionKey) + if err != nil { + return err + } + fmt.Printf("\n%s %s\n", internal.Logo, resp) + return nil + } + + // Interactive REPL mode. + fmt.Printf("%s Interactive mode (Ctrl+C or 'exit' to quit)\n\n", internal.Logo) + interactiveMode(gatewayURL, sessionKey) + return nil +} + +// checkGateway pings the /health endpoint to confirm the gateway is up. +func checkGateway(gatewayURL string) error { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, gatewayURL+"/health", nil) + if err != nil { + return err + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return err + } + resp.Body.Close() + return nil +} + +// sendMessage sends a single message to the gateway /chat endpoint and returns +// the text response. The request timeout is 5 minutes to allow long-running +// tool chains. +func sendMessage(gatewayURL, message, sessionKey string) (string, error) { + body, _ := json.Marshal(agent.ChatAPIRequest{ + Message: message, + Session: sessionKey, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, gatewayURL+"/chat", bytes.NewReader(body)) + if err != nil { + return "", fmt.Errorf("creating request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return "", fmt.Errorf("sending request: %w", err) + } + defer resp.Body.Close() + + var chatResp agent.ChatAPIResponse + if err := json.NewDecoder(resp.Body).Decode(&chatResp); err != nil { + return "", fmt.Errorf("decoding response: %w", err) + } + + if chatResp.Error != "" { + return "", fmt.Errorf("gateway error: %s", chatResp.Error) + } + + return chatResp.Response, nil +} + +// interactiveMode starts a readline-powered REPL that sends each line to the +// gateway and prints the reply. Falls back to simple stdin reading if readline +// is unavailable. +func interactiveMode(gatewayURL, sessionKey string) { + prompt := fmt.Sprintf("%s You: ", internal.Logo) + + rl, err := readline.NewEx(&readline.Config{ + Prompt: prompt, + HistoryFile: filepath.Join(os.TempDir(), ".picoclaw_chat_history"), + HistoryLimit: 200, + InterruptPrompt: "^C", + EOFPrompt: "exit", + }) + if err != nil { + fmt.Printf("Error initializing readline: %v\n", err) + fmt.Println("Falling back to simple input mode...") + simpleInteractiveMode(gatewayURL, sessionKey) + return + } + defer rl.Close() + + for { + line, err := rl.Readline() + if err != nil { + if err == readline.ErrInterrupt || err == io.EOF { + fmt.Println("\nGoodbye!") + return + } + fmt.Printf("Error reading input: %v\n", err) + continue + } + + input := strings.TrimSpace(line) + if input == "" { + continue + } + if input == "exit" || input == "quit" { + fmt.Println("Goodbye!") + return + } + + response, err := sendMessage(gatewayURL, input, sessionKey) + if err != nil { + fmt.Printf("Error: %v\n", err) + continue + } + + fmt.Printf("\n%s %s\n\n", internal.Logo, response) + } +} + +// simpleInteractiveMode is a fallback REPL using plain bufio. +func simpleInteractiveMode(gatewayURL, sessionKey string) { + reader := bufio.NewReader(os.Stdin) + for { + fmt.Printf("%s You: ", internal.Logo) + line, err := reader.ReadString('\n') + if err != nil { + if err == io.EOF { + fmt.Println("\nGoodbye!") + return + } + fmt.Printf("Error reading input: %v\n", err) + continue + } + + input := strings.TrimSpace(line) + if input == "" { + continue + } + if input == "exit" || input == "quit" { + fmt.Println("Goodbye!") + return + } + + response, err := sendMessage(gatewayURL, input, sessionKey) + if err != nil { + fmt.Printf("Error: %v\n", err) + continue + } + + fmt.Printf("\n%s %s\n\n", internal.Logo, response) + } +} + + + diff --git a/cmd/picoclaw/internal/gateway/helpers.go b/cmd/picoclaw/internal/gateway/helpers.go index 747f7d44e..612587225 100644 --- a/cmd/picoclaw/internal/gateway/helpers.go +++ b/cmd/picoclaw/internal/gateway/helpers.go @@ -33,6 +33,7 @@ import ( "github.com/sipeed/picoclaw/pkg/heartbeat" "github.com/sipeed/picoclaw/pkg/logger" "github.com/sipeed/picoclaw/pkg/media" + "github.com/sipeed/picoclaw/pkg/nodes" "github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/state" "github.com/sipeed/picoclaw/pkg/tools" @@ -60,8 +61,31 @@ func gatewayCmd(debug bool) error { } msgBus := bus.NewMessageBus() + + // Prepare the openclaw node WebSocket server. + // The handler is mounted on the shared HTTP server (cfg.Gateway.Host:Port) + // rather than a separate listener, so nodes and the rest-API share one port. + // Nodes (headless Linux, iOS, Android, macOS) connect using the + // openclaw gateway protocol (version 3); picoclaw acts as the server. + var nodeRegistry *nodes.Registry + var nodeSrv *nodes.Server + if cfg.Nodes.Enabled { + nodeRegistry = nodes.NewRegistry() + nodeSrv = nodes.NewServer(nodes.ServerConfig{ + Enabled: true, + Host: cfg.Gateway.Host, // bind address follows gateway (shared server) + Port: cfg.Gateway.Port, // same port as gateway + Token: cfg.Nodes.Token, + }, nodeRegistry) + } + agentLoop := agent.NewAgentLoop(cfg, msgBus, provider) + // Attach node registry to agent loop so NodesTool is available. + if nodeRegistry != nil { + agentLoop.SetNodeRegistry(nodeRegistry) + } + // Print agent startup info fmt.Println("\n📦 Agent Status:") startupInfo := agentLoop.GetStartupInfo() @@ -169,8 +193,17 @@ func gatewayCmd(debug bool) error { fmt.Println("✓ Device event service started") } - // Setup shared HTTP server with health endpoints and webhook handlers + // Setup shared HTTP server with health endpoints and webhook handlers. + // Handlers must be registered before SetupHTTPServer so they are on the shared mux. healthServer := health.NewServer(cfg.Gateway.Host, cfg.Gateway.Port) + healthServer.RegisterHandler("/chat", agentLoop.NewChatHTTPHandler()) + openAIHandler := agentLoop.NewOpenAIChatHandler(agent.OpenAIChatHandlerConfig{ + Token: cfg.Gateway.Token, + }) + healthServer.RegisterHandler("/v1/chat/completions", openAIHandler) + if nodeSrv != nil { + healthServer.RegisterHandler("/", nodeSrv.Handler()) + } addr := fmt.Sprintf("%s:%d", cfg.Gateway.Host, cfg.Gateway.Port) channelManager.SetupHTTPServer(addr, healthServer) @@ -178,8 +211,17 @@ func gatewayCmd(debug bool) error { fmt.Printf("Error starting channels: %v\n", err) return err } - - fmt.Printf("✓ Health endpoints available at http://%s:%d/health and /ready\n", cfg.Gateway.Host, cfg.Gateway.Port) + // HTTP server is started inside channelManager.StartAll() + fmt.Printf("✓ HTTP server started on %s:%d\n", cfg.Gateway.Host, cfg.Gateway.Port) + fmt.Printf(" • /health /ready – health probes\n") + fmt.Printf(" • /chat – chat API (picoclaw chat)\n") + fmt.Printf(" • /v1/chat/completions – OpenAI-compatible chat completions\n") + if cfg.Gateway.Token != "" { + fmt.Printf(" ↳ bearer token auth enabled\n") + } + if nodeSrv != nil { + fmt.Printf(" • / – node WebSocket (openclaw protocol v3)\n") + } go agentLoop.Run(ctx) diff --git a/cmd/picoclaw/main.go b/cmd/picoclaw/main.go index 6db69c990..f07be3753 100644 --- a/cmd/picoclaw/main.go +++ b/cmd/picoclaw/main.go @@ -15,6 +15,7 @@ import ( "github.com/sipeed/picoclaw/cmd/picoclaw/internal" "github.com/sipeed/picoclaw/cmd/picoclaw/internal/agent" "github.com/sipeed/picoclaw/cmd/picoclaw/internal/auth" + "github.com/sipeed/picoclaw/cmd/picoclaw/internal/chat" "github.com/sipeed/picoclaw/cmd/picoclaw/internal/cron" "github.com/sipeed/picoclaw/cmd/picoclaw/internal/gateway" "github.com/sipeed/picoclaw/cmd/picoclaw/internal/migrate" @@ -36,6 +37,7 @@ func NewPicoclawCommand() *cobra.Command { cmd.AddCommand( onboard.NewOnboardCommand(), agent.NewAgentCommand(), + chat.NewChatCommand(), auth.NewAuthCommand(), gateway.NewGatewayCommand(), status.NewStatusCommand(), diff --git a/config/config.example.json b/config/config.example.json index d885ef94b..c39bbd043 100644 --- a/config/config.example.json +++ b/config/config.example.json @@ -260,5 +260,8 @@ "gateway": { "host": "127.0.0.1", "port": 18790 + }, + "nodes": { + "enabled": false } } diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 00b0f096a..055b3a80d 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -11,6 +11,7 @@ import ( "encoding/json" "errors" "fmt" + "net/http" "path/filepath" "strings" "sync" @@ -24,6 +25,7 @@ import ( "github.com/sipeed/picoclaw/pkg/constants" "github.com/sipeed/picoclaw/pkg/logger" "github.com/sipeed/picoclaw/pkg/media" + "github.com/sipeed/picoclaw/pkg/nodes" "github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/routing" "github.com/sipeed/picoclaw/pkg/skills" @@ -42,6 +44,7 @@ type AgentLoop struct { fallback *providers.FallbackChain channelManager *channels.Manager mediaStore media.MediaStore + nodeRegistry *nodes.Registry // optional; non-nil when nodes server is enabled } // processOptions configures how a message is processed @@ -85,6 +88,20 @@ func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers } } +// SetNodeRegistry attaches a node Registry to the agent loop so that the +// NodesTool becomes available. Call this after NewAgentLoop but before Run. +func (al *AgentLoop) SetNodeRegistry(nodeReg *nodes.Registry) { + al.nodeRegistry = nodeReg + // Register NodesTool with every agent instance. + for _, agentID := range al.registry.ListAgentIDs() { + agent, ok := al.registry.GetAgent(agentID) + if !ok { + continue + } + agent.Tools.Register(tools.NewNodesTool(nodeReg)) + } +} + // registerSharedTools registers tools that are shared across all agents (web, message, spawn). func registerSharedTools( cfg *config.Config, @@ -1350,3 +1367,61 @@ func extractParentPeer(msg bus.InboundMessage) *routing.RoutePeer { } return &routing.RoutePeer{Kind: parentKind, ID: parentID} } + +// ChatAPIRequest is the JSON body for POST /chat. +type ChatAPIRequest struct { + Message string `json:"message"` + Session string `json:"session"` +} + +// ChatAPIResponse is the JSON response for POST /chat. +type ChatAPIResponse struct { + Response string `json:"response"` + Session string `json:"session"` + Error string `json:"error,omitempty"` +} + +// NewChatHTTPHandler returns an http.HandlerFunc that exposes a simple +// synchronous chat endpoint: POST /chat {"message":"...", "session":"..."}. +// The handler calls ProcessDirect on the agent loop and waits for the reply. +func (al *AgentLoop) NewChatHTTPHandler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + var req ChatAPIRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + json.NewEncoder(w).Encode(ChatAPIResponse{Error: "invalid request body: " + err.Error()}) + return + } + + if req.Message == "" { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + json.NewEncoder(w).Encode(ChatAPIResponse{Error: "message is required"}) + return + } + + sessionKey := req.Session + if sessionKey == "" { + sessionKey = "cli:default" + } + + response, err := al.ProcessDirect(r.Context(), req.Message, sessionKey) + w.Header().Set("Content-Type", "application/json") + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + json.NewEncoder(w).Encode(ChatAPIResponse{Error: err.Error()}) + return + } + + json.NewEncoder(w).Encode(ChatAPIResponse{ + Response: response, + Session: sessionKey, + }) + } +} diff --git a/pkg/agent/openai_http.go b/pkg/agent/openai_http.go new file mode 100644 index 000000000..11a287e7d --- /dev/null +++ b/pkg/agent/openai_http.go @@ -0,0 +1,304 @@ +// PicoClaw - Ultra-lightweight personal AI agent +// License: MIT +// +// Copyright (c) 2026 PicoClaw contributors + +package agent + +import ( + "encoding/json" + "fmt" + "math/rand" + "net/http" + "strings" + "time" +) + +// openAIChatMessage is a single message in an OpenAI chat completion request. +type openAIChatMessage struct { + Role string `json:"role"` + Content any `json:"content"` // string or []contentPart + Name string `json:"name,omitempty"` +} + +// openAIChatRequest is the JSON body for POST /v1/chat/completions. +type openAIChatRequest struct { + Model string `json:"model"` + Messages []openAIChatMessage `json:"messages"` + Stream bool `json:"stream"` + User string `json:"user,omitempty"` +} + +// extractOpenAITextContent extracts plain text from an OpenAI message content +// field, which may be a plain string or an array of typed content parts. +func extractOpenAITextContent(content any) string { + switch v := content.(type) { + case string: + return v + case []any: + var parts []string + for _, part := range v { + m, ok := part.(map[string]any) + if !ok { + continue + } + t, _ := m["type"].(string) + switch t { + case "text", "input_text": + if text, ok := m["text"].(string); ok { + parts = append(parts, text) + } + } + } + return strings.Join(parts, "\n") + } + return "" +} + +// buildPromptFromOpenAIMessages extracts the latest user/tool message and an +// optional extra system prompt from an OpenAI-format messages array. +// +// The approach mirrors openclaw's buildAgentPrompt: +// - Collect all system/developer messages into a combined extra system prompt. +// - Find the last user or tool message to use as the actual query. +func buildPromptFromOpenAIMessages(messages []openAIChatMessage) (userMessage, extraSystemPrompt string) { + var systemParts []string + + // Last user/tool message index (search backwards). + lastUserIdx := -1 + for i := len(messages) - 1; i >= 0; i-- { + role := strings.TrimSpace(messages[i].Role) + if role == "user" || role == "tool" || role == "function" { + content := strings.TrimSpace(extractOpenAITextContent(messages[i].Content)) + if content != "" { + lastUserIdx = i + break + } + } + } + + if lastUserIdx >= 0 { + userMessage = strings.TrimSpace(extractOpenAITextContent(messages[lastUserIdx].Content)) + } + + for _, msg := range messages { + role := strings.TrimSpace(msg.Role) + if role == "system" || role == "developer" { + content := strings.TrimSpace(extractOpenAITextContent(msg.Content)) + if content != "" { + systemParts = append(systemParts, content) + } + } + } + extraSystemPrompt = strings.Join(systemParts, "\n\n") + return userMessage, extraSystemPrompt +} + +// writeOpenAISSE writes one SSE event and flushes. +func writeOpenAISSE(w http.ResponseWriter, data any) { + b, _ := json.Marshal(data) + fmt.Fprintf(w, "data: %s\n\n", b) + if f, ok := w.(http.Flusher); ok { + f.Flush() + } +} + +// writeOpenAIDone writes the SSE [DONE] terminator. +func writeOpenAIDone(w http.ResponseWriter) { + fmt.Fprint(w, "data: [DONE]\n\n") + if f, ok := w.(http.Flusher); ok { + f.Flush() + } +} + +// newOpenAIRunID generates a unique run/completion ID. +func newOpenAIRunID() string { + //nolint:gosec // non-crypto random is fine for a run ID + return fmt.Sprintf("chatcmpl-%d%d", time.Now().UnixNano(), rand.Int63n(100000)) +} + +// OpenAIChatHandlerConfig configures the /v1/chat/completions handler. +type OpenAIChatHandlerConfig struct { + // Token is an optional shared secret. When non-empty every request must + // carry "Authorization: Bearer ". Leave empty to allow all. + Token string +} + +// NewOpenAIChatHandler returns an http.HandlerFunc that exposes an +// OpenAI-compatible POST /v1/chat/completions endpoint. +// +// Features: +// - Non-streaming: standard JSON response (chat.completion object). +// - Streaming: SSE response (stream: true) — runs the agent synchronously +// then writes the full reply as consecutive SSE chunks. +// - Session routing: honour X-Picoclaw-Session-Key header; fall back to a +// key derived from the OpenAI "user" field; otherwise a per-request key. +// - Agent selection via the "model" field: "picoclaw:" or +// "agent:" selects a named agent; any other value uses default. +func (al *AgentLoop) NewOpenAIChatHandler(cfg OpenAIChatHandlerConfig) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + // Only accept POST on the exact path. + if r.URL.Path != "/v1/chat/completions" { + http.NotFound(w, r) + return + } + if r.Method != http.MethodPost { + w.Header().Set("Allow", "POST") + sendOpenAIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "Method not allowed") + return + } + + // ── Authentication ───────────────────────────────────────────────── + if cfg.Token != "" { + auth := strings.TrimSpace(r.Header.Get("Authorization")) + bearer := strings.TrimPrefix(auth, "Bearer ") + if bearer != cfg.Token { + sendOpenAIError(w, http.StatusUnauthorized, "authentication_error", "Unauthorized") + return + } + } + + // ── Parse body ────────────────────────────────────────────────────── + var req openAIChatRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + sendOpenAIError(w, http.StatusBadRequest, "invalid_request_error", + "invalid request body: "+err.Error()) + return + } + + userMessage, _ := buildPromptFromOpenAIMessages(req.Messages) + if userMessage == "" { + sendOpenAIError(w, http.StatusBadRequest, "invalid_request_error", + "Missing user message in `messages`.") + return + } + + // ── Session key ───────────────────────────────────────────────────── + sessionKey := strings.TrimSpace(r.Header.Get("X-Picoclaw-Session-Key")) + if sessionKey == "" && req.User != "" { + // Stable key per OpenAI "user" value. + sessionKey = "openai:user:" + req.User + } + if sessionKey == "" { + // Stateless: unique key per request (no history sharing). + sessionKey = fmt.Sprintf("openai:req:%d", time.Now().UnixNano()) + } + + // ── Model / run metadata ──────────────────────────────────────────── + model := req.Model + if model == "" { + model = "picoclaw" + } + runID := newOpenAIRunID() + created := int64(time.Now().Unix()) + + // ── Run the agent (synchronous) ──────────────────────────────────── + // Use "openai" as the channel so routing/history work naturally. + response, err := al.ProcessDirectWithChannel(r.Context(), userMessage, sessionKey, "openai", "api") + + // ── Streaming response (SSE) ──────────────────────────────────────── + if req.Stream { + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + w.WriteHeader(http.StatusOK) + + if err != nil { + // Emit the error as content and finish cleanly. + writeOpenAISSE(w, buildChunk(runID, model, created, + map[string]any{"role": "assistant", "content": fmt.Sprintf("Error: %v", err)}, + ptrStr("stop"))) + writeOpenAIDone(w) + return + } + + // 1) role chunk + writeOpenAISSE(w, buildChunk(runID, model, created, + map[string]any{"role": "assistant"}, nil)) + + // 2) content chunk + writeOpenAISSE(w, buildChunk(runID, model, created, + map[string]any{"content": response}, nil)) + + // 3) finish chunk + writeOpenAISSE(w, buildChunk(runID, model, created, + map[string]any{}, ptrStr("stop"))) + + writeOpenAIDone(w) + return + } + + // ── Non-streaming JSON response ───────────────────────────────────── + w.Header().Set("Content-Type", "application/json") + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": err.Error(), + "type": "api_error", + }, + }) + return + } + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(map[string]any{ + "id": runID, + "object": "chat.completion", + "created": created, + "model": model, + "choices": []any{ + map[string]any{ + "index": 0, + "message": map[string]any{ + "role": "assistant", + "content": response, + }, + "finish_reason": "stop", + }, + }, + "usage": map[string]any{ + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0, + }, + }) + } +} + +// ── helpers ────────────────────────────────────────────────────────────────── + +func sendOpenAIError(w http.ResponseWriter, status int, errType, msg string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": msg, + "type": errType, + }, + }) +} + +// buildChunk constructs an SSE chat.completion.chunk payload. +func buildChunk(id, model string, created int64, delta map[string]any, finishReason *string) map[string]any { + choice := map[string]any{ + "index": 0, + "delta": delta, + } + if finishReason != nil { + choice["finish_reason"] = *finishReason + } else { + choice["finish_reason"] = nil + } + return map[string]any{ + "id": id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": []any{choice}, + } +} + +func ptrStr(s string) *string { return &s } + + diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 155e50b39..faa0a531c 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -312,12 +312,11 @@ func (m *Manager) StartAll(ctx context.Context) error { defer m.mu.Unlock() if len(m.channels) == 0 { - logger.WarnC("channels", "No channels enabled") - return errors.New("no channels enabled") + logger.InfoC("channels", "No chat channels enabled; gateway will serve HTTP API only") + } else { + logger.InfoC("channels", "Starting all channels") } - logger.InfoC("channels", "Starting all channels") - dispatchCtx, cancel := context.WithCancel(ctx) m.dispatchTask = &asyncTask{cancel: cancel} diff --git a/pkg/config/config.go b/pkg/config/config.go index 55d0cfb2c..e25ee5c86 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -55,6 +55,7 @@ type Config struct { Providers ProvidersConfig `json:"providers,omitempty"` ModelList []ModelConfig `json:"model_list"` // New model-centric provider configuration Gateway GatewayConfig `json:"gateway"` + Nodes NodesConfig `json:"nodes"` Tools ToolsConfig `json:"tools"` Heartbeat HeartbeatConfig `json:"heartbeat"` Devices DevicesConfig `json:"devices"` @@ -488,8 +489,34 @@ func (c *ModelConfig) Validate() error { } type GatewayConfig struct { - Host string `json:"host" env:"PICOCLAW_GATEWAY_HOST"` - Port int `json:"port" env:"PICOCLAW_GATEWAY_PORT"` + Host string `json:"host" env:"PICOCLAW_GATEWAY_HOST"` + Port int `json:"port" env:"PICOCLAW_GATEWAY_PORT"` + // Token is an optional shared secret for the HTTP API. + // If non-empty, all /v1/chat/completions requests must provide + // "Authorization: Bearer ". Leave empty to allow all requests. + Token string `json:"token,omitempty" env:"PICOCLAW_GATEWAY_TOKEN"` +} + +// NodesConfig configures the openclaw node WebSocket server. +// Nodes (headless Linux, iOS, Android, macOS) connect to picoclaw using +// the openclaw gateway protocol (version 3). picoclaw acts as the server; +// nodes are clients with role="node". +type NodesConfig struct { + // Enabled controls whether the node WebSocket server starts. + Enabled bool `json:"enabled" env:"PICOCLAW_NODES_ENABLED"` + + // Host is the bind address (default "0.0.0.0"). + Host string `json:"host" env:"PICOCLAW_NODES_HOST"` + + // Port is the TCP listen port (default 18790). + // Note: 18789 is the standard openclaw gateway port; 18790 avoids collisions + // when picoclaw runs alongside openclaw. + Port int `json:"port" env:"PICOCLAW_NODES_PORT"` + + // Token is an optional shared secret for authentication. + // If non-empty, connecting nodes must provide it as auth.token. + // Leave empty to allow all connections (suitable for trusted networks). + Token string `json:"token,omitempty" env:"PICOCLAW_NODES_TOKEN"` } type BraveConfig struct { diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go index 44f4de7e9..0ec602a05 100644 --- a/pkg/config/defaults.go +++ b/pkg/config/defaults.go @@ -308,6 +308,12 @@ func DefaultConfig() *Config { Host: "127.0.0.1", Port: 18790, }, + Nodes: NodesConfig{ + Enabled: false, + Host: "0.0.0.0", + Port: 18790, // same port as gateway; node WS is served on the shared HTTP server + Token: "", + }, Tools: ToolsConfig{ MediaCleanup: MediaCleanupConfig{ Enabled: true, diff --git a/pkg/health/server.go b/pkg/health/server.go index 5609ebdf6..29aa7162f 100644 --- a/pkg/health/server.go +++ b/pkg/health/server.go @@ -12,6 +12,7 @@ import ( type Server struct { server *http.Server + mux *http.ServeMux mu sync.RWMutex ready bool checks map[string]Check @@ -34,6 +35,7 @@ type StatusResponse struct { func NewServer(host string, port int) *Server { mux := http.NewServeMux() s := &Server{ + mux: mux, ready: false, checks: make(map[string]Check), startTime: time.Now(), @@ -47,12 +49,18 @@ func NewServer(host string, port int) *Server { Addr: addr, Handler: mux, ReadTimeout: 5 * time.Second, - WriteTimeout: 5 * time.Second, + WriteTimeout: 30 * time.Second, // allow longer writes for chat responses } return s } +// RegisterHandler registers an additional HTTP handler on the server's mux. +// Must be called before Start/StartContext. +func (s *Server) RegisterHandler(pattern string, handler http.HandlerFunc) { + s.mux.HandleFunc(pattern, handler) +} + func (s *Server) Start() error { s.mu.Lock() s.ready = true @@ -155,11 +163,14 @@ func (s *Server) readyHandler(w http.ResponseWriter, r *http.Request) { }) } -// RegisterOnMux registers /health and /ready handlers onto the given mux. -// This allows the health endpoints to be served by a shared HTTP server. +// RegisterOnMux registers health endpoints and any handlers added via RegisterHandler +// onto the given mux. This allows the health server to be served by a shared HTTP server. +// Handlers added via RegisterHandler (e.g. /chat, /v1/chat/completions) are served via +// the catch-all "/" which forwards to the health server's mux. func (s *Server) RegisterOnMux(mux *http.ServeMux) { mux.HandleFunc("/health", s.healthHandler) mux.HandleFunc("/ready", s.readyHandler) + mux.Handle("/", s.mux) // catch-all for RegisterHandler routes (/chat, /v1/chat/completions, etc.) } func statusString(ok bool) string { diff --git a/pkg/nodes/protocol.go b/pkg/nodes/protocol.go new file mode 100644 index 000000000..887d16dbd --- /dev/null +++ b/pkg/nodes/protocol.go @@ -0,0 +1,264 @@ +// Package nodes implements the openclaw gateway WebSocket protocol server, +// allowing openclaw nodes (headless Linux, iOS, Android, macOS) to connect +// directly to picoclaw and expose their capabilities. +package nodes + +// ProtocolVersion is the openclaw gateway protocol version this server supports. +const ProtocolVersion = 3 + +// TickIntervalMs is how often (ms) the server sends a tick keepalive event. +// Must match openclaw's TICK_INTERVAL_MS = 30_000. +// The node client's watchdog closes the connection if no tick arrives within +// tickIntervalMs * 2 (i.e. 60 s), so we must send one every 30 s. +const TickIntervalMs = 30_000 + +// MaxPayloadBytes is the maximum allowed incoming frame size (512 KB). +const MaxPayloadBytes = 512 * 1024 + +// MaxBufferedBytes is the per-connection send buffer limit (1.5 MB). +const MaxBufferedBytes = 1536 * 1024 + +// Frame type constants matching the openclaw gateway protocol. +const ( + FrameTypeReq = "req" + FrameTypeRes = "res" + FrameTypeEvent = "event" +) + +// Method constants used in the protocol. +const ( + MethodConnect = "connect" + MethodNodeInvokeResult = "node.invoke.result" + MethodNodeEvent = "node.event" +) + +// Event constants sent from server to node. +const ( + EventConnectChallenge = "connect.challenge" + EventNodeInvokeRequest = "node.invoke.request" +) + +// Error code constants matching openclaw's ErrorCodes. +const ( + ErrCodeInvalidRequest = "INVALID_REQUEST" + ErrCodeNotPaired = "NOT_PAIRED" + ErrCodeUnavailable = "UNAVAILABLE" + ErrCodeTimeout = "TIMEOUT" +) + +// ReqFrame is a JSON-RPC style request frame sent by clients. +// Example: {"type":"req","id":"","method":"connect","params":{...}} +type ReqFrame struct { + Type string `json:"type"` + ID string `json:"id"` + Method string `json:"method"` + Params map[string]any `json:"params,omitempty"` +} + +// ResFrame is a response frame sent by the server. +// Example: {"type":"res","id":"","ok":true,"payload":{...}} +type ResFrame struct { + Type string `json:"type"` + ID string `json:"id"` + Ok bool `json:"ok"` + Payload any `json:"payload,omitempty"` + Error *ErrBody `json:"error,omitempty"` +} + +// EventFrame is an event frame sent by the server to nodes. +// Example: {"type":"event","event":"node.invoke.request","payload":{...}} +type EventFrame struct { + Type string `json:"type"` + Event string `json:"event"` + Payload any `json:"payload,omitempty"` +} + +// ErrBody is the error body in a response frame. +type ErrBody struct { + Code string `json:"code"` + Message string `json:"message"` +} + +// ConnectChallengePayload is the payload of the connect.challenge event. +// The server sends this immediately upon connection. +type ConnectChallengePayload struct { + Nonce string `json:"nonce"` + Ts int64 `json:"ts"` +} + +// ConnectParams is the parameters of the connect request sent by nodes. +// Matches the openclaw gateway protocol ConnectParams structure. +type ConnectParams struct { + // Protocol version range supported by the client. + MinProtocol int `json:"minProtocol"` + MaxProtocol int `json:"maxProtocol"` + + // Role: "node" or "operator" + Role string `json:"role"` + + // Scopes requested by the client. + Scopes []string `json:"scopes,omitempty"` + + // Client identity information. + Client ConnectClientInfo `json:"client"` + + // Optional auth credentials. + Auth *ConnectAuth `json:"auth,omitempty"` + + // Device identity (for nodes with a persistent device ID / key pair). + Device *ConnectDevice `json:"device,omitempty"` + + // Commands this node can handle (only valid for role=node). + Commands []string `json:"commands,omitempty"` + + // Capabilities this node exposes (e.g. "system", "camera"). + Caps []string `json:"caps,omitempty"` + + // Path env from the node host (for headless Linux nodes). + PathEnv string `json:"pathEnv,omitempty"` +} + +// ConnectClientInfo holds identifying information about the connecting client. +type ConnectClientInfo struct { + // Unique stable client identifier (e.g. "openclaw-node-host"). + ID string `json:"id"` + + // Human-readable display name (e.g. "My Raspberry Pi"). + DisplayName string `json:"displayName,omitempty"` + + // Platform: "linux", "darwin", "win32", "ios", "android". + Platform string `json:"platform,omitempty"` + + // Device family: "phone", "tablet", "desktop", "server", etc. + DeviceFamily string `json:"deviceFamily,omitempty"` + + // Hardware model identifier (e.g. "iPhone16,2"). + ModelIdentifier string `json:"modelIdentifier,omitempty"` + + // Client mode: "node", "operator", etc. + Mode string `json:"mode,omitempty"` + + // Client version string. + Version string `json:"version,omitempty"` + + // Instance ID: stable identifier for this running instance. + InstanceID string `json:"instanceId,omitempty"` +} + +// ConnectAuth holds authentication credentials. +type ConnectAuth struct { + // Shared token (set in gateway / picoclaw config). + Token string `json:"token,omitempty"` + + // Password auth (alternative to token). + Password string `json:"password,omitempty"` +} + +// ConnectDevice holds the cryptographic device identity of the connecting node. +// picoclaw uses this for pairing tracking; signature verification is optional. +type ConnectDevice struct { + // Stable device ID derived from the public key. + ID string `json:"id"` + + // Base64url-encoded Ed25519 public key. + PublicKey string `json:"publicKey"` + + // Base64url-encoded signature over the auth payload. + Signature string `json:"signature,omitempty"` + + // Timestamp (ms since epoch) when the signature was created. + SignedAt int64 `json:"signedAt"` + + // Nonce from the connect.challenge event (for replay protection). + Nonce string `json:"nonce,omitempty"` +} + +// HelloOkPayload is the payload returned in the successful connect response. +type HelloOkPayload struct { + Type string `json:"type"` + Protocol int `json:"protocol"` + Server HelloOkServer `json:"server"` + Features HelloOkFeatures `json:"features"` + // Policy tells the client the tick interval so its watchdog is calibrated. + Policy HelloOkPolicy `json:"policy"` +} + +// HelloOkPolicy carries server-enforced connection policy sent in hello-ok. +type HelloOkPolicy struct { + MaxPayload int `json:"maxPayload"` + MaxBufferedBytes int `json:"maxBufferedBytes"` + // TickIntervalMs is how often the server will send tick events (ms). + // The node watchdog disconnects if no tick arrives within 2× this value. + TickIntervalMs int `json:"tickIntervalMs"` +} + +// HelloOkServer holds server identity info in the hello-ok response. +type HelloOkServer struct { + Version string `json:"version"` + Host string `json:"host"` + ConnID string `json:"connId"` +} + +// HelloOkFeatures lists supported methods and events in the hello-ok response. +type HelloOkFeatures struct { + Methods []string `json:"methods"` + Events []string `json:"events"` +} + +// NodeInvokeRequestPayload is the payload of the node.invoke.request event. +// Server sends this to the node to invoke a command. +type NodeInvokeRequestPayload struct { + // Unique request ID for correlating the response. + ID string `json:"id"` + + // The node ID the request is destined for. + NodeID string `json:"nodeId"` + + // The command to invoke on the node (e.g. "system.run"). + Command string `json:"command"` + + // JSON-encoded parameters for the command (may be null). + ParamsJSON *string `json:"paramsJSON,omitempty"` + + // Optional timeout for the command in milliseconds. + TimeoutMs *int `json:"timeoutMs,omitempty"` + + // Optional idempotency key for deduplication. + IdempotencyKey *string `json:"idempotencyKey,omitempty"` +} + +// NodeInvokeResultParams is the parameters of the node.invoke.result request. +// The node sends this back to the server after executing a command. +type NodeInvokeResultParams struct { + // Matches the ID from NodeInvokeRequestPayload. + ID string `json:"id"` + + // The node ID (must match the invoking node's ID). + NodeID string `json:"nodeId"` + + // Whether the invocation succeeded. + Ok bool `json:"ok"` + + // Structured result payload (alternative to PayloadJSON). + Payload any `json:"payload,omitempty"` + + // JSON-encoded result payload (preferred over Payload for large results). + PayloadJSON *string `json:"payloadJSON,omitempty"` + + // Error info if Ok is false. + Error *InvokeError `json:"error,omitempty"` +} + +// InvokeError represents an error from a node invoke. +type InvokeError struct { + Code string `json:"code,omitempty"` + Message string `json:"message,omitempty"` +} + +// NodeEventParams is the parameters of the node.event request. +// Nodes send this to forward events to the server. +type NodeEventParams struct { + Event string `json:"event"` + PayloadJSON *string `json:"payloadJSON,omitempty"` + Payload any `json:"payload,omitempty"` +} diff --git a/pkg/nodes/registry.go b/pkg/nodes/registry.go new file mode 100644 index 000000000..18144f4d0 --- /dev/null +++ b/pkg/nodes/registry.go @@ -0,0 +1,450 @@ +package nodes + +import ( + "encoding/json" + "fmt" + "sync" + "time" + + "github.com/google/uuid" + "github.com/gorilla/websocket" +) + +// NodeSession holds the state of a connected node. +type NodeSession struct { + // NodeID is the stable identifier for this node (device.id or client.id). + NodeID string + + // ConnID is the unique connection identifier for this session. + ConnID string + + // DisplayName is the human-readable name of the node. + DisplayName string + + // Platform is the OS platform (linux, darwin, ios, android, etc.). + Platform string + + // DeviceFamily is the device category (phone, tablet, desktop, server, etc.). + DeviceFamily string + + // ModelIdentifier is the hardware model identifier. + ModelIdentifier string + + // Version is the client version string. + Version string + + // Commands lists the commands this node has declared it can handle. + Commands []string + + // Caps lists the capability groups this node exposes. + Caps []string + + // RemoteIP is the remote IP address of the connection (may be empty). + RemoteIP string + + // ConnectedAtMs is the unix millisecond timestamp when the node connected. + ConnectedAtMs int64 + + // conn is the underlying WebSocket connection. + conn *websocket.Conn + // mu protects conn sends. + mu sync.Mutex +} + +// SendEvent sends an event frame to this node. +// Returns an error if the send fails. +func (s *NodeSession) SendEvent(event string, payload any) error { + frame := EventFrame{ + Type: FrameTypeEvent, + Event: event, + Payload: payload, + } + data, err := json.Marshal(frame) + if err != nil { + return fmt.Errorf("marshal event: %w", err) + } + s.mu.Lock() + defer s.mu.Unlock() + return s.conn.WriteMessage(websocket.TextMessage, data) +} + +// HasCommand returns true if this node declared the given command. +func (s *NodeSession) HasCommand(cmd string) bool { + for _, c := range s.Commands { + if c == cmd { + return true + } + } + return false +} + +// pendingInvoke tracks an in-flight node invoke waiting for a result. +type pendingInvoke struct { + nodeID string + command string + resolve chan InvokeResult +} + +// InvokeResult holds the result of a node invoke operation. +type InvokeResult struct { + Ok bool + Payload any + PayloadJSON *string + Error *InvokeError +} + +// PairingRequest represents a node waiting for manual approval. +// When the server is configured with PairingMode="manual", new node connections +// are held here until an operator approves or rejects them. +type PairingRequest struct { + // RequestID is a unique ID for this pairing request. + RequestID string + + // NodeID is the connecting node's ID. + NodeID string + + // DisplayName is the connecting node's human-readable name. + DisplayName string + + // Platform is the OS of the connecting node. + Platform string + + // DeviceFamily is the device family of the connecting node. + DeviceFamily string + + // RemoteIP is the remote address of the connecting node. + RemoteIP string + + // RequestedAtMs is when the pairing request was created. + RequestedAtMs int64 + + // resolved is closed when the request is approved or rejected. + resolved chan bool +} + +// Registry tracks all connected nodes and manages in-flight invocations. +type Registry struct { + mu sync.RWMutex + nodesByID map[string]*NodeSession + nodesByConn map[string]string // connID -> nodeID + pendingByID map[string]*pendingInvoke + + pairingMu sync.RWMutex + pairingByID map[string]*PairingRequest // requestID -> PairingRequest + pairingByNodeID map[string]string // nodeID -> requestID +} + +// NewRegistry creates a new Registry. +func NewRegistry() *Registry { + return &Registry{ + nodesByID: make(map[string]*NodeSession), + nodesByConn: make(map[string]string), + pendingByID: make(map[string]*pendingInvoke), + pairingByID: make(map[string]*PairingRequest), + pairingByNodeID: make(map[string]string), + } +} + +// Register adds a new node session to the registry. +func (r *Registry) Register(connID string, conn *websocket.Conn, params *ConnectParams, remoteIP string) *NodeSession { + // Determine the stable node ID: prefer device.id, fall back to client.instanceId then client.id. + nodeID := "" + if params.Device != nil && params.Device.ID != "" { + nodeID = params.Device.ID + } else if params.Client.InstanceID != "" { + nodeID = params.Client.InstanceID + } else { + nodeID = params.Client.ID + } + + session := &NodeSession{ + NodeID: nodeID, + ConnID: connID, + DisplayName: params.Client.DisplayName, + Platform: params.Client.Platform, + DeviceFamily: params.Client.DeviceFamily, + ModelIdentifier: params.Client.ModelIdentifier, + Version: params.Client.Version, + Commands: params.Commands, + Caps: params.Caps, + RemoteIP: remoteIP, + ConnectedAtMs: time.Now().UnixMilli(), + conn: conn, + } + + r.mu.Lock() + r.nodesByID[nodeID] = session + r.nodesByConn[connID] = nodeID + r.mu.Unlock() + + return session +} + +// Unregister removes a node by connection ID. Returns the nodeID if found. +func (r *Registry) Unregister(connID string) string { + r.mu.Lock() + defer r.mu.Unlock() + + nodeID, ok := r.nodesByConn[connID] + if !ok { + return "" + } + delete(r.nodesByConn, connID) + delete(r.nodesByID, nodeID) + + // Fail any pending invokes for this node. + for id, pending := range r.pendingByID { + if pending.nodeID != nodeID { + continue + } + pending.resolve <- InvokeResult{ + Ok: false, + Error: &InvokeError{Code: ErrCodeUnavailable, Message: "node disconnected"}, + } + delete(r.pendingByID, id) + } + + return nodeID +} + +// Get returns the NodeSession for the given nodeID, or nil if not found. +func (r *Registry) Get(nodeID string) *NodeSession { + r.mu.RLock() + defer r.mu.RUnlock() + return r.nodesByID[nodeID] +} + +// GetByConn returns the NodeSession for the given connection ID, or nil. +func (r *Registry) GetByConn(connID string) *NodeSession { + r.mu.RLock() + nodeID := r.nodesByConn[connID] + r.mu.RUnlock() + if nodeID == "" { + return nil + } + return r.Get(nodeID) +} + +// ListConnected returns a snapshot of all currently connected node sessions. +func (r *Registry) ListConnected() []*NodeSession { + r.mu.RLock() + defer r.mu.RUnlock() + result := make([]*NodeSession, 0, len(r.nodesByID)) + for _, s := range r.nodesByID { + result = append(result, s) + } + return result +} + +// Invoke sends a command to a node and waits for the result. +// The server sends a node.invoke.request event to the node; the node +// executes the command and replies with a node.invoke.result request. +func (r *Registry) Invoke(nodeID, command string, params any, timeoutMs int) InvokeResult { + node := r.Get(nodeID) + if node == nil { + return InvokeResult{ + Ok: false, + Error: &InvokeError{Code: ErrCodeUnavailable, Message: "node not connected"}, + } + } + + requestID := uuid.New().String() + ch := make(chan InvokeResult, 1) + + pending := &pendingInvoke{ + nodeID: nodeID, + command: command, + resolve: ch, + } + + r.mu.Lock() + r.pendingByID[requestID] = pending + r.mu.Unlock() + + // Build the invoke request payload. + var paramsJSONPtr *string + if params != nil { + data, err := json.Marshal(params) + if err == nil { + s := string(data) + paramsJSONPtr = &s + } + } + if timeoutMs <= 0 { + timeoutMs = 30_000 + } + payload := NodeInvokeRequestPayload{ + ID: requestID, + NodeID: nodeID, + Command: command, + ParamsJSON: paramsJSONPtr, + TimeoutMs: &timeoutMs, + } + + // Send the event to the node. + if err := node.SendEvent(EventNodeInvokeRequest, payload); err != nil { + r.mu.Lock() + delete(r.pendingByID, requestID) + r.mu.Unlock() + return InvokeResult{ + Ok: false, + Error: &InvokeError{Code: ErrCodeUnavailable, Message: "failed to send invoke to node"}, + } + } + + // Wait for the result with timeout. + timer := time.NewTimer(time.Duration(timeoutMs) * time.Millisecond) + defer timer.Stop() + + select { + case result := <-ch: + return result + case <-timer.C: + r.mu.Lock() + delete(r.pendingByID, requestID) + r.mu.Unlock() + return InvokeResult{ + Ok: false, + Error: &InvokeError{Code: ErrCodeTimeout, Message: "node invoke timed out"}, + } + } +} + +// HandleInvokeResult processes a node.invoke.result response from a node. +// Returns true if the result was matched to a pending invoke. +func (r *Registry) HandleInvokeResult(p *NodeInvokeResultParams) bool { + r.mu.Lock() + pending, ok := r.pendingByID[p.ID] + if !ok { + r.mu.Unlock() + return false + } + if pending.nodeID != p.NodeID { + r.mu.Unlock() + return false + } + delete(r.pendingByID, p.ID) + r.mu.Unlock() + + result := InvokeResult{ + Ok: p.Ok, + Payload: p.Payload, + PayloadJSON: p.PayloadJSON, + Error: p.Error, + } + pending.resolve <- result + return true +} + +// SendEvent sends an event to the node with the given nodeID. +func (r *Registry) SendEvent(nodeID, event string, payload any) bool { + node := r.Get(nodeID) + if node == nil { + return false + } + return node.SendEvent(event, payload) == nil +} + +// Disconnect forcibly closes the WebSocket connection for the given nodeID. +// This is used to reject a connected node. +func (r *Registry) Disconnect(nodeID string) bool { + node := r.Get(nodeID) + if node == nil { + return false + } + node.mu.Lock() + defer node.mu.Unlock() + _ = node.conn.Close() + return true +} + +// --- Pairing queue (for manual pairing mode) --- + +// AddPairingRequest adds a node to the pending pairing queue and blocks until +// approved or rejected (or the channel is closed). Returns true if approved. +func (r *Registry) AddPairingRequest(nodeID, displayName, platform, deviceFamily, remoteIP string) *PairingRequest { + req := &PairingRequest{ + RequestID: uuid.New().String(), + NodeID: nodeID, + DisplayName: displayName, + Platform: platform, + DeviceFamily: deviceFamily, + RemoteIP: remoteIP, + RequestedAtMs: time.Now().UnixMilli(), + resolved: make(chan bool, 1), + } + r.pairingMu.Lock() + r.pairingByID[req.RequestID] = req + r.pairingByNodeID[nodeID] = req.RequestID + r.pairingMu.Unlock() + return req +} + +// WaitPairingDecision blocks until the pairing request is resolved. +// Returns true if approved, false if rejected. +func (r *Registry) WaitPairingDecision(req *PairingRequest) bool { + return <-req.resolved +} + +// ListPairingRequests returns all pending pairing requests. +func (r *Registry) ListPairingRequests() []*PairingRequest { + r.pairingMu.RLock() + defer r.pairingMu.RUnlock() + result := make([]*PairingRequest, 0, len(r.pairingByID)) + for _, req := range r.pairingByID { + result = append(result, req) + } + return result +} + +// ApprovePairing approves the pairing request with the given requestID. +// Returns false if the request is not found. +func (r *Registry) ApprovePairing(requestID string) bool { + r.pairingMu.Lock() + req, ok := r.pairingByID[requestID] + if ok { + delete(r.pairingByID, requestID) + delete(r.pairingByNodeID, req.NodeID) + } + r.pairingMu.Unlock() + if !ok { + return false + } + req.resolved <- true + return true +} + +// RejectPairing rejects the pairing request with the given requestID. +// Returns false if the request is not found. +func (r *Registry) RejectPairing(requestID string) bool { + r.pairingMu.Lock() + req, ok := r.pairingByID[requestID] + if ok { + delete(r.pairingByID, requestID) + delete(r.pairingByNodeID, req.NodeID) + } + r.pairingMu.Unlock() + if !ok { + return false + } + req.resolved <- false + return true +} + +// RemovePairingRequest removes a pending request without notifying (used on disconnect). +func (r *Registry) RemovePairingRequest(nodeID string) { + r.pairingMu.Lock() + reqID, ok := r.pairingByNodeID[nodeID] + if ok { + delete(r.pairingByNodeID, nodeID) + if req, exists := r.pairingByID[reqID]; exists { + delete(r.pairingByID, reqID) + // Non-blocking: resolve as rejected so the goroutine unblocks. + select { + case req.resolved <- false: + default: + } + } + } + r.pairingMu.Unlock() +} diff --git a/pkg/nodes/server.go b/pkg/nodes/server.go new file mode 100644 index 000000000..38c8239eb --- /dev/null +++ b/pkg/nodes/server.go @@ -0,0 +1,414 @@ +package nodes + +import ( + "context" + "encoding/json" + "fmt" + "net" + "net/http" + "os" + "strings" + "time" + + "github.com/google/uuid" + "github.com/gorilla/websocket" + + "github.com/sipeed/picoclaw/pkg/logger" +) + +// ServerConfig configures the nodes WebSocket server. +type ServerConfig struct { + // Enabled controls whether the node server starts at all. + Enabled bool `json:"enabled"` + + // Host is the bind address (e.g. "0.0.0.0" or "127.0.0.1"). + Host string `json:"host"` + + // Port is the TCP port to listen on (default: 18789). + Port int `json:"port"` + + // Token is an optional shared secret. If non-empty, connecting nodes must + // provide it as auth.token. Leave empty to allow all connections. + Token string `json:"token"` +} + +// Server is a WebSocket server that accepts openclaw node connections. +// It speaks the openclaw gateway protocol (version 3), allowing headless +// Linux nodes, iOS, Android, and macOS apps to connect as "node" role clients. +type Server struct { + cfg ServerConfig + registry *Registry + upgrader websocket.Upgrader + httpSrv *http.Server +} + +// NewServer creates a new Server with the given configuration. +// The returned server shares the provided Registry with other components. +func NewServer(cfg ServerConfig, registry *Registry) *Server { + return &Server{ + cfg: cfg, + registry: registry, + upgrader: websocket.Upgrader{ + // Allow all origins since nodes may connect from any host. + CheckOrigin: func(r *http.Request) bool { return true }, + // Match the openclaw gateway max payload (10 MB). + ReadBufferSize: 1024 * 1024, + WriteBufferSize: 1024 * 1024, + }, + } +} + +// Start begins listening for WebSocket connections. +// It returns after the listener is bound, running the accept loop in background. +// Cancel ctx to gracefully shut down the server. +func (s *Server) Start(ctx context.Context) error { + addr := fmt.Sprintf("%s:%d", s.cfg.Host, s.cfg.Port) + + mux := http.NewServeMux() + mux.HandleFunc("/", s.handleHTTP) + + s.httpSrv = &http.Server{ + Addr: addr, + Handler: mux, + } + + ln, err := net.Listen("tcp", addr) + if err != nil { + return fmt.Errorf("nodes server listen %s: %w", addr, err) + } + + logger.InfoCF("nodes", "Node server started", + map[string]any{"addr": addr, "token_required": s.cfg.Token != ""}) + + go func() { + if err := s.httpSrv.Serve(ln); err != nil && err != http.ErrServerClosed { + logger.ErrorCF("nodes", "Node server error", map[string]any{"error": err.Error()}) + } + }() + + go func() { + <-ctx.Done() + shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := s.httpSrv.Shutdown(shutCtx); err != nil { + logger.ErrorCF("nodes", "Node server shutdown error", map[string]any{"error": err.Error()}) + } + }() + + return nil +} + +// Handler returns the http.HandlerFunc that handles WebSocket node connections. +// Use this to mount the nodes endpoint onto an external HTTP server (e.g. the +// health/API server) instead of starting a dedicated listener with Start(). +// The caller is responsible for server lifecycle; this method is safe to call +// before or after Start(). +func (s *Server) Handler() http.HandlerFunc { + return s.handleHTTP +} + +// handleHTTP upgrades HTTP connections to WebSocket for node connections. +func (s *Server) handleHTTP(w http.ResponseWriter, r *http.Request) { + conn, err := s.upgrader.Upgrade(w, r, nil) + if err != nil { + // Upgrade errors are already written as HTTP responses. + return + } + + remoteIP := r.RemoteAddr + if forwarded := r.Header.Get("X-Forwarded-For"); forwarded != "" { + remoteIP = strings.SplitN(forwarded, ",", 2)[0] + } else if realIP := r.Header.Get("X-Real-IP"); realIP != "" { + remoteIP = realIP + } + // Strip port from remote addr. + if host, _, err := net.SplitHostPort(remoteIP); err == nil { + remoteIP = host + } + + go s.handleConnection(conn, remoteIP) +} + +// handleConnection runs the full lifecycle of a single node WebSocket connection. +func (s *Server) handleConnection(conn *websocket.Conn, remoteIP string) { + connID := uuid.New().String() + defer conn.Close() + + log := func(level, msg string, fields map[string]any) { + if fields == nil { + fields = map[string]any{} + } + fields["conn"] = connID + fields["remote"] = remoteIP + switch level { + case "info": + logger.InfoCF("nodes", msg, fields) + case "warn": + logger.WarnCF("nodes", msg, fields) + case "error": + logger.ErrorCF("nodes", msg, fields) + case "debug": + logger.DebugCF("nodes", msg, fields) + } + } + + send := func(v any) { + data, err := json.Marshal(v) + if err != nil { + return + } + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + log("warn", "send failed", map[string]any{"error": err.Error()}) + } + } + + sendRes := func(id string, ok bool, payload any, errBody *ErrBody) { + frame := ResFrame{ + Type: FrameTypeRes, + ID: id, + Ok: ok, + Payload: payload, + Error: errBody, + } + send(frame) + } + + // Step 1: Send connect.challenge immediately upon connection. + nonce := uuid.New().String() + send(EventFrame{ + Type: FrameTypeEvent, + Event: EventConnectChallenge, + Payload: ConnectChallengePayload{ + Nonce: nonce, + Ts: time.Now().UnixMilli(), + }, + }) + + log("debug", "sent connect.challenge", nil) + + // Step 2: Wait for the connect handshake request with a timeout. + handshakeDeadline := time.Now().Add(30 * time.Second) + if err := conn.SetReadDeadline(handshakeDeadline); err != nil { + log("warn", "set read deadline failed", map[string]any{"error": err.Error()}) + return + } + + _, rawMsg, err := conn.ReadMessage() + if err != nil { + log("debug", "read connect failed", map[string]any{"error": err.Error()}) + return + } + + // Remove deadline for subsequent messages. + if err := conn.SetReadDeadline(time.Time{}); err != nil { + log("warn", "clear read deadline failed", map[string]any{"error": err.Error()}) + } + + // Parse the frame. + var frame map[string]any + if err := json.Unmarshal(rawMsg, &frame); err != nil { + log("warn", "invalid JSON in connect frame", nil) + sendRes("", false, nil, &ErrBody{Code: ErrCodeInvalidRequest, Message: "invalid JSON"}) + return + } + + frameType, _ := frame["type"].(string) + frameMethod, _ := frame["method"].(string) + frameID, _ := frame["id"].(string) + + if frameType != FrameTypeReq || frameMethod != MethodConnect { + msg := fmt.Sprintf("expected connect request, got type=%s method=%s", frameType, frameMethod) + log("warn", "invalid handshake", map[string]any{"msg": msg}) + sendRes(frameID, false, nil, &ErrBody{Code: ErrCodeInvalidRequest, Message: msg}) + return + } + + // Parse ConnectParams from the "params" field. + paramsRaw, _ := frame["params"].(map[string]any) + paramsBytes, _ := json.Marshal(paramsRaw) + var params ConnectParams + if err := json.Unmarshal(paramsBytes, ¶ms); err != nil { + sendRes(frameID, false, nil, &ErrBody{Code: ErrCodeInvalidRequest, Message: "invalid connect params"}) + return + } + + // Step 3: Validate protocol version. + if params.MaxProtocol < ProtocolVersion || params.MinProtocol > ProtocolVersion { + msg := fmt.Sprintf("protocol mismatch: client supports [%d..%d], server requires %d", + params.MinProtocol, params.MaxProtocol, ProtocolVersion) + log("warn", "protocol mismatch", map[string]any{ + "min": params.MinProtocol, + "max": params.MaxProtocol, + }) + sendRes(frameID, false, nil, &ErrBody{Code: ErrCodeInvalidRequest, Message: msg}) + return + } + + // Step 4: Validate role (only "node" is accepted by this server). + if params.Role != "node" { + msg := fmt.Sprintf("unsupported role: %q (only 'node' is accepted)", params.Role) + log("warn", "invalid role", map[string]any{"role": params.Role}) + sendRes(frameID, false, nil, &ErrBody{Code: ErrCodeInvalidRequest, Message: msg}) + return + } + + // Step 5: Authenticate if a token is configured. + if s.cfg.Token != "" { + token := "" + if params.Auth != nil { + token = params.Auth.Token + } + if token != s.cfg.Token { + log("warn", "unauthorized connection", nil) + sendRes(frameID, false, nil, &ErrBody{Code: ErrCodeInvalidRequest, Message: "unauthorized"}) + return + } + } + + // Step 6: Validate nonce to prevent replay attacks (only for remote connections). + if params.Device != nil && params.Device.Nonce != "" { + if params.Device.Nonce != nonce { + log("warn", "nonce mismatch", map[string]any{ + "got": params.Device.Nonce, + "expected": nonce, + }) + sendRes(frameID, false, nil, &ErrBody{Code: ErrCodeInvalidRequest, Message: "device nonce mismatch"}) + return + } + } + + // Step 7: Register the node. + session := s.registry.Register(connID, conn, ¶ms, remoteIP) + + // Build hello-ok response matching the openclaw gateway protocol. + // policy.tickIntervalMs is critical: the node client watchdog closes the + // connection if no tick event arrives within tickIntervalMs*2. We must + // broadcast a tick every TickIntervalMs milliseconds to keep nodes alive. + hostname, _ := os.Hostname() + helloOk := HelloOkPayload{ + Type: "hello-ok", + Protocol: ProtocolVersion, + Server: HelloOkServer{ + Version: "picoclaw", + Host: hostname, + ConnID: connID, + }, + Features: HelloOkFeatures{ + Methods: []string{MethodNodeInvokeResult, MethodNodeEvent}, + Events: []string{EventConnectChallenge, EventNodeInvokeRequest, "tick"}, + }, + Policy: HelloOkPolicy{ + MaxPayload: MaxPayloadBytes, + MaxBufferedBytes: MaxBufferedBytes, + TickIntervalMs: TickIntervalMs, + }, + } + + sendRes(frameID, true, helloOk, nil) + + log("info", "node connected", map[string]any{ + "node_id": session.NodeID, + "display_name": session.DisplayName, + "platform": session.Platform, + "commands": session.Commands, + }) + + defer func() { + nodeID := s.registry.Unregister(connID) + if nodeID != "" { + log("info", "node disconnected", map[string]any{"node_id": nodeID}) + } + }() + + // Step 8a: Start tick keepalive goroutine. + // The openclaw node client's watchdog disconnects if it doesn't receive a + // tick event within tickIntervalMs*2. We send one every TickIntervalMs. + tickDone := make(chan struct{}) + go func() { + ticker := time.NewTicker(TickIntervalMs * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-tickDone: + return + case t := <-ticker.C: + send(EventFrame{ + Type: FrameTypeEvent, + Event: "tick", + Payload: map[string]any{"ts": t.UnixMilli()}, + }) + } + } + }() + defer close(tickDone) + + // Step 8b: Handle subsequent requests from the node. + for { + _, rawMsg, err := conn.ReadMessage() + if err != nil { + if !websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) { + log("debug", "connection read error", map[string]any{"error": err.Error()}) + } + return + } + + var req map[string]any + if err := json.Unmarshal(rawMsg, &req); err != nil { + sendRes("", false, nil, &ErrBody{Code: ErrCodeInvalidRequest, Message: "invalid JSON"}) + continue + } + + reqType, _ := req["type"].(string) + reqMethod, _ := req["method"].(string) + reqID, _ := req["id"].(string) + + if reqType != FrameTypeReq { + // Ignore non-request frames (events, etc.). + continue + } + + switch reqMethod { + case MethodNodeInvokeResult: + s.handleNodeInvokeResult(req, reqID, session, sendRes) + case MethodNodeEvent: + // Node is sending an event upward (e.g., voice, sensor data). + // For now, just acknowledge. Future: route to agent bus. + sendRes(reqID, true, map[string]any{"ok": true}, nil) + default: + log("warn", "unknown method", map[string]any{"method": reqMethod}) + sendRes(reqID, false, nil, &ErrBody{Code: ErrCodeInvalidRequest, + Message: fmt.Sprintf("unknown method: %s", reqMethod)}) + } + } +} + +// handleNodeInvokeResult processes a node.invoke.result request from a node. +func (s *Server) handleNodeInvokeResult( + req map[string]any, + reqID string, + session *NodeSession, + sendRes func(id string, ok bool, payload any, errBody *ErrBody), +) { + paramsRaw, _ := req["params"].(map[string]any) + paramsBytes, _ := json.Marshal(paramsRaw) + var p NodeInvokeResultParams + if err := json.Unmarshal(paramsBytes, &p); err != nil { + sendRes(reqID, false, nil, &ErrBody{Code: ErrCodeInvalidRequest, Message: "invalid params"}) + return + } + + // Security: the nodeId in the result must match the caller's nodeId. + if p.NodeID != session.NodeID { + sendRes(reqID, false, nil, &ErrBody{Code: ErrCodeInvalidRequest, Message: "nodeId mismatch"}) + return + } + + // Deliver the result to any waiting Invoke() call. + matched := s.registry.HandleInvokeResult(&p) + if !matched { + // Late-arriving result (after timeout) - acknowledged but ignored. + logger.DebugCF("nodes", "late invoke result ignored", + map[string]any{"id": p.ID, "node_id": p.NodeID}) + } + sendRes(reqID, true, map[string]any{"ok": true}, nil) +} diff --git a/pkg/tools/nodes.go b/pkg/tools/nodes.go new file mode 100644 index 000000000..0e5158444 --- /dev/null +++ b/pkg/tools/nodes.go @@ -0,0 +1,1037 @@ +package tools + +import ( + "context" + "encoding/base64" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "time" + + "github.com/sipeed/picoclaw/pkg/nodes" +) + +// NodesTool exposes all connected openclaw node capabilities to agents. +// +// Supported actions match the original openclaw nodes-tool.ts and canvas-tool.ts: +// +// status – list all connected nodes +// describe – describe a specific node +// pending – list nodes pending manual approval (pairing queue) +// approve – approve a pairing request by requestId +// reject – reject a pairing request by requestId +// notify – send a push notification to a node (system.notify) +// camera_snap – take a photo from a node camera (camera.snap) +// camera_list – list cameras on a node (camera.list) +// camera_clip – record a short video clip (camera.clip) +// screen_record – record the node screen (screen.record) +// location_get – get current GPS location from a node (location.get) +// run – execute a shell command on a node (system.run) +// device_torch – turn device flashlight on/off via on:bool (device.torch) +// canvas_present – show/open node canvas WebView (canvas.present) +// canvas_hide – hide the canvas WebView (canvas.hide) +// canvas_navigate – navigate canvas to a URL (canvas.navigate) +// canvas_eval – execute JavaScript in the canvas (canvas.eval) +// canvas_snapshot – capture a screenshot of the canvas (canvas.snapshot) +// canvas_a2ui_push – push A2UI JSONL messages to the canvas (canvas.a2ui.pushJSONL) +// canvas_a2ui_reset– reset the A2UI canvas state (canvas.a2ui.reset) +type NodesTool struct { + registry *nodes.Registry +} + +// NewNodesTool creates a NodesTool that reads from the given Registry. +func NewNodesTool(registry *nodes.Registry) *NodesTool { + return &NodesTool{registry: registry} +} + +func (t *NodesTool) Name() string { return "nodes" } + +func (t *NodesTool) Description() string { + return "Discover and control paired nodes (status/describe/pairing/notify/camera/screen/location/run/device/canvas)." +} + +func (t *NodesTool) Parameters() map[string]any { + return map[string]any{ + "type": "object", + "properties": map[string]any{ + "action": map[string]any{ + "type": "string", + "description": "Action to perform. " + + "Node discovery: status, describe. " + + "Pairing: pending, approve, reject. " + + "Notification: notify. " + + "Camera: camera_snap (photo), camera_list, camera_clip (video). " + + "Screen: screen_record. " + + "Location: location_get. " + + "Shell (desktop/server nodes only, NOT mobile): run. " + + "Device: device_torch (control flashlight/torch — on:true to turn on, on:false to turn off). " + + "Canvas/WebView (mobile & desktop nodes): " + + "canvas_present (open WebView, optional url+placement), " + + "canvas_hide (hide WebView), " + + "canvas_navigate (navigate to URL — use this to open a URL in the node browser), " + + "canvas_eval (execute JavaScript and return result), " + + "canvas_snapshot (screenshot the WebView), " + + "canvas_a2ui_push (push A2UI JSONL UI messages), " + + "canvas_a2ui_reset (reset A2UI state). ", + "enum": []string{ + "status", "describe", + "pending", "approve", "reject", + "notify", + "camera_snap", "camera_list", "camera_clip", + "screen_record", + "location_get", + "run", + "device_torch", + "canvas_present", "canvas_hide", "canvas_navigate", + "canvas_eval", "canvas_snapshot", + "canvas_a2ui_push", "canvas_a2ui_reset", + }, + }, + "node": map[string]any{ + "type": "string", + "description": "Target node ID or display name (required for most actions except status/pending).", + }, + "requestId": map[string]any{ + "type": "string", + "description": "Pairing request ID for approve/reject actions.", + }, + "timeoutMs": map[string]any{ + "type": "number", + "description": "Overall invoke timeout in milliseconds (default: 30000).", + }, + // notify + "title": map[string]any{ + "type": "string", + "description": "Notification title (notify action).", + }, + "body": map[string]any{ + "type": "string", + "description": "Notification body text (notify action).", + }, + "sound": map[string]any{ + "type": "string", + "description": "Notification sound name (notify action).", + }, + "priority": map[string]any{ + "type": "string", + "description": "Notification priority: passive | active | timeSensitive (notify action).", + "enum": []string{"passive", "active", "timeSensitive"}, + }, + "delivery": map[string]any{ + "type": "string", + "description": "Notification delivery channel: system | overlay | auto (notify action).", + "enum": []string{"system", "overlay", "auto"}, + }, + // camera_snap / camera_clip + "facing": map[string]any{ + "type": "string", + "description": "Camera facing: front | back | both (camera_snap: all three; camera_clip: front|back only).", + "enum": []string{"front", "back", "both"}, + }, + "maxWidth": map[string]any{ + "type": "number", + "description": "Maximum image width in pixels (camera_snap).", + }, + "quality": map[string]any{ + "type": "number", + "description": "JPEG quality 0–100 (camera_snap).", + }, + "delayMs": map[string]any{ + "type": "number", + "description": "Capture delay in milliseconds (camera_snap).", + }, + "deviceId": map[string]any{ + "type": "string", + "description": "Specific camera device ID (camera_snap / camera_clip).", + }, + "durationMs": map[string]any{ + "type": "number", + "description": "Recording duration in milliseconds (camera_clip / screen_record; default 3000/10000).", + }, + "includeAudio": map[string]any{ + "type": "boolean", + "description": "Whether to include audio (camera_clip / screen_record; default true).", + }, + // screen_record + "fps": map[string]any{ + "type": "number", + "description": "Recording frame rate (screen_record; default 10).", + }, + "screenIndex": map[string]any{ + "type": "number", + "description": "Screen index to record (screen_record; default 0).", + }, + "outPath": map[string]any{ + "type": "string", + "description": "Optional output file path (screen_record; default: temp file).", + }, + // location_get + "maxAgeMs": map[string]any{ + "type": "number", + "description": "Accept a cached location up to this many ms old (location_get).", + }, + "locationTimeoutMs": map[string]any{ + "type": "number", + "description": "Max time to wait for a fresh GPS fix (location_get).", + }, + "desiredAccuracy": map[string]any{ + "type": "string", + "description": "GPS accuracy hint: coarse | balanced | precise (location_get).", + "enum": []string{"coarse", "balanced", "precise"}, + }, + // run + "command": map[string]any{ + "type": "array", + "description": "Argv array for the command to run, e.g. [\"echo\",\"hello\"] (run action).", + "items": map[string]any{"type": "string"}, + }, + "cwd": map[string]any{ + "type": "string", + "description": "Working directory for the command (run action).", + }, + "env": map[string]any{ + "type": "array", + "description": "Extra environment variables as KEY=VALUE strings (run action).", + "items": map[string]any{"type": "string"}, + }, + "commandTimeoutMs": map[string]any{ + "type": "number", + "description": "Timeout for the remote command itself (run action).", + }, + // canvas_present + "url": map[string]any{ + "type": "string", + "description": "URL to load in the canvas (canvas_present / canvas_navigate).", + }, + "x": map[string]any{ + "type": "number", + "description": "Canvas placement x offset in pixels (canvas_present).", + }, + "y": map[string]any{ + "type": "number", + "description": "Canvas placement y offset in pixels (canvas_present).", + }, + "width": map[string]any{ + "type": "number", + "description": "Canvas placement width in pixels (canvas_present).", + }, + "height": map[string]any{ + "type": "number", + "description": "Canvas placement height in pixels (canvas_present).", + }, + // canvas_eval + "javaScript": map[string]any{ + "type": "string", + "description": "JavaScript code to execute in the canvas (canvas_eval).", + }, + // canvas_snapshot + "outputFormat": map[string]any{ + "type": "string", + "description": "Snapshot image format: png | jpg | jpeg (canvas_snapshot; default png).", + "enum": []string{"png", "jpg", "jpeg"}, + }, + // device_torch + "on": map[string]any{ + "type": "boolean", + "description": "Turn the torch on (true) or off (false) (device_torch action).", + }, + // canvas_a2ui_push + "jsonl": map[string]any{ + "type": "string", + "description": "A2UI JSONL payload string, one JSON object per line (canvas_a2ui_push).", + }, + }, + "required": []string{"action"}, + } +} + +// Execute dispatches the requested action to the appropriate registry call. +func (t *NodesTool) Execute(_ context.Context, args map[string]any) *ToolResult { + action, _ := args["action"].(string) + if action == "" { + return ErrorResult("action is required") + } + + timeoutMs := intArg(args, "timeoutMs", 30_000) + + switch action { + + // ── status ────────────────────────────────────────────────────────────── + case "status": + sessions := t.registry.ListConnected() + if len(sessions) == 0 { + return SilentResult("[]") + } + rows := make([]map[string]any, 0, len(sessions)) + for _, s := range sessions { + rows = append(rows, map[string]any{ + "nodeId": s.NodeID, + "displayName": s.DisplayName, + "platform": s.Platform, + "deviceFamily": s.DeviceFamily, + "version": s.Version, + "remoteIp": s.RemoteIP, + "connectedAtMs": s.ConnectedAtMs, + "caps": s.Caps, + "commands": s.Commands, + }) + } + return jsonResult(rows) + + // ── describe ───────────────────────────────────────────────────────────── + case "describe": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + s := t.registry.Get(nodeID) + if s == nil { + return ErrorResult(fmt.Sprintf("node %q not connected", nodeID)) + } + return jsonResult(map[string]any{ + "nodeId": s.NodeID, + "displayName": s.DisplayName, + "platform": s.Platform, + "deviceFamily": s.DeviceFamily, + "modelIdentifier": s.ModelIdentifier, + "version": s.Version, + "remoteIp": s.RemoteIP, + "connectedAtMs": s.ConnectedAtMs, + "caps": s.Caps, + "commands": s.Commands, + }) + + // ── pending ─────────────────────────────────────────────────────────────── + case "pending": + reqs := t.registry.ListPairingRequests() + if len(reqs) == 0 { + return SilentResult("[]") + } + rows := make([]map[string]any, 0, len(reqs)) + for _, r := range reqs { + rows = append(rows, map[string]any{ + "requestId": r.RequestID, + "nodeId": r.NodeID, + "displayName": r.DisplayName, + "platform": r.Platform, + "deviceFamily": r.DeviceFamily, + "remoteIp": r.RemoteIP, + "requestedAtMs": r.RequestedAtMs, + }) + } + return jsonResult(rows) + + // ── approve ─────────────────────────────────────────────────────────────── + case "approve": + reqID, _ := args["requestId"].(string) + if reqID == "" { + return ErrorResult("requestId is required for approve action") + } + if !t.registry.ApprovePairing(reqID) { + return ErrorResult(fmt.Sprintf("pairing request %q not found", reqID)) + } + return jsonResult(map[string]any{"ok": true, "requestId": reqID}) + + // ── reject ──────────────────────────────────────────────────────────────── + case "reject": + reqID, _ := args["requestId"].(string) + if reqID == "" { + return ErrorResult("requestId is required for reject action") + } + if !t.registry.RejectPairing(reqID) { + return ErrorResult(fmt.Sprintf("pairing request %q not found", reqID)) + } + return jsonResult(map[string]any{"ok": true, "requestId": reqID}) + + // ── notify ──────────────────────────────────────────────────────────────── + case "notify": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + title, _ := args["title"].(string) + body, _ := args["body"].(string) + if title == "" && body == "" { + return ErrorResult("title or body is required for notify action") + } + p := map[string]any{} + if title != "" { + p["title"] = title + } + if body != "" { + p["body"] = body + } + if s, ok := args["sound"].(string); ok && s != "" { + p["sound"] = s + } + if s, ok := args["priority"].(string); ok && s != "" { + p["priority"] = s + } + if s, ok := args["delivery"].(string); ok && s != "" { + p["delivery"] = s + } + result := t.registry.Invoke(nodeID, "system.notify", p, timeoutMs) + if !result.Ok { + return ErrorResult(invokeErrMsg(result)) + } + return jsonResult(map[string]any{"ok": true}) + + // ── camera_snap ─────────────────────────────────────────────────────────── + case "camera_snap": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + facing, _ := args["facing"].(string) + if facing == "" { + facing = "both" + } + facings := []string{} + switch facing { + case "front", "back": + facings = []string{facing} + default: // "both" + facings = []string{"front", "back"} + } + + p := map[string]any{"format": "jpg"} + if v := floatArg(args, "maxWidth"); v > 0 { + p["maxWidth"] = int(v) + } + if v := floatArg(args, "quality"); v > 0 { + p["quality"] = int(v) + } + if v := floatArg(args, "delayMs"); v > 0 { + p["delayMs"] = int(v) + } + if s, ok := args["deviceId"].(string); ok && s != "" { + p["deviceId"] = s + } + + type snapResult struct { + facing string + path string + width any + height any + } + var results []snapResult + + for _, f := range facings { + p["facing"] = f + res := t.registry.Invoke(nodeID, "camera.snap", p, timeoutMs) + if !res.Ok { + return ErrorResult(fmt.Sprintf("camera.snap facing=%s: %s", f, invokeErrMsg(res))) + } + payload, err := decodePayload(res) + if err != nil { + return ErrorResult(fmt.Sprintf("camera.snap parse: %v", err)) + } + b64, _ := payload["base64"].(string) + if b64 == "" { + return ErrorResult("camera.snap: empty base64 in response") + } + imgBytes, err := base64.StdEncoding.DecodeString(b64) + if err != nil { + imgBytes, err = base64.RawStdEncoding.DecodeString(b64) + if err != nil { + return ErrorResult("camera.snap: base64 decode failed") + } + } + path := tempMediaPath("snap", f, "jpg") + if err := os.WriteFile(path, imgBytes, 0o644); err != nil { + return ErrorResult(fmt.Sprintf("camera.snap: write file: %v", err)) + } + results = append(results, snapResult{ + facing: f, + path: path, + width: payload["width"], + height: payload["height"], + }) + } + + details := make([]map[string]any, 0, len(results)) + files := "" + for _, r := range results { + files += fmt.Sprintf("MEDIA:%s\n", r.path) + details = append(details, map[string]any{ + "facing": r.facing, + "path": r.path, + "width": r.width, + "height": r.height, + }) + } + detailJSON, _ := json.Marshal(details) + return SilentResult(files + string(detailJSON)) + + // ── camera_list ─────────────────────────────────────────────────────────── + case "camera_list": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + res := t.registry.Invoke(nodeID, "camera.list", map[string]any{}, timeoutMs) + if !res.Ok { + return ErrorResult(invokeErrMsg(res)) + } + payload, err := decodePayload(res) + if err != nil { + return jsonResult(map[string]any{}) + } + return jsonResult(payload) + + // ── camera_clip ─────────────────────────────────────────────────────────── + case "camera_clip": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + facing, _ := args["facing"].(string) + if facing == "" { + facing = "front" + } + if facing != "front" && facing != "back" { + return ErrorResult("camera_clip facing must be front or back") + } + durationMs := intArg(args, "durationMs", 3_000) + includeAudio := true + if v, ok := args["includeAudio"].(bool); ok { + includeAudio = v + } + p := map[string]any{ + "facing": facing, + "durationMs": durationMs, + "includeAudio": includeAudio, + "format": "mp4", + } + if s, ok := args["deviceId"].(string); ok && s != "" { + p["deviceId"] = s + } + res := t.registry.Invoke(nodeID, "camera.clip", p, timeoutMs+durationMs) + if !res.Ok { + return ErrorResult(invokeErrMsg(res)) + } + payload, err := decodePayload(res) + if err != nil { + return ErrorResult(fmt.Sprintf("camera.clip parse: %v", err)) + } + b64, _ := payload["base64"].(string) + if b64 == "" { + return ErrorResult("camera.clip: empty base64 in response") + } + vidBytes, err := base64.StdEncoding.DecodeString(b64) + if err != nil { + vidBytes, err = base64.RawStdEncoding.DecodeString(b64) + if err != nil { + return ErrorResult("camera.clip: base64 decode failed") + } + } + path := tempMediaPath("clip", facing, "mp4") + if err := os.WriteFile(path, vidBytes, 0o644); err != nil { + return ErrorResult(fmt.Sprintf("camera.clip: write file: %v", err)) + } + return SilentResult(fmt.Sprintf("FILE:%s\n%s", path, mustJSON(map[string]any{ + "facing": facing, + "path": path, + "durationMs": payload["durationMs"], + "hasAudio": payload["hasAudio"], + }))) + + // ── screen_record ───────────────────────────────────────────────────────── + case "screen_record": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + durationMs := intArg(args, "durationMs", 10_000) + fps := intArg(args, "fps", 10) + screenIndex := intArg(args, "screenIndex", 0) + includeAudio := true + if v, ok := args["includeAudio"].(bool); ok { + includeAudio = v + } + p := map[string]any{ + "durationMs": durationMs, + "fps": fps, + "screenIndex": screenIndex, + "includeAudio": includeAudio, + "format": "mp4", + } + res := t.registry.Invoke(nodeID, "screen.record", p, timeoutMs+durationMs) + if !res.Ok { + return ErrorResult(invokeErrMsg(res)) + } + payload, err := decodePayload(res) + if err != nil { + return ErrorResult(fmt.Sprintf("screen.record parse: %v", err)) + } + b64, _ := payload["base64"].(string) + if b64 == "" { + return ErrorResult("screen.record: empty base64 in response") + } + vidBytes, err := base64.StdEncoding.DecodeString(b64) + if err != nil { + vidBytes, err = base64.RawStdEncoding.DecodeString(b64) + if err != nil { + return ErrorResult("screen.record: base64 decode failed") + } + } + outPath, _ := args["outPath"].(string) + if outPath == "" { + outPath = tempMediaPath("screen", fmt.Sprintf("idx%d", screenIndex), "mp4") + } + if err := os.MkdirAll(filepath.Dir(outPath), 0o755); err != nil { + return ErrorResult(fmt.Sprintf("screen.record: mkdir: %v", err)) + } + if err := os.WriteFile(outPath, vidBytes, 0o644); err != nil { + return ErrorResult(fmt.Sprintf("screen.record: write file: %v", err)) + } + return SilentResult(fmt.Sprintf("FILE:%s\n%s", outPath, mustJSON(map[string]any{ + "path": outPath, + "durationMs": payload["durationMs"], + "fps": payload["fps"], + "screenIndex": payload["screenIndex"], + "hasAudio": payload["hasAudio"], + }))) + + // ── location_get ────────────────────────────────────────────────────────── + case "location_get": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + p := map[string]any{} + if v := floatArg(args, "maxAgeMs"); v > 0 { + p["maxAgeMs"] = int(v) + } + if v := floatArg(args, "locationTimeoutMs"); v > 0 { + p["timeoutMs"] = int(v) + } + if s, ok := args["desiredAccuracy"].(string); ok && s != "" { + p["desiredAccuracy"] = s + } + res := t.registry.Invoke(nodeID, "location.get", p, timeoutMs) + if !res.Ok { + return ErrorResult(invokeErrMsg(res)) + } + payload, _ := decodePayload(res) + return jsonResult(payload) + + // ── run ─────────────────────────────────────────────────────────────────── + case "run": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + node := t.registry.Get(nodeID) + if node == nil { + return ErrorResult(fmt.Sprintf("node %q not connected", nodeID)) + } + if !node.HasCommand("system.run") { + return ErrorResult(fmt.Sprintf( + "node %q does not support system.run (available commands: %v)", + nodeID, node.Commands)) + } + + rawCmd, ok := args["command"] + if !ok || rawCmd == nil { + return ErrorResult("command is required (argv array, e.g. [\"echo\",\"hello\"])") + } + cmdSlice, ok := rawCmd.([]any) + if !ok { + return ErrorResult("command must be an array of strings") + } + cmd := make([]string, 0, len(cmdSlice)) + for _, v := range cmdSlice { + cmd = append(cmd, fmt.Sprintf("%v", v)) + } + if len(cmd) == 0 { + return ErrorResult("command must not be empty") + } + + // approved=true tells the node-host that picoclaw has already + // authorized this invocation, bypassing its exec-approval prompt. + p := map[string]any{"command": cmd, "approved": true} + if v, ok := args["cwd"].(string); ok && v != "" { + p["cwd"] = v + } + if v, ok := args["env"].([]any); ok && len(v) > 0 { + envPairs := make([]string, 0, len(v)) + for _, e := range v { + envPairs = append(envPairs, fmt.Sprintf("%v", e)) + } + p["env"] = envPairs + } + if v := floatArg(args, "commandTimeoutMs"); v > 0 { + p["timeoutMs"] = int(v) + } + + invokeTimeout := timeoutMs + if v := floatArg(args, "invokeTimeoutMs"); v > 0 { + invokeTimeout = int(v) + } + + res := t.registry.Invoke(nodeID, "system.run", p, invokeTimeout) + if !res.Ok { + return ErrorResult(invokeErrMsg(res)) + } + payload, _ := decodePayload(res) + return jsonResult(payload) + + // ── device_torch ───────────────────────────────────────────────────────── + case "device_torch": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + + onVal, hasOn := args["on"].(bool) + if !hasOn { + return ErrorResult("device_torch requires on (boolean): true to turn on, false to turn off") + } + + res := t.registry.Invoke(nodeID, "device.torch", map[string]any{"on": onVal}, timeoutMs) + if !res.Ok { + return ErrorResult(invokeErrMsg(res)) + } + return jsonResult(map[string]any{"ok": true}) + + // ── canvas_present ──────────────────────────────────────────────────────── + case "canvas_present": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + p := map[string]any{} + if u, ok := args["url"].(string); ok && strings.TrimSpace(u) != "" { + p["url"] = strings.TrimSpace(u) + } + // Optional placement object; only set if at least one coordinate is given. + placement := map[string]any{} + if v, ok := args["x"].(float64); ok { + placement["x"] = int(v) + } + if v, ok := args["y"].(float64); ok { + placement["y"] = int(v) + } + if v, ok := args["width"].(float64); ok { + placement["width"] = int(v) + } + if v, ok := args["height"].(float64); ok { + placement["height"] = int(v) + } + if len(placement) > 0 { + p["placement"] = placement + } + res := t.registry.Invoke(nodeID, "canvas.present", p, timeoutMs) + if !res.Ok { + return ErrorResult(invokeErrMsg(res)) + } + return jsonResult(map[string]any{"ok": true}) + + // ── canvas_hide ─────────────────────────────────────────────────────────── + case "canvas_hide": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + res := t.registry.Invoke(nodeID, "canvas.hide", map[string]any{}, timeoutMs) + if !res.Ok { + return ErrorResult(invokeErrMsg(res)) + } + return jsonResult(map[string]any{"ok": true}) + + // ── canvas_navigate ─────────────────────────────────────────────────────── + case "canvas_navigate": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + u, ok := args["url"].(string) + if !ok || strings.TrimSpace(u) == "" { + return ErrorResult("url is required for canvas_navigate") + } + res := t.registry.Invoke(nodeID, "canvas.navigate", map[string]any{"url": strings.TrimSpace(u)}, timeoutMs) + if !res.Ok { + return ErrorResult(invokeErrMsg(res)) + } + return jsonResult(map[string]any{"ok": true}) + + // ── canvas_eval ─────────────────────────────────────────────────────────── + case "canvas_eval": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + js, ok := args["javaScript"].(string) + if !ok || strings.TrimSpace(js) == "" { + return ErrorResult("javaScript is required for canvas_eval") + } + res := t.registry.Invoke(nodeID, "canvas.eval", map[string]any{"javaScript": js}, timeoutMs) + if !res.Ok { + return ErrorResult(invokeErrMsg(res)) + } + payload, _ := decodePayload(res) + // Return the result string if present, otherwise ok. + if result, ok := payload["result"].(string); ok && result != "" { + return SilentResult(result) + } + return jsonResult(map[string]any{"ok": true}) + + // ── canvas_snapshot ─────────────────────────────────────────────────────── + case "canvas_snapshot": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + // Normalise format: jpg/jpeg → jpeg; anything else → png. + formatRaw := "png" + if f, ok := args["outputFormat"].(string); ok { + formatRaw = strings.ToLower(strings.TrimSpace(f)) + } + format := "png" + if formatRaw == "jpg" || formatRaw == "jpeg" { + format = "jpeg" + } + p := map[string]any{"format": format} + if v := floatArg(args, "maxWidth"); v > 0 { + p["maxWidth"] = int(v) + } + if v := floatArg(args, "quality"); v > 0 { + p["quality"] = int(v) + } + res := t.registry.Invoke(nodeID, "canvas.snapshot", p, timeoutMs) + if !res.Ok { + return ErrorResult(invokeErrMsg(res)) + } + payload, err := decodePayload(res) + if err != nil { + return ErrorResult(fmt.Sprintf("canvas.snapshot parse: %v", err)) + } + b64, _ := payload["base64"].(string) + if b64 == "" { + return ErrorResult("canvas.snapshot: empty base64 in response") + } + imgBytes, err := base64.StdEncoding.DecodeString(b64) + if err != nil { + imgBytes, err = base64.RawStdEncoding.DecodeString(b64) + if err != nil { + return ErrorResult("canvas.snapshot: base64 decode failed") + } + } + ext := format // "png" or "jpeg" + if ext == "jpeg" { + ext = "jpg" + } + path := tempMediaPath("canvas", "snapshot", ext) + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return ErrorResult(fmt.Sprintf("canvas.snapshot: mkdir: %v", err)) + } + if err := os.WriteFile(path, imgBytes, 0o644); err != nil { + return ErrorResult(fmt.Sprintf("canvas.snapshot: write file: %v", err)) + } + return SilentResult(fmt.Sprintf("MEDIA:%s\n%s", path, mustJSON(map[string]any{ + "format": format, + "path": path, + }))) + + // ── canvas_a2ui_push ────────────────────────────────────────────────────── + case "canvas_a2ui_push": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + jsonl, ok := args["jsonl"].(string) + if !ok || strings.TrimSpace(jsonl) == "" { + return ErrorResult("jsonl is required for canvas_a2ui_push") + } + res := t.registry.Invoke(nodeID, "canvas.a2ui.pushJSONL", map[string]any{"jsonl": jsonl}, timeoutMs) + if !res.Ok { + return ErrorResult(invokeErrMsg(res)) + } + return jsonResult(map[string]any{"ok": true}) + + // ── canvas_a2ui_reset ───────────────────────────────────────────────────── + case "canvas_a2ui_reset": + nodeID, err := t.resolveNode(args) + if err != nil { + return ErrorResult(err.Error()) + } + res := t.registry.Invoke(nodeID, "canvas.a2ui.reset", map[string]any{}, timeoutMs) + if !res.Ok { + return ErrorResult(invokeErrMsg(res)) + } + return jsonResult(map[string]any{"ok": true}) + + default: + return ErrorResult(fmt.Sprintf("unknown action: %q", action)) + } +} + +// ── helpers ──────────────────────────────────────────────────────────────────── + +// resolveNode resolves "node" arg to a nodeID by matching connected sessions. +// resolveNode resolves the "node" argument to a nodeID. +// Mirrors openclaw's resolveNodeIdFromList logic: +// - "current" / "default" / empty → picks the only connected node, or the +// first one when multiple are connected (auto-default). +// - Exact nodeId match. +// - Exact remoteIp match. +// - Normalized display-name match (lower-case, non-alnum → "-"). +// - Partial nodeId prefix match (query length ≥ 6). +func (t *NodesTool) resolveNode(args map[string]any) (string, error) { + ref, _ := args["node"].(string) + sessions := t.registry.ListConnected() + + // Auto-default: "current", "default", or empty query → pick the single + // connected node, or the first one if there are multiple. + if ref == "" || ref == "current" || ref == "default" { + if len(sessions) == 0 { + return "", fmt.Errorf("no nodes are currently connected") + } + return sessions[0].NodeID, nil + } + + qNorm := normalizeNodeKey(ref) + + var matches []*nodes.NodeSession + for _, s := range sessions { + if s.NodeID == ref { + return s.NodeID, nil // exact ID → return immediately + } + if s.RemoteIP == ref { + matches = append(matches, s) + continue + } + if s.DisplayName != "" && normalizeNodeKey(s.DisplayName) == qNorm { + matches = append(matches, s) + continue + } + // Partial prefix match (at least 6 chars to avoid false positives). + if len(ref) >= 6 && strings.HasPrefix(s.NodeID, ref) { + matches = append(matches, s) + } + } + + if len(matches) == 1 { + return matches[0].NodeID, nil + } + if len(matches) > 1 { + names := make([]string, len(matches)) + for i, s := range matches { + if s.DisplayName != "" { + names[i] = s.DisplayName + } else { + names[i] = s.NodeID + } + } + return "", fmt.Errorf("ambiguous node %q (matches: %s)", ref, strings.Join(names, ", ")) + } + + // Build a known-nodes hint for the error message. + known := make([]string, 0, len(sessions)) + for _, s := range sessions { + if s.DisplayName != "" { + known = append(known, s.DisplayName) + } else { + known = append(known, s.NodeID) + } + } + hint := "" + if len(known) > 0 { + hint = fmt.Sprintf(" (connected: %s)", strings.Join(known, ", ")) + } + return "", fmt.Errorf("no connected node matches %q%s", ref, hint) +} + +// normalizeNodeKey converts a string to a lowercase slug (non-alnum → "-"), +// matching openclaw's normalizeNodeKey used for display-name comparison. +func normalizeNodeKey(s string) string { + var b strings.Builder + prevDash := true // suppress leading dashes + for _, r := range strings.ToLower(s) { + if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') { + b.WriteRune(r) + prevDash = false + } else if !prevDash { + b.WriteByte('-') + prevDash = true + } + } + // Trim trailing dash. + result := b.String() + return strings.TrimRight(result, "-") +} + +// jsonResult marshals v and wraps it in a SilentResult. +func jsonResult(v any) *ToolResult { + data, err := json.MarshalIndent(v, "", " ") + if err != nil { + return ErrorResult(fmt.Sprintf("json marshal error: %v", err)) + } + return SilentResult(string(data)) +} + +// mustJSON marshals v, returning "{}" on error. +func mustJSON(v any) string { + data, _ := json.Marshal(v) + return string(data) +} + +// invokeErrMsg formats an error message from an InvokeResult. +func invokeErrMsg(res nodes.InvokeResult) string { + if res.Error != nil { + if res.Error.Message != "" { + return res.Error.Message + } + return res.Error.Code + } + return "unknown error" +} + +// decodePayload extracts the payload map from an InvokeResult. +// It tries PayloadJSON first, then Payload directly. +func decodePayload(res nodes.InvokeResult) (map[string]any, error) { + if res.PayloadJSON != nil && *res.PayloadJSON != "" { + var m map[string]any + if err := json.Unmarshal([]byte(*res.PayloadJSON), &m); err != nil { + return nil, err + } + return m, nil + } + if res.Payload != nil { + if m, ok := res.Payload.(map[string]any); ok { + return m, nil + } + // Re-marshal round-trip. + data, _ := json.Marshal(res.Payload) + var m map[string]any + if err := json.Unmarshal(data, &m); err != nil { + return nil, err + } + return m, nil + } + return map[string]any{}, nil +} + +// tempMediaPath returns a temp file path for node media files. +func tempMediaPath(kind, label, ext string) string { + ts := time.Now().UnixMilli() + name := fmt.Sprintf("node_%s_%s_%d.%s", kind, label, ts, ext) + return filepath.Join(os.TempDir(), "picoclaw", name) +} + +// intArg reads an integer parameter with a fallback default. +func intArg(args map[string]any, key string, defaultVal int) int { + if v, ok := args[key].(float64); ok && v > 0 { + return int(v) + } + return defaultVal +} + +// floatArg reads a numeric parameter (JSON numbers arrive as float64). +func floatArg(args map[string]any, key string) float64 { + v, _ := args[key].(float64) + return v +}