class PlannerContext(BaseModel):
    """规划器上下文对象，保存历史规划信息"""
    planning_history: List[Dict[str, Any]] = []        # 所有规划尝试的完整记录
    rejected_strategies: Dict[str, str] = {}            # 被拒绝的策略及原因
    long_term_objectives: List[str] = []                # 长期战略目标
    latest_reflection_report: Optional[Dict[str, Any]]  # 最新反思报告
    previous_planning_session: Optional[Dict[str, Any]] # 上一次规划会话

class ReflectorContext(BaseModel):
    """反思器上下文对象，保存历史反思信息"""
    reflection_log: List[Dict[str, Any]] = []           # 反思日志
    validated_patterns: List[Dict[str, Any]] = []       # 已验证的攻击模式
    persistent_insights: List[Dict[str, Any]] = []      # 持久化洞察

class ContextCompressor:
    """
    基于三种策略进行智能压缩：
    1. 消息数量阈值（默认 50 条）
    2. 执行轮次阈值（每 10 轮）
    3. 估算 token 超限（默认 100000）
    """

def should_compress(self, messages, execution_count) -> tuple[bool, str]:
    # 策略 1: 消息数量 > 50
    if len(messages) > self.message_threshold:
        return True, f"消息数量过多"
    # 策略 2: 每 10 轮定期压缩
    if execution_count > 0 and execution_count % self.compress_interval == 0:
        return True, f"定期压缩"
    # 策略 3: 估算 token > 100000
    estimated_tokens = total_chars // 4
    if estimated_tokens > self.token_threshold:
        return True, f"估算 token 超限"

def attribute(self, ...):
    # 必须逐层排查，低层未排除不能归因到高层
    l1_result = self._check_l1_tool_failure(...)  # 先排查工具问题
    if l1_result: return l1_result

    l2_result = self._check_l2_prerequisite_failure(...)  # 再排查前提条件
    if l2_result: return l2_result

    l3_result = self._check_l3_environment(...)  # 再排查环境因素
    if l3_result: return l3_result

    # 只有排除 L0-L3 后，才能归因到 L4
    if hypothesis and evidence:
        l4_result = self._check_l4_hypothesis(...)
        # L4 需要 failed_attempts >= 3 才触发

    # 只有多个 L4 失败形成模式（l4_failure_count >= 3），才归因到 L5
    if strategy:
        l5_result = self._check_l5_strategy(...)

# Reflector 返回的 rejected_staged_nodes 列表
rejected_nodes = reflection_result.get("rejected_staged_nodes", [])
if rejected_nodes:
    self.veto_count += 1
    # 直接移除被拒绝的节点
    tasks = [t for t in tasks if t.get("id") not in rejected_nodes]

# 基于 Reflector 的情报摘要触发动态规划
if reflection_result.get("intelligence_summary"):
    dynamic_plan_result = await self.planner.dynamic_plan(
        goal=goal,
        graph_summary=graph_summary,
        intelligence_summary=reflection_result.get("intelligence_summary"),
        failure_patterns_summary=reflection_result.get("attribution_result"),
        failed_nodes=[t for t in tasks if t.get("status") == "failed"]
    )

class AntiAddiction:
    def __init__(self,
        similarity_threshold: float = 0.85,  # 相似度阈值
        max_similar_count: int = 18,          # 最大连续相似次数
        command_timeout: int = 300,            # 命令超时 (秒)
        max_retries: int = 3                   # 最大重试次数
    ):

def _is_similar_to_last_command(self, current_cmd: str) -> bool:
    # 1. 命令太短 (< 10 字符) 不检测
    if len(current_cmd) < self._min_cmd_length:
        return False
    # 2. 用 SequenceMatcher 计算与上一条命令的相似度
    similarity = SequenceMatcher(None, current_cmd, last_cmd).ratio()
    # 3. 相似度 >= 0.85 判定为相似命令
    return similarity >= self.similarity_threshold

class CommandHistory(BaseModel):
    commands: List[str] = []
    max_length: int = 18  # 18 条命令的滑动窗口

    def add(self, command: str) -> None:
        self.commands.append(command)
        if len(self.commands) > self.max_length:
            self.commands.pop(0)  # FIFO 弹出最早的命令

def create_retry_decorator(self, func_name: str = "") -> Callable:
    return retry(
        retry=retry_if_exception_type(Exception),
        wait=wait_exponential_jitter(
            initial=0.3,   # 首次重试等待 0.3 秒
            max=10,         # 最大等待 10 秒
            jitter=0.5      # ±0.5 秒随机抖动
        ),
        stop=stop_after_attempt(self.max_retries),  # 最多重试 3 次
        reraise=True,
    )

@staticmethod
def is_retryable_error(exception) -> bool:
    # ConnectionError → 可重试
    # TimeoutError → 可重试
    # HTTP 408, 404, 429, 500-526 → 可重试

async def execute_with_timeout(self, coro, timeout=None, func_name=""):
    timeout = timeout or self.command_timeout  # 默认 300 秒
    return await asyncio.wait_for(coro, timeout=timeout)

if tool_name == "execute_shell_command":
    command = arguments.get("command", "")
    warning = self.anti_addiction.check_and_record(command)
    if warning:
        logger.warning(f"检测到循环模式: {warning[:100]}...")

# 所有工具调用都包装重试装饰器
retry_decorator = self.anti_addiction.create_retry_decorator(func_name=tool_name)

@retry_decorator
async def _execute():
    # 实际执行逻辑

class AIPentestMCPServer:
    def __init__(self):
        self.server = Server("ai-pentest-server")
        self.python_executor: PythonExecutor | None = None
        self.browser: BrowserAutomation | None = None
        self.terminal: Terminal | None = None
        self.proxy: Proxy | None = None
        self.knowledge_base: KnowledgeBase | None = None
        self.notes_storage: NoteStorage | None = None
        self.recon_workflow: ReconWorkflow | None = None

if tool_name == "execute_python_code":
    @retry_decorator
    async def _execute():
        outputs = self.python_executor.execute_code(
            session_name=arguments.get("session_name", "default"),
            code=arguments.get("code", ""),
            timeout=arguments.get("timeout", 120)
        )
        return "\n\n".join(
            f"Output: {output.get('text', '')}"
            for output in outputs
        )

elif tool_name == "execute_shell_command":
    @retry_decorator
    async def _execute():
        session_id = self.terminal.new_session()           # 创建新终端会话
        self.terminal.send_keys(session_id=session_id,     # 发送命令
            keys=arguments.get("command", ""), enter=True)
        await asyncio.sleep(2)                              # 等待执行
        output = self.terminal.get_output(session_id)       # 获取输出
        return output

self.proxy = Proxy(
    url=config.proxy_caido_url,
    token=config.proxy_caido_token
) if config.proxy_caido_token else None

- CyberspaceMapper    # 空间测绘（FOFA/Quake）
- FingerprintScanner  # 指纹识别
- DirectoryScanner    # 目录扫描
- JSAnalyzer          # JS 分析
- AuthBypassFuzzer    # 鉴权绕过 Fuzz
- ReconWorkflow       # 侦察工作流编排