# 框架总览：一次对话从输入到产出

> 本文把整个框架串起来：单轮/多轮上下文、事件层、agent 运行、协作、DAG，
> 在一次对话的时间轴上如何咬合。文件和符号引用指向当前代码；易漂移的行号不作为依据。
> 已设计但尚未落地的部分集中在末尾「已知边界」。

**贯穿全文的一句话：整个框架靠一个进程级事件总线把各子系统解耦相连。** dispatcher、agent loop、工具执行、存储、协作彼此不直接调用对方的 UI/广播逻辑，而是 `emit` 一个事件，谁关心谁订阅（`openprogram/events/bus.py` 的 `EventBus.emit` / `EventBus.subscribe`）。事件层有两条 lane：**异步旁观**（`EventBus`，谁也拦不住正在发生的事）和**同步问询**（`tool_gate`，全框架唯一能拦住工具执行的点）。

---

## 第一部分　一次对话的生命周期（主线）

以「用户输入一句话」为起点，真实数据流如下。每步标注：发生在哪、emit 什么事件、单轮 vs 多轮差异、分支（branch）如何体现。

### 数据流（时间轴）

```
用户输入一句话
   │
   ▼  [入口] process_user_turn(req)              `openprogram/agent/dispatcher/__init__.py` 的 `process_user_turn`（同步，内部起 asyncio loop 跑到完）
   │
   ├─▶ 1. session 建立/加载                       :175 get_session / :177 create_session
   │
   ├─▶ 2. 历史解析（分支在此第一次体现）           :186–198
   │       branch_from=INHERIT_PARENT → db.get_branch() 沿活跃分支父链回走（= 多轮）
   │       branch_from=None           → []（根级 fork，从空起）
   │       branch_from="<node_id>"    → db.get_branch(sid,node)（兄弟 fork，看到分叉点为止）
   │
   ├─▶ 3. 持久化 user 消息（先写后跑，崩了也留痕）   :258–312；写成 DAG Call 节点 caller=ROOT（:298）
   │       emit chat_ack(:314) + user_message(:326)  ← 让 UI 实时显示
   │       emit_safe("user.prompt_submitted")        :346  ← 进事件总线
   │
   ├─▶ 4. 绑定 turn 上下文（ContextVar）            :366–436
   │       _current_turn_id.set(assistant_msg_id) :379   ← turn 内任何协程都读得到同一 turn id
   │       _store.set(SessionNodeWriter)            :435   ← 深层 runtime/工具/Agent method 写同一 DAG
   │       assistant_msg_id = user_msg_id+"_reply" :164
   │       写 assistant 占位行 + set_head           :460；status="running" :464
   │
   ├─▶ 5. ★ 调模型前：上下文引擎先跑一遍 ★          `openprogram/agent/dispatcher/loop_runner.py` 的 `run_loop_blocking`
   │       a. ContextEngine.prepare(...)           :885  ← DAG 历史渲染成 LLM messages（TurnPrep）
   │       b. should_auto_compact(prep)?           :896
   │            是 → snip（免费删最老 turn）不够 → compact（LLM 压缩）→ 重新 prepare（:907/:941/:976）
   │       c. 拼出 prompt（loop 内只加一次，避免破坏 prompt 缓存）
   │       d. agent_loop([prompt], context, ...)   :1074 → 进核心循环（见下 §核心循环）
   │
   ├─▶ 6. 持久化 assistant 消息                     persist_assistant_message :676（persistence.py:31）
   │
   ├─▶ 7. finalize（收尾，context 子系统在这里做 3 件事）  finalize_turn :711 / dispatcher/finalize.py:175
   │       · head/token 更新
   │       · ContextCommit 回填                      finalize.py:283
   │       · ctx_engine.after_turn(...)             finalize.py:308 → engine.py:437
   │            ↑ after_turn 把 provider usage 回灌进 tracker
   │       · git commit_turn(:356) + 项目自动提交(:369 commit_turn_changes)
   │       · auto-title、usage 反馈、快照淘汰
   │       db.update_session(status="idle"/"done")  :727/:729
   │
   └─▶ emit chat_response(:735) → 返回 TurnResult(:739)
```

### 入口：dispatcher

`process_user_turn(req, *, on_event, cancel_event)` 是全框架**唯一**对话入口（`openprogram/agent/dispatcher/__init__.py` 的 `process_user_turn`）。它是**同步**的，便于 channel worker 线程直接调用，内部自起 asyncio loop 跑 `agent_loop` 到完。返回 `TurnResult`（`dispatcher/types.py:102`）。

### turn_id 绑 ContextVar——框架解耦的另一根脊柱

