# 08 - 并发安全策略

> 更新说明（2026-04-17）：
> 本文档中关于 `UDS` / `memory.sock` 的历史表述，现统一视为设计草案。
> 实际跨平台实现方案已切换为 `127.0.0.1` loopback 上的 `TCP` 动态端口，并通过本地 endpoint 文件做服务发现。

## 核心设计：Memory Actor 独立运行时

Memory 系统采用 **Actor 模型**独立运行，所有外部访问通过消息通道进行，彻底消除共享状态并发冲突。

### 为什么不用文件锁

| 方案 | 问题 |
|------|------|
| `RwLock<MemoryStore>` | 进程内可用，但跨进程（GUI + CLI + Server）无效 |
| 文件锁 (`fs2`) | NFS 不可靠、锁粒度粗、死锁风险、性能开销 |
| **Actor + Channel** | 单一 owner 串行处理，天然无冲突 |

### 设计原则

1. **单一写入者**：Memory 数据只有 Actor 内部可直接读写磁盘
2. **消息驱动**：外部通过 `MemoryHandle` 发送命令，Actor 串行处理
3. **读写分离**：查询返回 snapshot/clone，调用方拿到的数据不持有引用
4. **非阻塞主流程**：写入命令 fire-and-forget，不阻塞用户交互

---

## 架构总览

```
┌─────────────────────────────────────────────────┐
│  调用方（Runtime / Assembler / Organizer）        │
│                                                   │
│  let handle = MemoryHandle::clone();              │
│  let injection = handle.load_injection(...).await;│
│  handle.write_episode(episode);  // fire & forget │
└──────────────────┬────────────────────────────────┘
                   │ mpsc::Sender<MemoryCommand>
                   ▼
┌─────────────────────────────────────────────────┐
│  MemoryActor（独立 tokio task）                    │
│                                                   │
│  loop {                                           │
│    match cmd_rx.recv().await {                    │
│      LoadInjection { .. } => { ... }             │
│      WriteEpisode { .. } => { ... }              │
│      Recall { .. } => { ... }                    │
│      Ruminate { .. } => { ... }                  │
│      Shutdown => break,                           │
│    }                                              │
│  }                                                │
│                                                   │
│  内部持有：MemoryStore（索引 + 磁盘读写）           │
│           无需任何锁                               │
└─────────────────────────────────────────────────┘
```

---

## 消息协议

### 命令定义

```rust
use tokio::sync::{mpsc, oneshot};

/// Memory Actor 接收的命令
pub enum MemoryCommand {
    // ── 查询类（需要响应）──
    LoadInjection {
        session_id: String,
        workspace_id: Option<String>,
        reply: oneshot::Sender<Vec<String>>,
    },
    Recall {
        anchors: RecallAnchors,
        limit: usize,
        reply: oneshot::Sender<Vec<RecallHit>>,
    },
    LoadDepth2 {
        node_ids: Vec<String>,
        reply: oneshot::Sender<Vec<ExpandedMemory>>,
    },

    // ── 写入类（fire-and-forget）──
    WriteEpisode {
        episode: Episode,
    },
    SaveDecision {
        decision: Decision,
    },
    UpsertEntity {
        entity: Entity,
    },
    UpdateInjection {
        level: InjectionLevel,
        target_id: String,
        content: String,
    },

    // ── 反刍类 ──
    RunMicroRumination {
        turn_result: Box<TurnResult>,
    },
    RunMesoRumination {
        session_id: String,
        workspace_id: String,
    },
    RunMetaRumination,

    // ── 生命周期 ──
    Shutdown,
}

#[derive(Debug, Clone)]
pub enum InjectionLevel {
    Profile,
    Workspace,
    Session,
}
```

### 查询 vs 写入的区别

| 类型 | 机制 | 调用方是否等待 | 失败影响 |
|------|------|--------------|---------|
| 查询 | `oneshot::Sender` 回传结果 | 是（`.await`） | 返回空默认值 |
| 写入 | 发送即返回 | 否 | 静默日志警告 |
| 反刍 | 发送即返回 | 否 | 静默日志警告 |

---

## MemoryHandle（调用方持有）

