function: gateway support nodes to control android node or linux node

This commit is contained in:
陈波0668000637 2026-02-28 14:58:01 +08:00
parent 2d1ea48234
commit f22f6006f8
22 changed files with 3757 additions and 12 deletions

3
.cursorindexingignore Normal file
View file

@ -0,0 +1,3 @@
# Don't index SpecStory auto-save files, but allow explicit context inclusion via @ references
.specstory/**

4
.specstory/.gitignore vendored Normal file
View file

@ -0,0 +1,4 @@
# SpecStory project identity file
/.project.json
# SpecStory explanation file
/.what-is-this.md

View file

@ -0,0 +1,266 @@
<!-- Generated by SpecStory, Markdown v2.1.0 -->
<!-- cursor Session 2194b767-5f09-47e4-ab91-543ae429d4f8 (2026-02-25 09:23Z) -->
# Current branch status (2026-02-25 09:23Z)
_**User (2026-02-25 09:23Z)**_
查看当前分支情况
---
_**Agent (model claude-4.5-sonnet-thinking, mode Agent)**_
<think><details><summary>Thought Process</summary>
用户要查看当前分支情况。我需要运行 git 命令来查看当前分支状态。最常用的命令是:
- `git branch` - 查看本地分支
- `git status` - 查看当前分支和状态
- `git branch -a` - 查看所有分支(包括远程)
让我先运行 git status 和 git branch 来查看当前分支情况。</details></think>
---
我来帮你查看当前的分支情况。
---
<tool-use data-tool-type="shell" data-tool-name="run_terminal_cmd">
<details><summary>Tool use: **run_terminal_cmd** • Run command: cd /home/0668001505/Desktop/picoclaw/picoclaw && git branch -a</summary>
```bash
cd /home/0668001505/Desktop/picoclaw/picoclaw && git branch -a
```
```
* main
remotes/origin/HEAD -&gt; origin/main
remotes/origin/feat/agent-browser-tool
remotes/origin/feat/mcp-context7-integration-cleanup
remotes/origin/main
remotes/origin/refactor/channel-system
```
</details>
</tool-use>
---
<tool-use data-tool-type="shell" data-tool-name="run_terminal_cmd">
<details><summary>Tool use: **run_terminal_cmd** • Run command: cd /home/0668001505/Desktop/picoclaw/picoclaw && git status</summary>
```bash
cd /home/0668001505/Desktop/picoclaw/picoclaw && git status
```
```
位于分支 main
您的分支与上游分支 'origin/main' 一致。
未跟踪的文件:
(使用 "git add &lt;文件&gt;..." 以包含要提交的内容)
.cursorindexingignore
.specstory/
提交为空,但是存在尚未跟踪的文件(使用 "git add" 建立跟踪)
```
</details>
</tool-use>
---
<think><details><summary>Thought Process</summary>
好的,我已经获取了分支信息。让我总结一下:
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/
现在我用中文总结给用户。</details></think>
---
根据查看结果,当前分支情况如下:
## 📍 当前分支
- **当前所在分支**: `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)**_
<think><details><summary>Thought Process</summary>
用户希望我参考 ZeroClaw 的描述风格,给出当前项目 picoclaw 的描述。我需要先了解一下 picoclaw 项目的实际情况,然后基于项目的真实特性,仿照 ZeroClaw 的描述风格来撰写。
让我先查看一下项目的基本信息,比如 README、package.json 或其他配置文件。</details></think>
---
我来帮你了解项目并给出描述。首先让我查看一下项目的基本信息。
---
<tool-use data-tool-type="read" data-tool-name="read_file">
<details><summary>Tool use: **read_file** • Read file: /home/0668001505/Desktop/picoclaw/picoclaw/README.md</summary>
</details>
</tool-use>
---
Could not find file '/home/0668001505/Desktop/picoclaw/picoclaw/package.json' in the workspace.
---
<tool-use data-tool-type="generic" data-tool-name="list_dir">
<details>
<summary>Tool use: **list_dir** • Listed directory, 0 results</summary>
No results found
</details>
</tool-use>
---
<think><details><summary>Thought Process</summary>
好的,现在我了解了 picoclaw 项目的情况。这是一个用 Go 语言编写的超轻量级 AI 助手,与 ZeroClawRust 编写)的定位类似,但更注重极致的资源效率。
让我根据 README 的内容,参考 ZeroClaw 的描述风格,给出 picoclaw 的项目描述。</details></think>
---
根据项目信息,我为你撰写了 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-NanoRISC-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 证明了即使在最受限的环境中,也能构建功能完整的自主智能体系统。
---

View file

@ -0,0 +1,155 @@
<!-- Generated by SpecStory, Markdown v2.1.0 -->
<!-- cursor Session c87ea7a0-bb08-49c5-a773-26182e9af799 (2026-02-25 10:14Z) -->
# 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)**_
<think><details><summary>Thought Process</summary>
用户想要我参考 ZeroClaw 的描述,为当前项目 picoclaw 提供一个类似的项目描述。
让我先探索一下这个项目的结构和内容,了解它的特点和功能。
我需要:
1. 查看项目结构
2. 了解项目的主要文件和配置
3. 基于 ZeroClaw 的描述风格,为 picoclaw 创建一个类似的描述
让我先探索项目结构。</details></think>
---
我来帮您探索 picoclaw 项目并给出描述。让我先查看项目结构。
---
<tool-use data-tool-type="generic" data-tool-name="list_dir">
<details>
<summary>Tool use: **list_dir** • Listed directory, 0 results</summary>
No results found
</details>
</tool-use>
---
<tool-use data-tool-type="read" data-tool-name="read_file">
<details><summary>Tool use: **read_file** • Read file: /home/0668001505/Desktop/picoclaw/picoclaw/README.md</summary>
</details>
</tool-use>
---
<tool-use data-tool-type="generic" data-tool-name="glob_file_search">
<details>
<summary>Tool use: **glob_file_search** • Searched codebase "Cargo.toml" • **0** result in **1** directory</summary>
Directory: **/home/0668001505/Desktop/picoclaw/picoclaw** (0 file)
</details>
</tool-use>
---
<think><details><summary>Thought Process</summary>
好的,我现在了解了 picoclaw 项目的结构。让我查看更多关键文件来了解项目的核心特性。</details></think>
---
<tool-use data-tool-type="read" data-tool-name="read_file">
<details><summary>Tool use: **read_file** • Read file: /home/0668001505/Desktop/picoclaw/picoclaw/go.mod</summary>
</details>
</tool-use>
---
<tool-use data-tool-type="read" data-tool-name="read_file">
<details><summary>Tool use: **read_file** • Read file: /home/0668001505/Desktop/picoclaw/picoclaw/README.zh.md</summary>
</details>
</tool-use>
---
<think><details><summary>Thought Process</summary>
好的,现在我对 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 风格的描述:</details></think>
---
根据我对项目的分析,以下是参考 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 自举开发)。
---

