diff --git a/.claude/skills/learned/SKILL.md b/.claude/skills/learned/SKILL.md new file mode 100644 index 00000000..ea63251d --- /dev/null +++ b/.claude/skills/learned/SKILL.md @@ -0,0 +1,73 @@ +--- +name: learned-skills-index +description: Index directory for automatically learned skills from execution feedback +type: index +--- + +# Learned Skills Index + +This directory contains skills that have been automatically extracted from successful +execution sessions. These skills represent patterns that led to quality improvements +and are available for retrieval in future tasks. + +## How Skills Are Learned + +1. **Execution**: Claude Code runs a task with `--loop` enabled +2. **Feedback Collection**: Each iteration's quality scores and improvements are recorded +3. **Pattern Extraction**: Successful patterns are extracted from quality-improving iterations +4. **Skill Generation**: Patterns are compiled into a learnable skill definition +5. **Promotion Gate**: Skills must meet quality thresholds before promotion + +## Skill Status + +| Status | Meaning | +|--------|---------| +| **Pending** | Skill extracted but not yet promoted (needs more validation) | +| **Promoted** | Skill has been validated and can be applied to new tasks | +| **Archived** | Skill deprecated or superseded by newer learning | + +## Quality Thresholds for Promotion + +- Minimum quality score: 85.0 +- Minimum successful applications: 2 +- Minimum success rate: 70% + +## Directory Structure + +``` +learned/ +├── SKILL.md # This index file +├── learned-backend-auth/ # Example learned skill +│ ├── SKILL.md # Skill definition +│ └── metadata.json # Machine-readable metadata +└── learned-frontend-form/ # Another learned skill + ├── SKILL.md + └── metadata.json +``` + +## Integration + +Learned skills are automatically retrieved based on: +- Task description keywords +- File types being modified +- Domain context + +To manually query learned skills: +```python +from core.skill_persistence import retrieve_skills_for_task + +skills = retrieve_skills_for_task( + task_description="your task here", + domain="backend" # optional +) +``` + +## Safety + +All learned skills include full provenance: +- Source session ID +- Source repository +- Timestamp of learning +- Quality progression + +This enables rollback and audit of any behavioral changes from learned skills. diff --git a/.claude/skills/sc-implement/SKILL.md b/.claude/skills/sc-implement/SKILL.md index d7f3e173..7a3ac744 100644 --- a/.claude/skills/sc-implement/SKILL.md +++ b/.claude/skills/sc-implement/SKILL.md @@ -88,8 +88,52 @@ This skill requires evidence. You MUST: /sc:implement "enterprise auth system" --orchestrate --strategy systematic --delegate ``` +## Loop Mode & Learning + +When using `--loop`, this skill integrates with the skill persistence layer for cross-session learning: + +### How Learning Works + +1. **Feedback Recording** - Each iteration's quality scores and improvements are persisted +2. **Skill Extraction** - Successful patterns are extracted when quality threshold is met +3. **Skill Retrieval** - Relevant learned skills are injected into subsequent tasks +4. **Effectiveness Tracking** - Applied skills are tracked for success rate + +### Loop Flags + +| Flag | Type | Default | Description | +|------|------|---------|-------------| +| `--loop` | int | 3 | Enable iterative improvement (max 5) | +| `--learn` | bool | true | Enable learning from this session | +| `--auto-promote` | bool | false | Auto-promote high-quality skills | + +### Example with Learning + +```bash +# Iterative implementation with learning +/sc:implement auth flow --loop 3 --learn + +# View learned skills +python scripts/skill_learn.py '{"command": "stats"}' + +# Retrieve relevant skills +python scripts/skill_learn.py '{"command": "retrieve", "task": "auth"}' +``` + +### Learned Skills Location + +Promoted skills are stored in: +``` +.claude/skills/learned/ +├── SKILL.md # Index +├── learned-backend-auth/ # Example promoted skill +│ ├── SKILL.md +│ └── metadata.json +``` + ## Resources - [PERSONAS.md](PERSONAS.md) - Available persona definitions - [scripts/select_agent.py](scripts/select_agent.py) - Agent selection logic - [scripts/evidence_gate.py](scripts/evidence_gate.py) - Evidence validation +- [scripts/skill_learn.py](scripts/skill_learn.py) - Skill learning management diff --git a/.claude/skills/sc-implement/scripts/skill_learn.py b/.claude/skills/sc-implement/scripts/skill_learn.py new file mode 100644 index 00000000..b49874d1 --- /dev/null +++ b/.claude/skills/sc-implement/scripts/skill_learn.py @@ -0,0 +1,466 @@ +#!/usr/bin/env python3 +""" +Skill Learning CLI for SuperClaude. + +Provides commands for managing learned skills: +- list: Show learned skills and their status +- promote: Promote a skill to permanent status +- stats: Show learning statistics +- retrieve: Find relevant skills for a task +- export: Export skills to SKILL.md files + +Usage: + python skill_learn.py '{"command": "list"}' + python skill_learn.py '{"command": "promote", "skill_id": "learned-abc123"}' + python skill_learn.py '{"command": "stats"}' + python skill_learn.py '{"command": "retrieve", "task": "implement auth", "domain": "backend"}' +""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path +from typing import Any + +# Load skill_persistence directly to avoid core/__init__.py import issues +SUPERCLAUD_ROOT = Path(__file__).parent.parent.parent.parent.parent +SKILL_PERSISTENCE_PATH = SUPERCLAUD_ROOT / "core" / "skill_persistence.py" + +spec = importlib.util.spec_from_file_location("skill_persistence", SKILL_PERSISTENCE_PATH) +sp = importlib.util.module_from_spec(spec) +sys.modules["skill_persistence"] = sp +spec.loader.exec_module(sp) + +LearnedSkill = sp.LearnedSkill +PromotionGate = sp.PromotionGate +SkillExtractor = sp.SkillExtractor +SkillRetriever = sp.SkillRetriever +SkillStore = sp.SkillStore + + +def handle_list(args: dict[str, Any]) -> dict[str, Any]: + """List learned skills with optional filters.""" + store = SkillStore() + + domain = args.get("domain") + promoted_only = args.get("promoted_only", False) + min_quality = args.get("min_quality", 0.0) + + conn = store._get_connection() + + sql = "SELECT * FROM learned_skills WHERE quality_score >= ?" + params: list[Any] = [min_quality] + + if domain: + sql += " AND domain = ?" + params.append(domain) + + if promoted_only: + sql += " AND promoted = 1" + + sql += " ORDER BY quality_score DESC, learned_at DESC" + + rows = conn.execute(sql, params).fetchall() + + skills = [] + for row in rows: + skill = store._row_to_skill(row) + skills.append( + { + "skill_id": skill.skill_id, + "name": skill.name, + "domain": skill.domain, + "quality_score": skill.quality_score, + "promoted": skill.promoted, + "patterns_count": len(skill.patterns), + "learned_at": skill.learned_at, + "source_session": skill.source_session[:8] if skill.source_session else "", + } + ) + + return { + "success": True, + "count": len(skills), + "skills": skills, + } + + +def handle_promote(args: dict[str, Any]) -> dict[str, Any]: + """Promote a skill to permanent status.""" + skill_id = args.get("skill_id") + reason = args.get("reason", "Manual promotion") + + if not skill_id: + return {"success": False, "error": "skill_id is required"} + + store = SkillStore() + gate = PromotionGate(store) + + skill = store.get_skill(skill_id) + if skill is None: + return {"success": False, "error": f"Skill not found: {skill_id}"} + + # Check if already promoted + if skill.promoted: + return {"success": False, "error": "Skill already promoted"} + + # Evaluate + can_promote, eval_reason = gate.evaluate(skill) + if not can_promote: + return { + "success": False, + "error": f"Skill does not meet promotion criteria: {eval_reason}", + "skill_id": skill_id, + "quality_score": skill.quality_score, + } + + # Promote + path = gate.promote(skill, reason) + if path is None: + return {"success": False, "error": "Promotion failed"} + + return { + "success": True, + "skill_id": skill_id, + "name": skill.name, + "promoted_to": str(path), + "reason": reason, + } + + +def handle_stats(args: dict[str, Any]) -> dict[str, Any]: + """Get learning statistics.""" + store = SkillStore() + conn = store._get_connection() + + # Skill counts + skill_stats = conn.execute(""" + SELECT + COUNT(*) as total, + SUM(CASE WHEN promoted = 1 THEN 1 ELSE 0 END) as promoted, + AVG(quality_score) as avg_quality, + MAX(quality_score) as max_quality, + MIN(quality_score) as min_quality + FROM learned_skills + """).fetchone() + + # Domain breakdown + domain_stats = conn.execute(""" + SELECT domain, COUNT(*) as count, AVG(quality_score) as avg_quality + FROM learned_skills + GROUP BY domain + ORDER BY count DESC + """).fetchall() + + # Feedback stats + feedback_stats = conn.execute(""" + SELECT + COUNT(*) as total_feedback, + COUNT(DISTINCT session_id) as sessions, + AVG(quality_after - quality_before) as avg_improvement + FROM iteration_feedback + """).fetchone() + + # Application stats + app_stats = conn.execute(""" + SELECT + COUNT(*) as total_applications, + SUM(CASE WHEN was_helpful = 1 THEN 1 ELSE 0 END) as helpful, + SUM(CASE WHEN was_helpful = 0 THEN 1 ELSE 0 END) as unhelpful, + AVG(quality_impact) as avg_impact + FROM skill_applications + """).fetchone() + + # Recent learning activity + recent = conn.execute(""" + SELECT skill_id, name, quality_score, learned_at + FROM learned_skills + ORDER BY learned_at DESC + LIMIT 5 + """).fetchall() + + return { + "success": True, + "skills": { + "total": skill_stats["total"] or 0, + "promoted": skill_stats["promoted"] or 0, + "pending": (skill_stats["total"] or 0) - (skill_stats["promoted"] or 0), + "avg_quality": round(skill_stats["avg_quality"] or 0, 1), + "max_quality": round(skill_stats["max_quality"] or 0, 1), + "min_quality": round(skill_stats["min_quality"] or 0, 1), + }, + "domains": [ + { + "domain": row["domain"] or "general", + "count": row["count"], + "avg_quality": round(row["avg_quality"] or 0, 1), + } + for row in domain_stats + ], + "feedback": { + "total_records": feedback_stats["total_feedback"] or 0, + "sessions": feedback_stats["sessions"] or 0, + "avg_improvement": round(feedback_stats["avg_improvement"] or 0, 2), + }, + "applications": { + "total": app_stats["total_applications"] or 0, + "helpful": app_stats["helpful"] or 0, + "unhelpful": app_stats["unhelpful"] or 0, + "success_rate": round( + (app_stats["helpful"] or 0) / app_stats["total_applications"] + if app_stats["total_applications"] + else 0, + 2, + ), + "avg_quality_impact": round(app_stats["avg_impact"] or 0, 2), + }, + "recent_skills": [ + { + "skill_id": row["skill_id"], + "name": row["name"], + "quality": row["quality_score"], + "learned_at": row["learned_at"], + } + for row in recent + ], + } + + +def handle_retrieve(args: dict[str, Any]) -> dict[str, Any]: + """Retrieve relevant skills for a task.""" + task = args.get("task", "") + domain = args.get("domain") + files = args.get("files", []) + max_skills = args.get("max_skills", 5) + promoted_only = args.get("promoted_only", False) + + if not task: + return {"success": False, "error": "task is required"} + + store = SkillStore() + retriever = SkillRetriever(store) + + results = retriever.retrieve( + task_description=task, + file_paths=files, + domain=domain, + max_skills=max_skills, + promoted_only=promoted_only, + ) + + skills = [] + for skill, score in results: + skills.append( + { + "skill_id": skill.skill_id, + "name": skill.name, + "relevance": round(score, 2), + "quality_score": skill.quality_score, + "domain": skill.domain, + "promoted": skill.promoted, + "patterns": skill.patterns[:3], + "anti_patterns": skill.anti_patterns[:2], + "conditions": skill.applicability_conditions, + } + ) + + return { + "success": True, + "task": task, + "domain": domain, + "found": len(skills), + "skills": skills, + } + + +def handle_export(args: dict[str, Any]) -> dict[str, Any]: + """Export a skill to SKILL.md format.""" + skill_id = args.get("skill_id") + output_dir = args.get("output_dir") + + if not skill_id: + return {"success": False, "error": "skill_id is required"} + + store = SkillStore() + skill = store.get_skill(skill_id) + + if skill is None: + return {"success": False, "error": f"Skill not found: {skill_id}"} + + skill_md = skill.to_skill_md() + + if output_dir: + # Use skill_id for directory name to prevent path traversal + output_path = Path(output_dir) / skill.skill_id / "SKILL.md" + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(skill_md) + + return { + "success": True, + "skill_id": skill_id, + "exported_to": str(output_path), + } + + return { + "success": True, + "skill_id": skill_id, + "skill_md": skill_md, + } + + +def handle_pending(args: dict[str, Any]) -> dict[str, Any]: + """List skills pending promotion.""" + store = SkillStore() + gate = PromotionGate(store) + pending = gate.list_pending() + + skills = [] + for skill in pending: + can_promote, reason = gate.evaluate(skill) + effectiveness = store.get_skill_effectiveness(skill.skill_id) + + skills.append( + { + "skill_id": skill.skill_id, + "name": skill.name, + "quality_score": skill.quality_score, + "domain": skill.domain, + "can_promote": can_promote, + "reason": reason, + "applications": effectiveness["applications"], + "success_rate": round(effectiveness["success_rate"], 2), + "learned_at": skill.learned_at, + } + ) + + return { + "success": True, + "count": len(skills), + "skills": skills, + } + + +def handle_delete(args: dict[str, Any]) -> dict[str, Any]: + """Delete a learned skill.""" + skill_id = args.get("skill_id") + + if not skill_id: + return {"success": False, "error": "skill_id is required"} + + store = SkillStore() + conn = store._get_connection() + + # Check if exists + skill = store.get_skill(skill_id) + if skill is None: + return {"success": False, "error": f"Skill not found: {skill_id}"} + + # Delete from database + try: + conn.execute("DELETE FROM learned_skills WHERE skill_id = ?", (skill_id,)) + conn.execute("DELETE FROM skill_applications WHERE skill_id = ?", (skill_id,)) + conn.commit() + except Exception as e: + return {"success": False, "error": f"Database error: {e}"} + + # Remove files if promoted (use skill_id for directory name, not skill.name) + if skill.promoted: + # Use skill_id for directory name to match PromotionGate.promote() + skill_dir = Path.home() / ".claude" / "skills" / "learned" / skill.skill_id + if skill_dir.exists(): + import shutil + + shutil.rmtree(skill_dir) + + return { + "success": True, + "skill_id": skill_id, + "name": skill.name, + "deleted": True, + } + + +HANDLERS = { + "list": handle_list, + "promote": handle_promote, + "stats": handle_stats, + "retrieve": handle_retrieve, + "export": handle_export, + "pending": handle_pending, + "delete": handle_delete, +} + + +def main(): + """Main entry point.""" + if len(sys.argv) < 2: + print( + json.dumps( + { + "success": False, + "error": 'Usage: skill_learn.py \'{"command": "...", ...}\'', + "available_commands": list(HANDLERS.keys()), + } + ) + ) + sys.exit(1) + + try: + args = json.loads(sys.argv[1]) + except json.JSONDecodeError as e: + print( + json.dumps( + { + "success": False, + "error": f"Invalid JSON: {e}", + } + ) + ) + sys.exit(1) + + command = args.get("command") + if not command: + print( + json.dumps( + { + "success": False, + "error": "Missing 'command' field", + "available_commands": list(HANDLERS.keys()), + } + ) + ) + sys.exit(1) + + handler = HANDLERS.get(command) + if not handler: + print( + json.dumps( + { + "success": False, + "error": f"Unknown command: {command}", + "available_commands": list(HANDLERS.keys()), + } + ) + ) + sys.exit(1) + + try: + result = handler(args) + print(json.dumps(result, indent=2)) + except Exception as e: + print( + json.dumps( + { + "success": False, + "error": str(e), + "command": command, + } + ) + ) + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/core/loop_orchestrator.py b/core/loop_orchestrator.py index 8e84ea3a..ed17a97f 100644 --- a/core/loop_orchestrator.py +++ b/core/loop_orchestrator.py @@ -14,8 +14,10 @@ - Claude Code executes Skills and processes signals """ +from __future__ import annotations + import time -from typing import Any, Callable +from typing import Any, Callable, Dict, Optional from .pal_integration import PALReviewSignal, incorporate_pal_feedback from .quality_assessment import QualityAssessor @@ -53,7 +55,7 @@ class LoopOrchestrator: execute the sc-implement skill and return evidence. """ - def __init__(self, config: LoopConfig | None = None): + def __init__(self, config: Optional[LoopConfig] = None): """ Initialize the loop orchestrator. @@ -268,7 +270,7 @@ def _record_iteration( success: bool, termination: str, changed_files: list[str], - pal_signal: dict[str, Any] | None = None, + pal_signal: Optional[Dict[str, Any]] = None, ) -> None: """Record an iteration result.""" input_quality = self.score_history[-2] if len(self.score_history) >= 2 else 0.0 diff --git a/core/quality_assessment.py b/core/quality_assessment.py index c3884544..e7f43804 100644 --- a/core/quality_assessment.py +++ b/core/quality_assessment.py @@ -5,11 +5,13 @@ to assess quality of code changes during loop iterations. """ +from __future__ import annotations + import json import subprocess import sys from pathlib import Path -from typing import Any +from typing import Any, Optional from .types import QualityAssessment @@ -33,7 +35,7 @@ def __init__(self, threshold: float = 70.0): self.threshold = threshold self.evidence_gate_path = self._find_evidence_gate() - def _find_evidence_gate(self) -> Path | None: + def _find_evidence_gate(self) -> Optional[Path]: """ Locate the evidence_gate.py script. diff --git a/core/skill_learning_integration.py b/core/skill_learning_integration.py new file mode 100644 index 00000000..30fcedd6 --- /dev/null +++ b/core/skill_learning_integration.py @@ -0,0 +1,453 @@ +""" +Skill Learning Integration for SuperClaude Loop Orchestrator. + +This module integrates the skill persistence layer with the loop orchestrator, +enabling cross-session learning by: + +1. Recording iteration feedback after each iteration +2. Extracting learned skills from successful sessions +3. Retrieving and injecting relevant skills at loop start +4. Tracking skill application effectiveness + +Usage: + from core.skill_learning_integration import LearningLoopOrchestrator + + orchestrator = LearningLoopOrchestrator(config) + result = orchestrator.run(context, skill_invoker) + # Skills are automatically learned and retrieved +""" + +from __future__ import annotations + +import uuid +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional + +from .loop_orchestrator import LoopOrchestrator, create_skill_invoker_signal +from .skill_persistence import ( + IterationFeedback, + LearnedSkill, + PromotionGate, + SkillExtractor, + SkillRetriever, + SkillStore, +) +from .types import ( + LoopConfig, + LoopResult, + TerminationReason, +) + + +class LearningLoopOrchestrator(LoopOrchestrator): + """ + Extended loop orchestrator with learning capabilities. + + Wraps the standard LoopOrchestrator to add: + - Feedback persistence after each iteration + - Skill extraction from successful sessions + - Relevant skill retrieval at loop start + - Application tracking for skill effectiveness + """ + + def __init__( + self, + config: Optional[LoopConfig] = None, + store: Optional[SkillStore] = None, + enable_learning: bool = True, + auto_promote: bool = False, + ): + """ + Initialize the learning-enabled orchestrator. + + Args: + config: Loop configuration + store: Skill store instance (uses default if None) + enable_learning: Whether to record feedback and extract skills + auto_promote: Whether to automatically promote high-quality skills + """ + super().__init__(config) + + self.enable_learning = enable_learning + self.auto_promote = auto_promote + self.store = store or SkillStore() + self.extractor = SkillExtractor(self.store) + self.retriever = SkillRetriever(self.store) + self.promotion_gate = PromotionGate(self.store) + + # Session tracking + self.session_id = str(uuid.uuid4())[:12] + self.repo_path = self._detect_repo_path() + self.domain = "general" + + # Track applied skills for effectiveness measurement + self._applied_skills: List[LearnedSkill] = [] + self._initial_quality: float = 0.0 + + def _detect_repo_path(self) -> str: + """Detect the current repository path.""" + cwd = Path.cwd() + + # Walk up to find .git directory + for parent in [cwd] + list(cwd.parents): + if (parent / ".git").exists(): + return str(parent) + + return str(cwd) + + def run( + self, + initial_context: Dict[str, Any], + skill_invoker: Callable[[Dict[str, Any]], Dict[str, Any]], + ) -> LoopResult: + """ + Execute the agentic loop with learning. + + Args: + initial_context: Initial task context + skill_invoker: Function that invokes Skills + + Returns: + LoopResult with final output and learning metadata + """ + # Detect domain from context + self.domain = self._detect_domain(initial_context) + + # Retrieve and inject relevant skills + if self.enable_learning: + initial_context = self._inject_relevant_skills(initial_context) + + # Run the standard loop + result = super().run(initial_context, skill_invoker) + + # Record all iteration feedback + if self.enable_learning: + self._record_all_feedback(result) + + # Extract skill if successful + if self.enable_learning and result.termination_reason == TerminationReason.QUALITY_MET: + self._extract_and_save_skill(result) + + # Track skill application effectiveness + if self.enable_learning and self._applied_skills: + self._record_skill_effectiveness(result) + + return result + + def _detect_domain(self, context: Dict[str, Any]) -> str: + """Detect domain from task context.""" + task = context.get("task", "").lower() + files = context.get("changed_files", []) + + # Check task keywords + domain_keywords = { + "backend": ["api", "server", "database", "endpoint", "rest", "graphql"], + "frontend": ["ui", "component", "react", "vue", "css", "html", "form"], + "infrastructure": ["deploy", "docker", "kubernetes", "ci", "cd", "terraform"], + "testing": ["test", "spec", "coverage", "mock", "fixture"], + "security": ["auth", "security", "encrypt", "permission", "access"], + "data": ["data", "pipeline", "etl", "analytics", "ml", "model"], + } + + for domain, keywords in domain_keywords.items(): + if any(kw in task for kw in keywords): + return domain + + # Check file extensions + extensions = set() + for f in files: + ext = Path(f).suffix.lstrip(".") + if ext: + extensions.add(ext) + + ext_domains = { + "py": "backend", + "ts": "frontend", + "tsx": "frontend", + "jsx": "frontend", + "go": "backend", + "rs": "backend", + "tf": "infrastructure", + "yaml": "infrastructure", + "sql": "data", + } + + for ext in extensions: + if ext in ext_domains: + return ext_domains[ext] + + return "general" + + def _inject_relevant_skills(self, context: Dict[str, Any]) -> Dict[str, Any]: + """Retrieve and inject relevant learned skills into context.""" + task = context.get("task", "") + files = context.get("changed_files", []) + + # Retrieve relevant skills + skill_results = self.retriever.retrieve( + task_description=task, + file_paths=files, + domain=self.domain, + max_skills=3, + promoted_only=False, # Include pending skills for now + ) + + if not skill_results: + return context + + # Track applied skills (effectiveness will be recorded at end of session) + self._applied_skills = [skill for skill, _score in skill_results] + + # Build skill injection + skill_context = [] + for skill, score in skill_results: + skill_info = { + "name": skill.name, + "relevance": f"{score:.0%}", + "patterns": skill.patterns[:3], # Top 3 patterns + "anti_patterns": skill.anti_patterns[:2], # Top 2 anti-patterns + "conditions": skill.applicability_conditions, + } + skill_context.append(skill_info) + # NOTE: Application recording moved to _record_skill_effectiveness() + # to avoid double-counting and include outcome data + + # Inject into context + context = context.copy() + context["learned_skills"] = skill_context + context["learning_context"] = ( + f"Found {len(skill_context)} relevant learned skills. " + "Consider applying their patterns and avoiding their anti-patterns." + ) + + return context + + def _record_all_feedback(self, result: LoopResult) -> None: + """Record all iteration feedback to the store.""" + for iter_result in result.iteration_history: + feedback = IterationFeedback( + session_id=self.session_id, + iteration=iter_result.iteration, + quality_before=iter_result.input_quality, + quality_after=iter_result.output_quality, + improvements_applied=iter_result.improvements_applied, + improvements_needed=[], # Already applied + changed_files=iter_result.changed_files, + test_results={}, # Would need to extract from output + duration_seconds=iter_result.time_taken, + success=iter_result.success, + termination_reason=iter_result.termination_reason, + ) + self.store.save_feedback(feedback) + + def _extract_and_save_skill(self, result: LoopResult) -> Optional[LearnedSkill]: + """Extract a learned skill from a successful session.""" + skill = self.extractor.extract_from_session( + session_id=self.session_id, + repo_path=self.repo_path, + domain=self.domain, + ) + + if skill is None: + return None + + # Save the skill + self.store.save_skill(skill) + + # Auto-promote if enabled and meets criteria + if self.auto_promote: + should_promote, reason = self.promotion_gate.evaluate(skill) + if should_promote: + self.promotion_gate.promote(skill, reason) + + return skill + + def _record_skill_effectiveness(self, result: LoopResult) -> None: + """Record how effective the applied skills were.""" + final_quality = result.final_assessment.overall_score + # Get initial quality from first iteration, or use 0 if no iterations + initial_quality = ( + result.iteration_history[0].input_quality if result.iteration_history else 0.0 + ) + quality_impact = final_quality - initial_quality + was_helpful = result.termination_reason == TerminationReason.QUALITY_MET + + for skill in self._applied_skills: + self.store.record_skill_application( + skill_id=skill.skill_id, + session_id=self.session_id, + was_helpful=was_helpful, + quality_impact=quality_impact, + feedback=f"Final quality: {final_quality:.1f}, Termination: {result.termination_reason.value}", + ) + + +def create_learning_invoker_signal( + context: Dict[str, Any], + learned_skills: Optional[List[Dict[str, Any]]] = None, +) -> Dict[str, Any]: + """ + Create a signal for Claude Code with learned skill context. + + Extends create_skill_invoker_signal with learned skill information. + + Args: + context: Context for skill execution + learned_skills: List of learned skill info dicts + + Returns: + Signal dict for Claude Code + """ + signal = create_skill_invoker_signal(context) + + if learned_skills: + signal["learned_context"] = { + "skills_applied": len(learned_skills), + "patterns_to_follow": [ + pattern for skill in learned_skills for pattern in skill.get("patterns", []) + ], + "patterns_to_avoid": [ + pattern for skill in learned_skills for pattern in skill.get("anti_patterns", []) + ], + } + + return signal + + +# --- CLI Entry Point --- + + +def run_learning_loop( + task: str, + max_iterations: int = 3, + quality_threshold: float = 70.0, + enable_learning: bool = True, + auto_promote: bool = False, +) -> Dict[str, Any]: + """ + Run an agentic loop with learning enabled. + + This is the main entry point for --loop with learning. + + Args: + task: Task description + max_iterations: Maximum iterations + quality_threshold: Quality threshold to meet + enable_learning: Whether to enable learning + auto_promote: Whether to auto-promote skills + + Returns: + Result dict with loop outcome and learning metadata + """ + config = LoopConfig( + max_iterations=max_iterations, + quality_threshold=quality_threshold, + ) + + orchestrator = LearningLoopOrchestrator( + config=config, + enable_learning=enable_learning, + auto_promote=auto_promote, + ) + + # Create initial context + initial_context = { + "task": task, + "improvements_needed": [], + "changed_files": [], + } + + # Placeholder skill invoker (Claude Code would provide this) + def placeholder_invoker(ctx: dict[str, Any]) -> dict[str, Any]: + """Placeholder - Claude Code provides actual invoker.""" + return { + "changes": [], + "tests": {"ran": False}, + "lint": {"ran": False}, + "changed_files": [], + } + + result = orchestrator.run(initial_context, placeholder_invoker) + + return { + "success": result.termination_reason == TerminationReason.QUALITY_MET, + "termination_reason": result.termination_reason.value, + "iterations": result.total_iterations, + "final_score": result.final_assessment.overall_score, + "session_id": orchestrator.session_id, + "skills_applied": len(orchestrator._applied_skills), + "learning_enabled": enable_learning, + } + + +# --- Utility Functions --- + + +def list_pending_skills() -> List[Dict[str, Any]]: + """List skills pending promotion review.""" + store = SkillStore() + gate = PromotionGate(store) + pending = gate.list_pending() + + return [ + { + "skill_id": skill.skill_id, + "name": skill.name, + "quality_score": skill.quality_score, + "source_session": skill.source_session, + "learned_at": skill.learned_at, + "patterns": len(skill.patterns), + } + for skill in pending + ] + + +def promote_skill(skill_id: str, reason: str = "") -> bool: + """Manually promote a skill.""" + store = SkillStore() + gate = PromotionGate(store) + + skill = store.get_skill(skill_id) + if skill is None: + return False + + path = gate.promote(skill, reason) + return path is not None + + +def get_skill_stats() -> Dict[str, Any]: + """Get overall skill learning statistics.""" + store = SkillStore() + conn = store._get_connection() + + # Count skills + skill_counts = conn.execute(""" + SELECT + COUNT(*) as total, + SUM(CASE WHEN promoted = 1 THEN 1 ELSE 0 END) as promoted, + AVG(quality_score) as avg_quality + FROM learned_skills + """).fetchone() + + # Count feedback + feedback_count = conn.execute("SELECT COUNT(*) FROM iteration_feedback").fetchone()[0] + + # Count applications + app_stats = conn.execute(""" + SELECT + COUNT(*) as total, + SUM(CASE WHEN was_helpful = 1 THEN 1 ELSE 0 END) as helpful + FROM skill_applications + """).fetchone() + + return { + "total_skills": skill_counts["total"] or 0, + "promoted_skills": skill_counts["promoted"] or 0, + "avg_quality": skill_counts["avg_quality"] or 0.0, + "total_feedback_records": feedback_count or 0, + "total_applications": app_stats["total"] or 0, + "helpful_applications": app_stats["helpful"] or 0, + "success_rate": ( + (app_stats["helpful"] or 0) / app_stats["total"] if app_stats["total"] else 0 + ), + } diff --git a/core/skill_persistence.py b/core/skill_persistence.py new file mode 100644 index 00000000..ade34c07 --- /dev/null +++ b/core/skill_persistence.py @@ -0,0 +1,1046 @@ +""" +Skill Persistence Layer for SuperClaude + +Enables cross-session learning by: +1. Storing learned patterns from successful iterations +2. Extracting generalizable skills from execution results +3. Retrieving relevant skills for new tasks +4. Gating skill promotion through quality thresholds + +Architecture: + SkillStore (SQLite) ─> SkillExtractor ─> SkillRetriever + │ + v + PromotionGate (PAL/tests) + │ + v + .claude/skills/learned/ + +Compatible with Python 3.9+ +""" + +from __future__ import annotations + +import hashlib +import json +import sqlite3 +import sys +import threading +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +# Thread-local storage for SQLite connections +_local = threading.local() + + +@dataclass +class LearnedSkill: + """A skill extracted from successful execution patterns.""" + + skill_id: str + name: str + description: str + triggers: list[str] + domain: str + source_session: str + source_repo: str + learned_at: str + patterns: list[str] # Successful patterns/strategies + anti_patterns: list[str] # What to avoid + quality_score: float + iteration_count: int + provenance: dict[str, Any] # Full trace for auditability + applicability_conditions: list[str] # When this skill applies + promoted: bool = False + promotion_reason: str = "" + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> LearnedSkill: + return cls(**data) + + def to_skill_md(self) -> str: + """Generate SKILL.md content for this learned skill.""" + triggers_str = ", ".join(self.triggers) + patterns_str = "\n".join(f"- {p}" for p in self.patterns) + anti_patterns_str = "\n".join(f"- {p}" for p in self.anti_patterns) + conditions_str = "\n".join(f"- {c}" for c in self.applicability_conditions) + + return f"""--- +name: {self.name} +description: {self.description} +learned: true +source_session: {self.source_session} +learned_at: {self.learned_at} +quality_score: {self.quality_score} +--- + +# {self.name.replace("-", " ").title()} + +{self.description} + +## Domain + +{self.domain} + +## Triggers + +{triggers_str} + +## Learned Patterns + +These patterns were extracted from successful executions: + +{patterns_str} + +## Anti-Patterns + +Avoid these approaches (they failed or caused issues): + +{anti_patterns_str} + +## Applicability Conditions + +This skill applies when: + +{conditions_str} + +## Provenance + +- **Source Session**: `{self.source_session}` +- **Source Repository**: `{self.source_repo}` +- **Learned At**: {self.learned_at} +- **Quality Score**: {self.quality_score}/100 +- **Iterations**: {self.iteration_count} +- **Promoted**: {self.promoted} ({self.promotion_reason or "pending"}) + +## Integration + +This is a **learned skill** automatically extracted from execution feedback. +It should be reviewed periodically and may be promoted to a permanent skill +after sufficient validation. +""" + + +@dataclass +class IterationFeedback: + """Feedback from a single loop iteration.""" + + session_id: str + iteration: int + quality_before: float + quality_after: float + improvements_applied: list[str] + improvements_needed: list[str] + changed_files: list[str] + test_results: dict[str, Any] + duration_seconds: float + success: bool + termination_reason: str + timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> IterationFeedback: + return cls(**data) + + +class SkillStore: + """ + SQLite-backed persistent storage for learned skills. + + Thread-safe with connection-per-thread pattern. + """ + + DEFAULT_DB_PATH = Path.home() / ".claude" / "learned_skills.db" + + def __init__(self, db_path: Optional[Path] = None): + self.db_path = db_path or self.DEFAULT_DB_PATH + self.db_path.parent.mkdir(parents=True, exist_ok=True) + self._init_schema() + + def _get_connection(self) -> sqlite3.Connection: + """Get thread-local database connection with WAL mode for concurrency.""" + if not hasattr(_local, "connection") or _local.connection is None: + _local.connection = sqlite3.connect( + str(self.db_path), check_same_thread=False, timeout=30.0 + ) + _local.connection.row_factory = sqlite3.Row + # Enable WAL mode for better concurrent access + _local.connection.execute("PRAGMA journal_mode=WAL") + return _local.connection + + def close(self) -> None: + """Close the thread-local database connection.""" + if hasattr(_local, "connection") and _local.connection is not None: + _local.connection.close() + _local.connection = None + + def __enter__(self) -> "SkillStore": + return self + + def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: + self.close() + + def _init_schema(self) -> None: + """Initialize database schema.""" + conn = self._get_connection() + conn.executescript(""" + CREATE TABLE IF NOT EXISTS learned_skills ( + skill_id TEXT PRIMARY KEY, + name TEXT NOT NULL, + description TEXT, + triggers TEXT, -- JSON array + domain TEXT, + source_session TEXT, + source_repo TEXT, + learned_at TEXT, + patterns TEXT, -- JSON array + anti_patterns TEXT, -- JSON array + quality_score REAL, + iteration_count INTEGER, + provenance TEXT, -- JSON object + applicability_conditions TEXT, -- JSON array + promoted INTEGER DEFAULT 0, + promotion_reason TEXT, + created_at TEXT DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT DEFAULT CURRENT_TIMESTAMP + ); + + CREATE TABLE IF NOT EXISTS iteration_feedback ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + iteration INTEGER, + quality_before REAL, + quality_after REAL, + improvements_applied TEXT, -- JSON array + improvements_needed TEXT, -- JSON array + changed_files TEXT, -- JSON array + test_results TEXT, -- JSON object + duration_seconds REAL, + success INTEGER, + termination_reason TEXT, + timestamp TEXT, + created_at TEXT DEFAULT CURRENT_TIMESTAMP + ); + + CREATE TABLE IF NOT EXISTS skill_applications ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + skill_id TEXT NOT NULL, + session_id TEXT NOT NULL, + applied_at TEXT, + was_helpful INTEGER, -- 1=yes, 0=no, NULL=unknown + quality_impact REAL, -- delta in quality score + feedback TEXT, + FOREIGN KEY (skill_id) REFERENCES learned_skills(skill_id) + ); + + CREATE INDEX IF NOT EXISTS idx_feedback_session + ON iteration_feedback(session_id); + CREATE INDEX IF NOT EXISTS idx_skills_domain + ON learned_skills(domain); + CREATE INDEX IF NOT EXISTS idx_skills_promoted + ON learned_skills(promoted); + CREATE INDEX IF NOT EXISTS idx_applications_skill + ON skill_applications(skill_id); + """) + conn.commit() + + # --- Skill CRUD Operations --- + + def save_skill(self, skill: LearnedSkill) -> bool: + """Save or update a learned skill. Returns True on success.""" + conn = self._get_connection() + try: + conn.execute( + """ + INSERT OR REPLACE INTO learned_skills ( + skill_id, name, description, triggers, domain, + source_session, source_repo, learned_at, patterns, + anti_patterns, quality_score, iteration_count, + provenance, applicability_conditions, promoted, + promotion_reason, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + skill.skill_id, + skill.name, + skill.description, + json.dumps(skill.triggers), + skill.domain, + skill.source_session, + skill.source_repo, + skill.learned_at, + json.dumps(skill.patterns), + json.dumps(skill.anti_patterns), + skill.quality_score, + skill.iteration_count, + json.dumps(skill.provenance), + json.dumps(skill.applicability_conditions), + 1 if skill.promoted else 0, + skill.promotion_reason, + datetime.now(timezone.utc).isoformat(), + ), + ) + conn.commit() + return True + except sqlite3.Error as e: + print(f"[SkillStore] Failed to save skill {skill.skill_id}: {e}", file=sys.stderr) + return False + + def get_skill(self, skill_id: str) -> Optional[LearnedSkill]: + """Retrieve a skill by ID.""" + conn = self._get_connection() + row = conn.execute( + "SELECT * FROM learned_skills WHERE skill_id = ?", (skill_id,) + ).fetchone() + return self._row_to_skill(row) if row else None + + def get_promoted_skills(self) -> list[LearnedSkill]: + """Get all promoted skills.""" + conn = self._get_connection() + rows = conn.execute( + "SELECT * FROM learned_skills WHERE promoted = 1 ORDER BY quality_score DESC" + ).fetchall() + return [self._row_to_skill(row) for row in rows] + + def get_skills_by_domain(self, domain: str) -> list[LearnedSkill]: + """Get skills matching a domain.""" + conn = self._get_connection() + rows = conn.execute( + "SELECT * FROM learned_skills WHERE domain = ? ORDER BY quality_score DESC", (domain,) + ).fetchall() + return [self._row_to_skill(row) for row in rows] + + def search_skills( + self, + query: str, + domain: Optional[str] = None, + min_quality: float = 70.0, + promoted_only: bool = False, + ) -> List[LearnedSkill]: + """Search skills by trigger keywords and filters.""" + conn = self._get_connection() + + sql = "SELECT * FROM learned_skills WHERE quality_score >= ?" + params: list[Any] = [min_quality] + + if domain: + sql += " AND domain = ?" + params.append(domain) + + if promoted_only: + sql += " AND promoted = 1" + + sql += " ORDER BY quality_score DESC" + + rows = conn.execute(sql, params).fetchall() + + # Filter by trigger match + query_terms = set(query.lower().split()) + results = [] + for row in rows: + skill = self._row_to_skill(row) + skill_triggers = set(t.lower() for t in skill.triggers) + if query_terms & skill_triggers: + results.append(skill) + + return results + + def _row_to_skill(self, row: sqlite3.Row) -> LearnedSkill: + """Convert database row to LearnedSkill object.""" + return LearnedSkill( + skill_id=row["skill_id"], + name=row["name"], + description=row["description"] or "", + triggers=json.loads(row["triggers"] or "[]"), + domain=row["domain"] or "", + source_session=row["source_session"] or "", + source_repo=row["source_repo"] or "", + learned_at=row["learned_at"] or "", + patterns=json.loads(row["patterns"] or "[]"), + anti_patterns=json.loads(row["anti_patterns"] or "[]"), + quality_score=row["quality_score"] or 0.0, + iteration_count=row["iteration_count"] or 0, + provenance=json.loads(row["provenance"] or "{}"), + applicability_conditions=json.loads(row["applicability_conditions"] or "[]"), + promoted=bool(row["promoted"]), + promotion_reason=row["promotion_reason"] or "", + ) + + # --- Iteration Feedback --- + + def save_feedback(self, feedback: IterationFeedback) -> bool: + """Record iteration feedback for learning. Returns True on success.""" + conn = self._get_connection() + try: + conn.execute( + """ + INSERT INTO iteration_feedback ( + session_id, iteration, quality_before, quality_after, + improvements_applied, improvements_needed, changed_files, + test_results, duration_seconds, success, termination_reason, + timestamp + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + feedback.session_id, + feedback.iteration, + feedback.quality_before, + feedback.quality_after, + json.dumps(feedback.improvements_applied), + json.dumps(feedback.improvements_needed), + json.dumps(feedback.changed_files), + json.dumps(feedback.test_results), + feedback.duration_seconds, + 1 if feedback.success else 0, + feedback.termination_reason, + feedback.timestamp, + ), + ) + conn.commit() + return True + except sqlite3.Error as e: + print( + f"[SkillStore] Failed to save feedback for session {feedback.session_id}: {e}", + file=sys.stderr, + ) + return False + + def get_session_feedback(self, session_id: str) -> list[IterationFeedback]: + """Get all feedback for a session.""" + conn = self._get_connection() + rows = conn.execute( + "SELECT * FROM iteration_feedback WHERE session_id = ? ORDER BY iteration", + (session_id,), + ).fetchall() + return [ + IterationFeedback( + session_id=row["session_id"], + iteration=row["iteration"], + quality_before=row["quality_before"], + quality_after=row["quality_after"], + improvements_applied=json.loads(row["improvements_applied"] or "[]"), + improvements_needed=json.loads(row["improvements_needed"] or "[]"), + changed_files=json.loads(row["changed_files"] or "[]"), + test_results=json.loads(row["test_results"] or "{}"), + duration_seconds=row["duration_seconds"], + success=bool(row["success"]), + termination_reason=row["termination_reason"] or "", + timestamp=row["timestamp"] or "", + ) + for row in rows + ] + + # --- Skill Application Tracking --- + + def record_skill_application( + self, + skill_id: str, + session_id: str, + was_helpful: Optional[bool] = None, + quality_impact: Optional[float] = None, + feedback: str = "", + ) -> bool: + """Record when a skill was applied and its effectiveness. Returns True on success.""" + conn = self._get_connection() + try: + conn.execute( + """ + INSERT INTO skill_applications ( + skill_id, session_id, applied_at, was_helpful, + quality_impact, feedback + ) VALUES (?, ?, ?, ?, ?, ?) + """, + ( + skill_id, + session_id, + datetime.now(timezone.utc).isoformat(), + 1 if was_helpful else (0 if was_helpful is False else None), + quality_impact, + feedback, + ), + ) + conn.commit() + return True + except sqlite3.Error as e: + print( + f"[SkillStore] Failed to record application for skill {skill_id}: {e}", + file=sys.stderr, + ) + return False + + def get_skill_effectiveness(self, skill_id: str) -> Dict[str, Any]: + """Calculate skill effectiveness metrics.""" + conn = self._get_connection() + row = conn.execute( + """ + SELECT + COUNT(*) as applications, + SUM(CASE WHEN was_helpful = 1 THEN 1 ELSE 0 END) as helpful_count, + SUM(CASE WHEN was_helpful = 0 THEN 1 ELSE 0 END) as unhelpful_count, + AVG(quality_impact) as avg_quality_impact + FROM skill_applications + WHERE skill_id = ? + """, + (skill_id,), + ).fetchone() + + return { + "applications": row["applications"] or 0, + "helpful_count": row["helpful_count"] or 0, + "unhelpful_count": row["unhelpful_count"] or 0, + "success_rate": ( + (row["helpful_count"] or 0) / row["applications"] if row["applications"] else 0 + ), + "avg_quality_impact": row["avg_quality_impact"] or 0.0, + } + + def get_bulk_skill_effectiveness(self, skill_ids: List[str]) -> Dict[str, Dict[str, Any]]: + """ + Calculate skill effectiveness metrics for multiple skills in one query. + + Returns a dict mapping skill_id -> effectiveness metrics. + This avoids N+1 queries when scoring multiple skills. + """ + if not skill_ids: + return {} + + conn = self._get_connection() + placeholders = ",".join("?" for _ in skill_ids) + rows = conn.execute( + f""" + SELECT + skill_id, + COUNT(*) as applications, + SUM(CASE WHEN was_helpful = 1 THEN 1 ELSE 0 END) as helpful_count, + SUM(CASE WHEN was_helpful = 0 THEN 1 ELSE 0 END) as unhelpful_count, + AVG(quality_impact) as avg_quality_impact + FROM skill_applications + WHERE skill_id IN ({placeholders}) + GROUP BY skill_id + """, + skill_ids, + ).fetchall() + + results: Dict[str, Dict[str, Any]] = {} + for row in rows: + apps = row["applications"] or 0 + helpful = row["helpful_count"] or 0 + results[row["skill_id"]] = { + "applications": apps, + "helpful_count": helpful, + "unhelpful_count": row["unhelpful_count"] or 0, + "success_rate": helpful / apps if apps else 0, + "avg_quality_impact": row["avg_quality_impact"] or 0.0, + } + + # Fill in missing skill_ids with default values + for skill_id in skill_ids: + if skill_id not in results: + results[skill_id] = { + "applications": 0, + "helpful_count": 0, + "unhelpful_count": 0, + "success_rate": 0, + "avg_quality_impact": 0.0, + } + + return results + + +class SkillExtractor: + """ + Extracts generalizable skills from iteration feedback. + + Analyzes patterns across iterations to identify: + - Successful strategies that led to quality improvements + - Anti-patterns that caused regressions or stalls + - Conditions under which patterns apply + """ + + def __init__(self, store: SkillStore): + self.store = store + + def extract_from_session( + self, session_id: str, repo_path: str = "", domain: str = "general" + ) -> Optional[LearnedSkill]: + """ + Extract a learned skill from a completed session. + + Returns None if the session doesn't have enough signal + or didn't achieve quality threshold. + """ + feedback_list = self.store.get_session_feedback(session_id) + + if not feedback_list: + return None + + # Only extract from successful sessions + final_feedback = feedback_list[-1] + if not final_feedback.success or final_feedback.quality_after < 70.0: + return None + + # Need at least 2 iterations to learn from + if len(feedback_list) < 2: + return None + + # Extract patterns + patterns = self._extract_patterns(feedback_list) + anti_patterns = self._extract_anti_patterns(feedback_list) + triggers = self._extract_triggers(feedback_list) + conditions = self._extract_conditions(feedback_list) + + # Generate skill ID and name + skill_id = self._generate_skill_id(session_id, patterns) + name = self._generate_skill_name(patterns, domain) + + return LearnedSkill( + skill_id=skill_id, + name=name, + description=f"Learned skill extracted from session {session_id[:8]}", + triggers=triggers, + domain=domain, + source_session=session_id, + source_repo=repo_path, + learned_at=datetime.now(timezone.utc).isoformat(), + patterns=patterns, + anti_patterns=anti_patterns, + quality_score=final_feedback.quality_after, + iteration_count=len(feedback_list), + provenance={ + "session_id": session_id, + "repo_path": repo_path, + "iterations": len(feedback_list), + "quality_progression": [ + {"iteration": f.iteration, "before": f.quality_before, "after": f.quality_after} + for f in feedback_list + ], + "total_duration": sum(f.duration_seconds for f in feedback_list), + "termination_reason": final_feedback.termination_reason, + }, + applicability_conditions=conditions, + promoted=False, + promotion_reason="", + ) + + def _extract_patterns(self, feedback_list: list[IterationFeedback]) -> list[str]: + """Extract successful improvement patterns.""" + patterns = [] + + for i, feedback in enumerate(feedback_list): + # Quality improved this iteration + if feedback.quality_after > feedback.quality_before: + for improvement in feedback.improvements_applied: + patterns.append(f"[Iter {feedback.iteration}] {improvement}") + + # Deduplicate while preserving order + seen = set() + unique_patterns = [] + for p in patterns: + # Normalize for dedup + normalized = p.split("] ", 1)[-1].lower().strip() + if normalized not in seen: + seen.add(normalized) + unique_patterns.append(p) + + return unique_patterns[:10] # Limit to top 10 patterns + + def _extract_anti_patterns(self, feedback_list: list[IterationFeedback]) -> list[str]: + """Extract patterns that didn't work or caused issues.""" + anti_patterns = [] + + for feedback in feedback_list: + # Quality decreased or stalled + if feedback.quality_after <= feedback.quality_before: + for improvement in feedback.improvements_applied: + anti_patterns.append(f"[Failed] {improvement}") + + # Improvements still needed at end + if not feedback.success: + for needed in feedback.improvements_needed: + anti_patterns.append(f"[Unresolved] {needed}") + + return anti_patterns[:5] # Limit to top 5 anti-patterns + + def _extract_triggers(self, feedback_list: list[IterationFeedback]) -> list[str]: + """Extract trigger keywords from changed files and improvements.""" + triggers = set() + + for feedback in feedback_list: + # Extract from file paths + for file_path in feedback.changed_files: + parts = Path(file_path).parts + for part in parts: + if part not in {"src", "lib", "test", "tests", "spec", "."}: + triggers.add(part.lower()) + + # Extract extension + ext = Path(file_path).suffix.lstrip(".") + if ext: + triggers.add(ext) + + # Extract from improvements + for improvement in feedback.improvements_applied: + words = improvement.lower().split() + for word in words: + if len(word) > 3 and word.isalpha(): + triggers.add(word) + + # Filter common words + stopwords = {"the", "and", "for", "with", "from", "this", "that", "have", "been"} + triggers = triggers - stopwords + + return sorted(triggers)[:15] # Limit to 15 triggers + + def _extract_conditions(self, feedback_list: list[IterationFeedback]) -> list[str]: + """Extract applicability conditions.""" + conditions = [] + + # Analyze file types + extensions = set() + for feedback in feedback_list: + for file_path in feedback.changed_files: + ext = Path(file_path).suffix + if ext: + extensions.add(ext) + + if extensions: + conditions.append(f"File types: {', '.join(sorted(extensions))}") + + # Analyze test presence + had_tests = any(feedback.test_results.get("ran", False) for feedback in feedback_list) + if had_tests: + conditions.append("Project has test suite") + + # Analyze iteration count + if len(feedback_list) >= 3: + conditions.append("Complex task requiring multiple iterations") + + return conditions + + def _generate_skill_id(self, session_id: str, patterns: list[str]) -> str: + """Generate unique skill ID.""" + content = f"{session_id}:{':'.join(patterns)}" + return f"learned-{hashlib.sha256(content.encode()).hexdigest()[:12]}" + + def _generate_skill_name(self, patterns: list[str], domain: str) -> str: + """Generate human-readable skill name.""" + if not patterns: + return f"learned-{domain}-skill" + + # Extract key words from first pattern + first_pattern = patterns[0].split("] ", 1)[-1] if patterns else "" + words = [w for w in first_pattern.split()[:3] if len(w) > 2] + name_part = "-".join(words).lower() if words else "general" + + return f"learned-{domain}-{name_part}" + + +class SkillRetriever: + """ + Retrieves relevant learned skills for a given task context. + + Uses trigger matching and domain filtering to find applicable skills. + """ + + def __init__(self, store: SkillStore): + self.store = store + + def retrieve( + self, + task_description: str, + file_paths: Optional[List[str]] = None, + domain: Optional[str] = None, + max_skills: int = 3, + promoted_only: bool = True, + ) -> List[Tuple[LearnedSkill, float]]: + """ + Retrieve relevant skills for a task. + + Returns list of (skill, relevance_score) tuples. + """ + # Extract search terms from task + search_terms = self._extract_search_terms(task_description, file_paths) + + # Get candidate skills + candidates = self.store.search_skills( + query=" ".join(search_terms), + domain=domain, + min_quality=50.0, + promoted_only=promoted_only, + ) + + if not candidates: + return [] + + # Batch fetch effectiveness data to avoid N+1 queries + skill_ids = [s.skill_id for s in candidates] + effectiveness_map = self.store.get_bulk_skill_effectiveness(skill_ids) + + # Score and rank using pre-fetched effectiveness + scored = [] + for skill in candidates: + effectiveness = effectiveness_map.get(skill.skill_id, {}) + score = self._score_relevance(skill, search_terms, file_paths, effectiveness) + if score > 0: + scored.append((skill, score)) + + # Sort by score descending + scored.sort(key=lambda x: x[1], reverse=True) + + return scored[:max_skills] + + def _extract_search_terms(self, task_description: str, file_paths: Optional[List[str]]) -> set: + """Extract search terms from task context.""" + terms = set() + + # From task description + words = task_description.lower().split() + for word in words: + if len(word) > 3 and word.isalpha(): + terms.add(word) + + # From file paths + if file_paths: + for path in file_paths: + parts = Path(path).parts + for part in parts: + if part not in {"src", "lib", "test", "."}: + terms.add(part.lower()) + + ext = Path(path).suffix.lstrip(".") + if ext: + terms.add(ext) + + return terms + + def _score_relevance( + self, + skill: LearnedSkill, + search_terms: set, + file_paths: Optional[List[str]], + effectiveness: Optional[Dict[str, Any]] = None, + ) -> float: + """Score skill relevance to current context.""" + score = 0.0 + + # Trigger match (40%) + skill_triggers = set(t.lower() for t in skill.triggers) + trigger_overlap = len(search_terms & skill_triggers) + if skill_triggers: + score += 0.4 * (trigger_overlap / len(skill_triggers)) + + # Quality score (30%) + score += 0.3 * (skill.quality_score / 100.0) + + # Promoted bonus (20%) + if skill.promoted: + score += 0.2 + + # Effectiveness history (10%) - use pre-fetched data if available + if effectiveness is None: + effectiveness = self.store.get_skill_effectiveness(skill.skill_id) + if effectiveness.get("applications", 0) > 0: + score += 0.1 * effectiveness.get("success_rate", 0) + + return score + + +class PromotionGate: + """ + Gates skill promotion based on quality thresholds and validation. + + Skills must pass validation before being promoted to permanent status. + """ + + # Promotion thresholds + MIN_QUALITY_SCORE = 85.0 + MIN_APPLICATIONS = 2 + MIN_SUCCESS_RATE = 0.7 + + def __init__(self, store: SkillStore, skills_dir: Optional[Path] = None): + self.store = store + self.skills_dir = skills_dir or (Path.home() / ".claude" / "skills" / "learned") + + def evaluate(self, skill: LearnedSkill) -> Tuple[bool, str]: + """ + Evaluate if a skill should be promoted. + + Returns (should_promote, reason). + """ + reasons = [] + + # Check quality score + if skill.quality_score < self.MIN_QUALITY_SCORE: + reasons.append( + f"Quality score {skill.quality_score:.1f} below threshold {self.MIN_QUALITY_SCORE}" + ) + + # Check application history + effectiveness = self.store.get_skill_effectiveness(skill.skill_id) + + if effectiveness["applications"] < self.MIN_APPLICATIONS: + reasons.append( + f"Only {effectiveness['applications']} applications, need {self.MIN_APPLICATIONS}" + ) + elif effectiveness["success_rate"] < self.MIN_SUCCESS_RATE: + reasons.append( + f"Success rate {effectiveness['success_rate']:.1%} below {self.MIN_SUCCESS_RATE:.0%}" + ) + + if reasons: + return False, "; ".join(reasons) + + return True, "Meets all promotion criteria" + + def promote(self, skill: LearnedSkill, reason: str = "") -> Optional[Path]: + """ + Promote a skill to permanent status. + + Creates SKILL.md in the learned skills directory. + Returns the path to the created skill, or None on failure. + + Uses skill_id for directory name to prevent path traversal attacks. + Promotion is atomic: if file write fails, DB change is rolled back. + """ + # Evaluate first + should_promote, eval_reason = self.evaluate(skill) + + if not should_promote: + return None + + # Store original state for rollback + original_promoted = skill.promoted + original_reason = skill.promotion_reason + + # Update skill status + skill.promoted = True + skill.promotion_reason = reason or eval_reason + + # Use skill_id for directory name (safe hash, no path traversal risk) + # Keep skill.name in SKILL.md content for human readability + skill_dir = self.skills_dir / skill.skill_id + skill_md_path = skill_dir / "SKILL.md" + metadata_path = skill_dir / "metadata.json" + + try: + # Create skill directory and files + skill_dir.mkdir(parents=True, exist_ok=True) + skill_md_path.write_text(skill.to_skill_md()) + metadata_path.write_text(json.dumps(skill.to_dict(), indent=2)) + + # Save to database only after files are successfully written + if not self.store.save_skill(skill): + # DB save failed, rollback files + raise IOError("Database save failed") + + return skill_md_path + + except (IOError, OSError) as e: + # Rollback skill state + skill.promoted = original_promoted + skill.promotion_reason = original_reason + + # Clean up any partially created files + try: + if metadata_path.exists(): + metadata_path.unlink() + if skill_md_path.exists(): + skill_md_path.unlink() + if skill_dir.exists() and not any(skill_dir.iterdir()): + skill_dir.rmdir() + except OSError: + pass # Best effort cleanup + + print(f"[PromotionGate] Failed to promote skill {skill.skill_id}: {e}", file=sys.stderr) + return None + + def list_pending(self) -> List[LearnedSkill]: + """List skills pending promotion review.""" + conn = self.store._get_connection() + rows = conn.execute( + """ + SELECT * FROM learned_skills + WHERE promoted = 0 AND quality_score >= ? + ORDER BY quality_score DESC + """, + (self.MIN_QUALITY_SCORE - 10,), + ).fetchall() + + return [self.store._row_to_skill(row) for row in rows] + + +# --- Convenience Functions --- + + +def get_default_store() -> SkillStore: + """Get the default skill store instance.""" + return SkillStore() + + +def learn_from_session( + session_id: str, repo_path: str = "", domain: str = "general", auto_promote: bool = False +) -> Optional[LearnedSkill]: + """ + Convenience function to extract and optionally promote a skill from a session. + + Usage: + from core.skill_persistence import learn_from_session + + skill = learn_from_session( + session_id="abc123", + repo_path="/path/to/repo", + domain="backend", + auto_promote=True + ) + """ + store = get_default_store() + extractor = SkillExtractor(store) + + skill = extractor.extract_from_session(session_id, repo_path, domain) + + if skill is None: + return None + + store.save_skill(skill) + + if auto_promote: + gate = PromotionGate(store) + gate.promote(skill) + + return skill + + +def retrieve_skills_for_task( + task_description: str, file_paths: Optional[List[str]] = None, domain: Optional[str] = None +) -> List[LearnedSkill]: + """ + Convenience function to retrieve relevant skills for a task. + + Usage: + from core.skill_persistence import retrieve_skills_for_task + + skills = retrieve_skills_for_task( + task_description="Implement user authentication", + file_paths=["src/auth/login.py"], + domain="backend" + ) + """ + store = get_default_store() + retriever = SkillRetriever(store) + + results = retriever.retrieve( + task_description=task_description, + file_paths=file_paths, + domain=domain, + promoted_only=False, # Include non-promoted for now + ) + + return [skill for skill, _score in results] diff --git a/core/types.py b/core/types.py index 98aeba94..bfbe09fe 100644 --- a/core/types.py +++ b/core/types.py @@ -4,9 +4,11 @@ Ported from archive/python-sdk-v5/Quality/quality_scorer.py """ +from __future__ import annotations + from dataclasses import dataclass, field from enum import Enum -from typing import Any +from typing import Any, Dict, Optional class TerminationReason(Enum): @@ -49,7 +51,7 @@ class LoopConfig: quality_threshold: float = 70.0 oscillation_window: int = 3 stagnation_threshold: float = 2.0 - timeout_seconds: float | None = None + timeout_seconds: Optional[float] = None pal_review_enabled: bool = True pal_model: str = "gpt-5" @@ -107,7 +109,7 @@ class IterationResult: time_taken: float = 0.0 success: bool = False termination_reason: str = "" - pal_review: dict[str, Any] | None = None + pal_review: Optional[Dict[str, Any]] = None changed_files: list[str] = field(default_factory=list) diff --git a/tests/core/test_skill_learning_integration.py b/tests/core/test_skill_learning_integration.py new file mode 100644 index 00000000..cc36a1cf --- /dev/null +++ b/tests/core/test_skill_learning_integration.py @@ -0,0 +1,254 @@ +"""Tests for the skill learning integration module.""" + +from __future__ import annotations + +import pytest + +from core.skill_learning_integration import ( + LearningLoopOrchestrator, + create_learning_invoker_signal, + get_skill_stats, + list_pending_skills, + promote_skill, + run_learning_loop, +) +from core.skill_persistence import LearnedSkill, SkillStore +from core.types import LoopConfig + +# --- Fixtures --- + + +@pytest.fixture +def temp_store(tmp_path): + """Create a temporary skill store.""" + db_path = tmp_path / "test_learning.db" + store = SkillStore(db_path) + yield store + store.close() + + +@pytest.fixture +def learning_orchestrator(temp_store): + """Create a learning orchestrator with temp store.""" + config = LoopConfig(max_iterations=3, quality_threshold=70.0) + return LearningLoopOrchestrator( + config=config, + store=temp_store, + enable_learning=True, + auto_promote=False, + ) + + +@pytest.fixture +def sample_skill(): + """Create a sample skill for testing.""" + return LearnedSkill( + skill_id="test-learn-001", + name="Test Learning Skill", + description="A test skill", + triggers=["test", "learn"], + domain="testing", + source_session="session-123", + source_repo="/path/to/repo", + learned_at="2025-01-01T00:00:00Z", + patterns=["Pattern 1"], + anti_patterns=["Anti-pattern 1"], + quality_score=90.0, + iteration_count=3, + provenance={}, + applicability_conditions=[], + promoted=False, + promotion_reason="", + ) + + +# --- LearningLoopOrchestrator Tests --- + + +class TestLearningLoopOrchestrator: + """Tests for LearningLoopOrchestrator.""" + + def test_init_with_defaults(self): + """Test initialization with default values.""" + orchestrator = LearningLoopOrchestrator() + assert orchestrator.enable_learning is True + assert orchestrator.auto_promote is False + assert orchestrator.session_id is not None + + def test_init_with_custom_config(self, temp_store): + """Test initialization with custom config.""" + config = LoopConfig(max_iterations=5, quality_threshold=80.0) + orchestrator = LearningLoopOrchestrator( + config=config, + store=temp_store, + enable_learning=False, + auto_promote=True, + ) + assert orchestrator.config.max_iterations == 5 + assert orchestrator.config.quality_threshold == 80.0 + assert orchestrator.enable_learning is False + assert orchestrator.auto_promote is True + + def test_detect_domain_from_task(self, learning_orchestrator): + """Test domain detection from task description.""" + # Test backend keywords + context = {"task": "implement REST API endpoint"} + domain = learning_orchestrator._detect_domain(context) + assert domain == "backend" + + # Test frontend keywords + context = {"task": "create React component"} + domain = learning_orchestrator._detect_domain(context) + assert domain == "frontend" + + # Test testing keywords + context = {"task": "write unit tests"} + domain = learning_orchestrator._detect_domain(context) + assert domain == "testing" + + def test_detect_domain_from_files(self, learning_orchestrator): + """Test domain detection from file extensions.""" + context = {"task": "update code", "changed_files": ["app.tsx", "styles.css"]} + domain = learning_orchestrator._detect_domain(context) + assert domain == "frontend" + + context = {"task": "update code", "changed_files": ["main.py", "utils.py"]} + domain = learning_orchestrator._detect_domain(context) + assert domain == "backend" + + def test_detect_domain_default(self, learning_orchestrator): + """Test domain defaults to general.""" + context = {"task": "do something", "changed_files": []} + domain = learning_orchestrator._detect_domain(context) + assert domain == "general" + + def test_inject_relevant_skills(self, learning_orchestrator, temp_store, sample_skill): + """Test skill injection into context.""" + temp_store.save_skill(sample_skill) + context = {"task": "test learning"} + result = learning_orchestrator._inject_relevant_skills(context) + # May or may not find skills depending on matching + assert "task" in result + + def test_inject_skills_empty_when_none(self, learning_orchestrator): + """Test skill injection returns original context when no skills match.""" + context = {"task": "completely unrelated task xyz123"} + result = learning_orchestrator._inject_relevant_skills(context) + assert result.get("learned_skills") is None or len(result.get("learned_skills", [])) == 0 + + def test_repo_path_detection(self, learning_orchestrator): + """Test repository path detection.""" + # Should detect some path (cwd or git root) + assert learning_orchestrator.repo_path is not None + assert len(learning_orchestrator.repo_path) > 0 + + +# --- Utility Function Tests --- + + +class TestUtilityFunctions: + """Tests for utility functions.""" + + def test_create_learning_invoker_signal_basic(self): + """Test creating invoker signal without learned skills.""" + context = {"task": "implement feature", "files": ["main.py"]} + signal = create_learning_invoker_signal(context) + assert "task" in signal or "context" in str(signal).lower() + + def test_create_learning_invoker_signal_with_skills(self): + """Test creating invoker signal with learned skills.""" + context = {"task": "implement feature"} + learned_skills = [ + {"name": "skill1", "patterns": ["p1", "p2"], "anti_patterns": ["a1"]}, + {"name": "skill2", "patterns": ["p3"], "anti_patterns": []}, + ] + signal = create_learning_invoker_signal(context, learned_skills) + assert "learned_context" in signal + assert signal["learned_context"]["skills_applied"] == 2 + assert len(signal["learned_context"]["patterns_to_follow"]) == 3 + assert len(signal["learned_context"]["patterns_to_avoid"]) == 1 + + def test_list_pending_skills(self, tmp_path, monkeypatch): + """Test listing pending skills.""" + monkeypatch.setattr( + "core.skill_persistence.SkillStore.DEFAULT_DB_PATH", + tmp_path / "pending_test.db", + ) + # Should return empty list when no skills + pending = list_pending_skills() + assert isinstance(pending, list) + + def test_get_skill_stats(self, tmp_path, monkeypatch): + """Test getting skill statistics.""" + monkeypatch.setattr( + "core.skill_persistence.SkillStore.DEFAULT_DB_PATH", + tmp_path / "stats_test.db", + ) + stats = get_skill_stats() + assert "total_skills" in stats + assert "promoted_skills" in stats + assert "total_applications" in stats + assert stats["total_skills"] == 0 + + def test_promote_skill_nonexistent(self, tmp_path, monkeypatch): + """Test promoting a non-existent skill returns False.""" + monkeypatch.setattr( + "core.skill_persistence.SkillStore.DEFAULT_DB_PATH", + tmp_path / "promote_test.db", + ) + result = promote_skill("nonexistent-skill-id") + assert result is False + + +# --- Run Learning Loop Tests --- + + +class TestRunLearningLoop: + """Tests for run_learning_loop function.""" + + def test_run_learning_loop_basic(self, tmp_path, monkeypatch): + """Test basic learning loop execution.""" + monkeypatch.setattr( + "core.skill_persistence.SkillStore.DEFAULT_DB_PATH", + tmp_path / "loop_test.db", + ) + result = run_learning_loop( + task="test task", + max_iterations=1, + quality_threshold=70.0, + enable_learning=True, + ) + assert "success" in result + assert "termination_reason" in result + assert "iterations" in result + assert "session_id" in result + + def test_run_learning_loop_with_auto_promote(self, tmp_path, monkeypatch): + """Test learning loop with auto-promote enabled.""" + monkeypatch.setattr( + "core.skill_persistence.SkillStore.DEFAULT_DB_PATH", + tmp_path / "autopromote_test.db", + ) + result = run_learning_loop( + task="implement auth", + max_iterations=2, + quality_threshold=70.0, + enable_learning=True, + auto_promote=True, + ) + assert "learning_enabled" in result + assert result["learning_enabled"] is True + + def test_run_learning_loop_disabled(self, tmp_path, monkeypatch): + """Test learning loop with learning disabled.""" + monkeypatch.setattr( + "core.skill_persistence.SkillStore.DEFAULT_DB_PATH", + tmp_path / "disabled_test.db", + ) + result = run_learning_loop( + task="simple task", + max_iterations=1, + quality_threshold=70.0, + enable_learning=False, + ) + assert result["learning_enabled"] is False diff --git a/tests/core/test_skill_persistence.py b/tests/core/test_skill_persistence.py new file mode 100644 index 00000000..f1275bcb --- /dev/null +++ b/tests/core/test_skill_persistence.py @@ -0,0 +1,453 @@ +"""Tests for the skill persistence layer.""" + +from __future__ import annotations + +import pytest + +from core.skill_persistence import ( + IterationFeedback, + LearnedSkill, + PromotionGate, + SkillExtractor, + SkillRetriever, + SkillStore, + get_default_store, + learn_from_session, + retrieve_skills_for_task, +) + +# --- Fixtures --- + + +@pytest.fixture +def temp_db(tmp_path): + """Create a temporary database for testing.""" + db_path = tmp_path / "test_skills.db" + store = SkillStore(db_path) + yield store + store.close() + + +@pytest.fixture +def sample_skill(): + """Create a sample learned skill.""" + return LearnedSkill( + skill_id="test-skill-001", + name="Test Skill", + description="A test skill for unit testing", + triggers=["test", "unit", "pytest"], + domain="testing", + source_session="session-abc123", + source_repo="/path/to/repo", + learned_at="2025-01-01T00:00:00Z", + patterns=["Use pytest fixtures", "Mock external calls"], + anti_patterns=["Don't test implementation details"], + quality_score=85.0, + iteration_count=3, + provenance={"iterations": 3, "repo": "/path/to/repo"}, + applicability_conditions=["Python projects", "Has test suite"], + promoted=False, + promotion_reason="", + ) + + +@pytest.fixture +def sample_feedback(): + """Create sample iteration feedback.""" + return IterationFeedback( + session_id="session-abc123", + iteration=1, + quality_before=50.0, + quality_after=75.0, + improvements_applied=["Added tests", "Fixed linting"], + improvements_needed=["Add docstrings"], + changed_files=["test_foo.py", "foo.py"], + test_results={"ran": True, "passed": 10, "failed": 0}, + duration_seconds=30.5, + success=True, + termination_reason="", + ) + + +@pytest.fixture +def high_quality_skill(): + """Create a high-quality skill that meets promotion criteria.""" + return LearnedSkill( + skill_id="high-quality-001", + name="High Quality Skill", + description="A high quality skill", + triggers=["quality", "best"], + domain="backend", + source_session="session-xyz789", + source_repo="/path/to/repo", + learned_at="2025-01-01T00:00:00Z", + patterns=["Pattern 1", "Pattern 2"], + anti_patterns=["Anti-pattern 1"], + quality_score=90.0, + iteration_count=5, + provenance={}, + applicability_conditions=[], + promoted=False, + promotion_reason="", + ) + + +# --- SkillStore Tests --- + + +class TestSkillStore: + """Tests for SkillStore.""" + + def test_init_creates_database(self, tmp_path): + """Test that initializing SkillStore creates parent directory.""" + db_path = tmp_path / "subdir" / "skills.db" + store = SkillStore(db_path) + # Parent directory should be created on init + assert db_path.parent.exists() + # Database file is created lazily on first connection + conn = store._get_connection() + assert conn is not None + store.close() + + def test_save_and_get_skill(self, temp_db, sample_skill): + """Test saving and retrieving a skill.""" + assert temp_db.save_skill(sample_skill) is True + retrieved = temp_db.get_skill(sample_skill.skill_id) + assert retrieved is not None + assert retrieved.name == sample_skill.name + assert retrieved.domain == sample_skill.domain + assert retrieved.quality_score == sample_skill.quality_score + + def test_get_nonexistent_skill(self, temp_db): + """Test retrieving a non-existent skill returns None.""" + result = temp_db.get_skill("nonexistent-skill") + assert result is None + + def test_get_promoted_skills(self, temp_db, sample_skill): + """Test getting promoted skills.""" + sample_skill.promoted = True + temp_db.save_skill(sample_skill) + promoted = temp_db.get_promoted_skills() + assert len(promoted) == 1 + assert promoted[0].skill_id == sample_skill.skill_id + + def test_get_skills_by_domain(self, temp_db, sample_skill): + """Test getting skills by domain.""" + temp_db.save_skill(sample_skill) + skills = temp_db.get_skills_by_domain("testing") + assert len(skills) == 1 + assert skills[0].domain == "testing" + + def test_search_skills(self, temp_db, sample_skill): + """Test searching skills.""" + temp_db.save_skill(sample_skill) + results = temp_db.search_skills("test", min_quality=50.0) + assert len(results) == 1 + + def test_search_skills_by_domain(self, temp_db, sample_skill): + """Test searching skills with domain filter.""" + temp_db.save_skill(sample_skill) + results = temp_db.search_skills("test", domain="testing") + assert len(results) == 1 + results = temp_db.search_skills("test", domain="backend") + assert len(results) == 0 + + def test_search_promoted_only(self, temp_db, sample_skill): + """Test searching only promoted skills.""" + temp_db.save_skill(sample_skill) + results = temp_db.search_skills("test", promoted_only=True) + assert len(results) == 0 + sample_skill.promoted = True + temp_db.save_skill(sample_skill) + results = temp_db.search_skills("test", promoted_only=True) + assert len(results) == 1 + + def test_save_and_get_feedback(self, temp_db, sample_feedback): + """Test saving and retrieving feedback.""" + assert temp_db.save_feedback(sample_feedback) is True + feedback_list = temp_db.get_session_feedback(sample_feedback.session_id) + assert len(feedback_list) == 1 + assert feedback_list[0].quality_before == sample_feedback.quality_before + assert feedback_list[0].quality_after == sample_feedback.quality_after + + def test_record_skill_application(self, temp_db, sample_skill): + """Test recording skill application.""" + temp_db.save_skill(sample_skill) + result = temp_db.record_skill_application( + skill_id=sample_skill.skill_id, + session_id="session-123", + was_helpful=True, + quality_impact=10.0, + feedback="Great skill!", + ) + assert result is True + + def test_get_skill_effectiveness(self, temp_db, sample_skill): + """Test getting skill effectiveness metrics.""" + temp_db.save_skill(sample_skill) + temp_db.record_skill_application( + sample_skill.skill_id, "s1", was_helpful=True, quality_impact=10.0 + ) + temp_db.record_skill_application( + sample_skill.skill_id, "s2", was_helpful=True, quality_impact=5.0 + ) + temp_db.record_skill_application( + sample_skill.skill_id, "s3", was_helpful=False, quality_impact=-2.0 + ) + effectiveness = temp_db.get_skill_effectiveness(sample_skill.skill_id) + assert effectiveness["applications"] == 3 + assert effectiveness["helpful_count"] == 2 + assert effectiveness["unhelpful_count"] == 1 + + def test_get_bulk_skill_effectiveness(self, temp_db, sample_skill, high_quality_skill): + """Test bulk effectiveness fetch.""" + temp_db.save_skill(sample_skill) + temp_db.save_skill(high_quality_skill) + temp_db.record_skill_application( + sample_skill.skill_id, "s1", was_helpful=True, quality_impact=10.0 + ) + results = temp_db.get_bulk_skill_effectiveness( + [sample_skill.skill_id, high_quality_skill.skill_id] + ) + assert sample_skill.skill_id in results + assert high_quality_skill.skill_id in results + assert results[sample_skill.skill_id]["applications"] == 1 + assert results[high_quality_skill.skill_id]["applications"] == 0 + + def test_context_manager(self, tmp_path): + """Test SkillStore as context manager.""" + db_path = tmp_path / "context_test.db" + with SkillStore(db_path) as _store: + # Trigger connection to create the database + _store._get_connection() + assert db_path.parent.exists() + + def test_close_method(self, tmp_path): + """Test explicit close method.""" + db_path = tmp_path / "close_test.db" + store = SkillStore(db_path) + store.close() + # Should not raise even if called again + store.close() + + +# --- LearnedSkill Tests --- + + +class TestLearnedSkill: + """Tests for LearnedSkill dataclass.""" + + def test_to_skill_md(self, sample_skill): + """Test converting skill to SKILL.md format.""" + md = sample_skill.to_skill_md() + assert "---" in md + assert "name: Test Skill" in md + assert "Use pytest fixtures" in md + assert "Don't test implementation details" in md + assert "session-abc123" in md + + +# --- SkillExtractor Tests --- + + +class TestSkillExtractor: + """Tests for SkillExtractor.""" + + def test_extract_from_session_no_feedback(self, temp_db): + """Test extraction with no feedback returns None.""" + extractor = SkillExtractor(temp_db) + result = extractor.extract_from_session("nonexistent-session") + assert result is None + + def test_extract_from_session_insufficient_improvement(self, temp_db): + """Test extraction with insufficient improvement returns None.""" + extractor = SkillExtractor(temp_db) + feedback = IterationFeedback( + session_id="low-quality-session", + iteration=1, + quality_before=50.0, + quality_after=55.0, # Only 5 point improvement + improvements_applied=[], + improvements_needed=[], + changed_files=[], + test_results={}, + duration_seconds=10.0, + success=True, + termination_reason="quality_threshold_met", + ) + temp_db.save_feedback(feedback) + result = extractor.extract_from_session("low-quality-session") + assert result is None + + def test_extract_from_successful_session(self, temp_db): + """Test extraction from a successful session.""" + extractor = SkillExtractor(temp_db) + # Create feedback that meets criteria + feedback1 = IterationFeedback( + session_id="good-session", + iteration=1, + quality_before=50.0, + quality_after=70.0, + improvements_applied=["Added type hints", "Fixed imports"], + improvements_needed=[], + changed_files=["main.py", "utils.py"], + test_results={"ran": True}, + duration_seconds=30.0, + success=True, + termination_reason="", + ) + feedback2 = IterationFeedback( + session_id="good-session", + iteration=2, + quality_before=70.0, + quality_after=85.0, + improvements_applied=["Added tests"], + improvements_needed=[], + changed_files=["test_main.py"], + test_results={"ran": True}, + duration_seconds=20.0, + success=True, + termination_reason="quality_threshold_met", + ) + temp_db.save_feedback(feedback1) + temp_db.save_feedback(feedback2) + result = extractor.extract_from_session("good-session", domain="backend") + assert result is not None + assert result.domain == "backend" + assert result.quality_score >= 70.0 + + +# --- SkillRetriever Tests --- + + +class TestSkillRetriever: + """Tests for SkillRetriever.""" + + def test_retrieve_no_skills(self, temp_db): + """Test retrieval with no skills returns empty list.""" + retriever = SkillRetriever(temp_db) + results = retriever.retrieve("implement auth") + assert results == [] + + def test_retrieve_matching_skills(self, temp_db, sample_skill): + """Test retrieval finds matching skills.""" + temp_db.save_skill(sample_skill) + retriever = SkillRetriever(temp_db) + results = retriever.retrieve("unit testing", promoted_only=False) + assert len(results) >= 1 + skill, score = results[0] + assert skill.skill_id == sample_skill.skill_id + assert score > 0 + + def test_retrieve_with_domain_filter(self, temp_db, sample_skill, high_quality_skill): + """Test retrieval respects domain filter.""" + temp_db.save_skill(sample_skill) + temp_db.save_skill(high_quality_skill) + retriever = SkillRetriever(temp_db) + results = retriever.retrieve("skill", domain="testing", promoted_only=False) + assert all(s.domain == "testing" for s, _ in results) + + def test_retrieve_with_file_paths(self, temp_db, sample_skill): + """Test retrieval considers file paths.""" + temp_db.save_skill(sample_skill) + retriever = SkillRetriever(temp_db) + # Use task description that matches skill triggers + results = retriever.retrieve( + "unit test pytest", file_paths=["test_foo.py", "conftest.py"], promoted_only=False + ) + # Results depend on matching; just verify the method works + assert isinstance(results, list) + + +# --- PromotionGate Tests --- + + +class TestPromotionGate: + """Tests for PromotionGate.""" + + def test_evaluate_low_quality(self, temp_db, sample_skill): + """Test evaluation rejects low quality skills.""" + sample_skill.quality_score = 70.0 # Below threshold + temp_db.save_skill(sample_skill) + gate = PromotionGate(temp_db) + can_promote, reason = gate.evaluate(sample_skill) + assert can_promote is False + assert "quality" in reason.lower() + + def test_evaluate_insufficient_applications(self, temp_db, high_quality_skill): + """Test evaluation rejects skills with too few applications.""" + temp_db.save_skill(high_quality_skill) + gate = PromotionGate(temp_db) + can_promote, reason = gate.evaluate(high_quality_skill) + assert can_promote is False + assert "application" in reason.lower() + + def test_evaluate_meets_criteria(self, temp_db, high_quality_skill): + """Test evaluation approves qualifying skills.""" + temp_db.save_skill(high_quality_skill) + # Add successful applications + for i in range(3): + temp_db.record_skill_application( + high_quality_skill.skill_id, f"session-{i}", was_helpful=True, quality_impact=5.0 + ) + gate = PromotionGate(temp_db) + can_promote, reason = gate.evaluate(high_quality_skill) + assert can_promote is True + + def test_promote_skill(self, temp_db, high_quality_skill, tmp_path): + """Test promoting a skill creates files.""" + temp_db.save_skill(high_quality_skill) + for i in range(3): + temp_db.record_skill_application( + high_quality_skill.skill_id, f"s-{i}", was_helpful=True, quality_impact=5.0 + ) + gate = PromotionGate(temp_db, skills_dir=tmp_path) + path = gate.promote(high_quality_skill, "Test promotion") + assert path is not None + assert path.exists() + assert "SKILL.md" in str(path) + + def test_list_pending(self, temp_db, sample_skill, high_quality_skill): + """Test listing pending skills.""" + temp_db.save_skill(sample_skill) + temp_db.save_skill(high_quality_skill) + gate = PromotionGate(temp_db) + pending = gate.list_pending() + assert len(pending) >= 1 + + +# --- Convenience Function Tests --- + + +class TestConvenienceFunctions: + """Tests for module-level convenience functions.""" + + def test_get_default_store(self, tmp_path, monkeypatch): + """Test get_default_store returns a store.""" + # Override the default path + monkeypatch.setattr( + "core.skill_persistence.SkillStore.DEFAULT_DB_PATH", + tmp_path / "default_test.db", + ) + store = get_default_store() + assert store is not None + store.close() + + def test_learn_from_session_no_data(self, tmp_path, monkeypatch): + """Test learn_from_session with no data returns None.""" + monkeypatch.setattr( + "core.skill_persistence.SkillStore.DEFAULT_DB_PATH", + tmp_path / "learn_test.db", + ) + result = learn_from_session("nonexistent-session") + assert result is None + + def test_retrieve_skills_for_task(self, tmp_path, monkeypatch): + """Test retrieve_skills_for_task works.""" + monkeypatch.setattr( + "core.skill_persistence.SkillStore.DEFAULT_DB_PATH", + tmp_path / "retrieve_test.db", + ) + # Should return empty list when no skills exist + result = retrieve_skills_for_task("implement auth") + assert result == []