DPE 一致性不变量¶

读者任何修改 tfrobot/brain/memory/ 写路径、或新增「复用 DPE schema」业务领域的工程师。

核心问题 DPE（Document / Page / Element 三层 ORM）的 Merkle 一致性是 upsert_doc 增量同步、跨次入库 LLM 0 调用、外部 ID 稳定性的地基不变量——任何打破它的写路径都会让 upsert_doc 短路退化为 _full_rebuild，让 GraphIndex 重抽全文 LLM。本文给出不变量定义 + 两条合法 caller 路径的判别规则 + 维护它们的实现纪律。

设计理念¶

DPE 的全部价值来自一条增量同步通路：相同 file_uri 重新入库未变文档时 0 次 LLM 调用（PRD docs/refactor/dpe_incremental_sync.md §4 G1）。这条通路的实现机制是三层内容 hash 构成的 Merkle 树：

Document.doc_hash  =  hash(file_uri, file_type, doc_metadata,
                            [page_hash for page in pages.sorted_by_number])
Page.page_hash     =  hash(title, [ele.content_hash for ele in elements.sorted_by_seq_in_page])
Element.content_hash = hash(text-or-image-payload)

upsert_doc 的短路逻辑是：DB 里 stored 的 doc_hash == new_doc.doc_hash → status="unchanged" 直接返回；不等则按 page_hash 第一段 LCS 找出"稳定 page"O(1) 整页跳过。这两个短路成立的前提是「DB 里 stored 的 hash 永远反映 DB 里实际存放的内容」——这就是 DPE 一致性不变量。

如果 DB 里 stored 的 doc_hash 跟实际 page 序列对不上（比如某条业务路径加了 element 但没上溯重算），下一次 upsert_doc 就会撞两种场景之一：

新文档巧合算出同样的 doc_hash → 实际有差异但 short-circuit 返回 unchanged → silent 数据漂移
算出不同 doc_hash → 但 page_hash 短路也失效（因为 page_hash 也 stale）→ 退化到全文 LCS → GraphIndex 重抽 LLM 全文

两种结果都把 DPE 增量同步的全部收益清零。所以本不变量在工程上是硬护栏——任何修改 SqlDocElement / SqlDocPage / SqlDocument 的写路径都必须满足它。

核心概念¶

概念	定义	算自哪里
`content_hash`	element 的内容指纹（不含位置）	`HashStrategy.element_hash(ele)`，loader 出口 `attach()` 时算好
`page_hash`	聚合本 page 内 elements 的 content_hash 序列 + page.title	`HashStrategy.page_hash(page)`
`doc_hash`	聚合本 doc 内 pages 的 page_hash 序列 + 文档元信息	`HashStrategy.doc_hash(doc)`
`hash_strategy_uri`	算法身份指纹（`hash-strategy://default?algo_version=v1`）	strategy 实例的 URI 序列化
DPE 一致性不变量	DB 里 stored 的 `page_hash` / `doc_hash` 任意时刻反映 DB 里实际 elements / pages 的内容	——
业务局部更新 caller	持有「element + 父链 ID」就足够，没有完整 doc snapshot	chat `add_msg` / 改 `conversation.title`
文档级 reconcile caller	持有完整 `Document` Pydantic 树	loader 重新入库

不变量的精确陈述（PRD §5.3.2.6）：

在不持有外部事务的任意时刻 t，对 DB 里任意 doc_id / page_id，下式都为真：

stored_page_hash(page_id) == HashStrategy(stored_strategy_uri).page_hash(stored_page_at_t)

stored_doc_hash(doc_id) == HashStrategy(stored_strategy_uri).doc_hash(stored_doc_at_t)

例外只有「外部事务中间态」——caller 持外部 session 顺序调多个原语时允许中间步骤 stale，事务收尾必须重算上溯。

两类 caller 的合法路径¶

DPE 写路径有且只有两种合法形态。所有新功能必须先归类到其中一种——任何「我想再造第三种」的冲动都意味着没读 D20-extended（PRD §5.3.2.6）。

模式 A：文档级 reconcile —— `upsert_doc`¶

适用场景：调用方持有完整 Document Pydantic 树（pages + elements 全列），表达"这就是文档最新全貌，请把 DB 调成这个状态"的语义。

典型 caller：

Loader 重新入库（PRD / OCR / 网页抓取）
测试 fixture 装配整文档
数据迁移工具

正确入口：

from tfrobot.schema.document.base import Document
from tfrobot.schema.document.hasher import DefaultHashStrategy

