架构定位

能力扩展

通过能力包、外部证据、持久状态和标准协议扩展 Agent 能做的事情。

如何为 Agent 增加 Skill、检索、记忆和外部协议能力？

概念校准

记忆

它是什么: 跨步骤或会话保存、检索并使用状态的机制，必须定义写入、读取和遗忘策略。
它不是什么: 不只是压缩历史对话，也不应把所有旧信息重新塞回上下文。

上下文

它是什么: 一次模型调用可见的信息集合，包括指令、对话、工具 Schema 与结果、检索证据和当前状态。
它不是什么: 不只是聊天记录，也不等于模型能够永久记住的全部信息。

可观测性

它是什么: 通过结构化日志、Trace、指标和版本信息解释系统在一次运行中发生了什么。
它不是什么: 不只是保存终端输出；没有关联 ID、状态转换和敏感信息边界的日志难以诊断。

状态、会话、记忆与上下文压缩

设计会话状态、Checkpoint、短期记忆、长期记忆、压缩和重放。

本章只做一件事长任务可以暂停、恢复、压缩并重放关键决策。

开始前，Forge 已经具备已完成 Module 09 的“RAG、检索与知识评测”，其通过验收的 solution 是本章起点。

完成后，Forge 将能够让长任务可以暂停、恢复和重放，同时控制上下文增长。

卡住时的最小恢复点只保存显式状态和最后一个 Checkpoint，不启用长期记忆。

下载本章实验仓库Python 3.12 · pytest · Pydantic · SQLite · deterministic Mock

阶段能力链

01区分当前状态与历史事实，并用不可变事件确定性重建 RunState02区分会话和单次运行，并显式管理运行开始、结束与并发边界03保存带版本和完整性校验的最小 Checkpoint，并在损坏或版本不兼容时拒绝恢复04用 Checkpoint 加尾部事件恢复运行，并证明恢复结果与完整事件重放一致05把可恢复运行状态与可检索长期记忆分开，并实现作用域、过期和删除语义06只压缩普通对话材料，同时逐字保留指令、权限决定和承诺，并用保真检查阻止静默丢失07用可审计事件日志重放运行，并把记忆删除同步传播到依赖它的派生摘要缓存

每一步只增加一种可验证能力；后一步建立在前一步已经通过的代码和测试上。

概念校准\n\n在开始操作代码前，先把本章涉及的概念放回正确的工程边界。\n\n
\n\n### 记忆 memory\n\n- 它是什么: 跨步骤或会话保存、检索并使用状态的机制，必须定义写入、读取、冲突、保留和遗忘策略。\n- 它不是什么: 它不只是历史摘要，也不应把全部旧记录放进每次上下文；保存的数据只有被正确检索和使用时才形成有效记忆。\n- 与相邻概念的关系: State 表示当前进度，Checkpoint 保存一致快照，Memory 跨边界复用信息，Context 只承载本次调用选择出的部分。\n- 在 HeatStack Forge 中的位置: Forge 把任务状态、用户确认和长期偏好分开存储，使用版本与作用域检索，并让任务可暂停、恢复和重放。\n- 典型误用与修正: 误用是每轮自动写入全部模型输出。修正方法是设置写入门槛、来源、过期和冲突策略，并测试陈旧数据不能覆盖当前事实。\n\n### 上下文 context\n\n- 它是什么: 一次模型调用实际可见的信息集合，包括指令、对话片段、工具 Schema 与结果、检索证据、当前状态和输出约束。\n- 它不是什么: 它不只是聊天记录，也不是数据库中的全部信息；只有经过选择并放入本次调用的信息才属于本次上下文。\n- 与相邻概念的关系: RAG 选择外部证据，Memory 保存跨步骤状态，Tool Schema 描述可用动作；三者都可供给上下文，但职责不同。\n- 在 HeatStack Forge 中的位置: Forge 的 context builder 按任务、权限、预算和来源优先级组装输入，并记录每段信息的来源与版本。\n- 典型误用与修正: 误用是把更多 Token 当成更完整的理解。修正方法是建立选择、截断、引用和敏感信息过滤规则，并用固定夹具验证。\n\n### 可观测性 observability\n\n- 它是什么: 通过结构化日志、Trace、指标、状态转换和版本信息解释一次运行发生了什么，并支持定位失败与比较版本。\n- 它不是什么: 它不只是终端输出，也不是无限记录所有输入；缺少关联 ID、状态转换或敏感信息边界的日志既难排查又可能泄露数据。\n- 与相邻概念的关系: Harness 产生运行事件，Agent Loop 与工具调用形成 Trace，Evaluation 聚合指标，Memory 与恢复依赖版本记录。\n- 在 HeatStack Forge 中的位置: Forge 为每次任务生成 operation_id，记录阶段、工具、耗时、成本、错误类别、状态变化和证据路径，同时过滤敏感信息。\n- 典型误用与修正: 误用是等失败后再临时加日志。修正方法是在合同设计时定义事件与关联字段，并用故障注入确认能重建时间线。\n\n

STAGE 01

区分当前状态与历史事实，并用不可变事件确定性重建 RunState

先看一个具体问题

如何从事件序列重建当前状态？

你正在构建一个长任务执行系统，需要记录运行过程中的关键事实，例如运行创建、步骤开始、权限撤销等。这些事实一旦发生就不可更改，但当前状态（如运行是否还在进行、当前步骤是什么、是否允许写入）会随着事件不断变化。

如果直接修改一个可变的状态对象，当系统崩溃或需要审计时，你无法知道状态是如何一步步变成现在这样的。因此，你需要将事件作为不可变的历史记录，并通过重放事件来确定性重建状态。

本阶段你将实现一个事件溯源的核心：定义事件和状态模型，并编写 apply_event 函数，使每个事件都能更新状态。同时，你需要保证事件序列连续且运行终止后不再接受新事件。

先做判断

在实现 reduce_events 之前，请预测：如果事件序列中存在间隙（例如序列 1 后直接是序列 3），系统应该如何处理？

忽略间隙，继续应用后续事件
抛出异常，因为状态无法可靠重建
自动填补缺失的事件
仅记录警告，但继续执行

判断依据: 正确答案是抛出异常。事件序列必须连续，因为每个事件都依赖于前一个事件产生的状态。如果存在间隙，说明历史记录不完整，无法确定中间发生了什么，因此必须拒绝重建，而不是猜测或忽略。

它是什么

状态是当前进度的投影，事件是历史事实

状态（RunState）表示某一时刻运行的最新情况，例如运行状态、当前步骤、是否允许写入。它是从事件序列中推导出来的，而不是独立存储的。

事件（RunEvent）是不可变的历史记录，每个事件包含序列号、类型和负载。事件一旦创建就不能修改，因此可以安全地重放。

通过 apply_event 函数，将事件逐个应用到状态上，最终得到当前状态。这个过程是确定性的：给定相同的事件序列，总是得到相同的状态。

它不是什么

状态不是历史记录，事件也不是可变对象

状态不是所有事件的列表，它只包含当前需要的信息。历史事件可以保留用于审计，但状态是精简的投影。

事件不是可变的，一旦创建就不能修改。如果发现错误，不能修改旧事件，而应该追加一个新事件来纠正。

状态不能直接从历史记录中读取，必须通过重放事件来重建，否则无法保证一致性和可审计性。

它与相邻概念的关系

事件驱动状态转换

每个事件都对应一个状态转换。例如，RUN_CREATED 事件创建初始状态，STEP_STARTED 事件更新当前步骤，WRITE_PERMISSION_REVOKED 事件将 write_allowed 设为 False。

事件序列必须连续，因为每个事件都基于前一个事件产生的状态。如果序列有间隙，就无法确定中间发生了什么，因此必须拒绝。

运行终止后（COMPLETED 或 FAILED），不能再接受新事件，因为状态已经最终确定，继续应用事件会导致不一致。

本阶段的边界决定: 当事件序列不连续或运行已终止时，必须抛出异常，而不是尝试修复或忽略。这保证了状态重建的可靠性和可审计性。

实验 ID: state-memory-context-01-state-and-events
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/state-memory-context/01-state-and-events/
验收: 待验收

教师示范

目标: 演示如何使用 reduce_events 从三个事件重建 RunState，并验证状态字段。

假设我们有一个运行，首先创建运行（序列1），然后开始一个步骤（序列2），最后撤销写入权限（序列3）。

我们使用 reduce_events 函数，传入这三个事件，它将依次调用 apply_event 来更新状态。

最终状态应该是：运行状态为 RUNNING，当前步骤为 “inspect”，write_allowed 为 False，版本号为 3。

实施步骤

定义三个事件：RUN_CREATED、STEP_STARTED、WRITE_PERMISSION_REVOKED。
调用 reduce_events(events)。
断言返回的状态字段符合预期。

真实 solution 代码: forge/state/__init__.py

from .models import EventKind, RunEvent, RunState, RunStatus, apply_event, reduce_events

__all__ = ["EventKind", "RunEvent", "RunState", "RunStatus", "apply_event", "reduce_events"]

真实 solution 代码: forge/state/models.py

from __future__ import annotations

from enum import StrEnum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field

class RunStatus(StrEnum):
    CREATED = "created"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

class EventKind(StrEnum):
    RUN_CREATED = "run_created"
    STEP_STARTED = "step_started"
    WRITE_PERMISSION_REVOKED = "write_permission_revoked"
    RUN_COMPLETED = "run_completed"
    RUN_FAILED = "run_failed"

class RunEvent(BaseModel):
    model_config = ConfigDict(frozen=True)
    event_id: str = Field(min_length=1)
    run_id: str = Field(min_length=1)
    sequence: int = Field(ge=1)
    kind: EventKind
    payload: dict[str, Any] = Field(default_factory=dict)

class RunState(BaseModel):
    model_config = ConfigDict(frozen=True)
    run_id: str
    status: RunStatus
    version: int
    last_event_id: str
    current_step: str | None = None
    write_allowed: bool = True
    failure_reason: str | None = None

