# 传统做法：朴素重试，生产级Agent的死亡陷阱
@retry(max_attempts=3, backoff=exponential)
def call_tool(tool_name, params):
    response = tool_client.call(tool_name, params)
    return parse_response(response)

# 问题：重试不感知上下文状态，不感知工具依赖关系，
# 不区分"可重试"和"不可重试"的失败类型

graph TB
    subgraph "七层降级阶梯"
        T1["Tier 1: 全功能调用<br/>Full Tool Execution"]
        T2["Tier 2: 替代路由<br/>Fallback Provider"]
        T3["Tier 3: 缓存命中<br/>Semantic Cache"]
        T4["Tier 4: 近似结果<br/>Approximate Result"]
        T5["Tier 5: 静态兜底<br/>Static Fallback"]
        T6["Tier 6: 延迟执行<br/>Deferred/DLQ"]
        T7["Tier 7: 人类接管<br/>Human Escalation"]
    end

    T1 -->|"超时/限流"| T2
    T2 -->|"备选不可用"| T3
    T3 -->|"缓存未命中"| T4
    T4 -->|"近似不可接受"| T5
    T5 -->|"无预置答案"| T6
    T6 -->|"SLA即将违规"| T7

    style T1 fill:#22c55e,color:#000
    style T2 fill:#84cc16,color:#000
    style T3 fill:#eab308,color:#000
    style T4 fill:#f97316,color:#000
    style T5 fill:#ef4444,color:#fff
    style T6 fill:#dc2626,color:#fff
    style T7 fill:#991b1b,color:#fff

class Tier1FullExecution:
    def __init__(self, tool_registry, health_monitor):
        self.tools = tool_registry
        self.health = health_monitor
        self.timeout_budget = 5000  # ms, 工具调用的SLA窗口

    async def execute(self, tool_call: ToolCall) -> ToolResult:
        # 健康预检：在调用前检查工具端点状态
        health_score = await self.health.check(tool_call.tool_name)

        if health_score < 0.3:
            # 工具已病入膏肓，直接跳过Tier 1，节省超时等待
            raise PreemptiveFallback(
                f"Tool {tool_call.tool_name} health={health_score:.2f}, "
                f"skip to fallback tier"
            )

        # 设置自适应超时：根据历史P95动态调整
        adaptive_timeout = self._estimate_timeout(tool_call)

        async with asyncio.timeout(adaptive_timeout):
            result = await self.tools.call(tool_call)

        # 结果验证：不是"返回了"就行，还要验证语义正确性
        validation = await self._validate_result(result, tool_call)
        if not validation.passed:
            raise SemanticDrift(validation.reason)

        return result.with_metadata(tier=1, latency_ms=adaptive_timeout)

    def _estimate_timeout(self, call: ToolCall) -> float:
        """根据工具历史P95延迟动态设置超时，而非硬编码固定值"""
        p50 = self.health.get_latency_p50(call.tool_name)
        p95 = self.health.get_latency_p95(call.tool_name)
        # 经验公式：取P95的1.5倍，但不超过SLA窗口的80%
        return min(p95 * 1.5, self.timeout_budget * 0.8)

