第九章：智能体会话恢复与错误处理

# apps/backend/agents/base.py（简化）

# 子任务级别的重试限制
MAX_SUBTASK_RETRIES = 5      # 超过此次数标记为"卡死"

# 工具并发错误的重试配置（指数退避）
MAX_CONCURRENCY_RETRIES = 5
INITIAL_RETRY_DELAY_SECONDS = 2   # 依次：2s, 4s, 8s, 16s, 32s
MAX_RETRY_DELAY_SECONDS = 32      # 退避上限

# 暂停/恢复机制的文件标志
RATE_LIMIT_PAUSE_FILE = "RATE_LIMIT_PAUSE"  # 后端创建：已触发速率限制
AUTH_FAILURE_PAUSE_FILE = "AUTH_PAUSE"       # 后端创建：认证失败
RESUME_FILE = "RESUME"                       # 前端创建：用户已处理，请继续

# 等待超时
MAX_RATE_LIMIT_WAIT_SECONDS = 7200           # 最多等待 2 小时
RATE_LIMIT_CHECK_INTERVAL_SECONDS = 30       # 每 30 秒检查一次 RESUME 文件

# apps/backend/agents/session.py（简化）

except Exception as e:
    is_concurrency = is_tool_concurrency_error(e)   # 400 工具并发
    is_rate_limit = is_rate_limit_error(e)           # 429 配额耗尽
    is_auth = is_authentication_error(e)             # 401 认证失败

    if is_concurrency:
        error_type = "tool_concurrency"
    elif is_rate_limit:
        error_type = "rate_limit"
    elif is_auth:
        error_type = "authentication"
    else:
        error_type = "other"

# apps/backend/agents/base.py（简化）

def sanitize_error_message(error_message: str, max_length: int = 500) -> str:
    """在输出前对错误消息中的敏感信息进行脱敏。"""
    # 模式 1: sk-... 格式的 API 密钥（OpenAI/Anthropic 格式）
    sanitized = re.sub(
        r"\bsk-[a-zA-Z0-9._\-]{20,}\b",
        "[REDACTED_API_KEY]",
        error_message
    )
    # 模式 2: Bearer token
    sanitized = re.sub(
        r"\bBearer\s+[a-zA-Z0-9._\-]{20,}\b",
        "Bearer [REDACTED_TOKEN]",
        sanitized
    )
    # 模式 3: token=/secret= 赋值形式
    sanitized = re.sub(
        r"(token[=:]\s*)[a-zA-Z0-9._\-]{20,}\b",
        r"\1[REDACTED_TOKEN]",
        sanitized,
        flags=re.IGNORECASE,
    )
    return sanitized[:max_length]  # 同时截断超长消息

# 脱敏必须在打印到 stdout 之前发生，因为 stdout 会被前端捕获
sanitized_error = sanitize_error_message(str(e))

# 这里在 Python 中运行（100% 可靠），而非依赖智能体合规性。

# 恢复动作执行逻辑（apps/backend/agents/session.py，简化）

def _execute_recovery_action(recovery_action, recovery_manager, ...):
    if recovery_action.action == "rollback":
        # 代码回滚到上一个"好的"commit
        recovery_manager.rollback_to_commit(recovery_action.target)

    elif recovery_action.action == "retry":
        # 重置子任务状态为 pending，用不同方法重试
        reset_subtask(spec_dir, project_dir, subtask_id)

    elif recovery_action.action in ("skip", "escalate"):
        # 标记为"卡死"，需要人工介入
        recovery_manager.mark_subtask_stuck(subtask_id, recovery_action.reason)

# apps/backend/agents/session.py（简化）

is_concurrency_error = (error_info and error_info.get("type") == "tool_concurrency")