def apply_event(state: RunState | None, event: RunEvent) -> RunState:
    if state is None:
        if event.kind != EventKind.RUN_CREATED or event.sequence != 1:
            raise ValueError("the first event must create the run at sequence 1")
        return RunState(run_id=event.run_id, status=RunStatus.CREATED, version=1, last_event_id=event.event_id, write_allowed=bool(event.payload.get("write_allowed", True)))
    if event.run_id != state.run_id:
        raise ValueError("an event cannot be applied to a different run")
    if event.sequence != state.version + 1:
        raise ValueError("event sequence must be contiguous")
    if state.status in {RunStatus.COMPLETED, RunStatus.FAILED}:
        raise ValueError("a terminal run cannot accept more events")
    update: dict[str, Any] = {"version": event.sequence, "last_event_id": event.event_id}
    if event.kind == EventKind.STEP_STARTED:
        step = str(event.payload.get("step", "")).strip()
        if not step:
            raise ValueError("step_started requires a non-empty step")
        update.update(status=RunStatus.RUNNING, current_step=step)
    elif event.kind == EventKind.WRITE_PERMISSION_REVOKED:
        update["write_allowed"] = False
    elif event.kind == EventKind.RUN_COMPLETED:
        update.update(status=RunStatus.COMPLETED, current_step=None)
    elif event.kind == EventKind.RUN_FAILED:
        reason = str(event.payload.get("reason", "")).strip()
        if not reason:
            raise ValueError("run_failed requires a reason")
        update.update(status=RunStatus.FAILED, failure_reason=reason, current_step=None)
    elif event.kind == EventKind.RUN_CREATED:
        raise ValueError("a run can only be created once")
    return state.model_copy(update=update)

def reduce_events(events: list[RunEvent], initial: RunState | None = None) -> RunState:
    state = initial
    for event in events:
        state = apply_event(state, event)
    if state is None:
        raise ValueError("at least one event is required")
    return state

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

RUN_CREATED 事件创建初始状态，设置 run_id 和 write_allowed（默认为 True）。

STEP_STARTED 事件将状态更新为 RUNNING，并设置 current_step 为 “inspect”。

WRITE_PERMISSION_REVOKED 事件将 write_allowed 设为 False。

最终状态版本号为 3，因为最后一个事件的序列号是 3。

带提示练习

补全 apply_event 中缺失的序列连续性检查，使 test_sequence_gaps_are_rejected 通过。

打开 forge/state/models.py，找到 apply_event 函数。
在检查 run_id 之后、检查终端状态之前，添加序列连续性检查：如果 event.sequence != state.version + 1，则抛出 ValueError。
运行测试 test_sequence_gaps_are_rejected，确认通过。

逐级提示

提示1：状态对象有一个 version 字段，表示已应用的最新事件序列号。
提示2：新事件的序列号必须比当前版本大1。
提示3：错误消息中应包含 “contiguous” 一词，以满足测试断言。

完成标准

test_sequence_gaps_are_rejected 通过
apply_event 在序列不连续时抛出 ValueError

故障注入与排查

故障夹具位于 fixtures/failure.json，它模拟了事件日志跳过序列2的情况。运行测试 test_sequence_gaps_are_rejected 会失败，因为 reduce_events 接受了序列1后直接是序列3的事件，而没有抛出 ValueError。你需要诊断并修复这个问题。

本阶段你已经学习了事件序列必须连续的原则，并且知道 apply_event 函数负责应用事件。故障夹具中的缺陷是缺少连续性检查，这正是你在 guidedPractice 中补全的内容。通过检查 apply_event 的代码，你可以发现缺少 if event.sequence != state.version + 1 这一行，并修复它。

已验证的故障合同

目标测试: test_sequence_gaps_are_rejected
故障文件: forge/state/models.py

诊断顺序

运行故障命令：python -m pytest -q ../tests/test_stage.py::test_sequence_gaps_are_rejected
观察输出：测试失败，AssertionError: a sequence gap must be rejected
检查 forge/state/models.py 中的 apply_event 函数，发现缺少序列连续性检查。
确认在 run_id 检查之后没有 if event.sequence != state.version + 1 这一行。

修复原则: 在 apply_event 函数中，在检查 run_id 之后、检查终端状态之前，添加以下代码：if event.sequence != state.version + 1: raise ValueError(“event sequence must be contiguous”)。

python -m pytest -q ../tests/test_stage.py::test_sequence_gaps_are_rejected

迁移挑战

办公研究场景：一个运行首先创建（序列1），然后开始检索文档（序列2），接着撤销写入权限（序列3），最后生成简报（序列4）。

改变的条件: 事件序列的内容和顺序与原始示例不同，但事件类型和约束相同。

使用 reduce_events 从上述事件序列重建状态，并验证 write_allowed 为 False，当前步骤为 “generate_brief”，状态为 RUNNING。

完成标准

状态对象的 write_allowed 属性为 False
状态对象的 current_step 属性为 “generate_brief”
状态对象的 status 属性为 RunStatus.RUNNING

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么事件序列必须连续？如果运行已经完成，为什么不能接受更多事件？

一个有力回答应包含

事件序列必须连续，因为每个事件都基于前一个事件产生的状态。如果存在间隙，就无法确定中间发生了什么，状态重建将不可靠。
运行完成后，状态已经最终确定，继续应用事件会导致状态不一致，例如在 COMPLETED 后又有 STEP_STARTED，这违反了业务规则。

掌握证据

所有三个测试通过：test_events_rebuild_the_same_state_deterministically、test_sequence_gaps_are_rejected、test_terminal_runs_reject_later_events
能够解释状态与事件的区别：状态是当前进度的投影，事件是不可变的历史事实。
能够说明为什么事件序列必须连续以及终端状态不能接受新事件。

STAGE 02

区分会话和单次运行，并显式管理运行开始、结束与并发边界

先看一个具体问题

为什么需要区分会话和运行？

在上一阶段，我们建立了状态和事件模型，但还没有管理多个运行的能力。现在，一个用户会话可能包含多次运行，例如一次失败后重试，或者并行处理多个任务。如果每次运行都使用相同的 run_id，历史记录就会混在一起，无法区分哪次运行产生了哪些事件。

本阶段要求实现一个 Session 容器，它能够启动多个运行、完成运行，并保证 run_id 在会话内唯一。测试 test_session_tracks_multiple_runs_without_merging_them 验证会话能正确跟踪多个运行而不合并它们；测试 test_duplicate_run_id_is_rejected 验证重复的 run_id 会被拒绝。

先做判断

在实现 start_run 时，如果传入一个已经存在的 run_id，应该发生什么？

静默忽略，不添加新运行
抛出 ValueError，因为 run_id 必须唯一
覆盖旧的运行记录
自动生成一个新的 run_id

判断依据: 正确答案是抛出 ValueError。因为测试 test_duplicate_run_id_is_rejected 明确要求重复 run_id 被拒绝，并且错误消息中包含 ‘unique’。静默忽略或覆盖都会导致历史丢失或混乱，自动生成新 ID 则违背了调用者的意图。

它是什么

会话是运行的容器

Session 是一个不可变的数据结构，包含 session_id 和一个 runs 元组。每个 run 是一个 RunRecord，包含 run_id 和 status。start_run 函数向会话中添加一个新的 RunRecord，并返回新的 Session 实例。finish_run 函数将指定 run 的状态改为终态（completed、failed 或 cancelled）。

会话的生命周期由这些函数显式管理：创建会话、启动运行、完成运行。运行 ID 在会话内必须唯一，这是通过 start_run 中的检查保证的。

它不是什么

会话不是单次运行

会话不是一次具体的执行，而是多次执行的容器。运行是单次执行，有明确的状态转换。混淆两者会导致状态管理混乱，例如无法区分不同运行的事件。

会话也不等同于聊天记录或上下文；它只保存运行的结构信息，不保存具体的事件内容。

它与相邻概念的关系

会话、运行和状态的关系

Session 包含多个 RunRecord，每个 RunRecord 有 run_id 和 status。start_run 和 finish_run 是操作 Session 的函数，它们返回新的 Session 实例，保持不可变性。active_run_ids 属性从 runs 中筛选出状态为 active 的运行 ID。

这种设计使得会话状态可以安全地传递和共享，而不会意外修改。

本阶段的边界决定: 在 start_run 中，必须检查 run_id 是否已存在；如果存在，抛出 ValueError。这是会话内运行 ID 唯一性的边界。

实验 ID: state-memory-context-02-session-lifecycle
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/state-memory-context/02-session-lifecycle/
验收: 待验收

教师示范

目标: 演示如何创建会话、启动两个运行、完成一个运行，并验证活动运行列表。

假设我们有一个会话 session-a，需要依次启动 run-1 和 run-2，然后完成 run-1。最终，活动运行应该只有 run-2，并且 runs 列表按启动顺序包含 run-1（completed）和 run-2（active）。

这个例子展示了 Session 的基本用法，以及 start_run 和 finish_run 如何协作管理运行状态。

实施步骤

创建 Session 实例：session = Session(session_id=“session-a”)
启动第一个运行：session = start_run(session, “run-1”)
启动第二个运行：session = start_run(session, “run-2”)
完成第一个运行：session = finish_run(session, “run-1”, “completed”)
验证 active_run_ids 为 (“run-2”,)，runs 列表顺序正确。

真实 solution 代码: forge/state/session.py

from __future__ import annotations
from pydantic import BaseModel, ConfigDict, Field

class RunRecord(BaseModel):
    model_config = ConfigDict(frozen=True)
    run_id: str
    status: str = "active"

class Session(BaseModel):
    model_config = ConfigDict(frozen=True)
    session_id: str
    runs: tuple[RunRecord, ...] = Field(default_factory=tuple)
    @property
    def active_run_ids(self) -> tuple[str, ...]:
        return tuple(run.run_id for run in self.runs if run.status == "active")

def start_run(session: Session, run_id: str) -> Session:
    if any(run.run_id == run_id for run in session.runs):
        raise ValueError("run_id must be unique inside a session")
    return session.model_copy(update={"runs": (*session.runs, RunRecord(run_id=run_id))})

def finish_run(session: Session, run_id: str, status: str) -> Session:
    if status not in {"completed", "failed", "cancelled"}:
        raise ValueError("finish status must be terminal")
    found = False
    updated = []
    for run in session.runs:
        if run.run_id == run_id:
            found = True
            if run.status != "active":
                raise ValueError("run is already terminal")
            updated.append(run.model_copy(update={"status": status}))
        else:
            updated.append(run)
    if not found:
        raise KeyError(run_id)
    return session.model_copy(update={"runs": tuple(updated)})

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

start_run 每次添加一个新的 RunRecord，状态默认为 active。finish_run 找到对应的 run_id，将其状态改为 completed。由于 Session 是不可变的，每次操作都返回新实例。

active_run_ids 属性通过列表推导筛选出状态为 active 的 run_id，因此最终只包含 run-2。

