Agent 上下文包生成器：文件选择、规则注入与 Token 裁剪

任务：修复 Issue #456 "用户登录时偶发 500 错误"

Agent 拿到的信息：
  - Issue 标题和描述（200 tokens）
  - 整个仓库文件列表（Agent 自己决定读哪些）

Agent 的行为：
  1. 读 src/auth/login.py          → 2000 tokens
  2. 读 src/auth/session.py        → 1500 tokens
  3. 读 src/auth/middleware.py      → 800 tokens
  4. 读 src/db/connection.py        → 1200 tokens（因为猜测是 DB 问题）
  5. 读 src/db/models/user.py       → 900 tokens
  6. 读 config.yaml                 → 500 tokens
  7. 读 README.md                   → 300 tokens
  8. 读 src/auth/__init__.py        → 200 tokens
  9. 读 src/utils/logger.py         → 600 tokens
  10. 读 src/api/routes/auth.py     → 1100 tokens
  ...（持续 20+ 次文件读取）

总计消耗：~15,000 tokens（仅文件读取），Agent 才开始动手修复

                Issue #456
                PR #789
                Agent 历史会话
                     │
                     ▼
       ┌───────────────────────────┐
       │  Phase 1: 候选文件收集     │
       │                           │
       │  • Issue 关键词提取        │
       │  • Git blame 关联文件      │
       │  • 导入依赖分析            │
       │  • 最近变更文件            │
       └───────────┬───────────────┘
                   │ ~50 个候选文件
                   ▼
       ┌───────────────────────────┐
       │  Phase 2: 相关性评分       │
       │                           │
       │  • 关键词匹配度            │
       │  • 导入链路距离            │
       │  • 变更频率权重            │
       │  • Issue 提及权重          │
       └───────────┬───────────────┘
                   │ 按分数排序
                   ▼
       ┌───────────────────────────┐
       │  Phase 3: 敏感信息过滤     │
       │                           │
       │  • 密钥/凭证文件排除       │
       │  • 生产配置排除            │
       │  • 客户数据排除            │
       └───────────┬───────────────┘
                   │ 过滤后
                   ▼
       ┌───────────────────────────┐
       │  Phase 4: Token 预算裁剪   │
       │                           │
       │  • 按优先级分配 Token      │
       │  • 高优先文件完整保留      │
       │  • 低优先文件截断/摘要     │
       │  • 规则文件按层级注入      │
       └───────────┬───────────────┘
                   │
                   ▼
            上下文包 (Context Package)
            ~4000-6000 tokens

# context-packages.yaml
packages:
  bugfix:
    required_files:
      # 这些文件无论如何都要包含
      - pattern: "CLAUDE.md"
        max_lines: 200
      - pattern: ".claude/rules/*.md"
        max_lines: 100
    scoring:
      # 文件评分权重
      keyword_match: 0.35        # 文件名/内容与 Issue 关键词的匹配度
      import_distance: 0.25      # 与已知相关文件的导入链路距离
      change_frequency: 0.15     # 近 30 天的变更频率
      issue_reference: 0.15      # Issue/PR 中直接提及的文件
      test_proximity: 0.10       # 是否有对应的测试文件
    budget:
      total_tokens: 6000
      files_allocation: 0.60     # 文件内容占 60%
      rules_allocation: 0.20     # 规则占 20%
      history_allocation: 0.15   # 历史上下文占 15%
      reserve: 0.05              # 预留 5%
    max_files: 12
    min_files: 3

  feature:
    required_files:
      - pattern: "CLAUDE.md"
        max_lines: 200
      - pattern: "docs/architecture.md"
        max_lines: 150
    scoring:
      keyword_match: 0.30
      import_distance: 0.30
      change_frequency: 0.10
      issue_reference: 0.20
      test_proximity: 0.10
    budget:
      total_tokens: 8000
      files_allocation: 0.55
      rules_allocation: 0.20
      history_allocation: 0.20
      reserve: 0.05
    max_files: 20
    min_files: 5

import re
import os
from dataclasses import dataclass, field
from pathlib import Path

@dataclass
class FileCandidate:
    """候选文件"""
    path: str
    score: float = 0.0
    reasons: list[str] = field(default_factory=list)
    token_estimate: int = 0