View file

@ -0,0 +1,145 @@
<!-- Generated by SpecStory, Markdown v2.1.0 -->
<!-- cursor Session 54f6d339-bad5-48ae-8793-64f74601ff20 (2026-02-25 10:32Z) -->
# 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)**_
<think><details><summary>Thought Process</summary>
用户要求我用中文总结这个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修复和性能优化</details></think>
---
基于这些提交记录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)**_
<think><details><summary>Thought Process</summary>
用户要求用100字总结v0.1.1版本的主要改动。我需要更简洁地概括核心内容。
核心内容:
1. 多平台支持QQ、钉钉、Slack、飞书、Telegram
2. Agent功能增强记忆系统、Cron工具
3. OAuth认证
4. 迁移命令
5. Bug修复和优化
用中文100字左右总结</details></think>
---
v0.1.1版本主要更新:
**多平台集成**新增QQ频道、钉钉、Slack、飞书支持改进Telegram集成。
**Agent增强**实现记忆系统、动态上下文压缩、Cron定时任务工具。
**认证与工具**添加OAuth登录、`picoclaw migrate`迁移命令。
**安全优化**:强化工作空间目录边界控制。
**开发改进**添加CI构建、支持darwin arm64、修复数据竞争和心跳服务等多项Bug。
---

View file

