openprogram/context
August 23, 2026 · View on GitHub
OpenProgram 每个 turn 喂给 LLM 的内容由这个包决定：对话历史怎么存、节点怎么组织、系统提示怎么拼、token 预算怎么分、什么时候触发 compaction、compaction 之后历史怎么持久化。
整个包分两层：
数据底座（DAG） 节点定义、SQLite 存储、Session 管理、Runtime
↑
上下文编排 system_prompt / budget / aging / summarize / persistence
底座定义"对话历史长什么样、存哪里、怎么读写"——所有内容是一张有向无环图（DAG），只有一种节点类型 Call，靠 role 字段（user / llm / code）区分用户消息、LLM 调用、代码调用。时间顺序由节点的 seq 整数表达；两套真正的"边"是 called_by（谁调起我，嵌套关系）和 reads（这次 LLM 调用读了哪些节点，prompt 引用）。
编排层定义"每个 turn 给 LLM 看什么"——拿底座里的历史，按 token 预算切片、把旧 tool_result 脱水、必要时调 LLM 做增量总结，最后把折叠结果写回底座成为新节点。
设计目标：把 Claude Code、Hermes、OpenClaw 三个参考系统各自的强项拿过来，叠加 OpenProgram 独有的扁平 DAG 模型 + DAG re-parent 持久化。
0. 数据底座（DAG）
0.1 节点模型 nodes.py
只有一种节点类型 Call：
@dataclass
class Call:
id: str
seq: int # append 时单调递增 — 唯一的时间排序依据
created_at: float
role: str # "user" | "llm" | "code"
name: str # 模型 id / 函数名 / 用户名
input: Any # 这次调用收到的输入
output: Any # 这次调用产出的输出
called_by: str # 调起我的节点 id（嵌套边）
reads: list[str] # 这次 LLM 调用读了哪些节点（上下文边）
metadata: dict # passthrough 字段袋
三种 role 共用同一个数据结构：
role="user" ─ 用户消息 output = 消息文本
role="llm" ─ 一次 LLM 调用 input = {system: "..."}
output = 模型回复
reads = prompt 中包含的节点 ids
called_by = 调起这次 LLM 调用的 @agentic_function id（如果有）
role="code" ─ 一次函数调用 input = arguments dict
output = 函数返回值
called_by = 调起这个函数的 LLM 或父函数 id
DAG 上的"边"有三套（权威定义见
docs/reference/design/runtime/dag/overview.md
§3，本文其余段落里出现的旧名 called_by 一律读作 caller）：
caller 调用关系图 "谁调起我" 形成 fan-out 的嵌套树
predecessor 对话链 "我在聊天里接在谁后面" 分支靠它区分（同一前驱多个孩子 = fork）
reads 上下文引用 "我看了哪些节点" fan-in 的 LLM prompt 组成
时间顺序由 seq 表达，不是图的边。seq 是节点 append 到 DB 时单调递增的整数，按 seq 升序排就是发生顺序。predecessor 表达的是对话顺序而非时间顺序：两者的区别正是分支能存在的原因——retry 出来的兄弟节点 seq 更大，但 predecessor 与被替换的节点相同。
caller 与 predecessor 都是 Call 的顶层字段，metadata 里没有镜像。
Graph 是一个轻量容器：nodes: dict[id → Call] + _next_seq 整数。add(node) 自动分配 seq；update(node_id, **fields) 用于"入口 append 占位、出口 update output"的 @agentic_function 生命周期（这是 DAG 中唯一违反 append-only 的操作，专门支持实时观察）。
算法 helper：
last_user_message(graph) 找最近一条 user-role Call
linear_back_to(graph, target) seq ≥ target.seq 的所有节点，按 seq 升序
branch_terminals(spawn_id, graph) 沿 called_by 链找每个分支的末端
branch_internal(spawn_id, term, g) 单个分支从 spawn 到 terminal 的内部节点链
fold_history(current_id, graph) 把历史按 turn 折叠：每个 prior turn 留 (user, final-llm) 对
compute_reads(graph, 按 expose / render_range 算出
head_seq=..., @agentic_function 下次 LLM 调用的 reads
frame_entry_seq=...,
render_range=...)
Backward-compat 工厂：保留 UserMessage(content=...) / ModelCall(model=..., reads=..., output=...) / FunctionCall(function_name=..., arguments=..., result=..., called_by=...) 三个名字作为工厂函数，内部都返回 Call 实例。让老代码改动最小。但 isinstance(node, UserMessage) 不再可用——必须用 node.is_user() / is_llm() / is_code() 检查 role。
Call 类上还有几个 property 别名：.content → .output、.model / .function_name → .name、.arguments → .input、.result → .output、.system_prompt → .input["system"]。
0.2 SQLite 存储 storage.py
GraphStore(db_path, session_id) 是一个 session 对应一个 GraphStore 实例。三张表：
sessions id / title / created_at / updated_at / model / agent_id /
source / extra_json / last_node_id
nodes id / session_id / type / predecessor / created_at / seq / data_json
↑ ↑ ↑
所有节点 id role 存 metadata.parent_id（legacy 消息树边）
nodes_fts FTS5 虚表 text / session_id / node_id 全文搜索
nodes.type 列存 role 字符串（user / llm / code）；nodes.seq 是时间排序；data_json 把非列字段（input / output / reads / called_by / metadata）打包成 JSON。FTS5 索引节点的可搜索文本（output + name + 等）。
nodes.predecessor 列保留但装的是 metadata.parent_id——legacy 消息树的父 id。dispatcher / channels / webui 对消息树的依赖（fork / rewind / list_branches 等）通过这个 SQL 索引继续工作，不影响新 DAG 模型的纯净性。
append(node) 是单调 append-only——重复 append 同 id 报错。append 时自动给 node.seq 赋值（如果 caller 没指定）。update(node_id, **fields) 用于 @agentic_function 出口填占位节点的 output，同时刷 FTS。load() 把整个 session 重建成内存 Graph。search(query, limit) 走 FTS5 在 session 内搜，模块级 search_across_sessions(db, query) 跨 session 搜。
0.3 Session 抽象 session.py
DagSession = SessionMeta + Graph + GraphStore + DagRuntime，一个对象包一整个会话的所有状态。
DagSessionManager(db_path) 是多会话管理器：create() / load() / list() / rename() / delete() / exists()，一个 SQLite 文件可以存任意多个 session，按 updated_at 排序返回。
0.4 SessionDB 适配器 session_db.py
DagSessionDB 提供跟老 SessionDB 同名同签名的 API——create_session / append_message / get_messages / get_branch / set_head / list_branches / get_branch_token_stats / search_messages 等 14 个方法，但底下走 DAG schema。
外部 from openprogram.agent.session_db import SessionDB 其实是 SessionDB = DagSessionDB 的别名（agent/session_db.py 是个 16 行 re-export）。dispatcher / channels / webui 代码一行没改、全部跑在 DAG 后端上。
message dict ↔ Call 的双向映射在这一层：
- legacy
role="user"↔Call(role="user", output=content) - legacy
role="assistant"↔Call(role="llm", name=model, output=content) - legacy
role="tool"↔Call(role="code", name=function, input=arguments, output=result) - legacy
role="system"↔Call(role="llm", metadata.role="system")（roundtrip 保留原 role） - 非核心字段（source / attachments /
parent_id/_titledflag / ...）一律塞node.metadata，读出时 hoist 回 dict 顶层
get_branch(session_id, head_msg_id) 沿 metadata.parent_id 链回溯——保留 chat 消息树的 fork/rewind 语义；默认 head 读 sessions 表的 last_node_id。
0.5 LLM Runtime runtime.py
DagRuntime(provider_call, graph, store, default_model) — 干净的 DAG 原生 runtime：
exec(content, reads, model, system, tools)— 拿reads列表的节点 id 渲染成 messages，调provider_call，把回复写成一个新的 llm-roleCall节点 append 到 graph，同时持久化到 storeadd_user_message(content)— 在 graph 上加一条 user-role Callrecord_function_call(name, arguments, called_by, result)— 调用方自己执行完函数后调这个登记结果
DagRuntime 是个"给定 graph + reads，调一次 LLM"的纯函数 wrapper，不感知 @agentic_function，reads 谁来算由调用方负责——chat.py 的 chat 循环用它。跟它并列的是 agentic_programming/runtime.py 里的 Runtime（provider 基类，@agentic_function 注入的就是它）：Runtime.exec 自己感知 @agentic_function——从 ContextVar 取当前 graph/frame、调 compute_reads 算 reads、把 llm 节点写回 DAG（见 §0.7）。两者都不再有任何内存里的 tree Context。
0.6 Chat 循环 chat.py
chat_turn(user_input, runtime, tools, max_iterations) — 一个 LLM↔tool 循环：
循环：
1. fold_history 折叠之前 turn 的细节，加上本轮用户输入
2. runtime.exec → 拿到 LLM 回复
3. parse_tool_call(reply) → 看 LLM 要不要调工具
- 调 → 执行 + 写 code Call + 继续循环
- 不调（纯文本）→ 返回回复，结束
4. 最多 max_iterations 轮，超过强制停
parse_tool_call(text) 容忍 LLM 用 bare JSON、围栏 JSON、或者文本里夹 JSON 多种格式表达"调用工具"。
0.7 @agentic_function ↔ DAG 集成
集成靠两个 ContextVar，没有粘合层（旧的 bridge.py 已删）：
_store（context/storage.py）—— 本回合的GraphStore。_call_id（agentic_programming/function.py）—— 当前正在执行的@agentic_function的 code 节点 id；写在它内部的任何节点都拿这个做called_by。
turn 入口（dispatcher.process_user_turn，以及 webui 的 run 路径）：
_store.set(GraphStore(db_path, session_id))—— 装上本会话的 DAG storerun路径还会_call_id.set(命令消息 id)，把整个 program 执行子树挂到那条命令下finally里 reset 这两个 token
@agentic_function 装饰器（function.py）：
- 入口生成
pending_id，_append_function_call_entry→ 若_store已装，store.append(Call(role=code, output=None, status="running", called_by=_call_id.get()))——webui / 调试器能实时看到正在跑的函数。函数 docstring 一并写进该节点的metadata.doc，render_dag_messages渲染 code 节点时把它拼在函数名(参数)前面,所以函数自己的 LLM 调用能在上下文里看到"这个函数是干什么的" - 入口
_call_id.set(pending_id)——内部的runtime.exec/ 嵌套@agentic_function据此做called_by标记 - 装饰器的
system=在调用期间盖到注入的 runtime 上(_apply_system/_restore_system,save/restore),runtime.exec读runtime.system才拿得到——所以@agentic_function(system=...)是经由 runtime 生效的,不进 DAG 节点 - 出口
_update_function_call_exit→store.update(pending_id, output=..., status=...)——填回同一个节点的 output / status，不写第二个节点；异常路径写output={"error": ...}+status="error" _store没装时全部 no-op（standalone 跑 @agentic_function 不依赖持久化）
Runtime.exec（agentic_programming/runtime.py）：
store = _store.get()；为None时退化成普通 LLM 调用，完全不碰 DAG- 装了 store：从
_call_id拿当前 frame 节点 →compute_reads(...)算出本次要读哪些节点 →render.py::render_dag_messages渲染成 messages → 调 provider - 成功后 append 一个
Call(role=llm, name=model, output=reply, called_by=_call_id.get())到 store；顶层（无 frame）时called_by=""
compute_reads(graph, head_seq, frame_entry_seq, render_range)（在 nodes.py）—— 纯函数，按 DAG 结构算下一次 LLM 调用要读哪些节点：
顶层聊天 seq ≤ head_seq 的所有节点，按 seq 升序
@agentic_function 内部 frame_entry_seq 之前 + frame 内部新增节点
expose='io' 的 code Call 把它内部 llm Call（called_by == this）从 reads 里去掉
expose='full' 保留内部所有 llm Call
render_range['callers'] pre-frame 节点最多保留多少（默认 None 不限，0 = 完全隔离）
render_range['subcalls'] in-frame 节点最多保留多少（默认 -1 不限；frame 自然看见自己进度，子函数内部由各自 expose 负责藏）
所有节点始终写进 DAG（不影响存储），expose 只影响算 reads 时的可见性过滤。compute_reads + render_dag_messages 是 runtime.exec 的 prompt 拼接路径。
总体流程
每个 turn 进入 dispatcher 后：
engine.on_session_start(session_id)— 预热 UsageTracker 缓存engine.prepare(agent, session, history, model, tools)— 同步走 6 步组装出TurnPrep- dispatcher 检查
prep.budget_pct≥ 0.80（AUTO_COMPACT_PCT）→ 内联跑engine.compact(user_initiated=False)，把 LLM summary 写进 DAG，再次prepare- 否则继续
- dispatcher 把
prep.agent_messages喂给 agent_loop，跑 LLM engine.after_turn(session_id, usage, prep, on_event)— 真实 provider usage 喂回 UsageTracker；越过 0.70 阈值时发compaction_recommended事件- 用户手动点
/compact→trigger_compaction()走engine.compact(user_initiated=True)
下面按职责分层逐层展开。
1. 系统提示装配
文件：system_prompt.py::build_system_prompt(agent)
把 5 类信息按固定顺序拼成一段，包在 ── Agent prompt ── 和 ── End of agent prompt ── 之间，模型一眼能看到边界。
装配顺序
[1] 身份 banner "You are <name> (agent_id=<id>). Users may address you via: <mentions>."
[2] 工作区文件 AGENTS.md → SOUL.md → USER.md （按顺序读三个，空的跳过）
[3] 内联 prompt agent.system_prompt （用户在 agents show 里编辑的）
[4] Skill 索引 "<available_skills>" + 每份启用 skill 的 name + 描述首行（截 250 字符）+ SKILL.md 绝对路径
[5] Memory 块 get_backend().system_prompt() （持久化 memory 快照）
为什么是这个顺序：越靠前的越稳定，prefix cache 命中率越高。身份和工作区文件是几乎不变的；inline prompt 偶尔改；skill 列表会随启用/禁用变；memory 每天都在长。把不变的放前面，整段 system prompt 的 cache 命中率最高。
实现细节
- 接受 AgentSpec 对象或 dict，内部用
_attr(obj, name, default)统一访问，webui 传 profile dict、CLI 传 AgentSpec 都能用 - 任何一层抛异常都被吞掉，最差情况退化到只返回
agent.system_prompt，绝不让系统提示装配失败拖崩整个 turn - 工作区文件读取走
openprogram.agent.management.workspace，命中文件系统但有内部缓存 - Skill 索引限 20 条，超出显示
... (+N more)，避免 skill 多的 agent 把 system prompt 撑爆
2. Token 预算
三个文件分工：
tokens.py 估算单条消息和整段历史的 token 数；提供 real_context_window
budget.py 把 context_window 切成 system / history / tools_schema / output_reserve 四段
usage.py 缓存 provider 返回的真实 usage，下一轮 prepare 时用真实数代替估算
2.1 真实 context window
real_context_window(model) 读 model.context_window（输入+输出的总容量），不是 model.max_tokens（这是输出上限，通常只占总窗口的 10–30%）。这点容易踩坑——之前的实现读 max_tokens，导致在 32K 的 max_tokens 上估算预算，但实际窗口是 200K，触发 compaction 时利用率只到 16%。
2.2 Token 估算
estimate_message_tokens(msg) 和 estimate_history_tokens(history)：
- 优先 tiktoken（OpenAI 系模型精确）
- 回退到字符比例估算：
_is_cjk(text)判断主体是不是 CJK，CJK 用 1.3 char/tok，ASCII 用 3.8 char/tok。一段中英混排会按 CJK 字符占比加权。 - 估算的是输入 token，不包含 prompt template overhead，所以会略低于 provider 报数，hybrid 模式（见 2.3）会修正这一点。
2.3 BudgetAllocator
BudgetAllocator.allocate(context_window, system_prompt, history, tools) 返回 BudgetAllocation：
context_window 模型真实总窗口
system_prompt 系统提示 token（estimate_message_tokens）
history 历史 token（estimate_history_tokens）
tools_schema 工具 JSON schema 的 token（json.dumps + 每工具 5 tok 描述 overhead）
output_reserve 预留给 assistant 输出，默认 16384，最低 25% 总窗口
input_used system + history + tools_schema
input_budget context_window - output_reserve
input_used_pct input_used / input_budget
headroom input_budget - input_used
output_reserve 的 25% 下限是必须的：模型生成时会一口气吐到 reserve 上限，如果 reserve 太小，长回答会被截断；太大，输入端可用 token 不够。25% 是 Claude Code 和 Hermes 实测下来的折中点。
2.4 UsageTracker
UsageTracker 维护一个线程安全的 in-memory 缓存，持久化到 SessionDB.extra_meta._usage：
last_prompt_tokens provider 上一轮报的 input_tokens
last_cache_read_tokens provider 上一轮报的 cache_read_tokens
cumulative_prompt_tokens 会话累计
cumulative_completion 会话累计
turn_count 这个会话总 turn 数
compaction_count 这个会话总 compact 次数
source "estimate" | "provider" | "cached"
Hybrid 估算：prepare() 里如果 usage.source == "provider"，说明上一轮拿到了真实数。我们信 provider 报的 last_prompt_tokens（作为已固定前缀的 token 数），再加上本轮新增消息的本地估算，得到一个"信前半、估后半"的混合数。比纯估算准很多，比每次重新估全部历史也快很多。
on_session_end(session_id) 会把 in-memory cache flush 回 SessionDB。
对比
OpenProgram Claude Code OpenClaw Hermes
真实 context_window 有 有 无 部分
Tools schema 计入预算 有 有 无 部分
Provider usage 反馈 有 部分 无 无
CJK-aware 字符估算 有 无 无 无
3. 历史瘦身（aging，不调 LLM）
文件：aging.py::TurnAger + references.py::ReferenceTracker
每个 turn 都跑一次，把旧的、大的、没人再用的 tool_result 块替换成占位符，省 token。不调 LLM、不动消息结构、不动 SessionDB——只在 prepare 的内存副本里替换 extra.blocks 里的 content 字段。
3.1 三闸门
一条 tool_result 块要被 redact，必须同时通过三个闸门：
turn 距离闸门 消息所在 turn 离当前 ≥ keep_recent_turns（默认 4 个 assistant turn）
wall-clock 闸门 消息 timestamp 离现在 ≥ keep_recent_seconds（默认 60 秒）
引用闸门 ReferenceTracker 没有把这条消息标记为 cited
每条单独失败都不动。turn 距离闸门防止 agent 还在用刚抓的数据；wall-clock 闸门防止 agent 在 10 秒内连发 8 个 tool_call 时把刚来的输出也 redact 掉；引用闸门保护 agent 还在分析的旧输出。
通过三闸门后，再看 tool_result 的 content token 数 ≥ 800 才动手——更小的 redact 收益太低（占位符本身也占字数）。
替换后的内容：[Old tool result content cleared (was N tokens)]，并在 block 上加 _redacted: true 标记，下次 prepare 时遇到这个标记就跳过（避免反复处理）。
3.2 保护前 N
protect_first_n 默认 2。前 2 条消息（通常是用户的初始任务描述）永远不被 aging 触碰，即使过了几小时也保留原文。否则在长会话里 LLM 会忘了一开始要干嘛。
3.3 ReferenceTracker
判断"旧消息是不是还在被后续消息引用"。算法不追求精确，只要 catch 90% 的真实引用：
Distinctive 子串提取（regex）：
路径 [A-Za-z0-9_./-]+/[A-Za-z0-9_./-]+ 长度 ≥ 8
hex/数字 id [a-f0-9]{6,} | [0-9]{6,}
反引号片段 `(.{3,40})`
CamelCase [A-Z][A-Za-z0-9]{3,} 长度 ≥ 4
去停用词（the / and / True / False / TextContent / AssistantMessage / ...），每条消息最多保留 32 个 distinctive 子串。
引用判定：扫历史一遍，对每对 (i, j) i<j，检查消息 i 的 distinctive 子串是否出现在消息 j 的 text 里。命中一次就把 i 标为 cited，标完就跳过后续检查。O(messages²) 但每条只比 32 个短串、有 early-exit，几百条消息毫秒级。
ReferenceMap 在 prepare 开头构建一次，aging 和 summarisation 都共享。
对比
OpenProgram Claude Code OpenClaw Hermes
Tool-result aging 有 有 无 无
wall-clock 闸门 有 有 无 无
turn 距离闸门 有 有 无 无
引用追踪保护 有 无 无 有
保护前 N 条 有 无 无 有
800-tok 大小门槛 有 无 无 无
4. 历史压缩（compaction，调 LLM）
文件：summarize.py::Summarizer + prompts.py + persistence.py::Persister
策略综合了 Claude Code、Hermes Agent、OpenClaw 三家的实战经验，每个参数都有明确出处：
参数 默认值 来源
──────────────────── ──────────── ────────────────────────────────────────
trigger_pct 0.80 Hermes 50% 太激进 cache 失效频繁
Claude Code 93% 太晚 compact 时已紧张
emergency_pct 0.95 紧急 fallback，保证下一轮不会爆窗
recommend_pct 0.70 只发 UI 事件不动手
keep_min_tokens 8_000 Claude Code 的 minTokens 思路
keep_max_tokens 40_000 Claude Code 的 maxTokens 直接采用
keep_ratio 0.10 按 window × 比例算 tail
keep_min_messages 5 Claude Code 的 minTextBlockMessages
防止 tail 全是大 tool_result 没真对话
protect_first_n 3 Hermes（任务描述+首回复+澄清）
protect_last_n 20 Hermes，作为 tail-budget 之外的兜底
min_prompt_budget 8_000 OpenClaw 的 small-window cap
min_prompt_ratio 0.25 head 至少留 25% 窗口给 system+新轮
4.1 触发判定（两段式）
if prep.budget_pct >= AUTO_COMPACT_PCT: # 0.80
# 主动 compact — 仍有 budget 留给 summary call
fire auto-compact
elif prep.budget_pct >= RECOMMEND_PCT: # 0.70
# 只发事件 UI 提示
emit compaction_recommended
两段触发的意义：80% 时主动 compact 就享受充足的 summary 预算（call 本身要占 8-15K tokens），不必等到窗口逼近上限——那时连做摘要的余量都没有了。
4.2 找切点 find_cut_index
六步算法：
# 1. 算理想 tail 大小：按比例 + 上下界 clamp
desired = clamp(window × keep_ratio, keep_min_tokens, keep_max_tokens)
# 2. 小窗口 cap：head 必须留出 min_prompt 空间
min_prompt = min(min_prompt_budget, window × min_prompt_ratio)
effective_keep = min(desired, window - min_prompt)
# 3. 从尾巴往前累积 token + 数 text-block 消息
# 双闸门：token 数够 AND text-block 消息数够
for i from end downto protect_first_n:
tail_tokens += estimate(messages[i])
if has_text_block(messages[i]):
tail_text_msgs += 1
if tail_tokens >= effective_keep and tail_text_msgs >= keep_min_messages:
cut = i; break
# 4. protect_last_n 兜底：cut 不能晚于 len - protect_last_n
cut = min(cut, len - protect_last_n)
# 5. protect_first_n 兜底：cut 不能早于 protect_first_n
cut = max(cut, protect_first_n)
# 6. snap 到 user 消息边界（tail 必须从 user 起）
while messages[cut].role != "user":
cut += 1
双闸门（token 数 + text-block 消息数）是 Claude Code 的关键设计。没有 min_messages 时一个 50K 的 tool_result 会单独满足 token budget，结果 tail 里实际对话只剩 1-2 条。加上 ≥5 条 text-block 这个第二闸门后，tail 一定有最少 5 轮真对话原文。
has_text_block 判定：role 是 user/assistant/system 且 content 非空。纯 tool_result wrapper（content="" + extra.blocks 里有 tool_result）不算。
不同窗口下的实际行为：
context_window desired_keep max_safe_keep effective_keep tail 占比
────────────── ────────────── ────────────── ────────────── ─────────
8K 800 → 8K min ~6K (cap) 6K 75%
32K 3.2K → 8K min 24K 8K 25%
64K 6.4K → 8K min 56K 8K 12.5%
200K 20K 192K 20K 10%
272K (GPT-5) 27.2K 264K 27.2K 10%
1M 100K → 40K max ~992K 40K 4%
小窗口（8K-32K）自动 cap，大窗口（>400K）自动 ceiling。中间区间按比例伸缩。
4.3 legacy keep_recent_tokens 覆盖
engine.compact(keep_recent_tokens=N) 调用方可以传一个固定数字强制覆盖整套自适应逻辑，跳过 ratio/clamp/cap 直接用 N。webui /compact <N> 按钮走这条路，给用户手动微调的逃生通道。
4.4 链式 summary
不是每次都从头总结，而是基于上一次的 summary 增量更新。
prepare 时 从 session.extra_meta._last_summary_text 读出上次 summary
Summarizer.summarise(previous_summary=...)
→ 命中走 UPDATE_PROMPT，没有走 FRESH_PROMPT
compact 完成后 把新 summary 写回 session.extra_meta._last_summary_text
FRESH_PROMPT 让 LLM 按 5 段结构产出第一次的 summary：
1. User intent 用户总目标（1-2 句）
2. Decisions 用户表达的每一条具体指令/约束（接近原话）
3. Work completed 动过的文件、跑过的命令、得出的结论（带路径和 id）
4. Outstanding 用户还在等什么、有什么 dangling 问题
5. Active context 需要知道的活动状态（"连着 db X，开着文件 Y，env Z 是 W"）
UPDATE_PROMPT 包一个 <previous-summary> 标签把上一版扔给 LLM，让它"合并新消息、删过期细节、刷进度"，同样输出 5 段结构。冲突时新消息为准。
SYSTEM_PROMPT 框定 summariser 的工作风格：要具体（路径、id、命令名、错误信息），不要 hedging、不要前言"Here is a summary..."、不要 moralising。
4.5 LLM 失败兜底
Summarizer.summarise 包了一层 try/except：
try:
text = await self._llm_summary(...) # 调 provider
fell_back = False
except Exception as e:
text = self._structural_summary(prefix) # deterministic 兜底
fell_back = True
err = f"{type(e).__name__}: {e}"
_structural_summary 按 role + 60 字符 head 列出每条要折叠的消息：
[user] explain the database schema in section 4
[assistant] The schema has three tables: users, sessions, messages...
[user] now refactor the message table to add...
这样即使 provider 401 / 网络挂 / token 超限，summary 也能产出某种总结，agent loop 不会因为 compaction 失败而崩溃。CompactResult 里带 fell_back_to_structural=true 标记和 error 字段，调用方可以日志告警。
4.6 可取消
summarise(cancel_event=threading.Event)。LLM 调用前后都检查 cancel_event.is_set()，set 了就 raise CancelledError 走 structural 兜底。用户在长 summary 跑到一半中断时，agent 不会卡死。
4.7 持久化到 DAG
Persister.insert_summary_node(session_id, summary_text, cut_idx, history)：
- 生成
summary_id = "summary_" + uuid4()[:10] - 写一条
compactionSummary行：
id summary_xxx
role system
content "[Previous conversation summary]\n<summary_text>"
parent_id None 这是关键——summary 作为新链的根
timestamp first_kept_ts - 1e-6 必须早于第一条 kept 消息
type compactionSummary
source compaction
extra {"compaction": true}
- Re-parent kept tail：
for original in history[cut_idx:]，每条复制一份，分配k_<uuid>新 id，parent_id指向链上前一节点。原行不删。 db.set_head(session_id, prev)推进到新链尾。
timestamp 的微妙之处：get_branch 走 parent_id 找到所有节点后用 ORDER BY timestamp ASC 排序返回。如果 summary 用 time.time()（=now）作 timestamp，而 kept tail 保留了原始 timestamp（小得多），summary 会排在 kept tail 后面——branch[0] 不是 summary 而是某条 kept 消息，模型看到的第一条就成了 user 的"turn 18"，完全失序。
解决方案：summary 的 timestamp 设为 history[cut_idx].timestamp - 1e-6，比第一条 kept 消息早一微秒。这是 SQL ORDER BY ASC 唯一能保证 summary 排第一的方式。
为什么 parent_id = None：如果 summary 的 parent_id 指向被折叠的最后一条原消息，get_branch 会从 head 沿 parent_id 走回去，一路走到 summary，然后继续走到 summary.parent，再继续走到 m0——把整段被折叠的历史又走出来了，compaction 等于白做。设成 None 才是真切断。
4.8 原始历史不丢
原历史行（m0..m_{cut-1}）还在 messages 表里，只是不在当前活动分支上。get_descendants(m0) 还能拿到。这意味着：
- 用户可以"checkout"到压缩前的状态对比
- 调试时能看出"LLM 是基于哪段原始上下文产生了这个 summary"
- 误压缩可以撤销（理论上）
代价：DB 体积不会因为 compact 而缩小。但 SQLite 不在乎几千条消息，活动分支变短才是 LLM 在乎的。
对比
OpenProgram Claude Code OpenClaw Hermes
两段式触发 (auto + emergency) 有 无 无 无
按 window 比例自适应 tail 有 部分 无 有
绝对值上下界 clamp 有 有 无 无
小窗口 cap（防死循环） 有 无 有 无
text-block 消息数双闸门 有 有 无 无
protect_first_n 保护任务描述 有 无 无 有
protect_last_n 保护最近 N 条 有 无 无 有
增量 summary 链 有 有 无 有
LLM 失败 structural 兜底 有 无 无 无
可取消的 summarisation 有 有 无 无
DAG re-parent 保留原始分支 有 无 无 无
压缩后可回放调试 有 无 无 无
5. 生命周期与插件化
文件：engine.py::ContextEngine（ABC）+ DefaultContextEngine（默认实现）
5.1 ABC 暴露的 hook
on_session_start(session_id) 会话载入/创建时调用一次，预热缓存
ingest(session_id, message) 新消息落 DB 时调用，默认 no-op；自定义引擎可以维护索引
prepare(agent, session, history, 每个 LLM 调用前跑，返回 TurnPrep
model, tools)
should_recommend(prep) budget_pct ≥ 0.70？
should_auto_compact(prep) budget_pct ≥ 0.80？
compact(agent, session_id, model, 触发 compaction，返回 CompactResult
on_event, previous_summary,
user_initiated, cancel_event,
keep_recent_tokens)
after_turn(session_id, usage, prep, LLM 返回后调用，喂真实 usage、发推荐事件
on_event)
on_session_end(session_id) 会话关闭时调用，flush in-memory state
每个 hook 都可以单独 override。DefaultContextEngine 把这些 hook 接到 6 个组件单例（usage / budgets / ager / summarizer / persister / references）上，子类只要换一个组件就能改一个维度的行为。
5.2 组件注入
DefaultContextEngine.__init__ 全部走 keyword-only 注入：
DefaultContextEngine(
usage_tracker=..., # 换 UsageTracker 子类
budget_allocator=..., # 换 BudgetAllocator 子类
ager=..., # 改 aging 策略
summarizer=..., # 换 summary 模型/prompt
persister=..., # 换持久化层
references=..., # 换引用追踪算法
recommend_pct=0.65, # 阈值微调
auto_compact_pct=0.80,
)
测试里这是主要的 stub 路径——传一个 fake summarizer 就能跑 compact 流程而不调 provider。
5.3 注册表与按 agent 选引擎
CONTEXT_ENGINE_REGISTRY: dict[str, ContextEngine] = {}
register_engine(engine) 把引擎按 engine.name 注册
get_engine(name) 按名取，找不到 fallback default
resolve_engine_for(agent) 按优先级取引擎
resolve_engine_for(agent) 的优先级：
1. agent.context_engine 字段 per-agent 显式指定
2. config.context.engine 全局配置
3. default_engine 兜底
dispatcher 每个 turn 调一次 resolve_engine_for(agent_profile)，所以同一进程不同 agent 走不同引擎是免费的。
对比
OpenProgram Claude Code OpenClaw Hermes
ContextEngine ABC 插件化 有 无 有 部分
Per-agent engine override 有 无 有 有
完整生命周期 hooks 有 无 有 部分
组件级注入 有 无 有 无
6. 文件清单
─── 数据底座（DAG） ─────────────────────────────────────────
nodes.py Call 节点类型 + Graph 容器 + helpers (last_user_message /
linear_back_to / branch_terminals / branch_internal /
fold_history / compute_reads) + backward-compat 工厂
(UserMessage / ModelCall / FunctionCall)
storage.py GraphStore：SQLite 三表（sessions/nodes/nodes_fts），
append + update + FTS5
session.py DagSession + DagSessionManager：一个 SQLite 文件多 session
session_db.py DagSessionDB：跟老 SessionDB 同 API 的适配器（chat msg ↔ Call 映射）
runtime.py DagRuntime：provider_call wrapper，exec → 写 llm-role Call
render.py render_dag_messages：把 reads 节点列表渲染成 provider messages
chat.py chat_turn：LLM↔tool 循环 + parse_tool_call
─── 上下文编排 ─────────────────────────────────────────────
__init__.py 公开 API：default_engine / resolve_engine_for / TurnPrep / CompactResult / ...
types.py 编排 dataclass：UsageState / BudgetAllocation / TurnPrep / CompactResult / ReferenceMap
tokens.py token 估算 + real_context_window + CJK 比例
usage.py UsageTracker：provider usage 缓存 + hybrid 估算
budget.py BudgetAllocator：context_window → 四段切分
references.py ReferenceTracker：distinctive substring 引用图
aging.py TurnAger：三闸门 + 保护前 N 的 tool_result redact
prompts.py SYSTEM / FRESH / UPDATE 三个 summariser 提示
summarize.py Summarizer：找切点 + LLM 调用 + structural 兜底
persistence.py Persister：写 compactionSummary 节点 + re-parent kept tail
system_prompt.py build_system_prompt：5 段分层装配
engine.py ContextEngine ABC + DefaultContextEngine + 注册表
7. 数据类型
UsageState last_prompt_tokens / last_cache_read_tokens
cumulative_prompt_tokens / cumulative_completion_tokens
turn_count / compaction_count
source: "estimate" | "provider" | "cached"
BudgetAllocation context_window / system_prompt / history / tools_schema / output_reserve
input_used / input_budget / input_used_pct / headroom
TurnPrep system_prompt: str
agent_messages: list[Message] 给 agent_loop 用
history_dicts: list[dict] aging 后的 dict 视图
budget: BudgetAllocation
usage: UsageState
tool_results_redacted: int
tokens_freed_by_aging: int
references_protected: int
summary_id: str | None 当前活动 summary id
decision_path: list[str] 遥测：这一轮做了哪些动作
budget_pct: float budget.input_used_pct
context_window: int budget.context_window
CompactResult ok: bool
summary_text / summary_id
summarised_count / summarised_tokens
tokens_before / tokens_after
duration_ms
used_previous_summary: bool
reason: "auto" | "manual" | "recovered"
error: str | None
fell_back_to_structural: bool
ReferenceMap cited_tool_use_ids: set[str]
quoted_snippets_by_msg: dict[str, set[str]]
last_built_at: float
DAG 节点类型（底座，nodes.py）：
Call id: str
seq: int 时间排序，append 时单调递增（-1 = 未存）
created_at: float
role: str "user" | "llm" | "code"
name: str model id / 函数名 / 用户名
input: Any prompt 信息 / arguments dict
output: Any reply text / result / None
called_by: str 调起我的节点 id（嵌套关系边）
reads: list[str] 这次 LLM 调用 prompt 包含的节点 id
metadata: dict passthrough 字段袋（source / expose / status / parent_id / ...）
# backward-compat factory functions（返回 Call）：
UserMessage(content) → Call(role="user", output=content)
ModelCall(model, reads, → Call(role="llm", name=model, reads=reads, output=output,
output, system_prompt) input={"system": system_prompt})
FunctionCall(function_name, → Call(role="code", name=function_name, input=arguments,
arguments, result, output=result, called_by=called_by)
called_by)
Backward-compat property accessors on Call：
.content → .output (UserMessage 风格)
.model → .name (ModelCall 风格)
.system_prompt → .input["system"]
.function_name → .name (FunctionCall 风格)
.arguments → .input
.result → .output
8. 与其他平台的整体对比
下面是把上面所有维度合并起来的总表。"有" = 原生支持，"部分" = 有但不完整，"无" = 不支持。
维度 OpenProgram Claude Code OpenClaw Hermes
───────────────────────────────── ─────────── ─────────── ──────── ──────
扁平 DAG (单一 Call 类型 + role 区分) 有 无 无 无
called_by / reads 双套边 有 无 无 无
seq 时间排序（不混入图结构） 有 无 无 无
统一持久化（chat + agent 同 DAG） 有 部分 无 无
@agentic_function 入口 placeholder 有 无 无 无
+ 出口 in-place update（实时观察）
FTS5 节点级全文搜索 有 无 无 无
SessionDB 兼容适配器 有 — — —
分层 system prompt 装配 有 有 有 有
工作区文件 (AGENTS/SOUL/USER.md) 有 有 有 无
Skill 索引 有 有 无 部分
持久化 memory 块 有 有 无 无
真实 context_window 预算 有 有 无 部分
Tools schema 计入预算 有 有 无 部分
Provider usage 反馈 有 部分 无 无
CJK-aware token 估算 有 无 无 无
Hybrid 估算 (信前缀 + 估增量) 有 无 无 无
Tool-result aging（保留结构） 有 有 无 无
wall-clock 时间闸门 有 有 无 无
turn 距离闸门 有 有 无 无
800-tok 大小门槛 有 无 无 无
引用追踪保护 有 无 无 有
保护前 N 条 有 无 无 有
自动 compact（阈值触发） 有 有 无 有
手动 /compact 有 有 无 有
推荐事件（达到 70% 提示） 有 部分 无 无
增量 summary 链 有 有 无 有
LLM 失败 structural 兜底 有 无 无 无
可取消的 summarisation 有 有 无 无
DAG re-parent 保留原始分支 有 无 无 无
压缩后可回放调试 有 无 无 无
ContextEngine ABC 插件化 有 无 有 部分
Per-agent engine override 有 无 有 有
完整生命周期 hooks 有 无 有 部分
组件级注入 有 无 有 无
OpenProgram 独有的设计点
- 扁平 DAG：所有事件都是同一种
Call节点，靠role字段区分（user / llm / code）——一种数据结构表达全部历史，分支/嵌套/合并都靠called_by和reads两套边表达 seq整数承担时间排序，跟"图的边"完全解耦——纯净 DAG 模型- 聊天对话 +
@agentic_function内部调用统一落在同一张 DAG，按 seq 自然交错，没有"主聊天 DAG vs agent 内部 DAG"的切分 @agentic_function入口 append placeholder + 出口 update output 模式——函数运行期间 DAG 上就能看到status="running"节点，方便实时观察 / 调试 / webui 进度展示- DAG re-parent 持久化（compact 后原始分支不丢，可回放）
- LLM summary 失败 structural 兜底（agent loop 永远不因 compaction 崩溃）
- 三闸门 + 保护前 N + 引用追踪同时启用
- CJK 字符比例独立估算
- 组件级注入（不仅是 engine 整体替换，单个组件也能换）
- Hybrid 估算（信 provider 的前缀 + 估本轮增量）
主要借鉴
- Claude Code 的三层 compaction（aging + auto + manual）和 wall-clock 闸门
- Hermes 的引用追踪 + 保护前 N + 增量 summary 链
- OpenClaw 的 ContextEngine ABC + 完整生命周期 hooks