阶段 3：历史容器与 Prompt 组装

对应代码：stage03_history.py

学习目标

建立"模型看到的是历史投影，不是事件流"的心智，并理解快照语义为什么是重试正确性的基础。

源码锚点

概念	位置	说明
历史投影	turn.rs 每轮的 clone_history().for_prompt(...)	分析文档 7.4 节
Prompt 字段表	分析文档 7.13 节	input / tools / base_instructions / output_schema
ContextManager	codex-rs/core/src/context/	token 记账、压缩窗口的家（教学版只留 list）

代码走读

• History.snapshot()：浅拷贝列表。这是整个重试机制的支点——第一次物理尝试构造 Prompt 后，流断开前已完成的 item 会追加进历史；重试时重新 snapshot()，新 Prompt 自动继承这些事实（7.18 节"可重试流错误如何保持因果连续"）。
• build_prompt(tools)：对应 build_prompt(input, tools, base_instructions, output_schema)。真实版还会按模型 input_modalities 过滤历史（丢掉不支持的模态），教学版省略。

演示的输出就是快照语义的直接证明：

prompt.input 长度 = 1     <- 快照之后追加的 item 不影响已构造的 Prompt
history 长度     = 2     <- 权威副本已更新
下一轮 prompt 长度 = 2     <- 新事实此时才对模型可见

运行与预期输出

python stage03_history.py

练习

1. 给 History 加 truncate_to(ordinal)：从某序号截断——这是 rollback 和 fork 的基础（分析文档 12.8 节 ForkBoundary）。
2. 加 token_estimate()（例如按字符数 / 4 粗估），并在超过阈值时打警告——对应 auto_compact_window 的记账（3.19 节）。
3. 思考题：为什么 ContextManager 的历史不直接用事件队列？（答：事件是展示与控制流，含大量 delta/审批请求噪声；模型输入需要的是规范化的 ResponseItem 序列——11.1 节"三张不同的表"。）

与真实实现的差距

• 模态过滤、token 记账、压缩窗口（SessionState.auto_compact_window）整块缺失——迷你版历史无上限，真实系统靠压缩防止膨胀（7.8 节）。
• 真实的 ContextManager 还维护 reference context、contextual fragment 注入等（9.12 节的过滤规则就是为它们服务的）。

代码

"""积木 3：历史容器与 Prompt 组装。

真实对应物：
  - codex-rs/core/src/context/（ContextManager：token 记账、压缩窗口、历史权威副本）
  - turn.rs 每轮的 clone_history().for_prompt(...)（分析文档 7.4 节）
  - build_prompt 的字段表（分析文档 7.13 节）

核心认知：模型每轮看到的是"历史的投影"，不是事件流。
快照语义保证：采样期间历史被追加，不影响本次 Prompt —— 这正是重试能继承已提交事实的原因。

运行：python stage03_history.py
"""

from __future__ import annotations

from dataclasses import dataclass

from stage01_model import Item, ToolSpec


@dataclass
class Prompt:
    """对应 7.13 节的 Prompt 结构。"""

    input: list[Item]
    tools: list[ToolSpec]
    base_instructions: str | None = None


class History:
    """对应 ContextManager 的最小子集：模型历史的唯一权威副本。"""

    def __init__(self, base_instructions: str | None = None):
        self._items: list[Item] = []
        self.base_instructions = base_instructions

    def record(self, *items: Item) -> None:
        self._items.extend(items)

    def snapshot(self) -> list[Item]:
        """对应 clone_history()：浅拷贝列表。

        追加发生在快照之后，不会"渗"进已构造的 Prompt；
        下一次采样重新 snapshot，才能看到新事实。
        """
        return list(self._items)

    def build_prompt(self, tools: list[ToolSpec]) -> Prompt:
        """对应 build_prompt(input, tools, base_instructions, output_schema)。"""
        return Prompt(
            input=self.snapshot(),
            tools=tools,
            base_instructions=self.base_instructions,
        )

    def __len__(self) -> int:
        return len(self._items)


async def demo() -> None:
    history = History(base_instructions="你是一个终端助手")
    history.record(Item(type="message", role="user", text="跑一下 echo"))

    tools = [ToolSpec("shell", "run a command", {})]
    prompt = history.build_prompt(tools)

    # 快照之后又有新事实到达（例如迟到的工具结果）
    history.record(Item(type="function_call_output", call_id="c1", output="hi"))

    print("prompt.input 长度 =", len(prompt.input))       # 1 —— 不受追加影响
    print("history 长度     =", len(history))             # 2 —— 权威副本已更新
    print("base_instructions =", prompt.base_instructions)

    next_prompt = history.build_prompt(tools)
    print("下一轮 prompt 长度 =", len(next_prompt.input))  # 2 —— 新事实此时才可见


if __name__ == "__main__":
    import asyncio

    asyncio.run(demo())