`_current_turn_id.set(assistant_msg_id)`（`:379`）是关键：ContextVar 沿 asyncio task 传播，turn 内**任何**协程（工具执行、`Agent` method、`send_message`）都读得到同一 turn id，从而把文件备份、子分支父锚点都归到正确的 assistant 消息上。同时绑 `_store`（`:435`）让深层 runtime 写同一 SQLite DAG，无需层层透传。`finally` 块成功/异常/提前 return 都会 `reset`。

### 上下文组装：单轮 / 多轮 / 分支

历史解析集中在 `:186–198`，**分支在数据流里第一次体现**：

| 情况 | branch_from | 取的历史 | 含义 |
|---|---|---|---|
| 普通追加 | `INHERIT_PARENT` | `db.get_branch(sid)` 沿活跃分支父链回走 | **多轮**：看到当前分支全部历史 |
| 根级 fork | `None` | `[]` | LLM 从空起 |
| 兄弟 fork | `"<node_id>"` | `db.get_branch(sid, node_id)` | 看到分叉点为止，不被活跃分支污染 |

**单轮 vs 多轮唯一差别就是 `get_branch` 回走出来的链长度**：单轮链上只有刚写的 user 节点（挂在 ROOT 下），多轮则是一条完整父链。

> **分支怎么"写"出来**（上面讲的是怎么"读"）：fork 的写入侧 = 给 user 节点指一个非 active-tail 的 caller（存储层 `set_head` 改 UI 指针），或 `send_message` 起新根（见 §⑤）。读=三种 get_branch，写=换 head 指针 / 新建根。

### ★ 调模型前：上下文引擎先跑一遍（每轮自动压缩主路径）★

这是最容易被忽略、但每个 turn 都发生的一层。step 5 的实体是 `_run_loop_blocking`（`openprogram/agent/dispatcher/loop_runner.py` 的 `run_loop_blocking`），它在进 `agent_loop` **之前**：

1. `ContextEngine.prepare(agent, session, history, model, tools)`（`:885` → `engine.py:194`）：把 DAG 历史**渲染成 LLM 输入** messages（默认走 DAG 渲染 `_build_messages_from_dag` `engine.py:558`；config `context.render="legacy"` 回退到 commit-chain）。返回 `TurnPrep`（`context/types.py:102`）。
2. `should_auto_compact(prep)`（`:896`）为真时——**这是上下文超预算时真正触发的链路**：先 `snip`（免费删最老 turn），不够再 `_ctx_engine.compact(...)`（LLM 压缩），然后**重新 prepare**（三段重试 `:907`/`:941`/`:976`）。
3. 拼出 prompt（loop 内只加一次，避免重复破坏 prompt 缓存）。
4. `agent_loop([prompt], context, config, ...)`（`:1074`）进核心循环。

> 这条「turn 开头先压一遍」和末尾「idle-gap microcompact」是两条不同的压缩路径，别混（见 §③）。

### 核心循环：调模型 → 工具 → 回灌 → 循环

`agent_loop`（`agent_loop.py:114`）建 `EventStream`，内部 `_run_loop`（`:205`）的内循环（`:236`）：

1. push `AgentEventTurnStart`（多轮每轮都 push，`:246`）。
2. `_stream_assistant_response` 调模型（`:260`）。emit `model.response_started`（`:442`）/ `model.response_completed`（`:466`）。
3. 抽 `ToolCall`（`:273`）。有工具 → `_execute_tool_calls`（`:278`），结果 append 回 context（**回灌**，`:288–290`）。
4. push `AgentEventTurnEnd`。
5. 无更多工具且无 steering/follow-up → break。

聊天轮次没有内层硬上限（与 Codex / DeepSeek 一致）。`runtime.exec` 仍默认 20。

### 工具执行：tool.before 拦截

`_execute_tool_calls`（`openprogram/agent/agent_loop.py` 的 `_execute_tool_calls`）对每个 tool call：

1. push `AgentEventToolStart`（`:675`）+ 插件 hook `TOOL_BEFORE_USE`（`:686`，best-effort）。
2. **事件层 tool.before**：`make_event("tool.before",...)` + `emit`（`:695`）——一份事件，异步旁观和同步问询共用。
3. **同步 gate**：`decide_tool_gate(before_ev)`（`:701`）。**全框架唯一能拦住工具执行的点**（`openprogram/events/tool_gate.py` 的 `decide_tool_gate`）：任一 gate 返回 deny 即拦（理由合并），gate 抛错按 allow（fail-open）。被拦 `raise ToolGateDenied`（`:708`），deny 理由作为 error tool result 回模型。**对 subagent 也生效**——gate 在 `permission_mode` approval 包装之外，`bypass` 关不掉它（`events/tool_gate.py:14–15`）。
4. `tool.execute(...)`（`:731`），前后做 cwd 快照 + 文件 checkpoint。
5. push `AgentEventToolEnd`（`:758/:766`）+ emit `tool.after`（`:772`）。
6. 组 `ToolResultMessage` 回灌（`:792`）。每次工具后检查 steering（`:806`），命中则跳过剩余工具、回灌 steering。

