my-pi-agent--session管理

功能设计

架构演进：从单文件到领域自洽子系统 (session/)

在经历 Tau 对齐深度重塑（阶段 18）后，原先散落在外的 session.py 与 session_store.py 已彻底物理下沉，解耦为职责高度内聚的 packages/my-agent-core/src/my_agent_core/session/ 完整自洽子系统：

packages/my-agent-core/src/my_agent_core/session/
├── entries.py   # 【数据实体层】9 种强类型多态 SessionEntry 判别联合体 (Discriminated Union)
├── tree.py      # 【算法层】纯内存 DAG 算法（LCA 公共祖先、带 seen 集合防死锁回溯，零 I/O）
├── memory.py    # 【状态投影层】SessionState 不可变事件溯源纯函数折叠聚合器 (from_entries)
├── storage.py   # 【存储抽象层】纯异步只追加存储协议 (SessionStorage / InMemorySessionStorage)
├── jsonl.py     # 【驱动实现层】行级追加持久化、跨进程文件锁 (.{name}.lock) 与碎片自愈清理
├── session.py   # 【单会话门面】Session 与 SessionTree 树状分支会话高层实现
├── store.py     # 【仓库管理层】SessionStore 会话仓库管理器（工作区天然隔离与短前缀寻址）
└── __init__.py  # 【统一导出层】静态导出全部核心类，外部统一由 my_agent_core.session 一站式导入

模块分工一览

1. entries.py：定义 9 种强类型多态实体，废除脆弱的第 0 行文件头字典，改由首条 SessionInfoEntry 承载会话元数据；
2. tree.py：纯内存算法，零文件依赖。提供带重复 ID 校验的 entries_by_id、带循环引用死锁拦截的 path_to_entry 与 lowest_common_ancestor 计算；
3. memory.py：SessionState 不可变快照，通过纯函数折叠聚合最新状态，自动将 CompactionEntry 折叠为摘要消息；
4. storage.py：定义只追加存储契约 SessionStorage，提供用于高速离线单测的 InMemorySessionStorage；
5. jsonl.py：负责物理文件的高并发安全追加，实现跨平台文件锁（Windows msvcrt / POSIX fcntl）与碎片自愈；
6. session.py：单会话高级门面，提供 Session 与 SessionTree，无缝桥接底层驱动与四层上下文压缩；
7. store.py：多会话仓库管理器，提供 SessionStore 与 SessionMeta，负责工作区隔离、模糊寻址与会话分叉。

数据结构：树 + current 指针

状态 A：current = c3（旧枝末端）

r1 ──► a1 ──► b2 ──► c3      ← current_id = c3，对话位置在旧枝
        │
        └──► b2'             ← 新枝存在，但没人指着它

get_current_path() 沿 c3 的 parent 回溯：c3 → b2 → a1 → r1
= [r1, a1, b2, c3]

状态 B：rewind("b2'") 切到新枝

r1 ──► a1 ──► b2 ──► c3      ← 旧枝留档（c3 还在 entries 里）
        │
        └──► b2'             ← current_id = b2'，对话位置切到新枝