# loader 出口 attach 三层 hash + seq_in_page
doc: Document = ...  # loader 产物
DefaultHashStrategy().attach(doc)  # 装配 content_hash / page_hash / doc_hash + seq_in_page

# 调 upsert_doc，由它内部 LCS reconcile
result = unified_memory.upsert_doc(doc)
# result.status ∈ {"unchanged", "added", "lcs_reconciled", "full_rebuild"}
# result.elements_added / elements_deleted / elements_synced 计数完整

为什么必须走 upsert_doc：

idempotent：同 file_uri 二次调用走 reconcile 而非撞 PG UNIQUE(file_uri)——PRD §3 P1 痛点
批量摊薄：upsert_doc 内部走 _xxx_with_doc_snapshot_in_session 私有原语 + 收尾一次性 _update_doc_with_doc_snapshot_in_session 重算，避免 N 次原语 × N 次上溯重算的 O(M²) 中间态浪费
跨 strategy 安全：hash_strategy_uri 不一致时自动 _full_rebuild，业务代码不需要懂跨版本兼容

禁止做的事：

拿到 Document 后自己拆 page/element 调 add_element / add_page 循环写入——这把 LCS 收益清零，每次 add_element 都触发上溯重算（O(M²) 中间态浪费），且非 LCS EQUAL 路径 GraphIndex 还会跑完整 LLM 抽取
走废弃的 add_doc / delete_doc + add_doc 组合——这两个方法已发 DeprecationWarning 委托给 upsert_doc，新代码不要碰

实施期纪律：

Document 进入 upsert_doc 前必须已经 HashStrategy.attach() 过；重复 attach 会触发 D20 §5.3.2.2「DB 是权威」违反
upsert_doc 全程包裹在文档级 advisory lock + 单事务里——外层 caller 不要自己再开 session

模式 B：业务局部更新 —— `add_element` / `add_page` / `update_page` / `delete_element` 等 patch 入口¶

适用场景：调用方没有也不需要持有完整 Document snapshot，只持有当前 element + 父链 ID（page_id / doc_id）或单字段。

典型 caller：

TFChat PGBufferStore.add_msg（chat 加新消息——没有也不该 reverse-load 整 doc）
TFChat UnifiedMemory.add_conversation / update_conversation（增 / 改一个 conversation page）
TFChat delete_element（删一条消息）

正确入口：

# 加消息：走 PGBufferStore.add_msg（业务转换层）
buffer_store.add_msg(msg)  # 内部：message_to_sql_element + add_element_in_session 原语 + propagate_hashes_in_session 上溯

# 改 conversation 标题
unified_memory.update_conversation(conversation_id, name="新标题")

# 单独加 element / page（不通过 buffer / conversation manager 时）
ele_id = unified_store.add_element(sql_doc_element)
page_id = unified_store.add_page(sql_doc_page)

这条路径的 invariant 维护契约：

每个 public add_xxx / update_xxx / delete_xxx 在 commit 前自维护父 hash：

调 Layer 1 数据原语（add_element_in_session / delete_page_in_session / ...）做实际 PG 写 + index hooks 派发
调 propagate_hashes_in_session(session, page_id)——同事务内先 UPDATE page_hash 再读 page_hash 序列重算 doc_hash
session.commit()

# UnifiedStore.add_element 简化骨架，展示模式 B 的标准两步
def add_element(self, element):
    with self.session_factory() as session:
        try:
            ele_id = add_element_in_session(session, element, indexes=self.indexes)
            propagate_hashes_in_session(session, element.page_id)  # ← 单调用 caller 必须紧跟上溯
            session.commit()
            return ele_id
        except Exception:
            session.rollback()
            raise

为什么不强迫这类 caller 走 upsert_doc：

D20 原契约（PRD §5.3.2.2）写作时漏掉了这一类 caller——按当时的 caller 集合枚举，所有 DPE 写路径都通过 upsert_doc。TFChat 数据复用 DPE schema 后这个假设破裂：chat add_msg 每次都 reverse-load 整 conversation 对应的 doc snapshot 然后调 upsert_doc 是范围错配——chat 没有也不该有"整文档级 reconcile"语义，业务模型也没有 doc snapshot 可拿。

D20-extended（PRD §5.3.2.6）的精确化是：复杂签名护栏作用域收窄到 _xxx_with_doc_snapshot_in_session 私有原语；public API 通过「自维护父 hash」实现等价护栏。两条 caller 路径都合法。

禁止做的事：

