800行代码实现 Open Claw 的 Tool、消息总线、子 Agent管理架构

export abstract class Tool {
  abstract readonly name: string;
  abstract readonly description: string;
  abstract readonly input_schema: AnthropicTool["input_schema"];

  abstract execute(args: Record<string, unknown>): Promise<unknown>;

  toSchema(): AnthropicTool {
    return {
      name: this.name,
      description: this.description,
      input_schema: this.input_schema,
    };
  }
}

export class ToolRegistry {
  private tools = new Map<string, Tool>();

  register(tool: Tool) {
    this.tools.set(tool.name, tool);
  }

  async execute(name: string, args: Record<string, unknown>) {
    const tool = this.tools.get(name);
    if (!tool) throw new Error(`Tool "${name}" not found`);
    return tool.execute(args);
  }

  getToolDefinition(): AnthropicTool[] {
    return Array.from(this.tools.values()).map((tool) => tool.toSchema());
  }

  exclude(names: string[]): ToolRegistry {
    const excludeSet = new Set(names);
    const filtered = new ToolRegistry();
    for (const [name, tool] of this.tools) {
      if (!excludeSet.has(name)) {
        filtered.register(tool);
      }
    }
    return filtered;
  }
}

const occurrences = content.split(oldText).length - 1;

if (occurrences === 0) {
  return `Error: old_text not found in ${filePath}`;
}
if (occurrences > 1) {
  return `Warning: old_text found ${occurrences} times in ${filePath}. Please provide a more unique text snippet. No changes made.`;
}

const updated = content.replace(oldText, newText);

const DANGEROUS_PATTERNS: RegExp[] = [
  /rm\s+(-[a-zA-Z]*f[a-zA-Z]*\s+)?(-[a-zA-Z]*r[a-zA-Z]*\s+)?\/($|\s)/,
  /rm\s+-[a-zA-Z]*rf?\s+~($|\/|\s)/,
  /mkfs\b/,
  /dd\s+if=/,
  /:\(\)\s*\{\s*:\|:\s*&\s*\}\s*;/,  // fork bomb
  />\s*\/dev\/[sh]d[a-z]/,            // 写入裸设备
  /chmod\s+-R\s+777\s+\//,
];

function truncateOutput(text: string): string {
  if (text.length <= MAX_OUTPUT_LENGTH) return text;
  const half = Math.floor(MAX_OUTPUT_LENGTH / 2);
  return (
    text.slice(0, half) +
    `\n\n--- truncated (${text.length} chars total) ---\n\n` +
    text.slice(-half)
  );
}

function htmlToText(html: string): string {
  return html
    .replace(/<script[\s\S]*?<\/script>/gi, "")
    .replace(/<style[\s\S]*?<\/style>/gi, "")
    .replace(/<(br|\/p|\/div|\/li|\/tr|\/h[1-6])[^>]*>/gi, "\n")
    .replace(/<[^>]+>/g, "")
    // HTML 实体解码...
    .replace(/[ \t]+/g, " ")
    .replace(/\n{3,}/g, "\n\n")
    .trim();
}

export class MessageTool extends Tool {
  constructor(private sendCallback: SendCallback) {
    super();
  }

  async execute(args: Record<string, unknown>): Promise<string> {
    const content = args.content as string;
    const channel = (args.channel as string) ?? "repl";
    const chatId = (args.chat_id as string) ?? "default";
    await this.sendCallback({ channel, chatId, content });
    return `Message sent to ${channel}:${chatId}`;
  }
}

private parseCronInterval(expr: string): number {
  const parts = expr.trim().split(/\s+/);
  if (parts.length !== 5) return 60_000;

  const [minute, hour] = parts;

  // */N * * * * → 每 N 分钟
  if (minute?.startsWith("*/") && hour === "*") {
    const n = parseInt(minute.slice(2), 10);
    if (!isNaN(n) && n > 0) return n * 60_000;
  }

  // 0 */N * * * → 每 N 小时
  if (minute === "0" && hour?.startsWith("*/")) {
    const n = parseInt(hour.slice(2), 10);
    if (!isNaN(n) && n > 0) return n * 3600_000;
  }

  if (minute === "*" && hour === "*") return 60_000;     // 每分钟
  if (minute === "0" && hour === "*") return 3600_000;    // 每小时
  if (minute === "0" && hour === "0") return 86400_000;   // 每天

  return 60_000; // 复杂表达式降级为每分钟
}