```rust
/// Memory 系统的客户端句柄，可任意 Clone 跨线程使用
#[derive(Clone)]
pub struct MemoryHandle {
    tx: mpsc::Sender<MemoryCommand>,
}

impl MemoryHandle {
    /// 加载注入上下文（查询，需等待响应）
    pub async fn load_injection(
        &self,
        session_id: &str,
        workspace_id: Option<&str>,
    ) -> Vec<String> {
        let (reply_tx, reply_rx) = oneshot::channel();
        let cmd = MemoryCommand::LoadInjection {
            session_id: session_id.to_string(),
            workspace_id: workspace_id.map(String::from),
            reply: reply_tx,
        };
        if self.tx.send(cmd).await.is_err() {
            tracing::warn!("Memory Actor 已关闭，返回空注入");
            return Vec::new();
        }
        reply_rx.await.unwrap_or_default()
    }

    /// 执行粗召回（查询，需等待响应）
    pub async fn recall(
        &self,
        anchors: RecallAnchors,
        limit: usize,
    ) -> Vec<RecallHit> {
        let (reply_tx, reply_rx) = oneshot::channel();
        let cmd = MemoryCommand::Recall {
            anchors,
            limit,
            reply: reply_tx,
        };
        if self.tx.send(cmd).await.is_err() {
            return Vec::new();
        }
        reply_rx.await.unwrap_or_default()
    }

    /// 写入 Episode（fire-and-forget）
    pub fn write_episode(&self, episode: Episode) {
        let _ = self.tx.try_send(MemoryCommand::WriteEpisode { episode });
    }

    /// 触发 Micro 反刍（fire-and-forget）
    pub fn run_micro_rumination(&self, turn_result: TurnResult) {
        let _ = self.tx.try_send(MemoryCommand::RunMicroRumination {
            turn_result: Box::new(turn_result),
        });
    }

    /// 优雅关闭
    pub async fn shutdown(&self) {
        let _ = self.tx.send(MemoryCommand::Shutdown).await;
    }
}
```

---

## MemoryActor（独立运行时）

```rust
pub struct MemoryActor {
    store: MemoryStore,
    cmd_rx: mpsc::Receiver<MemoryCommand>,
}

impl MemoryActor {
    /// 启动 Actor，返回 Handle
    pub fn spawn(workspace_id: Option<String>) -> Result<MemoryHandle> {
        let (tx, rx) = mpsc::channel(256); // 命令队列容量
        let store = MemoryStore::load(&workspace_id)?;

        let actor = Self { store, cmd_rx: rx };
        tokio::spawn(actor.run());

        Ok(MemoryHandle { tx })
    }

    async fn run(mut self) {
        while let Some(cmd) = self.cmd_rx.recv().await {
            match cmd {
                MemoryCommand::LoadInjection { session_id, workspace_id, reply } => {
                    let result = self.store.load_injection(&session_id, workspace_id.as_deref());
                    let _ = reply.send(result);
                }
                MemoryCommand::Recall { anchors, limit, reply } => {
                    let result = self.store.coarse_recall(&anchors, limit);
                    let _ = reply.send(result);
                }
                MemoryCommand::LoadDepth2 { node_ids, reply } => {
                    let result = self.store.expand_nodes(&node_ids);
                    let _ = reply.send(result);
                }
                MemoryCommand::WriteEpisode { episode } => {
                    if let Err(e) = self.store.save_episode(&episode) {
                        tracing::warn!("Episode 写入失败: {e}");
                    }
                }
                MemoryCommand::SaveDecision { decision } => {
                    if let Err(e) = self.store.save_decision(&decision) {
                        tracing::warn!("Decision 写入失败: {e}");
                    }
                }
                MemoryCommand::UpsertEntity { entity } => {
                    if let Err(e) = self.store.upsert_entity(&entity) {
                        tracing::warn!("Entity 更新失败: {e}");
                    }
                }
                MemoryCommand::UpdateInjection { level, target_id, content } => {
                    if let Err(e) = self.store.update_injection(level, &target_id, &content) {
                        tracing::warn!("Injection 更新失败: {e}");
                    }
                }
                MemoryCommand::RunMicroRumination { turn_result } => {
                    self.handle_micro_rumination(&turn_result).await;
                }
                MemoryCommand::RunMesoRumination { session_id, workspace_id } => {
                    self.handle_meso_rumination(&session_id, &workspace_id).await;
                }
                MemoryCommand::RunMetaRumination => {
                    self.handle_meta_rumination().await;
                }
                MemoryCommand::Shutdown => {
                    tracing::info!("Memory Actor 收到关闭指令");
                    break;
                }
            }
        }
        // Actor 退出前刷盘
        if let Err(e) = self.store.flush() {
            tracing::error!("Memory 刷盘失败: {e}");
        }
    }
}
```