get_current_path() 沿 b2' 的 parent 回溯：b2' → a1 → r1
= [r1, a1, b2']

                        entries        current_id        get_current_path()
状态 A（旧枝末端）      {r1,a1,b2,c3,b2'}   "c3"       [r1, a1, b2, c3]
状态 B（新枝末端）      {r1,a1,b2,c3,b2'}   "b2'"      [r1, a1, b2']
                         ↑ 完全一样          ↑ 唯一变化     ↑ 跟着指针走

SessionTree 对象
├── entries: dict                                  # 5 个节点平铺，树关系在各自 parent_id 里
│   ├── "r1" → SessionEntry(id="r1", parent_id=None,role="system",    content="sys")
│   ├── "a1" → SessionEntry(id="a1", parent_id="r1",role="user",      content="q1")
│   ├── "b2" → SessionEntry(id="b2", parent_id="a1",role="assistant",content="a1答案")
│   ├── "c3" → SessionEntry(id="c3", parent_id="b2", role="user",    content="q2")
│   └── "b2'"→ SessionEntry(id="b2'", parent_id="a1",role="assistant",content="换个答法")
├── current_id: ???                                  # ← 可变，下面两个状态看它
└── root_id: "r1"

为什么记录parent_id

它支撑的核心操作：沿祖先链回溯

整个设计围绕一个需求转：“取当前对话的完整上下文”。上下文在树里的定义就是”从根一路走到 current 的路径”——而路径就是沿 parent_id 连续向上跳：

def path_to_entry(entries: Sequence[SessionEntry], leaf_id: str) -> list[SessionEntry]:
    """纯内存纯函数：从根到指定叶节点的祖先回溯（带环路检测与 O(N) 优化）。"""
    by_id = entries_by_id(entries)  # 严格校验重复 ID
    path: list[SessionEntry] = []
    seen: set[str] = set()
    current_id: str | None = leaf_id
    
    while current_id is not None:
        if current_id in seen:
            raise SessionTreeError(f"Cycle detected at session entry: {current_id}")
        seen.add(current_id)
        entry = by_id.get(current_id)
        if entry is None:
            raise SessionTreeError(f"Missing parent entry: {current_id}")
        path.append(entry)
        current_id = entry.parent_id
        
    path.reverse()  # O(N) 翻转，避免频繁 insert(0, ...) 导致的 O(N^2) 内存拷贝
    return path

b2’ 的视角：只看得上自己的 parent b2’.parent_id = “a1” ──► a1.parent_id = “r1” ──► r1.parent_id = None 停 ⇒ 回溯结果 [r1, a1, b2’] ← 这就是”上下文”

parent_id 就是这条回溯链路的物理实现——每个节点只认识”我是谁的孩子”，整条链靠反复跳父节点拼出来。

为什么不存children（双向）

存 parent_id（现在）            存 parent + children（备选）
  b2.parent_id = "a1"             b2.parent_id = "a1"
  a1.children = {...}  ← 需要额外维护
  add_entry 时：                 add_entry 时：
    只写新节点.parent_id            ① 写新节点.parent_id
    一步                         ② 还要往父节点.children 里追加
                                 两处写，必须同步（一致性 bug 高发区）

只存 parent_id 的另一个好处：add 永远只写一个地方（新节点的 parent_id），不会有”父的 children 和子的 parent 对不上”的隐患。

代价：向下看不见

单向链的代价是——从 current 不知道下面有哪些节点（不知道谁是我的孩子）

r1 ──► a1 ──► b2 ──► c3
        │
        └──► b2'

• 站在 a1 上，按 parent_id 只能看到 r1（上），看不见 b2、b2’（下）
• 但”下”这件事我们不需要：切到 b2’ 靠 rewind(“b2’”) 直接按 id 跳（O(1)，不走路），不需要”从 a1 走过去”
• 真要看全树（比如列所有枝），直接遍历 entries dict 就行——entries 本身就是全量存储

想"从 a1 走到 b2'"？   → 不需要。rewind("b2'") 直接跳
想知道 b2' 存在吗？    → 查 entries["b2'"] 就行
想枚举 a1 的所有孩子？ → 遍历 entries 过滤 parent_id=="a1"（pig-mono 的
                        get_children 就是这么干的，一次 O(n) 扫描）

如何切换branch

def rewind(self, entry_id: str) -> None:
    if entry_id not in self.entries:
        raise ValueError(f"Entry {entry_id} not found")
    self.current_id = entry_id

切换 branch 的全部动作就是上面这 3 行——校验存在 + 给 current_id 重新赋值。没有移动节点、没有删除、没有重建路径。路径是”现算”的：切完指针，get_current_path() 沿新指针对应的 parent 链回溯，自动得到新枝的上下文。

压缩护栏（context 阶段加的）：Session.rewind 在 compaction_floor（压缩时刻 current id）非 None 时，只允许回 floor（含）之后长出的节点，压缩点及之前 → ValueError——换来缓存永不失效、无尾部真空（详见 context 笔记）。

① 切到新枝 b2'

session.rewind("b2'")          # current_id: "c3" → "b2'"

   r1 ──► a1 ──► b2 ──► c3     ← 旧枝留档
           │
           └──► b2'            ← current 切到这

此时 session.get_current_path_messages() 返回 [r1, a1, b2']——Agent 恢复上下文就是这条。

② 从新枝继续对话

agent.run("那再解释下")

# run() 开头：messages = session.get_current_path_messages()  → [r1, a1, b2']
# 循环内：add_message("user", "那再解释下")
#         → add_entry 的 parent_id 缺省 = current_id = "b2'"
#         → 新节点 e5 挂在 b2' 下，current_id = "e5"

   r1 ──► a1 ──► b2 ──► c3     ← 旧枝完全没动
           │
           └──► b2' ──► e5     ← 新枝继续长（current=e5）

③ 切回旧枝（再切换）

session.rewind("c3")           # current_id: "e5" → "c3"

   r1 ──► a1 ──► b2 ──► c3     ← current 又回到旧枝末端
           │
           └──► b2' ──► e5     ← 新枝还在，随时可再切

无限次往返切换，任何一次都不丢节点——entries 里 6 个节点始终齐全。

核心流程

消息落盘与持久化范式（只追加 Append-Only 模型）

在现代 Tau / Pi 架构中，核心持久化范式确立了一条铁律：“历史发生即不可变，彻底废除全量重写”。

Agent 运行产生新事件
       │
       ▼
Session.add_message / rewind / model_change
       │
       ├─► 内存：生成强类型多态 SessionEntry (MessageEntry, LeafEntry...)
       │
       ▼
SessionStorage.append(entry) (基于 jsonl.py 追加写入)
       │
       ├─► 1. 获取跨进程文件锁 (Windows msvcrt / POSIX fcntl)
       ├─► 2. 自动清理崩溃残留碎片 (_remove_incomplete_temp)
       ├─► 3. 以 "a" 模式追加单行 JSONL 到文件末尾
       └─► 4. 释放文件锁

为什么必须采用“只追加模型”？

1. 消除 O(N) 到 O(N²) 的写放大：早期每加一条消息就做一次临时文件全量写盘 + os.replace，当会话拥有上百条历史时，单次交互的 I/O 耗时极其严重。只追加写入单行耗时恒定为 O(1)；
2. 分支切换零开销：在只追加模型中，分支回溯（rewind）不需要重写文件，只需追加一条极其轻量的 LeafEntry(leaf_id=target_id) 记录指针跳转；
3. 支持断电与部分行撕裂自愈：JsonlSessionStorage 能宽容容忍文件末尾最后一行在断电时残留的半截撕裂数据，读取时自动忽略未闭合的末尾行，同时严格拒绝中间行损坏。

恢复（load）

Session.load(path)
   │
   ├─ 读所有行
   ├─ 第一行 = header ──校验──► 缺 id/created_at → ValueError（必要字段门禁，
   │                           type/version 已删：防误读靠目录隔离 + 必要字段）
   │
   ├─ 最后一行 JSON 损坏？  ──是──► 丢弃（尾行撕裂宽容：原子写下很少见，兜底）
   │        │否
   │        ▼
   ├─ SessionTree.from_jsonl_iter(其余行)   ← 重建 entries + root_id
   │
   ├─ header 里的 current_id / root_id 有效？ ──► 覆盖树里的指针（文件为准）
   │
   └─ 返回 Session（树 + current 指针 = 恢复时的对话位置 + compaction_floor 恢复）
   
磁盘文件（正式 .jsonl）                         内存
┌──────────────────────────────┐
│ {"id":...,"cwd":...,          │  ──②──► header dict（校验必要字段 id/created_at；
│  "current_id":"e5","root_id": │           type/version 已删）
│  "r1","compaction_floor":"e5"}│  ──⑦──► tree.current_id = "e5"
├──────────────────────────────┤          tree.root_id = "r1"
│ {"id":"r1","parent_id":null,  │          session.compaction_floor = "e5"
│  "role":"system",...}         │  ─┐
│ {"id":"a1","parent_id":"r1",..│   │──⑥──► SessionTree.from_jsonl_iter
│ {"id":"b2",...}               │   │        逐行 model_validate_json
│ ...                           │  ─┘        填 entries + 推断 root_id
└──────────────────────────────┘

关键设计：from_jsonl_iter 只填 entries/root，不推断 current——current 只信 header。pig-mono 用”叶子时间戳最新”启发式推断（兼容没有 header 的旧文件）；我的格式从第一版就有 header，不需要猜。

Agent 集成（持久化循环）

Agent(llm=llm, session=session, system_prompt="...")   ← session 必填（无内存模式）
   │
   ├─ __init__: 拼 system（system_prompt + skill 清单 + subagent 清单）
   │            + session.get_full_history_messages()（纯对话）
   │            → messages = [system] + 对话
   ▼
run("37*19=?")
   ├─ 同步 messages = [system 首条] + session.get_current_path_messages()   ← rewind 后同 Agent 续跑保留 system
   ├─ messages.append(user)        ──► session.add_message("user", ...)   ★落盘
   ├─ 循环:
   │    llm.chat(messages) → assistant
   │    messages.append(assistant) ──► session.add_message("assistant", ...) ★落盘
   │    if tool_calls:
   │       for tc: 执行工具 → tool 消息
   │       messages.append(tool)   ──► session.add_message("tool", ...)     ★落盘
   │    else: return  （最终回答）
   ▼
   文件 = header + user + assistant(tool_calls) + tool + assistant（纯对话，不含 system）
   （每步一条，current 指针一路指向最新）

Agent 与 Session 的接缝：Agent 的 self.messages（内存，发给 LLM）和 session 树（持久化）双写同步——add_message 在 append 后立即调。get_current_path_messages() 就是把树路径翻译回 list[Message]（metadata 承载 tool_calls/tool_call_id）的桥；get_full_history_messages() 是它的变体——沿路径过滤 type=“compaction” 缓存 entry，返回纯历史（Agent 恢复上下文用这个，避免缓存 entry 混进 transcript）。

2026-08-16 关键重构：system 从「session 的持久化内容」变成「Agent 的运行时配置」。原设计把 system_prompt 作为树的根 entry 存进文件（Session(path, system_prompt=...)），这带来一个问题——恢复会话时 system 是「建 session 时存的那个」，改了 skill/subagent 配置也不会变。重构后：Session 只存纯对话（去 system_prompt，树从空开始），system 由 Agent 每次构造时拼（system_prompt + skill/subagent 清单），再和恢复的对话合成 [system] + 对话。对齐 anthropic-sdk-python 的「system = agent 定义、session = 运行历史」拆分。

get_current_path 就是”把我当前对话的完整上下文取出来”这个动作的实现——每当”从会话恢复/继续对话”需要把树变成 LLM 能用的消息序列时，它就是必经之路。

rewind 后的续跑

session.rewind(a1.id)          ← 只改 current 指针 + save（header 的 current_id 更新）
   │
Agent.run("换个问法")
   ├─ 同步 messages = 当前路径 [r1, a1]      ← 新枝从 a1 长，旧枝 [b2,c3] 不动
   ├─ append user → add_message → parent = a1
   ▼
   树:  r1 → a1 → b2 → c3      （旧枝留档）
        r1 → a1 → b2'(新)       （新枝）

SessionEntry与SessionTree

SessionEntry

class BaseSessionEntry(BaseModel):
    """会话树所有实体的公共基类。"""
    model_config = ConfigDict(
        extra="forbid",            # 严禁未识别的脏字段注入
        populate_by_name=True,     # 支持驼峰 parentId / 蛇形 parent_id 互通
        alias_generator=to_camel,
    )
    id: str = Field(default_factory=lambda: uuid4().hex)
    parent_id: str | None = None
    timestamp: float = Field(default_factory=time.time)

# 9 种具体的多态判别条目：
class SessionInfoEntry(BaseSessionEntry):
    """会话元数据（替代旧版脆弱的第0行 Header 字典，成为流首项）。"""
    type: Literal["session_info", "sessionInfo"] = "session_info"
    title: str | None = None
    cwd: str | None = None

class MessageEntry(BaseSessionEntry):
    """核心对话消息载体（嵌套包装 Message，彻底解耦树节点与模型字段）。"""
    type: Literal["message"] = "message"
    message: Message

class ModelChangeEntry(BaseSessionEntry):
    """运行时切换大模型记录。"""
    type: Literal["model_change", "modelChange"] = "model_change"
    model: str
    provider: str | None = None

class ThinkingLevelChangeEntry(BaseSessionEntry):
    """动态调整推理思考深度（如 low, medium, high）。"""
    type: Literal["thinking_level_change", "thinkingLevelChange"] = "thinking_level_change"
    thinking_level: str

class CompactionEntry(BaseSessionEntry):
    """上下文压缩折叠记录（包含替换的条目 ID 列表与摘要文本）。"""
    type: Literal["compaction"] = "compaction"
    summary: str
    replaces_entry_ids: list[str] = Field(default_factory=list)

class BranchSummaryEntry(BaseSessionEntry):
    """分支探索总结（跨分支合并时使用）。"""
    type: Literal["branch_summary", "branchSummary"] = "branch_summary"
    summary: str

class LabelEntry(BaseSessionEntry):
    """用户检查点书签（如 v1.0, before-refactor）。"""
    type: Literal["label"] = "label"
    label: str

class LeafEntry(BaseSessionEntry):
    """当前活跃叶节点指针（分支回溯时仅追加一条即可，零文件重写）。"""
    type: Literal["leaf"] = "leaf"
    leaf_id: str

class CustomEntry(BaseSessionEntry):
    """扩展与遥测专用槽位（带 namespace 隔离）。"""
    type: Literal["custom"] = "custom"
    namespace: str
    data: dict[str, Any] = Field(default_factory=dict)

# 通过 Pydantic v2 discriminator="type" 组装强类型联合体：
SessionEntry = Annotated[
    SessionInfoEntry | MessageEntry | ModelChangeEntry | ThinkingLevelChangeEntry
    | CompactionEntry | BranchSummaryEntry | LabelEntry | LeafEntry | CustomEntry,
    Field(discriminator="type"),
]

9 种多态条目各司其职： 1. SessionInfoEntry：记录会话元数据（工作目录、创建时间、标题），天然作为文件的第一条记录，彻底消灭特殊的 Header； 2. MessageEntry：包裹真正的 Message（User、Assistant、Tool）； 3. ModelChangeEntry：记录模型切换事件； 4. ThinkingLevelChangeEntry：记录推理思考级别调整； 5. CompactionEntry：记录压缩摘要及它覆盖的历史 ID 列表； 6. BranchSummaryEntry：记录分支探索总结； 7. LabelEntry：用户设置的书签/检查点； 8. LeafEntry：活动叶子节点指针（分支切换全靠它）； 9. CustomEntry：第三方扩展的隔离插槽。

纯内存状态投影：SessionState 事件溯源折叠 (memory.py)

在多态只追加体系下，Agent 如何获取当前所指分支的最新完整上下文？ 答案就是：纯函数不可变折叠投影（Event Sourcing Fold）：

@dataclass(frozen=True, slots=True)
class SessionState:
    """不可变状态快照。"""
    messages: tuple[Message, ...] = ()
    model: str | None = None
    provider: str | None = None
    thinking_level: str | None = None
    label: str | None = None
    active_leaf_id: str | None = None

    @classmethod
    def from_entries(cls, entries: Sequence[SessionEntry], leaf_id: str | None = None) -> SessionState:
        """沿根节点到 leaf_id 的路径条目，线性折叠聚合出当前运行态！"""

• 遇 MessageEntry → 追加对话消息；
• 遇 CompactionEntry → 自动将 replaces_entry_ids 内的历史消息折叠为一条前置摘要消息；
• 遇 ModelChangeEntry → 刷新生效模型；
• 全流程无锁、纯内存计算，线程绝对安全！

为什么需要 SessionState？它扮演什么角色？

在传统的 Agent 设计中，“获取会话状态”非常简单，因为状态就是一个写死的变量（例如 self.model = "gpt-4o"，self.messages = [...]）。

但在只追加（Append-Only）事件溯源架构下，情况完全变了： - 物理磁盘里根本没有一个叫做“当前状态”的写死字段； - 硬盘里只有一条条按时间顺序追加的历史流水账：一会儿记了一条用户提问，一会儿记了中途切换模型，一会儿记了一次压缩，一会儿记了指针回退； - 每次 Agent 准备调用大模型前，必须有一个人拿着算盘，把这串流水账从头到尾拨一遍，算出一个最新的运行时状态。

SessionState 就是这个被计算出来的“不可变状态快照包”，它封装了 Agent 开展下一轮工作所需的 4 大核心数据：

1. messages: tuple[Message, ...]（当前纯净消息链）：
   • 沿活动分支从根到当前叶节点的有效消息；
   • 已经被自动处理好了：被压缩覆盖的旧消息已经被安全抹除，就地替换成了精炼的摘要 User 消息，零多余 Token；
2. model: str | None & provider: str | None（当前生效的大模型与厂商）：
   • 如果用户在第 10 轮对话中途调用 /model deepseek-chat，折叠器在扫描到 ModelChangeEntry 时会准确更新此字段，使 Agent 立即感知并切换客户端；
3. thinking_level: str | None（当前推理思考深度）：
   • 如 low, medium, high，反映当前分支最新的推理深度设定；
4. active_leaf_id: str | None（当前停留在哪个叶节点）：
   • 指明当前对话停留在树的哪一个分支节点，指示下一条新产生的消息应该把 parent_id 指向谁。

不可变（frozen=True）带来的工程红利

SessionState 被严格声明为 @dataclass(frozen=True, slots=True)： - 无锁并发安全：状态一旦通过纯函数计算得出，任何人都不准篡改它。多线程或多协程并发读取该快照时，零锁竞争、零副作用； - 支持“历史时光机（Time Travel）”漫游： 想看 5 步之前会话长什么样？只需调用 SessionState.from_entries(entries, leaf_id="msg_005")，算法会以第 5 步为终点进行回放折叠，瞬间穿越回当年的状态快照，极其适合用于多分支对比、探索回滚与可视化渲染！

核心魔法：四段压缩管线与 CompactionEntry 的协同机理

在框架中，我们在 context.py 中实现了四层廉价优先压缩管线（L3 磁盘溢出 ➔ L1 中间截断 ➔ L2 工具微紧凑 ➔ L4 大模型滚动摘要）：

• L3 / L1 / L2（纯内存临时修剪）： 前三层属于“免费/局部压缩”。它们完全不改变会话持久化文件，只在调用大模型前夕把长文本在内存里做非破坏性的视图修剪；
• L4（LLM 宏观滚动摘要）： 当免费手段用尽、Token 依然超标时，才真正花钱调大模型总结出一段 <summary>。此时系统就会生成一条强类型的 CompactionEntry 追加落盘！

# session/memory.py 内部的核心替换算法：
def _apply_compaction(items: list[tuple[str, Message]], entry: CompactionEntry) -> list[tuple[str, Message]]:
    """将 replaces_entry_ids 范围内的历史消息，就地折叠为一条 UserMessage 摘要。"""
    summary_msg = Message(role="user", content=f"Previous conversation summary:\n{entry.summary}")
    replaces_set = set(entry.replaces_entry_ids)

    new_items = []
    inserted = False
    for eid, msg in items:
        if eid in replaces_set:
            if not inserted:
                new_items.append((entry.id, summary_msg))  # 就地插入摘要
                inserted = True
            # 其余被替换的旧消息直接跳过！
        else:
            new_items.append((eid, msg))                   # 未被替换的消息完整保留
    return new_items

这一设计的极致之处在于： 1. 物理磁盘层面：所有被替换的旧消息一条都没删，完好无损（拥有绝对不可篡改的审计与复盘能力），写盘只需要在末尾追加这一条 CompactionEntry（耗时 0.1ms）； 2. 模型认知层面：通过 _apply_compaction 的纯函数折叠，大模型看到的视图已经被就地替换为了精炼摘要，不浪费一个多余 Token！

SessionTree

class SessionTree:
    # 状态
    entries: dict[str, SessionEntry]   # 全量节点
    current_id: str | None             # 当前指针
    root_id: str | None                # 根锚点
    # 方法
    add_entry(role, content, parent_id=None, **metadata) -> SessionEntry  # 增
    get_current_path() -> list[SessionEntry]                              # 查当前路径
    get_path_to_entry(entry_id) -> list[SessionEntry]                     # 查任意路径
    rewind(entry_id) -> None                                              # 切指针
    to_jsonl() -> str                                                     # 序列化
    from_jsonl_iter(lines) -> SessionTree                                 # 反序列化

会话流程中要做的事	用哪个方法	谁在调
新消息进来	add_entry	Session.add_message
回退 / 切枝	rewind	Session.rewind
取当前上下文发 LLM	get_current_path	Session.get_current_path_messages
fork 复制路径	get_path_to_entry	SessionStore.fork
落盘	to_jsonl（或直接遍历 entries）	Session.save
恢复	from_jsonl_iter	Session.load
重置	（无专用方法——Session.reset 重建 SessionTree()）	Session.reset

样例

加第二条：user 提问

e2 = tree.add_entry("user", "37*19=?")

SessionEntry：

SessionEntry(
    id="7b04e8aa",
    parent_id="f3a9c2d1",    # ← 缺省 = current_id = 上一条的 id
    timestamp="2026-08-10T15:30:13.001234",
    role="user",
    content="37*19=?",
    metadata={},
)

树（现在两个节点连起来了）：

entries = {
    "f3a9c2d1": SessionEntry(id="f3a9c2d1", parent_id=None,    role="system",   content="You are a helpful assistant."),
    "7b04e8aa": SessionEntry(id="7b04e8aa", parent_id="f3a9c2d1", role="user",  content="37*19=?"),
}
current_id = "7b04e8aa"
root_id    = "f3a9c2d1"

Session 类——「树 + 文件」的会话本体

💡 持久化演进注记（全量重写 ➔ 只追加 Append-Only）： - 早期设计：Session.save() 采用 NamedTemporaryFile + fsync + os.replace 原子全量重写整棵树（保证崩溃安全，但存在 O(N²) 写放大）； - Tau 对齐新架构：高层保留 Session 与 SessionTree 门面，底层存储全面升级为 JsonlSessionStorage 只追加模型。常规对话消息直接以追加模式写入单行，写盘耗时恒定为 O(1)；save() 保留作为兼容与快照初始化门面。

class Session:
    # 状态
    path: Path                     # 会话文件路径（一个会话 = 一个文件）
    id: str                        # 会话身份（时间戳-hex，如 "20260810-153012-a1b2c3d4"）
    created_at: str                # 创建时间（ISO）
    cwd: str                       # 工作目录（= workspace，记录在 header）
    tree: SessionTree              # 树（SessionTree 的全部能力在这里）
    compaction_floor: str | None   # 压缩时刻 current id（header 持久化，rewind 护栏）

    # 方法
    __init__(*, path, cwd=None, metadata=None)               # 新建（空树纯对话，不写盘）
    add_message(role, content, parent_id=None, **metadata) -> SessionEntry # 增 + 落盘
    rewind(entry_id) -> None                                 # 切指针 + 落盘（护栏）
    get_current_path_messages() -> list[Message]           # 查：路径 → Message 列表（含缓存）
    get_full_history_messages() -> list[Message]           # 查：过滤 type=compaction → 纯历史
    add_summary_cache(...) -> None                          # 增：写缓存 entry（context 阶段加，不动 current）
    get_latest_compaction_cache() -> dict | None            # 查：找最新缓存 entry（context 阶段加）
    save() -> None                                                # 持久化：原子全量重写
    load(path) -> Session        # classmethod                   # 持久化：从文件恢复
    reset() -> None                                               # 清空（纯对话 + 清 floor）

类别	方法	内部动作	谁在用
增	add_message	树 add + save	Agent.run
增	add_summary_cache	插缓存 entry + floor	ContextSessionBridge.write_compaction
切	rewind	树 rewind + save	宿主（切枝后续跑，护栏限制压缩点前）
查	get_current_path_messages	树路径 → Message	宿主（含缓存 entry 的原始路径）
查	get_full_history_messages	树路径 → Message（过滤）	Agent 构造/续跑（纯历史）
查	get_latest_compaction_cache	找最深 compaction	ContextSessionBridge.restore_cache
持久化	save / load	树 ↔︎ JSONL 文件	add_message 内部 / SessionStore.open
破坏	reset	清树 + 重写 + 清 floor	Agent.reset
构造	__init__	空树（纯对话，不含 system）	SessionStore.create

Session 与 SessionTree 的分工（一句话）

SessionTree = 纯树（组织 + 回溯 + 序列化），不知道文件、不知道 Message
Session     = 树 + 文件 + Message 翻译 + 缓存 entry（持久化外壳）
   ├─ 转发：add_message / rewind  → 树操作 + save()
   ├─ 独有：get_current_path_messages / get_full_history_messages（树路径 → Message 桥）
   ├─ 独有：add_summary_cache / get_latest_compaction_cache（context 缓存读写）
   └─ 独有：save / load / reset（文件全生命周期）

为什么要用树

树 = 允许「对话历史分叉」的数据结构。 序列（列表）只能一条道走到底，树可以在任意点分裂出多条分支。

先看序列的局限：rewind 的困境

假设一个对话历史（列表）： [user, assistant(tool_calls), tool(703), assistant(“703”), user, assistant(“坏了，重来”)]

现在你想 rewind（回退）到「user, assistant(“坏了，重来”)」之前，改成另一种问法：

列表： [… user, assistant(“坏了”)] 回退后：[… user] ← 新的问法从这开始

问题：回退 = 删掉「assistant(“坏了”)」这一条吗？ - 删掉 → 旧的那条分支就永久丢了（你想保留「试过错的路」对比，没了） - 不删 → 列表尾部挂着一条「当前对话已经不在那了」的僵尸消息

树解决这个：不删，而是分叉——

树：

root → user → assistant(703) → assistant("703") → user
                                                    ├─ assistant("坏了，重来")   ← 旧分支（保留）
                                                    └─ assistant("换个问法")     ← 新分支（当前）

回退 = 移动 current_id 指针到 user，然后从那里长出新分支。旧分支完整保留，随时能 branch_to 切回来看。

这就是树的第一个好处：rewind/分支是天然支持的——不需要删除数据，只是移动指针 + 长新枝。

树的其他好处

每个消息能回溯「上下文链」

树里每个 entry 带 parent_id，get_path_to_entry(id) 能回溯出从根到它的完整路径：

def get_path_to_entry(self, entry_id):
    """从根到该 entry 的完整路径"""
    while current.parent_id:
        path.insert(0, current)
        current = entries[current.parent_id]

这有什么用：你知道「当前会话」是哪条路径（get_current_conversation() = 从根到 current 的 path）。比如分支后，你只关心当前这条路径，而树能干净地给出它，不需要在列表里过滤僵尸消息。

支持「在任意点开新对话」

树让「从历史某点分叉开始新对话」变得自然（fork(entry_id)）——复制根到某点的路径成一个新 session。列表做不到（你得手动截断）。

sessionmanager

SessionStore = 会话文件的仓库管理员：它管理”一批会话文件”的生命周期（造/列/找/删/分叉），把”文件系统”的细节藏起来，让外面的人（宿主/Agent）只需要说”给我开个新会话 / 打开某某会话”，不用关心文件在哪、叫什么名。

它解决什么问题

Session（单会话）知道的是”我这个会话怎么写盘、怎么恢复”——但“磁盘上有哪些会话、我要用哪一个”它不知道。这正是 SessionStore 补的：

问题	SessionStore 的答案
怎么开一个新会话？	create() → 生成 id + 建文件 + 返回可用的 Session
有哪些会话？	list() → 扫目录，返回元信息（id/时间/条数）
我要接上之前的对话？	open("前缀") → 找到文件 → 恢复成 Session
这个会话不要了？	delete(id) → 删文件
想从某点开个新会话？	fork(id, entry_id) → 复制路径成新会话

class SessionMeta(BaseModel):
    # 状态（list() 的返回类型）
    id: str
    path: Path
    created_at: str
    entries: int

class SessionStore:
    # 状态
    workspace: Path        # 绑定的项目目录（默认 Path.cwd()）
    root: Path             # 会话目录 = workspace/root（绝对 root 直接用）

    # 方法
    __init__(root=".my_agent_core/sessions", workspace=None)    # 绑定项目
    create() -> Session                                          # 新会话（id 生成 + 落盘）
    list() -> list[SessionMeta]                                 # 列会话（新→旧）
    open(id_or_prefix) -> Session                               # 打开（恢复整棵树）
    delete(id_or_prefix) -> None                                # 删除文件
    fork(id_or_prefix, entry_id) -> Session                     # 分叉新会话
    _resolve(id_or_prefix) -> Path        # 私有：id/前缀 → 文件路径

SessionStore 在整个设计中的位置

SessionEntry  → 一条消息（数据）
SessionTree   → 消息组织成树（容器 + 算法）
Session       → 树 + 文件（单个会话的持久化外壳）
SessionStore  → 管理"一堆会话文件"（仓库）
                 └─ 它不管单条消息、不管树、只跟"会话文件"打交道

流程

store.create()
   │
   ├─ 造 id（时间戳-hex）→ 路径 <root>/<id>.jsonl
   ├─ new Session（cwd=workspace）
   ├─ session.id = sid → session.save()
   │     └─ save() 内部：Session 把自己的树写进文件（header + entry 行）
   └─ 返回可用的 Session

store.open("20260810-153")  # 支持 open() 或 open_session()
   ├─ _resolve：扫 *.jsonl，header.id.startswith("20260810-153") 唯一命中
   ├─ Session.load(文件)
   │     └─ 读 header → 重建树 → 恢复 current/root
   └─ 返回恢复的 Session（Agent 可以直接接上续聊）

store.list()
   ├─ 只读每个文件的 header
   └─ 返回 [SessionMeta(id, created_at, entries)] 按时间倒序

如何管理session

SessionStore 管理会话 (位于 packages/my-agent-core/src/my_agent_core/session/store.py)
├── ① 操作面（主动做）——会话的生命周期
│     create / create_session  造会话（id 生成 + 文件）
│     open / open_session      开会话（全 id / 前缀命中）
│     list                     列全部（元信息列表，按时间倒序）
│     delete                   删会话（物理删除对应 .jsonl）
│     fork                     分叉（从某 entry 复制路径造新会话）
└── ② 数据面（管文件）——workspace 目录隔离
      根目录：<workspace>/.my_agent_core/sessions/
      文件：一个会话一个 <id>.jsonl，各项目天然隔离

统一门面导出约定 (session/init.py)

在重构后，外部所有消费者（如 Agent、CLI 或测试）无需关心内部是 store.py 还是 session.py，统一从 my_agent_core.session 导入即可：

from my_agent_core.session import Session, SessionStore

# 一行代码创建基于当前 workspace 隔离的会话仓库
store = SessionStore()
session = store.create()

目录隔离

防什么：会话目录里混进的”非会话文件”（比如用户手放的一个 notes.jsonl、将来别的格式文件）被当成会话处理。

怎么防：只扫”专用目录”里符合 .jsonl 的文件，这个目录之外的东西根本不进名单。

# session/store.py（list 和 _resolve 都用）
for f in self.root.glob("*.jsonl"):

效果：

.my_agent_core/sessions/
├── 20260810-153012-a1b2c3d4.jsonl   ← 被扫到 ✓
├── notes.txt                        ← glob("*.jsonl") 不匹配，进不了名单 ✓
└── 随便.txt

读取容错

防什么：某个会话文件损坏（JSON 坏了 / header 缺字段）时，list() 整个崩掉——一个坏文件让所有会话都列不出来。

怎么防：读每个文件的 header 包 try/except，坏文件跳过、继续下一个。

# 我：except (JSONDecodeError, KeyError)（精确版）
try:
    with open(f) as fh:
        header = json.loads(fh.readline())
except (json.JSONDecodeError, KeyError):
    continue                              # 坏文件直接不列

效果：一个坏文件 → 它自己被跳过，其余会话照常列出。

workspace 过滤（跨项目隔离）

我们的目录布局已经把隔离做在了目录层

D:/code/python/my-pi-agent/.my_agent_core/sessions/    ← 项目 A 的会话
D:/blog/.my_agent_core/sessions/                        ← 项目 B 的会话

子代理独立持久化

为什么子代理也要有 session

子代理（task 委派）的语义是「fresh context + 用完返回摘要」——但「用完」不等于「不落盘」。对齐 Claude Code 的做法：每个子代理独立持久化成一个 session 文件，留着不删，这样：

1. 事后能翻子代理的完整对话（父只收到最终摘要，中间过程丢了就翻不到）
2. 将来能 resume（用文件反查继续）

落点：父 session 目录下的 subagents/

sessions/
├── <父sessionId>.jsonl               ← 父会话
└── subagents/
    └── agent-<task_id>.jsonl          ← 子代理独立会话（task_00000001 等）

子代理的元数据塞进 header（复用 Session 的 header 机制，不单独建 meta 文件）：

child_session = Session(
    path=parent.session.path.parent / "subagents" / f"agent-{task_id}.jsonl",
    metadata={
        "agent_type": subagent_type,          # 哪个子代理（Subagent.name）
        "spawn_depth": parent._spawn_depth + 1,  # 嵌套深度
        "parent_session_id": parent.session.id,  # 关联回父会话
    },
)

对照 Claude Code 的 agent-<id>.jsonl + .meta.json：它单独存 meta 文件，我们把 meta 塞进 Session 自己的 header（Session 新增 metadata 参数，save 写进 header、load 读回），少一个文件、复用现有 header 机制。

和父 session 的关键区别

	父 session	子代理 session
谁建	宿主（SessionStore.create）	TaskManager._run（spawn 时）
存什么	完整对话 + system 由 Agent 拼	子代理对话（system 由子 Agent 拼）
生命周期	持久，可 rewind/fork/resume	持久（留着不删），供追溯/将来 resume
污染边界	—	绝不写进父的树（fresh context 的关键）

子代理的 session 是「独立 + 临时（用完可弃但先留着）」的——它和父 session 的「持久化 + rewind + fork」是两码事。核心就一条：子代理的对话历史不能混进父的树，所以必须独立文件。