不要绕过 public API 直接拼 SQLAlchemy ORM 写 PG——session.add(SqlDocElement(...)) + session.commit() 一行会留下 stale page_hash，下一次 upsert_doc 短路返回 unchanged 把这次写入"吞掉"
不要在 hot loop 里循环调 update_page——每次都走 selectinload 全树拉 + 两表 merge（PRD §5.3.2.6 fix-review D1），O(N) 行变成 O(N²) PG round-trip。批量改走 upsert_doc
不要忘记 propagate_hashes_in_session——只调数据原语 add_element_in_session 不上溯，等于把 invariant 破了一半

三层架构与原语层职责¶

Layer 0 —— ORM schema（tfrobot/schema/document/models/）：SqlDocument / SqlDocPage / SqlDocElement。新加业务字段需评估是否进 hash 输入（多数否，详见 §5.3.2.4 sync_xxx 白名单）。

Layer 1 —— DPE 一致性原语（tfrobot/brain/memory/store/dpe_primitives.py，function 级、无状态）：

原语类别	函数	职责
数据原语	`add_element_in_session` / `add_page_in_session` / `delete_element_in_session` / `delete_page_in_session` / `delete_doc_in_session`	PG 写 + index hooks 派发；不维护父 hash（caller 决定）
Hash 重算原语	`recompute_page_hash_in_session` / `recompute_doc_hash_in_session`	走精确 SELECT + dataclass stub 避开 SQLAlchemy 关系链 lazy-load O(N) 退化（详 TFROB-430 落地经验）
上溯原语	`propagate_hashes_in_session(session, page_id)`	先 UPDATE `page_hash` 再读 page_hash 序列重算 `doc_hash`——次序不可换，否则 doc_hash 输入 stale
seq 分配原语	`next_seq_in_page`	`FOR UPDATE` 锁 page row + 读 `MAX(seq_in_page)`，保证并发 INSERT 唯一性
编排原语	`run_with_index_hooks_post_write_in_session` / `run_with_index_hooks_pre_write_in_session`	派发 index hooks + 失败补偿；delete 路径必须用 pre-write（hook 需要在 row 还在时读 parent 链）
补偿器	`Compensator` / `AsyncCompensator` / `walk_compensators`	LCS 跨原语编排失败时反向走累计补偿；frozen dataclass，不要 setattr 走 side-table

关键设计点：

原语全部 _in_session 语义——不开 session、不 commit；事务由 caller 持有
原语 docstring 显式写「调用方契约：单调用必须紧跟 propagate_hashes_in_session 否则违反 Merkle 一致性契约」
indexes 参数显式接受（None 表示无 hook）——4 个 Store 平等调用，避免引入新的 Store 抽象层

Layer 2 —— 4 个 DPE-aware Store（互不依赖、平等调用 Layer 1）：

Store	业务概念	调原语姿势
`BaseUnifiedStore`	doc / page / element CRUD + index hooks（embedding / keyword / graph / conversation）	`add_element_in_session(..., indexes=self.indexes)` + `propagate_hashes_in_session`
`doc_store.PostgresDocumentStore`	纯 doc / page CRUD（无业务 index）	`add_page_in_session(..., indexes=None)`
`PGBufferStore`	TFChat 消息（element）	`add_element_in_session(..., indexes=[ConversationIndex()])`
`PGBaseConversationManager`	TFChat 会话（page）	`add_page_in_session(..., indexes=[ConversationIndex()])`

对称纪律（PRD §5.3.2.6.4）：

层	`_xxx_in_session`（私有）	`xxx`（public）
element	不维护父 hash（LCS 编排末尾一次重算）	维护 page_hash + doc_hash
page	不维护父 hash	维护 doc_hash
doc	——	进入即 `doc.doc_hash = strategy.doc_hash(doc)` 重算（D20「传入 strategy 总是重算」）

Layer 3 —— 业务路径（UnifiedMemory wrapper / chat 业务转换层）：

只做业务概念到 DPE 映射：UserAndAssMsg → SqlDocElement / conversation_id → page_id / platform_id → doc_id
直接调 Layer 2 public API 即可合规——不直接维护 hash

写路径决策流程¶

新增任何修改 DPE schema 的代码前，按下表决策：

是否持有完整 Document Pydantic 树（pages + elements 全列）？
├─ 是 → upsert_doc(doc)（模式 A）
└─ 否 → 是否只持有 element / page 字段 + 父链 ID？
        ├─ 是 → public API（add_element / update_page / delete_element 等）（模式 B）
        └─ 否 → 你大概率没想清楚——回去重读 PRD §5.3.2.6

反模式判定（看到下列代码立刻 review 阻塞）：

