iAs/DAEMON_WORKER_PLAN.md

iAs 守护进程 / 调度进程分离 — 实现计划

生成于 2026-06-02，供新会话参考当前架构状态。

---

目标

ias daemon (常驻)                    ias worker (短命，每条消息 spawn)
┌──────────────────────┐            ┌──────────────────────┐
│ WeChat 长轮询         │            │ LLM 对话              │
│ 消息收发              │   Unix     │ 工具调用              │
│ 审批匹配              │◄─Domain──►│ 无 DB 连接             │
│ DB 持有 / 上下文预载  │  Socket   │ 无状态（纯计算）       │
│ 消息队列（按用户串行） │            │ stdin 读取任务/环境    │
└──────────────────────┘            └──────────────────────┘

• daemon 稳定、不更新。持有 WeChat 长连接和数据库。
• worker 每次收到消息 spawn，完成后销毁。代码更新 → 编译 → 下一条消息自动用新 worker。
• IPC：Unix Domain Socket，长度前缀帧 + JSON 消息。

---

当前架构（供参考）

src/
├── main.rs          CLI 入口 + 监听循环 + 工具执行器
├── cli.rs           clap 子命令 (login/listen/send/whoami/usage/service)
├── db/
│   ├── mod.rs       Database 连接池
│   └── models.rs    CRUD (auth/chat/usage/summary/memories/approvals)
├── llm/
│   ├── types.rs     Message / Role / Usage / StreamChunk
│   ├── provider.rs  create_provider / parse_chat_chunk
│   ├── deepseek.rs  DeepSeek 流式客户端
│   └── conversation.rs  Conversation (chat + chat_with_tools 工具循环)
├── context/
│   ├── types.rs     ChatSession
│   ├── builder.rs   token budget 构建 + 摘要触发
│   └── tools.rs     MemoryStore
├── tools/
│   ├── types.rs     RiskLevel / SkillSpec / SkillResult / ExecutionContext
│   ├── builtin.rs   BuiltinRegistry (get_current_datetime / manage_memos / query_weather)
│   ├── weather.rs   QWeather 客户端 (JWT + reqwest)
│   └── approval.rs  ApprovalManager (确认码 + oneshot 通道)
└── wechat/
    ├── types.rs     iLink API 协议类型
    └── client.rs    HTTP 客户端 (login/poll/send/notify)

---

分阶段实现

Phase 1: 协议层 — IPC 帧格式

目标：实现与业务无关的 send_frame / recv_frame。

新文件：src/ipc.rs

// 长度前缀帧
// [4 字节 u32 BE: payload_len]
// [N 字节: JSON payload]

pub async fn send_frame(stream: &mut UnixStream, msg: &serde_json::Value) -> Result<()>
pub async fn recv_frame(stream: &mut UnixStream) -> Result<serde_json::Value>

测试：

cargo test ipc

---

Phase 2: Worker — 无状态执行进程

目标：ias worker 子命令，接收 task JSON，执行 LLM，输出结果帧。

新文件：src/worker.rs

// ias worker --sock /tmp/ias_daemon.sock
pub async fn run(sock_path: &str) -> Result<()> {
    let mut stream = UnixStream::connect(sock_path).await?;

    // 1. 读 task
    let task: TaskFrame = recv_frame(&mut stream).await?;

    // 2. 注入环境变量（从 task.env 字段）
    for (k, v) in &task.env { std::env::set_var(k, v); }

    // 3. 构建 Conversation（复用现有代码）
    let mut conv = Conversation::new(config)?;
    conv.session().load_history(&task.history);
    conv.session().load_memories(&task.memories);
    conv.session().load_summaries(&task.summaries);

    // 4. 注册工具执行器（复用现有代码）
    conv.set_tool_executor(build_executor());

    // 5. LLM 对话 + 工具循环
    let (reply, used_tools, usage) = conv.chat_with_tools(&task.msg.text).await?;

    // 6. 输出结果帧
    send_frame(&mut stream, &OutputFrame::usage(&usage)).await?;
    send_frame(&mut stream, &OutputFrame::reply(&reply)).await?;
    send_frame(&mut stream, &OutputFrame::bye()).await?;

    Ok(())
}

改动：

• cli.rs：新增 Worker { sock_path: String } 子命令
• main.rs：Commands::Worker { sock_path } => worker::run(&sock_path).await

不依赖 daemon，可独立测试：

# 终端 1：模拟 daemon
socat UNIX-LISTEN:/tmp/ias.sock,fork EXEC:'echo done'

# 终端 2：运行 worker
ias worker --sock /tmp/ias.sock

---

Phase 3: Daemon — 常驻进程 + 消息队列

目标：ias daemon 子命令，持有 WeChat + DB，spawn worker。

新文件：src/daemon.rs