### 持久化 + finalize（context 子系统在这步做三件事）

- 出错路径（`:605–639`）：错误折进占位行或独立 error 节点，`head_id` 移到失败 turn（`:612/:618/:622`）。
- 成功路径：`persist_assistant_message`（`:676`，`dispatcher/persistence.py:31`）写 assistant 消息 → `finalize_turn`（`:711`，`dispatcher/finalize.py:175`）。**finalize 里 context 子系统做三件事**：① head/token 更新 ② **ContextCommit 回填**（`finalize.py:283`，把这轮压缩决策固化成不可变 per-turn commit）③ **`after_turn`**（`finalize.py:308` → `engine.py:437`，usage 回灌）。随后 git `commit_turn`（`finalize.py:356`）+ 项目自动提交（`:369`）。

### DAG 更新——贯穿全程，不是单独一步

user 节点（`:298`）、assistant 占位、每个工具结果、`Agent` method 内部节点，都通过 `_store` ContextVar 落入同一 `SessionNodeWriter`，turn 末 `commit_turn`（`openprogram/store/session/session_store.py` 的 `commit_turn`）把整棵工作树作为一次 turn 提交——append-only、无可变"当前态"镜像文件，两个 agent 并发写不会撞同一文件。

---

## 第二部分　分层参考（查细节）

### 事件全表（核心：子系统靠它解耦相连）

实际在发的事件（全仓 `emit_safe` / `emit_ws_frame` / `make_event` 扫描 + 核对）。两类：进总线的 typed 事件（异步旁观 + 同步问询）和透传前端的 `ws.frame`。

| 事件 type | 谁发（文件和符号） | 谁收 | 备注 |
|---|---|---|---|
| `user.prompt_submitted` | dispatcher `:346` | proactive observer（`proactive/state.py:61`） | 用户消息已提交 |
| `tool.before` | agent_loop `:695` | **tool_gate（同步）** + 旁观 | 唯一拦截位 |
| `tool.after` | agent_loop `:772` | 旁观 | 携带 `is_error` |
| `model.response_started` | agent_loop `:442` | 旁观 | 模型流开始 |
| `model.response_completed` | agent_loop `:466` | proactive 收尾策略 | 收尾时机检查 |
| `subagent.started` / `.ended` | task/runner `:115`（origin=`system`，session 显式传参，因 worker 线程 ContextVar 不可靠） | 旁观 | 子 agent 状态漏斗 |
| `branch.message_sent` | send_message `:266` | 旁观 + `ws.frame` | from/to/is_new |
| `question.asked` | questions `:164`（同时 `emit_ws_frame` `:161` 成前端卡片） | channels question bridge（`_question_bridge.py:43`） | 既进总线又发 ws 帧 |
| `question.replied` | questions `:275`（`resolve_question_and_broadcast:262`） | 前端 | **只走 ws 帧** |
| `question.rejected` | questions `:173/:276` | 前端按"收回"处理 | **只走 ws 帧** |
| `context.compacted` | context 引擎 | UI / 旁观 | 压缩已发生 |
| `file.changed` | functions watcher 等 | 旁观 | 文件变更 |
| `channel.message_inbound` | channels | 旁观 | 入站消息 |
| `memory.ingest_started` / `.ended` | memory | 旁观 | 记忆摄入 |
| `skills.changed` / `plugins.update_available` / `sessions.listed` / `branches.listed` | 各子系统 | UI / 旁观 | 列表与可用更新 |
| `ws.frame`（`openprogram/events/bus.py`） | 外部源 `emit_ws_frame`（`:118`） | `openprogram/webui/server.py`（原样广播） | 透传信封：外部源不直连 webui `_broadcast` |

**订阅侧实际位点**：proactive 引擎订阅**全部**事件再按 `on` 过滤（`proactive/engine.py:145`）；webui 只订 `ws.frame`（`server.py:1192`）；channels question bridge 只订 `question.asked`（`_question_bridge.py:43`）。

