graph TD
    A[用户意图/会议上下文] --> B{ReAct Loop}
    B --> C[Thought: 推理规划]
    C --> D[Action: 工具调用决策]
    D --> E[Execution: 统一网关调用]
    E --> F[Observation: 结果/异常]
    F --> B
    E --> G[链路追踪埋点]
    E --> H[熔断/降级判定]

{
  "trace_id": "a1b2c3d4e5f6...",
  "span_id": "span_001",
  "parent_span_id": "span_000",
  "attributes": {
    "meeting.id": "meet_20240115_001",
    "agent.role": "minutes_assistant",
    "react.step": 3,
    "react.thought": "需要调用转写服务获取全量文本",
    "tool.name": "asr.get_full_transcript",
    "tool.input.params_hash": "sha256:...",
    "tool.output.tokens": 12450,
    "tool.latency.ms": 1820,
    "error.type": "TIMEOUT",
    "error.retryable": true
  }
}

// 伪代码：工具级熔断器配置
CircuitBreakerConfig config = CircuitBreakerConfig.custom()
    .failureRateThreshold(50)                    // 失败率阈值
    .slowCallRateThreshold(80)                   // 慢调用阈值
    .slowCallDurationThreshold(Duration.ofSeconds(5))
    .waitDurationInOpenState(Duration.ofSeconds(30))
    .permittedNumberOfCallsInHalfOpenState(3)    // 半开态探测次数
    .slidingWindowType(SlidingWindowType.TIME_BASED)
    .slidingWindowSize(10)                       // 10 秒滑动窗口
    .recordExceptions(ConnectException.class, SocketTimeoutException.class)
    .build();

message ToolResult {
  bool success = 1;
  string data = 2;                 // 正常载荷
  ErrorDetail error = 3;           // 结构化错误
  bool degraded = 4;               // 是否降级数据
  map<string, string> metadata = 5; // 耗时、版本、节点IP等
}

message ErrorDetail {
  string code = 1;                 // 标准错误码
  string message = 2;              // 人类可读
  bool retryable = 3;              // 关键：指导 LLM 决策
  string suggested_action = 4;     // 建议：RETRY / CORRECT_PARAMS / SWITCH_TOOL / ABORT
}

// 扩展 W3C TraceContext，增加 Agent 协作语义
message AgentTraceContext {
  string trace_id = 1;           // 全局唯一，贯穿整个会议会话
  string parent_span_id = 2;     // 直接调用者的 SpanID
  string root_agent_id = 3;      // 入口 Agent ID（如 Supervisor）
  string current_agent_id = 4;   // 当前执行 Agent ID
  int32  collaboration_depth = 5; // 协作层级深度，防止循环调用
  map<string, string> agent_baggage = 6; // 透传业务上下文（会议ID、租户ID、权限Token）
}

# 伪代码：Supervisor 编排层熔断感知调度
async def orchestrate(dag: TaskDAG, ctx: AgentTraceContext):
    for node in dag.topological_sort():
        agent_health = circuit_breaker_aggregator.get_health(node.agent_id)
        if agent_health.state == CircuitState.OPEN:
            if node.is_critical:
                raise SessionDegradedException(f"Critical agent {node.agent_id} open")
            else:
                # 非关键节点：记录降级标记，跳过执行，使用默认值/缓存
                ctx.mark_degraded(node.agent_id, reason="circuit_open")
                dag.bypass_node(node)
                continue
        # 正常执行并更新健康度
        result = await node.execute(ctx)
        circuit_breaker_aggregator.record_result(node.agent_id, result)

-- Saga 实例表设计
CREATE TABLE saga_instance (
    saga_id           VARCHAR(64) PRIMARY KEY,      -- 格式: saga_{meeting_id}_{uuid}
    meeting_id        VARCHAR(64) NOT NULL,
    current_state     VARCHAR(32) NOT NULL,         -- RUNNING/COMPENSATING/COMPLETED/FAILED
    payload           JSONB NOT NULL,               // 当前上下文载荷（含各步骤输出）
    compensable_steps JSONB NOT NULL,               // 已执行可补偿步骤栈 [{step, tool, compensation_action, status}]
    created_at        TIMESTAMP,
    updated_at        TIMESTAMP,
    version           INT DEFAULT 0                 // 乐观锁
);

stateDiagram-v2
    [*] --> RUNNING: 启动
    RUNNING --> RUNNING: 步骤成功
    RUNNING --> COMPENSATING: 步骤失败/超时
    COMPENSATING --> COMPENSATING: 补偿步骤执行中
    COMPENSATING --> FAILED: 补偿彻底失败(需人工)
    COMPENSATING --> COMPLETED: 全部补偿成功(回滚完成)
    RUNNING --> COMPLETED: 所有步骤成功