带提示练习

在 start_run 中添加重复 run_id 检查，使 test_duplicate_run_id_is_rejected 通过。

打开 forge/state/session.py 文件。
在 start_run 函数中，在添加新 RunRecord 之前，检查 session.runs 中是否已有相同 run_id 的记录。
如果存在，抛出 ValueError，消息中包含 ‘unique’。
运行测试验证。

逐级提示

使用 any() 函数遍历 session.runs，比较 run.run_id 与传入的 run_id。
错误消息可以写成 “run_id must be unique inside a session”。

完成标准

test_duplicate_run_id_is_rejected 通过。
test_session_tracks_multiple_runs_without_merging_them 仍然通过。

故障注入与排查

故障夹具 fixtures/failure.json 中，forge/state/session.py 的 start_run 函数没有包含重复 run_id 检查。运行测试 test_duplicate_run_id_is_rejected 会失败，因为第二次调用 start_run 时，相同的 run_id 被追加到会话中，导致测试抛出 AssertionError。

本阶段的核心就是实现 run_id 唯一性检查。学习者已经通过 guidedPractice 在 forge/state/session.py 中添加了该检查，因此能够识别故障原因并修复。故障夹具中的 start_run 函数与 guidedPractice 之前的版本相同，缺少检查。

已验证的故障合同

目标测试: test_duplicate_run_id_is_rejected
故障文件: forge/state/session.py

诊断顺序

运行故障测试命令，观察 AssertionError。
检查 forge/state/session.py 中的 start_run 函数，发现没有检查重复 run_id。
对比正确实现，确认缺少 any() 检查。

修复原则: 在 forge/state/session.py 的 start_run 函数中添加重复检查：if any(run.run_id == run_id for run in session.runs): raise ValueError(“run_id must be unique inside a session”)。

python -m pytest -q ../tests/test_stage.py::test_duplicate_run_id_is_rejected

迁移挑战

在音乐创作场景中，使用 Session 管理多个曲目运行。每个曲目运行代表一次创作尝试，例如生成旋律、编曲等。你需要确保曲目运行 ID 在会话内唯一，例如不能有两个 ‘track-1’ 运行。

改变的条件: 场景从软件交付变为音乐创作，但 Session 和运行机制不变。

编写代码，使用 Session 启动多个曲目运行，并故意尝试启动重复的曲目运行 ID，验证会抛出 ValueError。

完成标准

能够创建 Session 并启动多个不同 ID 的运行。
尝试启动重复 ID 时抛出 ValueError。
活动运行列表正确。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么运行 ID 在会话内必须唯一？请结合本阶段实现的 start_run 检查机制，说明如果允许重复 ID 会导致哪些具体问题。

一个有力回答应包含

运行 ID 用于标识单次执行，如果重复，就无法区分不同运行的事件和状态。
唯一性保证历史记录不会混淆，例如重试时不会覆盖或合并之前的运行。
start_run 中的 any() 检查在添加新运行前验证 ID 不存在，从而维护会话内运行的可辨识性。

掌握证据

两个测试均通过。
能够解释会话和运行的区别。
能够在 start_run 中实现重复 ID 检查。

STAGE 03

保存带版本和完整性校验的最小 Checkpoint，并在损坏或版本不兼容时拒绝恢复

先看一个具体问题

保存的检查点可能被篡改或版本不兼容

在长任务执行中，我们需要保存当前状态以便暂停、恢复或重放。如果检查点文件被意外修改（例如磁盘错误或人为篡改），恢复时可能加载到错误的状态，导致后续决策基于不可信的数据。

此外，随着系统演进，检查点的结构可能变化。如果加载旧版本或未知版本的检查点，程序可能因字段缺失或类型不匹配而崩溃，或者静默地产生错误结果。

本阶段要求实现一个带版本号和 SHA-256 校验和的最小检查点模块，在加载时验证版本和完整性，拒绝损坏或不支持的快照。

先做判断

在实现检查点加载时，以下哪种做法能最可靠地防止加载被篡改的状态？

只检查 JSON 文件是否能被解析为 CheckpointEnvelope 模型。
在加载时重新计算状态的 SHA-256 校验和，并与存储的校验和比较。
信任文件系统权限，认为只有授权用户能修改文件。
在加载时只检查 schema_version 是否等于当前版本。

判断依据: 正确选项是重新计算校验和并比较。仅解析 JSON 或检查版本无法发现内容被篡改；文件系统权限不能防止所有意外修改；版本检查只能处理结构兼容性，不能保证数据完整性。

它是什么

检查点契约

检查点是一个包含 schema_version、state 和 checksum_sha256 的不可变信封（CheckpointEnvelope）。schema_version 标识结构版本，state 是当前运行状态，checksum_sha256 是对 schema_version 和 state 的规范化 JSON 表示计算的 SHA-256 哈希。

保存时，我们计算校验和并原子写入文件；加载时，先验证 schema_version 是否受支持，然后重新计算校验和并与存储值比较，任何不匹配都抛出 ValueError。

它不是什么

检查点不是什么

检查点不是状态的简单序列化，它必须包含完整性证明。它也不是长期记忆或会话历史的替代品；它只保存某一时刻的状态快照。

检查点不负责决定何时保存或恢复，也不处理并发写入冲突；这些策略由上层会话管理决定。

它与相邻概念的关系

与其他概念的关系

检查点与状态（RunState）紧密相关：它封装了状态并提供持久化。与上下文不同，检查点不会自动进入模型上下文；只有显式加载并选择部分状态后，才成为上下文的一部分。

可观测性方面，检查点的版本和校验和提供了审计线索，帮助诊断恢复失败的原因。

本阶段的边界决定: 当加载检查点时，必须同时验证 schema_version 和 checksum_sha256；任何一项失败都应拒绝恢复，而不是尝试部分加载或静默修复。

实验 ID: state-memory-context-03-checkpoint-contract
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/state-memory-context/03-checkpoint-contract/
验收: 待验收

教师示范

目标: 演示保存和加载检查点，并验证状态完全一致。

我们从一个简单的运行状态开始：先创建运行，然后开始一个步骤，最后撤销写权限。通过 reduce_events 得到 RunState。

使用 save_checkpoint 将状态保存到临时目录中的文件，返回 CheckpointEnvelope。然后使用 load_checkpoint 加载该文件，并断言加载的状态与保存的状态相等，且校验和长度为 64。

实施步骤

构造三个事件：RUN_CREATED（write_allowed=True）、STEP_STARTED（step=‘inspect’）、WRITE_PERMISSION_REVOKED。
调用 reduce_events 得到 state。
调用 save_checkpoint(path, state)，其中 path 是 tmp_path / ‘run-checkpoint.json’。
调用 load_checkpoint(path) 得到 loaded。
断言 loaded == saved、loaded.state == state、loaded.state.write_allowed is False、len(loaded.checksum_sha256) == 64。

真实 solution 代码: forge/state/checkpoint.py

from __future__ import annotations

import hashlib
import hmac
import json
import os
from pathlib import Path
from typing import Any

from pydantic import BaseModel, ConfigDict, Field

from .models import RunState

CHECKPOINT_SCHEMA_VERSION = 1


class CheckpointEnvelope(BaseModel):
    model_config = ConfigDict(frozen=True)

    schema_version: int = Field(ge=1)
    state: RunState
    checksum_sha256: str = Field(min_length=64, max_length=64)


def _canonical_payload(schema_version: int, state: RunState) -> bytes:
    payload: dict[str, Any] = {
        "schema_version": schema_version,
        "state": state.model_dump(mode="json"),
    }
    return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")


def checkpoint_checksum(schema_version: int, state: RunState) -> str:
    return hashlib.sha256(_canonical_payload(schema_version, state)).hexdigest()


def save_checkpoint(path: str | Path, state: RunState) -> CheckpointEnvelope:
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    envelope = CheckpointEnvelope(
        schema_version=CHECKPOINT_SCHEMA_VERSION,
        state=state,
        checksum_sha256=checkpoint_checksum(CHECKPOINT_SCHEMA_VERSION, state),
    )
    temporary = target.with_name(f".{target.name}.tmp")
    temporary.write_text(
        json.dumps(envelope.model_dump(mode="json"), ensure_ascii=False, sort_keys=True, indent=2) + "\n",
        encoding="utf-8",
    )
    os.replace(temporary, target)
    return envelope


def load_checkpoint(path: str | Path) -> CheckpointEnvelope:
    raw = json.loads(Path(path).read_text(encoding="utf-8"))
    envelope = CheckpointEnvelope.model_validate(raw)
    if envelope.schema_version != CHECKPOINT_SCHEMA_VERSION:
        raise ValueError(f"unsupported checkpoint schema version: {envelope.schema_version}")
    expected = checkpoint_checksum(envelope.schema_version, envelope.state)
    if not hmac.compare_digest(expected, envelope.checksum_sha256):
        raise ValueError("checkpoint checksum mismatch")
    return envelope

真实 solution 代码: forge/state/__init__.py

from .checkpoint import CheckpointEnvelope, checkpoint_checksum, load_checkpoint, save_checkpoint
from .models import EventKind, RunEvent, RunState, RunStatus, apply_event, reduce_events
from .session import RunRecord, Session, finish_run, start_run

__all__ = [
    "CheckpointEnvelope",
    "EventKind",
    "RunEvent",
    "RunRecord",
    "RunState",
    "RunStatus",
    "Session",
    "apply_event",
    "checkpoint_checksum",
    "finish_run",
    "load_checkpoint",
    "reduce_events",
    "save_checkpoint",
    "start_run",
]

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

save_checkpoint 内部创建 CheckpointEnvelope，计算规范化负载的 SHA-256 校验和，并原子写入文件。

load_checkpoint 读取 JSON，验证 schema_version 和校验和，然后返回信封。由于文件未被修改，校验和匹配，加载成功。

带提示练习

在 load_checkpoint 中恢复校验和比较，使 test_checkpoint_corruption_is_rejected 通过。

打开 forge/state/checkpoint.py，找到 load_checkpoint 函数。
在版本检查之后，调用 checkpoint_checksum(envelope.schema_version, envelope.state) 计算期望校验和。
使用 hmac.compare_digest(expected, envelope.checksum_sha256) 比较期望值与存储值。
如果不匹配，抛出 ValueError(“checkpoint checksum mismatch”)。
运行测试验证。

逐级提示