class HierarchicalCircuitBreaker:
    """
    三层嵌套断路器：
    - Per-Tool: 单个工具粒度的熔断（如search_api挂了）
    - Per-Provider: 提供商粒度的熔断（如Google全家桶挂了）
    - Per-Capability: 能力粒度的熔断（如所有搜索工具都挂了）
    """

    def __init__(self):
        # Per-Tool: 单工具连续5次失败则熔断
        self.tool_breaker = CircuitBreaker(
            failure_threshold=5, 
            recovery_timeout=30,  # 30秒后半开探测
            expected_exception=ToolError
        )
        # Per-Provider: 提供商连续10次跨工具失败则熔断
        self.provider_breaker = CircuitBreaker(
            failure_threshold=10,
            recovery_timeout=60,
            expected_exception=ProviderError
        )
        # Per-Capability: 能力域连续3个工具都熔断则能力降级
        self.capability_breaker = CircuitBreaker(
            failure_threshold=3,
            recovery_timeout=120,
            expected_exception=CapabilityError
        )

    async def call_with_fallback(self, tool_call: ToolCall) -> ToolResult:
        providers = self._get_ordered_providers(tool_call.tool_name)
        # 按优先级排列：primary -> secondary -> tertiary

        last_error = None
        for provider in providers:
            # 逐层检查断路器状态
            if self.tool_breaker.is_open(f"{provider}:{tool_call.tool_name}"):
                continue  # 跳过已熔断的工具-提供商组合
            if self.provider_breaker.is_open(provider):
                continue  # 跳过已熔断的提供商
            if self.capability_breaker.is_open(tool_call.capability):
                break  # 整个能力域都不可用，直接降级到Tier 3+

            try:
                result = await self._execute_via_provider(
                    tool_call, provider
                )
                # 成功则重置断路器计数
                self.tool_breaker.record_success(
                    f"{provider}:{tool_call.tool_name}"
                )
                return result.with_metadata(tier=2, provider=provider)
            except ToolError as e:
                self.tool_breaker.record_failure(
                    f"{provider}:{tool_call.tool_name}"
                )
                self.provider_breaker.record_failure(provider)
                last_error = e

        raise AllProvidersExhausted(
            f"All {len(providers)} providers failed for {tool_call.tool_name}"
        ) from last_error

class SemanticCacheTier:
    def __init__(self, vector_store, similarity_threshold=0.92):
        self.cache = vector_store  # Redis + 向量索引
        self.threshold = similarity_threshold
        self.ttl_policy = AdaptiveTTL()  # 根据数据变化频率动态调整TTL

    async def lookup(self, tool_call: ToolCall) -> Optional[ToolResult]:
        # Step 1: 精确匹配缓存（最快路径）
        exact_key = self._hash_call(tool_call)
        cached = await self.cache.get(exact_key)
        if cached and not self._is_stale(cached):
            return cached.with_metadata(tier=3, cache_type="exact")

        # Step 2: 语义匹配缓存（核心价值）
        query_embedding = await self._embed_tool_call(tool_call)
        similar = await self.cache.vector_search(
            query_embedding, 
            top_k=3,
            filter={"tool_name": tool_call.tool_name}
        )

        for hit in similar:
            if hit.similarity >= self.threshold:
                # 缓存命中，但需要验证时效性
                if self._is_result_fresh_enough(hit, tool_call):
                    return hit.result.with_metadata(
                        tier=3, 
                        cache_type="semantic",
                        similarity=hit.similarity
                    )

        return None  # 缓存未命中，降级到Tier 4

    def _is_result_fresh_enough(self, cached, current_call) -> bool:
        """
        关键决策：缓存的"新鲜度"要求因工具而异
        - 股价查询：TTL < 30秒
        - 天气查询：TTL < 1小时
        - 知识库查询：TTL < 24小时
        """
        tool_freshness = self.ttl_policy.get(tool_name=current_call.tool_name)
        age = time.now() - cached.timestamp
        return age < tool_freshness.max_age