class ContextScorer:
    """文件相关性评分器"""

    def __init__(self, repo_root: str, config: dict):
        self.repo_root = Path(repo_root)
        self.weights = config.get("scoring", {})

    def score_files(
        self,
        candidates: list[str],
        keywords: list[str],
        related_files: list[str],
        recent_changes: dict[str, int],
        issue_mentions: list[str]
    ) -> list[FileCandidate]:
        """对所有候选文件评分并排序"""
        results = []

        for file_path in candidates:
            candidate = FileCandidate(path=file_path)
            full_path = self.repo_root / file_path

            if not full_path.exists():
                continue

            # 估算 Token 数（粗略：1 token ≈ 4 chars / 1.3 中文字符）
            content = full_path.read_text(encoding="utf-8", errors="ignore")
            candidate.token_estimate = self._estimate_tokens(content)

            # 1. 关键词匹配
            kw_score = self._keyword_score(file_path, content, keywords)
            candidate.score += kw_score * self.weights.get("keyword_match", 0.35)
            if kw_score > 0.5:
                candidate.reasons.append(f"关键词匹配度: {kw_score:.2f}")

            # 2. 导入链路距离
            import_score = self._import_distance(file_path, related_files)
            candidate.score += import_score * self.weights.get("import_distance", 0.25)
            if import_score > 0:
                candidate.reasons.append(f"导入距离: {import_score:.2f}")

            # 3. 变更频率
            changes = recent_changes.get(file_path, 0)
            freq_score = min(changes / 10.0, 1.0)  # 10 次变更 = 满分
            candidate.score += freq_score * self.weights.get("change_frequency", 0.15)

            # 4. Issue 直接提及
            mention_score = 1.0 if file_path in issue_mentions else 0.0
            candidate.score += mention_score * self.weights.get("issue_reference", 0.15)
            if mention_score > 0:
                candidate.reasons.append("Issue 直接提及")

            # 5. 测试文件邻近
            test_score = self._test_proximity(file_path, candidates)
            candidate.score += test_score * self.weights.get("test_proximity", 0.10)

            results.append(candidate)

        # 按分数降序
        results.sort(key=lambda c: c.score, reverse=True)
        return results

    def _keyword_score(self, path: str, content: str, keywords: list[str]) -> float:
        """关键词匹配度：文件名匹配权重高，内容匹配权重低"""
        filename = os.path.basename(path).lower()
        path_lower = path.lower()
        score = 0.0

        for kw in keywords:
            kw_lower = kw.lower()
            if kw_lower in filename:
                score += 0.8  # 文件名匹配，高分
            elif kw_lower in path_lower:
                score += 0.4  # 路径匹配，中分
            elif kw_lower in content.lower():
                # 内容匹配，按出现次数计分
                count = content.lower().count(kw_lower)
                score += min(count * 0.05, 0.3)

        return min(score, 1.0)

    def _import_distance(self, path: str, related_files: list[str]) -> float:
        """导入链路距离：直接导入 = 1.0，间接导入 = 0.5，无关联 = 0"""
        if not related_files:
            return 0.0

        # 简化实现：检查文件名是否被 related_files 导入
        module_name = Path(path).stem
        score = 0.0

        for related in related_files:
            try:
                content = (self.repo_root / related).read_text(errors="ignore")
                if f"import {module_name}" in content or f"from {module_name}" in content:
                    score = max(score, 1.0)
                elif module_name in content:
                    score = max(score, 0.5)
            except (OSError, UnicodeDecodeError):
                continue

        return score

    def _test_proximity(self, path: str, all_candidates: list[str]) -> float:
        """测试文件邻近度：如果有对应的测试文件在候选列表中"""
        stem = Path(path).stem
        test_patterns = [
            f"test_{stem}.py", f"{stem}_test.py",
            f"test_{stem}.js", f"{stem}.test.js", f"{stem}.spec.js",
            f"test_{stem}.ts", f"{stem}.test.ts", f"{stem}.spec.ts",
        ]
        for pattern in test_patterns:
            if any(pattern in c for c in all_candidates):
                return 1.0
        return 0.0

    def _estimate_tokens(self, content: str) -> int:
        """粗略估算 Token 数"""
        # 英文约 1 token / 4 chars，中文约 1 token / 1.5 chars
        cjk_count = sum(1 for c in content if '\u4e00' <= c <= '\u9fff')
        ascii_count = len(content) - cjk_count
        return int(ascii_count / 4 + cjk_count / 1.5)