# ❌ 反模式 1：reverse-load doc 后调 upsert_doc 只为加一条消息
def add_msg_wrong(buffer_store, msg):
    doc = unified_memory.get_doc(buffer_store.doc_id)
    doc.pages[some_idx].elements.append(msg_to_element(msg))
    DefaultHashStrategy().attach(doc)
    unified_memory.upsert_doc(doc)  # 范围错配：N 条消息 × O(N+M) reconcile = O(N²)

# ❌ 反模式 2：自己拆 doc 循环调 add_element
def import_doc_wrong(unified_store, doc):
    for page in doc.pages:
        for ele in page.elements:
            unified_store.add_element(ele_to_orm(ele))
    # LCS 收益清零，每条 add_element 都触发上溯，O(M²) 中间态浪费
    # 且 page_hash / doc_hash 跟 strategy 算的对不上（attach 算法 vs 加法重算路径不同）

# ❌ 反模式 3：直接拼 ORM 写 PG，绕过 Layer 1 / Layer 2
def add_msg_bypass(session, ele_orm):
    session.add(ele_orm)
    session.commit()
    # page_hash / doc_hash 完全不更新——下次 upsert_doc 看到 stale 短路返回 unchanged

正路：

# ✅ 模式 A：loader 重新入库走 upsert_doc
def import_doc_right(unified_memory, doc):
    DefaultHashStrategy().attach(doc)
    return unified_memory.upsert_doc(doc)

# ✅ 模式 B：chat 加消息走 buffer_store.add_msg
def add_msg_right(buffer_store, msg):
    buffer_store.add_msg(msg)  # 内部 add_element_in_session + propagate_hashes_in_session

# ✅ 自定义 Store 加 element 时调原语 + 上溯
def custom_add_element(session_factory, element, indexes):
    with session_factory() as session:
        try:
            ele_id = add_element_in_session(session, element, indexes=indexes)
            propagate_hashes_in_session(session, element.page_id)
            session.commit()
            return ele_id
        except Exception:
            session.rollback()
            raise

运行时行为细节¶

post-write vs pre-write 派发顺序¶

add_xxx / update_xxx 用 post-write 派发（先写 PG 再调 index hook）——hook 观察"已写入"的 entity，可读 PG 自动生成的主键 / FK 状态。

delete_xxx 用 pre-write 派发（先调 index hook 再 session.delete）——hook 需要在 element 还在 DB 时读 page_id / parent_id 链以维护 entrance_ele_id 回退。

历史教训（TFROB-432 [C-3]）：早期 _delete_element_in_session 用 post-write，依赖 ConversationIndex.delete_element 长期 no-op 掩盖了这个 bug；C-3 实施 entrance_ele_id 多跳回退时暴露——必须把派发顺序矫正为 pre-write。新加 delete 类 hook 时务必走 run_with_index_hooks_pre_write_in_session。

`propagate_hashes_in_session` 次序¶

实现内部步骤：

new_page_hash = recompute_page_hash_in_session(session, page_id)
UPDATE doc_pages SET page_hash = :new_page_hash WHERE page_id = :page_id
session.flush()
doc_id = lookup_doc_id_in_session(session, page_id)
new_doc_hash = recompute_doc_hash_in_session(session, doc_id)（重新读 page_hash 序列——必须读到步骤 2 写入的新值）
UPDATE documents SET doc_hash = :new_doc_hash WHERE doc_id = :doc_id

次序不可换：先重算 doc_hash 再 UPDATE page_hash 会让 doc_hash 输入读到旧 page_hash 序列——上溯 stale，invariant 破。

LCS 编排为什么不调 `propagate_hashes_in_session`¶

upsert_doc._dispatch_lcs_in_session 走 N 次 add_element_in_session / delete_element_in_session 后由收尾 _update_doc_with_doc_snapshot_in_session 一次重算 doc_hash + 写 page_hash 全集——避免 O(M²) 中间态浪费。如果 LCS 编排里也调 propagate_hashes_in_session，那 N 次操作 × N 次上溯 = O(N²) 行扫描，文档越大塌得越狠。

这是「数据原语不主动维护父 hash」的根本原因（PRD §5.3.2.6.4 MUST 16）——它给 LCS 留出"批量摊薄"的窗口。单调用 caller（模式 B）紧跟一次 propagate_hashes_in_session 不付此代价。

Hash 重算的精确 SELECT 纪律¶

recompute_page_hash_in_session / recompute_doc_hash_in_session 用 dataclass stub 满足 HashStrategy duck type 契约，禁止用 Document.model_validate(sql_doc, from_attributes=True)——后者会触发 SQLAlchemy from_attributes lazy-load doc.pages.elements cascade，doc_hash 重算从 O(M_pages) 退化成 O(N_total_elements) 全表扫（Phase B benchmark 排查到的 artifact，详见 tests/benchmarks/buffer_dpe_invariants/reports/decision_rationale.md §4.2）。