class ApproximateResultTier:
    """
    近似策略矩阵：
    ┌─────────────┬──────────────────────┬──────────────────┐
    │ 策略        │ 方法                 │ 精度保证         │
    ├─────────────┼──────────────────────┼──────────────────┤
    │ 模型降级    │ GPT-5.5→Claude Haiku │ 推理正确率↓15%  │
    │ 工具简化    │ 全文搜索→标题搜索    │ 召回率↓40%      │
    │ 数据截断    │ 100条→Top 10条       │ 覆盖率↓90%      │
    │ 时间模糊    │ 实时→过去1小时均值   │ 时效性↓3600s    │
    └─────────────┴──────────────────────┴──────────────────┘
    """

    async def get_approximate(self, tool_call: ToolCall) -> ToolResult:
        strategy = self._select_strategy(tool_call)

        if strategy == "model_downgrade":
            # 用更便宜的模型生成近似答案
            result = await self._call_with_small_model(tool_call)
            quality = QualityLabel.APPROXIMATE_MODEL

        elif strategy == "tool_simplification":
            # 用简化版工具获取部分结果
            simplified_call = self._simplify_call(tool_call)
            result = await self.tools.call(simplified_call)
            quality = QualityLabel.APPROXIMATE_TOOL

        elif strategy == "data_truncation":
            # 用已缓存的部分数据拼接近似结果
            result = await self._compose_from_fragments(tool_call)
            quality = QualityLabel.APPROXIMATE_DATA
        else:
            raise NoApproximationAvailable()

        # 关键：注入质量元数据，让Agent的推理层知道这是降级结果
        result.metadata = ResultMetadata(
            tier=4,
            quality=quality,
            accuracy_bounds=self._estimate_accuracy(result, strategy),
            caveat=f"Result generated via {strategy} strategy"
        )
        return result

# 静态兜底不是懒惰的硬编码，而是一个可运营的知识库
STATIC_FALLBACKS = {
    "weather_query": {
        "response": "当前天气服务暂时不可用。以下是本季节{city}的平均天气概况：{seasonal_data}",
        "data_source": "seasonal_averages_db",  # 离线预置的季节平均数据
        "quality": "informational_only",
        "disclaimer": "此数据为历史均值，不代表实时天气"
    },
    "stock_price": {
        "response": None,  # 金融数据不允许静态兜底，直接跳到Tier 6
        "escalate": True
    },
    "document_search": {
        "response": "搜索服务暂时不可用。以下是相关主题的热门文档：{popular_docs}",
        "data_source": "trending_docs_cache",
        "quality": "relevance_approximate"
    }
}

class AgentDeadLetterQueue:
    """
    Agent的死信队列与传统MQ的DLQ有三个关键区别：
    1. 携带完整上下文快照（不只是消息体）
    2. 支持降级重放（用简化策略重试）
    3. 有成本上限（Agent重试要花LLM推理费）
    """

    async def enqueue(self, failed_task: AgentTask, context: ExecutionContext):
        # 序列化完整上下文，而非仅任务描述
        envelope = DLQEnvelope(
            task=failed_task,
            context_snapshot=context.freeze(),  # 冻结此刻的所有变量
            failure_trace=context.get_failure_trace(),
            retry_budget=self._calculate_budget(failed_task),
            priority=self._calculate_priority(failed_task),
            deadline=failed_task.user_facing_sla - timedelta(minutes=5)
        )

        await self.dlq.put(envelope)

        # 向用户返回确认：任务已接受，将在X时间内完成
        return DeferredResponse(
            task_id=envelope.id,
            estimated_completion=envelope.deadline,
            status="queued_for_retry"
        )

    async def process_dlq(self):
        """DLQ消费者：按优先级和预算处理积压任务"""
        while batch := await self.dlq.get_batch(max_items=10):
            for envelope in batch:
                if envelope.retry_budget <= 0:
                    # 预算耗尽，升级到Tier 7
                    await self.escalate_to_human(envelope)
                    continue

                try:
                    # 使用降级策略重试
                    result = await self._retry_with_degraded_strategy(envelope)
                    await self._notify_completion(envelope, result)
                except Exception:
                    envelope.retry_budget -= 1
                    envelope.retry_count += 1
                    await self.dlq.requeue(envelope, delay=2**envelope.retry_count)

class HumanEscalation:
    async def escalate(self, envelope: DLQEnvelope):
        handoff = HumanHandoff(
            # 不是把原始任务扔给人类，而是提供决策摘要
            summary=self._generate_summary(envelope),
            # 列出系统已尝试的策略和失败原因
            attempted_tiers=envelope.attempted_tiers,
            # 为人类提供决策选项，而非开放式问题
            suggested_actions=self._suggest_actions(envelope),
            # 附带系统对每个选项的风险评估
            risk_assessment=self._assess_risks(envelope),
            # 超时自动处理（如果人类也没响应）
            auto_action=self._get_safe_default(envelope),
            timeout=timedelta(hours=2)
        )

        await self.handoff_queue.send(handoff)

