Agent 运行时会遇到各种意外：某个专家调用的工具接口挂了、证据不够、Critic 反复不通过、某一步 LLM 突然烧了很多钱……没有一个单点故障应该导致整个 Agent 崩溃或失控。

解决思路借用军事术语——防御纵深（defense in depth）：设多道独立的闸门，就算一道被突破，后面还有下一道兜着。

代码在 packages/ai-trust-toolkit/src/ai_trust_toolkit/failsafe/。它的 __init__.py 开头就列全了这几道闸。

代码 failsafe/node_fallback.py。先看配置类 FallbackConfig（node_fallback.py:12）和包装器 wrap_with_fallback（node_fallback.py:33）的真身：

@dataclass
class FallbackConfig:                         # node_fallback.py:12
    state_field: str                          # 这个节点往哪个 state 字段写（如 'trace_finding'）
    fallback_finding: dict = field(default_factory=lambda: {
        "fallback": True,                     # ← 关键标记！Critic L1 靠它排除假证据
        "health": "broken", "confidence": "低",
        "evidence_ids": [], "hint": "节点执行失败，建议人工核实",
    })
    log_level: int = logging.WARNING

def wrap_with_fallback(node_fn, config):      # node_fallback.py:33
    async def wrapped(state: dict) -> dict:
        try:
            return await node_fn(state)       # 正常路径：原样返回
        except Exception as e:                # ← 兜住"任何"异常
            logger.log(config.log_level, "Node failed for state_field=%s: %s",
                       config.state_field, e, exc_info=True)
            return {config.state_field: dict(config.fallback_finding, error=str(e))}
    wrapped.__name__ = getattr(node_fn, "__name__", "wrapped_node")
    wrapped.__wrapped__ = node_fn             # 留原函数引用，便于测试/内省
    return wrapped

回忆 Day 05：专家工厂 make_specialist_node 内部自动给每个专家包了这道闸（specialists/factory.py:125）。所以哪怕 trace 专家接口彻底挂了，它也只返回一个 health=broken 的降级 finding，其它 3 个专家照常工作。

Day 07 见过它的三态行为，今天看它作为"失败闸门"的真身。代码 failsafe/retry_router.py：工厂 build_critic_router（retry_router.py:12）返回真正的路由函数 critic_router（retry_router.py:35）：

def build_critic_router(*, pass_dest="writeback", retry_dest="synthesizer",
                        end_dest="__end__", max_retries=2):   # retry_router.py:12
    def critic_router(state: dict) -> str:                    # retry_router.py:35
        if state.get("critique_passed"):
            return pass_dest                    # 通过 → writeback
        if state.get("retry_count", 0) >= max_retries:
            return end_dest                     # 触顶 → END（诚实收尾）
        return retry_dest                       # 没到顶 → synthesizer 重做
    critic_router.__name__ = "critic_router"
    return critic_router

should_writeback（retry_router.py:46）是写进记忆库前的最后一道门——只有"质检通过且信息充分"才准入库：

def should_writeback(state, *, require_critique_passed=True,
                     require_info_sufficient=True,
                     conclusion_field="conclusion") -> bool:   # retry_router.py:46
    if require_critique_passed and not state.get("critique_passed"):
        return False                            # 没过 Critic → 不写
    conclusion = state.get(conclusion_field) or {}
    return not (require_info_sufficient
                and not conclusion.get("info_sufficient", True))

代码 failsafe/budget_gate.py。先看异常类 BudgetExceeded（budget_gate.py:42）——它带上 scope/used/limit 三个字段，出错时一眼看清"哪个范围、花了多少、上限多少"：

class BudgetExceeded(Exception):              # budget_gate.py:42
    def __init__(self, *, scope: str, used: float, limit: float):
        self.scope = scope                    # "per_invoke" 或 "per_node"
        self.used = used
        self.limit = limit
        super().__init__(f"budget_gate scope={scope} exceeded: ${used:.4f}/${limit:.4f}")

核心是上下文管理器 budget_gate（budget_gate.py:63），用 @contextmanager 装饰，yield 前后各做一次成本快照对比：

@contextmanager
def budget_gate(*, per_invoke=None, per_node=None, on_exceed="raise"):  # budget_gate.py:63
    tracker = current_tracker()               # 从 ContextVar 取当前成本追踪器
    if tracker is None:
        yield                                 # ← 边界：没绑 tracker（测试/独立调用）静默放行
        return
    start_cost = tracker.cost_usd             # 进块时拍快照
    yield                                     # ← with 块内的代码在这里执行
    used_in_block = tracker.cost_usd - start_cost   # 出块后算差值
    if per_node is not None and used_in_block > per_node:
        _trigger("per_node", tracker, used_in_block, per_node, on_exceed)
    if per_invoke is not None and tracker.cost_usd > per_invoke:
        _trigger("per_invoke", tracker, tracker.cost_usd, per_invoke, on_exceed)

触发处理 _trigger（budget_gate.py:103）：先打 Prometheus 计数器、记 warning 日志，最后按 on_exceed 决定抛不抛：

def _trigger(scope, tracker, used, limit, on_exceed):   # budget_gate.py:103
    BUDGET_EXCEEDED_TOTAL.labels(agent=..., tenant=..., scope=scope).inc()  # 指标
    log.warning("budget_gate %s exceeded: $%.4f / $%.4f ...", scope, used, limit)
    if on_exceed == "raise":
        raise BudgetExceeded(scope=scope, used=used, limit=limit)

除了上面 6 道，还有一道正交的安全闸——敏感度分级闸门（sensitivity/，架构文档里的 DD-001 第 7 道闸）。它在请求进来时就对输入内容分级：

分类器 RuleBasedClassifier（classifier.py:81）用正则规则做，按 L2→L1→L0 优先级判定。两个红线设计值得学：

• 它绝不 import anthropic/langchain（classifier.py:10）——纯规则、极轻、无外部依赖，绝不能因为分类器本身故障而漏过机密。
• 分类结果的 reasons 只放"命中了哪条规则名"，绝不包含原始的敏感内容——避免日志二次泄漏。
• 它必须排在预算闸门之前：先拦机密输入，防止恶意 input 白白消耗预算。

还有个互补的 api/pii_scrub.py：不管字段叫什么，只要字符串里出现 email/手机号/身份证/token，落库前统一脱敏成 <redacted:类别>。

框架里跟"钱/量"相关的机制有 4 个，容易搞混，一张表理清（api/budget.py 的 docstring 有对照）：

此外 v0.6 的"成本感知路由"（cost/router.py）：预算剩余低于阈值（默认 10%）时，get_llm("auto") 自动从贵的 Sonnet 降到便宜的 Haiku——Day 11 细讲。

✋ 动手

# 1. 看 6 道闸门清单
sed -n '1,12p' packages/ai-trust-toolkit/src/ai_trust_toolkit/failsafe/__init__.py

# 2. 读节点兜底、证据开关、Critic 路由
cat packages/ai-trust-toolkit/src/ai_trust_toolkit/failsafe/node_fallback.py
sed -n '10,66p' packages/ai-trust-toolkit/src/ai_trust_toolkit/failsafe/synth_safeguard.py

# 3. 读预算硬熔断
sed -n '42,110p' packages/ai-trust-toolkit/src/ai_trust_toolkit/failsafe/budget_gate.py

# 4. 读敏感度分类器（纯规则、不碰 LLM）
sed -n '1,96p' packages/ai-trust-toolkit/src/ai_trust_toolkit/sensitivity/classifier.py

# 5. 跑失败闸门单测
uv run pytest packages/ai-trust-toolkit/tests/test_failsafe.py -q