checkpoint_checksum 函数已经存在，它接受 schema_version 和 state 并返回十六进制字符串。
hmac.compare_digest 用于常量时间比较，避免时序攻击。
确保在返回 envelope 之前进行检查。

完成标准

test_checkpoint_corruption_is_rejected 通过。
load_checkpoint 在篡改后抛出包含 ‘checksum mismatch’ 的 ValueError。

故障注入与排查

故障夹具 fixtures/failure.json 描述了一个检查点损坏场景：状态被手动编辑但未重新计算 SHA-256 摘要，恢复必须停止。当前 faulty 代码在 load_checkpoint 中跳过了校验和比较，导致 test_checkpoint_corruption_is_rejected 失败。可执行故障夹具修改 forge/state/checkpoint.py，并由 test_checkpoint_corruption_is_rejected 稳定复现。已验证机制标识为 load_checkpoint、hmac.compare_digest。

本阶段的核心就是实现完整性校验。学习者已经通过 guidedPractice 在 load_checkpoint 中添加了校验和比较，因此能够识别出 faulty 代码中缺失的检查，并知道如何修复。

已验证的故障合同

目标测试: test_checkpoint_corruption_is_rejected
故障文件: forge/state/checkpoint.py

诊断顺序

运行故障测试：python -m pytest -q ../tests/test_stage.py::test_checkpoint_corruption_is_rejected，观察失败信息 ‘Failed: DID NOT RAISE ValueError’。
检查 load_checkpoint 函数，发现只有 schema_version 检查，没有校验和比较。
对比正确实现，确认缺少 expected = checkpoint_checksum(…) 和 hmac.compare_digest 检查。

修复原则: 在 load_checkpoint 中，版本检查后添加：expected = checkpoint_checksum(envelope.schema_version, envelope.state)；if not hmac.compare_digest(expected, envelope.checksum_sha256): raise ValueError(“checkpoint checksum mismatch”)。

python -m pytest -q ../tests/test_stage.py::test_checkpoint_corruption_is_rejected

迁移挑战

在软件交付场景中，你正在为前端仓库添加功能。运行状态包括当前分支、修改的文件列表和测试结果。你需要保存一个检查点，然后模拟文件被篡改，验证加载失败。

改变的条件: 场景数据从通用运行状态变为软件交付的具体状态，但检查点机制不变。

使用现有的 save_checkpoint 和 load_checkpoint，创建一个包含软件交付状态的检查点，手动修改 JSON 文件中的某个字段（例如将 write_allowed 改为 True），然后尝试加载，确认抛出 ValueError。

完成标准

成功保存检查点并加载未修改的文件。
修改文件后加载抛出包含 ‘checksum mismatch’ 的 ValueError。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么使用 SHA-256 而不是简单哈希（如 CRC32）来保护检查点完整性？

一个有力回答应包含

SHA-256 是密码学哈希，具有抗碰撞性，难以找到两个不同输入产生相同哈希。
CRC32 设计用于检测随机错误，不是抗篡改的，攻击者可以轻易构造碰撞。
使用 SHA-256 可以确保即使攻击者知道算法，也无法在不改变哈希的情况下修改状态。

掌握证据

所有三个测试通过：test_checkpoint_round_trip_preserves_exact_state、test_checkpoint_corruption_is_rejected、test_unknown_checkpoint_version_is_rejected。
能够解释检查点中版本号和校验和的作用，以及为什么需要两者。
在故障实验室中成功诊断并修复缺失的校验和比较。

STAGE 04

用 Checkpoint 加尾部事件恢复运行，并证明恢复结果与完整事件重放一致

先看一个具体问题

如何证明从检查点恢复的状态与完整事件重放完全一致？

在长任务执行中，系统会定期保存检查点（Checkpoint）以支持暂停和恢复。当任务中断后，我们希望通过加载检查点并重放检查点之后的事件（尾部事件）来恢复运行状态。

然而，仅仅加载检查点并应用尾部事件并不能保证恢复结果正确。如果尾部事件不完整（例如遗漏了某个事件），恢复出的状态可能与从头重放所有事件得到的状态不一致。

本阶段要求你实现一个验证函数 verify_recovery_equivalent，它比较“检查点+尾部事件”恢复的状态与“完整事件重放”得到的状态，若不一致则抛出 RecoveryMismatch 异常。

先做判断

在实现恢复等价性验证之前，请预测：如果从检查点恢复时遗漏了一个尾部事件，会发生什么？

恢复的状态可能与完整重放不一致，但系统无法自动发现
恢复的状态一定与完整重放一致，因为检查点包含了所有必要信息
系统会抛出异常，因为检查点校验和会检测到事件缺失
恢复的状态可能不一致，但可以通过日志手动检查

判断依据: 正确答案是第一个选项。检查点只保存了某个时刻的状态快照，它无法知道之后应该发生哪些事件。如果尾部事件不完整，恢复出的状态就会缺少那些事件的影响，导致与完整重放不一致。因此需要显式比较来发现这种不一致。

它是什么

恢复等价性验证是什么

恢复等价性验证是一种安全机制，它通过比较两种计算路径的结果来确保恢复正确：路径一是从检查点加载状态并应用尾部事件；路径二是从头重放完整事件历史。

如果两条路径得到的状态完全相同，则证明检查点和尾部事件足以重建完整状态；如果不同，则说明尾部事件有缺失或顺序错误，必须抛出异常。

它不是什么

恢复等价性验证不是什么

它不是检查点校验和验证。校验和只能保证检查点文件未被篡改，不能保证尾部事件完整。

它也不是简单的状态比较，而是必须实际执行完整重放，因为状态可能包含复杂对象，仅靠表面字段无法判断深层一致性。

它与相邻概念的关系

与其他概念的关系

检查点（Checkpoint）保存了某个时刻的状态快照，是恢复的起点。

事件重放（Event Replay）通过依次应用事件来构建状态，是完整路径。

恢复等价性验证将两者结合：检查点提供加速，尾部事件提供增量，完整重放提供基准。

本阶段的边界决定: 当且仅当检查点加尾部事件恢复的状态与完整事件重放的状态完全相等时，才认为恢复等价；否则必须抛出 RecoveryMismatch 异常，阻止错误状态继续传播。

实验 ID: state-memory-context-04-recovery-equivalence
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/state-memory-context/04-recovery-equivalence/
验收: 待验收

教师示范

目标: 演示如何从检查点加尾部事件恢复，并与完整重放比较，确保结果一致。

假设一个运行历史包含四个事件：RUN_CREATED、STEP_STARTED(inspect)、WRITE_PERMISSION_REVOKED、STEP_STARTED(report)。我们在第二个事件后保存了检查点。

现在任务中断，我们加载检查点，并重放尾部事件（第三个和第四个事件）。恢复出的状态应该与从头重放所有四个事件得到的状态完全相同。

如果尾部事件遗漏了第四个事件，恢复出的状态将停留在第三个事件之后，与完整重放不一致，此时验证函数应抛出 RecoveryMismatch。

实施步骤

使用 save_checkpoint 在第二个事件后保存检查点。
调用 recover_state 加载检查点并应用尾部事件，得到恢复状态。
调用 reduce_events 重放完整事件历史，得到基准状态。
比较两个状态，若相等则返回恢复状态，否则抛出 RecoveryMismatch。

真实 solution 代码: forge/state/recovery.py

from __future__ import annotations

from pathlib import Path

from .checkpoint import load_checkpoint
from .models import RunEvent, RunState, reduce_events


class RecoveryMismatch(RuntimeError):
    pass


def recover_state(checkpoint_path: str | Path, tail_events: list[RunEvent]) -> RunState:
    checkpoint = load_checkpoint(checkpoint_path)
    if tail_events and tail_events[0].sequence != checkpoint.state.version + 1:
        raise ValueError("tail events must continue immediately after the checkpoint")
    if not tail_events:
        return checkpoint.state
    return reduce_events(tail_events, initial=checkpoint.state)


def verify_recovery_equivalent(
    checkpoint_path: str | Path,
    tail_events: list[RunEvent],
    full_history: list[RunEvent],
) -> RunState:
    recovered = recover_state(checkpoint_path, tail_events)
    replayed = reduce_events(full_history)
    if recovered != replayed:
        raise RecoveryMismatch(
            "checkpoint recovery differs from full event replay; keep the original checkpoint and inspect the tail boundary"
        )
    return recovered

真实 solution 代码: forge/state/__init__.py

from .checkpoint import CheckpointEnvelope, checkpoint_checksum, load_checkpoint, save_checkpoint
from .models import EventKind, RunEvent, RunState, RunStatus, apply_event, reduce_events
from .recovery import RecoveryMismatch, recover_state, verify_recovery_equivalent
from .session import RunRecord, Session, finish_run, start_run

__all__ = [
    "CheckpointEnvelope",
    "EventKind",
    "RecoveryMismatch",
    "RunEvent",
    "RunRecord",
    "RunState",
    "RunStatus",
    "Session",
    "apply_event",
    "checkpoint_checksum",
    "finish_run",
    "load_checkpoint",
    "recover_state",
    "reduce_events",
    "save_checkpoint",
    "start_run",
    "verify_recovery_equivalent",
]

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

该测试验证了当尾部事件完整时，恢复状态与完整重放一致。

verify_recovery_equivalent 内部先调用 recover_state 得到恢复状态，再调用 reduce_events 得到重放状态，然后比较两者。

如果相等，返回恢复状态；如果不相等，抛出 RecoveryMismatch。

带提示练习

在 verify_recovery_equivalent 中添加完整重放比较，使 test_missing_tail_event_is_detected_by_equivalence_check 通过。

打开 forge/state/recovery.py 文件。
在 verify_recovery_equivalent 函数中，调用 recover_state 获取恢复状态。
调用 reduce_events(full_history) 获取完整重放状态。
比较两个状态，如果不相等，抛出 RecoveryMismatch 异常，消息中包含 ‘differs from full event replay’。
运行测试确认通过。

逐级提示

使用 != 运算符比较两个 RunState 对象。
异常消息必须包含 ‘differs from full event replay’ 以匹配测试。
确保在比较之前已经计算了 replayed 状态。

完成标准

test_missing_tail_event_is_detected_by_equivalence_check 通过
test_checkpoint_plus_tail_matches_full_replay 仍然通过
test_tail_must_start_at_checkpoint_boundary 仍然通过

故障注入与排查