@dataclass
class RuleFile:
    path: str
    priority: int        # 1 = 最高，必须包含
    max_lines: int
    content: str = ""

class RuleInjector:
    """规则文件分层注入"""

    RULE_LAYERS = [
        # (目录模式, 优先级, 最大行数)
        ("CLAUDE.md",                    1, 200),   # 项目根规则
        (".claude/rules/core/*.md",      2, 100),   # 核心规则
        (".claude/rules/style/*.md",     3, 80),    # 风格规则
        (".claude/rules/testing/*.md",   4, 80),    # 测试规则
        (".claude/rules/security/*.md",  5, 100),   # 安全规则
        ("docs/architecture.md",         6, 150),   # 架构文档
        ("CONTRIBUTING.md",              7, 100),   # 贡献指南
    ]

    def collect_rules(self, repo_root: Path) -> list[RuleFile]:
        """收集所有规则文件"""
        rules = []
        for pattern, priority, max_lines in self.RULE_LAYERS:
            if "*" in pattern:
                for f in repo_root.glob(pattern):
                    rules.append(RuleFile(str(f.relative_to(repo_root)), priority, max_lines))
            else:
                full = repo_root / pattern
                if full.exists():
                    rules.append(RuleFile(pattern, priority, max_lines))
        return rules

    def inject(self, rules: list[RuleFile], token_budget: int) -> list[RuleFile]:
        """在 Token 预算内注入尽可能多的规则"""
        result = []
        used_tokens = 0

        # 按优先级排序
        rules.sort(key=lambda r: r.priority)

        for rule in rules:
            full_path = Path(rule.path)
            if not full_path.exists():
                continue

            content = full_path.read_text(encoding="utf-8", errors="ignore")
            lines = content.splitlines()

            # 截断到 max_lines
            if len(lines) > rule.max_lines:
                content = "\n".join(lines[:rule.max_lines])
                content += f"\n... (截断，共 {len(lines)} 行)"

            tokens = estimate_tokens(content)
            if used_tokens + tokens <= token_budget:
                rule.content = content
                result.append(rule)
                used_tokens += tokens
            elif rule.priority <= 2:
                # 优先级 1-2 的规则强制包含，即使超出预算
                rule.content = content
                result.append(rule)
                used_tokens += tokens

        return result

import re

SENSITIVE_PATTERNS = [
    # 密钥和凭证
    re.compile(r'(?i)(password|secret|api[_-]?key|token|credential)\s*[=:]\s*["\'][^"\']+["\']'),
    re.compile(r'(?i)AWS_(SECRET_)?ACCESS_KEY\s*='),
    re.compile(r'(?i)(GITHUB_TOKEN|NPM_TOKEN|DATABASE_URL)\s*='),
    # 私钥
    re.compile(r'-----BEGIN (RSA |EC |DSA )?PRIVATE KEY-----'),
    # 数据库连接串（含密码）
    re.compile(r'(postgres|mysql|mongodb)://[^:]+:[^@]+@'),
]

SENSITIVE_FILES = [
    "*.env", ".env.*", "*.pem", "*.key", "*.p12", "*.pfx",
    "**/secrets/**", "**/credentials/**", "**/.aws/**",
    "id_rsa", "id_ed25519", "*.keystore",
]

class SensitiveFilter:
    """敏感信息过滤器"""

    def is_sensitive_file(self, path: str) -> bool:
        """文件路径是否匹配敏感模式"""
        from fnmatch import fnmatch
        return any(fnmatch(path, pat) for pat in SENSITIVE_FILES)

    def contains_sensitive_data(self, content: str) -> tuple[bool, list[str]]:
        """文件内容是否包含敏感数据，返回 (是否敏感, 匹配项列表)"""
        matches = []
        for pattern in SENSITIVE_PATTERNS:
            found = pattern.findall(content)
            if found:
                matches.extend(found[:3])  # 最多报告 3 个匹配
        return (len(matches) > 0, matches)

    def sanitize(self, content: str) -> str:
        """脱敏处理：将敏感值替换为占位符"""
        result = content
        for pattern in SENSITIVE_PATTERNS:
            result = pattern.sub("[REDACTED]", result)
        return result

@dataclass
class ContextPackage:
    """最终的上下文包"""
    files: list[dict]          # 文件内容
    rules: list[dict]          # 规则内容
    history: list[dict]        # 历史上下文
    total_tokens: int
    metadata: dict             # 生成元数据