**事件契约**：`emit` 是 fire-and-forget，handler 抛错绝不反噬发射方（`events/bus.py:141–157` + `_call:182–198` 打 stderr）；async handler 无 loop 时跳过；`emit_safe`（`:96`）整体 try/swallow——「事件层绝不破坏调用方代码路径」。

---

### ① 存储层　SessionStore，分支 = (session, head)

**职责**：每 session 一个 git 仓 + 内存索引；append-only、无可变当前态镜像。
**关键文件**：`store/session/session_store.py`、`store/session/memory_index.py`。
**关键机制**：
- `_open(sid)`（`:413`）返回 `(GitSession, SessionMemoryIndex)`，LRU 缓存、容量淘汰；索引可从 git 无损重建。
- **分支 = (session, head 指针)**：`get_branch(sid[, head])`（`:817` / `memory_index.py:101`）沿 `predecessor`/`caller` 父链回走；`set_head`（`:880`）改的是 meta 里单值的 UI 指针。fork 只是从某 node 起再走一条链，互不污染。
- `commit_turn(sid, msg)`（`:504`）：把工作树作为一次 turn 提交。
**对外事件**：`sessions.listed`、`branches.listed`。

### ② 事件层　总线 + tool.before 拦截

**职责**：进程级 fan-out，解耦所有子系统。
**关键文件**：`openprogram/events/`（bus.py / tool_gate.py）、`agent/questions.py`。
**关键机制**：
- `EventBus`（`:129`）：typed `subscribe(handler, types=...)`（`:159`）+ legacy channel `on`（`:208`）；进程单例 `get_event_bus()`（`:241`，双检锁）。
- **tool.before 同步拦截**：`register_tool_gate`（`events/tool_gate.py:38`）/ `decide_tool_gate`（`:53`），取最严、fail-open。gate 必须快，不许调 LLM / 慢 IO。
- 问询子系统：`QuestionRegistry`（`questions.py:61`，进程级待答表、claim-once、线程安全）。
**对外事件**：见上表。

### ③ 上下文引擎　组装 / 压缩 / ContextCommit

**职责**：把 DAG 历史渲染成 LLM 输入，并做压缩决策。
**关键文件**：`context/engine.py`、`context/microcompact.py`、`context/references.py`、`context/render.py`、`context/commit/`、`context/rules/`、`context/tool_aging/`；finalize 在 `agent/dispatcher/finalize.py`（不在 context/ 下）。
**关键机制**：
- `ContextEngine.prepare(...)`（`engine.py:194`）返回 `TurnPrep`（`context/types.py:102`）。默认从 DAG 渲染（`_build_messages_from_dag:558`），失败回退 legacy（`:218–220`，免得一个坏 commit 拖垮整个 turn）。
- **两条压缩路径**（别混）：
  - **turn-start auto-compact（主）**：`should_auto_compact` 为真 → `snip`（免费删最老 turn）→ 不够则 `compact`（LLM 压缩）→ re-prepare。在 `_run_loop_blocking` 里，每轮调模型前（见第一部分 step 5）。
  - **idle-gap microcompact（次）**：`microcompact.py:76`，**仅空闲 > 3600s 触发**（`GAP_THRESHOLD_SECONDS:45`，prompt 缓存到期后清理"零额外成本"），保留最近 5 个 tool_result，更老的大结果替占位符。**非破坏性**（返回拷贝，不动 DAG 节点）。
- **引用扫描**：`ReferenceTracker.build`（`references.py:114`）廉价子串扫描，标记被后文引用的 tool_result 避免被压。（注：目前结果仅用于日志，ContextCommit 规则尚未消费它，见已知边界。）
- **ContextCommit**（`commit/types.py:104`）：每 turn 的压缩决策固化成不可变 commit；`finalize_turn` 回填（`dispatcher/finalize.py:283`）。
- **`after_turn`**（`engine.py:437`，finalize 里调 `dispatcher/finalize.py:308`）：usage 回灌。
**对外事件**：`context.compacted`。

### ④ agent 运行　loop + 工具注册 + 子 agent

