确认点触发条件设计：手动触发、自动触发、基于超时阈值的触发逻辑

1. 触发条件原理概述

确认点的触发条件是整个中断确认机制的核心决策层。根据触发来源与判定方式，可划分为三种独立模式：手动触发依赖外部交互信号；自动触发由内部事件或条件匹配驱动；超时阈值触发基于时间度量自动激活。三种模式可共存于同一个确认点实例中，通过优先级仲裁决定最终执行路径。

2. 触发条件判断流程

下图展示了确认点从运行状态监控到最终回调执行的完整判断路径，三种触发条件在类型判断节点后并行等待，任一条件满足即进入执行阶段。

3. 手动触发实现

手动触发通过外部显式调用来激活确认点，适用于需要人工干预或第三方系统协同的场景。实现上提供同步与异步两种调用入口，并支持携带上下文原因。

class ConfirmPoint:
    """确认点核心类 - 手动触发实现"""
    
    def __init__(self, name: str, callback=None):
        self.name = name
        self.callback = callback
        self._triggered = False
        self._trigger_type = None
        self._trigger_meta = {}
    
    def trigger_manual(self, reason: str = "", operator: str = "system") -> bool:
        """
        手动触发确认点。
        
        参数:
            reason:  触发原因描述
            operator: 操作者标识 (user_id / service_name)
        
        返回:
            True 表示触发成功并执行回调
        """
        if self._triggered:
            print(f"[确认点:{self.name}] 已被触发，忽略重复手动请求")
            return False
        
        self._triggered = True
        self._trigger_type = "manual"
        self._trigger_meta = {
            "reason": reason,
            "operator": operator,
            "timestamp": time.time()
        }
        
        print(f"[确认点:{self.name}] 手动触发 | 操作者:{operator} | 原因:{reason}")
        
        if self.callback:
            self.callback(self)
        return True
    
    def trigger_manual_async(self, reason: str = "", operator: str = "system") -> asyncio.Task:
        """异步触发手动确认点，不阻塞调用方"""
        return asyncio.create_task(
            self._async_trigger_manual(reason, operator)
        )
    
    async def _async_trigger_manual(self, reason: str, operator: str):
        """异步触发内部实现"""
        if self._triggered:
            return False
        self._triggered = True
        self._trigger_type = "manual"
        self._trigger_meta = {
            "reason": reason,
            "operator": operator,
            "timestamp": time.time()
        }
        if self.callback:
            if asyncio.iscoroutinefunction(self.callback):
                await self.callback(self)
            else:
                self.callback(self)
        return True

@app.post("/agent/confirm-point/{name}/trigger")
async def trigger_confirm_point(
    name: str,
    reason: str = Body("", description="触发原因"),
    operator: str = Body("api-user", description="操作者标识")
):
    agent = get_current_agent()
    confirm_point = agent.get_confirm_point(name)
    if not confirm_point:
        raise HTTPException(status_code=404, detail="确认点不存在")
    
    success = confirm_point.trigger_manual(reason=reason, operator=operator)
    return {
        "status": "ok" if success else "ignored",
        "trigger_type": "manual",
        "confirm_point": name
    }

4. 自动触发实现

自动触发基于Agent内部事件总线或状态机变化进行条件匹配。当注册的事件类型到达且满足预设条件时，确认点自动激活，无需外部干预。

class AutoTriggerRule:
    """自动触发规则定义"""
    
    def __init__(self, event_type: str, condition: callable = None):
        self.event_type = event_type
        self.condition = condition  # callable(event_data) -> bool
        self.priority = 0

class ConfirmPointAuto(ConfirmPoint):
    """支持自动触发的确认点"""
    
    def __init__(self, name: str, callback=None):
        super().__init__(name, callback)
        self._auto_rules: list[AutoTriggerRule] = []
        self._event_bus = None
    
    def register_auto_rule(self, rule: AutoTriggerRule):
        """注册自动触发规则"""
        self._auto_rules.append(rule)
        self._auto_rules.sort(key=lambda r: r.priority, reverse=True)
    
    def bind_event_bus(self, event_bus):
        """绑定事件总线，开始监听"""
        self._event_bus = event_bus
        for rule in self._auto_rules:
            event_bus.subscribe(rule.event_type, self._on_event)
    
    def _on_event(self, event_data: dict):
        """事件回调：条件匹配后自动触发"""
        if self._triggered:
            return
        
        event_type = event_data.get("type")
        matched_rules = [
            r for r in self._auto_rules
            if r.event_type == event_type
        ]
        
        for rule in matched_rules:
            if rule.condition and not rule.condition(event_data):
                continue
            
            # 条件满足，自动触发
            self._triggered = True
            self._trigger_type = "auto"
            self._trigger_meta = {
                "event_type": event_type,
                "event_data": event_data,
                "timestamp": time.time()
            }
            
            print(f"[确认点:{self.name}] 自动触发 | 事件:{event_type}")
            
            if self.callback:
                self.callback(self)
            break