if is_concurrency_error:
    # 直接重置为 pending，不走 check_and_recover() 恢复逻辑
    reset_subtask(spec_dir, project_dir, subtask_id)

    # 同时重置实现计划中的状态
    for phase in plan.get("phases", []):
        for subtask in phase.get("subtasks", []):
            if subtask.get("id") == subtask_id:
                subtask["status"] = "pending"
                subtask["started_at"] = None
                subtask["completed_at"] = None

    # 原子写入（防止文件损坏）
    write_json_atomic(plan_path, plan, indent=2)

# 成功路径和失败路径都调用 extract_session_insights()
extracted_insights = await extract_session_insights(
    spec_dir=spec_dir,
    project_dir=project_dir,
    subtask_id=subtask_id,
    session_num=session_num,
    commit_before=commit_before,
    commit_after=commit_after,
    success=False,        # 失败的会话
    recovery_manager=recovery_manager,
)

# 然后保存到记忆系统
await save_session_memory(
    ...
    success=False,
    discoveries=extracted_insights,  # 失败原因也是"发现"
)

# lab_09_retry.py
import re
import time
import asyncio
import random
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable


class ErrorType(Enum):
    CONCURRENCY = "concurrency"   # 可立即重试（有上限）
    RATE_LIMIT = "rate_limit"     # 等待后重试
    AUTH = "auth"                 # 不可重试，需人工处理
    TRANSIENT = "transient"       # 可重试
    PERMANENT = "permanent"       # 不可重试


@dataclass
class ErrorInfo:
    error_type: ErrorType
    original_message: str
    sanitized_message: str
    retryable: bool
    retry_delay: float = 0.0


def sanitize_error_message(message: str) -> str:
    """
    对错误消息进行脱敏处理。

    TODO: 实现以下脱敏规则（5-8 行）：
    1. 替换 sk-xxx 格式的 API 密钥为 [REDACTED_API_KEY]
    2. 替换 Bearer xxx 格式的 token
    3. 替换 token=xxx 或 token:xxx 格式的值
    4. 截断到 500 字符

    提示：使用 re.sub() 和正则表达式
    """
    # YOUR CODE HERE
    pass


def classify_error(error: Exception) -> ErrorType:
    """
    对异常进行分类。

    TODO: 根据以下规则分类（8-12 行）：
    - 消息包含 "400" 且 "concurrent" → CONCURRENCY
    - 消息包含 "429" 或 "rate limit" → RATE_LIMIT
    - 消息包含 "401" 或 "unauthorized" → AUTH
    - ConnectionError, TimeoutError → TRANSIENT
    - 其他 → PERMANENT

    提示：检查 str(error).lower() 中的关键词
    """
    # YOUR CODE HERE
    pass


def build_error_info(error: Exception, retry_delay: float = 0.0) -> ErrorInfo:
    """根据异常构建 ErrorInfo 对象。"""
    error_type = classify_error(error)
    original = str(error)
    sanitized = sanitize_error_message(original)

    retryable = error_type in (
        ErrorType.CONCURRENCY,
        ErrorType.RATE_LIMIT,
        ErrorType.TRANSIENT,
    )

    return ErrorInfo(
        error_type=error_type,
        original_message=original,
        sanitized_message=sanitized,
        retryable=retryable,
        retry_delay=retry_delay,
    )


# === 验证 Part A ===
def test_sanitize():
    msg = "Request failed with key sk-ant-api03-abcdef1234567890xyz Bearer eyJ0eXAiOiJKV1Q"
    result = sanitize_error_message(msg)
    assert "[REDACTED_API_KEY]" in result, "API key should be redacted"
    assert "[REDACTED_TOKEN]" in result, "Bearer token should be redacted"
    assert "sk-ant-api03" not in result, "Raw API key should not appear"
    print("[PASS] sanitize_error_message")

def test_classify():
    cases = [
        (Exception("400 concurrent tool calls exceeded"), ErrorType.CONCURRENCY),
        (Exception("429 rate limit exceeded"), ErrorType.RATE_LIMIT),
        (Exception("401 unauthorized access"), ErrorType.AUTH),
        (ConnectionError("connection refused"), ErrorType.TRANSIENT),
        (ValueError("unknown error"), ErrorType.PERMANENT),
    ]
    for error, expected in cases:
        result = classify_error(error)
        assert result == expected, f"Expected {expected}, got {result} for {error}"
    print("[PASS] classify_error")

