一. 文件锁避免重复实例
let _pid_lock = match ironclaw::bootstrap::PidLock::acquire() {//尝试获取
Ok(lock) => Some(lock),
Err(ironclaw::bootstrap::PidLockError::AlreadyRunning { pid }) => {//已有运行，则取消本次运行
anyhow::bail!(
"Another IronClaw instance is already running (PID {}). \
If this is incorrect, remove the stale PID file: {}",
pid,
ironclaw::bootstrap::pid_lock_path().display()
);
}
Err(e) => {//无权限等错误到这
eprintln!("Warning: Could not acquire PID lock: {}", e);
eprintln!("Continuing without PID lock protection.");
None
}
};二. Agent startup
1. Enhanced first-run detection
避免没有必须的配置的情况，去运行agent
if !cli.no_onboard
&& let Some(reason) = ironclaw::setup::check_onboard_needed()
{
wizard.run().await?;//同直接ironclaw onboard，走一遍引导2. 最终Agent参数获取
let toml_path = cli.config.as_deref();//同Onboard Setting获取的文件
let runtime_overrides = ironclaw::config::RuntimeConfigOverrides {//这里是运行时安全配置重写
deployment: cli.deployment_mode,//包括部署模式
profile: cli.runtime_profile,//运行画像 ，这些都会影响后面agent的行为决策
yolo_disclosure_acknowledged: if cli.yolo_disclosure {
Some(true)
} else {
None
},
};
let config = match Config::from_env_with_toml(toml_path)//将用从toml读取构造的setting给config配置
.await
.and_then(|c| c.with_runtime_overrides(&runtime_overrides))
Settings.llm_backend → LlmConfig.backend:
1. Settings::default()
2. Profile preset
3. TOML 文件
4. 从 OS 凭据存储注入（Keychain / Linux credentials）
5. 从加密 DB 注入的 LLM 密钥
6. 用户显式设置的环境变量（.env / IRONCLAW_* / shell）— 最高3. 日志广播初始化
生产者（产生日志方）：业务代码直接调 tracing::info!() / error!() 等宏，不需要感知 web、订阅者、SSE 这些东西。tracing-subscriber 框架会把日志路由到 Layer，Layer 再推给 broadcaster。
消费者（需要日志方）：SSE/WebSocket handler 调用 subscribe() 建立订阅，broadcaster 维护一个活跃连接列表，每次收到新日志就遍历推送。
简单说就是：
业务代码 (tracing!宏)
↓
tracing Layer (log_layer.rs 的 on_event)
↓
Broadcaster::send() ← 所有在线订阅者收到
↓
SSE handler / WebSocket handler (subscribe → receiver)
↓
客户端浏览器
Layer 本身充当了桥接层——一边接 tracing 框架的被动回调，一边接主动的订阅/广播机制，业务代码完全不用改。
let log_broadcaster = Arc::new(LogBroadcaster::new());三. AppBuilder(构建核心组件)
buildAll()
init_database//连接数据库，获取句柄；清理"残留 sandbox 任务"——上次进程崩溃可能留下未结束的 sandbox job 记录
**************************
r#"
UPDATE agent_jobs SET
status = 'interrupted',//设置状态为中断
failure_reason = 'Process restarted',
completed_at = NOW()
WHERE source = 'sandbox' AND status IN ('running', 'creating')//任务来源是沙箱、状态为运行或创建中init_secrets(&mut self)
拿master key：环境变量、Keychain (macOS) / Credentials file (Linux)
拿到后let crypto = Arc::new(SecretsCrypto::new(master_key.clone())?);构造加密引擎
构造真正的持久化 secrets storelet store = create_secrets_store(crypto, handles);
如果本进程自动生成了新 key，但 DB 里已有历史密钥行，新 key 解不开旧行——继续往下走会静默遮盖无法恢复的数据。一旦发现，回滚 key 持久化（让下次启动重新触发检测），并 fail-
closed。
crate::config::inject_llm_keys_from_secrets(secrets.as_ref(), &self.config.owner_id)注入配置将加密llm key
除了固定读的，还有动态发现的(provider.json文件里，有配置什么模型需要什么变量)
DB 里是加密的，内存里是明文的——明文只存在于进程运行时的 INJECTED_VARS，进程退出就清掉。
Setting的llm_custom_providers和llm_builtin_overrides
self.config
.re_resolve_llm_with_secrets(
settings_store,
&owner_id,
toml_path,
Some(secrets.as_ref()),
true,
)
.await
若缺失补上config.llmconfig的apikey
let registry = ironclaw_llm::ProviderRegistry::load();
let has_dedicated_config = registry
.find(self.config.llm.backend.as_str())
.is_some_and(|d| d.protocol.has_dedicated_config());
if !has_dedicated_config && self.config.llm.provider.is_none() {检测当前选的provicer有无keypub async fn init_llm( 构造主模型，廉价模型，模型调用记录, 详见llm链一篇
let (
safety,
tools,
embeddings,
workspace,
builder,
credential_registry,
http_interceptor,
workspace_resolver,
) = self.init_tools(&llm, cheap_llm.as_ref()).await?;初始化tools
let embeddings = ironclaw_embeddings::create_provider(还会创建向量化模型
&self.config.embeddings,
ironclaw_embeddings::ProviderDeps {
session: self.session.clone(),
bedrock_setup,
},
)
.await;//注册消息总结hook 每天一结，追加写入, 放到消息管理器上
// Create hook registry early so runtime extension activation can register hooks.
let hooks = Arc::new(HookRegistry::new());
// Register session summary hook (writes conversation summary on session end).
if let (Some(db), Some(ws_resolver)) = (&self.db, &workspace_resolver) {
let summary_llm = cheap_llm
.as_ref()
.map(Arc::clone)
.unwrap_or_else(|| Arc::clone(&llm));
hooks
.register(Arc::new(crate::hooks::SessionSummaryHook::new(
Arc::clone(db) as Arc<dyn crate::db::ConversationStore>,
Arc::clone(ws_resolver),
summary_llm,
)))
.await;
}
/// Writes a conversation summary to workspace when a session ends.
///
/// Uses the LLM to generate a brief summary of the most recent
/// conversation, then appends it to `daily/{date}-session-summary.md`.
let agent_session_manager =
Arc::new(AgentSessionManager::new().with_hooks(Arc::clone(&hooks)));
pub struct ConversationMessage {
pub id: Uuid,
pub role: String,
pub content: String,
pub created_at: DateTime<Utc>,
}//用户偏好覆盖默认偏好：tool权限相关
// Build the workspace-backed `SettingsStore` BEFORE init_extensions so
// tools registered there (`register_permission_tools`,
// `upgrade_tool_list`) can be wired with the adapter from the start.
// The same adapter instance is then exposed on `AppComponents.settings_store`
// and reused by main.rs (e.g. for the SIGHUP reload handler).
let (settings_store, settings_cache): (
Option<Arc<dyn crate::db::SettingsStore + Send + Sync>>,
Option<Arc<crate::db::cached_settings::CachedSettingsStore>>,
) = match (&workspace, &self.db) {let ownership_cache = Arc::new(crate::ownership::OwnershipCache::new());
/// In-process cache: `(channel, external_id)` → [`UserId`].
///
/// All methods take `&self` — interior mutability via `RwLock`.
pub struct OwnershipCache {
identities: RwLock<HashMap<(String, String), UserId>>,
}
● OwnershipCache 的作用
这是个写穿透缓存,把 (channel, external_id) → Identity 的映射常驻内存,避免每次请求都打 DB。
1. 解决的痛点
IronClaw 是多渠道接入(Telegram / Slack / Web / Webhook / ...),每个渠道给用户发消息时带的是渠道侧的 ID:
- Telegram 给的是 chat_id = 123456789(整数,跟 IronClaw 没关系)
- Slack 给的是 slack_user_id = U07ABCD...
- Web 是 session cookie
这些 ID 每条消息都来一次,而系统需要回答的核心问题是:
▎ "这个 Telegram chat_id 到底对应 IronClaw 里的哪个 user?是不是已经配对过了?是不是管理员?"
如果每次消息都 SELECT FROM channel_identities WHERE channel=$1 AND external_id=$2,高 QPS 下 DB 压力会爆。
2. 缓存了什么
从名字拆解:
┌────────────────────────┬───────────────────────────────────────────┐
│ 字段 │ 含义 │
├────────────────────────┼───────────────────────────────────────────┤
│ (channel, external_id) │ 渠道类型 + 该渠道里的用户 ID(联合主键) │
├────────────────────────┼───────────────────────────────────────────┤
│ → Identity │ IronClaw 内部的 user_id、role、配对状态等 │
└────────────────────────┴───────────────────────────────────────────┘
配对流程(Task 8 那条注释也提到了)在第一次配对成功后把结果写进缓存;之后所有消息都先查缓存,miss 才打 DB。
3. 为什么在 build_all 里预先建好
代码注释里那句 // Pre-allocated here so all subsystems can hold an Arc to the same cache instance 是关键:
- 配对流程要写
- 消息路由要读
- 权限校验要读
- ExtensionManager 也要读(在 init_extensions 里 PairingStore::new(db, ownership_cache.clone()) 就用上了)
这些模块持有的是 Arc<OwnershipCache>,不是引用——所以必须先把 Arc 造好,才能在多个子系统间共享同一个实例。否则各持一份,缓存就废了。
4. 跟 PairingStore 的关系
init_extensions 里:
let ps = Arc::new(crate::pairing::PairingStore::new(
Arc::clone(db),
Arc::clone(&ownership_cache), // <-- 共享同一个 cache
));
em = em.with_pairing_store(Arc::clone(&ps));
PairingStore 负责 DB 持久化(配对状态写表),OwnershipCache 负责加速查询(在内存里直接拿到 Identity)。两者是经典的 "DB 是 source of truth,cache 是 hot path" 模式。
5. 失效语义
写穿透缓存通常有这些问题,代码里没全看到(可以从 src/ownership/ 看),但常见处理:
- 配对完成时:写 DB + 写缓存
));
em = em.with_pairing_store(Arc::clone(&ps));
PairingStore 负责 DB 持久化(配对状态写表),OwnershipCache 负责加速查询(在内存里直接拿到 Identity)。两者是经典的 "DB 是 source of truth,cache 是 hot path" 模式。
5. 失效语义
写穿透缓存通常有这些问题,代码里没全看到(可以从 src/ownership/ 看),但常见处理:
- 配对完成时:写 DB + 写缓存
- 取消配对时:写 DB + cache.invalidate((channel, external_id))
- 跨实例不命中:重启后 cache 是空的,首次访问会回源 DB,然后填回缓存
一句话
▎ OwnershipCache 是 (channel, external_id) → Identity 的进程内写穿透缓存,在 build_all 里预分配一个 Arc,让配对/路由/权限/ExtensionManager 等所有子系统共享同一个实例,避免每条消息都查 channel_identities
▎ 表。 /// Phase 5: Load WASM tools, MCP servers, and create extension manager.
pub async fn init_extensions(
&self,
tools: &Arc<ToolRegistry>,
hooks: &Arc<HookRegistry>,
settings_store_override: Option<Arc<dyn crate::db::SettingsStore + Send + Sync>>,
ownership_cache: Arc<crate::ownership::OwnershipCache>,
) -> Result<
(
Arc<McpSessionManager>,
Arc<McpProcessManager>,
Option<Arc<WasmToolRuntime>>,
Option<Arc<ExtensionManager>>,
Vec<crate::extensions::RegistryEntry>,
Vec<String>,
),
anyhow::Error,
> {
还注册了对extMannager的管理工具，以及配对等工具
▎ init_extensions = ①建 MCP session/process manager + ②初始化 WASM runtime + ③tokio::join 并发扫 WASM tools 目录 / 加载 owner 的 MCP server clients（建 client + 拉 tools/list） + ④用前面建好的
▎ manager/runtime/catalog 构造 ExtensionManager（含空的 McpClientStore）+ ⑤把启动期 clients 注入 store 并注册 wrappers（canonical 命名匹配）+ ⑥在 ToolRegistry 上挂权限 / discovery
▎ 工具，返回元组给上层组装。
**********
对于cataLog
不是重复加载，是两个独立阶段各自按需取一份。wizard.run() 是 onboarding 入口、只用一次，wizard 退出时它的 catalog 实例就丢；init_extensions 是主循环启动阶段，要构造 ExtensionManager、需要
▎ Vec<RegistryEntry> 类型（wizard 里的 catalog 类型对不上）。RegistryCatalog::load_or_embedded() 是无状态纯函数，没有全局缓存，加载本身几毫秒 —— 没人愿意为这点开销引入全局 LazyLock 单例。
******主要是能力清单去展示！！！
▎ Wizard 选完后三类安装：①bundled（编译期内嵌的 .wasm，fs::copy 落 ~/.ironclaw/{tools,channels}/，不走网络）②有 artifact URL + SHA256（reqwest::get 下载 → SHA256 校验 → tar.gz 解压或裸 .wasm
▎ 写盘，校验失败或下载失败时退回本地 cargo build）③没 artifact（直接本地编译）。文件名统一用 manifest.name，MCP server 走配置文件而不是二进制，不进 installer。 生成默认模板
▎ 拿现成模板 → 填默认内容 → 给旧内容生成"指纹"向量,全部自动、不能错也不阻塞启动。
---
详细版
想象你新装了一个笔记 App,首次打开时它做了这些事:
① 可选:从外部目录"塞"现成笔记进来
if let Ok(import_dir) = std::env::var("WORKSPACE_IMPORT_DIR") {
ws.import_from_directory(import_path).await
}
- 只在设置了环境变量 WORKSPACE_IMPORT_DIR 时触发
- 场景:Docker 镜像或部署脚本想预装团队定制的笔记模板(比如写好的"项目说明.md"、"工具使用规范.md")
- 重要规则:只导入"还没有"的笔记,绝不覆盖你已写的内容
- 放在最前面,优先于下面的默认种子
② 默认种子:空 workspace 写入自带模板
match ws.seed_if_empty().await { ... }
- 如果你的笔记空间是空的,自动塞入一些通用模板(身份文件 AGENTS.md、SOUL.md 之类)
- 名字已经说明——if_empty,空才填
- 失败只警告,不阻断启动
③ 后台任务:给已写内容生成"AI 检索指纹"
if embeddings.is_some() {
tokio::spawn(async move {
ws_bg.backfill_embeddings().await
});
}
- 只有配了 embedding 服务才跑
- 做了什么:把以前写过、但还没生成"向量指纹"的笔记补上向量(就是给文字算一串数字,后面做语义搜索用)
- tokio::spawn 丢后台,不阻塞启动——可以慢慢算
---
为啥要这么设计
┌─────────────────────────────────┬────────────────────────────────────────────┐
│ 设计 │ 原因 │
├─────────────────────────────────┼────────────────────────────────────────────┤
│ if let Some(ref ws) = workspace │ 没数据库时整个跳过——没存东西的地方就不折腾 │
├─────────────────────────────────┼────────────────────────────────────────────┤
│ import 在 seed 之前 │ 部署方定制的优先,系统默认垫底 │
├─────────────────────────────────┼────────────────────────────────────────────┤
│ seed_if_empty │ 不动用户已有内容 │
├─────────────────────────────────┼────────────────────────────────────────────┤
│ tokio::spawn 后台跑回填 │ embedding 计算可能慢,启动不能等 │
├─────────────────────────────────┼────────────────────────────────────────────┤
│ 失败只 warn! │ 启动体验优先,数据任务下次再补 │
└─────────────────────────────────┴────────────────────────────────────────────┘
一句话总结:这是 workspace(持久记忆)的"首次填充 + 数据补全"流程,启动时跑一次,把 workspace 从空状态变成可用状态,并保证后续语义搜索能命中所有历史内容。加载skill系统
// Skills system
let (skill_registry, skill_catalog) = if self.config.skills.enabled {
let mut registry = SkillRegistry::new(self.config.skills.local_dir.clone())
.with_installed_dir(self.config.skills.installed_dir.clone())
.with_bundled_content(crate::skills::bundled::load_bundled_skills())
.with_max_scan_depth(self.config.skills.max_scan_depth);
let loaded = registry.discover_all().await;//发现
if !loaded.is_empty() {
tracing::debug!("Loaded {} skill(s): {}", loaded.len(), loaded.join(", "));
}
// Register credential mappings from skill frontmatter into the
// shared registry so the HTTP tool can auto-inject credentials.
crate::skills::register_skill_credentials(registry.skills(), &credential_registry);//注册凭据到路由器
if let Some(db) = self.db.as_ref() {
crate::skills::persist_skill_auth_descriptors(
registry.skills(),
Some(db.as_ref()),
&self.config.owner_id,
)
.await;
}
let registry = Arc::new(std::sync::RwLock::new(registry));
let catalog = ironclaw_skills::catalog::shared_catalog();//待发现的skill
tools.register_skill_tools(Arc::clone(®istry), Arc::clone(&catalog));//注册skill tool工具
(Some(registry), Some(catalog))
} else { let context_manager = Arc::new(ContextManager::new(self.config.agent.max_parallel_jobs));//创建上下文管理器和token花销统计
let cost_guard = Arc::new(crate::agent::cost_guard::CostGuard::new(
crate::agent::cost_guard::CostGuardConfig {
max_cost_per_day_cents: self.config.agent.max_cost_per_day_cents,
max_actions_per_hour: self.config.agent.max_actions_per_hour,
max_cost_per_user_per_day_cents: self.config.agent.max_cost_per_user_per_day_cents,
},
));四. tunnel setup
// ── Tunnel setup ───────────────────────────────────────────────────
let (config, active_tunnel) = if enable_non_cli {
ironclaw::tunnel::start_managed_tunnel(config).await
} else {
(config, None)
};
隧道相关五. Orchestrator / container job manager
// ── Orchestrator / container job manager ────────────────────────────
// Orchestrator starts an internal HTTP API (default 0.0.0.0:50051) for
// sandbox worker communication. Skip it entirely under --cli-only to
// honour the "no network listeners" contract.
let (container_job_manager, job_event_tx, prompt_queue, docker_status) = if enable_non_cli {
let orch = ironclaw::orchestrator::setup_orchestrator(
&config,
&components.llm,
components.db.as_ref(),
components.secrets_store.as_ref(),
)
.await;
(
orch.container_job_manager,
orch.job_event_tx,
orch.prompt_queue,
orch.docker_status,
)
/// Detect Docker availability, create the container job manager, and start
/// the orchestrator internal API in the background.
pub async fn setup_orchestrator(
config: &crate::config::Config,
llm: &Arc<dyn LlmProvider>,
db: Option<&Arc<dyn Database>>,
secrets_store: Option<&Arc<dyn SecretsStore + Send + Sync>>,
) -> OrchestratorSetup {
let prompt_queue = Arc::new(Mutex::new(
HashMap::<Uuid, VecDeque<api::PendingPrompt>>::new(),
));
let docker_status = if config.sandbox.enabled {
let (job_event_tx, container_job_manager) = if config.sandbox.enabled && docker_status.is_ok() {
let (tx, _) = broadcast::channel(256);
let job_event_tx = Some(tx);
let token_store = TokenStore::new();
let orchestrator_port = resolve_orchestrator_port();
let job_config = ContainerJobConfig {
};
let jm = Arc::new(ContainerJobManager::new(job_config, token_store.clone()));
let orchestrator_state = api::OrchestratorState {
llm: Arc::clone(llm),
job_manager: Arc::clone(&jm),
token_store,
job_event_tx: job_event_tx.clone(),
prompt_queue: Arc::clone(&prompt_queue),
store: db.cloned(),
secrets_store: secrets_store.cloned(),
job_owner_cache: Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())),
};
tokio::spawn(async move {//内部监听容器的http请求
if let Err(e) = OrchestratorApi::start(orchestrator_state, orchestrator_port).await {
tracing::error!("Orchestrator API failed: {}", e);
}
});
if config.claude_code.enabled {
tracing::info!(
"Claude Code sandbox mode available (model: {}, max_turns: {})",
config.claude_code.model,
config.claude_code.max_turns
);
}
if config.acp.enabled {
tracing::info!("ACP agent sandbox mode available");
}
(job_event_tx, Some(jm))
} else {
(None, None)
};
OrchestratorSetup {
container_job_manager,
job_event_tx,
prompt_queue,
docker_status,
}
}
▎ OrchestratorApi 是"装着敏感资源（LLM key、events、prompts）的 host 端 HTTP 服务器"——sandbox 容器里跑的 worker 必须通过它才能借到这些能力，且每个 job 用独立 bearer token
▎ 限定作用域，既能让容器干活，又不让它碰到真正的密钥和跨租户状态。
******************************
# `job_event_tx` 链路：worker → host → SSE
```
[WORKER] sandbox 容器内的 worker 进程
│
│ HTTP POST /worker/{job_id}/event
│ Authorization: Bearer <job-token> ← 启动时由 host 注入
│ body: JobEventPayload
▼
[HOST] src/orchestrator/api.rs::job_event_handler
│
│ worker_auth_middleware 校验 token
│ 构造 AppEvent (JobMessage / JobToolUse / JobStatus / JobResult …)
▼
[HOST] state.job_event_tx.send((job_id, user_id, event))
│ broadcast::channel(256) ← 在 src/orchestrator/mod.rs:109 创建
▼
[HOST] SseManager (订阅者) ← src/main.rs:872 注入
│
│ broadcast_for_user(user_id, event) ← 按 user_id 路由
▼
[HOST] state.sse (broadcast hub)
│
│ SSE 帧 type: job_message / job_tool_use / …
▼
[USER] 浏览器
```
---
# `prompt_queue` 链路：host 内部 push → worker 主动 pull
```
[USER] 浏览器
│
│ POST /api/jobs/{id}/prompt { content, done }
│ Authorization: Bearer <GATEWAY_AUTH_TOKEN>
▼
[HOST] src/channels/web/features/jobs/mod.rs::jobs_prompt_handler
│
│ 校验 ownership + 仅当 job 模式 ∈ {claude_code, acp*}
│ let mut queue = state.prompt_queue.lock().await
│ queue.entry(job_id).or_default().push_back(PendingPrompt{content, done})
▼
[HOST] GatewayState.prompt_queue
│ Option<PromptQueue> (Arc<Mutex<HashMap<Uuid, VecDeque<PendingPrompt>>>>)
│
│ ─────────── 同一 Arc ───────────
│
│ Arc::clone() 在 src/main.rs:1022 注入到 OrchestratorState
▼
[HOST] OrchestratorState.prompt_queue
│
│ ← 同一份 in-memory 队列（host 进程内共享）
│
│ ───────── worker 拉取 ─────────
│
│ HTTP GET /worker/{job_id}/prompt
│ Authorization: Bearer <job-token>
▼
[HOST] src/orchestrator/api.rs::get_prompt_handler
│
│ worker_auth_middleware 校验 token
│ state.prompt_queue.lock().await
│ queue.get_mut(&job_id).pop_front()
│
│ 200 OK { content, done } ← 队列非空
│ 204 No Content ← 队列空
▼
[WORKER] sandbox 容器内的 worker (claude_bridge / acp_bridge)
│
│ 拿到 prompt 后继续干活
```
**另一条 push 路径**（不走 web，直接调 tool）：
```
[HOST] src/tools/builtin/job.rs::queue_prompt_to_job
│ ToolDispatcher.dispatch("queue_prompt_to_job", …)
│
│ 校验 ownership
│ state.prompt_queue.lock().await
│ queue.entry(job_id).or_default().push_back(PendingPrompt{…})
▼
[HOST] (同上，写入同一份队列)
```
---
# 关键标注速查
| 节点 | 角色 | 文件 |
| --------------------------------- | -------------------------- | ------------------------------------------- |
| worker POST /event | **WORKER** | 容器内 worker 进程 |
| `job_event_handler` | **HOST** | `src/orchestrator/api.rs` |
| `job_event_tx.send` | **HOST**（broadcast 通道） | `src/orchestrator/api.rs:448` |
| SseManager 订阅 | **HOST** | `src/channels/web/platform/sse.rs` |
| 浏览器收 SSE | **USER** | — |
| 浏览器 POST /api/jobs/{id}/prompt | **USER → HOST** | — |
| `jobs_prompt_handler` 写入队列 | **HOST** | `src/channels/web/features/jobs/mod.rs:702` |
| `queue_prompt_to_job` 工具 | **HOST**（agent loop 内） | `src/tools/builtin/job.rs:1668` |
| `prompt_queue` 数据结构 | **HOST**（in-memory 共享） | `OrchestratorState` 字段 |
| worker GET /prompt | **WORKER** | 容器内 worker 进程 |
| `get_prompt_handler` | **HOST** | `src/orchestrator/api.rs:467` |六. Channel
// Default user ID for extension operations (single-user mode).
let ext_user_id = config.owner_id.clone();
// Startup-active WASM channels are resolved lazily inside the
// `enable_non_cli && wasm_channels_enabled` gate below. Defaulting to
// an empty set here keeps the later auto-activation block (gated on
// `wasm_channel_runtime_state`) compiling without computing — and
// potentially failing on — settings-store reads in `--cli-only` /
// `WASM_CHANNELS_ENABLED=false` runs.
let mut startup_active_wasm_channels: std::collections::HashSet<String> =
std::collections::HashSet::new();
let channels = ChannelManager::new();//管理器收集各种支持的channel
if tui_mode && cli.message.is_none()//tui channel
channels.add(Box::new(tui_channel)).await;
let shared_routine_engine_slot: ironclaw::channels::web::platform::state::RoutineEngineSlot =
Arc::new(tokio::sync::RwLock::new(None));
// Collect webhook route fragments; a single WebhookServer hosts them all.
let mut webhook_routes: Vec<axum::Router> = Vec::new();
webhook_routes.push(webhooks::routes(ToolWebhookState {
tools: Arc::clone(&components.tools),
routine_engine: Arc::clone(&shared_routine_engine_slot),
user_id: config.owner_id.clone(),
secrets_store: components.secrets_store.clone(),
}));
******************加载wasm channel
let wasm_result = ironclaw::channels::wasm::setup_wasm_channels(//从~/.ironclaw/channel加载，并注册路由
&config,
&components.secrets_store,
components.extension_manager.as_ref(),
components.db.as_ref(),
&channel_names,
&startup_active_wasm_channels,
Arc::clone(&components.ownership_cache),
)
.await;
// Start the unified webhook server if any routes were registered.//包括wasmChannel和tool的
let webhook_server: Option<Arc<tokio::sync::Mutex<WebhookServer>>> = if !webhook_routes
.is_empty()
{ if config.sandbox.enabled {//gatewaychannel，会在沙箱模式下用job_event_tx 订阅，得到任务执行情况，然后广播
gw = gw.with_prompt_queue(Arc::clone(&prompt_queue));
if let Some(ref tx) = job_event_tx {
let mut rx = tx.subscribe();//订阅
let gw_state = Arc::clone(gw.state());
tokio::spawn(async move {
while let Ok((_job_id, user_id, event)) = rx.recv().await {
// Reuse the gateway's central status-event router so
// the sandbox dispatch path inherits the same drop /
// WARN / broadcast policy as `Channel::send_status`.
// Empty `user_id` collapses into the None arm via
// `dispatch_status_event`'s `!is_empty()` filter.
let user_id_opt = (!user_id.is_empty()).then_some(user_id.as_str());
ironclaw::channels::web::dispatch_status_event(//广播
&gw_state.sse,
gw_state.multi_tenant_mode,
user_id_opt,
event,
);
}
});
}
}
## 沙箱模式（worker 跑在 Docker 容器内）
```
sandbox worker 进程 (Docker container)
↓ post_event(event_type, data) [worker/container.rs:319,347]
HTTP POST /event → orchestrator/api.rs
↓ 解析 event_type 字符串 → 构造 AppEvent [orchestrator/api.rs:343,358,1058,1112]
↓ state.job_event_tx.send((job_id, user_id, event)) [orchestrator/api.rs:448]
broadcast::Sender<(Uuid, String, AppEvent)>
↓ tx.subscribe() [main.rs:1024-1028]
main.rs tokio::spawn 桥接任务
↓ dispatch_status_event(...) [main.rs:1037]
↓ gw_state.sse.broadcast_for_user(user_id_opt, event) [web/CLAUDE.md "central status-event router"]
SSE 广播
↓ subscribe() 闭包过滤 uid / verbose [platform/sse.rs:217-289]
浏览器
```
## 非沙箱模式（worker 跑在主进程内）
```
主进程
↓
agentic_loop 跑 JobDelegate [worker/job.rs:252]
↓ 在 run_agentic_loop() 里 emit StatusUpdate [worker/job.rs:159,172]
↓ JobDelegate::handle_status [worker/job.rs:246]
↓ sse.broadcast(event) [worker/job.rs:246]
SSE 广播（直接调，跨进程都不需要）
↓ subscribe() 闭包过滤
浏览器
```
## 对比要点
| 维度 | 沙箱 | 非沙箱 |
| -------- | ------------------------------------- | ---------------------------------- |
| 起点 | `worker/container.rs::post_event` | `worker/job.rs` 内的 `JobDelegate` |
| 跨进程 | 是 (HTTP) | 否 (同进程) |
| 通道 | `broadcast::Sender` (`job_event_tx`) | 直接函数调用 |
| 桥接任务 | `main.rs:1024` tokio::spawn | **不存在** |
| 转换位置 | `orchestrator/api.rs` 解析 event_type | `JobDelegate` 直接构造 `AppEvent` |
| 中转函数 | `dispatch_status_event` | 无 |七. hook
let hook_bootstrap = bootstrap_hooks(
&components.hooks,
components.workspace.as_ref(),
&config.wasm.tools_dir,
&config.channels.wasm_channels_dir,
&active_tool_names,
&loaded_wasm_channel_names,
&components.dev_loaded_tool_names,
)
.await;
启动时把内置、WASM 工具/通道（已激活的）、workspace 用户自定义三类 hook 配置全部解析并注册到 HookRegistry 里；任何一份配置损坏都不阻塞启动，只增加 summary.errors 计数。/// Register bundled hooks, then load plugin and workspace hook bundles.
pub async fn bootstrap_hooks(
registry: &Arc<HookRegistry>,
workspace: Option<&Arc<Workspace>>,
wasm_tools_dir: &Path,
wasm_channels_dir: &Path,
active_tool_names: &[String],
active_channel_names: &[String],
dev_loaded_tool_names: &[String],
) -> HookBootstrapSummary {
let mut summary = HookBootstrapSummary::default();
let bundled = register_bundled_hooks(registry).await;
summary.bundled_hooks += bundled.hooks;
summary.outbound_webhooks += bundled.outbound_webhooks;
summary.errors += bundled.errors;
let plugin = register_plugin_bundles(
registry,
wasm_tools_dir,
wasm_channels_dir,
active_tool_names,
active_channel_names,
dev_loaded_tool_names,
)
.await;
summary.plugin_hooks += plugin.hooks;
summary.outbound_webhooks += plugin.outbound_webhooks;
summary.errors += plugin.errors;
if let Some(workspace) = workspace {
let workspace_loaded = register_workspace_bundles(registry, workspace).await;
summary.workspace_hooks += workspace_loaded.hooks;
summary.outbound_webhooks += workspace_loaded.outbound_webhooks;
summary.errors += workspace_loaded.errors;
}
summary
}
## 答：是，配置不同，执行逻辑完全不同
`register_plugin_bundle_from_capabilities_file` 把 bundle 拆成 `bundle.rules` 和 `bundle.outbound_webhooks` 两类配置（`register_bundle`，149-187 行），每类配置编译成**不同的 `Hook` trait 实现**注册到 `HookRegistry`。注册完之后，运行时调度器看到的是"另一个 hook"，**不会**知道它来自哪个配置文件。
## RuleHook 执行逻辑（`bundled.rs:386-425`）
纯函数式、**同步 in-process** 改写内容：
```
1. extract_primary_content(event) // 取出事件的主文本
2. if when_regex 存在且不匹配 → return HookOutcome::ok() // 守卫不通过，整条规则 no-op
3. if reject_reason 存在 → return HookOutcome::reject(...) // 命中守卫就拒绝
4. 对 modified 应用 replacements[] (按顺序) // 正则替换
5. 拼接 prepend / append
6. if modified != 原 content → HookOutcome::modify(modified)
else → HookOutcome::ok()
```
**核心特征**：
- **可能改变事件** —— 返回 `HookOutcome::Modify(new_text)` / `Reject(reason)`，会**阻塞并影响主流程**。这是 `transform` 类 hook 的本质：返回值会被 `HookRegistry` 用来改写后续处理。
- **不发起 I/O** —— 全在内存里跑 regex，廉价。
- **同步语义** —— 因为可能修改事件，下游必须等它返回才知道内容。
## OutboundWebhookHook 执行逻辑（`bundled.rs:534-591`）
完全不同的形态 —— **异步 fire-and-forget HTTP POST**：
```
1. 把 event 序列化成 OutboundWebhookPayload (摘要是结构化、去除敏感字段)
2. semaphore.try_acquire_owned() // 超过 max_in_flight 直接丢弃（不阻塞主流程）
3. tokio::spawn { // 关键：派发到独立任务，主流程立刻返回
client.post(url).json(&payload).send().await
}
4. return HookOutcome::ok() // 立即返回，不等待 HTTP 完成
```
**核心特征**：
- **永远不影响事件** —— 始终 `HookOutcome::ok()`，不可能 `Modify` / `Reject`。
- **不阻塞主流程** —— 投递是 `tokio::spawn` 出去的后台任务，hook execute 本身只是组装 payload + 拿 permit。
- **重试/超时由 `reqwest::Client::timeout` 控制**（`from_config:466-470` 里设置）。
- **可以丢弃** —— 超过并发上限的请求直接丢 + warn log（`semaphore.try_acquire_owned` 失败，547-556 行）。这保护主流程不被慢/挂的 webhook 拖死。
- **网络策略拦截** —— `dispatch_client_for_target`（567-577 行）允许 hook 在发起请求前接受运行时网络策略检查（`sandbox` 相关的 `NetworkPolicyDecider`），被拦截就在后台任务里 warn 后 return，不影响主流程。
## 关键差异
| 维度 | RuleHook | OutboundWebhookHook |
| ----------------- | ------------------------------------------- | -------------------------------------------------- |
| 配置名 | `HookRuleConfig` (191-220) | `OutboundWebhookConfig` (231-250) |
| 触发的 hook point | 用户可配 | 用户可配 |
| 是否改事件 | **是** (`Modify` / `Reject`) | **否**（永远 `ok`） |
| 阻塞主流程 | **是** —— 调用方必须等返回值 | **否** —— `tokio::spawn` 后立刻返回 |
| 失败处理 | 通过 `failure_mode` (FailOpen / FailClosed) | **丢弃**（不返回错误给主流程） |
| 资源上限 | `timeout_ms` | `timeout_ms` + `max_in_flight` (semaphore) |
| 网络 I/O | **无** | **有** (HTTP POST) |
| 默认 priority | `DEFAULT_RULE_PRIORITY` | `DEFAULT_WEBHOOK_PRIORITY`（一般更低，因为不阻塞） |
## 注册函数里那 187 行的形态
`register_bundle` 内部就是这两个 `for` 循环的并列结构：
```rust
for rule in bundle.rules { // → RuleHook
RuleHook::from_config(...).map(register_with_priority)
}
for webhook in bundle.outbound_webhooks { // → OutboundWebhookHook
OutboundWebhookHook::from_config(...).map(register_with_priority)
}
```
**两类 hook 走的是完全独立的编译路径**，共享的只是"配置名带 source 前缀" (`format!("{}::{}", source, config.name)`) 和"按 priority 排序"这两点。`HookRegistry` 那边不区分 —— 任何实现了 `Hook` trait 的对象（包括 `AuditLogHook`、`RuleHook`、`OutboundWebhookHook`、`SessionSummaryHook` 以及任何用户自定义 impl）注册进去就是平级的"某个 hook"。
## 一句话总结
> `RuleHook` 是在**主流程内同步改写**事件内容（regex 替换/拒绝/前置后置拼接），`OutboundWebhookHook` 是把事件**异步外发到 HTTP 端点**（fire-and-forget，可丢弃）。两者都从同一份 `HookBundleConfig` 解析而来，但一旦注册进 `HookRegistry`，调度器就只认 `Hook` trait 接口，看不出它们的来源（bundled / plugin / workspace）或类型。
*****************
capabilities.json → HookBundleConfig → Hook trait impl → HookRegistry
(配置) (中间表示) (可执行对象) (调度器)
1. 启动/运行时读一份 JSON 配置（HookBundleConfig 上有 rules[] 和 outbound_webhooks[] 两个列表）
2. 每条 rule 调 RuleHook::from_config() 编译成 RuleHook（regex 预编译、timeout 验证），注册进 HookRegistry
3. 每条 webhook 调 OutboundWebhookHook::from_config() 编译成 OutboundWebhookHook（创建 reqwest::Client + semaphore），注册进 HookRegistry
之后运行时触发的过程中，HookRegistry 只看到一堆实现了 Hook trait 的对象，按 priority 排好顺序，按 hook point 分组 —— 不再区分来源是 bundled / plugin / workspace，也不区分是 RuleHook 还是
OutboundWebhookHook。这就是"配置→对象→统一执行"的模式。
************
典型用途 │ 内容脱敏、关键词拦截、改写、拼接 │ 审计/通知：把生命周期事件外发到 Slack/SIEM/IM 机器人八. Scheduler and job tool
// Lazy scheduler slot — filled after Agent::new creates the Scheduler.
// Allows CreateJobTool to dispatch local jobs via the Scheduler even though
// the Scheduler is created after tools are registered (chicken-and-egg).
let scheduler_slot: ironclaw::tools::builtin::SchedulerSlot =
Arc::new(tokio::sync::RwLock::new(None));
// Register job tools even under --cli-only so scheduler-backed jobs remain available.
// Sandbox-only dependencies are injected only when the container manager is running.
components.tools.register_job_tools(
Arc::clone(&components.context_manager),
Some(scheduler_slot.clone()),
container_job_manager.clone(),
components.db.clone(),
job_event_tx.clone(),
Some(channels.inject_sender()),
if config.sandbox.enabled && container_job_manager.is_some() {
Some(Arc::clone(&prompt_queue))
} else {
None
},
components.secrets_store.clone(),
); /// Register job management tools.
///
/// Job tools allow the LLM to create, list, check status, and cancel jobs.
/// When sandbox deps are provided, `create_job` automatically delegates to
/// Docker containers. Otherwise it dispatches via the Scheduler (which
/// persists to DB and spawns a worker).
#[allow(clippy::too_many_arguments)]
pub fn register_job_tools(
&self,
context_manager: Arc<ContextManager>,
scheduler_slot: Option<crate::tools::builtin::SchedulerSlot>,
job_manager: Option<Arc<ContainerJobManager>>,
store: Option<Arc<dyn Database>>,
job_event_tx: Option<
tokio::sync::broadcast::Sender<(uuid::Uuid, String, ironclaw_common::AppEvent)>,
>,
inject_tx: Option<tokio::sync::mpsc::Sender<crate::channels::IncomingMessage>>,
prompt_queue: Option<PromptQueue>,
secrets_store: Option<Arc<dyn SecretsStore + Send + Sync>>,
) {
let mut create_tool = CreateJobTool::new(Arc::clone(&context_manager));
if let Some(slot) = scheduler_slot {
create_tool = create_tool.with_scheduler_slot(slot);
}
// Clone before moving into create_tool so cancel_job can also use them.
let jm_for_cancel = job_manager.clone();
let store_for_cancel = store.clone();
if let Some(jm) = job_manager {
create_tool = create_tool.with_sandbox(jm, store.clone());
}
if let (Some(etx), Some(itx)) = (job_event_tx, inject_tx) {
create_tool = create_tool.with_monitor_deps(etx, itx);
}
if let Some(secrets) = secrets_store {
create_tool = create_tool.with_secrets(secrets);
}
self.register_sync(Arc::new(create_tool));
self.register_sync(Arc::new(ListJobsTool::new(Arc::clone(&context_manager))));
self.register_sync(Arc::new(JobStatusTool::new(Arc::clone(&context_manager))));
let mut cancel_tool = CancelJobTool::new(Arc::clone(&context_manager));
if let Some(jm) = jm_for_cancel {
cancel_tool = cancel_tool.with_sandbox(jm, store_for_cancel);
}
self.register_sync(Arc::new(cancel_tool));
// Base tools: create, list, status, cancel
let mut job_tool_count = 4;
// Register event reader if store is available
if let Some(store) = store {
self.register_sync(Arc::new(JobEventsTool::new(
store,
Arc::clone(&context_manager),
)));
job_tool_count += 1;
}
// Register prompt tool if queue is available
if let Some(pq) = prompt_queue {
self.register_sync(Arc::new(JobPromptTool::new(
pq,
Arc::clone(&context_manager),
)));
job_tool_count += 1;
}
tracing::debug!("Registered {} job management tools", job_tool_count);
}
## 翻译
```rust
/// Register job management tools.
///
/// Job tools allow the LLM to create, list, check status, and cancel jobs.
/// When sandbox deps are provided, `create_job` automatically delegates to
/// Docker containers. Otherwise it dispatches via the Scheduler (which
/// persists to DB and spawns a worker).
```
> 注册 job 管理工具。
>
> job 工具让 LLM 能创建、列出、查看状态、取消 job。
> 当提供 sandbox 依赖时，`create_job` 会自动委派到 Docker 容器执行。
> 否则它通过 Scheduler 派发（Scheduler 会把任务持久化到数据库并启动 worker）。
## 几个关键点
### "Job tools" 是给 LLM 用的 tool 集合
LLM 在对话/agentic 循环中通过 `ToolDispatcher::dispatch()` 调用这些 tool 名（`create_job`、`list_jobs`、`job_status`、`cancel_job` 等），来实现"长跑/沙箱化"的子任务——这是 `.claude/rules/tools.md` 强调的"Everything Goes Through Tools"模式，agent 不能直接调内部 API，必须走 tool 调度器。
### `create_job` 的两条派发路径（同一个 tool 名字，两种实现）
调用方**无感**，因为 tool 名字始终是 `create_job`；区别在于注册 tool 时**闭包里捕获的 dispatch 逻辑不同**：
| 条件 | 实际跑什么 | 用途 |
| ---------------------------------------------------- | ----------- | ---------------------------------------------------------------------------------------------------------- |
| `sandbox deps` 提供了（有 `ContainerJobManager` 等） | Docker 容器 | 隔离执行环境，跑沙箱化代码（CLAUDE.md "Engine v2 Per-Project Sandbox" 描述的 `SANDBOX_ENABLED=true` 路径） |
| 没提供（默认/普通部署） | `Scheduler` | 进程内异步 worker 池，DB 持久化 + `tokio::spawn` 一个 `JobDelegate`（共享 `run_agentic_loop()`） |
### "persists to DB and spawns a worker" 的细节
`Scheduler::dispatch_job()` 在 `src/agent/CLAUDE.md` 里被特别强调为"首选入口"：
> **`dispatch_job()` — creates context, optionally sets metadata, persists to DB (so FK references from `job_actions`/`llm_calls` are valid immediately), then calls `schedule()`. Don't call `schedule()` directly unless you've already persisted.**
也就是说：先 INSERT 到 DB（让 `job_actions`、`llm_calls` 这些外键立即可引用），再 `spawn` worker。如果反过来，**FK 约束会在 worker 中途写入关联表时炸**。
### "auto-delegates" 的含义
"自动"不是说配置变更后 LLM 会动态感知，而是**注册阶段就确定**了（启动时 `sandbox` 模块是否存在决定了闭包里的分支）。tool 注册是一次性的、不可变的（除非显式重新注册）—— LLM 每次调 `create_job` 都走同一条路径。
### 一句话总结
> 这段注释在描述：注册一组让 LLM 管 jobs 的 tool 集合；其中 `create_job` 是个"开关式 tool"——同一个名字，闭包里根据启动时是否提供 sandbox 依赖来决定是走 Docker 容器还是进程内 Scheduler+DB 路径。
job_event_tx 是个 broadcast::Sender，由 sandbox worker（通过 orchestrator HTTP API）发送 (job_id, user_id, AppEvent)；唯一订阅者在 main.rs:1024 —— gateway 启动一个桥接任务把这些事件转交给
▎ Channel::send_status 这个统一入口广播到 SSE，从而让浏览器能收到 sandbox 容器内的进度/状态/工具调用事件。九. AgentStart
// Register message tool for sending messages to connected channels
components
.tools
.register_message_tools(Arc::clone(&channels), components.extension_manager.clone())
.await;
▎ 路径 A（普通回复）= 响应 user input；路径 B（message 工具）= 主动 / 跨 channel / 附件。
▎
▎ 区分的本质是 是否有 IncomingMessage 这个锚点：有 → 路径 A；没有 → 必须用工具或 broadcast_all 直接投递。两类方式互补，覆盖了所有"agent 想跟用户说话"的合法场景。
下面是对 `MessageTool::description()` 全文翻译的整理版，分 5 段呈现。
---
### 一、工具的定位
向 channel 发送**主动消息**。在当前会话中，使用普通的 assistant 输出即可回复；请在以下场景使用本工具：
- 主动通知
- routine/后台跟进
- 发送附件
- 向不同的 channel/收件人发送
言下之意：日常对话的直接回复**不要**走这个工具，让 LLM 直接输出文本即可；只有需要"主动做点什么"的场景才调用它。
---
### 二、参数省略时的默认行为
如果省略 `channel` 和 `target`，会在可用时**复用当前会话的 channel 和发送者/群组**。
这是 LLM 在大多数场景下能省略这两个参数的原因 —— agent loop 已经通过 `set_message_tool_context` 把当前 channel/target 注入到 `MessageTool` 的内部状态里。
---
### 三、只填 `target` 不填 `channel` 的风险
如果提供了 `target` 但没有提供 `channel`，并且没有解析到限定范围的 channel，消息可能会**广播到所有已连接的 channel**，而不是只发到某一个。
这是一个**重要的安全提醒**：LLM 在跨 channel 转发时如果忘了填 `channel`，后果是把消息群发给所有已连接的用户 / 群组 —— 包括不想发的那个。
---
### 四、附件支持
支持文件附件：
1. 先用 `http` 工具下载文件并使用 `save_to`（例如：`http GET https://picsum.photos/800/600 save_to=/tmp/photo.jpg`）
2. 然后把文件路径放到 `attachments` 数组里
图片在 Telegram 上会**作为照片发送**（不是文件形式），这是 Telegram 的特殊处理。
---
### 五、各 channel 的 `target` 格式
| Channel | target 格式 |
| ------------ | ------------------------------------------ |
| **Signal** | E.164 格式（如 `+1234567890`）或群组 ID |
| **Telegram** | 用户名或 chat ID |
| **Slack** | channel ID（`C0...`）或 user ID（`U0...`） |
注意区分：这里的 `target` 在 Slack 上是 **Slack 频道/用户的 ID**，不是 channel 名。在 Telegram 上是 **chat ID**，不是 @username（在很多场景下 username 也能用，但 chat ID 更稳）。
---
### 总结一句
> 这段描述的核心是教 LLM **三件事**：
> 1. **什么时候用**（主动 / 附件 / 跨 channel）
> 2. **省略参数会发生什么**（默认复用 vs 广播风险）
> 3. **每个 channel 的 target 长什么样**（避免格式错误）
格式约束尤其关键 —— LLM 填错 `target`（例如把 Slack 频道名填进去而不是 ID）会导致发送失败，而错误回包机制会让 LLM 自纠。兜底激活需要激活的channel
`PairingStore` 是**通道配对**（pairing）的统一入口，封装"DB 写 + 缓存读"两层。它的核心目的是：**让 channel（Slack relay、Telegram bot 等）识别"哪些外部 sender 是已配对的用户"**。
---
## 整体职责
```
外部 sender（Slack user ID、Telegram chat ID...）
│
▼
┌──────────────────────┐
│ PairingStore │ ← 你贴的这个 struct
│ db: Optional<DB> │
│ cache: 热路径缓存 │
└──────────────────────┘
│
├─ 读路径：cache 命中 → 直接返回
│ cache 未命中 → 查 DB → 回填 cache → 返回
│
└─ 写路径：直接落 DB
remove → 同步清 cache
approve → 不回填 cache（下一读再填）
```
---
## 两个字段的用途
### `db: Option<Arc<dyn Database>>`
**职责**：持久化所有配对状态。
- **`Some`** → 正常模式：写请求到 `channel_pairing_requests` 表，读查询配对关系
- **`None`** → 优雅降级模式：所有操作 no-op / 返回空，用于：
- `--cli-only` 启动（不需要配对功能）
- 测试场景（不需要真实 DB）
这是个标准的"feature-flagged 依赖"模式 —— `Option<Arc<dyn Database>>` 在 main.rs 里按是否有 DB 决定是否实例化。
### `cache: Arc<OwnershipCache>`
**职责**：热路径缓存，避免每次消息进来都查 DB。
关键不变量（来自刚才翻译的注释）：
- **`resolve_identity`**：读穿透，DB 命中时**回填缓存**
- **`remove`**：写完 DB 后**立即清缓存**（防止下次误命中已解绑的身份）
- **`approve`**：写完 DB 后**不回填缓存**（因为 approve 流程拿不到完整的外部 sender identity 信息，要等下次 resolve 时再回填）
---
## 为什么需要这个分层？
每个外部消息到达时（特别是 Slack relay、Telegram bot 这种"长连接 channel"），agent 都要判断：
> "这个 sender ID（`U0123...` / `123456789`）是不是已经配对过某个内部用户？如果是，路由到那个用户的 session。"
这个判断**每次消息都会触发**（每个 user message 至少查一次），而配对关系**很少变化**（用户配对一次后基本不动）。所以：
- **写少读多** → 必须有缓存
- **缓存可能 stale** → 必须能清缓存
- **降级必须平滑** → `db = None` 时不能 panic
这就是 `PairingStore` 这个封装层的存在意义。
---
## 一句话总结
| 字段 | 作用 |
| ------------------------------- | ------------------------------------------------------------- |
| `db: Option<Arc<dyn Database>>` | 持久化所有配对请求 / 批准 / 解绑，None 时降级 no-op |
| `cache: Arc<OwnershipCache>` | 热路径缓存外部 sender → 内部 user 的映射，避免每个消息都查 DB |
两个字段**必须配合**：
- 写路径只动 DB
- 读路径先查 cache，miss 才查 DB 并回填
- `remove` 是唯一同步清缓存的写操作（`approve` 不清，因为不回填）
***************************
/// WASM channel runtime.
///
/// Manages the Wasmtime engine and a cache of prepared channel modules.
pub struct WasmChannelRuntime {
/// Wasmtime engine with configured settings.
engine: Engine,
/// Runtime configuration.
config: WasmChannelRuntimeConfig,
/// Cache of prepared modules by name.
modules: RwLock<HashMap<String, Arc<PreparedChannelModule>>>,
}
**********************************
┌────────────────────┬──────────────────────────────────────────┐
│ 组件 │ 在 hot-activate 流程中的角色 │
├────────────────────┼──────────────────────────────────────────┤
│ ChannelManager │ 把 outgoing message 分发给正确的 channel │
├────────────────────┼──────────────────────────────────────────┤
│ WasmChannelRuntime │ 加载、编译、执行 WASM 模块 │
├────────────────────┼──────────────────────────────────────────┤
│ PairingStore │ 解析外部 sender → 内部 user 的映射 │
├────────────────────┼──────────────────────────────────────────┤
│ WasmChannelRouter │ 入站 HTTP 请求 → 正确 WASM 模块 （/chat/send会查对应正确的wasmChannel） │
└────────────────────┴──────────────────────────────────────────┘
*******************************
//配对流程
好，以**飞书（feishu）**为例，把整条链路重新走一遍。这次要明确 **WASM client 调宿主** 的具体位置。
---
## 飞书配对完整链路
```
飞书 sender 首次 DM（无 approved identity）
│
▼
┌──────────────────────────────────────────────────────────────────────┐
│ 1. 飞书 provider 把 webhook POST 到 /webhook/feishu │
└──────────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────┐
│ 2. WebhookServer (axum) → WasmChannelRouter.match("/webhook/feishu") │
│ src/channels/wasm/router.rs │
└──────────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────┐
│ 3. 路由转发到 feishu WasmChannel 实例 │
│ 调 WasmChannel::handle_webhook → 进入 WASM guest │
└──────────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────┐
│ 4. ⭐ feishu.wasm (guest) 在自己的 handler 里调： │
│ pairing_upsert_request("feishu", sender_id, meta_json) //handle_message │
│ │
│ 这是 WIT 定义的 host function export │
│ guest 侧通过 wit-bindgen 生成的 stub 调用 │
└──────────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────┐
│ 5. ⭐ 宿主侧实现在 src/channels/wasm/wrapper.rs:715-743 │
│ fn pairing_upsert_request(&mut self, channel, id, meta_json) │
│ ↓ block_in_place + handle.block_on(...) │
│ PairingStore::upsert_request("feishu", sender_id, meta) │
│ src/pairing/store.rs:85-110 │
└──────────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────┐
│ 6. DB 层：db.upsert_pairing_request(...) │
│ src/db/libsql/pairing.rs 或 src/db/postgres.rs │
│ INSERT INTO channel_pairing_requests (Pending) │
└──────────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────┐
│ 7. 返回配对码（如 "ABCD-1234"）给 guest //上面调pairing_upsert_request的地方，那里调宿主的http方法，返回给飞书 │
│ guest 把它发给飞书 sender（DM 内显示 / 引导到 web UI） // 发回给飞书 sender 本人 │
└──────────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────┐
│ 8. 用户在 web UI 输入配对码 → POST /api/pairing/approve │
│ src/pairing/approval.rs │
│ → PairingStore::approve("feishu", "ABCD-1234", owner_id) │
│ src/pairing/store.rs:115-134 │
│ normalize_submission → db.approve_pairing(...) │
└──────────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────┐
│ 9. 下一次飞书 DM 进来 │
│ feishu.wasm (guest) 调 pairing_resolve_identity("feishu", sender) │
│ ↓ 宿主 wrapper.rs:745-765 │
│ PairingStore::resolve_identity → cache miss → db → 回填 cache │
│ src/pairing/store.rs:55-81 │
└──────────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────┐
│ 10. guest 拿到 owner_user_id，进入正常消息处理 │
└──────────────────────────────────────────────────────────────────────┘
```
---
## WASM client（guest）调宿主配对代码
### Guest 侧（WASM 模块内部）
具体怎么调取决于 feishu.wasm 的实现。WIT 接口在 `channels/wit/channel_host.wit`（或类似路径），生成出来的 guest 侧 stub 形如：
```rust
// 在 feishu.wasm 内部（伪代码，实际由 wit-bindgen 生成）
match event {
FeishuEvent::DmMessage { sender_id, content } => {
// 调 host function
let upsert = pairing_upsert_request(
"feishu", // channel
sender_id, // external_id
json!({"display": sender_name}).to_string(), // meta
)?;
if upsert.created {
// 第一次：发配对码 DM 给 sender
send_dm(sender_id, format!("Your pairing code: {}", upsert.code)).await?;
} else {
// 已存在 pending 请求：复用 code
send_dm(sender_id, format!("Your pairing code is still: {}", upsert.code)).await?;
}
}
}
```
**关键点**：
- guest 调的是 WIT 生成的 stub 函数（不是 `self.pairing_upsert_request(...)`，而是一个自由函数 / namespace 函数）
- 返回 `PairingUpsertResult { code, created }` —— `created=true` 表示新建，`created=false` 表示复用现有 pending
- guest **不直接接触** DB 或 cache，全部通过 host function
### Host 侧实现
`src/channels/wasm/wrapper.rs:715-743` —— 完整代码：
```rust
fn pairing_upsert_request(
&mut self,
channel: String,
id: String,
meta_json: String,
) -> Result<near::agent::channel_host::PairingUpsertResult, String> {
// 1. 解析 meta JSON
let meta = if meta_json.is_empty() {
None
} else {
serde_json::from_str(&meta_json).ok()
};
// 2. 拿到 PairingStore 的 Arc 克隆
let store = self.pairing_store.clone();
// 3. ⭐ 同步 host → async 桥接
let handle = tokio::runtime::Handle::try_current()
.map_err(|_| "pairing host callback requires a Tokio runtime".to_string())?;
if handle.runtime_flavor() != tokio::runtime::RuntimeFlavor::MultiThread {
return Err("pairing host callback requires a multi-thread Tokio runtime".to_string());
}
// 4. 调 PairingStore
let result: Result<crate::db::PairingRequestRecord, crate::error::DatabaseError> =
tokio::task::block_in_place(move || {
handle.block_on(async move {
store.upsert_request(&channel, &id, meta).await
})
});
// 5. 转换成 WIT 返回类型
match result {
Ok(req) => Ok(near::agent::channel_host::PairingUpsertResult {
code: req.code,
created: req.created,
}),
Err(e) => Err(e.to_string()),
}
}
```
**这段代码就是"wasm client 调宿主"的实际位置**。
注意几个关键设计：
1. **`pairing_store: Arc<PairingStore>`** 是 `WasmChannelWrapper` 的字段（`wrapper.rs:132`），通过 `WasmChannelLoader` 在 setup 时注入（`setup.rs:90-95`）
2. **同步 host function 调 async DB** —— WASM host function 由 WIT 定义为同步（因为 WASM 调用约定限制），但底层 DB 操作是 async。这里用 `block_in_place + handle.block_on` 在 multi-thread runtime 里同步等待。如果 runtime 不是 multi-thread，直接报错（`wrapper.rs:729-731`）
3. **错误转换** —— `DatabaseError` 转 `String`，这是 WIT 边界的常见做法（WIT 接口不能用 Rust 错误类型）
---
## WIT 接口定义在哪？
WIT（WebAssembly Interface Type）文件描述了 host 暴露给 guest 的函数签名。通常在：
```
channels/wit/channel_host.wit (或类似路径)
```
WIT 内容大致：
```wit
// pairing.wit (片段)
pairing-upsert-request: func(
channel: string,
external-id: string,
meta-json: string,
) -> result<paring-upsert-result, string>;
record pairing-upsert-result {
code: string,
created: bool,
}
pairing-resolve-identity: func(
channel: string,
external-id: string,
) -> result<option<string>, string>;
pairing-read-allow-from: func(channel: string) -> result<list<string>, string>;
```
`wit-bindgen` 根据这个 WIT 自动生成：
- **Guest 侧 stub**（feishu.wasm 编译时链接）—— `pairing_upsert_request(...)` 自由函数
- **Host 侧 trait**（铁锈侧实现）—— `impl channel_host::ChannelHost for WasmChannelWrapper { fn pairing_upsert_request(...) }`
`wrapper.rs:715-743` 就是 host 侧 trait 的实现。
---
## 关键代码定位（飞书为例）
| 步骤 | 代码位置 |
| ----------------------------------- | ------------------------------------------------------------- |
| Webhook 路由匹配 | `src/channels/wasm/router.rs` (`match("/webhook/feishu")`) |
| 转发到 WasmChannel | `src/channels/wasm/router.rs` → `WasmChannel::handle_webhook` |
| **⭐ Guest 调 host function** | feishu.wasm 内部（WIT 生成的 stub） |
| **⭐ Host 实现** | `src/channels/wasm/wrapper.rs:715-743` |
| 同步桥接 | `wrapper.rs:727-735` (`block_in_place + handle.block_on`) |
| **⭐ PairingStore::upsert_request** | `src/pairing/store.rs:85-110` |
| DB INSERT | `src/db/libsql/pairing.rs` 或 `src/db/postgres.rs` |
| 配对码生成 | `src/code_challenge.rs` (`CodeChallengeFlow::issue_code`) |
| 用户 approve | `src/pairing/approval.rs` (HTTP 路由) |
| **⭐ PairingStore::approve** | `src/pairing/store.rs:115-134` |
| **⭐ 后续 resolve** | `src/pairing/store.rs:55-81` + `wrapper.rs:745-765` |
| Cache 回填 | `src/ownership.rs` (`OwnershipCache::insert`) |
---
## 一句话总结
> **"WASM client 调宿主配对"的代码 = `src/channels/wasm/wrapper.rs:715-743`**（`pairing_upsert_request` 方法）。Guest 侧是 WIT 生成的 stub（feishu.wasm 内部自由函数），宿主侧是 `WasmChannelWrapper` 实现 trait 方法。整条链路是：`飞书 webhook → axum 路由 → WasmChannel → feishu.wasm guest 调 pairing_upsert_request → WasmChannelWrapper::pairing_upsert_request (block_in_place 桥接 async) → PairingStore::upsert_request → DB INSERT Pending 行 → 返回配对码 → guest 发 DM 给 sender → 用户 UI 输入码 → approve → resolve_identity read-through → cache 回填`。
**************************
这是好问题 —— 这两个"存"存的是 **DB 里两张不同的表**：
| 操作 | 存的表 | 状态 |
| ---------------- | -------------------------- | --------------------- |
| `upsert_request` | `channel_pairing_requests` | **Pending**（待批准） |
| `approve` | `channel_identities` | **已批准 + 绑 owner** |
---
## 两张表的职责
### `channel_pairing_requests`（请求表）
存的是 **"未决请求"**：
| 字段 | 含义 |
| -------------- | ----------------------------------- |
| `id` | 请求 ID（UUID） |
| `channel` | `feishu` |
| `external_id` | 飞书 sender open_id |
| `code` | 配对码（如 `ABCD-1234`） |
| `meta` | 元信息（display name 等） |
| `state` | `Pending` / `Approved` / `Rejected` |
| `requested_at` | 请求时间 |
| `expires_at` | 过期时间（15 分钟） |
| `resolved_at` | 处理时间 |
| `resolved_by` | 哪个 owner 处理 |
每条记录是一次性"票据" —— 批准后变成 `Approved` 状态，但 **不会自动产生 sender → owner 的映射**。
### `channel_identities`（绑定表）
存的是 **"最终绑定关系"**：
| 字段 | 含义 |
| --------------- | ------------------- |
| `channel` | `feishu` |
| `external_id` | 飞书 sender open_id |
| `owner_user_id` | 绑定的内部 user |
每条记录是 `lookup key`，是 `resolve_identity` 真正查询的表。
---
## 为什么需要两张表？
两个表解决 **不同的问题**：
| 维度 | `channel_pairing_requests` | `channel_identities` |
| -------------- | ---------------------------------------------- | ---------------------------------------------- |
| **回答的问题** | "这个 sender 申请过配对吗？码是什么？" | "这个 sender 是哪个 owner 的？" |
| **生命周期** | 临时（15 分钟过期 + Approved/Rejected 后归档） | 长期（直到 remove 才删） |
| **查询模式** | 按 code 查（用户输入码时） | 按 `(channel, external_id)` 查（每次消息进来） |
| **写入频率** | 每次 sender 首次 DM | 每次 approve 成功（一次性） |
如果只用一个表：
- **只用 requests 表**：`resolve_identity` 要按 `(channel, external_id)` 扫整张表，状态机还得在每次查询时校验 `state='Approved'` + `expires_at > now()` —— 性能差、逻辑乱
- **只用 identities 表**：approve 时用什么 code 校验？必须有一行"待批准的票据"在前面
两张表是 **典型的"工作流表 + 终态表"分离**。
---
## 实际数据流（以飞书为例）
```
飞书 sender "ou_abc123" 首次 DM
│
▼
PairingStore::upsert_request("feishu", "ou_abc123", meta)
│
▼
INSERT INTO channel_pairing_requests
(channel='feishu', external_id='ou_abc123',
code='ABCD-1234', state='Pending',
expires_at=NOW()+15min)
│
▼
guest 拿到 code='ABCD-1234'，发给 sender
│
▼
用户在 web UI 输入 "ABCD-1234"
│
▼
PairingStore::approve("feishu", "ABCD-1234", owner_id="alice")
│
├─ 1. UPDATE channel_pairing_requests
│ SET state='Approved',
│ resolved_by='alice',
│ resolved_at=NOW()
│ WHERE channel='feishu' AND code='ABCD-1234'
│
└─ 2. INSERT INTO channel_identities
(channel='feishu', external_id='ou_abc123',
owner_user_id='alice')
ON CONFLICT DO UPDATE
```
approve 这一步**两张表都写**：
1. **更新 requests 表**：`Pending → Approved`（留个审计记录）
2. **插入 identities 表**：建立 sender → owner 的长期映射（这才是 resolve_identity 用的表）
---
## resolve_identity 实际查哪张表？
`PairingStore::resolve_identity`（`src/pairing/store.rs:55-81`）：
```rust
let identity = db.resolve_channel_identity(&channel, external_id).await?;
```
`resolve_channel_identity` 查的是 **`channel_identities`**（绑定表），不是 `channel_pairing_requests`。
也就是说：
- `requests` 表是 **"票据/审计"**
- `identities` 表是 **"运行时映射"**
`resolve_identity` 只看 identities。
---
## 一句话回答
> **`upsert_request` 存的是"待批准的票据"**（`channel_pairing_requests` 表，Pending 状态），**`approve` 存的是"最终的 sender → owner 绑定"**（`channel_identities` 表）。**两张表是不同生命周期**：requests 是临时票据（15 分钟过期），identities 是长期绑定（直到 remove 才删）。`approve` 同时更新 requests（标记 Approved 作审计）和插入 identities（建立运行时映射）。`resolve_identity` 只查 identities，不查 requests。
****************
send_message(sender_id, "open_id", ...) 里：
- URL = 固定的飞书发消息接口，域名从 workspace 配置读，路径硬编码；
- 收件人 = receive_id_type（说明 id 类型）+ receive_id（sender 本人的 open_id，来自 webhook 事件）；
- 鉴权 = 缓存的 tenant access token，走 app_id/app_secret 换取。
所有实际 HTTP 都通过 host function channel_host::http_request 出网（guest 在 WASM 沙箱里，不能自己开 socket，必须经宿主）。 // Wire SSE into plan_update tool for live plan progress broadcasting.
if let Some(ref sse) = sse_manager {计划更新工具
components.tools.register_plan_tools(Some(Arc::clone(sse)));
} // Snapshot memory for trace recording before the agent starts.
// The recorder lives in `ironclaw_llm` and must not depend on the
// host's `Workspace` type, so we materialise entries here.
if let Some(ref recorder) = components.recording_handle
&& let Some(ref ws) = components.workspace
{给 trace 录制器喂一份"agent 启动那一刻整个 workspace 长什么样"的副本，这样以后回放这条 trace 时能完整看到 agent 的初始知识背景，而不是只看 prompt 里实际出现的那几行。
match ws.list_all().await
recorder.snapshot_memory(entries).await;LLM provider 发出的每个 HTTP 请求都过这个钩子；钩子既能记下真实流量（录制），也能用历史录好的响应冒充新响应（回放），还能把出站请求重写到 mock 服务器（测试）——三件事共用一个 trait
接口，链式组合按顺序短路。
let http_interceptor = ironclaw::http_intercept::chain(
[
components.http_interceptor.clone(),
components
.recording_handle
.as_ref()
.map(|r| r.http_interceptor()),
]
.into_iter()
.flatten(),把 secrets store + skill registry + extension manager + tool registry 这四样东西打包成一个 AuthManager，作为引擎 / 网关 / 扩展运行时在所有"工具有没有凭据 / 要不要走 setup /
该解析到哪个扩展"问题上的单一权威入口。没有 secrets store 时直接是 None，整个 auth 流降级关闭。
let auth_manager = components.tools.secrets_store().cloned().map(|secrets| {
Arc::new(ironclaw::auth::extension::AuthManager::new(
secrets,
components.skill_registry.clone(),
components.extension_manager.clone(),
Some(Arc::clone(&components.tools)),
))
}); let mut agent = Agent::new(
config.agent.clone(),
deps,
channels,
Some(config.heartbeat.clone()),
Some(config.hygiene.clone()),
Some(config.routines.clone()),
Some(components.context_manager),
Some(session_manager),
);
// Fill the scheduler slot now that Agent (and its Scheduler) exist.
*scheduler_slot.write().await = Some(agent.scheduler());## 这是 sandbox 容器孤儿清理后台任务
`src\orchestrator\reaper.rs:1` 注释明说：
> **Problem:** If the agent process crashes between container creation and cleanup, containers are orphaned indefinitely.
>
> **Solution:** Background reaper task that: 1) Scans Docker for containers with the `ironclaw.job_id` label, 2) Checks if each job is active in the ContextManager, 3) Cleans up containers with inactive/missing jobs.
### 它解决的具体问题
sandbox job 创建 Docker 容器跑 agent / Claude Code / shell 等任务，正常流程是任务结束 → `ContainerJobManager` 收尾 → `docker stop` + `docker rm`。
但如果中间宿主进程崩了（OOM、kill -9、机器重启），**这些容器就留在 Docker 里没人收**——它们挂着 `ironclaw.job_id` 标签（worker 启动时打的），但对应 job 在 `ContextManager` 里已经查不到了。每次跑任务都开新容器，旧的就一直堆着，吃磁盘、占内存、占端口。
### 代码做的事
```rust
if let Some(ref jm) = container_job_manager {
let reaper_jm = Arc::clone(jm);
let reaper_config = ReaperConfig {
scan_interval: Duration::from_secs(config.sandbox.reaper_interval_secs),
orphan_threshold: Duration::from_secs(config.sandbox.orphan_threshold_secs),
..ReaperConfig::default()
};
let reaper_ctx = Arc::clone(&reaper_context_manager);
tokio::spawn(async move {
match SandboxReaper::new(reaper_jm, reaper_ctx, reaper_config).await {
Ok(reaper) => reaper.run().await,
Err(e) => tracing::error!("Sandbox reaper failed to initialize: {}", e),
}
});
}
```
拆开看：
1. **`if let Some(ref jm) = container_job_manager`**：sandbox 没启用（`SANDBOX_ENABLED=false`）或 `ContainerJobManager` 没构造出来时**根本不启动 reaper**——reaper 是 sandbox 专属，没 sandbox 就不需要它。
2. **克隆 `Arc`**：`reaper_jm` / `reaper_ctx`——`tokio::spawn` 要 `'static`，所以把所有权 move 进闭包。
3. **`ReaperConfig`**：三个旋钮：
- `scan_interval`：每多久扫一次 Docker。默认 300 秒（5 分钟）。
- `orphan_threshold`：容器创建时间距今多久 + 没活 job 就删。默认 600 秒（10 分钟）。
- `container_label`：找 Docker 容器用的 label key。默认 `"ironclaw.job_id"`（用 `..ReaperConfig::default()` 时保持默认）。
4. **`SandboxReaper::new(...).await`**：**立刻连 Docker**——失败的话 reaper 直接 log error 退出，不影响主进程启动。`bollard::Docker` 客户端拿不到就启动不了，这是合理降级（没 Docker 的环境本来也跑不了 sandbox 任务）。
5. **`reaper.run()`**：`tokio::spawn` 一个长生命周期后台任务，进入循环：
```rust
let mut interval = tokio::time::interval(scan_interval);
interval.set_missed_tick_behavior(MissedTickBehavior::Skip); // 扫一次超过 interval 时不补跑
loop {
interval.tick().await;
self.scan_and_reap().await;
}
```
`scan_interval == 0` 时**主动拒绝启动**（注释 73 行："validate to prevent tokio::time::interval panic"），避免误配置把进程搞崩。
### `scan_and_reap` 单次干的事（reaper.rs:90 起）
```
1. docker.list_containers(label=ironclaw.job_id) → 拿到所有 IronClaw 容器
2. context_manager.active_job_ids() → 当前活 job 集合
3. 对每个容器：
- 抽 label 里的 job_id
- 不在 active 集合里 AND 创建时间 > orphan_threshold → docker stop + docker rm
4. 记录统计（reaped / skipped / errors）
```
### 为什么 `orphan_threshold` 不为 0
如果阈值是 0，正常任务刚启动还没来得及在 `ContextManager` 注册的几十毫秒空窗里，reaper 会误删——这是经典 race。所以默认 600 秒留足余量；只有"超过 10 分钟还没活 job 跟着"的容器才是孤儿。
### `container_label` 为什么有默认值还要 `..ReaperConfig::default()`
符合 `error-handling.md` 的 fail-loud：旋钮没设就用合理默认，**不会因为少一个字段就构造不出来**。同时 `container_label` 是内部协议字段（与 worker 启动时打的 label 一致），外部配置里通常不动，留默认值最稳。
### 整体架构定位
```
agent 进程
├─ 主任务循环
├─ 心跳任务
├─ 各种 listener
└─ SandboxReaper (tokio::spawn) ← 这块加的就是这个
├─ 每 5 分钟扫一次 Docker
├─ 对比 ContextManager 的活 job 集合
└─ 删孤儿容器
```
属于"**主进程的生命周期应该负责的所有清理工作**"——主进程崩了也要有人兜底，所以单独立任务、不阻塞启动。
### 一句话总结
**每隔 N 分钟扫一次 Docker 找挂了 `ironclaw.job_id` 标签的容器，对比 `ContextManager` 里还活着的 job 集合，把"超过 orphan_threshold 还没活 job 跟着"的容器当孤儿删掉**——堵住"agent 进程崩在容器创建和清理中间"的资源泄漏口，沙箱没启用时整个块不启动。 // Give the agent the routine engine slot so it can expose the engine to the gateway.
agent.set_routine_engine_slot(shared_routine_engine_slot);
// Prepare SIGHUP handler for hot-reloading HTTP webhook config
// Broadcast channel for clean shutdown of background tasks
let (shutdown_tx, _) = tokio::sync::broadcast::channel::<()>(1); agent.run().await?;