// 注解式定义
@ToolAction(
    name = "calendar.create_event",
    idempotencyKey = "#meetingId + #params.title",
    compensation = @CompensationAction(
        tool = "calendar.delete_event",
        paramMapping = "@{eventId -> #result.event_id}" // 引用正向返回值
    )
)
public CalendarEvent createEvent(MeetingContext ctx, CalendarParams params) { ... }

func (e *SagaExecutor) compensate(ctx context.Context, saga *SagaInstance) error {
    // 逆序遍历已成功步骤
    for i := len(saga.CompensableSteps) - 1; i >= 0; i-- {
        step := saga.CompensableSteps[i]
        if step.Status == "COMPENSATED" { continue } // 幂等保护
        
        // 构造补偿参数：从正向结果反解
        compParams := e.resolveCompensationParams(step, saga.Payload)
        
        // 执行补偿（带重试、熔断、Trace）
        compCtx := trace.Inject(ctx, step.TraceContext)
        result, err := e.toolGateway.Invoke(compCtx, step.CompensationTool, compParams)
        
        // 记录补偿结果，更新状态
        saga.updateStepStatus(step.StepID, "COMPENSATED", result, err)
        if err != nil && !isRetryable(err) {
            // 非可重试错误：标记需人工介入，发送告警
            saga.markManualIntervention(step.StepID, err)
            return err
        }
    }
    return nil
}

# OPA/Rego 策略示例
package meeting.agent.tools

default allow = false

# 规则1：仅会议创建者/管理员可调用日程写入
allow {
    input.tool == "calendar.create_event"
    input.user.role in ["organizer", "admin"]
    input.meeting.tenant == input.user.tenant
}

# 规则2：敏感数据工具需脱敏权限
allow {
    input.tool == "crm.export_leads"
    input.user.has_permission("crm:export:sensitive")
    input.context.data_classification != "PUBLIC"
}

# 规则3：高风险工具需双因子确认（人工介入）
require_mfa {
    input.tool in ["finance.transfer", "hr.terminate_contract"]
    not input.context.mfa_verified
}

class TokenBudgetController:
    def __init__(self, meeting_id: str, total_budget: int = 128000):
        self.meeting_id = meeting_id
        self.total_budget = total_budget
        self.consumed = 0
        self.redis = RedisClient()

    def allocate(self, agent_id: str, estimated: int) -> bool:
        # Lua 脚本原子扣减
        script = """
        local key = KEYS[1]
        local budget = tonumber(ARGV[1])
        local estimated = tonumber(ARGV[2])
        local remaining = tonumber(redis.call('GET', key) or budget)
        if remaining >= estimated then
            redis.call('DECRBY', key, estimated)
            return 1
        end
        return 0
        """
        return self.redis.eval(script, 1, f"token_budget:{self.meeting_id}", self.total_budget, estimated)

    def refund(self, actual_used: int):
        self.redis.incrby(f"token_budget:{self.meeting_id}", actual_used)

stateDiagram-v2
    [*] --> STREAMING_ARGS: 收到 tool_calls.chunk (index=0)
    STREAMING_ARGS --> STREAMING_ARGS: 累积 arguments JSON 片段
    STREAMING_ARGS --> TOOL_EXECUTING: arguments 完整/流结束
    TOOL_EXECUTING --> TOOL_COMPLETED: 网关返回结果
    TOOL_COMPLETED --> [*]: Observation 回填上下文

# 路由规则配置 (GitOps 管理)
routing_rules:
  - condition: "task_type == 'transcript_correction'"
    model: "qwen2-7b-int4"
    fallback: "qwen2-72b"
  - condition: "task_type == 'minutes_generation' and context_tokens < 32k"
    model: "deepseek-v2-moe"
    fallback: "gpt-4o-mini"
  - condition: "task_type == 'strategic_insight' or context_tokens > 64k"
    model: "claude-3.5-sonnet"
    fallback: "gpt-4o"

# .gitlab-ci.yml 片段
chaos_validation:
  stage: validate
  image: chaos-mesh/kubectl
  script:
    - kubectl apply -f chaos/experiments/agent-tool-timeout.yaml
    - sleep 120  # 观测窗口
    - python scripts/verify_chaos_result.py 
        --experiment=agent-tool-timeout 
        --expected-circuit-break=true 
        --expected-fallback-tool="keyword_search" 
        --max-recovery-time=30s
    - kubectl delete -f chaos/experiments/agent-tool-timeout.yaml
  rules:
    - if: $CI_PIPELINE_SOURCE == "merge_request_event"