export class MessageBus {
  private listeners = new Map<string, Set<MessageHandler>>();
  private queue: InboundMessage[] = [];

  subscribe(channel: string, handler: MessageHandler): () => void {
    if (!this.listeners.has(channel)) {
      this.listeners.set(channel, new Set());
    }
    this.listeners.get(channel)!.add(handler);
    return () => { this.listeners.get(channel)?.delete(handler); };
  }

  async publish(message: InboundMessage): Promise<void> {
    const handlers = this.listeners.get(message.channel);
    if (handlers && handlers.size > 0) {
      for (const handler of handlers) {
        await handler(message);
      }
    } else {
      this.queue.push(message);
    }
  }

  drain(channel?: string): InboundMessage[] {
    if (!channel) {
      const msgs = [...this.queue];
      this.queue = [];
      return msgs;
    }
    const matched = this.queue.filter((m) => m.channel === channel);
    this.queue = this.queue.filter((m) => m.channel !== channel);
    return matched;
  }
}

const subagentTools = tools.exclude(["spawn", "message", "edit_file", "cron"]);

spawn(params: { task: string; label?: string; ... }): string {
  const id = `subagent-${++this.counter}`;
  const label = params.label ?? `Task ${this.counter}`;

  const promise = this.runSubagent(id, params.task, label, ...);

  promise.finally(() => {
    this.runningTasks.delete(id);
  });

  this.runningTasks.set(id, { id, label, promise });
  return id;
}

buildMessages: (_history, userMessage) => [
  { role: "user" as const, content: userMessage },
],

await this.bus.publish({
  channel: "system",
  senderId: "subagent",
  chatId: `${originChannel}:${originChatId}`,
  content: `[Subagent "${label}" (${id}) completed]\n\n${result}`,
});

let processing = false;
const pendingSubagentResults: InboundMessage[] = [];

async function drainPendingResults(): Promise<void> {
  while (pendingSubagentResults.length > 0) {
    const msg = pendingSubagentResults.shift()!;

    const systemContent = [
      "[SYSTEM NOTIFICATION - Subagent Result]",
      msg.content,
      "Please summarize the above subagent result for the user.",
    ].join("\n\n");

    const reply = await agent.run(systemContent, history);
    history.push({ role: "user", content: systemContent });
    history.push({ role: "assistant", content: reply });
    console.log(`\nBot > ${reply}\n`);
  }
}

async function tryDrainPending(): Promise<void> {
  if (processing) return;
  processing = true;
  try {
    await drainPendingResults();
  } finally {
    processing = false;
  }
  rl.prompt();
}

rl.on("line", async (input) => {
  // ...
  processing = true;
  try {
    const reply = await agent.run(trimmed, history);
    history.push({ role: "user", content: trimmed });
    history.push({ role: "assistant", content: reply });

    await drainPendingResults();
  } finally {
    processing = false;
  }
  rl.prompt();
});

bus.subscribe("system", (msg) => {
  pendingSubagentResults.push(msg);
  void tryDrainPending().catch(console.error);
});

tools.register(
  new MessageTool((msg) => {
    console.log(`\n[Message → ${msg.channel}:${msg.chatId}] ${msg.content}\n`);
  }),
);

终端 stdin
   │
   ▼
REPL 主循环 (index.ts)
   │  ┌──────────────────────── history (共享，互斥访问)
   │  │
   ▼  ▼
AgentLoop.run() ──→ Tool 调用
   │                  ├── 文件/命令/网络工具 → 直接返回结果
   │                  ├── SpawnTool → SubagentManager.spawn()
   │                  │                  └── 子 AgentLoop (独立 ReAct)
   │                  │                        └── bus.publish("system", 结果)
   │                  ├── MessageTool → sendCallback → stdout
   │                  └── CronTool → CronService
   │                                    └── setInterval → bus.publish("system", 触发通知)
   │
   │  ◄── bus.subscribe("system") ◄── pendingSubagentResults 队列
   │
   ▼
stdout 输出