class ContextBuilder:
    """上下文包构建器"""

    def build(
        self,
        task_config: dict,
        scored_files: list[FileCandidate],
        rules: list[RuleFile],
        history: list[dict],
    ) -> ContextPackage:
        """构建最终上下文包"""
        budget = task_config["budget"]
        total = budget["total_tokens"]

        file_budget = int(total * budget["files_allocation"])
        rule_budget = int(total * budget["rules_allocation"])
        history_budget = int(total * budget["history_allocation"])

        # 1. 注入规则（优先保证）
        injected_rules = self._inject_rules(rules, rule_budget)
        rule_tokens_used = sum(estimate_tokens(r["content"]) for r in injected_rules)

        # 2. 分配剩余预算给文件
        remaining_for_files = file_budget + max(0, rule_budget - rule_tokens_used)
        selected_files = self._select_files(scored_files, remaining_for_files,
                                             task_config.get("max_files", 12))

        # 3. 历史上下文
        selected_history = self._select_history(history, history_budget)

        # 4. 元数据
        metadata = {
            "total_tokens": sum(
                estimate_tokens(f.get("content", "")) for f in selected_files
            ) + rule_tokens_used + sum(
                estimate_tokens(h.get("content", "")) for h in selected_history
            ),
            "files_count": len(selected_files),
            "rules_count": len(injected_rules),
            "history_count": len(selected_history),
            "file_budget": file_budget,
            "rule_budget": rule_budget,
        }

        return ContextPackage(
            files=selected_files,
            rules=injected_rules,
            history=selected_history,
            total_tokens=metadata["total_tokens"],
            metadata=metadata,
        )

    def _select_files(
        self, files: list[FileCandidate], budget: int, max_files: int
    ) -> list[dict]:
        """在预算内选择文件，高分文件完整保留，低分文件截断"""
        result = []
        used = 0

        for f in files[:max_files]:
            content = (Path(f.path)).read_text(encoding="utf-8", errors="ignore")
            tokens = f.token_estimate

            if used + tokens <= budget:
                # 预算充足，完整保留
                result.append({"path": f.path, "content": content, "score": f.score})
                used += tokens
            elif f.score >= 0.5:
                # 高分文件但预算不够，截断
                max_chars = (budget - used) * 4  # 粗略转换
                if max_chars > 200:
                    truncated = content[:max_chars] + f"\n... (截断，共 {len(content)} 字符)"
                    result.append({"path": f.path, "content": truncated, "score": f.score})
                    used = budget
                    break
            # 低分文件 + 预算不够 → 直接跳过

        return result

    def _inject_rules(self, rules: list[RuleFile], budget: int) -> list[dict]:
        injector = RuleInjector()
        injected = injector.inject(rules, budget)
        return [{"path": r.path, "content": r.content} for r in injected]

    def _select_history(self, history: list[dict], budget: int) -> list[dict]:
        """选择历史上下文（Issue 描述、PR 评论、会话摘要）"""
        result = []
        used = 0
        for h in history:
            tokens = estimate_tokens(h.get("content", ""))
            if used + tokens <= budget:
                result.append(h)
                used += tokens
            else:
                break
        return result

FILE_TYPE_WEIGHTS = {
    ".py": 1.3, ".js": 1.3, ".ts": 1.3, ".tsx": 1.3,
    ".go": 1.3, ".rs": 1.3, ".java": 1.3,
    ".md": 0.6, ".txt": 0.5,
    ".yaml": 0.8, ".json": 0.8, ".toml": 0.8,
    ".html": 0.9, ".css": 0.9,
}

def adjust_score(candidate: FileCandidate) -> float:
    ext = Path(candidate.path).suffix.lower()
    weight = FILE_TYPE_WEIGHTS.get(ext, 1.0)
    return candidate.score * weight

try:
    import tiktoken
    encoder = tiktoken.encoding_for_model("claude-sonnet-4-20250514")
    def estimate_tokens(content: str) -> int:
        return len(encoder.encode(content))
except ImportError:
    def estimate_tokens(content: str) -> int:
        cjk = sum(1 for c in content if '\u4e00' <= c <= '\u9fff')
        ascii_len = len(content) - cjk
        return int(ascii_len / 4 + cjk / 1.5)