# 接续 lab_09_retry.py

@dataclass
class RetryConfig:
    max_retries: int = 5
    initial_delay: float = 2.0
    max_delay: float = 32.0
    backoff_factor: float = 2.0


class RetrySession:
    """
    带指数退避的会话重试管理器。

    记录每次尝试的结果，提供退避延迟计算。
    """

    def __init__(self, task_id: str, config: RetryConfig = None):
        self.task_id = task_id
        self.config = config or RetryConfig()
        self.attempts: list[dict] = []
        self.consecutive_failures = 0

    def record_attempt(self, success: bool, error_info: ErrorInfo | None = None):
        """记录一次尝试结果。"""
        self.attempts.append({
            "attempt": len(self.attempts) + 1,
            "success": success,
            "error_type": error_info.error_type.value if error_info else None,
        })
        if success:
            self.consecutive_failures = 0
        else:
            self.consecutive_failures += 1

    def get_retry_delay(self) -> float:
        """
        计算下次重试的等待时间（指数退避）。

        TODO: 实现指数退避公式（3-5 行）：
        delay = initial_delay * (backoff_factor ** (consecutive_failures - 1))
        限制在 max_delay 内
        加入 ±20% 的随机抖动（jitter），防止惊群效应

        提示：使用 min(delay, max_delay) 和 random.uniform(0.8, 1.2)
        """
        # YOUR CODE HERE
        pass

    def should_retry(self, error_info: ErrorInfo) -> bool:
        """
        判断是否应该重试。

        TODO: 实现重试判断逻辑（5-8 行）：
        - 如果 error_type 为 AUTH → 不重试
        - 如果 error_type 为 PERMANENT → 不重试
        - 如果 consecutive_failures >= max_retries → 不重试
        - 如果 error_info.retryable → 重试
        """
        # YOUR CODE HERE
        pass

    @property
    def attempt_count(self) -> int:
        return len(self.attempts)


async def run_with_retry(
    task_fn: Callable,
    task_id: str,
    config: RetryConfig = None,
    verbose: bool = True,
) -> tuple[bool, Any, RetrySession]:
    """
    带重试的任务执行器。

    Args:
        task_fn: 异步任务函数，失败时抛出异常
        task_id: 任务标识符
        config: 重试配置
        verbose: 是否打印重试信息

    Returns:
        (success, result, session) 三元组
    """
    session = RetrySession(task_id, config)
    result = None

    while True:
        try:
            result = await task_fn()
            session.record_attempt(success=True)
            if verbose:
                print(f"  [OK] Attempt {session.attempt_count} succeeded")
            return True, result, session

        except Exception as e:
            error_info = build_error_info(e)
            session.record_attempt(success=False, error_info=error_info)

            if verbose:
                print(f"  [FAIL] Attempt {session.attempt_count}: "
                      f"{error_info.error_type.value} - {error_info.sanitized_message[:80]}")

            if not session.should_retry(error_info):
                if verbose:
                    reason = ("max retries exceeded"
                              if session.consecutive_failures >= session.config.max_retries
                              else "non-retryable error")
                    print(f"  [STOP] Not retrying: {reason}")
                return False, None, session

            delay = session.get_retry_delay()
            if verbose:
                print(f"  [WAIT] Retrying in {delay:.1f}s...")
            await asyncio.sleep(delay)


# === 验证 Part B ===
async def test_retry_success():
    """测试：第 3 次尝试成功"""
    call_count = 0

    async def flaky_task():
        nonlocal call_count
        call_count += 1
        if call_count < 3:
            raise ConnectionError("transient network error")
        return "SUCCESS"

    config = RetryConfig(max_retries=5, initial_delay=0.01, max_delay=0.1)
    success, result, session = await run_with_retry(flaky_task, "test-1", config, verbose=False)

    assert success, "Should succeed after retries"
    assert result == "SUCCESS"
    assert session.attempt_count == 3
    print("[PASS] retry_success after 3 attempts")


