# macOS 上 AI 代理的网关设置此技能为 Mac 上的 AI 编码代理构建持久网关。它将后台工作流程（Inngest、cron、管道）桥接到代理的会话中，并可选择将响应路由到外部通道（Telegram、WebSocket）。 ## 开始之前 **必填：** * macOS（首选 Apple Silicon） * pi 编码代理已安装并正在运行 * Redis 运行（本地、Docker 或 k8s — 如果需要，请参阅 `inngest-local` 技能） **可选（解锁更多功能）：** * Ingest 自托管（用于持久的事件驱动工作流程） * Tailscale（用于从手机/笔记本电脑进行安全远程访问） * Telegram 机器人令牌（用于移动通知/聊天） ## 关键设置注意事项 **“GATEWAY_ROLE=central”是始终在线会话所必需的。**如果没有它，会话将作为卫星运行，并会错过心跳、系统警报以及任何不专门针对它的事件。启动时设置： GATEWAY_ROLE=中央 pi **当 Inngest 在 Docker 中运行并且工作线程在主机上运行时，`serveHost` 是必需的。** SDK 通告 `localhost:3100` 作为其回调 URL，但 Docker 无法到达主机的环回。将其设置在您的 Hono 服务处理程序中： inngestServe({ 客户：摄取，功能，服务主机：“http://host.docker.internal:3100”， }) 然后强制重新同步： `curl -X PUT http://localhost:3100/api/inngest` **Bun 中的 ioredis 解析不稳定。** 如果出现“找不到模块 '@ioredis/commands'”，请显式安装它：面包添加@ioredis/命令 # 或：rm -rf node_modules && 包安装 **发布/订阅需要两个 ioredis 客户端。** 订阅的客户端无法运行 LRANGE、DEL 或其他命令。该扩展创建单独的“sub”和“cmd”客户端。 ## 意图对齐在构建任何内容之前，请询问用户这些问题以确定范围。根据他们的答案进行调整。 ### 问题 1：你的目标是什么？呈现这些选项： 1. **仅通知** — 后台作业完成，我想在不看终端的情况下了解它 2. **永远在线的代理** — 我想要一个持久会话，能够在终端关闭后继续存在、处理心跳、路由事件 3. **完整网关** — 始终在线 + 通过 Telegram/电话 + 多会话路由与我的代理交谈每个级别都建立在前一个级别的基础上。从他们现在需要的东西开始。 ### 问题 2：您的事件源是什么？ 1. **只是 cron/定时器** — 我想要一个定期检查系统运行状况的心跳 2. **Inngest 函数** — 我有持久的工作流程，可以发出完成事件 3. **混合** — Inngest + cron + 也许是 webhooks ### 问题 3：有多少个并发代理会话？ 1. **One** — 我一次运行一个 pi 会话 2. **多个** — 我经常在不同的终端上进行 2-5 个会话，处理不同的事情如果有多个：启用中央/卫星路由。如果是：更简单的单会话模式。 ## 架构层 ### 第 1 层：通知桥（最简单） **您得到什么：** 后台事件在您的 pi 会话中显示为消息。 **组件：** * Redis（已经运行） * 网关 pi 扩展（~100 行） *`pushGatewayEvent()`实用函数 **它是如何工作的：** 后台进程 → Redis LPUSH → pi 扩展在通知上耗尽 → 作为用户消息注入 **构建步骤：** 1.创建扩展目录： mkdir -p ~/.pi/agent/extensions/gateway 2. 创建`~/.pi/agent/extensions/gateway/package.json`： { "name": "网关扩展", “私人”：真实， “依赖项”：{ “ioredis”：“^5.4.2” } } 3.安装依赖项： cd ~/.pi/agent/extensions/gateway && npm install 4. 使用最小的桥创建 `~/.pi/agent/extensions/gateway/index.ts`：从“@mariozechner/pi-coding-agent”导入类型{ExtensionAPI，ExtensionContext}； const SESSION_ID = "主"; const EVENT_LIST = `代理：事件：${SESSION_ID}`; const NOTIFY_CHANNEL = `代理：通知：${SESSION_ID}`; 类型 RedisLike = { on(事件: 字符串, 监听器: (...args: 未知[]) => void): void; connect(): Promise; 订阅（频道：字符串）：Promise<未知>； lrange(key: string, start: number, stop: number): Promise; del(key: string): Promise<数字>; llen(key: string): Promise<数字>; 取消订阅（）：无效；断开连接（）：无效； }; 类型RedisCtor = new（选项：{主机：字符串；端口：数字；lazyConnect：布尔值}）=> RedisLike；让Redis：RedisCtor |空=空；让子：RedisLike |空=空；让 cmd: RedisLike |空=空；让 ctx: ExtensionContext |空=空；让 piRef: ExtensionAPI |空=空；接口系统事件{ id：字符串；类型：字符串；来源：字符串；有效负载：记录<字符串，未知>； ts：数字； } 函数formatEvents（事件：SystemEvent []）：字符串{ 返回 events.map((e) => { const time = new Date(e.ts).toLocaleTimeString("en-US", { hour12: false }); 返回 `- **[${time}] ${e.type}** (${e.source})`; }).join("\n"); } 异步函数rain(): Promise { if (!cmd || !piRef) 返回； const raw = wait cmd.lrange(EVENT_LIST, 0, -1); if (raw.length === 0) 返回; const events = raw.reverse().map(r => { 尝试 { 返回 JSON.parse(r) 作为 SystemEvent; } catch { 返回 null; } }).filter(Boolean) as SystemEvent[]; if (events.length === 0) { 等待 cmd.del(EVENT_LIST);返回; } 常量提示 = [ `## 🔔 ${events.length} 事件 — ${new Date().toISOString()}`, "", formatEvents(事件), "", “如果需要，请采取行动，否则请简要确认。”, ].join("\n"); if (ctx?.isIdle()) { piRef.sendUserMessage(提示); } 否则{ piRef.sendUserMessage(prompt, { DeliverAs: "followUp" }); } 等待 cmd.del(EVENT_LIST); } 导出默认函数（pi：ExtensionAPI）{ piRef = pi； pi.on("session_start", async (_event, _ctx) => { ctx = _ctx; 如果（！Redis）{ 尝试{ Redis = (等待 import("ioredis")).default as RedisCtor; } 捕获（错误）{ const 消息 = 错误实例错误？错误消息：字符串（错误）； _ctx.ui.notify(`网关扩展在没有 Redis 的情况下运行：${message}`, "警告"); 返回； } } sub = new Redis({ 主机：“localhost”，端口：6379，lazyConnect：true }); cmd = new Redis({ 主机：“localhost”，端口：6379，lazyConnect：true }); 等待 sub.connect(); 等待 cmd.connect(); 等待 sub.subscribe(NOTIFY_CHANNEL); sub.on("消息", () => { if (ctx?.isIdle())rain(); }); // 清空会话关闭时累积的所有内容 const挂起=等待cmd.llen(EVENT_LIST); 如果（挂起> 0）等待排水（）； ctx.ui.setStatus("网关", "🔗已连接"); }); pi.on("agent_end", async () => {rain(); }); pi.on("session_shutdown", async () => { if (sub) { sub.unsubscribe();子.disconnect(); } if (cmd) { cmd.disconnect(); } }); } 5. 从任何脚本推送事件：从“ioredis”导入 Redis； const redis = new Redis(); 异步函数pushEvent（类型：字符串，源：字符串，有效负载= {}）{ const event = { id: crypto.randomUUID(), 类型, 源, 负载, ts: Date.now() }; 等待 redis.lpush("agent:events:main", JSON.stringify(event)); 等待 redis.publish("agent:notify:main", JSON.stringify({ type })); } // 示例：下载完成时通知等待 PushEvent("download.complete", "my-script", { file: "video.mp4" }); 6. 重新启动 pi — 扩展会自动加载。 ### 第 2 层：始终保持心跳 **添加：** Cron 心跳、看门狗故障检测、启动顺序。 **附加组件：** * HEARTBEAT.md 检查表（由代理在每次心跳时读取） * 扩展中的看门狗定时器 * tmux 或 launchd 用于持久化 **构建步骤（在第 1 层之上）：** 1. 创建 `~/HEARTBEAT.md` （或者无论您的代理的家在哪里）： # 心跳检查表 ## 系统健康状况 - [ ] Redis 可达 - [ ] 后台工作人员正在响应 - [ ] 没有卡住的作业 ## 待处理工作 - [ ] 检查收件箱中是否有未处理的项目如果没有什么需要注意的，回复HEARTBEAT_OK。 2. 添加心跳 cron — 如果使用 Inngest：导出 const heartbeatCron = inngest.createFunction( { id: "系统心跳" }, [{ cron: "*/15 * * * *" }], 异步（{步骤}）=> { 等待step.run（“推送心跳”，async（）=> { 等待pushEvent("cron.heartbeat", "inngest", {}); }); } ）；或者在没有 Inngest 的情况下，使用简单的 cron/setInterval： # crontab -e */15 * * * * redis-cli LPUSH 代理:事件:main '{"id":"'$(uuidgen)'","type":"cron.heartbeat","source":"cron","payload":{},"ts":'$(date +%s000)'}' && redis-cli PUBLISH 代理:notify:main '{"type":"cron.heartbeat"}' 3. 将看门狗添加到扩展中（在 session_start 之后插入）：常量 WATCHDOG_THRESHOLD_MS = 30 * 60 * 1000； // 15 分钟间隔的 2 倍让lastHeartbeatTs = Date.now(); 让 watchdogAlarmFired = false; 设置间隔（（）=> { if (!piRef || !ctx) 返回； if (watchdogAlarmFired) 返回； const elapsed = Date.now() - lastHeartbeatTs; 如果（已过去> WATCHDOG_THRESHOLD_MS）{ watchdogAlarmFired = true； piRef.sendUserMessage(`## ⚠️ MISSED HEARTBEAT\n\n${Math.round(elapsed / 60000)} 分钟内没有心跳。检查您的worker/cron。`); } }, 5 * 60 * 1000); // 收到心跳后重置（在排出函数内）： // if (events.some(e => e.type === "cron.heartbeat")) { //lastHeartbeatTs = Date.now(); // watchdogAlarmFired = false; // } 4. 使其在终端关闭后仍然存在 - tmux： tmux 新会话 -d -s 代理 -x 120 -y 40“pi” # 附加：tmux Attach -t 代理 # 分离：Ctrl-B、D 或者 launchd 来实现完全永远在线： <字典> 标签com.you.agent-gateway 程序参数 <数组> <字符串>/bin/bash <字符串>-c tmux new-session -d -s agent -x 120 -y 40 "GATEWAY_ROLE=central pi" && 而 tmux has-session -t agent 2>/dev/null;睡5；完成 RunAtLoad 保持活动 StandardOutPath/tmp/agent-gateway.log StandardErrorPath/tmp/agent-gateway.log 加载它： launchctl 加载 ~/Library/LaunchAgents/com.you.agent-gateway.plist ### 第 3 层：多会话 + 中央/卫星 **添加：** 多个 pi 会话、智能事件路由。 **当您需要时：** 您同时运行 2 个以上 pi 会话 — 一个用于监督，其他用于编码任务。心跳应该只发送到监督会话。 **与第 1 层相比的主要变化：** 1. Session ID 变得基于角色： const ROLE = process.env.GATEWAY_ROLE ?? “卫星”; const SESSION_ID = ROLE ===“中央”？ "网关" : `pid-${process.pid}`; 2. Session在Redis集中注册：等待 cmd.sadd("代理:网关:会话", SESSION_ID); // 关闭时：等待 cmd.srem("代理:网关:会话", SESSION_ID); 3. `pushEvent` 扇出到目标：异步函数pushEvent（类型，源，有效负载，originSession？）{ const event = { id: crypto.randomUUID(), 类型, 源, 负载, ts: Date.now() }; const json = JSON.stringify(事件); const session = wait redis.smembers("agent:gateway:sessions"); const 目标 = new Set<字符串>(); if (sessions.includes("gateway")) Targets.add("gateway"); // 始终居中 if (originSession &&sessions.includes(originSession)) Targets.add(originSession); for (目标的 const sid) { 等待 redis.lpush(`代理:事件:${sid}`, json); 等待 redis.publish(`agent:notify:${sid}`, JSON.stringify({ type })); } } 4. 开始您的中央会话： GATEWAY_ROLE=central pi # 这个获取所有事件 5. 其他会话正常开始（卫星）： pi # 只获取它发起的事件 ### 第 4 层：外部渠道（Telegram、WebSocket） **添加：** 通过手机与您的代理交谈。这一层需要嵌入式守护进程方法——pi 作为 Node.js 进程内的库运行，而不是作为 TUI。有关完整架构，请参阅 joelclaw.com 文章“为您的 AI 代理构建网关”。 **关键部件：** * 来自 pi SDK 的 `createAgentSession()` — 无头代理会话 * grammY 代表 Telegram 机器人 * 序列化所有输入的命令队列（TUI、心跳、电报） * 出站路由器将响应发送回请求通道这是最复杂的一层。仅当您确实需要移动访问时才构建它。 ## 验证清单设置完成后，验证： * Pi 会话启动并加载扩展（检查状态栏🔗） * 推送测试事件: `redis-cli LPUSH agent:events:main '{"id":"test","type":"test","source":"manual","payload":{},"ts":0}'` \+ `redis-cli PUBLISH agent:notify:main test` * 事件在几秒钟内出现在 pi 会话中 *（第 2 层+）心跳按计划触发 *（第 2 层+）终止心跳源 — 看门狗警报在 30 分钟后触发 *（第 3 层+）中央会话接收心跳，卫星会话不接收心跳 ## 设置脚本（curl-first）对于自动设置，用户可以运行：卷曲-sL joelclaw.com/scripts/gateway-setup.sh |巴什或者具有特定级别：卷曲-sL joelclaw.com/scripts/gateway-setup.sh | bash -s -- 2 该脚本是幂等的，可检测 Redis、安装扩展并为第 2+ 层配置持久性。 ## 决策链（压缩 ADR）导致当前网关设计的顺序架构决策。每个都解决了上一次迭代中发现的实际问题。＃|美国存托凭证 |决定|问题已解决 |关键权衡 ---|---|---|---|--- 1 | [0010]() |混合 cron + 事件网关 |手动分流瓶颈|永远在线的法学硕士课程 = 昂贵。 Cron = 延迟。混合动力平衡了两者。 2 | [0018]() | Redis 事件桥（pi 扩展）| Inngest→pi 桥不存在 |仅扩展，没有单独的过程。 Redis 作为干净的接口边界。 3 | [0035]() |中环+卫星路由|心跳中断编码会话 |按角色扇出。中央获得一切，卫星只获得原点目标。 4 | [0036]()| launchd + tmux（已取代）|网关会话在终端关闭时终止 | Pi 需要 PTY。 tmux 提供了它。 launchd 在崩溃时重新启动。 5 | [0037]()| 3层看门狗| “谁看守守望者”|扩展看门狗（Inngest down）、launchd tripwire（pi down）、heartbeat（一切正常）。 6 | [0038]()|嵌入式 pi 守护进程（取代 0036）|没有移动访问，没有多渠道|将 pi 嵌入为库。 grammY 为电报。命令队列序列化所有输入。最复杂的层。 **完整上下文的阅读顺序：** 0010 → 0018 → 0035 → 0037 → 0038（跳过 0036，已取代） ## 已知限制 1. **排除竞争条件。** 扩展先执行“LRANGE”，然后执行“DEL”——不是原子的。这些调用之间推送的事件将被删除而不进行处理。内存中的“seenIds”重复数据删除可以防止重复传递，但不能防止丢失事件。修复：使用 `LRANGE` \+ `LTRIM` 或 Redis 事务。对单用户系统的影响很小，但却是真实的。 2. **Redis 连接恢复仅通知。** 如果 Redis 出现故障，扩展程序会捕获错误并记录下来，但不会自动重试或重新连接。 ioredis `retryStrategy` 在客户端级别处理重新连接，但中断期间累积的事件可能会丢失。 3. **看门狗间隔是硬编码的。** 检查间隔（5 分钟）和阈值（30 分钟）是扩展中的常量。应该可以通过环境变量或 Redis 配置进行配置。 4. **重启后不会持续进行重复数据删除。** `seenIds` 集存在于内存中，上限为 500。进程重新启动 = 重复数据删除重置。对于每 15 分钟心跳一次的用例来说，这很好。对于高频事件，它可能会导致重复。 ## 学分 * [OpenClaw]() — 网关守护进程模式、命令队列序列化、通道插件架构 * [pi 编码代理]() — 扩展 API、sendUserMessage()、会话生命周期 * [Inngest]() — 持久的事件驱动工作流程 * [joelclaw.com/building-a-gateway-for-your-ai-agent]() — 人类摘要

网关设置

包含技能