# 使用示例
def task_completed_condition(data):
    return data.get("status") == "success" and data.get("exit_code") == 0

rule = AutoTriggerRule(
    event_type="task.completed",
    condition=task_completed_condition
)

cp = ConfirmPointAuto("task-check", callback=my_callback)
cp.register_auto_rule(rule)
cp.bind_event_bus(agent.event_bus)

5. 超时阈值触发实现

超时阈值触发是确认点的保底机制，当在指定时间窗口内既无手动触发也无自动触发时，由定时器自动激活。实现上需要精确的时间度量、可配置的阈值参数以及重试策略。

class ConfirmPointTimeout(ConfirmPoint):
    """支持超时阈值触发的确认点"""
    
    def __init__(self, name: str, callback=None):
        super().__init__(name, callback)
        self._timeout_threshold: float = 30.0       # 超时阈值(秒)
        self._retry_count: int = 0                   # 重试次数
        self._retry_interval: float = 5.0            # 重试间隔(秒)
        self._max_retries: int = 3                   # 最大重试次数
        self._timer_task: asyncio.Task = None
        self._deadline: float = 0.0
    
    def configure_timeout(
        self,
        threshold: float = 30.0,
        max_retries: int = 3,
        retry_interval: float = 5.0
    ):
        """配置超时触发参数"""
        self._timeout_threshold = threshold
        self._max_retries = max_retries
        self._retry_interval = retry_interval
        print(f"[确认点:{self.name}] 超时配置 | 阈值:{threshold}s "
              f"| 重试:{max_retries}次 | 间隔:{retry_interval}s")
    
    async def start_timeout_monitor(self):
        """启动超时监控协程"""
        if self._triggered:
            return
        
        self._deadline = time.time() + self._timeout_threshold
        print(f"[确认点:{self.name}] 超时监控启动 | 截止:{self._deadline}")
        
        self._timer_task = asyncio.create_task(self._run_timeout_loop())
    
    async def _run_timeout_loop(self):
        """超时检测循环"""
        remaining = self._timeout_threshold
        
        for attempt in range(self._max_retries + 1):
            if self._triggered:
                return
            
            await asyncio.sleep(remaining)
            
            if self._triggered:
                return
            
            # 检查是否已超过截止时间
            if time.time() >= self._deadline:
                await self._fire_timeout_trigger(attempt)
                return
            else:
                # 未超时则重新计算剩余时间
                remaining = self._deadline - time.time()
        
        # 达到最大重试次数仍未触发则强制执行
        if not self._triggered:
            await self._fire_timeout_trigger(self._max_retries)
    
    async def _fire_timeout_trigger(self, attempt: int):
        """执行超时触发"""
        self._triggered = True
        self._trigger_type = "timeout"
        self._trigger_meta = {
            "threshold": self._timeout_threshold,
            "attempt": attempt,
            "deadline": self._deadline,
            "timestamp": time.time()
        }
        
        print(f"[确认点:{self.name}] 超时触发 | 阈值:{self._timeout_threshold}s "
              f"| 尝试:{attempt}/{self._max_retries}")
        
        if self.callback:
            if asyncio.iscoroutinefunction(self.callback):
                await self.callback(self)
            else:
                self.callback(self)
    
    async def cancel_timeout_monitor(self):
        """取消超时监控（用于手动提前触发时）"""
        if self._timer_task and not self._timer_task.done():
            self._timer_task.cancel()
            print(f"[确认点:{self.name}] 超时监控已取消")

配置文件示例（YAML）： 超时阈值参数通过配置文件加载，支持运行时热更新。

