上下文累积治理(CAG):为多智能体工作流构建不降级的敏感度与约束状态机
【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit
导读
本文围绕 agent-governance-toolkit 仓库中 Context Accumulation Governance(上下文累积治理,CAG)v1 的安全设计笔记展开,讲解其核心机制:ContextEnvelope如何以不可变、单调递增的方式跟踪工作流累积的标签、敏感度与硬性限制,以及aggregation rules、obligations、context_delegation如何让"逐一检查都允许、聚合后却构成敏感上下文"的空隙被关闭。读完本文,你将掌握 CAG 的完整数据流(累积 → 聚合评估 → 决策 → 委托继承 → 审计事件),理解它为何"fail-closed",并能将这套模型复用到你自己的多智能体治理设计里。
一、为什么需要"上下文累积"治理
在传统的策略评估模型中,每个动作是独立评估的:单次工具调用、单次读取、单次导出,各自对照策略判定是否允许。这种"逐动作隔离"模型存在两个结构性空隙:
- 聚合空隙(aggregation gap):单独来看都不越界的动作,累积起来可能构成敏感上下文。例如读取 10 个字段各自都是
public,但当这些字段组合在一起形成了可识别的个人信息(PII),工作流的整体敏感度已经发生了变化。 - 委托空隙(delegation gap):父智能体携带了限制(如
no_external_export),但在委托子智能体时,约束可能在链路上悄悄丢失,子智能体"逃逸"出父级限制。
CAG 的设计目标正是关闭这两个空隙:用一个随工作流演进的"运行状态"来门控后续动作与委托,而不是孤立地评估每个动作。原安全设计笔记(docs/security/context-accumulation-governance.md)明确指出:
Adds a first cut of context accumulation governance: a
ContextEnvelopethat tracks the labels and sensitivity a workflow has accumulated and gates later actions and delegations against that running state, rather than evaluating each action in isolation.
实现上的新增模块全部位于 agent-governance-python/agent-os/src/agent_os/policies/ 下,且没有引入任何新依赖——它复用了既有的DataClassification/DataLabel/ABACPolicy类型和 policy-engine 的result_labels。
各模块职责一览
| 模块 | 核心类型 / 函数 | 职责 |
|---|---|---|
| context_envelope.py | ContextEnvelope、fold、apply_restrictions、EnvelopeReference | 不可变、带版本号的累积状态载体;跨信任边界的投影句柄 |
| context_aggregation.py | AggregationRule、AggregationRuleSet、evaluate_aggregation | 组织编写的标签组合规则 + 单调后盾(escalation backstop) |
| context_accumulation.py | accumulate、decide_next、to_policy_action、ContextOutcome | 执行后累积真实结果标签,再门控下一动作 |
| obligations.py | Obligation、ObligationSet | constrain结果携带的前置义务 |
| context_delegation.py | merge_restrictions | 委托边界上的只增限制继承 |
| context_audit.py | context_event、CONTEXT_*常量 | 信封迁移的审计事件,事件自带敏感度下限 |
二、ContextEnvelope:不可变的累积状态载体
ContextEnvelope是 CAG 的核心数据结构,定义在 context_envelope.py。它是一个 frozen dataclass,字段如下:
| 字段 | 类型 | 说明 |
|---|---|---|
envelope_id | str | 该信封谱系的稳定标识符 |
workflow_id | str | 关联键(一个工作流可跨越多次工具调用/委托) |
labels | frozenset[str] | 已累积的DataLabel类别(如pii、financial) |
aggregate_sensitivity | DataClassification | 所有已折叠敏感度的运行最大值 |
restrictions | frozenset[str] | 只增集合的限制令牌 |
version | int | 单调计数器;每次 fold/应用后version + 1 |
parent_envelope_id | Optional[str] | 子(委托)信封上设置 |
created_at | str | 调用方提供的 ISO-8601 时间戳(不进入纯逻辑代码) |
两个核心不变量,构成了整个治理模型的安全基石:
- 敏感度是 max-lattice(最大格):
aggregate_sensitivity只升不降。fold()中joined_sensitivity = max(env.aggregate_sensitivity, new_sensitivity)(context_envelope.py),即使后续输入是低敏感度,也无法把信封的分类"拉低"——这直接封堵了"敏感度在累积过程中被降低"的攻击面。 - 限制是只增集合(grow-only set):
apply_restrictions()采用并集合并(context_envelope.py),信封中已存在的限制绝不会被丢弃。
因为标签并集(join)与敏感度最大值(meet)都是交换律与幂等律满足的操作,单写入者可以以任意顺序折叠增量,最终必然收敛到同一状态——这是fold被设计为纯函数的价值所在。
跨边界投影:EnvelopeReference
跨信任边界(例如进入证据收据、跨 mesh 传递)的内容不是信封本体,而是一个不透明句柄EnvelopeReference(context_envelope.py):
- 只携带不透明的
envelope_id(作为连接键)和粗粒度sensitivity层级(作为路由提示); - 刻意省略信封内容:labels、restrictions、版本谱系、工作流关联、时间戳一律不跨边界。
这意味着进程内的ContextEnvelope形状可以自由演进,而不改变收据所承诺的内容。envelope_reference(env)是唯一被认可的投影函数,纯投影、无 I/O、无签名、无副作用。引用方(issuer)拥有保留与可解析性责任;消费者遇到过期或无法解析的引用时,将其视为一次验证结果而非收据模式失败。
三、聚合评估:组织规则 + 单调后盾
聚合评估的逻辑在 context_aggregation.py 中,它负责回答:"当前信封累积的标签组合,是否已经越过某条组织规则的敏感度门槛?"
AggregationRule 与 AggregationRuleSet
@dataclass(frozen=True) class AggregationRule: name: str all_labels: frozenset[str] sets_sensitivity: DataClassification adds_restrictions: frozenset[str] = frozenset() def __post_init__(self) -> None: if not self.all_labels: raise ValueError( f"AggregationRule {self.name!r} must have " f"a non-empty all_labels set" )一条规则的语义是:当信封中同时存在all_labels中的全部标签时,该规则将敏感度至少提升到sets_sensitivity,并追加adds_restrictions。__post_init__拒绝空标签集——因为空标签集匹配任何信封,会静默瓦解后面的 escalation 后盾。
AggregationRuleSet只是规则的有序元组,规则按声明顺序评估(context_aggregation.py)。
evaluate_aggregation:结果敏感度 = max(当前, 所有命中规则)
def evaluate_aggregation(env, ruleset, n_category_threshold): sensitivity = env.aggregate_sensitivity restrictions = set(env.restrictions) applied = [] for rule in ruleset.rules: if rule.all_labels <= env.labels: sensitivity = max(sensitivity, rule.sets_sensitivity) restrictions |= set(rule.adds_restrictions) applied.append(rule.name) escalate = not applied and len(env.labels) >= n_category_threshold return AggregationResult(...)两个关键设计点:
- 无规则增长保持原分类:如果工作流一直在增长,但从未命中任何声明的组合规则,其敏感度就维持当前的运行最大值——敏感度提升只来自"逐数据分类的运行 max"或"命中规则"。测试
test_growth_without_rule_keeps_classification验证了这一点。 - 单调后盾(monotone backstop):当没有任何规则命中且累积的不同类别标签数 ≥
n_category_threshold时,escalate = True——未被规则覆盖的组合被升级去人工审查,而不是静默通过。test_backstop_escalates_on_n_distinct_categories覆盖了该行为(test_context_aggregation.py)。
聚合(声明的标签组合越过阈值)与推断(语义推导出新的敏感度)是两回事:CAG v1 只治理前者,推断检测明确不在范围内(模块 docstring 中注明)。
四、累积与决策:先执行、再累积、后门控
context_accumulation.py 定义了 CAG 的主流程,其核心原则是:敏感度只从动作"实际产生"的标签(result_labels)累积,绝不基于尚未运行的输出投影。
accumulate:执行后的折叠
def accumulate(env, result_labels, result_sensitivity, ruleset, n_category_threshold): folded = fold(env, result_labels, result_sensitivity) agg = evaluate_aggregation(folded, ruleset, n_category_threshold) raised = replace(folded, aggregate_sensitivity=agg.aggregate_sensitivity) return apply_restrictions(raised, agg.restrictions)流程:先fold真实结果 → 再对折叠后的信封做聚合评估 → 应用评估出的敏感度提升与只增限制。测试test_accumulate_folds_result_labels与test_accumulation_never_lowers验证了折叠正确性与"绝不降低"不变量。
受限动作表:令牌 → 动作映射
_RESTRICTED_ACTIONS: dict[str, str] = { "export": "no_external_export", "delegate": "no_external_delegation", "memory_write": "no_memory_write", }这张表把"动作令牌"映射到"会门控它的限制令牌"。即:当信封中已存在no_external_export时,export动作被门控。
decide_next:双重触发 + 硬门控
def decide_next(env, action, ruleset, n_category_threshold, restricted_floor=DataClassification.RESTRICTED): agg = evaluate_aggregation(env, ruleset, n_category_threshold) if agg.escalate: return ContextDecision(ContextOutcome.ESCALATE, ..., reason="aggregation threshold crossed with no governing rule") gating = _RESTRICTED_ACTIONS.get(action) effective_restrictions = agg.restrictions # 评估后的限制集(超集) restriction_present = gating is not None and gating in effective_restrictions floor_triggered = gating is not None and agg.aggregate_sensitivity >= restricted_floor if restriction_present or floor_triggered: # 构造 ObligationSet,返回 CONSTRAIN ... return ContextDecision(ContextOutcome.ALLOW, ...)决策结果有四种 outcome:ALLOW/CONSTRAIN/DENY/ESCALATE(ContextOutcome枚举)。门控的语义是:
- 显式限制是硬门控:只要评估后的限制集中存在门控令牌(
restriction_present),无论当前聚合敏感度是否低于restricted_floor,动作都被门控。测试test_explicit_restriction_gates_below_floor专门验证"显式限制在阈值之下依然生效"。 - 敏感度下限是独立、附加的触发条件:对于
export/delegate/memory_write这类"流动型"动作,一旦聚合敏感度 ≥restricted_floor(默认RESTRICTED),即使没有显式限制也会被门控(test_floor_triggers_flow_action_without_explicit_restriction)。 - 关键实现细节:门控与义务都读取评估后的限制集
agg.restrictions而非信封原始值。因为evaluate_aggregation以env.restrictions为种子,agg.restrictions是超集——它只会门控更多,绝不少。若直接读信封,规则新增的门控令牌若未将敏感度推到下限,两个触发条件都不会生效,形成绕过(源码注释明确记录了这一修正动机)。
decide_next的完整行为由 test_context_accumulation.py 中的 10 个测试覆盖,包括:
test_next_action_gated_on_accumulated_state(基于累积状态门控下一动作)test_rule_added_restriction_gates_below_the_floor(规则新增的限制在阈值之下门控)test_unrelated_action_is_not_gated_by_someone_elses_restriction(无关动作不被他人限制误伤)test_floor_gated_decision_always_names_an_obligation(下限触发的决策必须命名义务)
五、Obligation:constrain 的承载通道与 fail-closed 语义
在 CAG 中,constrain不是一种新的策略裁决,而是"带义务的允许"(allow-with-obligations)。定义在 obligations.py:
@dataclass(frozen=True) class Obligation: key: str # 一个限制令牌 satisfied: bool @dataclass(frozen=True) class ObligationSet: obligations: tuple[Obligation, ...] = () result_labels: frozenset[str] = frozenset() @property def all_satisfied(self) -> bool: return all(o.satisfied for o in self.obligations)语义:动作仅在以下两种情况下被允许——宿主能够把伴随的限制/标签携带前进(存在义务通道),或每个义务都已声明式满足。二者皆不成立时,决策必须 fail-closed。
这里存在两个需要防御的失败模式:
- 无义务通道时 constrain 退化为 allow:声明的
PolicyEvaluation枚举没有义务通道。原设计笔记给出的缓解是:to_policy_action在缺少通道时把constrain映射为DENY。对应测试test_python_path_constrain_fails_closed。 - 空义务集合通过空真(vacuous truth)授予 allow:
ObligationSet.all_satisfied对空集合是真空满足的(all([]) == True)。缓解方案是:空义务集不满足允许条件,即空的constrain不能被当作无条件的 allow。对应测试test_empty_obligation_constrain_fails_closed。
源码中对空义务的防御体现在decide_next:当下限触发但信封没有任何已记录限制时,会主动把门控令牌本身命名为义务(if floor_triggered and gating is not None: obligation_keys.add(gating)),确保CONSTRAIN永远不会携带一个真空的义务集(test_floor_gated_decision_always_names_an_obligation)。
六、委托继承:只增并集,子级永不丢弃父级限制
委托是 CAG 的第二个核心场景。当一个工作流委托给子智能体时,子级的上下文信封必须继承父级的限制。
merge_restrictions(纯函数)
def merge_restrictions(parent: ContextEnvelope, child_declared: Iterable[str]) -> frozenset[str]: return parent.restrictions | frozenset(child_declared)在 context_delegation.py:子级的有效限制 = 父级限制 ∪ 子级声明限制。子级可以增加限制,永远不能移除任何一条从父级继承的限制。对应测试(test_context_delegation.py):
test_child_inherits_parent_restrictions(子级继承父级限制)test_child_cannot_drop_parent_restriction(子级不能丢弃父级限制)test_child_may_add_restrictions(子级可以增加限制)test_effective_restrictions_union_along_chain(沿整条委托链做并集)
DelegationChain.effective_restrictions:与既有验证器零耦合
在 agentmesh 集成侧,structural_authz_agentmesh.trust.DelegationChain新增了effective_restrictions方法(structural-authz-agentmesh/structural_authz_agentmesh/trust.py):
def effective_restrictions(self, parent_restrictions, child_declared): """Return a delegated child's effective restrictions: parent ∪ child. Grow-only restriction inheritance ... composes ALONGSIDE the scope attenuation in validate(); it does not call or modify validate() and shares none of its state.""" return frozenset(parent_restrictions) | frozenset(child_declared)设计上的一个关键红线是:不得回归既有的委托验证器。validate()的签名、返回类型、reason 字符串被测试断言为逐字节不变(test_validate_signature_and_reasons_unchanged),且既有 test_structural_authz.py 套件作为回归门继续全绿。validate()负责 scope、cycle、expiry、signature 检查(trust.py),effective_restrictions是新增的独立方法 + 纯自由函数,二者状态完全隔离。
这种"scope 衰减(validate)+ 限制继承(effective_restrictions)"的组合,形成了委托安全的两个正交维度:权限只减不增、限制只增不减。
七、审计事件:迁移本身是敏感工件
context_audit.py 为信封的每次迁移发出CONTEXT_*事件:
CONTEXT_ENVELOPE_CREATED = "CONTEXT_ENVELOPE_CREATED" CONTEXT_ENVELOPE_UPDATED = "CONTEXT_ENVELOPE_UPDATED" CONTEXT_AGGREGATION_ELEVATED = "CONTEXT_AGGREGATION_ELEVATED" CONTEXT_DELEGATED = "CONTEXT_DELEGATED" CONTEXT_REDACTED = "CONTEXT_REDACTED" DERIVED_ARTIFACT_LABELED = "DERIVED_ARTIFACT_LABELED"ContextEvent记录previous_sensitivity → new_sensitivity、labels_added、rules_applied、restrictions_added等字段。设计要点:迁移事件本身是敏感的——它点名了哪些标签与限制被累积,因此每个事件都携带自己的分类下限:classification = max(before.aggregate_sensitivity, after.aggregate_sensitivity)(context_audit.py)。事件永远不会比它所描述的数据保护得更少。测试 test_context_audit_events.py 中的test_event_carries_classification_floor验证了这一下限语义。
八、威胁模型:本控制的失败模式分析
CAG 是治理控制,因此相关的风险不是经典的注入或内存安全,而是控制是否可能 fail-open 或静默削弱既有保证。原设计笔记的纯逻辑审查(所有新文件中无子进程、反序列化、文件系统、网络或加密操作)确认没有经典漏洞面。
攻击面与缓解对照
| 风险 | 缓解 | 测试覆盖 |
|---|---|---|
| 无法承载义务的路径上 constrain 失败放开 | to_policy_action在缺少义务通道时把 constrain 映射为 DENY,fail-closed by construction | test_python_path_constrain_fails_closed |
| 空义务集通过真空真值授予 allow | 空义务不满足允许条件 | test_empty_obligation_constrain_fails_closed |
| 显式限制在敏感度阈值之下被忽略 | 存在的限制无论敏感度如何都门控其动作;下限是独立的附加触发,绝不抑制显式限制 | test_explicit_restriction_gates_below_floor |
| 累积过程中敏感度被降低 | max-lattice 连接永不下降 | test_sensitivity_is_max_lattice、test_accumulation_never_lowers |
| 委托子级丢弃父级限制 | 继承是只增并集(父级 ∪ 子级声明) | test_child_cannot_drop_parent_restriction、test_effective_restrictions_union_along_chain |
| 未被规则覆盖的组合静默通过 | 单调后盾在 n 个不同类别时升级审查 | test_backstop_escalates_on_n_distinct_categories |
validate()行为回归 | 只新增独立方法;validate 字节不变 | test_validate_signature_and_reasons_unchanged+ 既有test_structural_authz.py套件 |
明确不在 v1 范围内的内容(防过度依赖)
设计笔记明确列出以下推迟项,部署者不应过度依赖本控制的保证:
- 跨兄弟智能体、会话或独立工作流的累积(需要一个 per-principal 寄存器);
- 摄取时的标注(防止通过未标注存储或改写受限内容为新文本进行洗白);
- 检测未声明的敏感推断——这在一般情况下不可判定。设计约定:信封已敏感时产出的工件继承该分类(对应
DERIVED_ARTIFACT_LABELED事件)。
此外还有两个已形成结论但未实现的边界设计:
信封到证据收据的边界(职责契约):跨信任边界的签名且带版本的信封被推迟;在那之前,跨 mesh 的任何信封携带信息都是 advisory(对等方可能出示更弱的信封)。但边界的形状已定:引用治理上下文的收据持有不透明的envelope_id加上最多一个粗粒度、非敏感的敏感度层级或治理标签,绝不携带信封内容。跨边界工作因此是一份"职责契约"——issuer 身份、版本与单调性断言、重放与降级拒绝、验证位置——而非序列化的下游 schema。
信封生命周期与保留(开放):版本链不假设无界,且累积链本身是敏感工件(它同时是最佳审计线索和最软的目标)。v1 未敲定生命周期但留下了空间:棘轮(ratchet)是头部(head)的属性——最新版本已携带合并的标签、敏感度与只增限制,因此中间版本可以被压缩而不改变执行结果,保留完整轨迹成为审计选择而非正确性要求。降级步骤是压缩边界(跨越它会复活降级移除的内容),因此降级是任何未来剪枝的硬边界。剪枝后必须存活的最小集是活跃限制 + 解释每条限制为何成立的 reason anchor。折叠前缀的检查点需要可验证的摘要,而不只是签名(签名证明作者身份,不证明忠实性)。本版未实现任何压缩或检查点记录,这是文档化的扩展点。
九、测试覆盖与质量保证
该 PR 随附的测试覆盖(全部在 agent-os/tests/policies/ 下):
- 24 个新单元测试,覆盖:信封法则(折叠的交换律与幂等律、敏感度永不降低、限制永不丢弃)、聚合与升级后盾、fail-closed 的 constrain 映射、委托限制继承、审计事件形状。
- 既有
DelegationChain套件(test_structural_authz.py)作为回归门保持全绿,并有一条显式测试断言validate()的返回形状与 reason 字符串不变。 - 181 个既有策略测试确认新包导出不会破坏既有导入方。
- 所有测试在标准 CI 中运行,无需特殊硬件或外部服务。
信封法则的关键测试分布在 test_context_envelope.py(folding 的交换律/幂等律)、test_context_aggregation.py(规则触发、单调后盾、空标签拒绝)、test_context_delegation.py(继承与不可丢弃)、test_context_audit_events.py(事件形状与分类下限)。
十、设计要点总结与可复用模式
- 单调性是安全性的语法:敏感度用 max-lattice、限制用只增并集——"只升不降、只增不减"把"状态被悄悄削弱"从类型层面排除。
- 累积真实结果,而非投影:
accumulate在动作执行后折叠真实的result_labels,杜绝"用未发生的结果做决策"。 - fail-closed by construction:constrain 无义务通道映射为 DENY、空义务集不授予 allow、下限触发的 CONSTRAIN 必须命名义务——每个决策路径都闭合。
- 显式限制优先于阈值:下限是附加触发,绝不抑制显式限制;门控读取评估后的限制超集,避免"规则新增令牌但未推高敏感度"的绕过。
- 新代码与既有验证器零耦合:
effective_restrictions是独立方法,validate()字节不变并有测试断言。 - 审计事件自带分类下限:描述敏感数据的日志,其自身保护级别不低于数据本身。
- 边界形状先行:不透明
envelope_id+ 粗粒度敏感度层级,让进程内信封演进不影响收据承诺;跨边界是职责契约而非序列化 schema。
这套"信封 + 规则 + 义务 + 委托继承 + 审计"的模型,正是 OWASP Agentic Top 10 中"上下文中毒/敏感信息泄漏/不安全的委托"等场景的结构化回应。对希望为自主智能体构建治理层的工程师而言,agent-governance-python/agent-os/src/agent_os/policies/ 下的六个模块是一份可直接对照实现的参考骨架;原安全设计笔记(docs/security/context-accumulation-governance.md)则记录了其威胁模型与决策理由,适合作为后续审查与扩展的基线。
【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考