graph LR
    subgraph "Saga补偿模式对比"
        A["编排式Saga<br/>Orchestration"]
        B["协同式Saga<br/>Choreography"]
        C["预计算逆操作<br/>Pre-computed Inverse"]
    end

    A -->|"中央协调器"| D["优点：全局可见<br/>缺点：单点"]
    B -->|"事件驱动"| E["优点：松耦合<br/>缺点：调试难"]
    C -->|"工具自描述"| F["优点：零编排<br/>缺点：不是所有操作可逆"]

# Microsoft agent-governance-toolkit Saga模式
from agent_governance import SagaOrchestrator, CompensatingAction

saga = SagaOrchestrator()

@saga.step(compensate="cancel_flight")
async def book_flight(params):
    booking = await flight_api.book(params)
    return booking

@saga.step(compensate="cancel_hotel")  
async def book_hotel(params):
    booking = await hotel_api.book(params)
    return booking

@saga.step(compensate="send_cancellation_email")
async def send_confirmation(params):
    await email_service.send(params)
    return {"status": "sent"}

# 执行：任何步骤失败，自动触发已完成步骤的补偿
result = await saga.execute(
    steps=[book_flight, book_hotel, send_confirmation],
    params=travel_request
)
# 如果book_hotel失败，saga自动调用cancel_flight

class CostAwareDegradationRouter:
    """
    根据请求的业务价值动态选择降级深度
    """

    def route(self, request: ToolRequest) -> DegradationPlan:
        value = self._classify_value(request)

        if value == BusinessValue.CRITICAL:
            # 金融交易、医疗决策：走完整7层降级
            return DegradationPlan(
                max_tier=7,
                max_cost_per_retry=0.50,  # 单次重试最高$0.50
                max_total_cost=5.00,      # 单任务总预算$5
                require_human_escalation=True
            )
        elif value == BusinessValue.STANDARD:
            # 一般查询：走Tier 1-5
            return DegradationPlan(
                max_tier=5,
                max_cost_per_retry=0.10,
                max_total_cost=0.50,
                require_human_escalation=False
            )
        else:  # LOW_VALUE
            # 背景任务：Tier 1-3，快速失败
            return DegradationPlan(
                max_tier=3,
                max_cost_per_retry=0.02,
                max_total_cost=0.10,
                require_human_escalation=False
            )

graph TD
    A["主工具故障"] -->|"流量涌入"| B["缓存层"]
    B -->|"缓存穿透"| C["近似计算层"]
    C -->|"算力不足"| D["静态兜底层"]
    D -->|"模板耗尽"| E["DLQ"]
    E -->|"积压"| F["人类接管"]
    F -->|"排队超限"| G["系统全面崩溃"]

    style A fill:#ef4444,color:#fff
    style G fill:#991b1b,color:#fff

class DegradationLayerBudget:
    """
    每一层都有自己的容量上限和溢出策略
    """
    BUDGETS = {
        "tier1_full":        {"max_rps": 1000, "overflow": "tier2"},
        "tier2_fallback":    {"max_rps": 800,  "overflow": "tier3"},
        "tier3_cache":       {"max_rps": 5000, "overflow": "tier5"},  # 缓存层容量大
        "tier4_approximate": {"max_rps": 500,  "overflow": "tier5"},
        "tier5_static":      {"max_rps": 10000, "overflow": "tier6"}, # 静态层几乎无限
        "tier6_dlq":         {"max_queue": 50000, "overflow": "tier7"},
        "tier7_human":       {"max_concurrent": 50, "overflow": "reject"},
    }

    def check_capacity(self, tier: str, current_load: float) -> bool:
        budget = self.BUDGETS[tier]
        if current_load >= budget["max_rps"] * 0.9:
            # 层容量接近上限，触发层内降级
            return False
        return True