# config/confirm_points.yaml
confirm_points:
  task_execution_check:
    enabled: true
    trigger_modes:
      - manual
      - auto
      - timeout
    timeout:
      threshold: 30.0          # 超时阈值（秒）
      max_retries: 3           # 最大重试次数
      retry_interval: 5.0      # 重试间隔（秒）
    auto_rules:
      - event_type: "task.completed"
        condition: "status == success"
      - event_type: "task.error"
        condition: "fatal == true"
  
  resource_cleanup_confirm:
    enabled: true
    trigger_modes:
      - timeout                # 仅超时触发作为保底
    timeout:
      threshold: 120.0
      max_retries: 1
      retry_interval: 10.0

# 加载配置并启动超时监控
async def load_and_start_confirm_point(config_path: str):
    with open(config_path, "r") as f:
        config = yaml.safe_load(f)
    
    for name, cfg in config["confirm_points"].items():
        if not cfg["enabled"]:
            continue
        
        cp = ConfirmPointTimeout(name, callback=on_confirm)
        
        if "timeout" in cfg.get("trigger_modes", []):
            t = cfg["timeout"]
            cp.configure_timeout(
                threshold=t["threshold"],
                max_retries=t["max_retries"],
                retry_interval=t["retry_interval"]
            )
            await cp.start_timeout_monitor()
        
        agent.register_confirm_point(cp)

6. 调试与验证方法

针对三种触发条件提供差异化的调试手段，确保触发逻辑在生产环境可观测、可复现。

# 调试命令：查看确认点状态（CLI）
agent-cli confirm-point status --name task_execution_check

# 输出示例
+---------------------------+------------------+---------------------+
| 字段                      | 当前值            | 说明                |
+---------------------------+------------------+---------------------+
| name                      | task_exec_check  | 确认点名称          |
| triggered                 | false            | 是否已触发          |
| trigger_type              | none             | 触发方式            |
| timeout_threshold         | 30.0s            | 超时阈值            |
| timeout_remaining         | 12.4s            | 剩余超时时间        |
| auto_rules_registered     | 2                | 自动规则数          |
| manual_trigger_count      | 0                | 手动触发次数        |
+---------------------------+------------------+---------------------+

# 调试日志：开启确认点详细日志
export CONFIRM_POINT_LOG_LEVEL=DEBUG

# 日志输出示例
[2025-01-15 10:23:45.123] [DEBUG] [ConfirmPoint] task_exec_check 监控启动 | deadline=10:24:15
[2025-01-15 10:23:47.456] [DEBUG] [ConfirmPoint] task_exec_check 事件到达 | type=task.completed
[2025-01-15 10:23:47.457] [DEBUG] [ConfirmPoint] task_exec_check 条件评估 | rule=0 | match=true
[2025-01-15 10:23:47.458] [INFO ] [ConfirmPoint] task_exec_check 自动触发 | event=task.completed
[2025-01-15 10:23:47.459] [DEBUG] [ConfirmPoint] task_exec_check 超时监控已取消
[2025-01-15 10:23:47.460] [INFO ] [ConfirmPoint] task_exec_check 回调执行开始

7. 常见错误与解决方案

8. 性能与稳定性优化

• 定时器精度控制： 超时监控使用 asyncio.sleep 而非 time.sleep，避免阻塞事件循环；高精度场景改用 asyncio.loop.call_later 或操作系统高精度定时器。
• 事件总线背压： 自动触发场景下，事件到达速率可能超过确认点处理能力。引入有界队列（如 asyncio.Queue(maxsize=1024)）和丢弃策略，防止内存溢出。
• 确认点去重： 同一确认点在一次生命周期内只允许触发一次，通过 _triggered 原子标志位保证，避免重复回调导致状态混乱。
• 超时监控资源回收： 每个确认点的超时监控任务必须在触发或取消时显式清理，使用 try/finally 或 Task.add_done_callback 确保回收。
• 批量确认点调度： 当Agent管理大量确认点时（1000+），使用时间轮（Timing Wheel）替代独立 asyncio.create_task，将定时器开销从 O(n) 降为 O(1)。
• 条件函数沙箱： 自动触发条件函数应在受限沙箱中执行，设置超时保护（如 asyncio.wait_for(condition_fn(), timeout=0.1)），防止恶意或死循环条件拖垮事件循环。