---

## 跨进程场景：Leader 选举 + Actor 迁移

天工支持 GUI、TUI、Server 多模式运行，可能同时存在多个进程。采用 **Leader 选举**机制：全局只有一个进程持有 Memory Actor（Leader），其他进程通过 IPC 访问。当 Leader 退出时，剩余进程选举新 Leader 并接管 Actor。

### 整体架构

```
┌─────────┐  ┌─────────┐  ┌──────────┐
│  GUI    │  │  TUI    │  │  Server  │
│ Process │  │ Process │  │ Process  │
│         │  │         │  │          │
│ Handle  │  │ Handle  │  │  Handle  │
└────┬────┘  └────┬────┘  └────┬─────┘
     │            │            │
     │   IPC      │   IPC      │  in-process channel
     │            │            │
     └────────────┘            │
              │                │
              ▼                ▼
     ┌──────────────────────────────┐
     │  Server 进程（当前 Leader）    │
     │                              │
     │  MemoryActor（独立 task）      │
     │  MemoryIpcServer（UDS 监听）   │
     │  LeaderRegistry（注册表）      │
     └──────────────────────────────┘
```

### Leader 注册表

使用 lock 文件 + 注册表 JSON 文件实现分布式协调：

```
~/.tiangong/
  memory/
    leader.lock          ← flock 排他锁（选举互斥）
    leader.json          ← Leader 信息
    registry.json        ← 所有活跃进程注册表
    memory.sock          ← UDS 监听地址
```

```rust
/// Leader 信息
#[derive(Debug, Serialize, Deserialize)]
pub struct LeaderInfo {
    pub pid: u32,
    pub process_type: ProcessType,   // Gui / Tui / Server / Cli
    pub started_at: String,
    pub sock_path: String,
}

/// 进程注册信息
#[derive(Debug, Serialize, Deserialize)]
pub struct ProcessEntry {
    pub pid: u32,
    pub process_type: ProcessType,
    pub registered_at: String,
    pub priority: u8,    // 选举优先级
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
#[serde(rename_all = "lowercase")]
pub enum ProcessType {
    Server = 0,   // 最高优先级（长期运行）
    Gui    = 1,
    Tui    = 2,
    Cli    = 3,   // 最低优先级（生命周期短，但功能完整）
}
```

### 选举优先级

```
Server > GUI > TUI > CLI
```

理由：
- Server 通常作为后台服务长期运行，最适合持有 Actor
- GUI/TUI 是交互式应用，生命周期次之
- CLI 生命周期最短，但如果它是唯一活跃进程，也会成为 Leader 确保 Memory 可用

同优先级时，先注册的进程优先。所有模式都参与选举，保证 Memory 功能在任何运行模式下都不缺失。

---

### 启动流程

```rust
impl MemoryHandle {
    pub async fn connect_or_spawn(
        workspace_id: Option<String>,
        process_type: ProcessType,
    ) -> Result<Self> {
        let registry = ProcessRegistry::load()?;

        // 1. 注册自身
        registry.register(process_type)?;

        // 2. 尝试连接已有 Leader
        if let Some(leader) = registry.current_leader() {
            if leader.is_alive() {
                match Self::connect_ipc(&leader.sock_path).await {
                    Ok(handle) => return Ok(handle),
                    Err(e) => {
                        tracing::warn!("连接 Leader 失败: {e}");
                        // Leader 可能刚崩溃，尝试选举
                    }
                }
            }
        }

        // 3. 参与选举
        Self::try_become_leader(workspace_id, process_type).await
    }
}
```

### 选举算法

