feat(core): add skill persistence layer for cross-session learning - #25
Conversation
Implements ACE-inspired skill persistence enabling SuperClaude to learn from successful sessions and apply learned patterns to future tasks. New components: - core/skill_persistence.py: SQLite store, extractor, retriever, promotion gate - core/skill_learning_integration.py: LearningLoopOrchestrator integration - .claude/skills/sc-implement/scripts/skill_learn.py: CLI for skill management - .claude/skills/learned/SKILL.md: Index for promoted skills Key features: - Iteration feedback recording after each loop iteration - Skill extraction from successful sessions (quality >= 85) - Context-based skill retrieval with relevance scoring - Promotion gate with quality thresholds (85+ score, 2+ apps, 70% success) - Full provenance tracking for audit/rollback Safety measures: - Thread-local SQLite connections with WAL mode - Path traversal prevention (skill_id for filesystem paths) - Atomic promotion with rollback on failure - Error handling on all DB operations - Python 3.9 compatibility (Optional[] syntax) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Reviewer's GuideAdds a SQLite-backed skill persistence layer and learning integration around the loop orchestrator, plus a CLI for managing learned skills and associated docs, enabling SuperClaude to record iteration feedback, extract reusable skills, retrieve them by context, and gate promotion into a learned skills directory. Sequence diagram for LearningLoopOrchestrator run with skill persistencesequenceDiagram
actor User
participant CLI as sc_implement
participant LLO as LearningLoopOrchestrator
participant SR as SkillRetriever
participant SS as SkillStore
participant LO as LoopOrchestrator
participant SE as SkillExtractor
participant PG as PromotionGate
User->>CLI: invoke --loop --learn --auto-promote
CLI->>LLO: create LearningLoopOrchestrator(config, enable_learning, auto_promote)
LLO->>SS: SkillStore.__init__()
LLO->>SR: SkillRetriever.__init__(store)
LLO->>SE: SkillExtractor.__init__(store)
LLO->>PG: PromotionGate.__init__(store)
CLI->>LLO: run(initial_context, skill_invoker)
LLO->>LLO: _detect_domain(initial_context)
LLO->>SR: retrieve(task_description, file_paths, domain, max_skills, promoted_only=false)
SR->>SS: search_skills(query, domain, min_quality, promoted_only)
SS-->>SR: List[LearnedSkill]
SR->>SS: get_bulk_skill_effectiveness(skill_ids)
SS-->>SR: effectiveness_map
SR-->>LLO: List[(LearnedSkill, score)]
LLO->>LLO: _inject_relevant_skills(context)
LLO->>LO: super.run(context_with_skills, skill_invoker)
loop iterations
LO->>skill_invoker: invoke(context)
skill_invoker-->>LO: iteration_output
LO->>LO: _record_iteration(...)
end
LO-->>LLO: LoopResult
LLO->>SS: save_feedback(IterationFeedback) for each iteration
alt termination_reason == QUALITY_MET
LLO->>SE: extract_from_session(session_id, repo_path, domain)
SE->>SS: get_session_feedback(session_id)
SS-->>SE: list[IterationFeedback]
SE-->>LLO: LearnedSkill or None
opt skill extracted
LLO->>SS: save_skill(skill)
alt auto_promote
LLO->>PG: evaluate(skill)
PG-->>LLO: should_promote, reason
alt should_promote
LLO->>PG: promote(skill, reason)
PG->>SS: save_skill(promoted_skill)
SS-->>PG: ok
end
end
end
end
opt applied_skills
LLO->>SS: record_skill_application(skill_id, session_id, was_helpful, quality_impact, feedback) for each skill
end
LLO-->>CLI: LoopResult with learning metadata
CLI-->>User: display result and session_id
Sequence diagram for skill_learn.py retrieve and promote commandssequenceDiagram
actor Dev as Developer
participant SkillCLI as skill_learn.py
participant SS as SkillStore
participant SR as SkillRetriever
participant PG as PromotionGate
Dev->>SkillCLI: python skill_learn.py '{command: "retrieve", task: ...}'
SkillCLI->>SS: SkillStore.__init__()
SkillCLI->>SR: SkillRetriever.__init__(store)
SkillCLI->>SR: retrieve(task_description, file_paths, domain, max_skills, promoted_only)
SR->>SS: search_skills(query, domain, min_quality, promoted_only)
SS-->>SR: List[LearnedSkill]
SR->>SS: get_bulk_skill_effectiveness(skill_ids)
SS-->>SR: effectiveness_map
SR-->>SkillCLI: List[(LearnedSkill, score)]
SkillCLI-->>Dev: JSON skills with relevance and patterns
Dev->>SkillCLI: python skill_learn.py '{command: "promote", skill_id: ...}'
SkillCLI->>SS: SkillStore.__init__()
SkillCLI->>PG: PromotionGate.__init__(store)
SkillCLI->>SS: get_skill(skill_id)
SS-->>SkillCLI: LearnedSkill
SkillCLI->>PG: evaluate(skill)
PG-->>SkillCLI: can_promote, reason
alt can_promote
SkillCLI->>PG: promote(skill, reason)
PG->>SS: save_skill(promoted_skill)
SS-->>PG: ok
SkillCLI-->>Dev: JSON success with promoted_to path
else cannot promote
SkillCLI-->>Dev: JSON error with evaluation reason
end
ER diagram for learned skill persistence databaseerDiagram
learned_skills {
TEXT skill_id PK
TEXT name
TEXT description
TEXT triggers
TEXT domain
TEXT source_session
TEXT source_repo
TEXT learned_at
TEXT patterns
TEXT anti_patterns
REAL quality_score
INTEGER iteration_count
TEXT provenance
TEXT applicability_conditions
INTEGER promoted
TEXT promotion_reason
TEXT created_at
TEXT updated_at
}
iteration_feedback {
INTEGER id PK
TEXT session_id
INTEGER iteration
REAL quality_before
REAL quality_after
TEXT improvements_applied
TEXT improvements_needed
TEXT changed_files
TEXT test_results
REAL duration_seconds
INTEGER success
TEXT termination_reason
TEXT timestamp
TEXT created_at
}
skill_applications {
INTEGER id PK
TEXT skill_id FK
TEXT session_id
TEXT applied_at
INTEGER was_helpful
REAL quality_impact
TEXT feedback
}
learned_skills ||--o{ skill_applications : skill_id
iteration_feedback ||--o{ skill_applications : session_id
iteration_feedback ||--o{ iteration_feedback : session_id
Class diagram for core skill persistence layer and learning orchestratorclassDiagram
class SkillStore {
+Path db_path
+DEFAULT_DB_PATH
+__init__(db_path Optional[Path])
+close() None
+save_skill(skill LearnedSkill) bool
+get_skill(skill_id str) Optional[LearnedSkill]
+get_promoted_skills() list[LearnedSkill]
+get_skills_by_domain(domain str) list[LearnedSkill]
+search_skills(query str, domain Optional[str], min_quality float, promoted_only bool) List[LearnedSkill]
+save_feedback(feedback IterationFeedback) bool
+get_session_feedback(session_id str) list[IterationFeedback]
+record_skill_application(skill_id str, session_id str, was_helpful Optional[bool], quality_impact Optional[float], feedback str) bool
+get_skill_effectiveness(skill_id str) Dict[str, Any]
+get_bulk_skill_effectiveness(skill_ids List[str]) Dict[str, Dict[str, Any]]
-_get_connection() sqlite3.Connection
-_init_schema() None
-_row_to_skill(row sqlite3.Row) LearnedSkill
}
class LearnedSkill {
+str skill_id
+str name
+str description
+list[str] triggers
+str domain
+str source_session
+str source_repo
+str learned_at
+list[str] patterns
+list[str] anti_patterns
+float quality_score
+int iteration_count
+dict[str, Any] provenance
+list[str] applicability_conditions
+bool promoted
+str promotion_reason
+to_dict() dict[str, Any]
+to_skill_md() str
+from_dict(data dict[str, Any]) LearnedSkill
}
class IterationFeedback {
+str session_id
+int iteration
+float quality_before
+float quality_after
+list[str] improvements_applied
+list[str] improvements_needed
+list[str] changed_files
+dict[str, Any] test_results
+float duration_seconds
+bool success
+str termination_reason
+str timestamp
+to_dict() dict[str, Any]
+from_dict(data dict[str, Any]) IterationFeedback
}
class SkillExtractor {
+SkillStore store
+__init__(store SkillStore)
+extract_from_session(session_id str, repo_path str, domain str) Optional[LearnedSkill]
-_extract_patterns(feedback_list list[IterationFeedback]) list[str]
-_extract_anti_patterns(feedback_list list[IterationFeedback]) list[str]
-_extract_triggers(feedback_list list[IterationFeedback]) list[str]
-_extract_conditions(feedback_list list[IterationFeedback]) list[str]
-_generate_skill_id(session_id str, patterns list[str]) str
-_generate_skill_name(patterns list[str], domain str) str
}
class SkillRetriever {
+SkillStore store
+__init__(store SkillStore)
+retrieve(task_description str, file_paths Optional[List[str]], domain Optional[str], max_skills int, promoted_only bool) List[Tuple[LearnedSkill, float]]
-_extract_search_terms(task_description str, file_paths Optional[List[str]]) set
-_score_relevance(skill LearnedSkill, search_terms set, file_paths Optional[List[str]], effectiveness Optional[Dict[str, Any]]) float
}
class PromotionGate {
+SkillStore store
+Path skills_dir
+MIN_QUALITY_SCORE
+MIN_APPLICATIONS
+MIN_SUCCESS_RATE
+__init__(store SkillStore, skills_dir Optional[Path])
+evaluate(skill LearnedSkill) Tuple[bool, str]
+promote(skill LearnedSkill, reason str) Optional[Path]
+list_pending() List[LearnedSkill]
}
class LoopOrchestrator {
+LoopConfig config
+__init__(config Optional[LoopConfig])
+run(initial_context Dict[str, Any], skill_invoker Callable[[Dict[str, Any]], Dict[str, Any]]) LoopResult
-_record_iteration(iteration int, input_quality float, output_quality float, success bool, termination str, changed_files list[str], pal_signal Optional[Dict[str, Any]]) None
}
class LearningLoopOrchestrator {
+bool enable_learning
+bool auto_promote
+SkillStore store
+SkillExtractor extractor
+SkillRetriever retriever
+PromotionGate promotion_gate
+str session_id
+str repo_path
+str domain
+List[LearnedSkill] _applied_skills
+float _initial_quality
+__init__(config Optional[LoopConfig], store Optional[SkillStore], enable_learning bool, auto_promote bool)
+run(initial_context Dict[str, Any], skill_invoker Callable[[Dict[str, Any]], Dict[str, Any]]) LoopResult
-_detect_repo_path() str
-_detect_domain(context Dict[str, Any]) str
-_inject_relevant_skills(context Dict[str, Any]) Dict[str, Any]
-_record_all_feedback(result LoopResult) None
-_extract_and_save_skill(result LoopResult) Optional[LearnedSkill]
-_record_skill_effectiveness(result LoopResult) None
}
class SkillLearnCLI {
<<script>>
+handle_list(args dict[str, Any]) dict[str, Any]
+handle_promote(args dict[str, Any]) dict[str, Any]
+handle_stats(args dict[str, Any]) dict[str, Any]
+handle_retrieve(args dict[str, Any]) dict[str, Any]
+handle_export(args dict[str, Any]) dict[str, Any]
+handle_pending(args dict[str, Any]) dict[str, Any]
+handle_delete(args dict[str, Any]) dict[str, Any]
+main() None
}
SkillExtractor --> SkillStore : uses
SkillRetriever --> SkillStore : uses
PromotionGate --> SkillStore : uses
LearningLoopOrchestrator --|> LoopOrchestrator
LearningLoopOrchestrator --> SkillStore
LearningLoopOrchestrator --> SkillExtractor
LearningLoopOrchestrator --> SkillRetriever
LearningLoopOrchestrator --> PromotionGate
SkillStore --> LearnedSkill
SkillStore --> IterationFeedback
SkillLearnCLI ..> SkillStore
SkillLearnCLI ..> SkillRetriever
SkillLearnCLI ..> PromotionGate
SkillLearnCLI ..> LearnedSkill
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
|
Warning Rate limit exceeded@Tony363 has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 7 minutes and 1 seconds before requesting another review. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. 📒 Files selected for processing (1)
Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. 📝 WalkthroughWalkthroughAdds a persistence-backed learned-skills subsystem (extraction, retrieval, promotion), integrates learning into the loop orchestrator to inject and record learned skills, supplies a CLI for skill management, new docs describing learned skills, and consistent Optional-based typing updates across core types. Changes
Sequence Diagram(s)sequenceDiagram
participant User as User/CLI
participant Orch as LearningLoopOrchestrator
participant Store as SkillStore
participant Retriever as SkillRetriever
participant Extractor as SkillExtractor
participant Gate as PromotionGate
participant Invoker as SkillInvoker
User->>Orch: run_learning_loop(task)
Orch->>Store: get_skills_by_domain(detected_domain)
Store-->>Retriever: promoted_skills
Retriever->>Store: get_bulk_skill_effectiveness(skill_ids)
Store-->>Retriever: effectiveness_data
Retriever-->>Orch: ranked_skills
rect rgb(220,240,255)
Note over Orch,Invoker: Loop execution with injected learned_skills
Orch->>Invoker: invoke(context_with_learned_skills)
Invoker-->>Orch: iteration_result
end
rect rgb(230,255,230)
Note over Orch,Extractor: Post-run learning flow (on success)
Orch->>Store: save_feedback(iteration_feedback)
Orch->>Extractor: extract_from_session(session_id)
Extractor->>Store: get_session_feedback(session_id)
Store-->>Extractor: feedback_entries
Extractor-->>Store: save_skill(learned_skill)
Orch->>Store: record_skill_application(applied_skill_ids)
end
rect rgb(255,240,220)
Note over Orch,Gate: Promotion (if auto_promote)
Orch->>Gate: evaluate(skill)
Gate-->>Orch: evaluation_result
alt criteria met
Gate->>Gate: promote(skill) / write SKILL.md + metadata.json
Gate-->>Orch: promotion_path
end
end
Orch-->>User: LoopResult + session_id + applied_skills
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Poem
Pre-merge checks and finishing touches✅ Passed checks (3 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Hey - I've found 1 security issue, 4 other issues, and left some high level feedback:
Security issues:
- Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option. (link)
General comments:
- Several callers (e.g.
PromotionGate.list_pending,skill_learn.pyhandlers,get_skill_stats) reach intoSkillStore's private internals (_get_connection,_row_to_skill), which makes the persistence layer hard to evolve; consider exposing explicit public query methods onSkillStoreinstead of relying on its internal connection and row-conversion helpers. - In
LearningLoopOrchestrator,_initial_qualityis initialized but never set from any real assessment before_record_skill_effectivenessruns, soquality_impactis always measured from zero rather than the actual starting quality; it would be more meaningful to capture the initial quality (e.g. from the first assessment) and use that as the baseline. - The CLI (
skill_learn.py) dynamically loadscore/skill_persistence.pyvia a hard-codedSUPERCLAUD_ROOTandimportlib.util, which is brittle to project layout changes; consider using a standard package import path or a single shared entry point to these types to avoid path drift and duplicated import logic.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- Several callers (e.g. `PromotionGate.list_pending`, `skill_learn.py` handlers, `get_skill_stats`) reach into `SkillStore`'s private internals (`_get_connection`, `_row_to_skill`), which makes the persistence layer hard to evolve; consider exposing explicit public query methods on `SkillStore` instead of relying on its internal connection and row-conversion helpers.
- In `LearningLoopOrchestrator`, `_initial_quality` is initialized but never set from any real assessment before `_record_skill_effectiveness` runs, so `quality_impact` is always measured from zero rather than the actual starting quality; it would be more meaningful to capture the initial quality (e.g. from the first assessment) and use that as the baseline.
- The CLI (`skill_learn.py`) dynamically loads `core/skill_persistence.py` via a hard-coded `SUPERCLAUD_ROOT` and `importlib.util`, which is brittle to project layout changes; consider using a standard package import path or a single shared entry point to these types to avoid path drift and duplicated import logic.
## Individual Comments
### Comment 1
<location> `.claude/skills/sc-implement/scripts/skill_learn.py:28` </location>
<code_context>
+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"
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Repository root detection walks one directory too far, likely breaking the dynamic import of core/skill_persistence.py.
With the script at `.claude/skills/sc-implement/scripts/skill_learn.py`, five `.parent` hops puts you at the *parent* of the repo root, so `SKILL_PERSISTENCE_PATH` will resolve to `<repo_root_parent>/core/skill_persistence.py`, which won’t exist. Using four `.parent` hops should correctly locate the repo root:
```python
SUPERCLAUD_ROOT = Path(__file__).parent.parent.parent.parent
SKILL_PERSISTENCE_PATH = SUPERCLAUD_ROOT / "core" / "skill_persistence.py"
```
Consider also asserting `SKILL_PERSISTENCE_PATH.exists()` and failing with a clear error if it does not.
</issue_to_address>
### Comment 2
<location> `core/skill_learning_integration.py:87-89` </location>
<code_context>
+ 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:
</code_context>
<issue_to_address>
**issue (bug_risk):** Skill effectiveness uses a fixed initial quality of 0, which can distort the recorded quality_impact.
In `_record_skill_effectiveness`, `quality_impact` is calculated as `final_quality - self._initial_quality`, but `_initial_quality` is never updated and remains 0. That means `quality_impact` is effectively just `final_quality` for every session, so the effectiveness metrics are misleading. Consider initializing `_initial_quality` from the first real quality score (e.g., a baseline assessment or `iteration_history[0].input_quality`) before the loop, or derive the delta directly from `iteration_history` instead of a never-updated attribute.
</issue_to_address>
### Comment 3
<location> `.claude/skills/sc-implement/scripts/skill_learn.py:27` </location>
<code_context>
+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"
</code_context>
<issue_to_address>
**issue (complexity):** Consider replacing the custom importlib-based loading of `skill_persistence` with a standard import after adding the project root to `sys.path` to simplify and clarify the module dependency.
The main complexity spike is in the low‑level dynamic import of `skill_persistence`. You can keep the “avoid `core/__init__.py` issues” goal while using a much simpler and more idiomatic pattern.
Instead of building a spec, creating a module, injecting into `sys.modules`, and then copying attributes:
```python
import sys
import importlib.util
from pathlib import Path
# 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
```
you can:
1. Ensure the project root is on `sys.path`.
2. Use a normal import, which is easier to reason about and debug.
3. Avoid the extra indirection through `sp`.
For example:
```python
import sys
from pathlib import Path
# Ensure project root is importable
SUPERCLAUD_ROOT = Path(__file__).resolve().parents[5]
sys.path.insert(0, str(SUPERCLAUD_ROOT))
from core.skill_persistence import (
LearnedSkill,
PromotionGate,
SkillExtractor,
SkillRetriever,
SkillStore,
)
```
This keeps all functionality intact while:
- Removing the manual spec/module management.
- Making the dependency explicit and aligned with normal Python import semantics.
- Avoiding the extra `sp` module indirection that every handler depends on.
</issue_to_address>
### Comment 4
<location> `core/skill_learning_integration.py:323` </location>
<code_context>
+
+# --- CLI Entry Point ---
+
+def run_learning_loop(
+ task: str,
+ max_iterations: int = 3,
</code_context>
<issue_to_address>
**issue (complexity):** Consider splitting CLI/admin helpers and DB stats logic out of this orchestrator module so it stays focused on loop+learning integration while persistence and entrypoints live in their own layers.
The orchestrator class is a solid abstraction, but the module is doing too much: it mixes core loop integration, CLI entrypoint, and persistence utilities (including raw SQL). You can reduce complexity without changing behavior by:
1. **Splitting CLI / admin utilities into a separate module**
2. **Pushing the raw SQL stats logic into the persistence layer**
### 1. Move CLI + admin helpers into a separate module
Keep this module focused on the orchestrator and signal construction:
```python
# core/skill_learning_integration.py
class LearningLoopOrchestrator(LoopOrchestrator):
...
# unchanged
def create_learning_invoker_signal(
context: Dict[str, Any],
learned_skills: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
...
# unchanged
```
Create a new module for CLI/service-style helpers, e.g. `core/skill_learning_cli.py`:
```python
# core/skill_learning_cli.py
from typing import Any, Dict, List
from .skill_learning_integration import LearningLoopOrchestrator
from .types import LoopConfig, TerminationReason
from .skill_persistence import SkillStore, PromotionGate
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]:
config = LoopConfig(
max_iterations=max_iterations,
quality_threshold=quality_threshold,
)
orchestrator = LearningLoopOrchestrator(
config=config,
enable_learning=enable_learning,
auto_promote=auto_promote,
)
initial_context = {
"task": task,
"improvements_needed": [],
"changed_files": [],
}
def placeholder_invoker(ctx: dict[str, Any]) -> dict[str, Any]:
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,
}
def list_pending_skills() -> List[Dict[str, Any]]:
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:
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
```
This keeps the orchestrator file smaller and conceptually focused (loop + learning integration only).
### 2. Move stats query into the persistence layer
The raw SQL in `get_skill_stats` couples your integration module to DB schema and duplicates persistence concerns. Move that into `SkillStore` (or a dedicated repository) and call it from the CLI module:
```python
# core/skill_persistence.py
class SkillStore:
...
def get_skill_stats(self) -> Dict[str, Any]:
conn = self._get_connection()
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()
feedback_count = conn.execute(
"SELECT COUNT(*) FROM iteration_feedback"
).fetchone()[0]
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
),
}
```
Then the CLI/service function becomes a thin wrapper:
```python
# core/skill_learning_cli.py
from .skill_persistence import SkillStore
def get_skill_stats() -> Dict[str, Any]:
store = SkillStore()
return store.get_skill_stats()
```
This preserves all current behavior while:
- Making the orchestrator module smaller and easier to navigate.
- Centralizing DB schema knowledge in `SkillStore`, avoiding duplicated SQL and reducing maintenance cost.
- Making CLI-style entrypoints clearly separated from the core loop integration.
</issue_to_address>
### Comment 5
<location> `core/skill_persistence.py:504-514` </location>
<code_context>
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()
</code_context>
<issue_to_address>
**security (python.sqlalchemy.security.sqlalchemy-execute-raw-query):** Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
*Source: opengrep*
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| from typing import Any | ||
|
|
||
| # Load skill_persistence directly to avoid core/__init__.py import issues | ||
| SUPERCLAUD_ROOT = Path(__file__).parent.parent.parent.parent.parent |
There was a problem hiding this comment.
issue (bug_risk): Repository root detection walks one directory too far, likely breaking the dynamic import of core/skill_persistence.py.
With the script at .claude/skills/sc-implement/scripts/skill_learn.py, five .parent hops puts you at the parent of the repo root, so SKILL_PERSISTENCE_PATH will resolve to <repo_root_parent>/core/skill_persistence.py, which won’t exist. Using four .parent hops should correctly locate the repo root:
SUPERCLAUD_ROOT = Path(__file__).parent.parent.parent.parent
SKILL_PERSISTENCE_PATH = SUPERCLAUD_ROOT / "core" / "skill_persistence.py"Consider also asserting SKILL_PERSISTENCE_PATH.exists() and failing with a clear error if it does not.
| from pathlib import Path | ||
| from typing import Any | ||
|
|
||
| # Load skill_persistence directly to avoid core/__init__.py import issues |
There was a problem hiding this comment.
issue (complexity): Consider replacing the custom importlib-based loading of skill_persistence with a standard import after adding the project root to sys.path to simplify and clarify the module dependency.
The main complexity spike is in the low‑level dynamic import of skill_persistence. You can keep the “avoid core/__init__.py issues” goal while using a much simpler and more idiomatic pattern.
Instead of building a spec, creating a module, injecting into sys.modules, and then copying attributes:
import sys
import importlib.util
from pathlib import Path
# 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.SkillStoreyou can:
- Ensure the project root is on
sys.path. - Use a normal import, which is easier to reason about and debug.
- Avoid the extra indirection through
sp.
For example:
import sys
from pathlib import Path
# Ensure project root is importable
SUPERCLAUD_ROOT = Path(__file__).resolve().parents[5]
sys.path.insert(0, str(SUPERCLAUD_ROOT))
from core.skill_persistence import (
LearnedSkill,
PromotionGate,
SkillExtractor,
SkillRetriever,
SkillStore,
)This keeps all functionality intact while:
- Removing the manual spec/module management.
- Making the dependency explicit and aligned with normal Python import semantics.
- Avoiding the extra
spmodule indirection that every handler depends on.
|
|
||
| # --- CLI Entry Point --- | ||
|
|
||
| def run_learning_loop( |
There was a problem hiding this comment.
issue (complexity): Consider splitting CLI/admin helpers and DB stats logic out of this orchestrator module so it stays focused on loop+learning integration while persistence and entrypoints live in their own layers.
The orchestrator class is a solid abstraction, but the module is doing too much: it mixes core loop integration, CLI entrypoint, and persistence utilities (including raw SQL). You can reduce complexity without changing behavior by:
- Splitting CLI / admin utilities into a separate module
- Pushing the raw SQL stats logic into the persistence layer
1. Move CLI + admin helpers into a separate module
Keep this module focused on the orchestrator and signal construction:
# core/skill_learning_integration.py
class LearningLoopOrchestrator(LoopOrchestrator):
...
# unchanged
def create_learning_invoker_signal(
context: Dict[str, Any],
learned_skills: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
...
# unchangedCreate a new module for CLI/service-style helpers, e.g. core/skill_learning_cli.py:
# core/skill_learning_cli.py
from typing import Any, Dict, List
from .skill_learning_integration import LearningLoopOrchestrator
from .types import LoopConfig, TerminationReason
from .skill_persistence import SkillStore, PromotionGate
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]:
config = LoopConfig(
max_iterations=max_iterations,
quality_threshold=quality_threshold,
)
orchestrator = LearningLoopOrchestrator(
config=config,
enable_learning=enable_learning,
auto_promote=auto_promote,
)
initial_context = {
"task": task,
"improvements_needed": [],
"changed_files": [],
}
def placeholder_invoker(ctx: dict[str, Any]) -> dict[str, Any]:
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,
}
def list_pending_skills() -> List[Dict[str, Any]]:
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:
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 NoneThis keeps the orchestrator file smaller and conceptually focused (loop + learning integration only).
2. Move stats query into the persistence layer
The raw SQL in get_skill_stats couples your integration module to DB schema and duplicates persistence concerns. Move that into SkillStore (or a dedicated repository) and call it from the CLI module:
# core/skill_persistence.py
class SkillStore:
...
def get_skill_stats(self) -> Dict[str, Any]:
conn = self._get_connection()
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()
feedback_count = conn.execute(
"SELECT COUNT(*) FROM iteration_feedback"
).fetchone()[0]
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
),
}Then the CLI/service function becomes a thin wrapper:
# core/skill_learning_cli.py
from .skill_persistence import SkillStore
def get_skill_stats() -> Dict[str, Any]:
store = SkillStore()
return store.get_skill_stats()This preserves all current behavior while:
- Making the orchestrator module smaller and easier to navigate.
- Centralizing DB schema knowledge in
SkillStore, avoiding duplicated SQL and reducing maintenance cost. - Making CLI-style entrypoints clearly separated from the core loop integration.
🤖 PAL MCP Consensus Code ReviewOverviewThis PR implements a skill persistence layer enabling cross-session learning in SuperClaude. The implementation includes SQLite-based storage for learned skills, iteration feedback tracking, skill extraction/retrieval with relevance scoring, and a CLI management tool. The system integrates with the existing loop orchestrator to automatically learn from successful execution patterns. 🔴 Critical Issues1. SQL Injection Vulnerability in Dynamic Query Building
rows = conn.execute(f"""
...
WHERE skill_id IN ({placeholders})
GROUP BY skill_id
""", skill_ids).fetchall()
2. Path Traversal Prevention Incomplete
3. Database Connection Not Thread-Safe in Error Paths
🟠 High Priority4. Missing Input Validation in CLI
5. Insufficient Error Handling in Database Operations
except sqlite3.Error as e:
print(f"[SkillStore] Failed to save skill {skill.skill_id}: {e}", file=sys.stderr)
return False
6. Division by Zero Not Protected
"success_rate": round(
(app_stats["helpful"] or 0) / app_stats["total_applications"]
if app_stats["total_applications"] else 0,
2
),
7. Resource Cleanup Not Guaranteed
8. Hardcoded Credentials Path
🟡 Medium Priority9. Type Hints Inconsistent with Python 3.9 Compatibility
triggers: list[str] # Python 3.9+ syntax
10. Magic Numbers Throughout Code
11. Inefficient String Operations
12. Missing Docstring Parameter Documentation
13. No Logging Framework
14. Test Results Always Empty
test_results={}, # Would need to extract from output
15. Unused Imports
import os
import time
🟢 Positive Observations
📊 Review Summary
🎯 RecommendationsBefore Merge (Critical):
Short Term (High Priority):
Long Term (Medium Priority):
📝 Testing RequirementsCritical Missing Tests:
Test Coverage Target: Minimum 80% for database and security-critical code This review was generated using manual code analysis following PAL MCP principles. Recommendation: ✅ Approve with Conditions - Address critical security issues (#1-3) and add test coverage before production deployment. |
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (7)
.claude/skills/sc-implement/SKILL.md (1)
126-132: Add language specifier to fenced code block.The directory structure code block should have a language specifier for consistency and to satisfy markdownlint (MD040).
🔎 Proposed fix
-``` +```text .claude/skills/learned/ ├── SKILL.md # Index ├── learned-backend-auth/ # Example promoted skill │ ├── SKILL.md │ └── metadata.json</details> </blockquote></details> <details> <summary>.claude/skills/sc-implement/scripts/skill_learn.py (2)</summary><blockquote> `51-65`: **Accessing private methods (`_get_connection`, `_row_to_skill`) couples CLI to internal implementation.** Multiple handlers directly access `store._get_connection()` and `store._row_to_skill()`, which are internal implementation details. If the store's internals change, this script breaks. Consider adding public methods to `SkillStore` (e.g., `list_skills(filters)`, `get_stats()`) or at minimum document this coupling as intentional. Also applies to: 134-134, 346-346 --- `354-367`: **Delete handler lacks confirmation and could orphan related data.** The delete operation removes skills and applications from the database but doesn't verify if related `iteration_feedback` records reference this skill. Also, for destructive operations in a CLI, consider adding a `--force` flag or confirmation mechanism. </blockquote></details> <details> <summary>.claude/skills/learned/SKILL.md (1)</summary><blockquote> `37-46`: **Add language specifier to fenced code block.** The directory structure code block should have a language specifier for consistency and to satisfy markdownlint (MD040). <details> <summary>🔎 Proposed fix</summary> ```diff -``` +```text 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</details> </blockquote></details> <details> <summary>core/skill_learning_integration.py (2)</summary><blockquote> `420-457`: **`get_skill_stats` accesses private `_get_connection` method.** Similar to the CLI script, this function accesses internal implementation details. Consider adding a public `get_stats()` method to `SkillStore` to encapsulate this query logic. --- `363-371`: **Placeholder invoker always returns empty results.** The `placeholder_invoker` function will cause the loop to terminate after max iterations with zero quality. This is documented as a placeholder, but consider adding a docstring note that this function is only for demonstration/testing and should be replaced by Claude Code's actual invoker. </blockquote></details> <details> <summary>core/skill_persistence.py (1)</summary><blockquote> `959-968`: **Consider adding a public method to SkillStore for better encapsulation.** Lines 961 and 968 directly access private methods `_get_connection()` and `_row_to_skill()` from SkillStore. While this works given the tight coupling, it breaks encapsulation and could become fragile if SkillStore internals change. <details> <summary>💡 Suggested refactor</summary> Add a public method to SkillStore: ```python def get_skills_above_quality(self, min_quality: float, promoted: Optional[bool] = None) -> List[LearnedSkill]: """Get skills above a quality threshold, optionally filtered by promotion status.""" conn = self._get_connection() sql = "SELECT * FROM learned_skills WHERE quality_score >= ?" params = [min_quality] if promoted is not None: sql += " AND promoted = ?" params.append(1 if promoted else 0) sql += " ORDER BY quality_score DESC" rows = conn.execute(sql, params).fetchall() return [self._row_to_skill(row) for row in rows]Then update
list_pending():def list_pending(self) -> List[LearnedSkill]: """List skills pending promotion review.""" return self.store.get_skills_above_quality( min_quality=self.MIN_QUALITY_SCORE - 10, promoted=False )
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (8)
.claude/skills/learned/SKILL.md.claude/skills/sc-implement/SKILL.md.claude/skills/sc-implement/scripts/skill_learn.pycore/loop_orchestrator.pycore/quality_assessment.pycore/skill_learning_integration.pycore/skill_persistence.pycore/types.py
🧰 Additional context used
📓 Path-based instructions (1)
**/*.py
📄 CodeRabbit inference engine (AGENTS.md)
**/*.py: Python code targets 3.8+ with Black (88 cols), Flake8, and MyPy; use 4-space indents
Use snake_case for Python module and variable names
Use PascalCase for Python agent classes aligned with their persona names
Files:
core/loop_orchestrator.pycore/quality_assessment.pycore/types.pycore/skill_learning_integration.pycore/skill_persistence.py
🧬 Code graph analysis (3)
core/loop_orchestrator.py (1)
core/types.py (1)
LoopConfig(32-61)
.claude/skills/sc-implement/scripts/skill_learn.py (1)
core/skill_persistence.py (13)
LearnedSkill(39-126)PromotionGate(850-968)SkillRetriever(736-847)SkillStore(154-539)_get_connection(168-179)_row_to_skill(357-376)get_skill(296-303)evaluate(866-895)promote(897-957)retrieve(746-788)to_skill_md(66-126)list_pending(959-968)get_skill_effectiveness(468-490)
core/skill_persistence.py (2)
core/types.py (1)
to_dict(137-158)archive/python-sdk-v5/Telemetry/evidence_store.py (1)
query(256-369)
🪛 GitHub Actions: CI
.claude/skills/sc-implement/scripts/skill_learn.py
[error] 19-19: I001 Import block is un-sorted or un-formatted
🪛 GitHub Check: CodeQL
core/types.py
[notice] 11-11: Unused import
Import of 'List' is not used.
core/skill_learning_integration.py
[notice] 22-22: Unused import
Import of 'os' is not used.
[notice] 23-23: Unused import
Import of 'time' is not used.
[notice] 37-43: Unused import
Import of 'IterationResult' is not used.
Import of 'QualityAssessment' is not used.
core/skill_persistence.py
[notice] 32-32: Unused import
Import of 'Callable' is not used.
🪛 GitHub Check: Quality Gate
.claude/skills/sc-implement/scripts/skill_learn.py
[failure] 19-25: Ruff (I001)
.claude/skills/sc-implement/scripts/skill_learn.py:19:1: I001 Import block is un-sorted or un-formatted
core/types.py
[failure] 11-11: Ruff (F401)
core/types.py:11:31: F401 typing.List imported but unused
core/skill_learning_integration.py
[failure] 41-41: Ruff (F401)
core/skill_learning_integration.py:41:5: F401 .types.QualityAssessment imported but unused
[failure] 38-38: Ruff (F401)
core/skill_learning_integration.py:38:5: F401 .types.IterationResult imported but unused
[failure] 23-23: Ruff (F401)
core/skill_learning_integration.py:23:8: F401 time imported but unused
[failure] 22-22: Ruff (F401)
core/skill_learning_integration.py:22:8: F401 os imported but unused
core/skill_persistence.py
[failure] 32-32: Ruff (F401)
core/skill_persistence.py:32:25: F401 typing.Callable imported but unused
[failure] 22-32: Ruff (I001)
core/skill_persistence.py:22:1: I001 Import block is un-sorted or un-formatted
🪛 markdownlint-cli2 (0.18.1)
.claude/skills/learned/SKILL.md
37-37: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
.claude/skills/sc-implement/SKILL.md
126-126: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: Sourcery review
- GitHub Check: PAL MCP Consensus Code Review
- GitHub Check: CodeQL Analysis
🔇 Additional comments (17)
.claude/skills/sc-implement/SKILL.md (1)
91-133: Documentation for Loop Mode & Learning looks good.The new section clearly documents the learning workflow, loop flags, example usage, and directory structure. The resource link to
scripts/skill_learn.pyis appropriately added.Also applies to: 139-139
.claude/skills/sc-implement/scripts/skill_learn.py (2)
388-434: Main entry point and command routing look well-structured.Good error handling with JSON responses for invalid input, missing commands, and execution errors. The HANDLERS mapping provides clean extensibility.
289-293: Path handling in export is safe;skill_idformat is constrained by design.The code is secure. The
skill.skill_idretrieved from the database is generated internally via_generate_skill_id(), which produces formats likelearned-<12-char-hex>containing only alphanumerics and hyphens. Sinceskill_idvalues are never created from untrusted user input and follow a strictly safe format by design, path traversal through..or/separators is not possible.core/quality_assessment.py (1)
8-9: Typing updates align with Python 3.8+ compatibility.The change from
Path | NonetoOptional[Path]and the addition offrom __future__ import annotationsensure compatibility with Python 3.8/3.9 as per coding guidelines. As per coding guidelines, Python code targets 3.8+.Also applies to: 14-14, 38-38
core/types.py (1)
7-8: Optional typing updates for Python 3.8+ compatibility look good.The changes from union syntax (
float | None) toOptional[float]andOptional[Dict[str, Any]]ensure backward compatibility with Python 3.8/3.9.Also applies to: 54-54, 112-112
.claude/skills/learned/SKILL.md (1)
1-73: Learned Skills Index documentation is comprehensive and well-structured.The documentation clearly explains the learning workflow, skill lifecycle statuses, promotion thresholds, directory structure, integration points, and safety/provenance features. This provides good guidance for users of the skill persistence system.
core/loop_orchestrator.py (1)
17-18: Typing updates for Python 3.8+ compatibility are consistent.The changes from union syntax to
Optionaltyping align with the broader PR's approach to ensure compatibility with Python 3.8/3.9 as per coding guidelines.Also applies to: 20-20, 58-58, 273-273
core/skill_learning_integration.py (3)
46-56: LearningLoopOrchestrator design and integration look solid.The orchestrator cleanly extends
LoopOrchestratorwith learning capabilities: skill retrieval/injection, feedback recording, skill extraction, and effectiveness tracking. The separation of concerns is well-maintained.Also applies to: 102-139
141-183: Domain detection logic is practical and extensible.The keyword-based and file-extension-based domain detection provides reasonable defaults. The fallback to "general" domain is appropriate.
285-318:create_learning_invoker_signalprovides clean skill context injection.The function properly extends the base signal with learned patterns and anti-patterns for Claude Code to use during skill execution.
core/skill_persistence.py (7)
38-127: LGTM! Well-structured skill representation.The
LearnedSkilldataclass provides a comprehensive model with all necessary metadata for tracking skill provenance and effectiveness. Theto_skill_md()method generates well-formatted documentation.
492-539: LGTM! Excellent optimization to avoid N+1 queries.The
get_bulk_skill_effectivenessmethod batches effectiveness lookups and fills in missing entries with defaults, which prevents N+1 query problems when scoring multiple skills. The f-string SQL construction on line 504 is safe because placeholders are dynamically generated based on the count of skill_ids, and actual values are passed as parameterized queries.
555-618: LGTM! Comprehensive skill extraction with excellent provenance tracking.The extraction logic properly gates on success and quality thresholds, requires sufficient iterations, and captures detailed provenance including quality progression over time. This supports auditability and debugging of the learning system.
746-788: LGTM! Efficient retrieval with batch effectiveness lookup.The retrieval logic properly batches effectiveness lookups to avoid N+1 queries, scores skills with multiple relevance factors, and returns scored results for transparency. The separation of search and scoring is clean.
897-957: LGTM! Excellent security and atomicity guarantees.The promotion logic properly prevents path traversal by using
skill_id(a hash) for directory names rather than user-controllednamefields. The atomic promotion pattern (files first, then DB) with comprehensive rollback on failure ensures consistency. Best-effort cleanup of partial artifacts is appropriate.
1014-1040: LGTM! Clean convenience API.The convenience functions provide a user-friendly API with good documentation. Note that line 1038 uses
promoted_only=Falsewith a comment "Include non-promoted for now", suggesting this may change in the future. Consider whether this default aligns with the intended user experience.
168-179: Thread-safety implementation is correct; no changes needed.The use of
threading.local()for connection storage (line 35) combined withcheck_same_thread=Falseis a safe and correct pattern. Each thread is guaranteed an isolated connection through the thread-local namespace, preventing accidental sharing. The WAL mode pragma is appropriate for concurrent read access.
| from __future__ import annotations | ||
|
|
||
| import json | ||
| import sys | ||
| import importlib.util | ||
| from pathlib import Path | ||
| from typing import Any |
There was a problem hiding this comment.
Fix import block sorting to pass CI.
The pipeline is failing due to unsorted imports.
🔎 Proposed fix
from __future__ import annotations
import json
+import importlib.util
import sys
-import importlib.util
from pathlib import Path
from typing import Any📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| from __future__ import annotations | |
| import json | |
| import sys | |
| import importlib.util | |
| from pathlib import Path | |
| from typing import Any | |
| from __future__ import annotations | |
| import json | |
| import importlib.util | |
| import sys | |
| from pathlib import Path | |
| from typing import Any |
🧰 Tools
🪛 GitHub Actions: CI
[error] 19-19: I001 Import block is un-sorted or un-formatted
🪛 GitHub Check: Quality Gate
[failure] 19-25: Ruff (I001)
.claude/skills/sc-implement/scripts/skill_learn.py:19:1: I001 Import block is un-sorted or un-formatted
🤖 Prompt for AI Agents
In .claude/skills/sc-implement/scripts/skill_learn.py around lines 19 to 25, the
import block is not sorted which fails CI; reorder imports into standard-library
first (alphabetical), then third-party, then local application imports (each
group separated by a single blank line), and alphabetize names within each group
(e.g., from __future__ import annotations, then import importlib.util, import
json, import sys, from pathlib import Path, from typing import Any) so the file
conforms to the project's import-sorting rules.
- Sort imports alphabetically in skill_learn.py, skill_persistence.py - Remove unused imports: os, time, IterationResult, QualityAssessment - Remove unused Callable from skill_persistence.py - Remove unused List from types.py 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
| 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() |
There was a problem hiding this comment.
security (python.sqlalchemy.security.sqlalchemy-execute-raw-query): Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
Source: opengrep
🤖 PAL MCP Consensus Code ReviewOverviewThis PR introduces a comprehensive skill persistence layer enabling SuperClaude to learn from successful execution patterns across sessions. The implementation adds SQLite-backed storage (~1,000 LOC in skill_persistence.py), a learning-enabled orchestrator, and a CLI management tool. Key Components:
🔴 Critical IssuesNone identified - The implementation demonstrates strong security awareness and defensive programming practices. 🟠 High Priority1. SQL Injection Prevention in Dynamic Queries (skill_persistence.py:504)
2. Thread Safety Concerns (skill_persistence.py:182-185)
3. Missing Input Validation (skill_learn.py:290-293)
🟡 Medium Priority4. Exception Handling Too Broad (skill_learn.py:358, 425)
5. Missing Database Connection Pooling
6. Quality Threshold Magic Numbers (skill_persistence.py:858-860)
7. Inefficient Pattern Extraction (skill_persistence.py:632-638)
8. Missing Logging Infrastructure
🟢 Positive Observations
📊 Review Summary
💡 RecommendationsBefore Merge:
Post-Merge: ✅ Approval StatusAPPROVED with recommendations for minor improvements before merge. This PR implements a sophisticated learning system with strong security practices and clean architecture. The identified issues are minor and don't block merging. This review was generated using PAL MCP-style consensus analysis. |
There was a problem hiding this comment.
Actionable comments posted: 5
♻️ Duplicate comments (1)
core/skill_learning_integration.py (1)
85-86:_initial_qualityis never set, causing incorrectquality_impact.
_initial_qualityremains0.0throughout the run, soquality_impactin_record_skill_effectiveness()will always equal the final quality score rather than the actual improvement. Capture the initial quality before running the loop.🔎 Proposed fix
def run( self, initial_context: Dict[str, Any], skill_invoker: Callable[[Dict[str, Any]], Dict[str, Any]], ) -> LoopResult: # 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) + # Capture initial quality for effectiveness tracking + initial_assessment = self.assessor.assess(initial_context) + self._initial_quality = initial_assessment.overall_score # Run the standard loop result = super().run(initial_context, skill_invoker)Alternatively, use the first iteration's input quality from the result:
def _record_skill_effectiveness(self, result: LoopResult) -> None: """Record how effective the applied skills were.""" final_quality = result.final_assessment.overall_score - quality_impact = final_quality - self._initial_quality + initial_quality = ( + result.iteration_history[0].input_quality + if result.iteration_history else 0.0 + ) + quality_impact = final_quality - initial_quality
🧹 Nitpick comments (10)
core/skill_persistence.py (4)
62-64:from_dictwill raiseTypeErroron unexpected keys.If the stored JSON has extra fields (e.g., from schema evolution),
cls(**data)will fail. Consider filtering to known fields or using a more defensive approach.🔎 Proposed defensive fix
@classmethod def from_dict(cls, data: dict[str, Any]) -> LearnedSkill: - return cls(**data) + import inspect + valid_fields = {f.name for f in cls.__dataclass_fields__.values()} + filtered = {k: v for k, v in data.items() if k in valid_fields} + return cls(**filtered)Alternatively, if you prefer avoiding the import:
from dataclasses import fields ... valid_fields = {f.name for f in fields(cls)}
346-355: In-memory trigger filtering may not scale.The current approach fetches all skills matching the quality/domain filter and then filters in Python. For large skill sets, consider adding a full-text search index or filtering at the SQL level.
624-628: Unused loop variablei.The
enumerateprovides an indexithat is never used. Use_or removeenumerate.🔎 Proposed fix
- for i, feedback in enumerate(feedback_list): + for feedback in feedback_list:
959-968: Consider using public API instead of private methods.
list_pending()accessesstore._get_connection()andstore._row_to_skill(). Consider adding a public method toSkillStorefor fetching pending skills or using existing public methods likesearch_skills().🔎 Proposed approach
Add a public method to
SkillStore:# In SkillStore class def get_unpromoted_skills(self, min_quality: float = 0.0) -> list[LearnedSkill]: """Get unpromoted skills above a quality threshold.""" conn = self._get_connection() rows = conn.execute(""" SELECT * FROM learned_skills WHERE promoted = 0 AND quality_score >= ? ORDER BY quality_score DESC """, (min_quality,)).fetchall() return [self._row_to_skill(row) for row in rows]Then in
PromotionGate.list_pending():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] + return self.store.get_unpromoted_skills(self.MIN_QUALITY_SCORE - 10).claude/skills/sc-implement/scripts/skill_learn.py (3)
51-65: Use public API instead of raw SQL.The handler accesses
store._get_connection()and duplicates SQL logic that partially exists inSkillStore.search_skills(). Consider using the public API for maintainability.🔎 Proposed approach
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) + # Use public search_skills or add a list method to SkillStore + all_skills = store.search_skills( + query="", # Empty query to get all + domain=domain, + min_quality=min_quality, + promoted_only=promoted_only, + ) + + skills = [] + for skill in all_skills: skills.append({...})Note: This may require adjusting
search_skillsto handle empty queries or adding a dedicatedlist_skillsmethod.
131-173: Stats logic duplicatesget_skill_stats()inskill_learning_integration.py.Consider moving the stats query to
SkillStoreas a public method and calling it from both places to avoid duplication and private method access.
314-329: N+1 query for effectiveness data.Each skill triggers a separate
get_skill_effectiveness()call. Consider usingget_bulk_skill_effectiveness()for better performance.🔎 Proposed fix
skills = [] + skill_ids = [s.skill_id for s in pending] + effectiveness_map = store.get_bulk_skill_effectiveness(skill_ids) for skill in pending: can_promote, reason = gate.evaluate(skill) - effectiveness = store.get_skill_effectiveness(skill.skill_id) + effectiveness = effectiveness_map.get(skill.skill_id, {}) skills.append({ "skill_id": skill.skill_id, ...core/skill_learning_integration.py (3)
236-236: Test results are not captured.The
test_resultsfield is always empty. If test data is available initer_result, consider extracting it for richer skill learning context.
416-454: Consider moving stats query toSkillStore.This function accesses
store._get_connection()and duplicates SQL that also exists inskill_learn.py:handle_stats(). Consolidating inSkillStorewould improve maintainability.
377-377: Accessing private_applied_skillsattribute.Consider adding a public property to
LearningLoopOrchestratorto expose the count of applied skills.
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
.claude/skills/sc-implement/scripts/skill_learn.pycore/skill_learning_integration.pycore/skill_persistence.pycore/types.py
🚧 Files skipped from review as they are similar to previous changes (1)
- core/types.py
🧰 Additional context used
📓 Path-based instructions (1)
**/*.py
📄 CodeRabbit inference engine (AGENTS.md)
**/*.py: Python code targets 3.8+ with Black (88 cols), Flake8, and MyPy; use 4-space indents
Use snake_case for Python module and variable names
Use PascalCase for Python agent classes aligned with their persona names
Files:
core/skill_learning_integration.pycore/skill_persistence.py
🧬 Code graph analysis (1)
.claude/skills/sc-implement/scripts/skill_learn.py (1)
core/skill_persistence.py (14)
LearnedSkill(39-126)PromotionGate(850-968)SkillExtractor(542-733)SkillRetriever(736-847)SkillStore(154-539)_get_connection(168-179)_row_to_skill(357-376)get_skill(296-303)evaluate(866-895)promote(897-957)retrieve(746-788)to_skill_md(66-126)list_pending(959-968)get_skill_effectiveness(468-490)
🪛 GitHub Actions: CI
core/skill_learning_integration.py
[error] 1-1: Ruff formatting would modify this file. Run 'ruff format' to fix formatting.
core/skill_persistence.py
[error] 1-1: Ruff formatting would modify this file. Run 'ruff format' to fix formatting.
.claude/skills/sc-implement/scripts/skill_learn.py
[error] 1-1: Ruff formatting would modify this file. Run 'ruff format' to fix formatting.
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: Sourcery review
- GitHub Check: CodeQL Analysis
- GitHub Check: PAL MCP Consensus Code Review
🔇 Additional comments (7)
core/skill_persistence.py (4)
168-179: Thread-local connection pattern is well-implemented.The use of thread-local storage with
check_same_thread=Falseand WAL mode is appropriate for concurrent access. The connection caching per thread avoids connection overhead while maintaining thread safety.
181-191: LGTM!Context manager correctly delegates to
close()and doesn't suppress exceptions.
773-788: Good use of bulk fetch to avoid N+1 queries.The
get_bulk_skill_effectivenesspattern efficiently fetches all effectiveness data in one query before scoring, avoiding per-skill database calls.
973-975:get_default_store()creates a new instance each call.If this is intentional for isolation, it's fine. If a shared store is desired for connection reuse, consider using a module-level singleton or caching pattern.
.claude/skills/sc-implement/scripts/skill_learn.py (1)
388-434: LGTM!Clean CLI pattern with proper JSON input parsing, command routing, and error handling with informative messages.
core/skill_learning_integration.py (2)
98-135: LGTM!The run method correctly orchestrates the learning flow: domain detection → skill injection → loop execution → feedback recording → skill extraction → effectiveness tracking.
137-179: LGTM!Domain detection uses a reasonable heuristic combining task keywords and file extensions, with sensible fallback to "general".
| 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) |
There was a problem hiding this comment.
Add null check for spec and spec.loader.
spec_from_file_location can return None if the file doesn't exist, and spec.loader can be None. This could cause AttributeError at runtime.
🔎 Proposed fix
spec = importlib.util.spec_from_file_location("skill_persistence", SKILL_PERSISTENCE_PATH)
+if spec is None or spec.loader is None:
+ print(json.dumps({
+ "success": False,
+ "error": f"Could not load skill_persistence from {SKILL_PERSISTENCE_PATH}",
+ }))
+ sys.exit(1)
sp = importlib.util.module_from_spec(spec)
sys.modules["skill_persistence"] = sp
spec.loader.exec_module(sp)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| 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) | |
| spec = importlib.util.spec_from_file_location("skill_persistence", SKILL_PERSISTENCE_PATH) | |
| if spec is None or spec.loader is None: | |
| print(json.dumps({ | |
| "success": False, | |
| "error": f"Could not load skill_persistence from {SKILL_PERSISTENCE_PATH}", | |
| })) | |
| sys.exit(1) | |
| sp = importlib.util.module_from_spec(spec) | |
| sys.modules["skill_persistence"] = sp | |
| spec.loader.exec_module(sp) |
🤖 Prompt for AI Agents
In .claude/skills/sc-implement/scripts/skill_learn.py around lines 31 to 34, add
null checks for spec and spec.loader returned by
importlib.util.spec_from_file_location before using them: if spec is None or
spec.loader is None, raise a clear ImportError or log and exit with an
explanatory message (including the SKILL_PERSISTENCE_PATH), otherwise proceed to
create the module, register it in sys.modules and call
spec.loader.exec_module(sp) as before. Ensure the check prevents AttributeError
by aborting early with a helpful error when the file is missing or loader is
unavailable.
| 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}"} |
There was a problem hiding this comment.
Wrap deletes in a transaction for atomicity.
The two DELETE statements are separate; if the first succeeds but the second fails, the database will be in an inconsistent state.
🔎 Proposed fix
# 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()
+ with conn:
+ conn.execute("DELETE FROM skill_applications WHERE skill_id = ?", (skill_id,))
+ conn.execute("DELETE FROM learned_skills WHERE skill_id = ?", (skill_id,))
except Exception as e:
return {"success": False, "error": f"Database error: {e}"}Using with conn: ensures the operations are wrapped in a transaction that commits on success or rolls back on failure.
🤖 Prompt for AI Agents
In .claude/skills/sc-implement/scripts/skill_learn.py around lines 354-359 the
two DELETEs are executed separately which can leave the DB in an inconsistent
state if one succeeds and the other fails; wrap both DELETE statements in a
single transaction (e.g. use "with conn:" or explicit BEGIN/ROLLBACK/COMMIT) so
they either both succeed or are rolled back on error, remove the explicit
conn.commit() inside the try (or place it after the transactional block), and
keep the existing exception handling to return the database error on failure.
🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
🤖 PAL MCP Consensus Code ReviewOverviewThis PR introduces a comprehensive skill persistence layer for SuperClaude, enabling cross-session learning by extracting, storing, and retrieving successful execution patterns. The implementation includes ~2000 lines across 6 Python modules with SQLite-backed storage, skill extraction/retrieval algorithms, and a CLI management tool. 🔴 Critical Issues1. SQL Injection Vulnerability in SkillRetriever._score_relevance (CRITICAL)Location: core/skill_persistence.py:518-530 The code uses f-string interpolation to build SQL with a dynamic number of placeholders: placeholders = ",".join("?" for _ in skill_ids)
rows = conn.execute(
f"""
SELECT ...
FROM skill_applications
WHERE skill_id IN ({placeholders})
GROUP BY skill_id
""",
skill_ids,
).fetchall()Issue: While the placeholders themselves are safe, this pattern can be dangerous if Recommendation: Add input validation: if not all(isinstance(sid, str) for sid in skill_ids):
raise ValueError("skill_ids must be list of strings")2. Race Condition in PromotionGate.promote (HIGH)Location: core/skill_persistence.py:903-963 The promotion operation has a TOCTOU (Time-of-Check-Time-of-Use) race condition:
Issue: Between steps 1 and 2, another process could read the files and see a promoted skill that isn't in the database. The rollback on failure is good, but doesn't prevent the race window. Recommendation: Use database transactions with proper locking or implement file-based locking before writing. 3. Missing Test Coverage (HIGH PRIORITY)Files Modified: 8 Python files, 0 tests modified The PR adds complex learning logic with no test coverage:
Recommendation: Add comprehensive test suite covering:
🟠 High Priority4. Thread-Local Connection Pattern May Leak ConnectionsLocation: core/skill_persistence.py:168-177 Uses thread-local storage for SQLite connections with a Issue: Long-running threads that create connections but don't explicitly call Recommendation: Implement a context manager or use 5. Hardcoded Paths with Path Traversal RiskLocation: Multiple files
Issue: While Recommendation:
6. No Database Migration StrategyLocation: core/skill_persistence.py:191-253 The schema is created with Issue: Future schema changes will be difficult to deploy without data loss. No way to detect schema version mismatches. Recommendation: Add schema versioning table and migration framework (e.g., alembic or simple manual migrations). 7. Insufficient Error Handling in CLILocation: skill_learn.py:361-366, 449-462 Database errors are caught but don't rollback transactions properly. Generic Recommendation: Use specific exception types, ensure transactions are properly rolled back, and add logging for debugging. 🟡 Medium Priority8. Performance: N+1 Query Pattern (Mitigated)Location: core/skill_persistence.py:783-798 Good work implementing Suggestion: Always use bulk query in retrieval path, make single query a rare exception. 9. Magic Numbers and ThresholdsLocation: Throughout
Issue: Hardcoded thresholds make experimentation difficult. These should be configurable. Recommendation: Move to a 10. Typo in Variable NameLocation: skill_learn.py:28 SUPERCLAUD_ROOT = Path(__file__).parent.parent.parent.parent.parentIssue: Should be 11. SQLite WAL Mode May Cause Issues on Network FilesystemsLocation: core/skill_persistence.py:176 WAL (Write-Ahead Logging) mode improves concurrency but doesn't work reliably on NFS or other network filesystems. Recommendation: Add configuration option to disable WAL for network filesystem deployments. 12. Large String Concatenation in Pattern ExtractionLocation: core/skill_persistence.py:633-653 Uses list concatenation and set operations for deduplication. For large iteration counts, this could be inefficient. Suggestion: Use a more efficient deduplication algorithm or limit pattern extraction to recent iterations. 🟢 Positive Observations
📊 Review Summary
🎯 RecommendationAPPROVE WITH CONDITIONS This is a well-designed feature with solid architecture and security-conscious implementation. However, blocking issues before merge:
Nice-to-have (can be follow-up PRs):
This review was generated by PAL MCP Consensus Code Review. |
Adds comprehensive tests for skill_persistence.py and skill_learning_integration.py to meet 90% coverage requirement. Tests cover: - SkillStore CRUD operations and context management - SkillExtractor session extraction logic - SkillRetriever relevance scoring - PromotionGate evaluation and promotion flow - LearningLoopOrchestrator domain detection - Utility functions (stats, promote, retrieve) Coverage: 90.29% (threshold: 90%) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Code Review - PR #25: Skill Persistence LayerOverviewThis PR adds a sophisticated skill persistence layer for cross-session learning with SQLite storage, pattern extraction, skill retrieval, and quality-based promotion gates. Changes: 10 files, 2799 insertions, 8 deletions Critical Issues1. Path Traversal Risk (HIGH)
2. SQL Injection Pattern (CRITICAL)
3. Race Condition (HIGH)
High Priority
Medium Priority
Positive Observations✅ Excellent modular architecture Summary
Verdict: ✅ Approve with required changes Outstanding architecture. Must fix path traversal and document thread safety before merge. Comprehensive manual code review focusing on security, concurrency, and data integrity |
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (6)
tests/core/test_skill_learning_integration.py (3)
133-137: Simplify the assertion for checking empty learned_skills.The current assertion is verbose. A cleaner approach:
🔎 Proposed simplification
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 + assert not result.get("learned_skills")
152-156: Consider a more specific assertion for the invoker signal structure.The current assertion
"task" in signal or "context" in str(signal).lower()is loose and may mask regressions. If the signal has a defined structure, assert on specific expected keys.
206-254: Consider adding@pytest.mark.integrationmarker.Per the coding guidelines, tests that involve DB operations or broader workflows should be marked with
@pytest.mark.slowor@pytest.mark.integration. Theserun_learning_looptests interact with the SQLite store and orchestrate multiple components.🔎 Proposed addition
+@pytest.mark.integration class TestRunLearningLoop: """Tests for run_learning_loop function."""tests/core/test_skill_persistence.py (3)
217-223: Minor: Variable name convention.The underscore prefix
_storetypically indicates an unused variable, but here the store is used to trigger connection. Consider usingstoredirectly for clarity.🔎 Proposed fix
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: + with SkillStore(db_path) as store: # Trigger connection to create the database - _store._get_connection() + store._get_connection() assert db_path.parent.exists()
332-340: Consider using exact count assertion.Since only one skill is saved before retrieval,
assert len(results) >= 1could beassert len(results) == 1for precision. This catches regressions where duplicate entries might be returned.🔎 Proposed fix
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 + assert len(results) == 1 skill, score = results[0] assert skill.skill_id == sample_skill.skill_id assert score > 0
98-99: Consider adding@pytest.mark.integrationfor DB-backed test classes.Per coding guidelines, tests involving database operations should be marked appropriately. The
TestSkillStore,TestSkillExtractor,TestPromotionGate, and related classes interact heavily with SQLite.🔎 Proposed addition
+@pytest.mark.integration class TestSkillStore: """Tests for SkillStore."""Apply similar markers to other DB-backed test classes as appropriate.
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
tests/core/test_skill_learning_integration.pytests/core/test_skill_persistence.py
🧰 Additional context used
📓 Path-based instructions (2)
**/*.py
📄 CodeRabbit inference engine (AGENTS.md)
**/*.py: Python code targets 3.8+ with Black (88 cols), Flake8, and MyPy; use 4-space indents
Use snake_case for Python module and variable names
Use PascalCase for Python agent classes aligned with their persona names
Files:
tests/core/test_skill_learning_integration.pytests/core/test_skill_persistence.py
tests/**/*.py
📄 CodeRabbit inference engine (AGENTS.md)
tests/**/*.py: Mirror production paths when adding tests (tests//test_.py) and name test functions test_
Mark slower test journeys with @pytest.mark.slow or @pytest.mark.integration per pyproject.toml
Include fixtures that validate requires_evidence guardrails and .superclaude_metrics outputs whenever agent workflows, telemetry, or auto-implementation logic changes
Files:
tests/core/test_skill_learning_integration.pytests/core/test_skill_persistence.py
🧬 Code graph analysis (1)
tests/core/test_skill_learning_integration.py (1)
core/skill_persistence.py (3)
LearnedSkill(39-126)SkillStore(154-555)save_skill(257-295)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: Sourcery review
- GitHub Check: CodeQL Analysis
- GitHub Check: PAL MCP Consensus Code Review
🔇 Additional comments (8)
tests/core/test_skill_learning_integration.py (2)
1-17: LGTM! Clean imports and module structure.The imports are well-organized, and the module docstring clearly states the purpose.
21-62: Fixtures are well-structured with proper cleanup.Good use of
tmp_pathfor test isolation and properstore.close()in teardown. The fixtures provide appropriate test data for the learning integration tests.tests/core/test_skill_persistence.py (6)
1-17: LGTM! Comprehensive imports covering all persistence layer components.The imports correctly include all major classes and convenience functions that need testing.
22-92: Well-designed fixtures with appropriate test data.Fixtures provide good coverage of different skill states (regular vs. high-quality) and proper cleanup. The
sample_feedbackfixture nicely exercises theIterationFeedbackstructure.
185-200: Good coverage of effectiveness metrics calculation.The test verifies helpful/unhelpful counts and application tracking correctly. The assertions validate the core business logic of skill effectiveness evaluation.
282-317: Good multi-iteration extraction test.The test correctly sets up a session with multiple feedback iterations that meet the quality improvement criteria. The assertions verify the extracted skill has proper domain and quality score.
397-408: Good end-to-end promotion test.The test validates the full promotion flow: saving skill, recording applications, and verifying the SKILL.md file is created. This covers the critical path for skill persistence.
422-453: LGTM! Convenience functions are well-tested.Tests properly isolate the default store path via monkeypatch and verify expected behavior for empty/missing data scenarios.
The _initial_quality instance variable was never set, causing the skill effectiveness calculation to always use 0.0. Now correctly gets initial quality from the first iteration's input_quality. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
🤖 Comprehensive Code ReviewOverviewThis PR introduces a skill persistence layer for cross-session learning in SuperClaude. The implementation includes:
Files reviewed: 8 Python files (5 core modules, 2 test files, 1 CLI script) 🔴 Critical Issues1. SQL Injection Vulnerability in Bulk Query (CRITICAL - Security)Location: The placeholders = ",".join("?" for _ in skill_ids)
rows = conn.execute(
f"""
SELECT ...
WHERE skill_id IN ({placeholders})
""",
skill_ids,
)While this uses parameterized placeholders, it's vulnerable if Recommendation: Add input validation: if not isinstance(skill_ids, list):
raise ValueError("skill_ids must be a list")
for sid in skill_ids:
if not isinstance(sid, str) or not sid.startswith("learned-"):
raise ValueError(f"Invalid skill_id format: {sid}")2. Path Traversal Risk in CLI Export (HIGH - Security)Location: The export handler uses output_path = Path(output_dir) / skill.skill_id / "SKILL.md"While the code comment mentions "prevent path traversal", there's no actual validation. A malicious skill_id like Recommendation: Add explicit path validation: # Validate skill_id is safe
if ".." in skill.skill_id or "/" in skill.skill_id:
return {"success": False, "error": "Invalid skill_id"}3. Race Condition in Promotion Atomicity (MEDIUM - Reliability)Location: The promotion process writes files first, then updates the database: 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))
if not self.store.save_skill(skill):
raise IOError("Database save failed")If the database save fails, the rollback may not clean up files completely (race condition if another process reads the directory). Consider using a transaction or temporary directory approach. 🟠 High Priority4. Missing Database Connection PoolingLocation: The code uses thread-local storage for SQLite connections but doesn't implement proper connection pooling. Multiple threads could exhaust resources or create contention. Recommendation: Add connection limit or use a proper connection pool library. 5. Quality Score Calculation Missing Initial ValueLocation: In initial_quality = (
result.iteration_history[0].input_quality
if result.iteration_history
else 0.0 # ⚠️ This could be wrong
)Recommendation: Store 6. Unbounded Memory in Feedback StorageLocation: The system continuously appends to Recommendation: Implement a retention policy or archival system. 7. No Input Sanitization in CLILocation: JSON parsing errors are caught, but the error messages could leak sensitive information: except json.JSONDecodeError as e:
print(json.dumps({"error": f"Invalid JSON: {e}"}))Recommendation: Sanitize error messages to avoid information disclosure. 🟡 Medium Priority8. Type Safety IssuesMultiple locations use bare
Recommendation: Define TypedDict schemas for better type safety. 9. Magic Numbers Scattered Throughout Code
Recommendation: Extract to configuration class or constants file. 10. Inconsistent Error HandlingSome methods return Recommendation: Standardize on raising exceptions for errors and using return types for success cases. 11. Missing LoggingThe code uses Recommendation: Use Python's 12. WAL Mode Not VerifiedLocation: The code executes Recommendation: Check the pragma result and fall back gracefully. 🟢 Positive Observations
📊 Review Summary
🎯 RecommendationsBefore Merge (Critical):
Post-Merge (High Priority): Future Enhancements: 📝 Additional NotesThis is a well-architected feature with clear value for cross-session learning. The code quality is high, with excellent documentation and test coverage. The main concerns are around security hardening and operational robustness (unbounded growth, connection management). The PR demonstrates strong software engineering practices:
Recommendation: ✅ Approve with conditions - Address the 3 critical security issues before merge, and create follow-up issues for the high-priority items. This review was performed by manual code analysis focusing on security, architecture, performance, and maintainability. |
🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
🤖 Comprehensive Code Review - PR #25OverviewThis PR adds a skill persistence layer for cross-session learning in SuperClaude. The implementation includes:
Total Changes: 10 files, +2803 lines, -8 lines 🟢 Strengths & Positive ObservationsSecurity
Code Quality
Testing
🟡 Medium Priority Recommendations1. Database Connection Resource LeaksLocation: skill_persistence.py:168-177, skill_learn.py:51-65, 136 Issue: CLI commands and some operations don't explicitly close database connections. Example: # skill_learn.py:51
conn = store._get_connection() # Never closed
rows = conn.execute(sql, params).fetchall()Impact: Thread-local connections accumulate, consuming resources. Recommendation: # Use context manager pattern
def handle_list(args: dict[str, Any]) -> dict[str, Any]:
with SkillStore() as store:
conn = store._get_connection()
# ... operations
# Connection auto-closed2. Performance: N+1 Query Pattern in RetrievalLocation: skill_persistence.py:786-793 Issue: While bulk effectiveness fetch exists (line 505), retrieval could benefit from batch operations. Current: # Gets effectiveness one-by-one in scoring
for skill in candidates:
effectiveness = self.store.get_skill_effectiveness(skill.skill_id)Recommendation: Already implemented bulk fetch at line 784-785! Just needs to be used consistently. Status: ✅ Already resolved in code (effectiveness_map pattern) 3. Missing Index for PerformanceLocation: skill_persistence.py:244-251 Issue: Recommendation: CREATE INDEX IF NOT EXISTS idx_applications_session
ON skill_applications(session_id);4. Quality Impact Calculation LogicLocation: skill_learning_integration.py:265-272 Issue: Current: initial_quality = result.iteration_history[0].input_quality if result.iteration_history else 0.0Problem: If quality assessment runs after first changes, Recommendation: # Store baseline quality before loop starts
def run(self, initial_context, skill_invoker):
self._initial_quality = self.assessor.assess(initial_context).overall_score
# ... rest of loop5. Potential Race Condition in File CleanupLocation: skill_learn.py:368-375 Issue: Recommendation: try:
shutil.rmtree(skill_dir)
except FileNotFoundError:
pass # Already deleted, no problem6. Stopwords Too RestrictiveLocation: skill_persistence.py:697-698 Issue: Only 9 common words filtered. "code", "file", "make", etc. could dilute trigger quality. Recommendation: stopwords = {
"the", "and", "for", "with", "from", "this", "that", "have", "been",
"code", "file", "make", "add", "update", "fix", "change", "create"
}🟢 Low Priority Observations1. Hardcoded Path in CLILocation: skill_learn.py:28-29 Issue: Uses 5-level parent navigation which is fragile. Recommendation: Use 2. Magic Numbers in Skill GenerationLocation: skill_persistence.py:653, 670, 700 Issue: Hardcoded limits (10 patterns, 5 anti-patterns, 15 triggers) lack justification. Recommendation: Extract to class constants with docstring explaining rationale. 3. Incomplete Test Results in FeedbackLocation: skill_learning_integration.py:236 Issue: Comment says "Would need to extract from output" but field left empty. Recommendation: Either extract test results from skill invoker output or document why it's acceptable to omit. 🔵 Architecture & Design Observations1. Well-Designed Promotion AtomicityLocation: skill_persistence.py:903-963 Excellent: Files written first, then DB updated. On failure, files cleaned up and DB state rolled back. This prevents partial promotion states. 2. Smart Bulk Effectiveness PatternLocation: skill_persistence.py:505-555 Excellent: Avoids N+1 queries by fetching all skill effectiveness in one query, then filling in missing entries with defaults. This is a performance best practice. 3. Thread-Safety via Thread-Local StorageLocation: skill_persistence.py:34-35, 168-177 Good: Uses thread-local connections for SQLite. WAL mode enables concurrent reads. Proper pattern for multi-threaded environments. 4. Provenance TrackingLocation: skill_persistence.py:617-626 Excellent: Full audit trail of skill origins, quality progression, and source context. Critical for understanding learned behaviors. 📊 Review Summary
Overall Assessment: ⭐⭐⭐⭐⭐ (4.8/5) ✅ RecommendationAPPROVE with minor suggestions This is high-quality, production-ready code with strong security practices. The issues identified are minor and non-blocking:
The architecture is sound, security is excellent, and the test coverage is comprehensive. The skill persistence layer provides a solid foundation for cross-session learning. 🔍 Files ReviewedCore Implementation (1,047 lines):
Tooling (466 lines):
Tests (454 lines):
Documentation:
This review was performed through comprehensive manual analysis of code security, architecture, performance, and test coverage. |
Summary
Key Components
core/skill_persistence.pycore/skill_learning_integration.py.claude/skills/sc-implement/scripts/skill_learn.py.claude/skills/learned/SKILL.mdFeatures
Safety Measures
Test plan
from core.skill_persistence import SkillStorepython skill_learn.py '{"command": "stats"}'--loopflag on real task🤖 Generated with Claude Code
Summary by Sourcery
Introduce a SQLite-backed skill persistence layer and integrate it with the loop orchestrator to enable cross-session learning, retrieval, and management of learned skills.
New Features:
Enhancements:
Documentation:
Summary by CodeRabbit
✏️ Tip: You can customize this high-level summary in your review settings.