@ -0,0 +1,282 @@
<!-- Generated by SpecStory, Markdown v2.1.0 -->
<!-- cursor Session cd3f7560-ddfb-49a4-8624-a12bef19cb3e (2026-02-25 10:36Z) -->
# 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)**_
<think><details><summary>Thought Process</summary>
用户提供了一系列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更新
用户要求用中文回复,所以我需要用中文总结这些改动。</details></think>
---
根据这些提交记录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)**_
<think><details><summary>Thought Process</summary>
用户要求用100字左右进行总结之前的回答太详细了。我需要精简到大约100个汉字。
让我重新组织一下,只提取最核心的几点:
1. 工具系统重构
2. 新增多个渠道LINE、OneBot
3. 新增多个AI提供商DeepSeek、GitHub Copilot、Ollama等
4. 集成goreleaser
5. 安全和bug修复
100字左右的总结</details></think>
---
## 0.1.2版本主要改动总结
**核心重构**工具系统全面升级支持异步执行和新的ToolResult返回机制心跳服务增强并默认启用。
**新增渠道**LINE官方账号、OneBot支持优化Telegram和Discord消息处理。
**AI提供商扩展**新增DeepSeek、胜算云、GitHub Copilot、本地Ollama等多个提供商。
**基础设施**集成goreleaser自动化发布新增健康检查端点支持设备热插拔通知。
**安全加固**修复符号链接逃逸漏洞、并发安全问题加强权限控制和ACL检查。
---

View file

@ -15,6 +15,7 @@ import (
"github.com/sipeed/picoclaw/pkg/agent" "github.com/sipeed/picoclaw/pkg/agent"
"github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/bus"
"github.com/sipeed/picoclaw/pkg/logger" "github.com/sipeed/picoclaw/pkg/logger"
"github.com/sipeed/picoclaw/pkg/nodes"
"github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/providers"
) )
@ -51,6 +52,14 @@ func agentCmd(message, sessionKey, model string, debug bool) error {
defer msgBus.Close() defer msgBus.Close()
agentLoop := agent.NewAgentLoop(cfg, msgBus, provider) 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) // Print agent startup info (only for interactive mode)
startupInfo := agentLoop.GetStartupInfo() startupInfo := agentLoop.GetStartupInfo()
logger.InfoCF("agent", "Agent initialized", logger.InfoCF("agent", "Agent initialized",

View file

@ -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
}

View file

@ -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)
}
}

View file

@ -33,6 +33,7 @@ import (
"github.com/sipeed/picoclaw/pkg/heartbeat" "github.com/sipeed/picoclaw/pkg/heartbeat"
"github.com/sipeed/picoclaw/pkg/logger" "github.com/sipeed/picoclaw/pkg/logger"
"github.com/sipeed/picoclaw/pkg/media" "github.com/sipeed/picoclaw/pkg/media"
"github.com/sipeed/picoclaw/pkg/nodes"
"github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/providers"
"github.com/sipeed/picoclaw/pkg/state" "github.com/sipeed/picoclaw/pkg/state"
"github.com/sipeed/picoclaw/pkg/tools" "github.com/sipeed/picoclaw/pkg/tools"
@ -60,8 +61,31 @@ func gatewayCmd(debug bool) error {
} }
msgBus := bus.NewMessageBus() 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) 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 // Print agent startup info
fmt.Println("\n📦 Agent Status:") fmt.Println("\n📦 Agent Status:")
startupInfo := agentLoop.GetStartupInfo() startupInfo := agentLoop.GetStartupInfo()
@ -169,8 +193,17 @@ func gatewayCmd(debug bool) error {
fmt.Println("✓ Device event service started") 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 := 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) addr := fmt.Sprintf("%s:%d", cfg.Gateway.Host, cfg.Gateway.Port)
channelManager.SetupHTTPServer(addr, healthServer) channelManager.SetupHTTPServer(addr, healthServer)
@ -178,8 +211,17 @@ func gatewayCmd(debug bool) error {
fmt.Printf("Error starting channels: %v\n", err) fmt.Printf("Error starting channels: %v\n", err)
return err return err
} }
// HTTP server is started inside channelManager.StartAll()
fmt.Printf("✓ Health endpoints available at http://%s:%d/health and /ready\n", cfg.Gateway.Host, cfg.Gateway.Port) 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) go agentLoop.Run(ctx)