```rust
impl MemoryHandle {
    async fn try_become_leader(
        workspace_id: Option<String>,
        process_type: ProcessType,
    ) -> Result<Self> {
        let lock_path = memory_base_dir().join("leader.lock");

        // 使用 flock 排他锁保证选举互斥
        let lock_file = OpenOptions::new()
            .create(true)
            .write(true)
            .open(&lock_path)?;

        // 非阻塞尝试获取锁
        match lock_file.try_lock_exclusive() {
            Ok(()) => {
                // 获得选举锁，检查是否有更高优先级的进程
                let registry = ProcessRegistry::load()?;
                let candidates = registry.alive_processes();

                let highest = candidates.iter()
                    .min_by_key(|p| (p.priority, p.registered_at.clone()));

                let i_am_highest = highest
                    .map(|h| h.pid == std::process::id())
                    .unwrap_or(true);

                if i_am_highest {
                    // 我是最高优先级 → 成为 Leader
                    let handle = Self::become_leader(workspace_id).await?;
                    let _ = lock_file.unlock();
                    return Ok(handle);
                }

                // 不是最高优先级 → 释放锁，等待更高优先级进程成为 Leader
                let _ = lock_file.unlock();
                tokio::time::sleep(Duration::from_millis(200)).await;

                // 重试连接
                Self::connect_or_spawn(workspace_id, process_type).await
            }
            Err(_) => {
                // 其他进程正在选举，等待后重试
                tokio::time::sleep(Duration::from_millis(100)).await;
                Self::connect_or_spawn(workspace_id, process_type).await
            }
        }
    }

    async fn become_leader(workspace_id: Option<String>) -> Result<Self> {
        // 1. 启动 Actor
        let handle = MemoryActor::spawn(workspace_id)?;

        // 2. 启动 IPC Server
        let sock_path = memory_sock_path();
        MemoryIpcServer::start(&sock_path, handle.clone()).await?;

        // 3. 写入 Leader 信息
        let leader_info = LeaderInfo {
            pid: std::process::id(),
            process_type: current_process_type(),
            started_at: chrono::Local::now().naive_local().to_string(),
            sock_path: sock_path.to_string_lossy().to_string(),
        };
        write_leader_info(&leader_info)?;

        // 4. 启动心跳 + 监听选举
        tokio::spawn(leader_heartbeat(handle.clone()));

        Ok(handle)
    }
}
```

---

### Leader 健康检测与迁移

#### 心跳机制

Leader 进程定期更新心跳时间戳：

```rust
async fn leader_heartbeat(handle: MemoryHandle) {
    let mut interval = tokio::time::interval(Duration::from_secs(2));
    loop {
        interval.tick().await;
        if let Err(e) = update_heartbeat() {
            tracing::warn!("心跳更新失败: {e}");
        }
    }
}

/// 心跳文件
/// ~/.tiangong/memory/leader.json 中的 heartbeat_at 字段
fn update_heartbeat() -> Result<()> {
    let mut info = read_leader_info()?;
    info.heartbeat_at = chrono::Local::now().naive_local().to_string();
    write_leader_info(&info)
}
```

#### Follower 监控

非 Leader 进程（Follower）监控 Leader 健康状态：

```rust
async fn watch_leader(
    workspace_id: Option<String>,
    process_type: ProcessType,
    reconnect_tx: mpsc::Sender<MemoryHandle>,
) {
    let mut interval = tokio::time::interval(Duration::from_secs(5));
    loop {
        interval.tick().await;

        let leader_alive = check_leader_alive();

        if !leader_alive {
            tracing::info!("检测到 Leader 离线，尝试选举新 Leader");

            // 参与选举
            match MemoryHandle::try_become_leader(workspace_id.clone(), process_type).await {
                Ok(new_handle) => {
                    let _ = reconnect_tx.send(new_handle).await;
                    return; // 选举完成，退出监控
                }
                Err(e) => {
                    tracing::warn!("选举失败，继续监控: {e}");
                }
            }
        }
    }
}

fn check_leader_alive() -> bool {
    let info = match read_leader_info() {
        Ok(info) => info,
        Err(_) => return false,
    };

    // 检查心跳是否超时（10 秒无心跳视为死亡）
    let heartbeat = match NaiveDateTime::parse_from_str(&info.heartbeat_at, "%Y-%m-%d %H:%M:%S%.f") {
        Ok(t) => t,
        Err(_) => return false,
    };
    let elapsed = chrono::Local::now().naive_local() - heartbeat;
    if elapsed > chrono::Duration::seconds(10) {
        return false;
    }

    // 检查进程是否还活着
    is_process_alive(info.pid)
}

fn is_process_alive(pid: u32) -> bool {
    // Unix: kill(pid, 0) 检测进程存在性
    unsafe { libc::kill(pid as i32, 0) == 0 }
}
```

---

### Actor 状态迁移

