Sparksatchel

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg

# Virtual environments
venv/
ENV/
env/

# IDE
.vscode/
.idea/
*.swp
*.swo
*~

# Data directories (don't upload vector databases and history)
data/collections/
data/cache/
data/*.db
data/*.sqlite

# Model cache (large embedding models)
data/cache/models/
*.bin
*.pt
*.pth
*.onnx

# Logs
*.log

# OS
.DS_Store
Thumbs.db

# Temporary files
*.tmp
*.bak
*.swp
.cache/

# Test coverage
.coverage
htmlcov/
.pytest_cache/

# MyPy
.mypy_cache/
.dmypy.json
dmypy.json

# Non-core documentation files (keep only essential skill files)
DESIGN.md
GITHUB_DESCRIPTION.md

MIT License

Copyright (c) 2026 SparkSatchel Contributors

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

{
  "name": "sparksatchel",
  "version": "1.0.0",
  "description": "灵犀妙计 SparkSatchel - Intelligent skill retrieval system for AI IDEs",
  "bin": {
    "sparksatchel": "src/retriever.py"
  },
  "scripts": {
    "start": "python src/retriever.py",
    "install-deps": "pip install -r requirements.txt"
  },
  "keywords": [
    "skill-retrieval",
    "semantic-search",
    "claude-code",
    "cursor",
    "ai-ide",
    "embeddings",
    "bilingual"
  ],
  "author": "Codestyle Team",
  "license": "MIT",
  "repository": {
    "type": "git",
    "url": "https://github.com/gccszs/Spark-Satchel.git"
  },
  "engines": {
    "node": ">=14.0.0"
  }
}

# SparkSatchel 灵犀妙计 - 依赖列表
# Meta-Skill for intelligent skill retrieval and recommendation

# 核心依赖
chromadb>=0.5.0              # 向量数据库
sentence-transformers>=2.7.0 # Embedding 模型
numpy>=1.24.0                # 向量运算

# 可选依赖（增强功能）
tiktoken>=0.5.0              # 文本分词（可选）
openai>=1.0.0                # OpenAI API（可选，用于更精准的 embedding）

# 开发依赖
pytest>=7.4.0                # 测试框架
black>=23.0.0                # 代码格式化
mypy>=1.0.0                  # 类型检查

"""
Embedding Model Download Script

Download and setup embedding models for SparkSatchel.
Models will be downloaded to the project's data/cache/models directory.
"""

import os
import sys
from pathlib import Path


# Available models
MODELS = {
    "default": {
        "name": "paraphrase-multilingual-MiniLM-L12-v2",
        "size_mb": 470,
        "description": "Bilingual (Chinese/English), 50+ languages, balanced performance",
        "languages": ["Chinese", "English", "50+ others"],
        "dimension": 384
    },
    "chinese": {
        "name": "shibing624/text2vec-base-chinese",
        "size_mb": 110,
        "description": "Chinese-optimized, faster and more accurate for Chinese",
        "languages": ["Chinese"],
        "dimension": 768
    },
    "large": {
        "name": "intfloat/multilingual-e5-large",
        "size_mb": 1300,
        "description": "High accuracy multilingual, best for complex queries",
        "languages": ["100+ languages"],
        "dimension": 1024
    },
    "english": {
        "name": "all-MiniLM-L6-v2",
        "size_mb": 23,
        "description": "English only, ultra lightweight",
        "languages": ["English"],
        "dimension": 384
    }
}


def print_models():
    """Print available models"""
    print("\n📦 Available Embedding Models")
    print("=" * 60)
    for key, model in MODELS.items():
        print(f"\n[{key}] {model['name']}")
        print(f"    Size: {model['size_mb']}MB")
        print(f"    Languages: {', '.join(model['languages'])}")
        print(f"    Dimension: {model['dimension']}")
        print(f"    Description: {model['description']}")


def download_model(model_key: str = "default"):
    """Download specified model

    Args:
        model_key: Key from MODELS dict
    """
    if model_key not in MODELS:
        print(f"❌ Unknown model: {model_key}")
        print_models()
        sys.exit(1)

    model = MODELS[model_key]

    print(f"\n📥 Downloading model: {model['name']}")
    print(f"   Size: ~{model['size_mb']}MB")
    print(f"   This may take a few minutes...\n")

    try:
        from sentence_transformers import SentenceTransformer
        import torch

        # Check GPU availability
        device = "cuda" if torch.cuda.is_available() else "cpu"
        if device == "cuda":
            print(f"🚀 Using GPU for faster inference")

        # Download model
        model_obj = SentenceTransformer(model['name'], device=device)

        # Get cache location
        cache_path = Path(model_obj._cache_dir) if hasattr(model_obj, '_cache_dir') else Path.home() / ".cache" / "huggingface"

        print(f"\n✅ Model downloaded successfully!")
        print(f"   Cache location: {cache_path}")
        print(f"   Dimension: {model_obj.get_sentence_embedding_dimension()}")

        # Test encoding
        test_text = "Hello, 世界!"
        embedding = model_obj.encode(test_text)
        print(f"   Test: Encoded '{test_text}' -> vector shape {embedding.shape}")

        print(f"\n💡 To use this model, modify src/models/embedding.py:")
        print(f"   DEFAULT_MODEL = \"{model['name']}\"")

    except ImportError:
        print("\n❌ sentence-transformers not installed")
        print("   Run: pip install sentence-transformers")
        sys.exit(1)
    except Exception as e:
        print(f"\n❌ Download failed: {e}")
        sys.exit(1)


def main():
    """Main entry point"""
    import argparse

    parser = argparse.ArgumentParser(description="Download embedding models for SparkSatchel")
    parser.add_argument(
        "model",
        nargs="?",
        default="default",
        choices=list(MODELS.keys()),
        help="Model to download (default: default)"
    )
    parser.add_argument(
        "--list",
        action="store_true",
        help="List available models"
    )

    args = parser.parse_args()

    if args.list:
        print_models()
        sys.exit(0)

    download_model(args.model)


if __name__ == "__main__":
    main()

"""
置信度评估模块

评估技能推荐的置信度，综合考虑多种因素
"""

from typing import List, Optional
from dataclasses import dataclass
from src.storage.vector_db import SearchResult
from src.storage.history import SkillStats


@dataclass
class ConfidenceBreakdown:
    """置信度分解"""
    total: float                # 总置信度
    similarity: float           # 语义相似度
    historical: float           # 历史成功率
    relevance: float            # 相关性分数
    freshness: float            # 最近使用加成


class ConfidenceEvaluator:
    """置信度评估器"""

    # 权重配置
    WEIGHTS = {
        "similarity": 0.5,      # 语义相似度权重
        "historical": 0.3,      # 历史成功率权重
        "relevance": 0.15,      # 相关性权重
        "freshness": 0.05       # 最近使用加成
    }

    # 置信度阈值
    THRESHOLDS = {
        "high": 0.70,           # 高置信度阈值
        "medium": 0.40,         # 中置信度阈值
        "low": 0.0              # 低置信度阈值
    }

    def __init__(self, weights: dict = None):
        """初始化评估器

        Args:
            weights: 自定义权重
        """
        if weights:
            self.WEIGHTS.update(weights)

    def evaluate(
        self,
        search_result: SearchResult,
        stats: Optional[SkillStats] = None,
        num_candidates: int = 1
    ) -> ConfidenceBreakdown:
        """评估单个结果的置信度

        Args:
            search_result: 搜索结果
            stats: 技能统计信息
            num_candidates: 候选技能数量

        Returns:
            置信度分解
        """
        # 计算各维度分数
        similarity = self._similarity_score(search_result.similarity)
        historical = self._historical_score(stats)
        relevance = self._relevance_score(search_result, num_candidates)
        freshness = self._freshness_score(stats)

        # 加权求和
        total = (
            similarity * self.WEIGHTS["similarity"] +
            historical * self.WEIGHTS["historical"] +
            relevance * self.WEIGHTS["relevance"] +
            freshness * self.WEIGHTS["freshness"]
        )

        return ConfidenceBreakdown(
            total=min(total, 1.0),
            similarity=similarity,
            historical=historical,
            relevance=relevance,
            freshness=freshness
        )

    def evaluate_batch(
        self,
        search_results: List[SearchResult],
        all_stats: dict
    ) -> List[ConfidenceBreakdown]:
        """批量评估

        Args:
            search_results: 搜索结果列表
            all_stats: 所有技能的统计信息映射

        Returns:
            置信度分解列表
        """
        return [
            self.evaluate(
                result,
                all_stats.get(result.skill_name),
                len(search_results)
            )
            for result in search_results
        ]

    def _similarity_score(self, similarity: float) -> float:
        """计算相似度分数

        Args:
            similarity: 原始相似度

        Returns:
            归一化后的分数 [0, 1]
        """
        # 相似度已经归一化，但可以应用非线性变换
        # 使得高相似度更突出
        return similarity ** 0.8

    def _historical_score(self, stats: Optional[SkillStats]) -> float:
        """计算历史表现分数

        Args:
            stats: 技能统计

        Returns:
            历史分数 [0, 1]
        """
        if stats is None or stats.total_calls == 0:
            # 没有历史记录，给予中性分数
            return 0.5

        # 主要考虑成功率
        success_rate = stats.success_rate

        # 调用次数也影响可信度
        call_confidence = min(stats.total_calls / 50, 1.0)  # 50次调用达到最高可信度

        # 综合评分
        return success_rate * 0.8 + call_confidence * 0.2

    def _relevance_score(self, result: SearchResult, num_candidates: int) -> float:
        """计算相关性分数

        Args:
            result: 搜索结果
            num_candidates: 候选数量

        Returns:
            相关性分数 [0, 1]
        """
        score = 1.0

        # 关键词匹配加成
        # TODO: 实现更精确的关键词匹配

        # 候选数量影响（候选越少，说明越明确）
        if num_candidates == 1:
            score *= 1.2
        elif num_candidates <= 3:
            score *= 1.1
        elif num_candidates > 10:
            score *= 0.9

        return min(score, 1.0)

    def _freshness_score(self, stats: Optional[SkillStats]) -> float:
        """计算新鲜度分数（最近使用加成）

        Args:
            stats: 技能统计

        Returns:
            新鲜度分数 [0, 1]
        """
        if stats is None or not stats.last_called:
            return 0.5

        # TODO: 实现基于时间的衰减
        # 最近使用的技能有轻微加成
        return 0.5

    def get_confidence_level(self, confidence: float) -> str:
        """获取置信度等级

        Args:
            confidence: 置信度分数

        Returns:
            等级: "high", "medium", "low"
        """
        if confidence >= self.THRESHOLDS["high"]:
            return "high"
        elif confidence >= self.THRESHOLDS["medium"]:
            return "medium"
        else:
            return "low"

    def should_ask_user(self, confidence: float) -> bool:
        """判断是否需要询问用户

        Args:
            confidence: 置信度分数

        Returns:
            是否需要询问
        """
        return confidence < self.THRESHOLDS["high"]

"""
意图分析模块

从用户请求中提取意图，支持关键词匹配和语义分析
"""

import re
from typing import List, Optional, Dict, Tuple
from dataclasses import dataclass


@dataclass
class Intent:
    """用户意图"""
    primary: str           # 主要意图
    keywords: List[str]    # 提取的关键词
    entities: List[str]    # 实体（如文件名、格式等）
    confidence: float      # 意图识别置信度


class IntentAnalyzer:
    """意图分析器"""

    # 常见意图模式
    INTENT_PATTERNS = {
        "document_process": [
            r"处理.*文档", r"处理.*文件", r"转换.*格式",
            r"处理.*PDF", r"处理.*Word", r"处理.*Excel",
            r"extract.*document", r"process.*file"
        ],
        "project_create": [
            r"创建.*项目", r"初始化.*项目", r"新建.*项目",
            r"based on.*create", r"template.*project"
        ],
        "data_analysis": [
            r"分析.*数据", r"处理.*数据", r"统计.*",
            r"analyze.*data", r"process.*data"
        ],
        "search_skill": [
            r"查找.*技能", r"搜索.*技能", r"什么.*技能",
            r"find.*skill", r"search.*skill"
        ],
        "disk_clean": [
            r"清理.*磁盘", r"清理.*空间", r"删除.*缓存",
            r"clean.*disk", r"free.*space"
        ],
        "ai_collaborate": [
            r"多.*AI.*协作", r"AI.*协同", r"Agent.*协作",
            r"multi.*agent", r"AI.*collaborate"
        ]
    }

    # 实体提取模式
    ENTITY_PATTERNS = {
        "file_format": [
            r"\.(PDF|pdf|DOCX|docx|PPT|ppt|XLSX|xlsx|CSV|csv)",
            r"(PDF|Word|Excel|PowerPoint|文档|表格)"
        ],
        "programming_language": [
            r"\b(Python|Java|JavaScript|TypeScript|Go|Rust|C\+\+)\b"
        ]
    }

    def __init__(self):
        """初始化意图分析器"""
        self._compile_patterns()

    def _compile_patterns(self):
        """编译正则表达式"""
        self.compiled_intents = {}
        for intent, patterns in self.INTENT_PATTERNS.items():
            self.compiled_intents[intent] = [
                re.compile(p, re.IGNORECASE) for p in patterns
            ]

        self.compiled_entities = {}
        for entity_type, patterns in self.ENTITY_PATTERNS.items():
            self.compiled_entities[entity_type] = [
                re.compile(p, re.IGNORECASE) for p in patterns
            ]

    def analyze(self, user_request: str) -> Intent:
        """分析用户请求的意图

        Args:
            user_request: 用户请求文本

        Returns:
            意图对象
        """
        keywords = self._extract_keywords(user_request)
        entities = self._extract_entities(user_request)
        primary_intent = self._match_intent(user_request)
        confidence = self._calculate_confidence(user_request, primary_intent)

        return Intent(
            primary=primary_intent,
            keywords=keywords,
            entities=entities,
            confidence=confidence
        )

    def _extract_keywords(self, text: str) -> List[str]:
        """提取关键词

        Args:
            text: 输入文本

        Returns:
            关键词列表
        """
        # 简单实现：提取中文词汇和英文单词
        # 实际应该使用 jieba 或其他分词工具

        # 移除标点符号
        text = re.sub(r'[^\w\s\u4e00-\u9fff]', ' ', text)

        # 提取中文词汇（2-4字）
        chinese_words = re.findall(r'[\u4e00-\u9fff]{2,4}', text)

        # 提取英文单词
        english_words = re.findall(r'\b[a-zA-Z]{3,}\b', text)

        # 过滤常见词
        stop_words = {'这个', '那个', '可以', '需要', 'the', 'a', 'an', 'is', 'are'}
        keywords = [w for w in chinese_words + english_words if w.lower() not in stop_words]

        return list(set(keywords))

    def _extract_entities(self, text: str) -> List[str]:
        """提取实体

        Args:
            text: 输入文本

        Returns:
            实体列表
        """
        entities = []

        for entity_type, patterns in self.compiled_entities.items():
            for pattern in patterns:
                matches = pattern.findall(text)
                entities.extend(matches)

        return list(set(entities))

    def _match_intent(self, text: str) -> str:
        """匹配意图

        Args:
            text: 输入文本

        Returns:
            意图名称
        """
        best_intent = "general"
        best_score = 0

        for intent, patterns in self.compiled_intents.items():
            score = sum(1 for p in patterns if p.search(text))
            if score > best_score:
                best_score = score
                best_intent = intent

        return best_intent

    def _calculate_confidence(self, text: str, intent: str) -> float:
        """计算意图识别置信度

        Args:
            text: 输入文本
            intent: 匹配的意图

        Returns:
            置信度 [0, 1]
        """
        # 基于匹配模式数量
        patterns = self.compiled_intents.get(intent, [])
        match_count = sum(1 for p in patterns if p.search(text))

        if match_count == 0:
            return 0.3  # 默认置信度

        # 匹配越多，置信度越高
        confidence = min(0.5 + match_count * 0.15, 1.0)
        return confidence

    def extract_skill_hints(self, user_request: str) -> List[str]:
        """从请求中提取技能提示

        Args:
            user_request: 用户请求

        Returns:
            可能的技能名称列表
        """
        hints = []

        # 直接提到的技能名
        skill_names = re.findall(
            r'(\w+-?\w*\s*skill)',
            user_request,
            re.IGNORECASE
        )
        hints.extend([s.strip().lower() for s in skill_names])

        # 常见功能映射
        function_mapping = {
            "pdf": ["pdf-skill"],
            "文档": ["pdf-skill", "docx-skill"],
            "word": ["docx-skill"],
            "excel": ["xlsx-skill"],
            "表格": ["xlsx-skill"],
            "ppt": ["pptx-skill"],
            "演示": ["pptx-skill"],
            "清理": ["disk-cleaner"],
            "磁盘": ["disk-cleaner"],
            "协作": ["agent-call"],
            "项目": ["project-start-skill"]
        }

        for keyword, skills in function_mapping.items():
            if keyword.lower() in user_request.lower():
                hints.extend(skills)

        return list(set(hints))

"""
决策引擎

根据置信度和候选技能，做出推荐决策
"""

from typing import List, Optional, Dict, Any
from dataclasses import dataclass, field
from enum import Enum

from src.storage.vector_db import SearchResult, SkillMetadata
from src.storage.history import SkillStats
from src.analysis.confidence import ConfidenceEvaluator, ConfidenceBreakdown


class DecisionLevel(Enum):
    """决策等级"""
    AUTO_RECOMMEND = "auto"      # 自动推荐
    WITH_ALTERNATIVES = "alt"    # 带备选的推荐
    ASK_USER = "ask"             # 询问用户


@dataclass
class DecisionResult:
    """决策结果"""
    level: DecisionLevel         # 决策等级
    recommended_skill: str       # 推荐技能
    confidence: float            # 置信度
    breakdown: ConfidenceBreakdown  # 置信度分解

    # 中低置信度时提供
    alternative_skills: List[str] = field(default_factory=list)
    candidate_skills: List[Dict[str, Any]] = field(default_factory=list)

    # 推荐理由
    reasoning: str = ""

    # 是否需要用户确认
    requires_confirmation: bool = False


class DecisionEngine:
    """决策引擎"""

    def __init__(
        self,
        confidence_evaluator: ConfidenceEvaluator = None
    ):
        """初始化决策引擎

        Args:
            confidence_evaluator: 置信度评估器
        """
        self.evaluator = confidence_evaluator or ConfidenceEvaluator()

    def decide(
        self,
        search_results: List[SearchResult],
        stats_map: Dict[str, SkillStats]
    ) -> DecisionResult:
        """做出决策

        Args:
            search_results: 搜索结果列表
            stats_map: 技能统计映射

        Returns:
            决策结果
        """
        if not search_results:
            return self._no_match_result()

        # 评估所有结果
        evaluations = []
        for result in search_results:
            stats = stats_map.get(result.skill_name)
            breakdown = self.evaluator.evaluate(result, stats, len(search_results))
            evaluations.append((result, breakdown))

        # 按置信度排序
        evaluations.sort(key=lambda x: x[1].total, reverse=True)

        # 获取最佳结果
        best_result, best_breakdown = evaluations[0]

        # 根据置信度决定
        level = self.evaluator.get_confidence_level(best_breakdown.total)

        if level == "high":
            return self._auto_recommend(best_result, best_breakdown)
        elif level == "medium":
            alternatives = [e[0].skill_name for e in evaluations[1:3]]
            return self._recommend_with_alternatives(
                best_result, best_breakdown, alternatives
            )
        else:
            candidates = [
                {
                    "skill_name": r.skill_name,
                    "similarity": r.similarity,
                    "description": r.metadata.description
                }
                for r, _ in evaluations[:5]
            ]
            return self._ask_user(best_result, best_breakdown, candidates)

    def _auto_recommend(
        self,
        result: SearchResult,
        breakdown: ConfidenceBreakdown
    ) -> DecisionResult:
        """自动推荐

        Args:
            result: 最佳搜索结果
            breakdown: 置信度分解

        Returns:
            决策结果
        """
        reasoning = self._generate_reasoning(result, breakdown)

        return DecisionResult(
            level=DecisionLevel.AUTO_RECOMMEND,
            recommended_skill=result.skill_name,
            confidence=breakdown.total,
            breakdown=breakdown,
            reasoning=reasoning,
            requires_confirmation=False
        )

    def _recommend_with_alternatives(
        self,
        result: SearchResult,
        breakdown: ConfidenceBreakdown,
        alternatives: List[str]
    ) -> DecisionResult:
        """推荐带备选

        Args:
            result: 最佳搜索结果
            breakdown: 置信度分解
            alternatives: 备选技能列表

        Returns:
            决策结果
        """
        reasoning = self._generate_reasoning(result, breakdown, alternatives)

        return DecisionResult(
            level=DecisionLevel.WITH_ALTERNATIVES,
            recommended_skill=result.skill_name,
            confidence=breakdown.total,
            breakdown=breakdown,
            alternative_skills=alternatives,
            reasoning=reasoning,
            requires_confirmation=False
        )

    def _ask_user(
        self,
        result: SearchResult,
        breakdown: ConfidenceBreakdown,
        candidates: List[Dict[str, Any]]
    ) -> DecisionResult:
        """询问用户

        Args:
            result: 最佳搜索结果
            breakdown: 置信度分解
            candidates: 候选技能列表

        Returns:
            决策结果
        """
        reasoning = (
            f"找到 {len(candidates)} 个可能符合的技能，"
            f"需要你帮忙选择最合适的一个。"
        )

        return DecisionResult(
            level=DecisionLevel.ASK_USER,
            recommended_skill=result.skill_name,  # 仍然给出推荐
            confidence=breakdown.total,
            breakdown=breakdown,
            candidate_skills=candidates,
            reasoning=reasoning,
            requires_confirmation=True
        )

    def _no_match_result(self) -> DecisionResult:
        """无匹配结果

        Returns:
            决策结果
        """
        return DecisionResult(
            level=DecisionLevel.ASK_USER,
            recommended_skill="",
            confidence=0.0,
            breakdown=ConfidenceBreakdown(0, 0, 0, 0, 0),
            reasoning="未找到匹配的技能",
            requires_confirmation=True
        )

    def _generate_reasoning(
        self,
        result: SearchResult,
        breakdown: ConfidenceBreakdown,
        alternatives: List[str] = None
    ) -> str:
        """生成推荐理由

        Args:
            result: 搜索结果
            breakdown: 置信度分解
            alternatives: 备选技能

        Returns:
            推荐理由文本
        """
        parts = []

        # 主要理由
        parts.append(f"建议用 **{result.skill_name}**")

        # 描述
        if result.metadata.description:
            parts.append(f"- {result.metadata.description}")

        # 相似度
        if breakdown.similarity > 0.7:
            parts.append(f"- 与你的请求高度匹配（{breakdown.similarity:.0%}）")

        # 历史表现
        if breakdown.historical > 0.6:
            parts.append(f"- 历史使用效果良好（{breakdown.historical:.0%}）")

        # 备选
        if alternatives:
            alt_list = "、".join(alternatives[:3])
            parts.append(f"- 备选：{alt_list}")

        return "\n".join(parts)

"""
缓存管理模块

监控数据库大小，提供清理策略
"""

import os
from pathlib import Path
from typing import Optional, Callable, Dict, Any
from datetime import datetime, timedelta
from dataclasses import dataclass
from enum import Enum

from src.storage.history import HistoryTracker


class CleanupTrigger(Enum):
    """清理触发条件"""
    SIZE_LIMIT = "size"           # 大小超限
    COUNT_LIMIT = "count"         # 数量超限
    OLD_RECORDS = "old"           # 旧记录过多
    MANUAL = "manual"             # 手动触发


class CleanupStrategy:
    """清理策略"""

    @staticmethod
    def by_age(days: int = 30) -> dict:
        """按时间清理：删除 N 天前的记录

        Args:
            days: 保留最近 N 天的记录

        Returns:
            策略配置
        """
        return {
            "type": "age",
            "days": days,
            "description": f"删除 {days} 天前的记录"
        }

    @staticmethod
    def by_count(keep: int = 1000) -> dict:
        """按数量清理：保留最近 N 条

        Args:
            keep: 保留最近 N 条记录

        Returns:
            策略配置
        """
        return {
            "type": "count",
            "keep": keep,
            "description": f"保留最近 {keep} 条记录"
        }

    @staticmethod
    def by_success_rate(min_rate: float = 0.5) -> dict:
        """按成功率清理：删除低成功率记录

        Args:
            min_rate: 最低保留成功率

        Returns:
            策略配置
        """
        return {
            "type": "success_rate",
            "min_rate": min_rate,
            "description": f"删除成功率低于 {min_rate:.0%} 的记录"
        }

    @staticmethod
    def by_size(max_size_mb: int = 100) -> dict:
        """按大小清理：限制数据库大小

        Args:
            max_size_mb: 最大大小（MB）

        Returns:
            策略配置
        """
        return {
            "type": "size",
            "max_size_mb": max_size_mb,
            "description": f"限制数据库大小为 {max_size_mb}MB"
        }


@dataclass
class CleanupReport:
    """清理报告"""
    trigger: CleanupTrigger
    strategy_used: dict
    records_before: int
    records_after: int
    size_before_mb: float
    size_after_mb: float
    records_deleted: int


@dataclass
class HealthStatus:
    """缓存健康状态"""
    needs_cleanup: bool
    reason: str = ""
    suggestion: str = ""
    can_auto_cleanup: bool = True
    current_size_mb: float = 0
    record_count: int = 0


class CacheManager:
    """缓存管理器"""

    # 清理阈值
    THRESHOLDS = {
        "max_size_mb": 500,           # 最大数据库大小（MB）
        "max_records": 10000,          # 最大记录数
        "old_days": 30,                # 旧记录阈值（天）
        "old_ratio": 0.3               # 旧记录比例阈值
    }

    def __init__(
        self,
        history_tracker: HistoryTracker = None,
        data_dir: str = None
    ):
        """初始化缓存管理器

        Args:
            history_tracker: 历史记录追踪器
            data_dir: 数据目录
        """
        self.history = history_tracker or HistoryTracker()

        if data_dir is None:
            data_dir = os.path.join(
                os.path.dirname(__file__),
                "..", "..", "data"
            )

        self.data_dir = Path(data_dir)

    def check_health(self) -> HealthStatus:
        """检查缓存健康状态

        Returns:
            健康状态
        """
        # 获取当前状态
        size_bytes = self.history.get_db_size()
        size_mb = size_bytes / (1024 * 1024)
        record_count = self.history.get_record_count()

        # 检查各项阈值
        triggers = []

        if size_mb > self.THRESHOLDS["max_size_mb"]:
            triggers.append(
                f"数据库已达 {size_mb:.1f}MB"
            )

        if record_count > self.THRESHOLDS["max_records"]:
            triggers.append(
                f"已累积 {record_count:,} 条调用历史"
            )

        # TODO: 检查旧记录比例
        # old_ratio = self._calculate_old_record_ratio()
        # if old_ratio > self.THRESHOLDS["old_ratio"]:
        #     triggers.append(f"发现 {old_ratio:.0%} 的旧记录")

        if triggers:
            return HealthStatus(
                needs_cleanup=True,
                reason="；".join(triggers),
                description=self._suggest_cleanup(size_mb, record_count),
                can_auto_cleanup=self._is_auto_cleanup_safe(),
                current_size_mb=size_mb,
                record_count=record_count
            )

        return HealthStatus(
            needs_cleanup=False,
            current_size_mb=size_mb,
            record_count=record_count
        )

    def _suggest_cleanup(self, size_mb: float, record_count: int) -> str:
        """生成清理建议

        Args:
            size_mb: 当前大小
            record_count: 记录数量

        Returns:
            建议文本
        """
        if size_mb > self.THRESHOLDS["max_size_mb"]:
            return f"建议清理 30 天前的记录以释放空间"

        if record_count > self.THRESHOLDS["max_records"]:
            return f"建议保留最近 1000 条记录"

        return "建议进行缓存清理"

    def _is_auto_cleanup_safe(self) -> bool:
        """判断是否可以自动清理

        Returns:
            是否安全
        """
        # TODO: 实现更复杂的安全检查
        return True

    def cleanup(
        self,
        strategy: dict,
        dry_run: bool = False
    ) -> CleanupReport:
        """执行清理

        Args:
            strategy: 清理策略
            dry_run: 是否为演练模式（不实际删除）

        Returns:
            清理报告
        """
        # 记录清理前状态
        records_before = self.history.get_record_count()
        size_before = self.history.get_db_size()

        # 执行清理
        records_deleted = 0

        if strategy["type"] == "age":
            if not dry_run:
                records_deleted = self.history.cleanup_old_records(
                    days=strategy["days"]
                )
            else:
                records_deleted = self._estimate_age_cleanup(strategy["days"])

        elif strategy["type"] == "count":
            # TODO: 实现按数量清理
            pass

        elif strategy["type"] == "size":
            # TODO: 实现按大小清理
            pass

        # 记录清理后状态
        records_after = self.history.get_record_count()
        size_after = self.history.get_db_size()

        return CleanupReport(
            trigger=CleanupTrigger.MANUAL,
            strategy_used=strategy,
            records_before=records_before,
            records_after=records_after,
            size_before_mb=size_before / (1024 * 1024),
            size_after_mb=size_after / (1024 * 1024),
            records_deleted=records_deleted
        )

    def _estimate_age_cleanup(self, days: int) -> int:
        """估算按时间清理会删除多少记录

        Args:
            days: 天数

        Returns:
            估算的删除数量
        """
        # TODO: 实现估算逻辑
        return 0

    def auto_cleanup_if_needed(self) -> Optional[CleanupReport]:
        """如果需要则自动清理

        Returns:
            清理报告，如果不需要清理则返回 None
        """
        health = self.check_health()

        if not health.needs_cleanup:
            return None

        if not health.can_auto_cleanup:
            return None

        # 使用默认策略
        strategy = CleanupStrategy.by_age(days=30)
        return self.cleanup(strategy)

    def get_cache_stats(self) -> Dict[str, Any]:
        """获取缓存统计信息

        Returns:
            统计信息字典
        """
        health = self.check_health()

        return {
            "size_mb": health.current_size_mb,
            "record_count": health.record_count,
            "needs_cleanup": health.needs_cleanup,
            "reason": health.reason,
            "suggestion": health.suggestion
        }

    def clear_all(self) -> bool:
        """清空所有缓存

        Returns:
            是否成功
        """
        try:
            self.history.clear_all()
            return True
        except Exception:
            return False

"""
技能健康检查模块

检测技能是否缺失、损坏或版本不匹配
"""

import os
import hashlib
from pathlib import Path
from typing import List, Dict, Optional
from enum import Enum
from dataclasses import dataclass


class HealthStatus(Enum):
    """健康状态"""
    HEALTHY = "healthy"              # 技能正常
    MISSING = "missing"              # 技能不存在
    CORRUPTED = "corrupted"          # 技能损坏（如 SKILL.md 缺失）
    VERSION_MISMATCH = "outdated"    # 版本不匹配
    UNKNOWN = "unknown"              # 未知状态


@dataclass
class HealthReport:
    """健康报告"""
    skill_name: str
    status: HealthStatus
    message: str
    suggested_action: str = ""


class HealthChecker:
    """技能健康检查器"""

    def __init__(self, skills_dir: str = None):
        """初始化健康检查器

        Args:
            skills_dir: 技能目录路径
        """
        if skills_dir is None:
            skills_dir = os.path.expanduser("~/.claude/skills")

        self.skills_dir = Path(skills_dir)

    def check_skill(
        self,
        skill_name: str,
        expected_hash: str = None
    ) -> HealthReport:
        """检查单个技能的健康状态

        Args:
            skill_name: 技能名称
            expected_hash: 期望的文件哈希（用于版本检查）

        Returns:
            健康报告
        """
        skill_path = self.skills_dir / skill_name

        # 检查技能是否存在
        if not skill_path.exists():
            return HealthReport(
                skill_name=skill_name,
                status=HealthStatus.MISSING,
                message=f"技能目录不存在: {skill_path}",
                suggested_action="请重新安装该技能或检查技能目录"
            )

        # 检查关键文件
        required_files = ["SKILL.md"]
        missing_files = [
            f for f in required_files
            if not (skill_path / f).exists()
        ]

        if missing_files:
            return HealthReport(
                skill_name=skill_name,
                status=HealthStatus.CORRUPTED,
                message=f"缺少关键文件: {', '.join(missing_files)}",
                suggested_action="请重新安装该技能"
            )

        # 检查版本（如果提供了哈希）
        if expected_hash:
            current_hash = self._calculate_hash(skill_path)
            if current_hash != expected_hash:
                return HealthReport(
                    skill_name=skill_name,
                    status=HealthStatus.VERSION_MISMATCH,
                    message=f"技能版本已变更（哈希不匹配）",
                    suggested_action="建议更新技能元数据并迁移历史记录"
                )

        return HealthReport(
            skill_name=skill_name,
            status=HealthStatus.HEALTHY,
            message="技能健康",
            suggested_action=""
        )

    def check_all_skills(
        self,
        skill_list: List[str] = None,
        hash_map: Dict[str, str] = None
    ) -> List[HealthReport]:
        """检查所有技能的健康状态

        Args:
            skill_list: 技能名称列表，None 表示检查目录下所有技能
            hash_map: 技能哈希映射

        Returns:
            健康报告列表
        """
        if skill_list is None:
            skill_list = [
                d.name for d in self.skills_dir.iterdir()
                if d.is_dir() and not d.name.startswith(".")
            ]

        reports = []
        for skill_name in skill_list:
            expected_hash = hash_map.get(skill_name) if hash_map else None
            report = self.check_skill(skill_name, expected_hash)
            reports.append(report)

        return reports

    def _calculate_hash(self, skill_path: Path) -> str:
        """计算技能目录的哈希值

        Args:
            skill_path: 技能目录路径

        Returns:
            SHA256 哈希值
        """
        # 简单实现：计算 SKILL.md 的哈希
        skill_file = skill_path / "SKILL.md"
        if not skill_file.exists():
            return ""

        sha256 = hashlib.sha256()
        with open(skill_file, "rb") as f:
            sha256.update(f.read())

        return sha256.hexdigest()

    def get_unhealthy_skills(
        self,
        skill_list: List[str] = None
    ) -> List[HealthReport]:
        """获取所有不健康的技能

        Args:
            skill_list: 技能名称列表

        Returns:
            不健康技能的报告列表
        """
        all_reports = self.check_all_skills(skill_list)
        return [
            r for r in all_reports
            if r.status != HealthStatus.HEALTHY
        ]

    def fix_missing_skill(self, skill_name: str) -> bool:
        """尝试修复缺失的技能

        Args:
            skill_name: 技能名称

        Returns:
            是否修复成功
        """
        # TODO: 实现自动修复逻辑
        # 例如：从备份恢复、从 GitHub 重新下载等
        return False

"""
生命周期管理模块

处理技能的版本迁移、降级策略等生命周期问题
"""

from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum

from src.maintenance.health import HealthStatus, HealthChecker


class MigrationAction(Enum):
    """迁移动作"""
    NONE = "none"                  # 无需操作
    UPDATE_METADATA = "update"     # 更新元数据
    MIGRATE_HISTORY = "migrate"    # 迁移历史记录
    REMOVE_FROM_INDEX = "remove"   # 从索引移除


@dataclass
class MigrationPlan:
    """迁移计划"""
    skill_name: str
    action: MigrationAction
    reason: str
    steps: List[str]


class LifecycleManager:
    """生命周期管理器"""

    # 降级映射：主技能不可用时使用的备选技能
    FALLBACK_MAP = {
        "pdf-skill": ["docx-skill", "generic-document-skill"],
        "docx-skill": ["pdf-skill", "generic-document-skill"],
        "autogpt-agents": ["agent-call", "manual-creation"],
        "brainstorming": ["manual-thinking"],
        # 用户可以继续添加...
    }

    # 版本迁移映射
    VERSION_MIGRATIONS = {
        "pdf-skill": {
            "v1": "migrate_pdf_v1_to_v2",
            "v2": "current"
        }
        # 用户可以继续添加...
    }

    def __init__(self, health_checker: HealthChecker = None):
        """初始化生命周期管理器

        Args:
            health_checker: 健康检查器
        """
        self.health_checker = health_checker or HealthChecker()

    def plan_migration(
        self,
        skill_name: str,
        health_status: HealthStatus
    ) -> MigrationPlan:
        """规划迁移方案

        Args:
            skill_name: 技能名称
            health_status: 健康状态

        Returns:
            迁移计划
        """
        if health_status == HealthStatus.HEALTHY:
            return MigrationPlan(
                skill_name=skill_name,
                action=MigrationAction.NONE,
                reason="技能健康，无需迁移",
                steps=[]
            )

        elif health_status == HealthStatus.MISSING:
            return MigrationPlan(
                skill_name=skill_name,
                action=MigrationAction.REMOVE_FROM_INDEX,
                reason="技能已移除，从索引中删除",
                steps=[
                    f"1. 标记 {skill_name} 为已移除",
                    "2. 保留历史记录以供参考",
                    "3. 推荐备选技能"
                ]
            )

        elif health_status == HealthStatus.VERSION_MISMATCH:
            return MigrationPlan(
                skill_name=skill_name,
                action=MigrationAction.MIGRATE_HISTORY,
                reason="技能版本已更新",
                steps=[
                    "1. 计算新的文件哈希",
                    "2. 更新技能元数据",
                    "3. 迁移历史记录到新版本",
                    "4. 验证迁移结果"
                ]
            )

        elif health_status == HealthStatus.CORRUPTED:
            return MigrationPlan(
                skill_name=skill_name,
                action=MigrationAction.REMOVE_FROM_INDEX,
                reason="技能损坏",
                steps=[
                    "1. 从索引中移除",
                    "2. 建议用户重新安装"
                ]
            )

        return MigrationPlan(
            skill_name=skill_name,
            action=MigrationAction.NONE,
            reason="未知状态",
            steps=[]
        )

    def get_fallback_skills(self, skill_name: str) -> List[str]:
        """获取备选技能列表

        Args:
            skill_name: 主技能名称

        Returns:
            备选技能名称列表
        """
        return self.FALLBACK_MAP.get(skill_name, [])

    def add_fallback(self, primary: str, fallback: List[str]):
        """添加降级映射

        Args:
            primary: 主技能名称
            fallback: 备选技能列表
        """
        self.FALLBACK_MAP[primary] = fallback

    def check_migrations_needed(
        self,
        skill_list: List[str],
        hash_map: Dict[str, str]
    ) -> List[MigrationPlan]:
        """检查所有需要迁移的技能

        Args:
            skill_list: 技能列表
            hash_map: 技能哈希映射

        Returns:
            迁移计划列表
        """
        plans = []

        for skill_name in skill_list:
            expected_hash = hash_map.get(skill_name)
            report = self.health_checker.check_skill(skill_name, expected_hash)

            if report.status != HealthStatus.HEALTHY:
                plan = self.plan_migration(skill_name, report.status)
                plans.append(plan)

        return plans

    def migrate_skill_version(
        self,
        skill_name: str,
        old_version: str,
        new_version: str
    ) -> bool:
        """迁移技能版本

        Args:
            skill_name: 技能名称
            old_version: 旧版本
            new_version: 新版本

        Returns:
            是否迁移成功
        """
        # TODO: 实现实际的版本迁移逻辑
        # 这里需要更新数据库中的版本信息和哈希

        return True

    def deprecate_skill(
        self,
        skill_name: str,
        replacement: str = None
    ) -> bool:
        """弃用技能

        Args:
            skill_name: 要弃用的技能
            replacement: 替代技能

        Returns:
            是否成功
        """
        # TODO: 实现弃用逻辑
        # 1. 标记技能为已弃用
        # 2. 如果有替代技能，建立映射
        # 3. 更新推荐逻辑

        if replacement:
            self.add_fallback(skill_name, [replacement])

        return True

"""
Embedding 模型封装

支持中英双语的文本 embedding，使用 sentence-transformers
"""

from functools import lru_cache
from typing import List, Union
import numpy as np


class EmbeddingModel:
    """文本 Embedding 模型"""

    # 默认模型：支持中英双语的轻量级模型
    DEFAULT_MODEL = "paraphrase-multilingual-MiniLM-L12-v2"

    def __init__(self, model_name: str = DEFAULT_MODEL, device: str = "cpu"):
        """初始化 Embedding 模型

        Args:
            model_name: 模型名称，默认使用多语言模型
            device: 运行设备，"cpu" 或 "cuda"
        """
        self.model_name = model_name
        self.device = device
        self._model = None
        self._load_model()

    def _load_model(self):
        """延迟加载模型"""
        if self._model is None:
            try:
                from sentence_transformers import SentenceTransformer
                self._model = SentenceTransformer(self.model_name, device=self.device)
            except ImportError:
                raise ImportError(
                    "sentence-transformers 未安装，请运行: pip install sentence-transformers"
                )

    def encode(
        self,
        texts: Union[str, List[str]],
        normalize: bool = True,
        batch_size: int = 32
    ) -> np.ndarray:
        """将文本编码为向量

        Args:
            texts: 单个文本或文本列表
            normalize: 是否归一化向量
            batch_size: 批处理大小

        Returns:
            向量数组，形状为 (num_texts, embedding_dim)
        """
        if isinstance(texts, str):
            texts = [texts]

        embeddings = self._model.encode(
            texts,
            normalize_embeddings=normalize,
            batch_size=batch_size,
            show_progress_bar=False
        )

        return embeddings

    def similarity(self, text1: Union[str, np.ndarray], text2: Union[str, np.ndarray]) -> float:
        """计算两个文本的相似度

        Args:
            text1: 文本或向量
            text2: 文本或向量

        Returns:
            相似度分数 [0, 1]
        """
        vec1 = self.encode(text1) if isinstance(text1, str) else text1
        vec2 = self.encode(text2) if isinstance(text2, str) else text2

        # 余弦相似度
        return float(np.dot(vec1, vec2.T).flatten()[0])

    def similarities(
        self,
        query: Union[str, np.ndarray],
        candidates: List[str]
    ) -> List[float]:
        """计算查询与多个候选文本的相似度

        Args:
            query: 查询文本或向量
            candidates: 候选文本列表

        Returns:
            相似度分数列表
        """
        query_vec = self.encode(query) if isinstance(query, str) else query
        candidate_vecs = self.encode(candidates)

        # 批量计算余弦相似度
        similarities = np.dot(candidate_vecs, query_vec.T).flatten()
        return similarities.tolist()

    @property
    def dimension(self) -> int:
        """返回向量维度"""
        return self._model.get_sentence_embedding_dimension()

    @staticmethod
    @lru_cache(maxsize=1)
    def get_default() -> "EmbeddingModel":
        """获取默认的单例模型"""
        return EmbeddingModel()


# 便捷函数
def encode_text(text: str, model: EmbeddingModel = None) -> np.ndarray:
    """便捷函数：编码单个文本"""
    if model is None:
        model = EmbeddingModel.get_default()
    return model.encode(text)[0]


def compute_similarity(text1: str, text2: str, model: EmbeddingModel = None) -> float:
    """便捷函数：计算两个文本的相似度"""
    if model is None:
        model = EmbeddingModel.get_default()
    return model.similarity(text1, text2)

"""
SparkSatchel 灵犀妙计 - 主入口

智能技能检索与推荐系统
"""

import os
from pathlib import Path
from typing import List, Optional, Dict, Any
from datetime import datetime
from dataclasses import dataclass, field

from src.models.embedding import EmbeddingModel
from src.storage.vector_db import VectorStore, SkillMetadata, SearchResult
from src.storage.history import HistoryTracker, SkillCall, SkillStats
from src.analysis.intent import IntentAnalyzer
from src.analysis.confidence import ConfidenceEvaluator
from src.decision import DecisionEngine, DecisionResult
from src.maintenance.health import HealthChecker, HealthStatus
from src.maintenance.lifecycle import LifecycleManager
from src.maintenance.cache import CacheManager, CleanupStrategy


@dataclass
class RetrievalResult:
    """检索结果"""
    confidence: float                    # 置信度 0-1
    recommended_skill: str               # 推荐的技能
    reasoning: str                       # 推荐理由

    # 中置信度时提供备选
    alternative_skills: List[str] = field(default_factory=list)

    # 低置信度时提供候选列表
    candidate_skills: List[Dict[str, Any]] = field(default_factory=list)

    # 是否需要用户确认
    requires_confirmation: bool = False

    # 内部数据（调试用）
    intent: str = ""
    matched_skills: List[str] = field(default_factory=list)


class SparkSatchel:
    """灵犀妙计 - 智能技能检索器"""

    def __init__(
        self,
        skills_dir: str = None,
        data_dir: str = None,
        auto_load: bool = True
    ):
        """初始化 SparkSatchel

        Args:
            skills_dir: 技能目录
            data_dir: 数据目录
            auto_load: 是否自动加载现有技能
        """
        # 目录设置
        if skills_dir is None:
            skills_dir = os.path.expanduser("~/.claude/skills")
        self.skills_dir = Path(skills_dir)

        if data_dir is None:
            data_dir = os.path.join(
                os.path.dirname(__file__),
                "..", "data"
            )
        self.data_dir = Path(data_dir)
        self.data_dir.mkdir(parents=True, exist_ok=True)

        # 初始化组件
        self.embedding_model = EmbeddingModel.get_default()
        self.vector_store = VectorStore(
            persist_directory=str(self.data_dir / "collections"),
            embedding_model=self.embedding_model
        )
        self.history = HistoryTracker(
            db_path=str(self.data_dir / "history.db")
        )
        self.intent_analyzer = IntentAnalyzer()
        self.confidence_evaluator = ConfidenceEvaluator()
        self.decision_engine = DecisionEngine(self.confidence_evaluator)
        self.health_checker = HealthChecker(str(self.skills_dir))
        self.lifecycle_manager = LifecycleManager(self.health_checker)
        self.cache_manager = CacheManager(self.history, str(self.data_dir))

        # 加载技能
        if auto_load:
            self._load_existing_skills()

    def _load_existing_skills(self):
        """加载现有技能到向量库"""
        # TODO: 扫描技能目录并加载
        pass

    def retrieve(self, user_request: str) -> RetrievalResult:
        """检索并推荐技能

        Args:
            user_request: 用户请求文本

        Returns:
            检索结果
        """
        # 1. 意图分析
        intent = self.intent_analyzer.analyze(user_request)

        # 2. 向量检索
        search_results = self.vector_store.search(
            query=user_request,
            top_k=10,
            min_similarity=0.3
        )

        # 3. 获取历史统计
        skill_names = [r.skill_name for r in search_results]
        stats_map = {}
        for name in skill_names:
            stats = self.history.get_skill_stats(name)
            if stats:
                stats_map[name] = stats

        # 4. 决策
        decision = self.decision_engine.decide(search_results, stats_map)

        # 5. 记录这次检索
        self._record_retrieval(user_request, intent, decision)

        # 6. 构建返回结果
        return RetrievalResult(
            confidence=decision.confidence,
            recommended_skill=decision.recommended_skill,
            reasoning=decision.reasoning,
            alternative_skills=decision.alternative_skills,
            candidate_skills=decision.candidate_skills,
            requires_confirmation=decision.requires_confirmation,
            intent=intent.primary,
            matched_skills=[r.skill_name for r in search_results]
        )

    def _record_retrieval(
        self,
        user_request: str,
        intent,
        decision: DecisionResult
    ):
        """记录检索到历史

        Args:
            user_request: 用户请求
            intent: 意图对象
            decision: 决策结果
        """
        call = SkillCall(
            id=None,
            timestamp=datetime.now().isoformat(),
            user_request=user_request,
            intent=intent.primary,
            matched_skills=decision.alternative_skills + [decision.recommended_skill],
            recommended_skill=decision.recommended_skill,
            confidence=decision.confidence,
            user_accepted=False,  # 待用户确认后更新
            execution_success=None,
            user_feedback=""
        )

        # 保存到待确认队列
        # TODO: 实现待确认队列
        self._pending_call = call

    def feedback(
        self,
        skill_name: str,
        success: bool,
        feedback: str = ""
    ):
        """记录用户反馈

        Args:
            skill_name: 技能名称
            success: 是否成功
            feedback: 用户反馈
        """
        if not hasattr(self, '_pending_call'):
            return

        # 更新待确认的调用记录
        self._pending_call.user_accepted = True
        self._pending_call.execution_success = success
        self._pending_call.user_feedback = feedback

        # 记录到历史
        self.history.record_call(self._pending_call)

        # 清除待确认记录
        delattr(self, '_pending_call')

    def add_skill(self, metadata: SkillMetadata):
        """添加技能到向量库

        Args:
            metadata: 技能元数据
        """
        self.vector_store.add_skill(metadata)

    def remove_skill(self, skill_name: str, category: str):
        """从向量库移除技能

        Args:
            skill_name: 技能名称
            category: 技能分类
        """
        self.vector_store.remove_skill(skill_name, category)

    def check_health(self) -> Dict[str, Any]:
        """检查系统健康状态

        Returns:
            健康状态报告
        """
        # 检查缓存健康
        cache_health = self.cache_manager.check_health()

        # 检查技能健康
        all_stats = self.history.get_all_stats()
        skill_names = list({s.skill_name for s in all_stats})
        unhealthy = self.health_checker.get_unhealthy_skills(skill_names)

        return {
            "cache": {
                "needs_cleanup": cache_health.needs_cleanup,
                "size_mb": cache_health.current_size_mb,
                "record_count": cache_health.record_count,
                "reason": cache_health.reason
            },
            "skills": {
                "unhealthy_count": len(unhealthy),
                "unhealthy_list": [
                    {"skill": r.skill_name, "status": r.status.value}
                    for r in unhealthy[:10]  # 最多显示10个
                ]
            },
            "suggestion": self._generate_health_suggestion(cache_health, unhealthy)
        }

    def _generate_health_suggestion(
        self,
        cache_health,
        unhealthy_skills
    ) -> str:
        """生成健康建议

        Args:
            cache_health: 缓存健康状态
            unhealthy_skills: 不健康技能列表

        Returns:
            建议文本
        """
        suggestions = []

        if cache_health.needs_cleanup:
            suggestions.append(f"🗑️ {cache_health.suggestion}")

        if unhealthy_skills:
            suggestions.append(
                f"⚠️ 发现 {len(unhealthy_skills)} 个技能状态异常"
            )

        return "；".join(suggestions) if suggestions else "系统健康"

    def cleanup(self, strategy: dict = None):
        """执行缓存清理

        Args:
            strategy: 清理策略，None 则使用默认策略
        """
        if strategy is None:
            strategy = CleanupStrategy.by_age(days=30)

        report = self.cache_manager.cleanup(strategy)

        return {
            "records_deleted": report.records_deleted,
            "size_before_mb": report.size_before_mb,
            "size_after_mb": report.size_after_mb,
            "freed_mb": report.size_before_mb - report.size_after_mb
        }

    def get_stats(self) -> Dict[str, Any]:
        """获取系统统计信息

        Returns:
            统计信息
        """
        cache_stats = self.cache_manager.get_cache_stats()
        skill_counts = self.vector_store.get_skill_count()

        return {
            "cache": cache_stats,
            "skills": {
                "total": sum(skill_counts.values()),
                "by_category": skill_counts
            }
        }

    def index_skills(self, force: bool = False):
        """扫描并索引技能目录

        Args:
            force: 是否强制重新索引
        """
        indexed_count = 0

        for skill_path in self.skills_dir.iterdir():
            if not skill_path.is_dir() or skill_path.name.startswith("."):
                continue

            # 读取 SKILL.md
            skill_md = skill_path / "SKILL.md"
            if not skill_md.exists():
                continue

            # 解析技能元数据
            metadata = self._parse_skill_md(skill_path.name, skill_md)

            if metadata:
                self.add_skill(metadata)
                indexed_count += 1

        return indexed_count

    def _parse_skill_md(self, skill_name: str, skill_md_path: Path) -> Optional[SkillMetadata]:
        """解析 SKILL.md 文件

        Args:
            skill_name: 技能名称
            skill_md_path: SKILL.md 文件路径

        Returns:
            技能元数据
        """
        try:
            content = skill_md_path.read_text(encoding="utf-8")

            # 解析 YAML frontmatter
            # TODO: 实现 YAML 解析

            # 简单实现：提取描述
            description = ""
            for line in content.split("\n"):
                if "description:" in line.lower():
                    description = line.split(":", 1)[1].strip()
                    break

            return SkillMetadata(
                name=skill_name,
                path=str(skill_md_path.parent),
                description=description,
                tags=[],
                trigger_keywords=[],
                category="utility"  # 默认分类
            )

        except Exception:
            return None


# 单例模式
_instance = None


def get_instance() -> SparkSatchel:
    """获取 SparkSatchel 单例

    Returns:
        SparkSatchel 实例
    """
    global _instance
    if _instance is None:
        _instance = SparkSatchel()
    return _instance