pub async fn run(db: Arc<Database>, client: WeChatClient) -> ! {
    let sock_path = "/tmp/ias_daemon.sock";
    let _ = std::fs::remove_file(sock_path);
    let listener = UnixListener::bind(sock_path)?;

    let queue = MessageQueue::new();

    loop {
        tokio::select! {
            // WeChat 消息
            msg = recv_message(&client) => {
                if is_approval_reply(&msg) {
                    handle_approval(&msg).await;
                } else {
                    queue.enqueue(msg);
                    process_queue(&queue, &db).await;
                }
            }
            // Worker 连接（worker 启动后 connect）
            (stream, _) = listener.accept() => {
                let user_id = queue.pop_waiting();
                spawn_process(&user_id, stream, &db).await;
            }
        }
    }
}

消息队列设计（daemon.rs 内部）：

struct MessageQueue {
    // 每个用户最多一个活跃 worker
    active: HashSet<UserId>,
    // 等待队列
    pending: HashMap<UserId, VecDeque<IncomingMessage>>,
    // 准备就绪待处理的用户
    waiting: VecDeque<UserId>,
}

Daemon 处理一个消息的完整流程：

async fn spawn_process(user_id, stream, db) {
    // 1. 预载上下文（从 DB）
    let history = db::models::list_recent_chat_records(&db.pool, &user_id, 20).await?;
    let memories = db.memory_store.read_for(&user_id).await;
    let summaries = db::models::load_summaries(&db.pool, 5).await?;

    // 2. 构建 env（从当前进程环境变量）
    let env = HashMap::from([
        ("DEEPSEEK_API_KEY", std::env::var("DEEPSEEK_API_KEY")?),
        ("DEEPSEEK_MODEL", std::env::var("DEEPSEEK_MODEL")?),
        ("QWEATHER_API_HOST", std::env::var("QWEATHER_API_HOST")?),
        // ...
    ]);

    // 3. 发送 task 帧
    send_frame(&mut stream, &TaskFrame { user_id, msg, history, memories, summaries, env }).await?;

    // 4. 逐帧处理 worker 输出
    loop {
        match recv_frame(&mut stream).await? {
            OutputFrame::Chunk { text } => {
                // 暂存，等最终 reply 一起发送
                reply_buffer.push_str(&text);
            }
            OutputFrame::DbWrite { table, row } => {
                execute_db_write(&db, table, row).await;
            }
            OutputFrame::Reply { text } => {
                client.send_text(&user_id, &text, &ctx_token).await?;
                // 发送成功后记录聊天
            }
            OutputFrame::NeedApproval { tool, code, message } => {
                approval_manager.create(&user_id, &tool, &code).await?;
                client.send_text(&user_id, &message, None).await?;
                // Worker 退出。审批完成后再 spawn 新 worker。
                break;
            }
            OutputFrame::Bye => break,
        }
    }
}

改动：

• cli.rs：新增 Daemon 子命令（替代 Service）
• main.rs：Commands::Daemon => daemon::run(db, client).await

---

Phase 4: 审批跨进程

目标：worker 遇 High 风险工具 → 输出 need_approval 帧 → daemon 处理 → 重新 spawn worker。

流程：

Worker                          Daemon
  │                               │
  ├─ LLM 请求 manage_memos       │
  ├─ 遇 High 风险                 │
  ├─ write(need_approval) ──────→├─ 读 need_approval
  ├─ write(bye) ────────────────→├─ 创建审批记录到 DB
  │                               ├─ 发确认消息到微信
  │   (Worker 退出)               │
  │                               ├─ 等待用户回复...
  │                               ├─ 收到确认码 → 匹配
  │                               │
  │   (新 Worker spawn)           │
  │◄───── write(task + approved)─├─ task 帧中带 approved_tool
  ├─ LLM 继续执行                 │
  ├─ write(reply) ──────────────→├─ 发最终回复

task 帧扩展：

{
  "user_id": "...",
  "msg": {...},
  "history": [...],
  "approved_tool": "manage_memos",   // ← 新增
  "env": {...}
}

---

Phase 5: 流式输出

目标：LLM 每生成一个 token，通过 UDS 帧实时推送给 daemon，daemon 可选实时推用户。

当前行为：等全部生成完 → 一次发送。 流式行为：逐 chunk 发送 → daemon 攒到换行或完成 → 发送微信。

{"type":"chunk","text":"北"}
{"type":"chunk","text":"京今"}
{"type":"chunk","text":"天晴"}
// 可攒多个 chunk 再发一帧以减少帧开销
{"type":"chunk","text":"北京今天晴天，25°C"}
{"type":"reply","text":"北京今天晴天，25°C，微风。"}

微信不原生支持流式推送，所以收益有限。作为最后阶段可选项。

---

消息协议完整定义