**职责**：驱动「模型↔工具」循环，执行工具，派生子 agent。
**关键文件**：`agent/agent_loop.py`、`agent/agent.py`、`agent/sub_agent_run.py`、`providers/utils/event_stream.py`、`agent/management/gating.py`。
**关键机制**：
- `EventStream`（`event_stream.py:15`）：async-iterable + 终值；provider dict 事件归一成 typed。
- `agent_loop` / `_run_loop`（`:114/:205`，硬上限 50 `:226`）+ `_execute_tool_calls`（`:654`）。
- `Agent` 支持 `steer`（中途插队）/ `follow_up`（收尾追加）/ `prompt`。
- **工具/技能/MCP 静态准入**：`gate(name, disabled, allowed, categories)`（`management/gating.py:38`，fnmatch，解析序 disabled→allowed→categories）。注意：这是**静态准入**（agent.json 声明谁能用），与 ② 的 `tool_gate` 运行时拦截是两回事。
- **子 agent**：`run_agent_turn(sid, prompt, agent_id, branch_from, label)`（`sub_agent_run.py:41`）内部就是再调一次 `process_user_turn`（`:96`），用 `source="agent_spawn"`、`permission_mode="bypass"`（`:89`）。返回 `AgentTurnResult`（`:32`，`head_id`=新分支 tip）；`label` 经 `set_branch_name` 成命名分支（`:109`）。
**对外事件**：`model.response_started/completed`、`subagent.started/ended`（后者由 task/runner 发）。

### ⑤ 协作　send_message + 跨 session + 防护

**职责**：分支/会话之间投递消息、跑分支、把回复带回来。
**关键文件**：`programs/tools/send_message/send_message/send_message.py`、`programs/tools/send_message/list_agents/list_agents.py`。
**关键机制**：
- `send_message(message, to, agent_id)`（→ `_send_message_impl`）。
- to 语义（`_parse_to`）：`SID:HEAD`（投到已存在分支 = 从其 head 再跑一轮）或分支名。建分支归 `agent` 工具（spawn / 从节点 fork）；`to="new"` 语法直接报错并指向它。
- 父锚点 `_resolve_parent`（`:74`）读 dispatcher 的 session/turn ContextVar，**turn id 缺失时回退到 session head**（修了"no active parent turn"）。
- 投递一律异步：交给 task runner（`run_agent_turn_async`），跑完写 attach pointer 并 dispatch followup 回**发起方** session（回复自动回流）。
- **防护**：消息预算 `agent.max_messages`（默认 8，`depth.py`，子继承 count+1，A↔B 来回也计入）；自指守卫（投给自己当前 turn 直接拒）；目标 session 必须存在不静默创建；tool.before 拦截（值守可拦）。
**对外事件**：`branch.message_sent`；并 `emit_ws_frame("branch_message",...)`（`_emit_branch_ui`）在发起方聊天流显示「已发送」行。

---

## 已知边界

- **tool.before 仅「观察 + deny」，尚不能 mutate/veto-by-plugin**：插件 hook `TOOL_BEFORE_USE` 注释明确写 future-work（`agent_loop.py:682–684`）。
- **gate 的 "ask" 三态尚未完全接 ApprovalRegistry**：`Gate.ask` 注释「接 ApprovalRegistry，后续单元接」（`proactive/actions.py:29`）；"critical fail-closed" 分级也待规则层进场（`events/tool_gate.py:13`）。
- **引用扫描结果未被 ContextCommit 规则消费**：目前仅用于日志（`engine.py:204–212`）。
- **DAG 渲染回退路径与正常路径并存**：坏 commit 时 fall back 到 legacy（`engine.py:218–220`）。
- **核心入口无覆盖测试**：`process_user_turn` / `agent_loop` 若干路径标注「⚠️ no covering tests found」。
- **worker 线程 ContextVar 不可靠**：`subagent.started/ended` 因此由 session 显式传参（`agent/job/runner.py:111–115`）；`send_message._resolve_parent` 为此加了 head 回退。
- **`ContextEngine.after_turn` 有两层**：抽象基类桩 `engine.py:124` 为 `pass`；**具体引擎实现 `engine.py:437` 才是真正在干活的**（usage 回灌），由 `dispatcher/finalize.py:308` 调用。

---

## 主线锚点速查

dispatcher 入口 `openprogram/agent/dispatcher/__init__.py` 的 `process_user_turn`；turn_id 绑定 `:379`；历史/分支解析 `:186–198`；user 节点写入 `:298`；**调模型前 prepare/auto-compact** `_run_loop_blocking :885/:896/:1074`；finalize `:711`/`dispatcher/finalize.py:175`（ContextCommit 回填 `:283`、after_turn `:308`→`engine.py:437`）。事件总线 `events/bus.py:141/159/241`；tool.before 拦截 `agent_loop.py:695/:701` + `openprogram/events/tool_gate.py` 的 `decide_tool_gate`。上下文 `engine.py:194` + 两条压缩路径（auto-compact `openprogram/agent/dispatcher/loop_runner.py` 的 `run_loop_blocking` / `microcompact.py`）。agent loop `agent_loop.py:114/205/654`；子 agent `sub_agent_run.py:41`。协作 `send_message.py:186/393`，深度上限 `:35`。