test12322

assets/toyclaw_副本/README.md

@@ -0,0 +1,368 @@
+           # 🐾 ToyClaw
+
+**千行代码实现的 OpenClaw。**
+
+ToyClaw 是一个极简但功能完整的 AI Agent 框架，参考 [nanobot](https://github.com/HKUDS/nanobot) 架构设计，用一千多行 Python 实现了 Agent Loop、Tool Use、Cron 定时任务、Heartbeat 心跳、Sub-agent 后台任务、Web Search 等核心能力。
+
+---
+
+## 目录
+
+- [设计思路](#设计思路)
+- [技术架构](#技术架构)
+- [项目结构](#项目结构)
+- [快速开始](#快速开始)
+- [配置说明](#配置说明)
+- [内置工具](#内置工具)
+- [Skills 扩展](#skills-扩展)
+- [核心模块详解](#核心模块详解)
+- [测试](#测试)
+
+---
+
+## 设计思路
+
+### 为什么做这个？
+
+OpenClaw 已经达到了惊人的40w+行的代码量，抽象层套抽象层，耦合程度堪比仙人排队跳———套里还有套。它看起来很复杂，各种功能：聊天软件支持、多 Provider、Skills、MCP、长Memory、Cron定时任务、Sub-agent……但核心能力其实很简单。Agent Loop 就是一个 while 循环：LLM 思考，调用工具，拿到结果，再思考，直到给出最终回复。其他一切都是生产级需求的包装。
+
+我决定砍掉一切非核心复杂度，仅用 1000 行 Python 代码实现一个功能完整的OpenClaw。我把它命名为toyclaw。在openclaw自身的辅助下，总共花费时间不超过2小时，代码量千行出头，但是它能做的事情几乎和 nanobot 一样——支持工具调用、定时任务、skills扩展、Session 持久化等核心功能。
+
+### 核心取舍
+
+| 保留 | 砍掉 | 原因 |
+|------|------|------|
+| Agent Loop（LLM ↔ Tool） | MessageBus | CLI 直连 Agent，不需要消息总线 |
+| Tool Registry + 9 个内置工具 | MCP 协议 | 内置工具直接注册，不需要外部协议 |
+| Cron 定时任务 | 多 Channel（Telegram/Discord） | 只做 CLI 入口 |
+| Heartbeat 心跳 | Memory Consolidation | Agent 自己用 write_file 管理记忆 |
+| Sub-agent 后台任务 | 多 Provider / LiteLLM | 一个 OpenAI 兼容端点够用 |
+| Skills 动态发现 | 多模态（图片/语音） | 纯文本交互 |
+| Session 持久化 | OAuth / 权限系统 | 个人工具，无需鉴权 |
+
+### 设计原则
+
+1. **单文件可读** — 每个模块职责单一，打开就能读懂
+2. **asyncio 原生** — 全链路 async，Tool 执行、LLM 调用、Cron 定时器都不阻塞
+3. **依赖最小化** — 只用 4 个运行时依赖（openai / httpx / croniter / readability-lxml）
+4. **组合优于继承** — 组件通过回调函数组合，不搞深层继承链
+
+---
+
+## 技术架构
+
+```
+┌─────────────────────────────────────────────────────┐
+│                      CLI REPL                       │
+│              (cli.py — 交互/单次模式)                │
+└─────────────┬───────────────────────────────────────┘
+              │ user input
+              ▼
+┌─────────────────────────────────────────────────────┐
+│                   Agent Loop                        │
+│           (agent.py — 核心处理引擎)                  │
+│                                                     │
+│  ┌──────────┐    ┌──────────┐    ┌───────────────┐  │
+│  │ Context  │    │ Provider │    │ Tool Registry │  │
+│  │ Builder  │    │ (OpenAI) │    │  (9 tools)    │  │
+│  └──────────┘    └──────────┘    └───────────────┘  │
+└──────┬──────────────────────────────────┬───────────┘
+       │                                  │
+       ▼                                  ▼
+┌──────────────┐  ┌──────────┐  ┌─────────────────┐
+│   Session    │  │  Skills  │  │  Background Svc │
+│  (JSONL)     │  │  Loader  │  │                 │
+└──────────────┘  └──────────┘  │  ┌───────────┐  │
+                                │  │   Cron    │  │
+                                │  └───────────┘  │
+                                │  ┌───────────┐  │
+                                │  │ Heartbeat │  │
+                                │  └───────────┘  │
+                                │  ┌───────────┐  │
+                                │  │ Sub-agent │  │
+                                │  └───────────┘  │
+                                └─────────────────┘
+```
+
+### 数据流
+
+```
+User Input
+  → ContextBuilder.build_messages()  (system prompt + history + skills + runtime)
+    → Provider.chat()                (发送到 LLM)
+      → LLM 返回 tool_calls?
+        ├─ YES → ToolRegistry.execute() → 结果追加到 messages → 回到 Provider.chat()
+        └─ NO  → 最终回复 → Session 持久化 → 返回给用户
+```
+
+---
+
+## 项目结构
+
+```
+toyclaw/
+├── pyproject.toml                     # 构建配置 + 依赖声明
+├── README.md
+├── toyclaw/
+│   ├── __init__.py              (3)   # 版本号
+│   ├── __main__.py              (5)   # python -m toyclaw 入口
+│   ├── config.py               (43)   # JSON 配置加载
+│   ├── provider.py             (98)   # OpenAI SDK 封装
+│   ├── context.py             (129)   # System prompt + 消息组装
+│   ├── session.py              (94)   # JSONL 会话持久化
+│   ├── skills.py               (57)   # Skills 目录发现
+│   ├── agent.py               (184)   # 核心 Agent Loop
+│   ├── subagent.py            (125)   # 后台子 Agent
+│   ├── cron.py                (241)   # 定时任务引擎
+│   ├── heartbeat.py           (115)   # 心跳唤醒服务
+│   ├── cli.py                 (174)   # CLI REPL + 组件装配
+│   └── tools/
+│       ├── __init__.py          (1)
+│       ├── base.py             (77)   # Tool 抽象基类 + Registry
+│       └── builtin.py         (452)   # 9 个内置工具
+└── tests/
+    ├── test_functional.py             # 12 项单元测试
+    └── test_e2e.py                    # 4 项端到端测试
+```
+
+
+---
+
+## 快速开始
+
+### 环境要求
+
+- Python ≥ 3.11
+- 一个 OpenAI 兼容的 API 端点（OpenAI / DeepSeek / OpenRouter / 本地 vLLM 等）
+
+### 安装
+
+```bash
+git clone https://github.com/yourname/toyclaw.git
+cd toyclaw
+pip install -e .
+```
+
+### 配置
+
+创建配置文件 `~/.toyclaw/config.json`：
+
+```json
+{
+  "api_key": "sk-your-api-key",
+  "api_base": "https://api.openai.com/v1",
+  "model": "gpt-4o-mini"
+}
+```
+
+### 运行
+
+```bash
+# 交互模式
+toyclaw
+
+# 或者
+python -m toyclaw
+
+# 单次执行
+toyclaw -m "帮我写一个 hello world"
+
+# 指定配置文件
+toyclaw -c /path/to/config.json
+```
+
+### REPL 命令
+
+| 命令 | 说明 |
+|------|------|
+| `/new` | 清空当前会话，开始新对话 |
+| `exit` / `quit` / `/exit` / `:q` | 退出 |
+
+---
+
+## 配置说明
+
+`~/.toyclaw/config.json` 支持以下字段：
+
+```json
+{
+  "api_key": "",                  // 必填 — LLM API Key
+  "api_base": "https://api.openai.com/v1",  // API 端点
+  "model": "gpt-4o",             // 模型名称
+  "workspace": "~/.toyclaw/workspace",      // 工作区路径
+  "brave_api_key": null,          // Brave Search API Key（可选）
+  "max_iterations": 30,           // Agent 单轮最大迭代次数
+  "heartbeat_interval": 1800,     // 心跳间隔（秒）
+  "heartbeat_enabled": true       // 是否启用心跳
+}
+```
+
+### 工作区目录结构
+
+启动后自动创建：
+
+```
+~/.toyclaw/workspace/
+├── memory/          # 长期记忆（MEMORY.md 等）
+├── sessions/        # 会话历史（JSONL 格式）
+├── skills/          # 技能扩展目录
+├── cron/            # 定时任务持久化
+├── AGENTS.md        # Agent 指令文件（可选）
+├── SOUL.md          # 人格设定（可选）
+├── HEARTBEAT.md     # 心跳任务清单（可选）
+└── ...
+```
+
+---
+
+## 内置工具
+
+ToyClaw 提供 9 个内置工具，LLM 可自主决定何时调用：
+
+| 工具 | 名称 | 功能 |
+|------|------|------|
+| 📖 ReadFile | `read_file` | 读取文件内容（128KB 上限，支持路径解析） |
+| ✏️ WriteFile | `write_file` | 写入文件（自动创建父目录） |
+| 🔧 EditFile | `edit_file` | 精确搜索替换（失败时给出 diff 提示） |
+| 📁 ListDir | `list_dir` | 列出目录内容 |
+| 💻 Shell | `exec` | 执行 Shell 命令（带安全拦截 + 60s 超时） |
+| 🔍 WebSearch | `web_search` | Brave Search API 网页搜索 |
+| 🌐 WebFetch | `web_fetch` | 抓取网页并提取正文（readability 算法） |
+| ⏰ Cron | `cron` | 创建 / 列出 / 删除定时任务 |
+| 🚀 Spawn | `spawn` | 派生后台子 Agent 执行耗时任务 |
+
+### Shell 安全防护
+
+内置危险命令拦截，以下模式会被自动阻止：
+
+```
+rm -rf, del /f, rmdir /s, format, mkfs, diskpart,
+dd if=, shutdown, reboot, fork bomb
+```
+
+---
+
+## Skills 扩展
+
+在 `workspace/skills/` 下创建子目录，放入 `SKILL.md` 即可被自动发现：
+
+```
+workspace/skills/
+└── my_skill/
+    ├── SKILL.md          # 技能说明（必须存在）
+    ├── script.py         # 任意辅助文件
+    └── ...
+```
+
+Agent 的 system prompt 会自动包含所有已发现技能的摘要。LLM 可以通过 `read_file` 工具读取 `SKILL.md` 获取详细指令。
+
+---
+
+## 核心模块详解
+
+### Agent Loop（agent.py）⭐
+
+这是整个系统的心脏。核心逻辑只有一个循环：
+
+```python
+for _ in range(max_iterations):
+    response = await provider.chat(messages, tools)
+    if response.has_tool_calls:
+        for tc in response.tool_calls:
+            result = await tools.execute(tc.name, tc.arguments)
+            messages.append(tool_result)
+    else:
+        return response.content  # 最终回复
+```
+
+关键设计：
+- **最大迭代次数保护** — 防止 LLM 无限循环调用工具
+- **`<think>` 标签清理** — 兼容 DeepSeek 等带思考过程的模型
+- **Tool result 截断** — Session 历史中工具结果限制 500 字符，防止 context 爆炸
+- **Runtime context 注入** — 每次对话自动注入当前时间、频道信息
+
+### Provider（provider.py）
+
+薄封装层，将 OpenAI SDK 的响应标准化为 `LLMResponse` 数据类：
+
+```python
+@dataclass
+class LLMResponse:
+    content: str | None
+    tool_calls: list[ToolCallRequest]
+    finish_reason: str
+
+@dataclass
+class ToolCallRequest:
+    id: str
+    name: str
+    arguments: dict[str, Any]
+```
+
+兼容任何 OpenAI 格式的 API（OpenAI / DeepSeek / OpenRouter / vLLM / Ollama）。
+
+### Cron Service（cron.py）
+
+支持三种调度模式：
+
+| 模式 | 示例 | 说明 |
+|------|------|------|
+| `at` | `2026-03-10T10:30:00` | 一次性定时，执行后自动删除 |
+| `every` | `every_seconds=300` | 固定间隔循环 |
+| `cron` | `0 9 * * *` | 标准 cron 表达式（依赖 croniter） |
+
+任务持久化为 JSON，重启后自动恢复。触发时通过回调调用 `Agent.process()`，形成完整闭环。
+
+### Heartbeat（heartbeat.py）
+
+定期读取 `workspace/HEARTBEAT.md`，将内容交给 LLM 判断是否需要执行。LLM 通过专用 tool call 返回 `skip`（无事可做）或 `run`（有待办任务），避免无意义的 Agent 唤醒。
+
+### Sub-agent（subagent.py）
+
+后台 `asyncio.Task` 运行独立的 Agent Loop，拥有受限工具集（**不含 cron 和 spawn**，防止递归）。完成后通过 `on_complete` 回调通知主进程。
+
+### Session（session.py）
+
+JSONL 格式持久化：
+- 每个会话一个 `.jsonl` 文件
+- 第一行为 metadata（key, created_at, updated_at）
+- 后续每行一条消息
+- 加载时自动对齐到 user turn（丢弃开头孤立的 tool 结果）
+
+---
+
+## 测试
+
+### 单元测试（无需 API Key）
+
+```bash
+python tests/test_functional.py
+```
+
+覆盖 12 项：Config、Provider 数据类、ToolRegistry、文件工具（读/写/编辑/列目录）、Shell（含安全拦截）、Session 持久化、Skills 发现、Context 组装、Cron 全生命周期、CronTool 接口、Heartbeat、HTML 清洗。
+
+### 端到端测试（需要 API Key）
+
+```bash
+python tests/test_e2e.py
+```
+
+覆盖 4 项：LLM 直接对话 → Tool Call 生成 → 完整 Agent Loop → Agent 调用工具创建文件并验证。
+
+
+---
+
+## License
+
+MIT
+
+---
+
+*Built as a learning exercise — understanding AI agents by building one from scratch.*
+
+---
+
+## 致谢
+
+- [OpenClaw](https://github.com/OpenClaw) — ToyClaw 的灵感来源与构建过程中的得力助手，没有它就没有这个项目。
+- [nanobot](https://github.com/HKUDS/nanobot) — 优秀的轻量级 AI Agent 框架，ToyClaw 的架构设计参考了它的核心思路。

assets/toyclaw_副本/pyproject.toml

@@ -0,0 +1,19 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "toyclaw"
+version = "0.1.0"
+description = "Ultra-minimal personal AI assistant"
+requires-python = ">=3.11"
+license = "MIT"
+dependencies = [
+    "openai>=1.0",
+    "httpx>=0.25",
+    "croniter>=1.3",
+    "readability-lxml>=0.8",
+]
+
+[project.scripts]
+toyclaw = "toyclaw.cli:main"

assets/toyclaw_副本/tests/test_e2e.py

@@ -0,0 +1,118 @@
+"""End-to-end test: load config, call LLM, run a tool, get final answer."""
+import asyncio
+import io
+import sys
+from pathlib import Path
+
+# Fix Windows GBK encoding for emoji output
+sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
+sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
+
+async def main():
+    # 1. Load config
+    from toyclaw.config import load_config
+    cfg = load_config()
+    print(f"[config] model={cfg.model}, api_base={cfg.api_base}")
+    print(f"[config] workspace={cfg.workspace}")
+
+    # 2. Test provider directly — simple chat (no tools)
+    from toyclaw.provider import OpenAIProvider
+    provider = OpenAIProvider(
+        api_key=cfg.api_key,
+        api_base=cfg.api_base,
+        default_model=cfg.model,
+    )
+
+    print("\n--- Test 1: Simple chat (no tools) ---")
+    resp = await provider.chat(
+        messages=[
+            {"role": "system", "content": "You are a helpful assistant. Reply concisely."},
+            {"role": "user", "content": "What is 2+3? Just the number."},
+        ],
+    )
+    print(f"[LLM] content={resp.content!r}")
+    print(f"[LLM] finish_reason={resp.finish_reason}, tool_calls={len(resp.tool_calls)}")
+    assert resp.content and "5" in resp.content, f"Expected '5' in response, got: {resp.content}"
+    print("PASS: Simple chat works!\n")
+
+    # 3. Test with tools — LLM should call a tool
+    print("--- Test 2: Chat with tools (expect tool call) ---")
+    from toyclaw.tools.base import ToolRegistry
+    from toyclaw.tools.builtin import ReadFileTool, WriteFileTool, ListDirTool, ShellTool
+
+    ws = cfg.workspace
+    reg = ToolRegistry()
+    reg.register(ReadFileTool(workspace=ws))
+    reg.register(WriteFileTool(workspace=ws))
+    reg.register(ListDirTool(workspace=ws))
+    reg.register(ShellTool(workspace=ws))
+
+    schemas = reg.get_definitions()
+    print(f"[tools] {len(schemas)} tools registered: {reg.names}")
+
+    resp2 = await provider.chat(
+        messages=[
+            {"role": "system", "content": "You are an AI assistant with tools. Use them when needed."},
+            {"role": "user", "content": "Please list the files in the current workspace directory."},
+        ],
+        tools=schemas,
+    )
+    print(f"[LLM] content={resp2.content!r}")
+    print(f"[LLM] tool_calls={[(tc.name, tc.arguments) for tc in resp2.tool_calls]}")
+
+    if resp2.has_tool_calls:
+        print("PASS: LLM generated tool calls!")
+        # Execute the tool call
+        for tc in resp2.tool_calls:
+            result = await reg.execute(tc.name, tc.arguments)
+            print(f"[tool:{tc.name}] result={result[:200]!r}")
+    else:
+        print("INFO: LLM responded directly without tool calls (acceptable)")
+
+    # 4. Full agent loop test (via agent.py)
+    print("\n--- Test 3: Full Agent.process() ---")
+    from toyclaw.agent import Agent
+    from toyclaw.session import SessionManager
+
+    sess_mgr = SessionManager(ws)
+
+    agent = Agent(
+        provider=provider,
+        tools=reg,
+        workspace=ws,
+        session_mgr=sess_mgr,
+        max_iterations=cfg.max_iterations,
+    )
+
+    # Simple question — no tool needed
+    answer = await agent.process(
+        content="Hi! What's the capital of France? One word answer.",
+        session_key="e2e_test",
+    )
+    print(f"[agent] answer={answer!r}")
+    assert answer and len(answer) > 0, "Agent returned empty answer"
+    print("PASS: Agent.process() works!\n")
+
+    # 5. Agent with tool usage — ask it to create a file
+    print("--- Test 4: Agent creates a file via tool ---")
+    answer2 = await agent.process(
+        content="Create a file called 'e2e_test.txt' in the workspace with content 'Hello from toyclaw!'. Then confirm what you wrote.",
+        session_key="e2e_test",
+    )
+    print(f"[agent] answer={answer2[:300]!r}")
+    test_file = ws / "e2e_test.txt"
+    if test_file.exists():
+        content = test_file.read_text(encoding="utf-8")
+        print(f"[file] {test_file} content={content!r}")
+        assert "Hello from toyclaw" in content
+        print("PASS: Agent successfully created file via tool!")
+        test_file.unlink()  # cleanup
+    else:
+        print("WARN: File not created (agent may have used different path)")
+
+    print("\n" + "=" * 50)
+    print("All end-to-end tests completed!")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

assets/toyclaw_副本/tests/test_functional.py

@@ -0,0 +1,449 @@
+"""Functional tests for toyclaw components (no API key needed)."""
+import asyncio
+import json
+import os
+import shutil
+import sys
+import tempfile
+from pathlib import Path
+
+loop = asyncio.new_event_loop()
+asyncio.set_event_loop(loop)
+
+# ── 1. Config ──
+from toyclaw.config import Config, load_config
+
+
+def test_config():
+    # Default config
+    cfg = Config()
+    assert cfg.model == "gpt-4o"
+    assert cfg.max_iterations == 30
+    assert cfg.heartbeat_interval == 1800
+
+    # Load from a temp JSON file
+    tmpdir = tempfile.mkdtemp(prefix="oc_cfg_")
+    try:
+        p = Path(tmpdir) / "config.json"
+        p.write_text(json.dumps({"api_key": "sk-test", "model": "gpt-3.5-turbo"}))
+        cfg = load_config(p)
+        assert cfg.api_key == "sk-test"
+        assert cfg.model == "gpt-3.5-turbo"
+        # Defaults still applied for missing keys
+        assert cfg.max_iterations == 30
+        print("  config: Config + load_config OK")
+    finally:
+        shutil.rmtree(tmpdir)
+
+
+# ── 2. Provider data classes ──
+from toyclaw.provider import LLMResponse, ToolCallRequest
+
+
+def test_provider_dataclasses():
+    tc = ToolCallRequest(id="t1", name="read_file", arguments={"path": "x"})
+    assert tc.id == "t1" and tc.name == "read_file"
+    assert tc.arguments == {"path": "x"}
+
+    resp = LLMResponse(content="hi", tool_calls=[], finish_reason="stop")
+    assert resp.content == "hi"
+    assert not resp.has_tool_calls
+
+    resp2 = LLMResponse(tool_calls=[tc])
+    assert resp2.has_tool_calls
+    print("  provider: dataclasses OK")
+
+
+# ── 3. Tool registry ──
+from toyclaw.tools.base import ToolRegistry
+
+
+def test_tool_registry():
+    reg = ToolRegistry()
+    assert reg.get_definitions() == []
+    assert reg.names == []
+
+    # Execute on missing tool
+    result = loop.run_until_complete(reg.execute("nope", {}))
+    assert "not found" in result.lower()
+    print("  tools.base: ToolRegistry OK")
+
+
+# ── 4. Builtin tools ──
+from toyclaw.tools.builtin import (
+    EditFileTool,
+    ListDirTool,
+    ReadFileTool,
+    ShellTool,
+    WriteFileTool,
+)
+
+
+def test_builtin_tools():
+    tmpdir = Path(tempfile.mkdtemp(prefix="oc_test_"))
+    ws = tmpdir  # workspace
+
+    try:
+        # ReadFile — missing file
+        rt = ReadFileTool(workspace=ws)
+        result = loop.run_until_complete(rt.execute(path=str(tmpdir / "nope.txt")))
+        assert "not found" in result.lower()
+
+        # WriteFile
+        wt = WriteFileTool(workspace=ws)
+        wf = tmpdir / "out.txt"
+        result = loop.run_until_complete(wt.execute(path=str(wf), content="wrote this"))
+        assert wf.exists()
+        assert wf.read_text() == "wrote this"
+        print("  builtin: WriteFileTool OK")
+
+        # ReadFile — existing file
+        result = loop.run_until_complete(rt.execute(path=str(wf)))
+        assert "wrote this" in result
+        print("  builtin: ReadFileTool OK")
+
+        # EditFile
+        et = EditFileTool(workspace=ws)
+        result = loop.run_until_complete(et.execute(path=str(wf), old_text="wrote", new_text="edited"))
+        assert "edited this" in wf.read_text()
+        print("  builtin: EditFileTool OK")
+
+        # EditFile — old_text not found
+        result = loop.run_until_complete(et.execute(path=str(wf), old_text="NOPE", new_text="x"))
+        assert "not found" in result.lower()
+        print("  builtin: EditFileTool error case OK")
+
+        # ListDir
+        ld = ListDirTool(workspace=ws)
+        result = loop.run_until_complete(ld.execute(path=str(tmpdir)))
+        assert "out.txt" in result
+        print("  builtin: ListDirTool OK")
+
+        # ShellTool
+        st = ShellTool(workspace=ws)
+        result = loop.run_until_complete(st.execute(command="echo test123"))
+        assert "test123" in result
+        print("  builtin: ShellTool OK")
+
+        # ShellTool — deny pattern
+        result = loop.run_until_complete(st.execute(command="rm -rf /"))
+        assert "blocked" in result.lower()
+        print("  builtin: ShellTool safety guard OK")
+
+    finally:
+        shutil.rmtree(tmpdir)
+
+
+# ── 5. Tool registry integration ──
+def test_tool_registry_with_tools():
+    tmpdir = Path(tempfile.mkdtemp(prefix="oc_reg_"))
+    try:
+        reg = ToolRegistry()
+        reg.register(ReadFileTool(workspace=tmpdir))
+        reg.register(WriteFileTool(workspace=tmpdir))
+        reg.register(ListDirTool(workspace=tmpdir))
+
+        assert len(reg.names) == 3
+        assert "read_file" in reg.names
+
+        # Schema structure
+        schemas = reg.get_definitions()
+        assert len(schemas) == 3
+        assert all(s["type"] == "function" for s in schemas)
+        assert schemas[0]["function"]["name"] in ("read_file", "write_file", "list_dir")
+        print("  registry: register + schema OK")
+
+        # Execute via registry
+        loop.run_until_complete(reg.execute("write_file", {"path": str(tmpdir / "hi.txt"), "content": "hi"}))
+        result = loop.run_until_complete(reg.execute("read_file", {"path": str(tmpdir / "hi.txt")}))
+        assert "hi" in result
+        print("  registry: execute integration OK")
+    finally:
+        shutil.rmtree(tmpdir)
+
+
+# ── 6. Session ──
+from toyclaw.session import Session, SessionManager
+
+
+def test_session():
+    tmpdir = Path(tempfile.mkdtemp(prefix="oc_sess_"))
+    try:
+        mgr = SessionManager(workspace=tmpdir)
+
+        # New session
+        sess = mgr.get_or_create("test_conv")
+        assert isinstance(sess, Session)
+        assert sess.key == "test_conv"
+        assert len(sess.messages) == 0
+
+        # Add messages
+        sess.messages.append({"role": "user", "content": "hi"})
+        sess.messages.append({"role": "assistant", "content": "hello"})
+        sess.messages.append({"role": "user", "content": "q2"})
+        mgr.save(sess)
+
+        # get_history trims to user turn
+        hist = sess.get_history()
+        assert hist[0]["role"] == "user"
+        assert len(hist) == 3
+        print("  session: message append + get_history OK")
+
+        # Persistence — reload from disk
+        mgr2 = SessionManager(workspace=tmpdir)
+        sess2 = mgr2.get_or_create("test_conv")
+        assert len(sess2.messages) == 3
+        assert sess2.messages[0]["content"] == "hi"
+        print("  session: persistence OK")
+    finally:
+        shutil.rmtree(tmpdir)
+
+
+# ── 7. Skills ──
+from toyclaw.skills import SkillsLoader
+
+
+def test_skills():
+    tmpdir = Path(tempfile.mkdtemp(prefix="oc_skills_"))
+    try:
+        # No skills dir
+        loader = SkillsLoader(tmpdir)
+        assert loader.list_skills() == []
+
+        # Create a fake skill
+        sk = tmpdir / "skills" / "demo"
+        sk.mkdir(parents=True)
+        (sk / "SKILL.md").write_text("# Demo Skill\nDoes demo things.\n\nMore detail here.")
+        found = loader.list_skills()
+        assert len(found) == 1
+        assert found[0]["name"] == "demo"
+        assert "Does demo things" in found[0]["description"]
+
+        # build_summary
+        summary = loader.build_summary()
+        assert "demo" in summary
+        assert "Does demo things" in summary
+        print("  skills: SkillsLoader OK")
+    finally:
+        shutil.rmtree(tmpdir)
+
+
+# ── 8. Context ──
+from toyclaw.context import ContextBuilder
+
+
+def test_context():
+    tmpdir = Path(tempfile.mkdtemp(prefix="oc_ctx_"))
+    try:
+        # Bootstrap file
+        (tmpdir / "AGENTS.md").write_text("# Test Agent Instructions")
+
+        builder = ContextBuilder(tmpdir)
+        messages = builder.build_messages(
+            history=[],
+            user_message="hello",
+            skills_summary="- **test_skill**: does stuff",
+        )
+
+        # Should have system + user
+        assert len(messages) == 2
+        assert messages[0]["role"] == "system"
+        assert messages[1]["role"] == "user"
+
+        system = messages[0]["content"]
+        assert "ToyClaw" in system
+        assert "Test Agent Instructions" in system
+        assert "test_skill" in system
+        print("  context: ContextBuilder OK")
+
+        # add_assistant_message / add_tool_result
+        ContextBuilder.add_assistant_message(messages, "thinking...")
+        assert messages[-1]["role"] == "assistant"
+
+        ContextBuilder.add_tool_result(messages, "tc1", "read_file", "file content")
+        assert messages[-1]["role"] == "tool"
+        assert messages[-1]["tool_call_id"] == "tc1"
+        print("  context: message helpers OK")
+
+        # strip_runtime_tag
+        raw = messages[1]["content"]
+        assert "[Runtime Context]" in raw
+        stripped = ContextBuilder.strip_runtime_tag(raw)
+        assert stripped == "hello"
+        print("  context: strip_runtime_tag OK")
+    finally:
+        shutil.rmtree(tmpdir)
+
+
+# ── 9. Cron ──
+from toyclaw.cron import CronJob, CronSchedule, CronService
+
+
+def test_cron():
+    tmpdir = Path(tempfile.mkdtemp(prefix="oc_cron_"))
+    try:
+        store = tmpdir / "cron.json"
+        svc = CronService(store_path=store)
+
+        # Add a recurring job
+        sched = CronSchedule(kind="every", every_ms=30_000)
+        job = svc.add_job(name="test-job", schedule=sched, message="say hi")
+        assert isinstance(job, CronJob)
+        assert job.id
+        assert job.name == "test-job"
+
+        # List
+        jobs = svc.list_jobs()
+        assert len(jobs) == 1
+        assert jobs[0].id == job.id
+        print("  cron: add_job + list_jobs OK")
+
+        # Persistence
+        assert store.exists()
+        data = json.loads(store.read_text())
+        assert len(data) == 1
+        print("  cron: JSON persistence OK")
+
+        # Remove
+        ok = svc.remove_job(job.id)
+        assert ok
+        assert len(svc.list_jobs()) == 0
+
+        # Remove non-existent
+        ok = svc.remove_job("nope")
+        assert not ok
+        print("  cron: remove_job OK")
+
+        # One-shot schedule
+        import time
+        future_ms = int(time.time() * 1000) + 60_000
+        sched2 = CronSchedule(kind="at", at_ms=future_ms)
+        job2 = svc.add_job(name="once", schedule=sched2, message="one time", delete_after_run=True)
+        assert job2.delete_after_run
+        svc.remove_job(job2.id)
+
+        # Cron expression
+        sched3 = CronSchedule(kind="cron", expr="0 9 * * *")
+        job3 = svc.add_job(name="daily", schedule=sched3, message="good morning")
+        assert job3.next_run_ms is not None  # croniter computed it
+        svc.remove_job(job3.id)
+        print("  cron: all schedule types OK")
+    finally:
+        shutil.rmtree(tmpdir)
+
+
+# ── 10. CronTool ──
+from toyclaw.tools.builtin import CronTool
+
+
+def test_cron_tool():
+    tmpdir = Path(tempfile.mkdtemp(prefix="oc_ct_"))
+    try:
+        svc = CronService(store_path=tmpdir / "cron.json")
+        ct = CronTool(service=svc)
+        ct.set_context("cli", "test")
+
+        # Add via tool
+        result = loop.run_until_complete(ct.execute(action="add", message="remind me", every_seconds=60))
+        assert "created" in result.lower()
+        print("  cron_tool: add OK")
+
+        # List via tool
+        result = loop.run_until_complete(ct.execute(action="list"))
+        assert "remind me" in result
+        print("  cron_tool: list OK")
+
+        # Remove via tool
+        job_id = svc.list_jobs()[0].id
+        result = loop.run_until_complete(ct.execute(action="remove", job_id=job_id))
+        assert "removed" in result.lower()
+        print("  cron_tool: remove OK")
+
+        # Error: missing message
+        result = loop.run_until_complete(ct.execute(action="add"))
+        assert "error" in result.lower()
+        print("  cron_tool: error handling OK")
+    finally:
+        shutil.rmtree(tmpdir)
+
+
+# ── 11. Heartbeat ──
+from toyclaw.heartbeat import HeartbeatService
+
+
+def test_heartbeat():
+    tmpdir = Path(tempfile.mkdtemp(prefix="oc_hb_"))
+    try:
+        # HeartbeatService requires a provider; use a mock
+        class FakeProvider:
+            pass
+
+        hb = HeartbeatService(
+            workspace=tmpdir,
+            provider=FakeProvider(),
+            interval_s=1800,
+            enabled=False,
+        )
+        assert hb.interval_s == 1800
+        assert not hb.enabled
+        assert hb._file == tmpdir / "HEARTBEAT.md"
+        print("  heartbeat: init OK")
+
+        # With HEARTBEAT.md
+        (tmpdir / "HEARTBEAT.md").write_text("- Check email\n- Review tasks")
+        assert hb._file.exists()
+        print("  heartbeat: HEARTBEAT.md detected OK")
+    finally:
+        shutil.rmtree(tmpdir)
+
+
+# ── 12. WebFetchTool (strip HTML) ──
+from toyclaw.tools.builtin import WebFetchTool
+
+
+def test_web_fetch_strip():
+    wf = WebFetchTool()
+    html = "<html><body><h1>Title</h1><p>Hello <b>world</b></p></body></html>"
+    text = wf._strip(html)
+    assert "Title" in text
+    assert "Hello" in text
+    assert "world" in text
+    assert "<" not in text  # all tags stripped
+    print("  web_fetch: HTML strip OK")
+
+
+# ── Run all ──
+if __name__ == "__main__":
+    tests = [
+        test_config,
+        test_provider_dataclasses,
+        test_tool_registry,
+        test_builtin_tools,
+        test_tool_registry_with_tools,
+        test_session,
+        test_skills,
+        test_context,
+        test_cron,
+        test_cron_tool,
+        test_heartbeat,
+        test_web_fetch_strip,
+    ]
+    passed = 0
+    failed = 0
+    for t in tests:
+        name = t.__name__
+        try:
+            t()
+            passed += 1
+        except Exception as e:
+            print(f"  FAIL {name}: {e}")
+            import traceback
+            traceback.print_exc()
+            failed += 1
+
+    print(f"\n{'=' * 50}")
+    print(f"Results: {passed} passed, {failed} failed out of {len(tests)}")
+    if failed:
+        sys.exit(1)
+    else:
+        print("All tests passed!")

assets/toyclaw_副本/toyclaw/__init__.py

@@ -0,0 +1,3 @@
+"""ToyClaw — ultra-minimal personal AI assistant."""
+
+__version__ = "0.1.0"

assets/toyclaw_副本/toyclaw/__main__.py

@@ -0,0 +1,5 @@
+"""Entry point for `python -m toyclaw`."""
+
+from toyclaw.cli import main
+
+main()

assets/toyclaw_副本/toyclaw/agent.py

@@ -0,0 +1,185 @@
+"""Agent loop — the core processing engine."""
+
+from __future__ import annotations
+
+import json
+import re
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Awaitable, Callable
+
+from toyclaw.context import ContextBuilder
+from toyclaw.provider import LLMResponse, OpenAIProvider
+from toyclaw.session import Session, SessionManager
+from toyclaw.skills import SkillsLoader
+from toyclaw.tools.base import ToolRegistry
+
+_TOOL_RESULT_MAX = 500  # chars to keep per tool result in session history
+
+
+class Agent:
+    """The agent loop: receive message → build context → LLM ↔ tools → respond.
+
+    This is the heart of ToyClaw.  Everything else feeds into or out of it.
+    """
+
+    def __init__(
+        self,
+        provider: OpenAIProvider,
+        tools: ToolRegistry,
+        workspace: Path,
+        session_mgr: SessionManager | None = None,
+        max_iterations: int = 30,
+        memory_window: int = 100,
+        on_output: Callable[[str], Awaitable[None]] | None = None,
+    ):
+        self.provider = provider
+        self.tools = tools
+        self.workspace = workspace
+        self.sessions = session_mgr or SessionManager(workspace)
+        self.max_iterations = max_iterations
+        self.memory_window = memory_window
+        self.on_output = on_output  # callback for async output (subagent results, cron)
+
+        self._ctx = ContextBuilder(workspace)
+        self._skills = SkillsLoader(workspace)
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    async def process(
+        self,
+        content: str,
+        session_key: str = "cli:direct",
+        channel: str = "cli",
+        chat_id: str = "direct",
+    ) -> str:
+        """Process a single user message and return the final reply."""
+        # Set tool contexts
+        self._set_tool_context(channel, chat_id)
+
+        session = self.sessions.get_or_create(session_key)
+        history = session.get_history(max_messages=self.memory_window)
+        print("\n history: " + str(history))
+
+        skills_summary = self._skills.build_summary()
+        messages = self._ctx.build_messages(
+            history=history,
+            user_message=content,
+            channel=channel,
+            chat_id=chat_id,
+            skills_summary=skills_summary,
+        )
+        print("\n messages: " + str(messages))
+        final, _, all_msgs = await self._run_loop(messages)
+
+        if final is None:
+            final = "I've completed processing but have no response to give."
+
+        self._save_turn(session, all_msgs, skip=1 + len(history))
+        self.sessions.save(session)
+        return final
+
+    # ------------------------------------------------------------------
+    # Agent loop
+    # ------------------------------------------------------------------
+
+    async def _run_loop(
+        self, messages: list[dict[str, Any]]
+    ) -> tuple[str | None, list[str], list[dict[str, Any]]]:
+        """Run the LLM ↔ tool iteration loop.
+
+        Returns (final_content, tools_used, messages).
+        """
+        tools_used: list[str] = []
+        final: str | None = None
+
+        for _iter in range(self.max_iterations):
+            response = await self.provider.chat(
+                messages=messages,
+                tools=self.tools.get_definitions(),
+            )
+            print("\n response: " + str(response))
+            if response.has_tool_calls:
+                # Append assistant message with tool calls
+                tc_dicts = [
+                    {
+                        "id": tc.id,
+                        "type": "function",
+                        "function": {
+                            "name": tc.name,
+                            "arguments": json.dumps(tc.arguments, ensure_ascii=False),
+                        },
+                    }
+                    for tc in response.tool_calls
+                ]
+                self._ctx.add_assistant_message(messages, response.content, tc_dicts)
+
+                # Execute each tool
+                for tc in response.tool_calls:
+                    tools_used.append(tc.name)
+                    result = await self.tools.execute(tc.name, tc.arguments)
+                    self._ctx.add_tool_result(messages, tc.id, tc.name, result)
+            else:
+                # No tool calls → final response
+                clean = self._strip_think(response.content)
+                if response.finish_reason == "error":
+                    final = clean or "Sorry, I encountered an error."
+                    break
+                self._ctx.add_assistant_message(messages, clean)
+                final = clean
+                break
+
+        if final is None:
+            final = (
+                f"Reached the maximum number of iterations ({self.max_iterations}). "
+                "Try breaking the task into smaller steps."
+            )
+        return final, tools_used, messages
+
+    # ------------------------------------------------------------------
+    # Helpers
+    # ------------------------------------------------------------------
+
+    def _set_tool_context(self, channel: str, chat_id: str) -> None:
+        """Update routing context on tools that need it."""
+        for name in ("cron", "spawn"):
+            tool = self.tools.get(name)
+            if tool and hasattr(tool, "set_context"):
+                tool.set_context(channel, chat_id)
+
+    def _save_turn(
+        self, session: Session, messages: list[dict[str, Any]], skip: int
+    ) -> None:
+        """Persist new messages from this turn into the session."""
+        now = datetime.now()
+        for m in messages[skip:]:
+            entry = dict(m)
+            role = entry.get("role")
+            content = entry.get("content")
+
+            # Skip empty assistant messages (no content, no tool_calls)
+            if role == "assistant" and not content and not entry.get("tool_calls"):
+                continue
+
+            # Strip runtime context tag from user messages
+            if role == "user" and isinstance(content, str):
+                entry["content"] = ContextBuilder.strip_runtime_tag(content)
+                if not entry["content"]:
+                    continue
+
+            # Truncate oversized tool results
+            if role == "tool" and isinstance(content, str) and len(content) > _TOOL_RESULT_MAX:
+                entry["content"] = content[:_TOOL_RESULT_MAX] + "\n... (truncated)"
+
+            entry.setdefault("timestamp", now.isoformat())
+            session.messages.append(entry)
+        session.updated_at = now
+
+    @staticmethod
+    def _strip_think(text: str | None) -> str | None:
+        """Remove <think>…</think> blocks that some models embed."""
+        if not text:
+            return None
+        return re.sub(r"<think>[\s\S]*?</think>", "", text).strip() or None

assets/toyclaw_副本/toyclaw/cli.py

@@ -0,0 +1,174 @@
+"""CLI entry-point — interactive REPL and one-shot mode."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import sys
+from pathlib import Path
+
+from toyclaw.agent import Agent
+from toyclaw.config import Config, load_config
+from toyclaw.cron import CronService
+from toyclaw.heartbeat import HeartbeatService
+from toyclaw.provider import OpenAIProvider
+from toyclaw.session import SessionManager
+from toyclaw.skills import SkillsLoader
+from toyclaw.subagent import SubagentManager
+from toyclaw.tools.base import ToolRegistry
+from toyclaw.tools.builtin import (
+    CronTool,
+    EditFileTool,
+    ListDirTool,
+    ReadFileTool,
+    ShellTool,
+    SpawnTool,
+    WebFetchTool,
+    WebSearchTool,
+    WriteFileTool,
+)
+
+log = logging.getLogger("toyclaw")
+
+_EXIT_CMDS = {"exit", "quit", "/exit", "/quit", ":q"}
+
+
+# ===================================================================
+# Bootstrap helpers
+# ===================================================================
+
+def _ensure_workspace(ws: Path) -> None:
+    for sub in ("memory", "sessions", "skills", "cron"):
+        (ws / sub).mkdir(parents=True, exist_ok=True)
+
+
+def _build_stack(cfg: Config):
+    """Wire up every component and return (agent, cron_service, heartbeat)."""
+    ws = cfg.workspace
+    _ensure_workspace(ws)
+
+    provider = OpenAIProvider(api_key=cfg.api_key, api_base=cfg.api_base, default_model=cfg.model)
+    session_mgr = SessionManager(ws)
+
+    # Async output callback (for subagent results & cron)
+    async def _print_async(text: str) -> None:
+        print(f"\n{text}\n> ", end="", flush=True)
+
+    # Cron
+    cron_svc = CronService(store_path=ws / "cron" / "jobs.json")
+    subagent_mgr = SubagentManager(
+        provider=provider, workspace=ws,
+        on_complete=_print_async, brave_api_key=cfg.brave_api_key,
+    )
+
+    # Tools
+    tools = ToolRegistry()
+    for cls in (ReadFileTool, WriteFileTool, EditFileTool, ListDirTool):
+        tools.register(cls(workspace=ws))
+    tools.register(ShellTool(workspace=ws))
+    tools.register(WebSearchTool(api_key=cfg.brave_api_key))
+    tools.register(WebFetchTool())
+    tools.register(CronTool(service=cron_svc))
+    tools.register(SpawnTool(manager=subagent_mgr))
+
+    agent = Agent(
+        provider=provider, tools=tools, workspace=ws,
+        session_mgr=session_mgr, max_iterations=cfg.max_iterations,
+        on_output=_print_async,
+    )
+
+    # Wire cron callback → agent.process
+    async def _on_cron(job) -> str | None:
+        result = await agent.process(
+            content=job.message,
+            session_key=f"cron:{job.id}",
+            channel=job.channel,
+            chat_id=job.chat_id,
+        )
+        if result:
+            await _print_async(f"⏰ [{job.name}]: {result}")
+        return result
+
+    cron_svc.on_job = _on_cron
+
+    # Heartbeat
+    heartbeat = HeartbeatService(
+        workspace=ws, provider=provider,
+        on_execute=lambda tasks: agent.process(tasks, session_key="heartbeat:main"),
+        on_notify=_print_async,
+        interval_s=cfg.heartbeat_interval,
+        enabled=cfg.heartbeat_enabled,
+    )
+
+    return agent, cron_svc, heartbeat
+
+
+# ===================================================================
+# Main
+# ===================================================================
+
+async def _async_main(cfg: Config, one_shot: str | None = None) -> None:
+    agent, cron_svc, heartbeat = _build_stack(cfg)
+
+    # Start background services
+    await cron_svc.start()
+    await heartbeat.start()
+
+    try:
+        if one_shot:
+            resp = await agent.process(one_shot)
+            print(resp)
+            return
+
+        print("🐾 ToyClaw ready. Type 'exit' to quit.\n")
+        loop = asyncio.get_event_loop()
+        while True:
+            try:
+                user_input = await loop.run_in_executor(None, lambda: input("> "))
+            except (EOFError, KeyboardInterrupt):
+                print()
+                break
+            text = user_input.strip()
+            if not text:
+                continue
+            if text.lower() in _EXIT_CMDS:
+                break
+            if text == "/new":
+                session = agent.sessions.get_or_create("cli:direct")
+                session.clear()
+                agent.sessions.save(session)
+                print("New session started.\n")
+                continue
+            resp = await agent.process(text)
+            print(f"\n{resp}\n")
+    finally:
+        cron_svc.stop()
+        heartbeat.stop()
+
+
+def main() -> None:
+    """CLI entry-point."""
+    logging.basicConfig(level=logging.WARNING, format="%(levelname)s: %(message)s")
+
+    # Minimal arg parsing (no argparse dependency bloat)
+    args = sys.argv[1:]
+    config_path: Path | None = None
+    one_shot: str | None = None
+
+    i = 0
+    while i < len(args):
+        if args[i] in ("-c", "--config") and i + 1 < len(args):
+            config_path = Path(args[i + 1])
+            i += 2
+        elif args[i] in ("-m", "--message") and i + 1 < len(args):
+            one_shot = args[i + 1]
+            i += 2
+        else:
+            i += 1
+
+    cfg = load_config(config_path)
+    if not cfg.api_key:
+        print("Error: api_key not set. Configure ~/.toyclaw/config.json")
+        sys.exit(1)
+
+    asyncio.run(_async_main(cfg, one_shot))

assets/toyclaw_副本/toyclaw/config.py

@@ -0,0 +1,43 @@
+"""Configuration loader."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from pathlib import Path
+
+_DEFAULT_DIR = Path.home() / ".toyclaw"
+
+
+@dataclass
+class Config:
+    """Application configuration."""
+
+    api_key: str = ""
+    api_base: str = "https://api.openai.com/v1"
+    model: str = "gpt-4o"
+    workspace: Path = field(default_factory=lambda: _DEFAULT_DIR / "workspace")
+    brave_api_key: str | None = None
+    max_iterations: int = 30
+    heartbeat_interval: int = 1800  # seconds
+    heartbeat_enabled: bool = True
+
+
+def load_config(path: Path | None = None) -> Config:
+    """Load config from JSON file. Missing keys use defaults."""
+    path = path or _DEFAULT_DIR / "config.json"
+    if not path.exists():
+        return Config()
+    raw = json.loads(path.read_text(encoding="utf-8"))
+    kwargs: dict = {}
+    for key in ("api_key", "api_base", "model", "brave_api_key"):
+        if key in raw:
+            kwargs[key] = raw[key]
+    for key in ("max_iterations", "heartbeat_interval"):
+        if key in raw:
+            kwargs[key] = int(raw[key])
+    if "heartbeat_enabled" in raw:
+        kwargs["heartbeat_enabled"] = bool(raw["heartbeat_enabled"])
+    if "workspace" in raw:
+        kwargs["workspace"] = Path(raw["workspace"]).expanduser()
+    return Config(**kwargs)

assets/toyclaw_副本/toyclaw/context.py

@@ -0,0 +1,129 @@
+"""Context builder — assembles system prompt and message lists."""
+
+from __future__ import annotations
+
+import platform
+import time
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+_BOOTSTRAP_FILES = ("AGENTS.md", "SOUL.md", "USER.md", "IDENTITY.md")
+_RUNTIME_TAG = "[Runtime Context]"
+
+
+class ContextBuilder:
+    """Builds system prompt + message lists for the LLM."""
+
+    def __init__(self, workspace: Path):
+        self.workspace = workspace
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def build_messages(
+        self,
+        history: list[dict[str, Any]],
+        user_message: str,
+        channel: str | None = None,
+        chat_id: str | None = None,
+        skills_summary: str = "",
+    ) -> list[dict[str, Any]]:
+        """Return the full message list: [system, ...history, user]."""
+        system = self._build_system_prompt(skills_summary)
+        runtime = self._runtime_context(channel, chat_id)
+        merged_user = f"{runtime}\n\n{user_message}"
+        return [
+            {"role": "system", "content": system},
+            *history,
+            {"role": "user", "content": merged_user},
+        ]
+
+    @staticmethod
+    def add_assistant_message(
+        messages: list[dict[str, Any]],
+        content: str | None,
+        tool_calls: list[dict[str, Any]] | None = None,
+    ) -> list[dict[str, Any]]:
+        msg: dict[str, Any] = {"role": "assistant", "content": content}
+        if tool_calls:
+            msg["tool_calls"] = tool_calls
+        messages.append(msg)
+        return messages
+
+    @staticmethod
+    def add_tool_result(
+        messages: list[dict[str, Any]],
+        tool_call_id: str,
+        tool_name: str,
+        result: str,
+    ) -> list[dict[str, Any]]:
+        messages.append({
+            "role": "tool",
+            "tool_call_id": tool_call_id,
+            "name": tool_name,
+            "content": result,
+        })
+        return messages
+
+    # ------------------------------------------------------------------
+    # Internals
+    # ------------------------------------------------------------------
+
+    def _build_system_prompt(self, skills_summary: str) -> str:
+        ws = str(self.workspace.expanduser().resolve())
+        system = platform.system()
+        runtime = f"{'macOS' if system == 'Darwin' else system} {platform.machine()}"
+
+        parts: list[str] = [
+            f"# ToyClaw 🐾\n\n"
+            f"You are ToyClaw, a helpful AI assistant.\n\n"
+            f"## Runtime\n{runtime}, Python {platform.python_version()}\n\n"
+            f"## Workspace\n"
+            f"Your workspace is at: {ws}\n"
+            f"- Long-term memory: {ws}/memory/MEMORY.md\n"
+            f"- History log: {ws}/memory/HISTORY.md\n\n"
+            f"## Guidelines\n"
+            f"- State intent before tool calls, but NEVER predict results.\n"
+            f"- Before modifying a file, read it first.\n"
+            f"- If a tool call fails, analyze the error before retrying.\n"
+            f"- Ask for clarification when the request is ambiguous.",
+        ]
+
+        # Load workspace bootstrap files
+        for fname in _BOOTSTRAP_FILES:
+            fp = self.workspace / fname
+            if fp.exists():
+                parts.append(f"## {fname}\n\n{fp.read_text(encoding='utf-8')}")
+
+        # Memory context
+        mem_file = self.workspace / "memory" / "MEMORY.md"
+        if mem_file.exists():
+            mem = mem_file.read_text(encoding="utf-8").strip()
+            if mem:
+                parts.append(f"## Long-term Memory\n\n{mem}")
+
+        if skills_summary:
+            parts.append(f"## Skills\n\n"
+                         f"Read a skill's SKILL.md with read_file to use it.\n\n"
+                         f"{skills_summary}")
+
+        return "\n\n---\n\n".join(parts)
+
+    @staticmethod
+    def _runtime_context(channel: str | None, chat_id: str | None) -> str:
+        now = datetime.now().strftime("%Y-%m-%d %H:%M (%A)")
+        tz = time.strftime("%Z") or "UTC"
+        lines = [_RUNTIME_TAG, f"Time: {now} ({tz})"]
+        if channel and chat_id:
+            lines.append(f"Channel: {channel}  Chat: {chat_id}")
+        return "\n".join(lines)
+
+    @staticmethod
+    def strip_runtime_tag(content: str) -> str:
+        """Strip the runtime context prefix from stored user messages."""
+        if isinstance(content, str) and content.startswith(_RUNTIME_TAG):
+            parts = content.split("\n\n", 1)
+            return parts[1].strip() if len(parts) > 1 else ""
+        return content

assets/toyclaw_副本/toyclaw/cron.py

@@ -0,0 +1,241 @@
+"""Cron service — lightweight scheduled-task engine backed by asyncio."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import time
+import uuid
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Awaitable, Callable, Literal
+
+
+# ===================================================================
+# Data types
+# ===================================================================
+
+@dataclass
+class CronSchedule:
+    kind: Literal["at", "every", "cron"]
+    at_ms: int | None = None       # one-shot timestamp (ms)
+    every_ms: int | None = None    # interval (ms)
+    expr: str | None = None        # cron expression
+
+
+@dataclass
+class CronJob:
+    id: str
+    name: str
+    schedule: CronSchedule
+    message: str = ""
+    channel: str = "cli"
+    chat_id: str = "direct"
+    enabled: bool = True
+    delete_after_run: bool = False
+    next_run_ms: int | None = None
+    last_run_ms: int | None = None
+
+
+# ===================================================================
+# Helpers
+# ===================================================================
+
+def _now_ms() -> int:
+    return int(time.time() * 1000)
+
+
+def _next_run(sched: CronSchedule, now_ms: int) -> int | None:
+    """Compute the next fire time in milliseconds."""
+    if sched.kind == "at":
+        return sched.at_ms if sched.at_ms and sched.at_ms > now_ms else None
+    if sched.kind == "every" and sched.every_ms and sched.every_ms > 0:
+        return now_ms + sched.every_ms
+    if sched.kind == "cron" and sched.expr:
+        try:
+            from datetime import datetime
+
+            from croniter import croniter
+
+            base = datetime.fromtimestamp(now_ms / 1000).astimezone()
+            return int(croniter(sched.expr, base).get_next(datetime).timestamp() * 1000)
+        except Exception:
+            return None
+    return None
+
+
+# ===================================================================
+# Service
+# ===================================================================
+
+class CronService:
+    """Persists jobs to a JSON file; fires them via an asyncio timer."""
+
+    def __init__(
+        self,
+        store_path: Path,
+        on_job: Callable[[CronJob], Awaitable[str | None]] | None = None,
+    ):
+        self._path = store_path
+        self.on_job = on_job
+        self._jobs: list[CronJob] = []
+        self._running = False
+        self._timer: asyncio.Task[None] | None = None
+
+    # -- lifecycle -----------------------------------------------------
+
+    async def start(self) -> None:
+        self._load()
+        self._recompute()
+        self._save()
+        self._running = True
+        self._arm()
+
+    def stop(self) -> None:
+        self._running = False
+        if self._timer:
+            self._timer.cancel()
+            self._timer = None
+
+    # -- public API (called by CronTool) -------------------------------
+
+    def list_jobs(self) -> list[CronJob]:
+        self._load()
+        return [j for j in self._jobs if j.enabled]
+
+    def add_job(
+        self,
+        name: str,
+        schedule: CronSchedule,
+        message: str,
+        channel: str = "cli",
+        chat_id: str = "direct",
+        delete_after_run: bool = False,
+    ) -> CronJob:
+        self._load()
+        job = CronJob(
+            id=uuid.uuid4().hex[:8],
+            name=name,
+            schedule=schedule,
+            message=message,
+            channel=channel,
+            chat_id=chat_id,
+            delete_after_run=delete_after_run,
+            next_run_ms=_next_run(schedule, _now_ms()),
+        )
+        self._jobs.append(job)
+        self._save()
+        self._arm()
+        return job
+
+    def remove_job(self, job_id: str) -> bool:
+        self._load()
+        before = len(self._jobs)
+        self._jobs = [j for j in self._jobs if j.id != job_id]
+        if len(self._jobs) < before:
+            self._save()
+            self._arm()
+            return True
+        return False
+
+    # -- internals -----------------------------------------------------
+
+    def _load(self) -> None:
+        if not self._path.exists():
+            self._jobs = []
+            return
+        try:
+            raw = json.loads(self._path.read_text(encoding="utf-8"))
+            self._jobs = [
+                CronJob(
+                    id=j["id"],
+                    name=j["name"],
+                    schedule=CronSchedule(
+                        kind=j["schedule"]["kind"],
+                        at_ms=j["schedule"].get("at_ms"),
+                        every_ms=j["schedule"].get("every_ms"),
+                        expr=j["schedule"].get("expr"),
+                    ),
+                    message=j.get("message", ""),
+                    channel=j.get("channel", "cli"),
+                    chat_id=j.get("chat_id", "direct"),
+                    enabled=j.get("enabled", True),
+                    delete_after_run=j.get("delete_after_run", False),
+                    next_run_ms=j.get("next_run_ms"),
+                    last_run_ms=j.get("last_run_ms"),
+                )
+                for j in raw.get("jobs", [])
+            ]
+        except Exception:
+            self._jobs = []
+
+    def _save(self) -> None:
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        data = {
+            "jobs": [
+                {
+                    "id": j.id,
+                    "name": j.name,
+                    "schedule": {"kind": j.schedule.kind, "at_ms": j.schedule.at_ms,
+                                 "every_ms": j.schedule.every_ms, "expr": j.schedule.expr},
+                    "message": j.message,
+                    "channel": j.channel,
+                    "chat_id": j.chat_id,
+                    "enabled": j.enabled,
+                    "delete_after_run": j.delete_after_run,
+                    "next_run_ms": j.next_run_ms,
+                    "last_run_ms": j.last_run_ms,
+                }
+                for j in self._jobs
+            ]
+        }
+        self._path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
+
+    def _recompute(self) -> None:
+        now = _now_ms()
+        for j in self._jobs:
+            if j.enabled:
+                j.next_run_ms = _next_run(j.schedule, now)
+
+    def _arm(self) -> None:
+        """Schedule the next timer tick."""
+        if self._timer:
+            self._timer.cancel()
+        times = [j.next_run_ms for j in self._jobs if j.enabled and j.next_run_ms]
+        if not times or not self._running:
+            return
+        delay_s = max(0, (min(times) - _now_ms())) / 1000
+
+        async def _tick() -> None:
+            await asyncio.sleep(delay_s)
+            if self._running:
+                await self._fire_due()
+
+        self._timer = asyncio.create_task(_tick())
+
+    async def _fire_due(self) -> None:
+        self._load()
+        now = _now_ms()
+        due = [j for j in self._jobs if j.enabled and j.next_run_ms and now >= j.next_run_ms]
+        for job in due:
+            await self._execute(job)
+        self._save()
+        self._arm()
+
+    async def _execute(self, job: CronJob) -> None:
+        job.last_run_ms = _now_ms()
+        if self.on_job:
+            try:
+                await self.on_job(job)
+            except Exception:
+                pass  # best-effort
+
+        # Reschedule or disable
+        if job.schedule.kind == "at":
+            if job.delete_after_run:
+                self._jobs = [j for j in self._jobs if j.id != job.id]
+            else:
+                job.enabled = False
+                job.next_run_ms = None
+        else:
+            job.next_run_ms = _next_run(job.schedule, _now_ms())

assets/toyclaw_副本/toyclaw/heartbeat.py

@@ -0,0 +1,115 @@
+"""Heartbeat service — periodic agent wake-up."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from pathlib import Path
+from typing import Awaitable, Callable
+
+log = logging.getLogger(__name__)
+
+_HEARTBEAT_TOOL = [
+    {
+        "type": "function",
+        "function": {
+            "name": "heartbeat",
+            "description": "Report heartbeat decision after reviewing tasks.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "action": {
+                        "type": "string",
+                        "enum": ["skip", "run"],
+                        "description": "skip = nothing to do, run = has active tasks",
+                    },
+                    "tasks": {
+                        "type": "string",
+                        "description": "Summary of active tasks (required when action=run)",
+                    },
+                },
+                "required": ["action"],
+            },
+        },
+    }
+]
+
+
+class HeartbeatService:
+    """Reads HEARTBEAT.md periodically; asks the LLM to decide skip/run."""
+
+    def __init__(
+        self,
+        workspace: Path,
+        provider: "toyclaw.provider.OpenAIProvider",  # noqa: F821
+        on_execute: Callable[[str], Awaitable[str]] | None = None,
+        on_notify: Callable[[str], Awaitable[None]] | None = None,
+        interval_s: int = 1800,
+        enabled: bool = True,
+    ):
+        self.workspace = workspace
+        self.provider = provider
+        self.on_execute = on_execute
+        self.on_notify = on_notify
+        self.interval_s = interval_s
+        self.enabled = enabled
+        self._running = False
+        self._task: asyncio.Task[None] | None = None
+
+    @property
+    def _file(self) -> Path:
+        return self.workspace / "HEARTBEAT.md"
+
+    async def start(self) -> None:
+        if not self.enabled:
+            return
+        self._running = True
+        self._task = asyncio.create_task(self._loop())
+        log.info("Heartbeat started (every %ds)", self.interval_s)
+
+    def stop(self) -> None:
+        self._running = False
+        if self._task:
+            self._task.cancel()
+            self._task = None
+
+    # ------------------------------------------------------------------
+
+    async def _loop(self) -> None:
+        while self._running:
+            try:
+                await asyncio.sleep(self.interval_s)
+                if self._running:
+                    await self._tick()
+            except asyncio.CancelledError:
+                break
+            except Exception:
+                log.exception("Heartbeat error")
+
+    async def _tick(self) -> None:
+        if not self._file.exists():
+            return
+        content = self._file.read_text(encoding="utf-8").strip()
+        if not content:
+            return
+
+        # Phase 1: decision via tool call
+        resp = await self.provider.chat(
+            messages=[
+                {"role": "system", "content": "You are a heartbeat agent. Call the heartbeat tool to report your decision."},
+                {"role": "user", "content": f"Review HEARTBEAT.md and decide:\n\n{content}"},
+            ],
+            tools=_HEARTBEAT_TOOL,
+        )
+        if not resp.has_tool_calls:
+            return
+        args = resp.tool_calls[0].arguments
+        if args.get("action") != "run":
+            return
+
+        # Phase 2: execute
+        tasks = args.get("tasks", content)
+        if self.on_execute:
+            result = await self.on_execute(tasks)
+            if result and self.on_notify:
+                await self.on_notify(result)

assets/toyclaw_副本/toyclaw/provider.py

@@ -0,0 +1,98 @@
+"""LLM provider — thin wrapper around the OpenAI SDK."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from typing import Any
+
+from openai import AsyncOpenAI
+
+
+@dataclass
+class ToolCallRequest:
+    """A single tool-call request from the LLM."""
+
+    id: str
+    name: str
+    arguments: dict[str, Any]
+
+
+@dataclass
+class LLMResponse:
+    """Normalised response from an LLM provider."""
+
+    content: str | None = None
+    tool_calls: list[ToolCallRequest] = field(default_factory=list)
+    finish_reason: str = "stop"
+
+    @property
+    def has_tool_calls(self) -> bool:
+        return len(self.tool_calls) > 0
+
+
+class OpenAIProvider:
+    """OpenAI-compatible chat-completion provider.
+
+    Works with any endpoint that speaks the OpenAI API format
+    (OpenRouter, DeepSeek, local vLLM, etc.).
+    """
+
+    def __init__(
+        self,
+        api_key: str,
+        api_base: str = "https://api.openai.com/v1",
+        default_model: str = "gpt-4o",
+    ):
+        self.default_model = default_model
+        self._client = AsyncOpenAI(api_key=api_key, base_url=api_base)
+
+    async def chat(
+        self,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]] | None = None,
+        model: str | None = None,
+        temperature: float = 0.1,
+        max_tokens: int = 4096,
+    ) -> LLMResponse:
+        """Send a chat-completion request and return a normalised response."""
+        kwargs: dict[str, Any] = {
+            "model": model or self.default_model,
+            "messages": messages,
+            "temperature": temperature,
+            "max_tokens": max_tokens,
+        }
+        if tools:
+            kwargs["tools"] = tools
+            kwargs["tool_choice"] = "auto"
+        try:
+            resp = await self._client.chat.completions.create(**kwargs)
+            return self._parse(resp)
+        except Exception as exc:
+            return LLMResponse(content=f"Error calling LLM: {exc}", finish_reason="error")
+
+    # ------------------------------------------------------------------
+    # Internals
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _parse(resp: Any) -> LLMResponse:
+        choice = resp.choices[0]
+        msg = choice.message
+        tool_calls: list[ToolCallRequest] = []
+        if msg.tool_calls:
+            for tc in msg.tool_calls:
+                args = tc.function.arguments
+                if isinstance(args, str):
+                    try:
+                        args = json.loads(args)
+                    except json.JSONDecodeError:
+                        args = {}
+                tool_calls.append(
+                    ToolCallRequest(id=tc.id, name=tc.function.name, arguments=args)
+                )
+        return LLMResponse(
+            content=msg.content,
+            tool_calls=tool_calls,
+            finish_reason=choice.finish_reason or "stop",
+        )

assets/toyclaw_副本/toyclaw/session.py

@@ -0,0 +1,94 @@
+"""Session management — conversation history with JSONL persistence."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+
+@dataclass
+class Session:
+    """A single conversation session."""
+
+    key: str
+    messages: list[dict[str, Any]] = field(default_factory=list)
+    created_at: datetime = field(default_factory=datetime.now)
+    updated_at: datetime = field(default_factory=datetime.now)
+
+    def get_history(self, max_messages: int = 100) -> list[dict[str, Any]]:
+        """Return recent messages, aligned to start on a user turn."""
+        sliced = self.messages[-max_messages:]
+        # Drop leading non-user messages to avoid orphaned tool results
+        for i, m in enumerate(sliced):
+            if m.get("role") == "user":
+                sliced = sliced[i:]
+                break
+        out: list[dict[str, Any]] = []
+        for m in sliced:
+            entry: dict[str, Any] = {"role": m["role"], "content": m.get("content", "")}
+            for k in ("tool_calls", "tool_call_id", "name"):
+                if k in m:
+                    entry[k] = m[k]
+            out.append(entry)
+        return out
+
+    def clear(self) -> None:
+        self.messages.clear()
+        self.updated_at = datetime.now()
+
+
+class SessionManager:
+    """Manages sessions persisted as JSONL files."""
+
+    def __init__(self, workspace: Path):
+        self._dir = workspace / "sessions"
+        self._dir.mkdir(parents=True, exist_ok=True)
+        self._cache: dict[str, Session] = {}
+
+    def get_or_create(self, key: str) -> Session:
+        if key in self._cache:
+            return self._cache[key]
+        session = self._load(key) or Session(key=key)
+        self._cache[key] = session
+        return session
+
+    def save(self, session: Session) -> None:
+        path = self._path(session.key)
+        with open(path, "w", encoding="utf-8") as f:
+            meta = {
+                "_type": "metadata",
+                "key": session.key,
+                "created_at": session.created_at.isoformat(),
+                "updated_at": session.updated_at.isoformat(),
+            }
+            f.write(json.dumps(meta, ensure_ascii=False) + "\n")
+            for msg in session.messages:
+                f.write(json.dumps(msg, ensure_ascii=False) + "\n")
+        self._cache[session.key] = session
+
+    # ------------------------------------------------------------------
+
+    def _path(self, key: str) -> Path:
+        safe = key.replace(":", "_").replace("/", "_")
+        return self._dir / f"{safe}.jsonl"
+
+    def _load(self, key: str) -> Session | None:
+        path = self._path(key)
+        if not path.exists():
+            return None
+        try:
+            messages, created = [], None
+            for line in path.read_text(encoding="utf-8").splitlines():
+                if not line.strip():
+                    continue
+                data = json.loads(line)
+                if data.get("_type") == "metadata":
+                    created = datetime.fromisoformat(data["created_at"]) if data.get("created_at") else None
+                else:
+                    messages.append(data)
+            return Session(key=key, messages=messages, created_at=created or datetime.now())
+        except Exception:
+            return None

assets/toyclaw_副本/toyclaw/skills.py

@@ -0,0 +1,57 @@
+"""Skills discovery — scan workspace/skills/ for SKILL.md files."""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+
+class SkillsLoader:
+    """Discover and summarise skills for the system prompt."""
+
+    def __init__(self, workspace: Path):
+        self._dir = workspace / "skills"
+
+    def list_skills(self) -> list[dict[str, str]]:
+        """Return [{name, path, description}, ...] for each discovered skill."""
+        if not self._dir.is_dir():
+            return []
+        skills: list[dict[str, str]] = []
+        for d in sorted(self._dir.iterdir()):
+            if not d.is_dir():
+                continue
+            md = d / "SKILL.md"
+            if md.exists():
+                desc = self._extract_description(md)
+                skills.append({"name": d.name, "path": str(md), "description": desc})
+        return skills
+
+    def build_summary(self) -> str:
+        """Build a concise summary for the system prompt."""
+        skills = self.list_skills()
+        if not skills:
+            return ""
+        lines: list[str] = []
+        for s in skills:
+            lines.append(f"- **{s['name']}**: {s['description']}\n  Path: {s['path']}")
+        return "\n".join(lines)
+
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _extract_description(path: Path) -> str:
+        """Extract description from SKILL.md — first non-empty, non-heading line."""
+        try:
+            text = path.read_text(encoding="utf-8")
+            # Strip optional YAML frontmatter
+            if text.startswith("---"):
+                m = re.match(r"^---\n.*?\n---\n", text, re.DOTALL)
+                if m:
+                    text = text[m.end():]
+            for line in text.splitlines():
+                line = line.strip()
+                if line and not line.startswith("#"):
+                    return line[:120]
+            return path.parent.name
+        except Exception:
+            return path.parent.name

assets/toyclaw_副本/toyclaw/subagent.py

@@ -0,0 +1,125 @@
+"""Sub-agent manager — run tasks in background async tasks."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import uuid
+from pathlib import Path
+from typing import Any, Awaitable, Callable
+
+from toyclaw.provider import OpenAIProvider
+from toyclaw.tools.base import ToolRegistry
+from toyclaw.tools.builtin import (
+    EditFileTool,
+    ListDirTool,
+    ReadFileTool,
+    ShellTool,
+    WebFetchTool,
+    WebSearchTool,
+    WriteFileTool,
+)
+
+
+class SubagentManager:
+    """Spawns lightweight background sub-agents."""
+
+    def __init__(
+        self,
+        provider: OpenAIProvider,
+        workspace: Path,
+        on_complete: Callable[[str], Awaitable[None]] | None = None,
+        brave_api_key: str | None = None,
+        max_iterations: int = 15,
+    ):
+        self.provider = provider
+        self.workspace = workspace
+        self.on_complete = on_complete
+        self.brave_api_key = brave_api_key
+        self.max_iterations = max_iterations
+        self._tasks: dict[str, asyncio.Task[None]] = {}
+
+    async def spawn(
+        self,
+        task: str,
+        label: str | None = None,
+        origin_channel: str = "cli",
+        origin_chat_id: str = "direct",
+    ) -> str:
+        """Spawn a background sub-agent. Returns immediately."""
+        task_id = uuid.uuid4().hex[:8]
+        display = label or (task[:30] + "..." if len(task) > 30 else task)
+
+        bg = asyncio.create_task(self._run(task_id, task, display))
+        self._tasks[task_id] = bg
+        bg.add_done_callback(lambda _t: self._tasks.pop(task_id, None))
+
+        return f"Sub-agent [{display}] started (id: {task_id}). Will notify when done."
+
+    # ------------------------------------------------------------------
+
+    async def _run(self, task_id: str, task: str, label: str) -> None:
+        """Execute a sub-agent loop and announce the result."""
+        try:
+            result = await self._agent_loop(task)
+        except Exception as exc:
+            result = f"Error: {exc}"
+
+        announcement = f"🔔 Background task [{label}] finished:\n{result}"
+        if self.on_complete:
+            try:
+                await self.on_complete(announcement)
+            except Exception:
+                pass  # best-effort delivery
+
+    async def _agent_loop(self, task: str) -> str:
+        """Minimal agent loop (no spawn, no cron — prevent recursion)."""
+        tools = self._build_tools()
+        messages: list[dict[str, Any]] = [
+            {"role": "system", "content": self._system_prompt()},
+            {"role": "user", "content": task},
+        ]
+
+        for _ in range(self.max_iterations):
+            resp = await self.provider.chat(messages=messages, tools=tools.get_definitions())
+
+            if resp.has_tool_calls:
+                tc_dicts = [
+                    {
+                        "id": tc.id,
+                        "type": "function",
+                        "function": {
+                            "name": tc.name,
+                            "arguments": json.dumps(tc.arguments, ensure_ascii=False),
+                        },
+                    }
+                    for tc in resp.tool_calls
+                ]
+                messages.append({"role": "assistant", "content": resp.content, "tool_calls": tc_dicts})
+
+                for tc in resp.tool_calls:
+                    result = await tools.execute(tc.name, tc.arguments)
+                    messages.append({"role": "tool", "tool_call_id": tc.id, "name": tc.name, "content": result})
+            else:
+                return resp.content or "Task completed (no output)."
+
+        return "Sub-agent reached iteration limit."
+
+    def _build_tools(self) -> ToolRegistry:
+        """Build a restricted tool set — NO spawn, NO cron."""
+        reg = ToolRegistry()
+        ws = self.workspace
+        for cls in (ReadFileTool, WriteFileTool, EditFileTool, ListDirTool):
+            reg.register(cls(workspace=ws))
+        reg.register(ShellTool(workspace=ws))
+        reg.register(WebSearchTool(api_key=self.brave_api_key))
+        reg.register(WebFetchTool())
+        return reg
+
+    def _system_prompt(self) -> str:
+        return (
+            "# Sub-agent\n\n"
+            "You are a background sub-agent spawned to complete a specific task.\n"
+            "Stay focused. Your final text response will be reported back.\n\n"
+            f"## Workspace\n{self.workspace}"
+        )

assets/toyclaw_副本/toyclaw/tools/__init__.py

@@ -0,0 +1 @@
+"""Tool abstraction and registry."""

assets/toyclaw_副本/toyclaw/tools/base.py

@@ -0,0 +1,77 @@
+"""Tool base class and registry."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class Tool(ABC):
+    """Abstract base class for agent tools."""
+
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        ...
+
+    @property
+    @abstractmethod
+    def description(self) -> str:
+        ...
+
+    @property
+    @abstractmethod
+    def parameters(self) -> dict[str, Any]:
+        ...
+
+    @abstractmethod
+    async def execute(self, **kwargs: Any) -> str:
+        ...
+
+    def to_schema(self) -> dict[str, Any]:
+        """Convert to OpenAI function-calling schema."""
+        return {
+            "type": "function",
+            "function": {
+                "name": self.name,
+                "description": self.description,
+                "parameters": self.parameters,
+            },
+        }
+
+
+class ToolRegistry:
+    """Dynamic registry for agent tools."""
+
+    _HINT = "\n\n[Analyze the error above and try a different approach.]"
+
+    def __init__(self) -> None:
+        self._tools: dict[str, Tool] = {}
+
+    def register(self, tool: Tool) -> None:
+        self._tools[tool.name] = tool
+
+    def get(self, name: str) -> Tool | None:
+        return self._tools.get(name)
+
+    def get_definitions(self) -> list[dict[str, Any]]:
+        """Return all tool schemas in OpenAI format."""
+        return [t.to_schema() for t in self._tools.values()]
+
+    async def execute(self, name: str, params: dict[str, Any]) -> str:
+        """Look up a tool by *name*, run it, and return the string result."""
+        tool = self._tools.get(name)
+        if not tool:
+            avail = ", ".join(self._tools)
+            return f"Error: Tool '{name}' not found. Available: {avail}"
+        try:
+            result = await tool.execute(**params)
+            if isinstance(result, str) and result.startswith("Error"):
+                return result + self._HINT
+            return result
+        except Exception as exc:
+            return f"Error executing {name}: {exc}" + self._HINT
+
+    @property
+    def names(self) -> list[str]:
+        return list(self._tools)

assets/toyclaw_副本/toyclaw/tools/builtin.py

@@ -0,0 +1,452 @@
+"""Built-in tools: filesystem, shell, web, cron, spawn."""
+
+from __future__ import annotations
+
+import asyncio
+import difflib
+import html as html_mod
+import json
+import os
+import re
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+from urllib.parse import urlparse
+
+import httpx
+
+from toyclaw.tools.base import Tool
+
+if TYPE_CHECKING:
+    from toyclaw.cron import CronService
+    from toyclaw.subagent import SubagentManager
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _resolve(path: str, workspace: Path) -> Path:
+    p = Path(path).expanduser()
+    if not p.is_absolute():
+        p = workspace / p
+    return p.resolve()
+
+
+# ===================================================================
+# Filesystem tools
+# ===================================================================
+
+class ReadFileTool(Tool):
+    name = "read_file"
+    description = "Read the contents of a file."
+    parameters = {
+        "type": "object",
+        "properties": {"path": {"type": "string", "description": "File path to read"}},
+        "required": ["path"],
+    }
+    _MAX = 128_000
+
+    def __init__(self, workspace: Path):
+        self._ws = workspace
+
+    async def execute(self, path: str, **kw: Any) -> str:
+        try:
+            fp = _resolve(path, self._ws)
+            if not fp.exists():
+                return f"Error: File not found: {path}"
+            if not fp.is_file():
+                return f"Error: Not a file: {path}"
+            text = fp.read_text(encoding="utf-8")
+            if len(text) > self._MAX:
+                return text[: self._MAX] + f"\n... (truncated, {len(text)} chars total)"
+            return text
+        except Exception as exc:
+            return f"Error reading file: {exc}"
+
+
+class WriteFileTool(Tool):
+    name = "write_file"
+    description = "Write content to a file. Creates parent directories if needed."
+    parameters = {
+        "type": "object",
+        "properties": {
+            "path": {"type": "string", "description": "File path"},
+            "content": {"type": "string", "description": "Content to write"},
+        },
+        "required": ["path", "content"],
+    }
+
+    def __init__(self, workspace: Path):
+        self._ws = workspace
+
+    async def execute(self, path: str, content: str, **kw: Any) -> str:
+        try:
+            fp = _resolve(path, self._ws)
+            fp.parent.mkdir(parents=True, exist_ok=True)
+            fp.write_text(content, encoding="utf-8")
+            return f"Wrote {len(content)} bytes to {fp}"
+        except Exception as exc:
+            return f"Error writing file: {exc}"
+
+
+class EditFileTool(Tool):
+    name = "edit_file"
+    description = "Edit a file by replacing old_text with new_text (must match exactly)."
+    parameters = {
+        "type": "object",
+        "properties": {
+            "path": {"type": "string", "description": "File path"},
+            "old_text": {"type": "string", "description": "Exact text to find"},
+            "new_text": {"type": "string", "description": "Replacement text"},
+        },
+        "required": ["path", "old_text", "new_text"],
+    }
+
+    def __init__(self, workspace: Path):
+        self._ws = workspace
+
+    async def execute(self, path: str, old_text: str, new_text: str, **kw: Any) -> str:
+        try:
+            fp = _resolve(path, self._ws)
+            if not fp.exists():
+                return f"Error: File not found: {path}"
+            content = fp.read_text(encoding="utf-8")
+            if old_text not in content:
+                return self._hint(old_text, content, path)
+            if content.count(old_text) > 1:
+                return f"Error: old_text appears {content.count(old_text)} times; provide more context."
+            fp.write_text(content.replace(old_text, new_text, 1), encoding="utf-8")
+            return f"Edited {fp}"
+        except Exception as exc:
+            return f"Error editing file: {exc}"
+
+    @staticmethod
+    def _hint(old: str, content: str, path: str) -> str:
+        lines, old_lines = content.splitlines(True), old.splitlines(True)
+        best_ratio, best_i = 0.0, 0
+        for i in range(max(1, len(lines) - len(old_lines) + 1)):
+            r = difflib.SequenceMatcher(None, old_lines, lines[i : i + len(old_lines)]).ratio()
+            if r > best_ratio:
+                best_ratio, best_i = r, i
+        if best_ratio > 0.5:
+            diff = "\n".join(difflib.unified_diff(
+                old_lines, lines[best_i : best_i + len(old_lines)],
+                fromfile="old_text", tofile=f"{path} (line {best_i + 1})", lineterm="",
+            ))
+            return f"Error: old_text not found in {path}.\nBest match ({best_ratio:.0%}) at line {best_i + 1}:\n{diff}"
+        return f"Error: old_text not found in {path}. No similar text found."
+
+
+class ListDirTool(Tool):
+    name = "list_dir"
+    description = "List directory contents."
+    parameters = {
+        "type": "object",
+        "properties": {"path": {"type": "string", "description": "Directory path"}},
+        "required": ["path"],
+    }
+
+    def __init__(self, workspace: Path):
+        self._ws = workspace
+
+    async def execute(self, path: str, **kw: Any) -> str:
+        try:
+            dp = _resolve(path, self._ws)
+            if not dp.is_dir():
+                return f"Error: Not a directory: {path}"
+            items = sorted(dp.iterdir())
+            if not items:
+                return f"Directory {path} is empty"
+            return "\n".join(
+                f"{'📁' if i.is_dir() else '📄'} {i.name}" for i in items
+            )
+        except Exception as exc:
+            return f"Error listing directory: {exc}"
+
+
+# ===================================================================
+# Shell
+# ===================================================================
+
+_DENY_PATTERNS = [
+    r"\brm\s+-[rf]{1,2}\b",
+    r"\bdel\s+/[fq]\b",
+    r"\brmdir\s+/s\b",
+    r"(?:^|[;&|]\s*)format\b",
+    r"\b(mkfs|diskpart)\b",
+    r"\bdd\s+if=",
+    r">\s*/dev/sd",
+    r"\b(shutdown|reboot|poweroff)\b",
+    r":\(\)\s*\{.*\};\s*:",
+]
+
+
+class ShellTool(Tool):
+    name = "exec"
+    description = "Execute a shell command and return its output."
+    parameters = {
+        "type": "object",
+        "properties": {
+            "command": {"type": "string", "description": "Shell command to execute"},
+        },
+        "required": ["command"],
+    }
+
+    def __init__(self, workspace: Path, timeout: int = 60):
+        self._ws = workspace
+        self._timeout = timeout
+
+    async def execute(self, command: str, **kw: Any) -> str:
+        lower = command.strip().lower()
+        for pat in _DENY_PATTERNS:
+            if re.search(pat, lower):
+                return "Error: Command blocked by safety guard."
+        try:
+            proc = await asyncio.create_subprocess_shell(
+                command,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                cwd=str(self._ws),
+            )
+            try:
+                stdout, stderr = await asyncio.wait_for(proc.communicate(), self._timeout)
+            except asyncio.TimeoutError:
+                proc.kill()
+                return f"Error: Command timed out after {self._timeout}s"
+            parts: list[str] = []
+            if stdout:
+                parts.append(stdout.decode("utf-8", errors="replace"))
+            if stderr and stderr.strip():
+                parts.append(f"STDERR:\n{stderr.decode('utf-8', errors='replace')}")
+            if proc.returncode:
+                parts.append(f"\nExit code: {proc.returncode}")
+            result = "\n".join(parts) or "(no output)"
+            if len(result) > 10_000:
+                result = result[:10_000] + "\n... (truncated)"
+            return result
+        except Exception as exc:
+            return f"Error executing command: {exc}"
+
+
+# ===================================================================
+# Web
+# ===================================================================
+
+_UA = "Mozilla/5.0 (compatible; ToyClaw/0.1)"
+
+
+class WebSearchTool(Tool):
+    name = "web_search"
+    description = "Search the web via Brave Search API. Returns titles, URLs, snippets."
+    parameters = {
+        "type": "object",
+        "properties": {
+            "query": {"type": "string", "description": "Search query"},
+            "count": {"type": "integer", "description": "Number of results (1-10)"},
+        },
+        "required": ["query"],
+    }
+
+    def __init__(self, api_key: str | None = None):
+        self._key = api_key
+
+    async def execute(self, query: str, count: int = 5, **kw: Any) -> str:
+        key = self._key or os.environ.get("BRAVE_API_KEY", "")
+        if not key:
+            return "Error: Brave Search API key not configured."
+        count = max(1, min(count, 10))
+        try:
+            async with httpx.AsyncClient(timeout=10) as c:
+                r = await c.get(
+                    "https://api.search.brave.com/res/v1/web/search",
+                    params={"q": query, "count": count},
+                    headers={"Accept": "application/json", "X-Subscription-Token": key},
+                )
+                r.raise_for_status()
+            results = r.json().get("web", {}).get("results", [])[:count]
+            if not results:
+                return f"No results for: {query}"
+            lines = [f"Results for: {query}\n"]
+            for i, item in enumerate(results, 1):
+                lines.append(f"{i}. {item.get('title', '')}\n   {item.get('url', '')}")
+                if desc := item.get("description"):
+                    lines.append(f"   {desc}")
+            return "\n".join(lines)
+        except Exception as exc:
+            return f"Error: {exc}"
+
+
+class WebFetchTool(Tool):
+    name = "web_fetch"
+    description = "Fetch a URL and extract readable text content."
+    parameters = {
+        "type": "object",
+        "properties": {
+            "url": {"type": "string", "description": "URL to fetch"},
+        },
+        "required": ["url"],
+    }
+    _MAX = 50_000
+
+    async def execute(self, url: str, **kw: Any) -> str:
+        parsed = urlparse(url)
+        if parsed.scheme not in ("http", "https"):
+            return f"Error: Only http/https allowed, got '{parsed.scheme}'"
+        try:
+            from readability import Document
+
+            async with httpx.AsyncClient(follow_redirects=True, timeout=30) as c:
+                r = await c.get(url, headers={"User-Agent": _UA})
+                r.raise_for_status()
+            ctype = r.headers.get("content-type", "")
+            if "application/json" in ctype:
+                text = json.dumps(r.json(), indent=2, ensure_ascii=False)
+            elif "text/html" in ctype or r.text[:256].lower().startswith(("<!doctype", "<html")):
+                doc = Document(r.text)
+                raw_html = doc.summary()
+                text = f"# {doc.title()}\n\n{self._strip(raw_html)}" if doc.title() else self._strip(raw_html)
+            else:
+                text = r.text
+            if len(text) > self._MAX:
+                text = text[: self._MAX] + "\n... (truncated)"
+            return text
+        except Exception as exc:
+            return f"Error fetching {url}: {exc}"
+
+    @staticmethod
+    def _strip(html: str) -> str:
+        """Strip HTML tags and collapse whitespace."""
+        text = re.sub(r"<script[\s\S]*?</script>", "", html, flags=re.I)
+        text = re.sub(r"<style[\s\S]*?</style>", "", text, flags=re.I)
+        text = re.sub(r"<[^>]+>", "", text)
+        text = html_mod.unescape(text)
+        return re.sub(r"\n{3,}", "\n\n", text).strip()
+
+
+# ===================================================================
+# Cron tool (interface to CronService)
+# ===================================================================
+
+class CronTool(Tool):
+    name = "cron"
+    description = "Schedule reminders and recurring tasks. Actions: add, list, remove."
+    parameters = {
+        "type": "object",
+        "properties": {
+            "action": {
+                "type": "string",
+                "enum": ["add", "list", "remove"],
+                "description": "Action to perform",
+            },
+            "message": {"type": "string", "description": "Reminder message (for add)"},
+            "every_seconds": {"type": "integer", "description": "Interval in seconds (recurring)"},
+            "cron_expr": {"type": "string", "description": "Cron expression, e.g. '0 9 * * *'"},
+            "at": {"type": "string", "description": "ISO datetime for one-time, e.g. '2026-03-10T10:30:00'"},
+            "job_id": {"type": "string", "description": "Job ID (for remove)"},
+        },
+        "required": ["action"],
+    }
+
+    def __init__(self, service: CronService):
+        self._svc = service
+        self._channel = "cli"
+        self._chat_id = "direct"
+
+    def set_context(self, channel: str, chat_id: str) -> None:
+        self._channel = channel
+        self._chat_id = chat_id
+
+    async def execute(self, action: str, **kw: Any) -> str:
+        if action == "add":
+            return self._add(**kw)
+        if action == "list":
+            return self._list()
+        if action == "remove":
+            return self._remove(kw.get("job_id"))
+        return f"Unknown action: {action}"
+
+    def _add(
+        self,
+        message: str = "",
+        every_seconds: int | None = None,
+        cron_expr: str | None = None,
+        at: str | None = None,
+        **_: Any,
+    ) -> str:
+        if not message:
+            return "Error: message is required"
+        from toyclaw.cron import CronSchedule
+
+        delete_after = False
+        if every_seconds:
+            sched = CronSchedule(kind="every", every_ms=every_seconds * 1000)
+        elif cron_expr:
+            sched = CronSchedule(kind="cron", expr=cron_expr)
+        elif at:
+            from datetime import datetime
+            try:
+                dt = datetime.fromisoformat(at)
+            except ValueError:
+                return f"Error: invalid datetime '{at}'"
+            sched = CronSchedule(kind="at", at_ms=int(dt.timestamp() * 1000))
+            delete_after = True
+        else:
+            return "Error: provide every_seconds, cron_expr, or at"
+        job = self._svc.add_job(
+            name=message[:30],
+            schedule=sched,
+            message=message,
+            channel=self._channel,
+            chat_id=self._chat_id,
+            delete_after_run=delete_after,
+        )
+        return f"Created job '{job.name}' (id: {job.id})"
+
+    def _list(self) -> str:
+        jobs = self._svc.list_jobs()
+        if not jobs:
+            return "No scheduled jobs."
+        return "Scheduled jobs:\n" + "\n".join(
+            f"- {j.name} (id: {j.id}, {j.schedule.kind})" for j in jobs
+        )
+
+    def _remove(self, job_id: str | None) -> str:
+        if not job_id:
+            return "Error: job_id is required"
+        return f"Removed job {job_id}" if self._svc.remove_job(job_id) else f"Job {job_id} not found"
+
+
+# ===================================================================
+# Spawn tool (interface to SubagentManager)
+# ===================================================================
+
+class SpawnTool(Tool):
+    name = "spawn"
+    description = (
+        "Spawn a background sub-agent for a complex or time-consuming task. "
+        "The sub-agent will complete the task independently and report back."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "task": {"type": "string", "description": "Task description for the sub-agent"},
+            "label": {"type": "string", "description": "Short label (for display)"},
+        },
+        "required": ["task"],
+    }
+
+    def __init__(self, manager: SubagentManager):
+        self._mgr = manager
+        self._channel = "cli"
+        self._chat_id = "direct"
+
+    def set_context(self, channel: str, chat_id: str) -> None:
+        self._channel = channel
+        self._chat_id = chat_id
+
+    async def execute(self, task: str, label: str | None = None, **kw: Any) -> str:
+        return await self._mgr.spawn(
+            task=task,
+            label=label,
+            origin_channel=self._channel,
+            origin_chat_id=self._chat_id,
+        )