当新 Leader 被选举出来时，需要重建 Actor 状态：

```rust
impl MemoryActor {
    /// 从磁盘重建状态（新 Leader 启动时）
    pub fn spawn_from_disk(workspace_id: Option<String>) -> Result<MemoryHandle> {
        // MemoryStore::load() 从磁盘重新加载索引和数据
        // 由于旧 Leader 退出前会 flush，数据是完整的
        Self::spawn(workspace_id)
    }
}
```

**关键保证**：
- 旧 Leader **正常退出**：收到 `Shutdown` 命令 → `flush()` 刷盘 → 新 Leader 从磁盘加载完整数据
- 旧 Leader **异常崩溃**：队列中未处理的命令会丢失，但磁盘上的数据仍然一致（因为每次写入都是原子操作 write-tmp + rename）
- 迁移期间的**命令丢失**：Follower 检测到 Leader 离线到新 Leader 就绪之间（约 5-10 秒），fire-and-forget 的写入命令会丢失。这是可接受的，因为写入本身就是"尽力而为"

---

### Handle 自动重连

`MemoryHandle` 内部支持自动切换后端（从 IPC 客户端切换到本地 Actor 或新的 IPC 连接）：

```rust
#[derive(Clone)]
pub struct MemoryHandle {
    inner: Arc<RwLock<HandleInner>>,
}

enum HandleInner {
    /// 本地 Actor（当前进程是 Leader）
    Local {
        tx: mpsc::Sender<MemoryCommand>,
    },
    /// IPC 客户端（当前进程是 Follower）
    Ipc {
        client: IpcClient,
    },
}

impl MemoryHandle {
    /// 当检测到后端失效时，自动重连
    async fn reconnect(&self, workspace_id: Option<String>, process_type: ProcessType) {
        match Self::connect_or_spawn(workspace_id, process_type).await {
            Ok(new_handle) => {
                let mut inner = self.inner.write().await;
                *inner = new_handle.take_inner();
                tracing::info!("Memory Handle 已重连");
            }
            Err(e) => {
                tracing::error!("Memory Handle 重连失败: {e}");
            }
        }
    }
}
```

---

### 完整生命周期时序

```
═══ 场景：Server 先启动，GUI 后启动，Server 先退出 ═══

T0: Server 启动
    ├── register(Server, pid=100)
    ├── 无 Leader → 选举 → 成为 Leader
    ├── MemoryActor::spawn()
    ├── MemoryIpcServer::start(memory.sock)
    └── leader.json = { pid: 100, type: server }

T1: GUI 启动
    ├── register(Gui, pid=200)
    ├── 发现 Leader(pid=100) 存活
    ├── connect_ipc(memory.sock) → 成功
    └── 启动 watch_leader() 监控

T2: Server 正常退出
    ├── handle.shutdown()
    │   └── Actor flush() 刷盘
    ├── 删除 memory.sock
    └── 更新 leader.json = null

T3: GUI 的 watch_leader 检测到 Leader 离线
    ├── try_become_leader(Gui)
    ├── 获取 leader.lock
    ├── 检查候选者 → 仅 GUI(pid=200)
    ├── become_leader()
    │   ├── MemoryActor::spawn_from_disk()  ← 从磁盘重建
    │   ├── MemoryIpcServer::start(memory.sock)
    │   └── leader.json = { pid: 200, type: gui }
    └── Handle 从 IPC 切换到 Local

T4: TUI 启动
    ├── register(Tui, pid=300)
    ├── 发现 Leader(pid=200) 存活
    └── connect_ipc(memory.sock) → 成功

T5: GUI 异常崩溃
    ├── memory.sock 断开（OS 自动关闭 fd）
    └── leader.json 心跳停止更新

T6: TUI 的 watch_leader 检测到心跳超时
    ├── try_become_leader(Tui)
    ├── 清理残留 memory.sock
    ├── become_leader()
    │   ├── MemoryActor::spawn_from_disk()
    │   │   └── 磁盘数据完整（原子写入保证）
    │   │   └── 崩溃时队列中未处理的命令丢失（可接受）
    │   └── MemoryIpcServer::start(memory.sock)
    └── TUI 成为新 Leader
```

---

### CLI 场景

CLI 执行单次命令后退出，选举优先级最低，但 **Memory 功能不可缺失**。CLI 启动时遵循与其他模式相同的 `connect_or_spawn` 流程：