TaskFrame (Daemon → Worker)

{
  "type": "task",
  "user_id": "wxid_abc",
  "msg": {
    "from": "wxid_abc",
    "text": "北京天气",
    "account_id": "bot_123",
    "context_token": "token_xyz"
  },
  "history": [
    {"role": "user", "content": "你好"},
    {"role": "assistant", "content": "你好！有什么可以帮你？"}
  ],
  "memories": ["用户叫张伟", "住在北京"],
  "summaries": ["之前讨论过天气偏好..."],
  "approved_tool": null,
  "env": {
    "DEEPSEEK_API_KEY": "sk-xxx",
    "DEEPSEEK_MODEL": "deepseek-v4-flash",
    "QWEATHER_API_HOST": "ky5ctpp742.re.qweatherapi.com",
    "QWEATHER_JWT_KEY_ID": "KF5AJT3JE9",
    "QWEATHER_JWT_PROJECT_ID": "4E2DWXQAVM",
    "QWEATHER_JWT_PRIVATE_KEY_FILE": "/home/xiao/project/iAs/qweather/ed25519-private.pem"
  }
}

OutputFrame (Worker → Daemon)

{"type": "chunk", "text": "..."}

{"type": "db_write", "table": "chat_records", "row": {...}}
{"type": "db_write", "table": "llm_usage", "row": {...}}

{"type": "reply", "text": "最终回复文本"}

{"type": "need_approval", "tool": "manage_memos", "code": "A1B2C3",
 "message": "⚠️ 确认码 A1B2C3，回复确认码继续。"}

{"type": "bye"}

DB Write 表定义

table	row 字段
chat_records	direction, user_id, account_id, text, source, context_token, message_id
llm_usage	user_id, model, provider, prompt_tokens, completion_tokens, cache_hit_tokens, cache_miss_tokens
user_memories	user_id, content

---

文件清单

文件	Phase	说明
src/ipc.rs	1	UnixStream + 长度前缀帧 + send_frame/recv_frame
src/worker.rs	2	Worker 主逻辑：读 task → LLM → 写结果帧
src/daemon.rs	3	Daemon 主逻辑：WeChat 轮询 + 消息队列 + spawn worker
src/cli.rs	1-3	新增 Daemon、Worker 子命令，保留 Login/Send/Usage
src/main.rs	1-3	命令路由，移除 Service（由 Daemon 替代）
Cargo.toml	1	新增 tokio/io-util（如果有）

不改动的文件：

• src/llm/ — 全部复用
• src/tools/ — 全部复用
• src/db/ — 全部复用（仅 daemon 侧调用）
• src/wechat/ — 全部复用（仅 daemon 侧调用）
• src/context/ — 需小改（ChatSession 支持从历史加载）

---

测试方法

# 1. 启动 daemon
ias daemon --sock /tmp/ias.sock &

# 2. 手动测试 worker（不通过 daemon）
echo '{"type":"task","user_id":"test","msg":{"text":"北京天气"},...}' \
  | socat - UNIX-CONNECT:/tmp/ias.sock

# 3. 完整流程（模拟微信消息）
ias send wxid_test "北京天气"
# daemon 收到 → spawn worker → 回复

---

关键实现细节

ChatSession 从预载数据初始化

当前 ChatSession 在收到第一条消息时通过 load_recent_messages 从 DB 加载历史。Worker 模式需要从 task 帧直接从内存加载：

// 在 ChatSession 或 WorkerContext 中新增
pub fn load_from_history(&mut self, history: &[HistoryEntry]) {
    for entry in history {
        match entry.role {
            "user" => self.add_user(&entry.content),
            "assistant" => self.add_assistant(&entry.content),
            _ => {}
        }
    }
}

Worker 中的工具执行器

Worker 中注册的工具执行器不连接 DB：

// read_memories / write_memory → 返回消息让 daemon 处理
// 在 worker 上下文中，只返回 db_write 帧
fn build_worker_executor() -> ToolExecutor {
    Arc::new(move |name, args| {
        Box::pin(async move {
            match name {
                "read_memories" => {
                    // memories 已在 task 帧中预载 → 直接返回
                    Ok(format!("长期记忆: {:?}", loaded_memories))
                }
                "write_memory" => {
                    // 返回 db_write 帧，由 daemon 执行
                    Ok(json!({"type":"db_write","table":"user_memories","row":{...}}).to_string())
                }
                // 其他 builtin 工具直接执行
                _ => builtin::execute(name, args)
            }
        })
    })
}

---

工作量估计

Phase	内容	估计
1	IPC 帧格式	30 min
2	Worker 独立运行	1 h
3	Daemon + 消息队列	2 h
4	审批跨进程	1 h
5	流式输出	30 min（可选）
合计	Phase 1-4	~5 h