当前 faulty 版本的 recovery.py 中，verify_recovery_equivalent 函数只返回 recover_state 的结果，没有与完整重放比较。因此，当尾部事件缺失时，测试 test_missing_tail_event_is_detected_by_equivalence_check 期望抛出 RecoveryMismatch，但实际没有抛出，导致测试失败。可执行故障夹具修改 forge/state/recovery.py，并由 test_missing_tail_event_is_detected_by_equivalence_check 稳定复现。已验证机制标识为 verify_recovery_equivalent、full_history。

本阶段已经学习了恢复等价性的概念，并知道必须比较检查点加尾部事件与完整重放。faulty 版本缺少这个比较步骤，因此我们可以通过添加比较逻辑来修复。

已验证的故障合同

目标测试: test_missing_tail_event_is_detected_by_equivalence_check
故障文件: forge/state/recovery.py

诊断顺序

运行测试 test_missing_tail_event_is_detected_by_equivalence_check，观察失败信息：DID NOT RAISE RecoveryMismatch。
检查 verify_recovery_equivalent 函数实现，发现它只调用了 recover_state 并直接返回结果。
确认缺少对 full_history 的重放和比较。

修复原则: 在 verify_recovery_equivalent 中，计算 replayed = reduce_events(full_history)，然后比较 recovered 和 replayed，如果不相等则抛出 RecoveryMismatch 异常。这个修复确保任何尾部事件缺失或顺序错误都会被检测到，从而防止错误状态被当作正确恢复结果使用。

python -m pytest -q ../tests/test_stage.py::test_missing_tail_event_is_detected_by_equivalence_check

迁移挑战

办公研究场景：你正在处理一份决策简报，事件历史包括：创建运行、开始研究步骤、撤销写权限、开始报告步骤。你在第二个事件后保存了检查点。现在需要从检查点恢复，但故意遗漏最后一个事件（开始报告步骤），然后调用 verify_recovery_equivalent 验证等价性。

改变的条件: 场景数据从软件交付变为办公研究，事件内容不同，但结构相同。

编写一个测试或脚本，使用办公研究的事件历史，在第二个事件后保存检查点，然后使用缩短的尾部事件（只包含第三个事件）调用 verify_recovery_equivalent，预期抛出 RecoveryMismatch。

完成标准

正确构造办公研究事件历史。
在第二个事件后保存检查点。
使用缩短的尾部事件调用 verify_recovery_equivalent。
断言抛出 RecoveryMismatch 异常。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么尾部事件必须从检查点边界之后立即开始？

一个有力回答应包含

检查点保存了某个版本的状态，其 version 字段表示已应用的事件序列号。
尾部事件必须从 version+1 开始，否则会跳过事件或重复应用事件，导致状态不一致。
recover_state 函数中的检查确保了尾部事件的第一个序列号等于 checkpoint.state.version + 1，否则抛出 ValueError。

掌握证据

所有三个测试通过：test_checkpoint_plus_tail_matches_full_replay、test_tail_must_start_at_checkpoint_boundary、test_missing_tail_event_is_detected_by_equivalence_check。
能够解释恢复等价性的原理：比较检查点+尾部事件与完整重放。
能够诊断并修复缺少完整重放比较的故障。
能够在不同场景下应用恢复等价性验证。

STAGE 05

把可恢复运行状态与可检索长期记忆分开，并实现作用域、过期和删除语义

先看一个具体问题

长期记忆与运行状态分离，并确保记忆只在正确作用域和有效期内可检索

在上一阶段，我们实现了可恢复的运行状态，但还没有长期记忆。现在需要新增 forge/memory.py，定义 MemoryScope（RUN、SESSION、USER）和 MemoryRecord，并实现 MemoryStore 的 put、query、delete、purge_expired 方法。

关键约束：查询时必须按作用域和主体过滤，并排除已过期记忆；删除操作必须显式且幂等。测试 test_memory_is_queried_only_inside_its_scope_and_subject、test_expired_memory_is_hidden_and_can_be_purged、test_deletion_is_explicit_and_idempotent 将验证这些行为。

当前起始代码没有 forge/memory.py，运行测试会报 ModuleNotFoundError。你需要从零创建该文件，并修改 forge/init.py 导出相关符号。

先做判断

在实现 MemoryStore.query 时，如果只按 scope 和 subject_id 过滤，而不检查 expires_at，会发生什么？

已过期的记忆仍然会被返回，导致后续决策使用陈旧信息。
查询会抛出异常，因为 SQL 语句不完整。
已过期的记忆会自动被数据库删除，所以不会返回。
查询结果会按创建时间排序，但不会包含过期记忆。

判断依据: 正确选项是第一个。如果查询不检查 expires_at，那么即使记忆已过期，只要 scope 和 subject_id 匹配，它就会被返回。这会导致过期记忆污染当前上下文，影响决策。测试 test_expired_memory_is_hidden_and_can_be_purged 专门验证这一点。

它是什么

记忆边界模型

记忆是一个带有作用域和生命周期的数据记录。作用域（RUN、SESSION、USER）定义了记忆的可见范围：RUN 仅在一次运行内可见，SESSION 在一次会话内可见，USER 跨会话长期可见。

每条记忆都有 created_at 和可选的 expires_at。查询时，系统根据当前时间过滤掉已过期的记忆。删除操作通过 memory_id 精确删除，并且重复删除不会报错，返回 False 表示未删除任何记录。

它不是什么

记忆边界不是什么

记忆不是简单的聊天记录压缩，也不是把所有历史信息都塞进上下文。它必须定义写入、读取、冲突、保留和遗忘策略。

记忆不等于运行状态。运行状态（如当前步骤、重试次数）是易失的，而记忆是持久化的，可以跨运行复用。

它与相邻概念的关系

记忆与其他概念的关系

记忆与上下文：上下文是一次模型调用可见的信息集合，记忆是上下文的可能来源之一。只有通过查询并放入上下文的记忆才会影响本次调用。

记忆与状态：状态表示当前进度，记忆保存跨步骤或会话的信息。状态可以通过检查点恢复，而记忆通过作用域和过期策略管理生命周期。

本阶段的边界决定: 在实现 MemoryStore.query 时，必须同时过滤 scope、subject_id 和 expires_at，确保只返回当前有效且属于请求作用域的记忆。

实验 ID: state-memory-context-05-memory-boundaries
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/state-memory-context/05-memory-boundaries/
验收: 待验收

教师示范

目标: 演示创建 MemoryStore、插入不同作用域的记忆，并查询特定作用域。

我们首先创建 forge/memory.py 文件，定义 MemoryScope 枚举和 MemoryRecord 模型。MemoryScope 包含 RUN、SESSION、USER 三个值。MemoryRecord 包含 memory_id、scope、subject_id、content、created_at 和可选的 expires_at。

然后实现 MemoryStore 类，构造函数接收数据库路径，创建 SQLite 连接和表。put 方法插入记录，query 方法按作用域和主体查询，delete 方法按 ID 删除，purge_expired 方法删除所有过期记录。

在 worked example 中，我们插入一条 RUN 作用域的记忆和一条 SESSION 作用域的记忆，然后查询 RUN 作用域，验证只返回 RUN 记忆。

实施步骤

创建 forge/memory.py，定义 MemoryScope 和 MemoryRecord。
实现 MemoryStore.init，建立数据库连接和表结构。
实现 put 方法，插入记录并处理唯一性冲突。
实现 query 方法，按 scope 和 subject_id 过滤，并排除过期记录。
实现 delete 和 purge_expired 方法。
在 forge/init.py 中导出 MemoryRecord、MemoryScope、MemoryStore。

真实 solution 代码: forge/memory.py

from __future__ import annotations

import sqlite3
from datetime import datetime, timezone
from enum import StrEnum
from pathlib import Path

from pydantic import BaseModel, ConfigDict, Field


class MemoryScope(StrEnum):
    RUN = "run"
    SESSION = "session"
    USER = "user"


class MemoryRecord(BaseModel):
    model_config = ConfigDict(frozen=True)

    memory_id: str = Field(min_length=1)
    scope: MemoryScope
    subject_id: str = Field(min_length=1)
    content: str = Field(min_length=1)
    created_at: datetime
    expires_at: datetime | None = None


class MemoryStore:
    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.connection = sqlite3.connect(self.path)
        self.connection.row_factory = sqlite3.Row
        self.connection.execute("PRAGMA journal_mode=WAL")
        self.connection.execute(
            """
            CREATE TABLE IF NOT EXISTS memories (
                memory_id TEXT PRIMARY KEY,
                scope TEXT NOT NULL,
                subject_id TEXT NOT NULL,
                content TEXT NOT NULL,
                created_at TEXT NOT NULL,
                expires_at TEXT
            )
            """
        )
        self.connection.execute(
            "CREATE INDEX IF NOT EXISTS idx_memories_scope_subject ON memories(scope, subject_id)"
        )
        self.connection.commit()

    def close(self) -> None:
        self.connection.close()

    def put(self, record: MemoryRecord) -> None:
        if record.expires_at is not None and record.expires_at <= record.created_at:
            raise ValueError("expires_at must be later than created_at")
        try:
            self.connection.execute(
                "INSERT INTO memories(memory_id, scope, subject_id, content, created_at, expires_at) VALUES (?, ?, ?, ?, ?, ?)",
                (
                    record.memory_id,
                    record.scope.value,
                    record.subject_id,
                    record.content,
                    record.created_at.isoformat(),
                    record.expires_at.isoformat() if record.expires_at else None,
                ),
            )
            self.connection.commit()
        except sqlite3.IntegrityError as error:
            raise ValueError("memory_id must be unique") from error

    def query(
        self,
        scope: MemoryScope,
        subject_id: str,
        *,
        now: datetime | None = None,
    ) -> list[MemoryRecord]:
        current = (now or datetime.now(timezone.utc)).isoformat()
        rows = self.connection.execute(
            """
            SELECT memory_id, scope, subject_id, content, created_at, expires_at
            FROM memories
            WHERE scope = ? AND subject_id = ?
              AND (expires_at IS NULL OR expires_at > ?)
            ORDER BY created_at, memory_id
            """,
            (scope.value, subject_id, current),
        ).fetchall()
        return [MemoryRecord.model_validate(dict(row)) for row in rows]

    def delete(self, memory_id: str) -> bool:
        cursor = self.connection.execute("DELETE FROM memories WHERE memory_id = ?", (memory_id,))
        self.connection.commit()
        return cursor.rowcount == 1

    def purge_expired(self, *, now: datetime | None = None) -> int:
        current = (now or datetime.now(timezone.utc)).isoformat()
        cursor = self.connection.execute(
            "DELETE FROM memories WHERE expires_at IS NOT NULL AND expires_at <= ?",
            (current,),
        )
        self.connection.commit()
        return cursor.rowcount