1. 如果有 Leader 在线 → 连接 IPC，正常使用 Memory
2. 如果无 Leader → CLI 自己成为 Leader，启动进程内 Actor
3. CLI 退出时 → flush 刷盘 → 其他进程检测到 Leader 离线后选举新 Leader

```rust
// CLI 使用与其他模式完全相同的入口
let handle = MemoryHandle::connect_or_spawn(
    workspace_id,
    ProcessType::Cli,  // 优先级最低，但功能完整
).await?;

// CLI 执行完毕后正常关闭
handle.shutdown().await;
```

**设计原则**：所有运行模式（GUI、TUI、Server、CLI）都必须接入 Memory，不允许任何模式跳过或忽略 Memory 功能。CLI 虽然生命周期短，但其执行过程中产生的 Episode 同样有价值，必须被记录。

---

## 命令队列背压

```rust
// 队列容量 256，超出时的处理策略
const CHANNEL_CAPACITY: usize = 256;

impl MemoryHandle {
    /// 写入类命令使用 try_send，队列满时丢弃
    pub fn write_episode(&self, episode: Episode) {
        match self.tx.try_send(MemoryCommand::WriteEpisode { episode }) {
            Ok(()) => {}
            Err(mpsc::error::TrySendError::Full(_)) => {
                tracing::warn!("Memory 命令队列已满，丢弃 Episode 写入");
            }
            Err(mpsc::error::TrySendError::Closed(_)) => {
                tracing::warn!("Memory Actor 已关闭");
            }
        }
    }

    /// 查询类命令使用 send（async），允许等待队列腾出空间
    pub async fn load_injection(&self, ...) -> Vec<String> {
        // send().await 会在队列有空间时继续
        // 但设置超时防止无限等待
        let result = tokio::time::timeout(
            Duration::from_millis(100),
            self.tx.send(cmd),
        ).await;
        // ...
    }
}
```

---

## 生命周期管理

```
进程启动
  ├── ProcessRegistry.register(process_type)
  ├── 检测 Leader 是否存活
  │     ├── Leader 存活 → connect_ipc() → Follower 模式
  │     │     └── 启动 watch_leader() 后台监控
  │     └── Leader 不存在/已死 → try_become_leader()
  │           ├── 获取 leader.lock（选举互斥）
  │           ├── 选出最高优先级进程
  │           ├── MemoryActor::spawn()
  │           ├── MemoryIpcServer::start()
  │           └── 写入 leader.json + 启动心跳
  ├── ... 正常业务运行 ...
  │     ├── handle.load_injection().await   ← 查询
  │     ├── handle.write_episode(...)       ← 写入
  │     └── handle.run_micro_rumination(..) ← 反刍
  └── 进程退出
        ├── Leader 退出：
        │     ├── handle.shutdown() → Actor flush()
        │     ├── 删除 memory.sock
        │     ├── 清除 leader.json
        │     └── Follower 自动检测并选举新 Leader
        └── Follower 退出：
              └── registry.unregister() → 从注册表移除
```

---

## 与文件锁方案的对比

| 维度 | 文件锁方案 | Actor + Channel 方案 |
|------|-----------|---------------------|
| 进程内并发 | 需要 RwLock | 天然串行，无冲突 |
| 跨进程并发 | 文件锁（不可靠） | IPC 统一入口 |
| 写入冲突 | 锁竞争 + 重试 | 队列串行，不可能冲突 |
| 读写一致性 | 依赖锁正确使用 | Actor 内部状态始终一致 |
| 性能 | 锁获取 + 释放开销 | Channel send 开销（< 1μs） |
| 复杂度 | 低（但容易出 bug） | 中（但正确性有保证） |
| 外部依赖 | `fs2` | 无（仅 tokio） |
| 降级策略 | 跳过写入 | 队列满时丢弃 |

---

## 原则

- Memory 写入失败**不影响主流程**，写入是"尽力而为"
- 查询超时（100ms）后返回空默认值，不阻塞 Prompt 装配
- Actor 退出前必须刷盘，保证已接收的写入不丢失
- Leader 迁移期间（5-10 秒）丢失的 fire-and-forget 命令可接受
- 选举使用 flock 互斥 + 优先级排序，保证全局唯一 Leader
- **所有模式都必须接入 Memory**，CLI 也参与选举（最低优先级），不允许跳过