View file

@ -15,6 +15,7 @@ import (
"github.com/sipeed/picoclaw/cmd/picoclaw/internal" "github.com/sipeed/picoclaw/cmd/picoclaw/internal"
"github.com/sipeed/picoclaw/cmd/picoclaw/internal/agent" "github.com/sipeed/picoclaw/cmd/picoclaw/internal/agent"
"github.com/sipeed/picoclaw/cmd/picoclaw/internal/auth" "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/cron"
"github.com/sipeed/picoclaw/cmd/picoclaw/internal/gateway" "github.com/sipeed/picoclaw/cmd/picoclaw/internal/gateway"
"github.com/sipeed/picoclaw/cmd/picoclaw/internal/migrate" "github.com/sipeed/picoclaw/cmd/picoclaw/internal/migrate"
@ -36,6 +37,7 @@ func NewPicoclawCommand() *cobra.Command {
cmd.AddCommand( cmd.AddCommand(
onboard.NewOnboardCommand(), onboard.NewOnboardCommand(),
agent.NewAgentCommand(), agent.NewAgentCommand(),
chat.NewChatCommand(),
auth.NewAuthCommand(), auth.NewAuthCommand(),
gateway.NewGatewayCommand(), gateway.NewGatewayCommand(),
status.NewStatusCommand(), status.NewStatusCommand(),

View file

@ -260,5 +260,8 @@
"gateway": { "gateway": {
"host": "127.0.0.1", "host": "127.0.0.1",
"port": 18790 "port": 18790
},
"nodes": {
"enabled": false
} }
} }

View file

@ -11,6 +11,7 @@ import (
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"net/http"
"path/filepath" "path/filepath"
"strings" "strings"
"sync" "sync"
@ -24,6 +25,7 @@ import (
"github.com/sipeed/picoclaw/pkg/constants" "github.com/sipeed/picoclaw/pkg/constants"
"github.com/sipeed/picoclaw/pkg/logger" "github.com/sipeed/picoclaw/pkg/logger"
"github.com/sipeed/picoclaw/pkg/media" "github.com/sipeed/picoclaw/pkg/media"
"github.com/sipeed/picoclaw/pkg/nodes"
"github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/providers"
"github.com/sipeed/picoclaw/pkg/routing" "github.com/sipeed/picoclaw/pkg/routing"
"github.com/sipeed/picoclaw/pkg/skills" "github.com/sipeed/picoclaw/pkg/skills"
@ -42,6 +44,7 @@ type AgentLoop struct {
fallback *providers.FallbackChain fallback *providers.FallbackChain
channelManager *channels.Manager channelManager *channels.Manager
mediaStore media.MediaStore mediaStore media.MediaStore
nodeRegistry *nodes.Registry // optional; non-nil when nodes server is enabled
} }
// processOptions configures how a message is processed // 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). // registerSharedTools registers tools that are shared across all agents (web, message, spawn).
func registerSharedTools( func registerSharedTools(
cfg *config.Config, cfg *config.Config,
@ -1350,3 +1367,61 @@ func extractParentPeer(msg bus.InboundMessage) *routing.RoutePeer {
} }
return &routing.RoutePeer{Kind: parentKind, ID: parentID} 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,
})
}
}

304
pkg/agent/openai_http.go Normal file
View file

@ -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 <token>". 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:<agentId>" or
// "agent:<agentId>" 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 }

View file