真实 solution 代码: forge/__init__.py

"""HeatStack Forge teaching package."""

from .memory import MemoryRecord, MemoryScope, MemoryStore

__all__ = ["MemoryRecord", "MemoryScope", "MemoryStore"]

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

MemoryStore.query 使用 SQL 的 WHERE 子句同时过滤 scope、subject_id 和 expires_at。expires_at 为 NULL 表示永不过期，否则必须大于当前时间。

测试中插入 RUN 和 SESSION 记忆，查询 RUN 作用域时只返回 RUN 记忆，因为 SQL 过滤了 scope。

带提示练习

在 MemoryStore.query 中恢复过期过滤，使 test_expired_memory_is_hidden_and_can_be_purged 通过。

打开 forge/memory.py，找到 query 方法。
检查 SQL 查询语句，确认 WHERE 子句包含 expires_at 条件。
如果缺少，添加条件：AND (expires_at IS NULL OR expires_at > ?)。
在 execute 调用中传入 current 作为第三个参数。
运行测试验证。

逐级提示

提示1：当前时间通过 now 参数传入，如果为 None 则使用 datetime.now(timezone.utc)。
提示2：SQL 中 expires_at 存储为 ISO 格式字符串，比较时直接使用字符串比较即可。
提示3：确保参数顺序与 SQL 中的占位符一致。

完成标准

test_expired_memory_is_hidden_and_can_be_purged 通过。
query 方法返回的列表中不包含已过期记忆。

故障注入与排查

故障夹具会修改 forge/memory.py，随后 test_expired_memory_is_hidden_and_can_be_purged 会稳定失败。请先复现并保留失败输出，再用本阶段已经实现的机制定位和修复原因。

本阶段的核心是记忆边界，包括过期过滤。故障正是由于 query 方法缺少过期条件导致的。学习者已经通过 guided practice 恢复了过期过滤，因此能够识别并修复该缺陷。

已验证的故障合同

目标测试: test_expired_memory_is_hidden_and_can_be_purged
故障文件: forge/memory.py

诊断顺序

运行故障测试：python -m pytest -q ../tests/test_stage.py::test_expired_memory_is_hidden_and_can_be_purged
观察断言失败：返回列表包含 ‘expired’ 和 ‘valid’，而预期只有 ‘valid’。
检查 forge/memory.py 的 query 方法，发现 SQL 缺少 expires_at 条件。
确认修复：添加 AND (expires_at IS NULL OR expires_at > ?) 并绑定 current。

修复原则: 在 query 方法的 SQL 查询中添加过期过滤条件 AND (expires_at IS NULL OR expires_at > ?)，并在 execute 调用中传入 current 作为第三个参数。这样，查询就会排除所有已过期的记忆，只返回当前有效的记录。

python -m pytest -q ../tests/test_stage.py::test_expired_memory_is_hidden_and_can_be_purged

迁移挑战

在音乐创作场景中，你需要分别保存用户偏好和单次曲目制作记录。用户偏好属于 USER 作用域，曲目制作记录属于 RUN 作用域。

改变的条件: 场景数据从软件交付变为音乐创作，但继续使用同一个 MemoryStore API、作用域和过期规则。

创建 USER 记忆：subject_id 为 ‘user-1’，content 为 ‘prefers minor keys’，created_at 为 2026-08-17T09:00:00Z，expires_at 为 2026-08-17T09:05:00Z。再创建不过期的 RUN 记忆：subject_id 为 ‘run-1’，content 为 ‘draft saved’。以 now=2026-08-17T10:00:00Z 查询 USER/user-1 和 RUN/run-1；前者必须因过期而不可见，后者仍必须可检索。

完成标准

以 now=2026-08-17T10:00:00Z 查询 USER 作用域的 ‘user-1’ 时返回空列表，因为该记忆已在 2026-08-17T09:05:00Z 过期。
以同一个 now=2026-08-17T10:00:00Z 查询 RUN 作用域的 ‘run-1’ 时，返回 content 为 ‘draft saved’ 的不过期记忆。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么需要区分 RUN、SESSION 和 USER 作用域？

一个有力回答应包含

不同作用域对应不同的生命周期和可见范围：RUN 仅在一次运行内有效，SESSION 在一次会话内有效，USER 跨会话长期有效。
区分作用域可以避免信息泄露和污染：例如，一个运行的临时决策不应影响其他运行或用户长期偏好。
作用域也影响过期策略：RUN 记忆可能在运行结束时自动清理，而 USER 记忆可能需要显式删除或长期保留。

掌握证据

所有三个测试通过。
学习者能够解释记忆边界，包括作用域、过期和删除语义。
学习者能够在转移挑战中正确使用 MemoryStore 存储和查询不同作用域的记忆。

STAGE 06

只压缩普通对话材料，同时逐字保留指令、权限决定和承诺，并用保真检查阻止静默丢失

先看一个具体问题

压缩上下文时如何防止权限决定被静默篡改？

在长任务中，上下文窗口有限，必须压缩普通消息以节省空间。但压缩不能影响关键信息：指令、权限决定和承诺必须逐字保留，否则系统可能执行未授权的操作。

当前代码中，forge/context.py 尚未创建，测试导入失败。你需要实现 ContextPacket、CompressedContext、compress_context 和 verify_critical_fidelity，确保压缩后的关键字段与源完全一致，并且任何改动都会触发保真检查失败。

先做判断

在实现 verify_critical_fidelity 时，如果压缩后的 permission_decisions 与源不同，应该发生什么？

静默接受，因为压缩可能改变权限
抛出 ValueError，因为权限决定必须逐字保留
记录警告但继续执行
自动恢复为源权限

判断依据: 正确选项是抛出 ValueError。权限决定属于关键上下文，任何改动都意味着安全风险，必须立即失败。静默接受或自动恢复都可能掩盖篡改。

它是什么

上下文压缩保真模型

上下文压缩将普通消息总结为短文本，同时将关键字段（指令、权限决定、承诺）原样复制到压缩结果中。

关键字段通过 SHA-256 指纹绑定，验证时逐字段比较并检查指纹，确保压缩过程没有丢失或篡改关键信息。

它不是什么

不是无差别总结

压缩不是对所有内容进行摘要。关键字段不能总结，因为总结可能改变语义，例如将“write_access=revoked”总结为“write access discussed”。

压缩也不是简单的截断。普通消息可以截断，但关键字段必须完整保留，否则后续步骤无法依赖这些信息。

它与相邻概念的关系

组件关系

ContextPacket 是源数据，包含所有字段。compress_context 生成 CompressedContext，其中关键字段直接复制，普通消息被总结。

verify_critical_fidelity 接收源和压缩结果，比较关键字段并验证指纹。如果任何关键字段不匹配或指纹不一致，抛出 ValueError。

本阶段的边界决定: 关键字段（指令、权限决定、承诺）必须逐字保留并指纹化；普通消息可以压缩。验证时，任何关键字段的差异或指纹不匹配都必须导致失败。

实验 ID: state-memory-context-06-context-compression
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/state-memory-context/06-context-compression/
验收: 待验收

教师示范

目标: 演示压缩上下文并验证关键字段逐字保留，确保压缩后的指令、权限决定和承诺与源完全一致。

我们从一个 ContextPacket 开始，它包含两条指令、一条权限决定、一条承诺和 20 条普通消息。

调用 compress_context 后，关键字段被原样复制，普通消息被总结为不超过 max_summary_chars 的字符串。

然后调用 verify_critical_fidelity，它比较关键字段并检查指纹，确保压缩结果可信。

实施步骤

创建 ContextPacket 实例，包含指令、权限决定、承诺和消息。
调用 compress_context，设置 max_summary_chars=140。
检查压缩结果的 instructions、permission_decisions、commitments 与源相同。
检查 message_summary 长度不超过 140，source_message_count 为 20。
调用 verify_critical_fidelity，确认不抛出异常。

真实 solution 代码: forge/context.py

from __future__ import annotations

import hashlib
import json

from pydantic import BaseModel, ConfigDict, Field


class ContextPacket(BaseModel):
    model_config = ConfigDict(frozen=True)

    instructions: tuple[str, ...] = Field(default_factory=tuple)
    permission_decisions: tuple[str, ...] = Field(default_factory=tuple)
    commitments: tuple[str, ...] = Field(default_factory=tuple)
    messages: tuple[str, ...] = Field(default_factory=tuple)


class CompressedContext(BaseModel):
    model_config = ConfigDict(frozen=True)

    instructions: tuple[str, ...]
    permission_decisions: tuple[str, ...]
    commitments: tuple[str, ...]
    message_summary: str
    source_message_count: int = Field(ge=0)
    critical_fingerprint: str = Field(min_length=64, max_length=64)


def _critical_payload(packet: ContextPacket) -> dict[str, tuple[str, ...]]:
    return {
        "instructions": packet.instructions,
        "permission_decisions": packet.permission_decisions,
        "commitments": packet.commitments,
    }