async def test_retry_exhausted():
    """测试：重试次数耗尽"""
    async def always_fail():
        raise ConnectionError("persistent error")

    config = RetryConfig(max_retries=3, initial_delay=0.01, max_delay=0.05)
    success, _, session = await run_with_retry(always_fail, "test-2", config, verbose=False)

    assert not success, "Should fail after max retries"
    assert session.attempt_count == 4  # 1 initial + 3 retries
    print("[PASS] retry_exhausted after max retries")


async def test_auth_no_retry():
    """测试：AUTH 错误不重试"""
    async def auth_fail():
        raise Exception("401 unauthorized - token invalid")

    config = RetryConfig(max_retries=5, initial_delay=0.01)
    success, _, session = await run_with_retry(auth_fail, "test-3", config, verbose=False)

    assert not success
    assert session.attempt_count == 1, "Should not retry auth errors"
    print("[PASS] auth_error_no_retry")

# 接续 lab_09_retry.py

class RecoveryAction(Enum):
    ROLLBACK = "rollback"
    RETRY = "retry"
    SKIP = "skip"
    ESCALATE = "escalate"


@dataclass
class RecoveryDecision:
    action: RecoveryAction
    reason: str
    target: str | None = None   # rollback 时为目标 commit hash


def decide_recovery(
    session: RetrySession,
    has_partial_commits: bool = False,
    is_skippable: bool = True,
    max_retries: int = 5,
) -> RecoveryDecision:
    """
    根据会话历史决定恢复动作。

    TODO: 实现恢复决策逻辑（10-15 行）：

    决策规则（按优先级）：
    1. 如果有部分 commit 且连续失败 >= 2 次
       → ROLLBACK（代码状态可能已损坏）
    2. 如果连续失败 < max_retries
       → RETRY（还有重试机会）
    3. 如果是可跳过的任务（is_skippable）
       → SKIP（跳过继续其他任务）
    4. 否则
       → ESCALATE（需要人工处理）

    提示：返回 RecoveryDecision 对象，
    reason 字段应包含有助于诊断的信息
    """
    # YOUR CODE HERE
    pass


# 运行所有测试
async def main():
    print("=== Part A: 错误分类测试 ===")
    test_sanitize()
    test_classify()

    print("\n=== Part B: 重试逻辑测试 ===")
    await test_retry_success()
    await test_retry_exhausted()
    await test_auth_no_retry()

    print("\n=== Part C: 恢复决策测试 ===")

    # 测试 1: 多次失败 → RETRY
    session = RetrySession("task-1")
    for _ in range(2):
        session.record_attempt(False, build_error_info(ConnectionError()))
    decision = decide_recovery(session, has_partial_commits=False)
    assert decision.action == RecoveryAction.RETRY, f"Expected RETRY, got {decision.action}"
    print("[PASS] decide_recovery: RETRY after 2 failures")

    # 测试 2: 有部分 commit 且失败 2 次 → ROLLBACK
    session = RetrySession("task-2")
    for _ in range(2):
        session.record_attempt(False, build_error_info(ConnectionError()))
    decision = decide_recovery(session, has_partial_commits=True)
    assert decision.action == RecoveryAction.ROLLBACK
    print("[PASS] decide_recovery: ROLLBACK with partial commits")

    # 测试 3: 超过最大重试 → SKIP
    session = RetrySession("task-3")
    for _ in range(6):
        session.record_attempt(False, build_error_info(ConnectionError()))
    decision = decide_recovery(session, is_skippable=True, max_retries=5)
    assert decision.action == RecoveryAction.SKIP
    print("[PASS] decide_recovery: SKIP after max retries")

    print("\n所有测试通过！")


if __name__ == "__main__":
    asyncio.run(main())