记忆污染、并发冲突与数据隔离

学习目标

识别记忆污染、过宽 scope、陈旧更新和跨租户泄露四类风险。
组合 provenance、命名空间、CAS/锁、tombstone 和审计，建立可纠错的 Memory 边界。
读源码时沿“写入来源→版本→检索过滤→Context 注入”检查污染路径。

前置知识

已阅读长期记忆写入、检索与更新。
已理解第04章的权限、审批、幂等和审计，以及本章的 Session、Context 和 State 责任边界。

三类故障和一个共同根因

MemoryPoisoning：错误或恶意事实被长期复用

记忆污染是错误、恶意或过度推断的内容进入了不该到达的 Memory scope，并在未来 Context 中被当作可信事实。攻击者可以伪造“系统规则”，普通模型也可能把猜测写成用户偏好。

白话说，污染就是“把一张没核实的纸条放进共享档案柜，后来每个人都把它当正式文件”。

Concurrency Conflict：更新互相覆盖

两个 Run 读取同一 version，分别写入不同修正；如果 Store 只按最后写入时间覆盖，先提交的事实可能被悄悄抹掉。需要 revision/CAS、锁或可解释的合并规则。

Tenant Isolation：范围穿透

用户、项目或租户 scope 配置错误，会让检索、缓存、摘要或 Context 把别人的 Memory 带进来。只在最终 UI 隐藏内容不够，底层查询和日志都应在正确 namespace 内完成。

共同根因：来源和所有权不明确

三类问题都来自同一个边界缺失：谁能写、写入属于哪个 scope、这个版本基于什么事实、未来谁能读。安全策略必须覆盖写入、更新、检索、注入和删除全链路。

从写入到隔离的防线

候选事实先做来源和策略校验，再在 namespace 内进行带版本的提交；查询先锁定 namespace，再排序和投影。冲突或污染信号进入隔离/人工队列，不应继续喂给 Model。

阅读提示：scope 过滤不是最后一步 UI 逻辑，而是检索入口；冲突不应靠“最新写入者获胜”静默解决。Quarantine 保留证据但阻止污染继续进入 Context。

防线如何协作

Provenance 和写入策略

把 user-confirmed、tool-verified、model-inferred、external-untrusted 等来源区分开；不同来源拥有不同可写 scope。模型推断可以作为候选，但不应无确认地升级为 tenant 规则。

Namespace 和最小权限

用结构化 namespace 表达 tenant、user、project、session，而不是把它们拼成可猜的自由文本。MemoryStore、缓存键、索引和日志都要使用同一 namespace，避免查询层隔离、缓存层串租户。

CAS、锁和合并

CAS 适合检测陈旧写入：expected_version 不匹配就拒绝。锁可以保护短事务，但不能替代 scope。可合并字段要有明确 reducer；偏好、权限和安全规则通常不适合无条件自动合并。

纠正和删除

发现污染后先阻止召回，再写入纠正版本或 tombstone，最后检查缓存和 Summary 是否仍引用旧版本。删除后的审计只保留处理事实，不应把敏感原文重新复制到诊断日志。

本地模拟：拒绝污染、隔离租户和陈旧更新

用途：下面的标准库示例把认证 principal、allowed namespaces、受信任调用上下文、版本 CAS 和检索过滤放在同一 Store；调用方不能把 actor/source 填进 Memory，create 与 update 也不能互相替代。它用确定性顺序模拟并发，不启动线程，也不触碰真实数据。

python

from __future__ import annotations

from dataclasses import dataclass


_TRUST_TOKEN = object()


@dataclass(frozen=True, init=False)
class TrustedInvocation:
    actor_id: str
    provenance: str
    namespace: str

    def __init__(
        self,
        actor_id: str,
        provenance: str,
        namespace: str,
        *,
        _token: object,
    ) -> None:
        if _token is not _TRUST_TOKEN:
            raise PermissionError("invocation must come from runtime")
        object.__setattr__(self, "actor_id", actor_id)
        object.__setattr__(self, "provenance", provenance)
        object.__setattr__(self, "namespace", namespace)


@dataclass(frozen=True)
class Principal:
    actor_id: str
    allowed_namespaces: frozenset[str]
    allowed_provenances: frozenset[str]


class Runtime:
    def __init__(self, principal: Principal) -> None:
        self.principal = principal

    def context(
        self,
        *,
        requested_namespace: str,
        requested_provenance: str,
    ) -> TrustedInvocation:
        if requested_namespace not in self.principal.allowed_namespaces:
            raise PermissionError("namespace not allowed")
        if requested_provenance not in self.principal.allowed_provenances:
            raise PermissionError("provenance not allowed")
        return TrustedInvocation(
            self.principal.actor_id,
            requested_provenance,
            requested_namespace,
            _token=_TRUST_TOKEN,
        )