@ -312,12 +312,11 @@ func (m *Manager) StartAll(ctx context.Context) error {
defer m.mu.Unlock() defer m.mu.Unlock()
if len(m.channels) == 0 { if len(m.channels) == 0 {
logger.WarnC("channels", "No channels enabled") logger.InfoC("channels", "No chat channels enabled; gateway will serve HTTP API only")
return errors.New("no channels enabled") } else {
logger.InfoC("channels", "Starting all channels")
} }
logger.InfoC("channels", "Starting all channels")
dispatchCtx, cancel := context.WithCancel(ctx) dispatchCtx, cancel := context.WithCancel(ctx)
m.dispatchTask = &asyncTask{cancel: cancel} m.dispatchTask = &asyncTask{cancel: cancel}

View file

@ -55,6 +55,7 @@ type Config struct {
Providers ProvidersConfig `json:"providers,omitempty"` Providers ProvidersConfig `json:"providers,omitempty"`
ModelList []ModelConfig `json:"model_list"` // New model-centric provider configuration ModelList []ModelConfig `json:"model_list"` // New model-centric provider configuration
Gateway GatewayConfig `json:"gateway"` Gateway GatewayConfig `json:"gateway"`
Nodes NodesConfig `json:"nodes"`
Tools ToolsConfig `json:"tools"` Tools ToolsConfig `json:"tools"`
Heartbeat HeartbeatConfig `json:"heartbeat"` Heartbeat HeartbeatConfig `json:"heartbeat"`
Devices DevicesConfig `json:"devices"` Devices DevicesConfig `json:"devices"`
@ -488,8 +489,34 @@ func (c *ModelConfig) Validate() error {
} }
type GatewayConfig struct { type GatewayConfig struct {
Host string `json:"host" env:"PICOCLAW_GATEWAY_HOST"` Host string `json:"host" env:"PICOCLAW_GATEWAY_HOST"`
Port int `json:"port" env:"PICOCLAW_GATEWAY_PORT"` 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 <token>". 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 { type BraveConfig struct {

View file

@ -308,6 +308,12 @@ func DefaultConfig() *Config {
Host: "127.0.0.1", Host: "127.0.0.1",
Port: 18790, 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{ Tools: ToolsConfig{
MediaCleanup: MediaCleanupConfig{ MediaCleanup: MediaCleanupConfig{
Enabled: true, Enabled: true,

View file

@ -12,6 +12,7 @@ import (
type Server struct { type Server struct {
server *http.Server server *http.Server
mux *http.ServeMux
mu sync.RWMutex mu sync.RWMutex
ready bool ready bool
checks map[string]Check checks map[string]Check
@ -34,6 +35,7 @@ type StatusResponse struct {
func NewServer(host string, port int) *Server { func NewServer(host string, port int) *Server {
mux := http.NewServeMux() mux := http.NewServeMux()
s := &Server{ s := &Server{
mux: mux,
ready: false, ready: false,
checks: make(map[string]Check), checks: make(map[string]Check),
startTime: time.Now(), startTime: time.Now(),
@ -47,12 +49,18 @@ func NewServer(host string, port int) *Server {
Addr: addr, Addr: addr,
Handler: mux, Handler: mux,
ReadTimeout: 5 * time.Second, ReadTimeout: 5 * time.Second,
WriteTimeout: 5 * time.Second, WriteTimeout: 30 * time.Second, // allow longer writes for chat responses
} }
return s 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 { func (s *Server) Start() error {
s.mu.Lock() s.mu.Lock()
s.ready = true 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. // RegisterOnMux registers health endpoints and any handlers added via RegisterHandler
// This allows the health endpoints to be served by a shared HTTP server. // 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) { func (s *Server) RegisterOnMux(mux *http.ServeMux) {
mux.HandleFunc("/health", s.healthHandler) mux.HandleFunc("/health", s.healthHandler)
mux.HandleFunc("/ready", s.readyHandler) mux.HandleFunc("/ready", s.readyHandler)
mux.Handle("/", s.mux) // catch-all for RegisterHandler routes (/chat, /v1/chat/completions, etc.)
} }
func statusString(ok bool) string { func statusString(ok bool) string {

264
pkg/nodes/protocol.go Normal file
View file

@ -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":"<uuid>","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":"<uuid>","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"`
}

450
pkg/nodes/registry.go Normal file
View file

@ -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()
}

414
pkg/nodes/server.go Normal file
View file

@ -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, &params); 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, &params, 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)
}

1037
pkg/tools/nodes.go Normal file

File diff suppressed because it is too large Load diff