新加 hash 重算辅助函数时必须沿用 dataclass stub 模式。

`ConversationIndex` 的注入语义¶

ConversationIndex 通过 __init_subclass__ 自动注册始终在 UnifiedStore.indexes 中，但 add_element / delete_element hook 内自查 element.page.doc.file_type == TFCHAT——非 TFCHAT 文档自动 no-op。这一选择保证 chat 路径在 unified_store / buffer_store 两条路径下都正确维护 entrance_ele_id，避免「某一条路径漏维护」的回归风险。

配置参数¶

DPE 不变量本身没有可配置项——它是 invariant 而非 feature。但下列与 invariant 相关的算法身份配置需要注意：

参数	类型	默认值	说明
`Document.hash_strategy_uri`	`str` (URI)	`hash-strategy://default?algo_version=v1`	strategy 身份；不一致时 `upsert_doc` 走 `_full_rebuild`（PRD §5.4.2 短路 1）
`DefaultHashStrategy.algo_version`	`str`	`"v1"`	算法版本；升级时 bump，跨 version doc_hash 不可比对
`DefaultHashStrategy.diff_mode`	`Literal["lcs", "full_rebuild_on_change"]`	`"lcs"`	image-source 等不可信 element diff 的 strategy 选 `full_rebuild_on_change`

修改任何 hash 输入字段（如新增 SqlDocElement.foo）必须同步评估是否进 element_hash / page_hash / doc_hash——多数业务字段不应进 hash（属 sync_element 白名单，PRD §5.3.2.4）。

与其他模块的协作¶

UnifiedStore.indexes：所有 index hook 透过 Layer 1 编排原语（run_with_index_hooks_xxx）派发，失败补偿走 Compensator 反向 walk
GraphIndex：DPE 不变量的最大受益方——upsert_doc 短路命中时跳过 GraphIndex 全文 LLM 重抽（每 element 两次 LLM 调用）
HashStrategy（tfrobot/schema/document/hasher.py）：所有 hash 计算唯一权威；新加文件类型走 __register_name__ + __init_subclass__ 自动注册，禁止在 caller 侧写 if file_type == ... 分支
PRD docs/refactor/dpe_incremental_sync.md：本文是 PRD §5.3.2.6 落地纪律的工程师向解读；细节决策记录见 PRD 附录 A.2 D20 / D14 / D16

历史教训¶

D20 原契约（PRD §5.3.2.2）写作时只 grep 了 update_xxx 的 caller（只有 upsert_doc），就推断 add/delete 也只有 upsert_doc 一类 caller——这是 caller 集合枚举的范围决策错误。TFChat 数据接入 DPE schema 这一场景在 D20 设计期还不存在，新场景出现后未触发 D20 假设的 review，导致 TFROB-417 / TFROB-429 / TFROB-430 直到撞 NOT NULL 才暴露问题。

教训凝结（PRD §5.3.2.6.6）：

caller 集合枚举要把"未来可能新增的 caller 形态"显式列出来——本文 §「两类 caller 的合法路径」即此承诺
每次有新 caller 接入 DPE schema 时（如未来其他领域复用 D/P/E），必须 review §5.3.2 全章假设是否仍成立——这一审视应作为「DPE schema 复用 PR」的强制 checklist 项
架构合理性 ≠ 安全冗余——D20 原契约的复杂签名护栏是为了防止业务代码无意中破坏 invariant，不是为了让业务代码受罪。当业务代码的合理需求被护栏挡住时，正确反应是给护栏开门（public API 自维护），而不是让业务代码爬窗户（绕过护栏 / 写补丁）

新接入 DPE schema 的 PR checklist

[ ] 这条 caller 路径属于模式 A（持完整 Document）还是模式 B（局部更新）？
[ ] 模式 A：是否走 upsert_doc 入口？是否在 loader 出口调过 HashStrategy.attach()？
[ ] 模式 B：是否走 Layer 2 public API？public API 内部是否调了 propagate_hashes_in_session 上溯？
[ ] 若新加自定义 Store：是否平等调用 Layer 1 原语 + 显式注入 indexes 参数？
[ ] 若新加 hash 输入字段：strategy attach / page_hash / doc_hash / element_hash / sync_element 白名单是否同步更新？
[ ] 是否补充了 buffer_dpe_invariants benchmark 覆盖新路径？