# iAs 进程拆分设计方案 ## 1. 设计结论 当前 `ias-service::run_service()` 同时承载微信长轮询、邮件轮询、AI 队列、HTTP 服务、control socket 和运行态快照，已经形成单进程强耦合。拆分后不再自研进程间队列，统一使用 **async-nats + NATS JetStream** 承担消息持久化、路由、Ack、重试和 DLQ。 目标边界： - **Daemon Control Plane**：只负责 worker 启停、健康检查、control socket、JetStream 初始化和状态汇总。 - **JetStream 数据面**：承载所有业务消息，worker 直接 publish/consume。 - **Worker**：承载业务逻辑，包括渠道、AI、工具、日志、定时任务和 HTTP。 Daemon 不调用 AI、不读写上下文、不调用微信、不处理邮件业务、不决定 typing 生命周期。 ## 2. 当前问题 - `core/queue.rs` 既是内存队列，又包含上下文记录、AI 调用、工具调用、typing 刷新和微信发送，职责过宽。 - 微信、邮件、AI、HTTP 共享同一 runtime，任何模块阻塞或崩溃都会影响整体。 - 邮件通知直接依赖微信 manager，缺少统一的消息路由。 - 现有 supervisor 只能管理单个后台服务进程，不能管理多个 worker。 - 内存 `mpsc` 不具备跨进程持久化、Ack、重试、DLQ 和恢复能力。 ## 3. 目标架构 保留一张核心图，后续实现以这张图为准。 ```mermaid flowchart TB User["微信用户"] --> WeChatAPI["WeChat API"] MailSrv["IMAP / 邮件服务器"] --> MailWorker["Mail Worker"] Browser["浏览器 / 管理页面"] --> HttpSvc["HTTP Service"] subgraph Control["控制面"] CLI["ias CLI"] Daemon["Daemon Control Plane
worker 启停 / health / control socket / JetStream bootstrap"] Runtime["runtime files
pid / state / logs"] CLI --> Daemon HttpSvc -- status --> Daemon Daemon --> Runtime end subgraph MQ["数据面：NATS JetStream"] Inbound["IA_INBOUND
inbound.wechat.* / inbound.mail.*"] Function["IA_FUNCTION
function.call.* / function.result.*"] Outbound["IA_OUTBOUND
outbound.wechat.* / outbound.mail.*"] Schedule["IA_SCHEDULE
schedule.trigger.*"] Logs["IA_LOGS
logs.*"] DLQ["IA_DLQ
dlq.*"] end subgraph Workers["Worker"] WeChatWorker["WeChat Worker"] MailWorker["Mail Worker"] Assistant["Assistant Worker"] FunctionWorker["Function Worker"] Scheduler["Scheduler Worker"] Logging["Logging Worker"] end subgraph Storage["存储"] Pg["PostgreSQL
账号 / 邮件 / 上下文 / 日志 / 定时任务"] JS["JetStream file store"] end WeChatAPI --> WeChatWorker WeChatWorker -- publish inbound --> Inbound MailWorker -- publish inbound/notification --> Inbound Inbound -- consume --> Assistant Assistant -- publish function call --> Function Function -- consume --> FunctionWorker FunctionWorker -- publish function result --> Function Function -- consume result --> Assistant Assistant -- publish outbound --> Outbound Outbound -- consume --> WeChatWorker WeChatWorker --> WeChatAPI Scheduler -- publish trigger --> Schedule Schedule --> Assistant Schedule --> FunctionWorker Daemon -. log .-> Logs WeChatWorker -. log .-> Logs MailWorker -. log .-> Logs Assistant -. log .-> Logs FunctionWorker -. log .-> Logs Scheduler -. log .-> Logs Logs --> Logging Logging --> Pg Inbound -. max deliver .-> DLQ Function -. max deliver .-> DLQ Outbound -. max deliver .-> DLQ Schedule -. max deliver .-> DLQ Inbound --- JS Function --- JS Outbound --- JS Schedule --- JS Logs --- JS DLQ --- JS Assistant <--> Pg WeChatWorker <--> Pg MailWorker <--> Pg Scheduler <--> Pg HttpSvc <--> Pg ``` ## 4. 核心组件 ### 4.1 Daemon Control Plane 职责： - 启动、停止、重启 worker 和 HTTP 服务。 - 初始化 JetStream stream、consumer、DLQ。 - 提供 control socket 给 CLI 和 HTTP 状态页。 - 汇总 worker 心跳、JetStream consumer lag、DLQ 数量和最近错误。 - 输出本地运行日志，并发布关键控制事件到 `IA_LOGS`。 禁止： - 不调用 `ias-ai`。 - 不调用 `ias-context`。 - 不调用 `ias-wechat`。 - 不解析业务消息 payload。 - 不直接消费或生产业务消息。 ### 4.2 JetStream 数据面 固定使用 `async-nats + NATS JetStream`。 Streams： - `IA_INBOUND`：渠道输入消息，例如 `inbound.wechat.*`、`inbound.mail.*`。 - `IA_FUNCTION`：工具调用和工具结果，例如 `function.call.*`、`function.result.*`。 - `IA_OUTBOUND`：渠道发送命令，例如 `outbound.wechat.*`。 - `IA_SCHEDULE`：定时任务触发事件，例如 `schedule.trigger.*`。 - `IA_LOGS`：结构化日志事件，例如 `logs.daemon.*`、`logs.worker.*`、`logs.audit.*`。 - `IA_DLQ`：死信消息，例如 `dlq.*`。 默认策略： - 使用 file storage。 - 使用 durable pull consumer。 - 使用 explicit ack。 - 普通消息 `AckWait` 默认 60 秒，AI 和工具类消息可单独加长。 - `MaxDeliver` 默认 5 次，超过后进入 `IA_DLQ`。 - 所有业务处理按 at-least-once 设计，下游必须幂等。 ### 4.3 Worker - **WeChat Worker**：微信长轮询、`updates_buf` 持久化、发布 inbound、消费 outbound、执行发送和 typing。 - **Mail Worker**：邮件轮询、邮件通知事件、邮件 inbound 事件。 - **Assistant Worker**：上下文、AI 调用、`/new`、active turn、typing 决策、工具编排、生成 outbound。 - **Function Worker**：执行工具、HTTP API、内部功能，返回 function result。 - **Scheduler Worker**：扫描定时任务，到期后发布 scheduled trigger。 - **Logging Worker**：消费 `IA_LOGS`，落库、写文件、聚合错误和审计记录。 - **HTTP Service**：REST API、SPA、日志/更新查询、daemon 状态代理。 ## 5. 消息标准 所有 JetStream payload 统一包一层业务消息信封，字段保持稳定： - `message_id`：消息幂等键。 - `topic`：业务 topic，与 JetStream subject 对齐。 - `trace_id`：跨 worker 链路追踪。 - `created_at_ms`：生成时间。 - `attempt_hint`：业务层可读的尝试次数提示，真实投递次数以 JetStream 为准。 - `payload`：业务内容。 常用幂等键： - 入站消息：`message_id = channel/account/external_message_id`。 - outbound 命令：`command_id`。 - function 调用：`call_id`。 - 定时触发：`fire_id = schedule_id + fire_time`。 - 日志事件：`event_id`。 消息内容可以使用 `serde_json`；不要在 daemon 中解析业务 payload。 ## 6. 日志设计 日志分两层： - **运行日志**：所有进程继续用 `tracing` 输出到 stdout/file，方便 systemd 和本地排障。 - **结构化事件日志**：关键事件发布到 `IA_LOGS`，由 Logging Worker 落库和归档。 日志事件来源： - daemon：进程启动/停止、worker 重启、JetStream 初始化、DLQ 操作。 - channel worker：长轮询错误、发送成功/失败、账号状态变化。 - assistant worker：AI 请求摘要、工具调用、上下文错误、`/new`。 - function worker：工具执行耗时、结果摘要、错误。 - scheduler worker：任务触发、跳过、重复触发拦截。 - HTTP service：管理操作、配置变更、API 错误。 日志要求： - 所有跨进程消息必须带 `trace_id`。 - `message_id`、`command_id`、`call_id`、`fire_id` 必须进入日志字段。 - 默认不记录完整用户正文，只记录摘要、长度和哈希。 - Logging Worker 失败不能阻塞业务流。 - `IA_LOGS` 必须设置保留周期和最大容量。 ## 7. 定时任务设计 Scheduler Worker 独立运行，不放在 daemon 内。 定时任务来源： - 数据库表，例如 `ias_schedules`。 - 少量系统级任务可来自配置文件或环境变量。 任务字段建议： - `schedule_id` - `name` - `schedule_type`：`cron` 或 `interval` - `expression` - `target_topic` - `payload` - `enabled` - `last_fired_at` - `next_fire_at` 触发流程： 1. Scheduler Worker 周期扫描启用任务。 2. 到期后生成 `fire_id = schedule_id + fire_time`。 3. 通过数据库唯一约束抢占触发权。 4. 发布 `ScheduledTrigger` 到 `IA_SCHEDULE`。 5. 下游 worker 根据 `target_topic` 处理。 防重复要求： - `fire_id` 必须唯一。 - JetStream publish 使用 `message_id` 去重窗口。 - 下游 worker 必须按 `fire_id` 幂等。 ## 8. 模块拆分 目标 crate： - `ias-daemon`：control daemon、worker registry、process manager、JetStream bootstrap。 - `ias-common`：消息类型、JetStream 薄封装、RuntimeSnapshot、control 协议。 - `ias-channel`：微信和邮件 worker。 - `ias-assistant`：AI、上下文、工具编排、typing 决策。 - `ias-function`：工具执行 worker。 - `ias-scheduler`：定时任务 worker。 - `ias-logging`：日志消费 worker。 - `ias-http`：HTTP 服务和 daemon 状态代理。 依赖约束： - `ias-daemon` 可以依赖 `ias-common` 和 `async-nats`。 - `ias-daemon` 禁止依赖 `ias-ai`、`ias-context`、`ias-wechat`、`ias-mail`。 - `ias-channel` 禁止依赖 `ias-ai` 和 `ias-context`。 - `ias-assistant` 禁止依赖 `ias-wechat`。 - `ias-logging` 和 `ias-scheduler` 不直接调用渠道 SDK 和 AI SDK。 ## 9. CLI 入口 保留现有账号管理命令，不复用 `ias channel` 作为 worker 启动命令。 新增或调整入口： - `ias service start|stop|restart|status|logs` - `ias daemon run` - `ias worker assistant` - `ias worker wechat --account-ids ...` - `ias worker mail` - `ias worker function --name ...` - `ias worker scheduler` - `ias worker logging` - `ias http --host 0.0.0.0:9003` ## 10. 部署配置 核心环境变量： - `NATS_URL=nats://127.0.0.1:4222` - `IAS_NATS_STREAM_PREFIX=IA` - `IAS_NATS_ACK_WAIT_SECS=60` - `IAS_NATS_MAX_DELIVER=5` - `IAS_CONTROL_SOCKET=/run/user/1000/ias/ias.sock` - `IAS_LOG_EVENT_ENABLED=true` - `IAS_LOG_EVENT_RETENTION_DAYS=30` - `IAS_SCHEDULER_ENABLED=true` - `IAS_SCHEDULER_TICK_SECS=10` - `HOST=0.0.0.0:9003` JetStream 部署方式： - NATS Server 作为独立基础设施部署，iAs 只通过 `NATS_URL` 连接。 - 开发和单机部署优先使用 Docker Compose，便于本地启动、清理和迁移。 - Docker Compose 中的 NATS 启动参数必须等价于 `nats-server --jetstream --store_dir=/data --http_port=8222`。 - Docker Compose 必须挂载持久化目录，例如 `./data/nats:/data`。 - 本机开发端口只绑定 `127.0.0.1:4222` 和 `127.0.0.1:8222`，不要暴露公网。 - 生产单机也可以用 systemd 管理 `nats-server`，推荐配置文件放在 `/etc/ias/nats-server.conf`，JetStream 数据放在 `/var/lib/ias/nats`。 - 跨主机或多节点 JetStream 集群不作为第一阶段要求；确实需要时必须补充鉴权、TLS、cluster routes、备份和监控设计。 - daemon 启动时只负责检查和创建 stream、consumer、DLQ，不负责启动 Docker 容器或 systemd 服务。 - NATS 不可用时，worker 必须退避重连；daemon 和 HTTP 状态接口显示 MQ 降级状态。 ## 11. 施工计划 Phase 0：边界收敛 - 确认 daemon 只做 control plane。 - 确认 JetStream stream、subject、consumer 命名。 - 确认 `worker` CLI 命名不与现有账号管理命令冲突。 Phase 1：公共协议 - 在 `ias-common` 定义业务消息信封、payload 类型、日志事件和定时触发事件。 - 封装 `JetStreamBus`，集中处理 publish、consumer、stream 初始化和错误映射。 - 增加消息序列化、幂等键生成和 subject 生成测试。 Phase 2：日志和定时基础 - 创建 `IA_LOGS` 和 `IA_SCHEDULE`。 - 实现 Logging Worker 原型。 - 实现 Scheduler Worker 原型。 - 将关键操作日志改成发布 `LogEvent`，保留本地 `tracing` 输出。 Phase 3：队列解耦 - 从 `core/queue.rs` 拆出 JetStream publish/consume 适配。 - 将 AI、上下文、工具编排、typing 决策迁移到 Assistant Worker。 - 将微信发送调用改成 `OutboundCommand`。 - 先保持单进程兼容模式，验证行为不变。 Phase 4：Daemon 和 Worker 提取 - 创建 `ias-daemon`，只迁移 control、worker registry、process manager、JetStream bootstrap。 - 创建 `ias-assistant`、`ias-channel`、`ias-scheduler`、`ias-logging`。 - 实现 worker 注册、心跳、JetStream Ack/Nack、断线重连。 Phase 5：HTTP 和部署 - 添加独立 `ias http` 入口。 - 添加 daemon control client。 - 更新 `service start/stop/status/logs`。 - 增加 NATS JetStream Docker Compose 和 systemd 部署说明。 Phase 6：端到端验证 - 微信消息到回复链路。 - AI 工具调用链路。 - worker 崩溃后的 redelivery。 - 定时任务触发和去重。 - 结构化日志落库和查询。 - HTTP 独立运行和 daemon 状态降级。 ## 12. 施工标准 ### 12.1 架构标准 - daemon 不在业务消息数据路径中。 - 所有业务消息必须经过 JetStream。 - worker 之间不得直接调用彼此内部 API。 - 新增业务能力优先通过新增 subject 和 worker 实现，不扩展 daemon 职责。 ### 12.2 依赖标准 - `ias-daemon` 只能依赖 `ias-common`、`async-nats`、日志和进程管理相关依赖。 - `ias-common` 可以定义协议和薄封装，但不能包含业务逻辑。 - 渠道 SDK 只允许出现在 channel worker。 - AI 和上下文依赖只允许出现在 assistant worker。 ### 12.3 消息标准 - 所有消息必须有 `message_id` 和 `trace_id`。 - 所有 outbound 必须有 `command_id`。 - 所有 function call 必须有 `call_id`。 - 所有定时触发必须有 `fire_id`。 - 所有 worker 按 at-least-once 投递语义实现幂等。 ### 12.4 日志标准 - 所有关键状态变化必须发 `LogEvent`。 - 日志字段必须包含相关 ID：`message_id`、`command_id`、`call_id`、`fire_id`。 - 默认不得记录完整用户正文。 - 日志流必须设置保留周期和容量上限。 ### 12.5 定时任务标准 - 定时任务触发权必须通过数据库唯一约束或等价机制保证。 - Scheduler Worker 可以多副本运行，但同一 `fire_id` 只能触发一次。 - 定时任务失败必须可重试或进入 DLQ。 ### 12.6 测试标准 - 每个 worker 至少有消息序列化和幂等测试。 - JetStream consumer Ack/Nack、redelivery、DLQ 必须有集成测试。 - daemon 边界必须通过依赖检查验证。 - 关键端到端链路必须有人工或自动化验证记录。 ### 12.7 部署标准 - NATS 必须作为独立进程或容器部署，不嵌入 iAs daemon。 - JetStream 必须开启 file storage，并配置持久化目录。 - 开发和单机默认使用 Docker Compose；生产单机可改用 systemd。 - NATS client 端口和监控端口默认只监听本机。 - NATS 跨主机访问必须启用鉴权；公网访问必须启用 TLS。 - iAs 代码只能负责连接、初始化和检查 JetStream，不能负责管理 Docker 或 systemd 生命周期。 ## 13. 验收标准 架构验收： - `ias-daemon` 编译依赖中不包含 `ias-ai`、`ias-context`、`ias-wechat`、`ias-mail`。 - 不存在自研 `ias-queue.sock` 业务数据面。 - 所有 MQ 访问通过 `async-nats + JetStream` 或 `ias-common` 的 JetStream 薄封装完成。 消息验收： - 没有 assistant worker 时，`IA_INBOUND` 能接收并保留消息。 - 没有 channel worker 时，`IA_OUTBOUND` 能保留 outbound 消息。 - worker 崩溃或断线后，未 Ack 消息能 redeliver。 - 超过最大投递次数的消息进入 `IA_DLQ`。 业务验收： - 微信消息能完成：inbound -> assistant -> outbound -> 微信回复。 - `/new` 仍能打断同一会话旧 turn。 - typing 由 assistant 生成命令，channel worker 执行。 - 邮件提醒不直接调用微信 manager，而是发布消息后由路由完成发送。 日志验收： - Logging Worker 能消费 `IA_LOGS`。 - 结构化日志能写入数据库或文件。 - 日志中能通过 `trace_id` 串起跨 worker 链路。 - 默认不会记录完整用户正文。 定时任务验收： - Scheduler Worker 能按 cron/interval 发布 `ScheduledTrigger`。 - 同一 `fire_id` 重复启动不会重复触发。 - 定时任务失败能重试或进入 DLQ。 运维验收： - HTTP 可以在 daemon 不在线时启动，状态接口返回降级结果。 - daemon 重启不影响 JetStream 中未 Ack 消息。 - Docker Compose 或 systemd 重启 NATS 后，JetStream file store 中的未处理消息仍保留。 - NATS 重启后 worker 能退避重连并恢复消费。 - NATS 端口默认只绑定本机；跨主机部署时必须验证鉴权配置。 - `service status` 能看到 daemon、worker、stream、consumer lag 和 DLQ 摘要。