def critical_fingerprint(packet: ContextPacket) -> str:
    encoded = json.dumps(
        _critical_payload(packet),
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def _summarize_messages(messages: tuple[str, ...], max_chars: int) -> str:
    distinct: list[str] = []
    for message in messages:
        normalized = " ".join(message.split())
        if normalized and normalized not in distinct:
            distinct.append(normalized)
    if not distinct:
        return ""
    summary = " | ".join(distinct)
    if len(summary) <= max_chars:
        return summary
    marker = " ...[compressed]"
    return summary[: max_chars - len(marker)].rstrip() + marker


def compress_context(packet: ContextPacket, *, max_summary_chars: int = 400) -> CompressedContext:
    if max_summary_chars < 80:
        raise ValueError("max_summary_chars must leave enough room for a useful summary")
    return CompressedContext(
        instructions=packet.instructions,
        permission_decisions=packet.permission_decisions,
        commitments=packet.commitments,
        message_summary=_summarize_messages(packet.messages, max_summary_chars),
        source_message_count=len(packet.messages),
        critical_fingerprint=critical_fingerprint(packet),
    )


def verify_critical_fidelity(source: ContextPacket, compressed: CompressedContext) -> None:
    if compressed.instructions != source.instructions:
        raise ValueError("compression changed instructions")
    if compressed.permission_decisions != source.permission_decisions:
        raise ValueError("compression changed permission decisions")
    if compressed.commitments != source.commitments:
        raise ValueError("compression changed commitments")
    if compressed.critical_fingerprint != critical_fingerprint(source):
        raise ValueError("critical context fingerprint mismatch")

真实 solution 代码: forge/__init__.py

"""HeatStack Forge teaching package."""

from .context import CompressedContext, ContextPacket, compress_context, verify_critical_fidelity
from .memory import MemoryRecord, MemoryScope, MemoryStore

__all__ = [
    "CompressedContext",
    "ContextPacket",
    "MemoryRecord",
    "MemoryScope",
    "MemoryStore",
    "compress_context",
    "verify_critical_fidelity",
]

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

compress_context 将关键字段直接赋值给 CompressedContext，因此它们与源完全相同。

verify_critical_fidelity 检查 instructions、permission_decisions、commitments 和指纹，全部匹配时通过。

带提示练习

在 verify_critical_fidelity 中恢复权限决定比较，使 test_changed_permission_decision_fails_fidelity_check 通过。

查看 forge/context.py 中 verify_critical_fidelity 的当前实现。
注意它缺少对 permission_decisions 的比较。
添加 if compressed.permission_decisions != source.permission_decisions: raise ValueError(“compression changed permission decisions”)。
运行测试确认通过。

逐级提示

比较 compressed.permission_decisions 和 source.permission_decisions。
错误消息应包含 ‘permission decisions’ 以匹配测试。

完成标准

test_changed_permission_decision_fails_fidelity_check 通过。
verify_critical_fidelity 在权限决定不同时抛出 ValueError。

故障注入与排查

故障夹具 fixtures/failure.json 描述了一个权限丢失场景：摘要丢弃了用户撤销写访问的决定。关键上下文必须逐字保留并在有损摘要之外进行指纹化。当前 faulty 代码中，verify_critical_fidelity 跳过了权限决定的比较，导致 test_changed_permission_decision_fails_fidelity_check 失败。可执行故障夹具修改 forge/context.py，并由 test_changed_permission_decision_fails_fidelity_check 稳定复现。已验证机制标识为 write_access、verify_critical_fidelity、compressed.permission_decisions、source.permission_decisions、permission_decisions。

本阶段已经实现了 compress_context 和 verify_critical_fidelity 的基础结构，学习者理解关键字段必须逐字保留。故障夹具中的缺陷是 verify_critical_fidelity 缺少对 permission_decisions 的检查，这正是本阶段要修复的机制。通过恢复该检查，可以确保权限决定不被静默篡改。

已验证的故障合同

目标测试: test_changed_permission_decision_fails_fidelity_check
故障文件: forge/context.py

诊断顺序

运行故障测试：python -m pytest -q ../tests/test_stage.py::test_changed_permission_decision_fails_fidelity_check
观察输出：Failed: DID NOT RAISE ValueError，说明 verify_critical_fidelity 没有抛出异常。
检查 forge/context.py 中 verify_critical_fidelity 的实现，发现缺少 permission_decisions 的比较。
确认其他关键字段（instructions、commitments）和指纹都有检查，唯独权限决定被遗漏。

修复原则: 在 verify_critical_fidelity 中添加对 permission_decisions 的逐字比较：if compressed.permission_decisions != source.permission_decisions: raise ValueError(“compression changed permission decisions”)。

python -m pytest -q ../tests/test_stage.py::test_changed_permission_decision_fails_fidelity_check

迁移挑战

软件交付场景：你正在为前端仓库添加功能，上下文包含代码审查指令和权限决定。压缩上下文后，有人篡改了权限决定，将 write_access=revoked 改为 write_access=allowed。

改变的条件: 场景数据从通用测试数据变为软件交付的具体指令和承诺，但篡改目标仍然是权限决定（permission_decisions），而不是指令。

使用 forge.context 中的 compress_context 压缩包含软件交付指令和承诺的 ContextPacket，其中 instructions 为 [“Review the frontend change before writing.”, “Keep test evidence for every modification.”]，permission_decisions 为 [“write_access=revoked”]，commitments 为 [“Do not publish partial results.”]。然后篡改压缩结果的 permission_decisions 为 [“write_access=allowed”]，并调用 verify_critical_fidelity 验证它抛出 ValueError。

完成标准

压缩结果的关键字段与源一致，且源使用指定的 instructions、permission_decisions 和 commitments。
篡改 permission_decisions 为 write_access=allowed 后 verify_critical_fidelity 抛出包含 ‘permission decisions’ 的 ValueError。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 为什么关键上下文必须逐字保留而不是总结？

一个有力回答应包含

总结可能改变语义，例如将“write_access=revoked”总结为“write access discussed”，导致系统误解权限状态。
关键上下文（指令、权限决定、承诺）直接影响系统行为，任何偏差都可能导致安全或合规问题。
逐字保留并配合指纹验证可以确保压缩过程没有丢失或篡改关键信息。

掌握证据

所有三个测试通过：test_compression_preserves_critical_context_verbatim、test_changed_permission_decision_fails_fidelity_check、test_fingerprint_prevents_silent_critical_context_replacement。
学习者能够解释为什么关键上下文必须逐字保留，以及指纹如何防止静默替换。
学习者能够在软件交付场景中应用压缩和保真检查，并识别权限篡改。

STAGE 07

用可审计事件日志重放运行，并把记忆删除同步传播到依赖它的派生摘要缓存

先看一个具体问题

删除记忆后，派生摘要缓存仍暴露已删除内容

在上一阶段，我们实现了上下文压缩，但系统仍缺少可审计的事件日志和安全的记忆删除机制。当前代码中，forge/replay.py 文件尚未创建，因此测试在导入时即失败（ModuleNotFoundError: No module named ‘forge.replay’）。

本阶段需要实现两个核心机制：一是基于哈希链的事件日志，确保日志可审计且防篡改；二是在删除记忆时，显式传播到所有依赖该记忆的派生摘要缓存，避免缓存继续暴露已删除的敏感信息。

具体而言，测试 test_deletion_propagates_to_derived_summary_cache 会创建一个用户记忆 private-note，并生成两个派生摘要：summary-1 引用了 private-note，summary-2 未引用。调用 delete_memory_everywhere 后，期望 summary-1 被失效（invalidated == 1），而 summary-2 保持不变。

当前故障夹具（fixtures/failure.json）显示，delete_memory_everywhere 在删除主存储中的记忆后直接返回，未调用 DerivedSummaryCache.invalidate_for_memory，导致测试失败。

先做判断

在实现 delete_memory_everywhere 时，如果只删除主存储中的记忆而不处理派生摘要缓存，会发生什么？

派生摘要缓存会自动更新，无需额外操作
派生摘要缓存仍保留已删除记忆的内容，导致数据泄露
系统会抛出异常，因为缓存与主存储不一致
删除操作会失败，因为缓存中存在引用

判断依据: 正确答案是第二个选项。派生摘要缓存是独立存储的，不会自动感知主存储的变化。如果不显式失效，缓存会继续暴露已删除的记忆内容，违反数据删除的完整性要求。

它是什么

事件日志与派生摘要缓存的关系

事件日志（EventJournal）是一个仅追加的、基于哈希链的审计记录。每条记录包含前一条记录的哈希（previous_hash）和当前事件的哈希（record_hash），任何篡改都会导致哈希不匹配而被检测到。

派生摘要缓存（DerivedSummaryCache）存储由多条记忆生成的摘要，并记录每条摘要引用的源记忆 ID 列表（source_memory_ids）。当删除某条记忆时，必须找到所有引用该记忆的摘要并将其删除，否则缓存会保留过时内容。

delete_memory_everywhere 函数负责协调删除操作：先删除主存储中的记忆，然后调用 summary_cache.invalidate_for_memory(memory_id) 失效所有相关摘要，并返回失效数量。

它不是什么

常见误解

事件日志不是普通的文本文件，它通过哈希链保证完整性，任何修改都会破坏链。

派生摘要缓存不会自动与主存储同步，必须显式失效。

删除记忆不是简单的单点操作，需要传播到所有依赖它的派生数据。

它与相邻概念的关系

组件协作

EventJournal 使用 RunEvent 和 reduce_events 重放状态，确保从日志恢复的状态与原始状态一致。

DerivedSummaryCache 通过 source_memory_ids 字段建立摘要与记忆的依赖关系，invalidate_for_memory 方法利用该字段查找并删除相关摘要。

delete_memory_everywhere 将 MemoryStore 和 DerivedSummaryCache 连接起来，实现跨存储的删除传播。

本阶段的边界决定: 在删除记忆时，必须同时处理主存储和所有派生缓存；仅删除主存储而忽略缓存会导致数据不一致和潜在泄露。

实验 ID: state-memory-context-07-replay-and-deletion
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/state-memory-context/07-replay-and-deletion/
验收: 待验收

教师示范

目标: 演示创建事件日志、追加事件并重放，验证状态一致。

首先，我们创建一个 EventJournal 实例，指定日志文件路径。然后，我们构造三个 RunEvent 对象，分别表示运行创建、步骤开始和写权限撤销。

通过 journal.append(event) 将事件依次追加到日志中。每次追加都会计算当前记录的哈希，并链接到前一条记录。

最后，调用 journal.replay() 读取所有事件并通过 reduce_events 重放，得到最终状态。我们验证重放状态与直接 reduce_events(events) 的结果一致。

实施步骤

创建 EventJournal 实例，指定临时文件路径。
构造三个 RunEvent 对象，序列号分别为 1、2、3。
依次调用 journal.append(event) 追加事件。
调用 journal.read_events() 读取所有事件，验证与原始事件列表相等。
调用 journal.replay() 重放，验证返回的 RunState 与 reduce_events(events) 一致。

真实 solution 代码: forge/replay.py

from __future__ import annotations

import hashlib
import json
import sqlite3
from pathlib import Path
from typing import Any

from .memory import MemoryStore
from .state.models import RunEvent, RunState, reduce_events

GENESIS_HASH = "0" * 64


def _record_hash(previous_hash: str, event_payload: dict[str, Any]) -> str:
    canonical = json.dumps(
        {"previous_hash": previous_hash, "event": event_payload},
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(canonical).hexdigest()


class EventJournal:
    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)

    def read_events(self) -> list[RunEvent]:
        if not self.path.exists():
            return []
        previous_hash = GENESIS_HASH
        events: list[RunEvent] = []
        for line_number, line in enumerate(self.path.read_text(encoding="utf-8").splitlines(), start=1):
            record = json.loads(line)
            if record.get("previous_hash") != previous_hash:
                raise ValueError(f"journal hash chain broken at line {line_number}")
            expected = _record_hash(previous_hash, record["event"])
            if record.get("record_hash") != expected:
                raise ValueError(f"journal record hash mismatch at line {line_number}")
            event = RunEvent.model_validate(record["event"])
            events.append(event)
            previous_hash = expected
        return events

    def append(self, event: RunEvent) -> str:
        events = self.read_events()
        if events:
            last = events[-1]
            if event.run_id != last.run_id:
                raise ValueError("one journal file may contain only one run_id")
            if event.sequence != last.sequence + 1:
                raise ValueError("journal events must use contiguous sequences")
            lines = self.path.read_text(encoding="utf-8").splitlines()
            previous_hash = json.loads(lines[-1])["record_hash"]
        else:
            if event.sequence != 1:
                raise ValueError("the first journal event must use sequence 1")
            previous_hash = GENESIS_HASH
        payload = event.model_dump(mode="json")
        record_hash = _record_hash(previous_hash, payload)
        record = {"previous_hash": previous_hash, "event": payload, "record_hash": record_hash}
        with self.path.open("a", encoding="utf-8", newline="\n") as handle:
            handle.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
        return record_hash

    def replay(self) -> RunState:
        return reduce_events(self.read_events())


class DerivedSummaryCache:
    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.connection = sqlite3.connect(self.path)
        self.connection.execute(
            """
            CREATE TABLE IF NOT EXISTS summaries (
                summary_id TEXT PRIMARY KEY,
                source_memory_ids TEXT NOT NULL,
                content TEXT NOT NULL
            )
            """
        )
        self.connection.commit()

    def close(self) -> None:
        self.connection.close()

    def put(self, summary_id: str, source_memory_ids: list[str], content: str) -> None:
        unique_ids = sorted(set(source_memory_ids))
        self.connection.execute(
            "INSERT OR REPLACE INTO summaries(summary_id, source_memory_ids, content) VALUES (?, ?, ?)",
            (summary_id, json.dumps(unique_ids), content),
        )
        self.connection.commit()

    def get(self, summary_id: str) -> str | None:
        row = self.connection.execute(
            "SELECT content FROM summaries WHERE summary_id = ?", (summary_id,)
        ).fetchone()
        return str(row[0]) if row else None

    def invalidate_for_memory(self, memory_id: str) -> int:
        rows = self.connection.execute("SELECT summary_id, source_memory_ids FROM summaries").fetchall()
        affected = [summary_id for summary_id, raw_ids in rows if memory_id in json.loads(raw_ids)]
        self.connection.executemany(
            "DELETE FROM summaries WHERE summary_id = ?", ((summary_id,) for summary_id in affected)
        )
        self.connection.commit()
        return len(affected)


def delete_memory_everywhere(
    memory_store: MemoryStore,
    summary_cache: DerivedSummaryCache,
    memory_id: str,
) -> tuple[bool, int]:
    deleted = memory_store.delete(memory_id)
    invalidated = summary_cache.invalidate_for_memory(memory_id)
    return deleted, invalidated

真实 solution 代码: forge/__init__.py

"""HeatStack Forge teaching package."""

from .context import CompressedContext, ContextPacket, compress_context, verify_critical_fidelity
from .memory import MemoryRecord, MemoryScope, MemoryStore
from .replay import DerivedSummaryCache, EventJournal, delete_memory_everywhere

__all__ = [
    "CompressedContext",
    "ContextPacket",
    "DerivedSummaryCache",
    "EventJournal",
    "MemoryRecord",
    "MemoryScope",
    "MemoryStore",
    "compress_context",
    "delete_memory_everywhere",
    "verify_critical_fidelity",
]

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

为什么这样实现

EventJournal.append 方法会检查事件序列的连续性，并计算哈希链。

read_events 方法在读取时验证哈希链，确保日志未被篡改。

replay 方法通过 reduce_events 将事件序列折叠为最终状态，实现确定性重放。

带提示练习

在 delete_memory_everywhere 中添加派生摘要缓存失效，使 test_deletion_propagates_to_derived_summary_cache 通过。

打开 forge/replay.py 文件，找到 delete_memory_everywhere 函数。
在函数中，调用 memory_store.delete(memory_id) 删除主存储中的记忆。
调用 summary_cache.invalidate_for_memory(memory_id) 失效所有引用该记忆的摘要，并获取失效数量。
返回 (deleted, invalidated) 元组。
运行测试验证。

逐级提示

invalidate_for_memory 方法已经实现，它会查找 source_memory_ids 中包含 memory_id 的摘要并删除。
确保返回的 invalidated 是实际失效的摘要数量。
不要忘记处理 memory_store.delete 返回 False 的情况（记忆不存在），此时 invalidated 应为 0。

完成标准

test_deletion_propagates_to_derived_summary_cache 测试通过。
delete_memory_everywhere 返回正确的 (deleted, invalidated) 元组。
summary-1 被删除，summary-2 保持不变。

故障注入与排查

故障夹具 fixtures/failure.json 模拟了一个常见缺陷：delete_memory_everywhere 在删除主存储中的记忆后，没有同步处理派生摘要缓存，导致缓存仍然保留已删除记忆的内容。测试 test_deletion_propagates_to_derived_summary_cache 因此失败，断言 invalidated == 1 但实际为 0。请根据测试输出和代码分析，诊断并修复该缺陷。可执行故障夹具修改 forge/replay.py，并由 test_deletion_propagates_to_derived_summary_cache 稳定复现。已验证机制标识为 delete_memory_everywhere、invalidate_for_memory。

本阶段已经实现了 DerivedSummaryCache.invalidate_for_memory 方法，它能够根据 source_memory_ids 找到所有引用指定记忆的摘要并删除。学习者只需在 delete_memory_everywhere 中调用该方法，即可修复故障。

已验证的故障合同

目标测试: test_deletion_propagates_to_derived_summary_cache
故障文件: forge/replay.py

诊断顺序

运行故障测试：python -m pytest -q ../tests/test_stage.py::test_deletion_propagates_to_derived_summary_cache
观察断言失败：assert invalidated == 1，实际为 0。
检查 delete_memory_everywhere 函数的实现，分析删除主存储记忆后是否遗漏了必要的传播步骤。
确认 DerivedSummaryCache 类提供了哪些方法可用于处理依赖关系。

修复原则: 在 delete_memory_everywhere 函数中，在 memory_store.delete(memory_id) 之后添加 invalidated = summary_cache.invalidate_for_memory(memory_id)，并返回 (deleted, invalidated)。

python -m pytest -q ../tests/test_stage.py::test_deletion_propagates_to_derived_summary_cache

迁移挑战

在办公研究场景中，用户要求删除一条敏感记忆（例如个人偏好），该记忆被用于生成决策简报摘要。你需要确保删除操作同时失效所有依赖该记忆的摘要缓存。

改变的条件: 场景数据从软件交付变为办公研究，记忆内容和摘要主题相应变化，但底层机制不变。

使用现有的 MemoryStore 和 DerivedSummaryCache，创建一条用户记忆（如 ‘private-note’），生成两个摘要：一个引用该记忆，另一个不引用。调用 delete_memory_everywhere 删除该记忆，验证引用它的摘要被失效，未引用的摘要保持不变。

完成标准

delete_memory_everywhere 返回 (True, 1)。
引用被删除记忆的摘要不再存在（get 返回 None）。
未引用被删除记忆的摘要内容保持不变。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

...                                                                      [100%]
3 passed in <duration>s

设计解释题: 哈希链如何检测日志篡改？为什么派生摘要缓存需要显式失效？

一个有力回答应包含

哈希链通过将每条记录的哈希链接到前一条记录，任何对历史记录的修改都会导致后续哈希不匹配，从而被检测到。
派生摘要缓存独立于主存储，不会自动感知删除操作，必须显式调用失效方法才能保证数据一致性。

掌握证据

所有三个测试通过（test_auditable_journal_replays_the_same_state、test_journal_tampering_is_detected_before_replay、test_deletion_propagates_to_derived_summary_cache）。
能够解释哈希链的防篡改原理。
能够解释删除记忆时传播到派生摘要缓存的必要性。

完成本章

让长任务可以暂停、恢复和重放，同时控制上下文增长。

FORGE / LOCAL CHECK

本地实验自检

未开始
2阅读中
3实验已下载
4测试结果已读取
5本地自检通过

verification.json 只在当前浏览器中解析，不会上传。这里验证的是实验合同，不是服务器认证或第三方背书。

选择 verification.json

主要层

依赖能力

为后续提供

受这些层约束

概念校准

记忆

上下文

可观测性

状态、会话、记忆与上下文压缩

先看一个具体问题

如何从事件序列重建当前状态？

先做判断

它是什么

状态是当前进度的投影，事件是历史事实

它不是什么

状态不是历史记录，事件也不是可变对象

它与相邻概念的关系

事件驱动状态转换

先看一个具体问题

为什么需要区分会话和运行？

先做判断

它是什么

会话是运行的容器

它不是什么

会话不是单次运行

它与相邻概念的关系

会话、运行和状态的关系

先看一个具体问题

保存的检查点可能被篡改或版本不兼容

先做判断

它是什么

检查点契约

它不是什么

检查点不是什么

它与相邻概念的关系

与其他概念的关系

先看一个具体问题

如何证明从检查点恢复的状态与完整事件重放完全一致？

先做判断

它是什么

恢复等价性验证是什么

它不是什么

恢复等价性验证不是什么

它与相邻概念的关系

与其他概念的关系

先看一个具体问题

长期记忆与运行状态分离，并确保记忆只在正确作用域和有效期内可检索

先做判断

它是什么

记忆边界模型

它不是什么

记忆边界不是什么

它与相邻概念的关系

记忆与其他概念的关系

先看一个具体问题

压缩上下文时如何防止权限决定被静默篡改？

先做判断

它是什么

上下文压缩保真模型

它不是什么

不是无差别总结

它与相邻概念的关系

组件关系

先看一个具体问题

删除记忆后，派生摘要缓存仍暴露已删除内容

先做判断

它是什么

事件日志与派生摘要缓存的关系

它不是什么

常见误解

它与相邻概念的关系

组件协作