@dataclass(frozen=True)
class Memory:
    key: str
    value: str
    namespace: str
    actor_id: str
    provenance: str
    version: int = 0
    deleted: bool = False


class SafeMemoryStore:
    TRUSTED_PROVENANCE = {"user-confirmed", "tool-verified"}

    def __init__(self) -> None:
        self.items: dict[tuple[str, str], Memory] = {}

    def _require_trusted(self, invocation: TrustedInvocation) -> None:
        if invocation.provenance not in self.TRUSTED_PROVENANCE:
            raise PermissionError("untrusted invocation provenance")

    def _commit(
        self,
        invocation: TrustedInvocation,
        key: str,
        value: str,
        expected_version: int,
        deleted: bool = False,
    ) -> Memory:
        identity = (invocation.namespace, key)
        current = self.items.get(identity)
        actual_version = 0 if current is None else current.version
        if expected_version != actual_version:
            raise RuntimeError("stale memory version")
        record = Memory(
            key,
            value,
            invocation.namespace,
            invocation.actor_id,
            invocation.provenance,
            actual_version + 1,
            deleted,
        )
        self.items[identity] = record
        return record

    def create(
        self,
        invocation: TrustedInvocation,
        key: str,
        value: str,
    ) -> Memory:
        self._require_trusted(invocation)
        if (invocation.namespace, key) in self.items:
            raise FileExistsError("memory already exists")
        return self._commit(invocation, key, value, expected_version=0)

    def update(
        self,
        invocation: TrustedInvocation,
        key: str,
        value: str,
        *,
        expected_version: int,
    ) -> Memory:
        self._require_trusted(invocation)
        if (invocation.namespace, key) not in self.items:
            raise KeyError("memory does not exist")
        return self._commit(
            invocation,
            key,
            value,
            expected_version=expected_version,
        )

    def read(self, invocation: TrustedInvocation, key: str) -> Memory | None:
        self._require_trusted(invocation)
        memory = self.items.get((invocation.namespace, key))
        if memory is None or memory.deleted:
            return None
        return memory


principal = Principal(
    actor_id="runtime:alice",
    allowed_namespaces=frozenset({"tenant-a/user-alice"}),
    allowed_provenances=frozenset({"user-confirmed", "tool-verified"}),
)
runtime = Runtime(principal)
store = SafeMemoryStore()
alice = runtime.context(
    requested_namespace="tenant-a/user-alice",
    requested_provenance="user-confirmed",
)
first = store.create(alice, "language", "中文")
print(first.version, first.actor_id, first.provenance)
try:
    store.create(alice, "language", "English")
except FileExistsError as error:
    print(type(error).__name__, str(error))
try:
    store.update(alice, "language", "English", expected_version=0)
except RuntimeError as error:
    print(type(error).__name__, str(error))
try:
    runtime.context(
        requested_namespace="tenant-b/user-bob",
        requested_provenance="user-confirmed",
    )
except PermissionError as error:
    print(type(error).__name__, str(error))
try:
    runtime.context(
        requested_namespace="tenant-a/user-alice",
        requested_provenance="model-inferred",
    )
except PermissionError as error:
    print(type(error).__name__, str(error))
# 输出：1 runtime:alice user-confirmed
# 输出：FileExistsError memory already exists
# 输出：RuntimeError stale memory version
# 输出：PermissionError namespace not allowed
# 输出：PermissionError provenance not allowed

示例中的 Runtime 持有已认证 Principal；context 只接收请求的 namespace/provenance，并在 principal 的 allowlist 内校验，actor_id 永远从 Principal 注入，不能由调用方自填。越权 namespace 和 model-inferred provenance 在产生 TrustedInvocation 前就被拒绝；Store 从合法上下文派生 actor_id 和 provenance，Memory payload 没有可自填的来源字段。create 拒绝已存在键，update 强制 expected_version CAS。生产实现还需从认证中间件生成 Principal，并补充原子事务、缓存失效、审计和人工恢复。

事故排查顺序

先止血：暂停可疑 Memory 的召回或把它放入 quarantine，避免继续进入 Context。
确认范围：检查 tenant/user/project/session namespace、缓存键和检索过滤。
还原来源：找到 source_event、写入者、版本、审批和工具调用。
处理冲突：拒绝陈旧写入，按业务规则选择纠正、合并或 tombstone。
清理派生物：刷新缓存、Summary 和 Context 索引，验证旧版本不再注入。
回归验证：用跨租户、并发、恶意来源和删除后的检索测试覆盖边界。

源码阅读心智模型

smolagents

重点检查应用层是否把模型文本、工具结果和用户确认混作一个记忆写入入口；轻量循环没有内建隔离时，namespace、provenance 和检索过滤通常需要由调用方补齐。

OpenAI Agents SDK

先区分 RunContextWrapper.context 的本地上下文与 Session history；前者不应被当作持久 Memory 授权。再从 RunResult.to_state 和 Session 的历史协调处追踪工具结果、审批和纠正如何进入下一次 Run。SDK 没有统一的 MemoryStore，namespace、provenance 和 CAS 仍是应用责任。

LangGraph

沿 configurable.thread_id、checkpoint_ns 和 checkpoint_id 定位 StateSnapshot；检查 reducer 合并是否覆盖了来源约束，再检查 Store 的 namespace 是否与 tenant/user scope 一致。checkpoint 隔离只保护图状态，不自动保证跨线程 Store 的 Memory 隔离。

DeepSeek Harness

沿 ctx.agents/agent-loop 的 Driver 运行入口、ctx.sessions/Session.append 和 SessionEvent 查谁产生候选事实，再沿 Plugin/Hook 查扩展是否绕过策略。持久化入口 ctx.sessionPersistence 的 create/open/stat/list 与 SessionHandle 的 read/append/flush/close 是审计锚点；全局插件缓存不能代替 namespace 隔离。

易混点

模型写出的“规则”不是可信规则：来源和授权先于内容格式。
版本检查不等于租户隔离：CAS 防止陈旧覆盖，namespace 防止越权读取，两个都需要。
最新写入不等于正确写入：时间顺序不能替代来源、审批和冲突策略。
过滤 UI 不等于隔离 Store：数据可能已在排序、缓存或日志中泄露。
删除 Memory 不等于清掉所有派生引用：Summary、缓存和 Context 索引也要失效。

课后小问

为什么模型推断的“安全规则”默认不能写入 tenant scope？
答案：模型输出缺少足够的权威来源和授权，可能是提示注入或误判。
解析：可以把它保存为待审核候选，但必须阻止它直接影响其他用户的 Context 或工具权限。
CAS 已经拒绝陈旧写入，为什么还需要 namespace？
答案：CAS 只解决同一记录的并发版本，不能证明调用者有权读取或修改该记录。
解析：一个跨租户请求可以带着正确的 expected_version 越权覆盖；scope 检查负责归属，CAS 负责版本，职责不同。
污染发现后为什么先暂停召回再修复？
答案：防止错误事实继续进入 Context 和后续 Memory 写入。
解析：若先修复但召回仍命中旧缓存，系统可能在修复期间继续扩大影响。止血、清理派生缓存和验证要有明确顺序。

本节小结

记忆污染、并发冲突和跨租户泄露都要求明确来源、scope、版本和所有权。
防线应覆盖候选写入、namespace、CAS/锁、检索过滤、Context 投影、纠正删除和审计。
冲突不能静默用最后写入者覆盖；污染先隔离，再纠正/撤回并清理派生引用。
源码阅读要沿全链路追踪：谁写 Memory、写到哪个 namespace、基于哪个版本、如何进入 Context。

快速回顾

能区分 provenance、namespace、CAS 和 tombstone 的责任。
能给出一次记忆污染事故的止血与恢复顺序。
能解释为什么“正确版本”仍可能来自错误租户。
至此完成第05章；下一章进入 MCP，学习跨进程能力如何继续携带权限和生命周期边界。

记忆污染、并发冲突与数据隔离 ​

学习目标 ​

前置知识 ​

三类故障和一个共同根因 ​

MemoryPoisoning：错误或恶意事实被长期复用 ​

Concurrency Conflict：更新互相覆盖 ​

Tenant Isolation：范围穿透 ​

共同根因：来源和所有权不明确 ​

从写入到隔离的防线 ​

防线如何协作 ​

Provenance 和写入策略 ​

Namespace 和最小权限 ​

CAS、锁和合并 ​

纠正和删除 ​

本地模拟：拒绝污染、隔离租户和陈旧更新 ​

事故排查顺序 ​

源码阅读心智模型 ​

smolagents ​

OpenAI Agents SDK ​

LangGraph ​

DeepSeek Harness ​

易混点 ​

课后小问 ​

本节小结 ​

快速回顾 ​