diff --git a/backend/app/mcp/mcp_bridge/index.ts b/backend/app/mcp/mcp_bridge/index.ts index 13d473d1..3d213209 100644 --- a/backend/app/mcp/mcp_bridge/index.ts +++ b/backend/app/mcp/mcp_bridge/index.ts @@ -57,7 +57,7 @@ const TOOLS = [ { name: "ucore_list_skills", description: - "List all registered uCore skills (currently 54). Returns id, name, description, category, and parameters for each skill.", + "List governed internal uCore capabilities. Returns id, name, description, category, and parameters.", inputSchema: { type: "object", properties: { @@ -72,14 +72,13 @@ const TOOLS = [ { name: "ucore_run_skill", description: - "Execute a named uCore skill by its ID. Skills include: ecosystem-audit, file_edit_enhancer, surface-registry, mcp_self_heal, dev-mode-executor, vault_discovery, and 48 more.", + "Execute a governed internal uCore capability by its ID.", inputSchema: { type: "object", properties: { skill_id: { type: "string", - description: - "The skill ID to run (e.g. 'ecosystem-audit', 'surface-registry', 'file_edit_enhancer')", + description: "The capability ID returned by ucore_list_skills", }, params: { type: "object", @@ -89,16 +88,6 @@ const TOOLS = [ required: ["skill_id"], }, }, - { - name: "ucore_surface_registry", - description: - "List all registered uCore surfaces (Groovebox, Developer, Server, System, Workflow, etc.) with their status, port, and health.", - inputSchema: { - type: "object", - properties: {}, - required: [], - }, - }, { name: "ucore_ollama_status", description: @@ -314,18 +303,6 @@ server.setRequestHandler(CallToolRequestSchema, async (request) => { }; } - case "ucore_surface_registry": { - const data = await apiPost("/api/skills/surface-registry/run", {}); - return { - content: [ - { - type: "text", - text: JSON.stringify(data, null, 2), - }, - ], - }; - } - case "ucore_ollama_status": { const data = await apiGet("/api/ollama/status"); const models = await apiGet("/api/ollama/models/available"); diff --git a/backend/app/skills/builtin/file_edit_enhancer.py b/backend/app/skills/builtin/file_edit_enhancer.py deleted file mode 100644 index 7bc305bc..00000000 --- a/backend/app/skills/builtin/file_edit_enhancer.py +++ /dev/null @@ -1,240 +0,0 @@ -"""file_edit_enhancer — Enhanced file editing with MCP integration. - -Provides intelligent file editing capabilities with: -- Multi-file batch operations -- Context-aware replacements -- Spool logging of edits -- Tasker integration for edit tracking - -Usage: - POST /api/skills/file_edit_enhancer/run - Body: { - "action": "batch_replace", - "edits": [ - {"file": "path/to/file.py", "old": "old text", "new": "new text"}, - ... - ], - "log_to_spool": true - } -""" -from __future__ import annotations - -import json -from datetime import UTC, datetime -from pathlib import Path -from typing import Any - -from app.core.settings import settings -from app.services.spool_writer import write_spool -from app.skills.base import BaseSkill, SkillMeta, SkillParam - -PROJECT_ROOT = settings.udos_root / "uCore" - - -class FileEditEnhancer(BaseSkill): - meta = SkillMeta( - id="file_edit_enhancer", - name="File Edit Enhancer", - description="Enhanced file editing with MCP integration, batch operations, and spool logging", - category="maintenance", - timeout=120, - params=[ - SkillParam( - name="action", - type="string", - required=True, - description="Action: batch_replace, smart_replace, or validate", - ), - SkillParam( - name="edits", - type="array", - required=False, - description="List of {file, old, new} edits for batch_replace", - ), - SkillParam( - name="file", - type="string", - required=False, - description="Single file path for smart_replace/validate", - ), - SkillParam( - name="log_to_spool", - type="boolean", - required=False, - default=True, - description="Log edits to spool for audit trail", - ), - SkillParam( - name="dry_run", - type="boolean", - required=False, - default=False, - description="Preview without writing changes", - ), - ], - requires_confirmation=False, - ) - - async def run(self, **kwargs) -> dict: - action = str(kwargs.get("action", "batch_replace")).strip().lower() - log_to_spool = bool(kwargs.get("log_to_spool", True)) - dry_run = bool(kwargs.get("dry_run", False)) - - if action == "batch_replace": - edits = kwargs.get("edits", []) - return await self._batch_replace(edits, log_to_spool, dry_run) - elif action == "smart_replace": - file_path = kwargs.get("file", "") - old_text = kwargs.get("old", "") - new_text = kwargs.get("new", "") - return await self._smart_replace(file_path, old_text, new_text, log_to_spool, dry_run) - elif action == "validate": - file_path = kwargs.get("file", "") - return self._validate(file_path) - else: - return {"success": False, "error": f"Unknown action: {action}"} - - async def _batch_replace( - self, - edits: list[dict[str, str]], - log_to_spool: bool, - dry_run: bool, - ) -> dict: - """Apply multiple replacements across files efficiently.""" - results = [] - errors = [] - - for edit in edits: - file_path = edit.get("file", "") - old_text = edit.get("old", "") - new_text = edit.get("new", "") - - if not file_path or not old_text: - errors.append(f"Missing file or old_text in edit: {edit}") - continue - - try: - path = Path(file_path).expanduser() - if not path.exists(): - errors.append(f"File not found: {file_path}") - continue - - content = path.read_text(encoding="utf-8") - if old_text not in content: - errors.append(f"Old text not found in {file_path}") - continue - - new_content = content.replace(old_text, new_text) - - if not dry_run: - path.write_text(new_content, encoding="utf-8") - - results.append({ - "file": file_path, - "replaced": True, - "dry_run": dry_run, - }) - - if log_to_spool and not dry_run: - write_spool( - level="INFO", - module="file_edit_enhancer", - message=f"Batch replaced in {file_path}", - ) - - except Exception as exc: - errors.append(f"Error editing {file_path}: {exc}") - - return { - "success": len(errors) == 0, - "action": "batch_replace", - "results": results, - "errors": errors, - "dry_run": dry_run, - } - - async def _smart_replace( - self, - file_path: str, - old_text: str, - new_text: str, - log_to_spool: bool, - dry_run: bool, - ) -> dict: - """Smart replacement with context validation.""" - if not file_path or not old_text: - return {"success": False, "error": "Missing file or old_text"} - - try: - path = Path(file_path).expanduser() - if not path.exists(): - return {"success": False, "error": f"File not found: {file_path}"} - - content = path.read_text(encoding="utf-8") - - # Validate context exists - if old_text not in content: - return {"success": False, "error": f"Old text not found in {file_path}"} - - # Count occurrences - occurrences = content.count(old_text) - - new_content = content.replace(old_text, new_text) - - if not dry_run: - path.write_text(new_content, encoding="utf-8") - - if log_to_spool and not dry_run: - write_spool( - level="INFO", - module="file_edit_enhancer", - message=f"Smart replaced {occurrences} occurrence(s) in {file_path}", - ) - - return { - "success": True, - "action": "smart_replace", - "file": file_path, - "occurrences": occurrences, - "dry_run": dry_run, - } - - except Exception as exc: - return {"success": False, "error": str(exc)} - - def _validate(self, file_path: str) -> dict: - """Validate file for editing issues.""" - if not file_path: - return {"success": False, "error": "Missing file path"} - - try: - path = Path(file_path).expanduser() - if not path.exists(): - return {"success": False, "error": f"File not found: {file_path}"} - - content = path.read_text(encoding="utf-8") - lines = content.split("\n") - - issues = [] - - # Check for common issues - for i, line in enumerate(lines, 1): - # Check for trailing whitespace - if line.rstrip() != line and line.strip(): - issues.append({"line": i, "type": "trailing_whitespace"}) - - # Check for tabs - if "\t" in line: - issues.append({"line": i, "type": "tabs"}) - - return { - "success": True, - "action": "validate", - "file": file_path, - "lines": len(lines), - "issues": issues, - "valid": len(issues) == 0, - } - - except Exception as exc: - return {"success": False, "error": str(exc)} diff --git a/backend/app/skills/builtin/skill_dead_code_archiver.py b/backend/app/skills/builtin/skill_dead_code_archiver.py deleted file mode 100644 index 2841b339..00000000 --- a/backend/app/skills/builtin/skill_dead_code_archiver.py +++ /dev/null @@ -1,389 +0,0 @@ -#!/usr/bin/env python3 -"""Dead Code / Legacy Archiver Skill — identify, archive, and document unused code. - -Scans source files to detect: -- Unused functions and methods -- Orphaned imports -- Dead code paths (unreachable code) -- Legacy code patterns (deprecated APIs, old patterns) -- Files with no references - -Archives findings to: -- tmp/dead_code_archive/ with timestamped reports -- Generates migration notes for legacy code -""" -from __future__ import annotations - -import ast -import json -import logging -import re -import time -from collections import defaultdict -from pathlib import Path -from typing import Any - -from app.skills.base import BaseSkill, SkillMeta, SkillParam -from app.skills.shared_utils import get_source_files as _get_source_files - -log = logging.getLogger("ucore.skills.dead_code_archiver") - -# Legacy patterns to detect -LEGACY_PATTERNS = { - "deprecated_imports": [ - (r"from\s+typing_extensions\s+import", "Use typing instead of typing_extensions"), - (r"import\s+requests", "Consider using httpx for async HTTP"), - (r"\.format\(", "Consider using f-strings instead of .format()"), - ], - "old_patterns": [ - (r"@asyncio\.coroutine", "Deprecated asyncio.coroutine decorator — use async def"), - (r"yield from", "Use async/await instead of yield from"), - (r"super\(.*\)\.__init__\(\)", "Use super().__init__() instead"), - ], -} - - -class DeadCodeArchiverSkill(BaseSkill): - meta = SkillMeta( - id="dead-code-archiver", - name="Dead Code / Legacy Archiver", - description=( - "Identify unused code, legacy patterns, and archive findings. " - "Generates migration notes and removal recommendations." - ), - category="developer", - params=[ - SkillParam( - name="action", - type="string", - description="Action: 'scan', 'archive', 'report'", - required=True, - ), - SkillParam( - name="target", - type="string", - description="Optional: specific directory to scan", - required=False, - ), - SkillParam( - name="include_legacy", - type="bool", - description="Include legacy pattern detection", - required=False, - default=True, - ), - SkillParam( - name="dry_run", - type="bool", - description="Preview without archiving", - required=False, - default=True, - ), - ], - timeout=180, - ) - - async def run(self, **kwargs) -> dict: - action = kwargs.get("action", "scan") - target = kwargs.get("target", "") - include_legacy = kwargs.get("include_legacy", True) - dry_run = kwargs.get("dry_run", True) - - if action == "scan": - return self._scan_dead_code(target, include_legacy) - if action == "report": - scan_result = self._scan_dead_code(target, include_legacy) - return { - "success": True, - "action": "report", - "report": scan_result, - "recommendations": self._generate_recommendations(scan_result), - } - if action == "archive": - scan_result = self._scan_dead_code(target, include_legacy) - return self._archive_findings(scan_result, dry_run) - return {"success": False, "error": f"Unknown action: {action}"} - - def _scan_dead_code(self, target: str, include_legacy: bool) -> dict[str, Any]: - """Scan for dead code and legacy patterns.""" - findings: dict[str, list] = { - "unused_functions": [], - "orphaned_imports": [], - "dead_code_paths": [], - "legacy_patterns": [], - "unreferenced_files": [], - } - stats = { - "files_scanned": 0, - "total_issues": 0, - "languages": defaultdict(int), - } - - source_files = self._get_source_files(target) - stats["files_scanned"] = len(source_files) - - # Build reference map - all_names: dict[str, list] = defaultdict(list) - for file_path in source_files: - try: - content = file_path.read_text() - if file_path.suffix == ".py": - names = self._extract_python_names(content) - for name in names: - all_names[name].append(str(file_path)) - stats["languages"]["python"] += 1 - except Exception: - continue - - # Check for unused functions - for file_path in source_files: - if file_path.suffix != ".py": - continue - try: - content = file_path.read_text() - tree = ast.parse(content) - - for node in ast.walk(tree): - if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): - func_name = node.name - # Check if function is used elsewhere - if len(all_names.get(func_name, [])) == 1: - # Only defined in this file, check if called - if not self._is_function_called(content, func_name): - findings["unused_functions"].append({ - "file": str(file_path), - "function": func_name, - "line": node.lineno, - "lines": len(node.body), - }) - except (SyntaxError, ValueError): - continue - - # Check for orphaned imports - findings["orphaned_imports"] = self._find_orphaned_imports(source_files) - - # Check for dead code paths - findings["dead_code_paths"] = self._find_dead_code_paths(source_files) - - # Check for legacy patterns - if include_legacy: - findings["legacy_patterns"] = self._find_legacy_patterns(source_files) - - # Check for unreferenced files - findings["unreferenced_files"] = self._find_unreferenced_files(source_files) - - stats["total_issues"] = sum(len(v) for v in findings.values()) - - return { - "success": True, - "findings": findings, - "stats": stats, - } - - def _get_source_files(self, target: str) -> list[Path]: - """Get all source files to scan.""" - return _get_source_files(target) - - def _extract_python_names(self, content: str) -> list[str]: - """Extract all defined names from Python content.""" - names = [] - try: - tree = ast.parse(content) - for node in ast.walk(tree): - if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): - names.append(node.name) - elif isinstance(node, ast.ClassDef): - names.append(node.name) - elif isinstance(node, ast.Assign): - for target in node.targets: - if isinstance(target, ast.Name): - names.append(target.id) - except (SyntaxError, ValueError): - pass - return names - - def _is_function_called(self, content: str, func_name: str) -> bool: - """Check if a function is called within its file.""" - # Simple regex check for function calls - call_pattern = re.compile(rf"\b{re.escape(func_name)}\s*\(") - return bool(call_pattern.search(content)) - - def _find_orphaned_imports(self, files: list[Path]) -> list[dict]: - """Find imports that are never used.""" - orphaned = [] - - for file_path in files: - if file_path.suffix != ".py": - continue - try: - content = file_path.read_text() - tree = ast.parse(content) - - for node in ast.walk(tree): - if isinstance(node, ast.Import): - for alias in node.names: - name = alias.asname or alias.name - if not self._is_name_used(content, name): - orphaned.append({ - "file": str(file_path), - "import": alias.name, - "line": node.lineno, - "type": "import", - }) - elif isinstance(node, ast.ImportFrom): - if node.module: - for alias in node.names: - name = alias.asname or alias.name - if not self._is_name_used(content, name): - orphaned.append({ - "file": str(file_path), - "import": f"{node.module}.{alias.name}", - "line": node.lineno, - "type": "from_import", - }) - except (SyntaxError, ValueError): - continue - - return orphaned - - def _is_name_used(self, content: str, name: str) -> bool: - """Check if a name is used in the content (excluding the import line).""" - # Remove import lines and check for usage - lines = content.splitlines() - code_lines = [ln for ln in lines if not ln.strip().startswith(("import ", "from "))] - code = "\n".join(code_lines) - return bool(re.search(rf"\b{re.escape(name)}\b", code)) - - def _find_dead_code_paths(self, files: list[Path]) -> list[dict]: - """Find unreachable code after return/break/continue.""" - dead_paths = [] - - for file_path in files: - if file_path.suffix != ".py": - continue - try: - content = file_path.read_text() - lines = content.splitlines() - - for i, line in enumerate(lines): - stripped = line.strip() - if stripped in ("return", "return None", "return 0", "return ''"): - # Check if next non-empty line has less indentation - j = i + 1 - while j < len(lines) and not lines[j].strip(): - j += 1 - if j < len(lines): - current_indent = len(line) - len(line.lstrip()) - next_indent = len(lines[j]) - len(lines[j].lstrip()) - if next_indent > current_indent: - dead_paths.append({ - "file": str(file_path), - "line": j + 1, - "issue": "Code after return statement", - "code": lines[j].strip()[:80], - }) - except Exception: - continue - - return dead_paths - - def _find_legacy_patterns(self, files: list[Path]) -> list[dict]: - """Find legacy code patterns.""" - legacy = [] - - for file_path in files: - try: - content = file_path.read_text() - for category, patterns in LEGACY_PATTERNS.items(): - for pattern, message in patterns: - matches = re.findall(pattern, content) - if matches: - legacy.append({ - "file": str(file_path), - "category": category, - "pattern": pattern, - "message": message, - "count": len(matches), - }) - except Exception: - continue - - return legacy - - def _find_unreferenced_files(self, files: list[Path]) -> list[dict]: - """Find files that are not imported or referenced.""" - unreferenced = [] - all_content = "" - - for file_path in files: - try: - all_content += file_path.read_text() + "\n" - except Exception: - continue - - for file_path in files: - if file_path.suffix != ".py": - continue - try: - stem = file_path.stem - # Check if module is imported anywhere - import_pattern = re.compile(rf"(?:import|from)\s+{re.escape(stem)}\b") - if not import_pattern.search(all_content): - # Check if it's a main file or __init__ - if stem not in ("__init__", "main", "app"): - unreferenced.append({ - "file": str(file_path), - "module": stem, - }) - except Exception: - continue - - return unreferenced - - def _generate_recommendations(self, scan_result: dict) -> list[str]: - """Generate recommendations based on scan results.""" - recs = [] - findings = scan_result.get("findings", {}) - - unused = findings.get("unused_functions", []) - if unused: - recs.append(f"Found {len(unused)} unused functions — consider removing or documenting") - - orphaned = findings.get("orphaned_imports", []) - if orphaned: - recs.append(f"Found {len(orphaned)} orphaned imports — clean up imports") - - dead = findings.get("dead_code_paths", []) - if dead: - recs.append(f"Found {len(dead)} dead code paths — review control flow") - - legacy = findings.get("legacy_patterns", []) - if legacy: - recs.append(f"Found {len(legacy)} legacy patterns — consider modernization") - - return recs - - def _archive_findings(self, scan_result: dict, dry_run: bool) -> dict: - """Archive findings to timestamped report file.""" - archive_dir = Path(__file__).parent.parent.parent.parent / "tmp" / "dead_code_archive" - archive_dir.mkdir(parents=True, exist_ok=True) - - timestamp = int(time.time()) - report_file = archive_dir / f"dead_code_{timestamp}.json" - - if dry_run: - return { - "success": True, - "action": "archive", - "dry_run": True, - "would_write": str(report_file), - "findings_count": scan_result.get("stats", {}).get("total_issues", 0), - } - - report_file.write_text(json.dumps(scan_result, indent=2)) - return { - "success": True, - "action": "archive", - "report_file": str(report_file), - "findings_count": scan_result.get("stats", {}).get("total_issues", 0), - } diff --git a/backend/app/skills/builtin/skill_duplicate_detector.py b/backend/app/skills/builtin/skill_duplicate_detector.py deleted file mode 100644 index e759828b..00000000 --- a/backend/app/skills/builtin/skill_duplicate_detector.py +++ /dev/null @@ -1,810 +0,0 @@ -#!/usr/bin/env python3 -"""Duplicate Code Detector Skill — find and analyze duplicate code patterns. - -Scans source files to detect: -- Exact duplicate functions/methods across files -- Similar functions/overlap (AST-based similarity scoring) -- Duplicate import patterns -- Repeated string literals -- Copy-pasted code blocks -- Variable consolidation/sharing/mapping ($VARIABLES) -- Script duplication across snacks, scripts, and docs - -Reports: file pairs, similarity scores, line ranges, and removal recommendations. -""" -# ruff: noqa: E501 -from __future__ import annotations - -import ast -import hashlib -import json -import logging -import re -import time -from collections import defaultdict -from difflib import SequenceMatcher -from pathlib import Path -from typing import Any - -from app.skills.base import BaseSkill, SkillMeta, SkillParam -from app.skills.shared_utils import get_source_files as _get_source_files - -log = logging.getLogger("ucore.skills.duplicate_detector") - -# Thresholds -SIMILARITY_THRESHOLD = 0.85 # 85% similarity for code blocks -MIN_LINES_FOR_DUPLICATE = 5 # Minimum lines to consider for duplication -MAX_LINE_LENGTH = 200 # Ignore very long lines (likely generated) -MIN_SIMILARITY_FOR_FUNCTION = 0.7 # 70% similarity for function overlap - - -class DuplicateDetectorSkill(BaseSkill): - meta = SkillMeta( - id="duplicate-detector", - name="Duplicate Code Detector", - description=( - "Find duplicate code patterns across Python/JS/TS files. " - "Reports similarity scores and removal recommendations. " - "Also detects $VARIABLE consolidation and script duplication." - ), - category="developer", - params=[ - SkillParam( - name="action", - type="string", - description="Action: 'scan', 'report', 'archive', 'variables', 'scripts'", - required=True, - ), - SkillParam( - name="target", - type="string", - description="Optional: specific directory or file pattern", - required=False, - ), - SkillParam( - name="min_lines", - type="int", - description="Minimum lines for duplicate detection", - required=False, - default=5, - ), - SkillParam( - name="similarity_threshold", - type="float", - description="Similarity threshold (0.0-1.0)", - required=False, - default=0.85, - ), - SkillParam( - name="dry_run", - type="bool", - description="Preview without archiving", - required=False, - default=True, - ), - ], - timeout=180, - ) - - async def run(self, **kwargs) -> dict: - action = kwargs.get("action", "scan") - target = kwargs.get("target", "") - min_lines = kwargs.get("min_lines", MIN_LINES_FOR_DUPLICATE) - threshold = kwargs.get("similarity_threshold", SIMILARITY_THRESHOLD) - dry_run = kwargs.get("dry_run", True) - - if action == "scan": - return self._scan_duplicates(target, min_lines, threshold) - if action == "report": - scan_result = self._scan_duplicates(target, min_lines, threshold) - return { - "success": True, - "action": "report", - "report": scan_result, - "recommendations": self._generate_recommendations(scan_result), - } - if action == "archive": - scan_result = self._scan_duplicates(target, min_lines, threshold) - return self._archive_duplicates(scan_result, dry_run) - if action == "variables": - source_files = self._get_source_files(target) - return { - "success": True, - "action": "variables", - "findings": self._find_variable_consolidation(source_files), - } - if action == "scripts": - return { - "success": True, - "action": "scripts", - "findings": self._find_script_duplication(target), - } - return {"success": False, "error": f"Unknown action: {action}"} - - def _scan_duplicates( - self, target: str, min_lines: int, threshold: float - ) -> dict[str, Any]: - """Scan for duplicate code patterns.""" - findings: dict[str, list] = { - "exact_duplicates": [], - "similar_blocks": [], - "similar_functions": [], - "duplicate_imports": [], - "repeated_literals": [], - "copy_paste_patterns": [], - "variable_consolidation": [], - "script_duplication": [], - } - stats = { - "files_scanned": 0, - "total_issues": 0, - "languages": defaultdict(int), - } - - # Get source files - source_files = self._get_source_files(target) - - # Group by language - py_files = [f for f in source_files if f.suffix == ".py"] - js_ts_files = [f for f in source_files if f.suffix in {".js", ".ts", ".jsx", ".tsx"}] - - # Scan Python files for function/method duplicates - py_findings = self._scan_python_duplicates(py_files, min_lines, threshold) - findings["exact_duplicates"].extend(py_findings["exact"]) - findings["similar_blocks"].extend(py_findings["similar"]) - stats["languages"]["python"] = len(py_files) - - # Scan JS/TS files for pattern duplicates - js_findings = self._scan_js_duplicates(js_ts_files, min_lines, threshold) - findings["exact_duplicates"].extend(js_findings["exact"]) - findings["similar_blocks"].extend(js_findings["similar"]) - stats["languages"]["javascript"] = len(js_ts_files) - - # Find similar functions (overlapping logic) - findings["similar_functions"] = self._find_similar_functions(py_files + js_ts_files, MIN_SIMILARITY_FOR_FUNCTION) - - # Find duplicate import patterns - findings["duplicate_imports"] = self._find_duplicate_imports(source_files) - - # Find repeated string literals - findings["repeated_literals"] = self._find_repeated_literals(source_files) - - # Find variable consolidation opportunities - findings["variable_consolidation"] = self._find_variable_consolidation(source_files) - - # Find script duplication across snacks/scripts/docs - findings["script_duplication"] = self._find_script_duplication(target) - - # Find large files that need modularization (>1000 lines) - findings["large_files"] = self._find_large_files(source_files) - - # Find duplicate functions within the same file - findings["duplicate_functions_in_file"] = self._find_duplicate_functions_in_file(py_files) - - # Find circular imports - findings["circular_imports"] = self._find_circular_imports(source_files) - - # Calculate stats - stats["files_scanned"] = len(source_files) - stats["total_issues"] = sum(len(v) for v in findings.values()) - - return { - "success": True, - "findings": findings, - "stats": stats, - } - - def _get_source_files(self, target: str) -> list[Path]: - """Get all source files to scan.""" - return _get_source_files(target) - - def _scan_python_duplicates( - self, files: list[Path], min_lines: int, threshold: float - ) -> dict[str, list]: - """Scan Python files for duplicate functions and code blocks.""" - exact_duplicates = [] - similar_blocks = [] - - # Hash-based exact duplicate detection - hash_to_files: dict[str, list] = defaultdict(list) - - for file_path in files: - try: - content = file_path.read_text() - tree = ast.parse(content) - - for node in ast.walk(tree): - if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): - func_code = ast.unparse(node) - if len(func_code.splitlines()) >= min_lines: - func_hash = hashlib.sha256(func_code.encode()).hexdigest() - hash_to_files[func_hash].append({ - "file": str(file_path), - "name": node.name, - "lines": len(func_code.splitlines()), - "start_line": node.lineno, - }) - except (SyntaxError, ValueError): - continue - - # Report exact duplicates - for func_hash, locations in hash_to_files.items(): - if len(locations) > 1: - exact_duplicates.append({ - "type": "exact_function_duplicate", - "locations": locations, - "hash": func_hash[:12], - }) - - # Similar block detection (simplified - based on normalized structure) - similar_blocks = self._find_similar_blocks(files, min_lines, threshold) - - return {"exact": exact_duplicates, "similar": similar_blocks} - - def _scan_js_duplicates( - self, files: list[Path], min_lines: int, threshold: float - ) -> dict[str, list]: - """Scan JS/TS files for duplicate functions and code blocks.""" - exact_duplicates = [] - similar_blocks = [] - - # Simple regex-based function extraction for JS/TS - func_pattern = re.compile( - r"(?:function\s+(\w+)|(\w+)\s*=\s*(?:async\s+)?\(|(\w+)\s*\([^)]*\)\s*=>)" - r"[\s\S]{0,5000}", - re.MULTILINE - ) - - hash_to_files: dict[str, list] = defaultdict(list) - - for file_path in files: - try: - content = file_path.read_text() - for match in func_pattern.finditer(content): - func_name = match.group(1) or match.group(2) or match.group(3) - if not func_name: - continue - func_code = match.group(0) - if len(func_code.splitlines()) >= min_lines: - func_hash = hashlib.sha256(func_code.encode()).hexdigest() - hash_to_files[func_hash].append({ - "file": str(file_path), - "name": func_name, - "lines": len(func_code.splitlines()), - }) - except Exception: - continue - - for func_hash, locations in hash_to_files.items(): - if len(locations) > 1: - exact_duplicates.append({ - "type": "exact_function_duplicate", - "locations": locations, - "hash": func_hash[:12], - }) - - similar_blocks = self._find_similar_blocks(files, min_lines, threshold) - return {"exact": exact_duplicates, "similar": similar_blocks} - - def _find_similar_blocks( - self, files: list[Path], min_lines: int, threshold: float - ) -> list[dict]: - """Find similar code blocks using normalized comparison.""" - similar = [] - block_hashes: dict[str, list] = defaultdict(list) - - for file_path in files: - try: - content = file_path.read_text() - lines = content.splitlines() - - # Extract blocks of consecutive non-empty lines - i = 0 - while i < len(lines): - if lines[i].strip() and not lines[i].strip().startswith("#"): - # Find block end - block_lines = [] - j = i - while j < len(lines) and lines[j].strip(): - if len(lines[j]) < MAX_LINE_LENGTH: - block_lines.append(lines[j]) - j += 1 - - if len(block_lines) >= min_lines: - # Normalize whitespace - normalized = "\n".join( - line.strip() for line in block_lines - ) - block_hash = hashlib.sha256( - normalized.encode() - ).hexdigest() - block_hashes[block_hash].append({ - "file": str(file_path), - "start_line": i + 1, - "end_line": j, - "lines": len(block_lines), - }) - i = j - else: - i += 1 - except Exception: - continue - - # Report similar blocks (same hash = identical after normalization) - for block_hash, locations in block_hashes.items(): - if len(locations) > 1: - similar.append({ - "type": "similar_block", - "locations": locations, - "hash": block_hash[:12], - }) - - return similar - - def _find_similar_functions( - self, files: list[Path], threshold: float - ) -> list[dict]: - """Find similar (overlapping) functions using AST-based similarity scoring.""" - function_signatures: list[dict] = [] - - for file_path in files: - try: - content = file_path.read_text() - tree = ast.parse(content) - - for node in ast.walk(tree): - if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): - func_code = ast.unparse(node) - if len(func_code.splitlines()) >= MIN_LINES_FOR_DUPLICATE: - # Extract signature info - args = [arg.arg for arg in node.args.args] - returns = ast.unparse(node.returns) if node.returns else None - function_signatures.append({ - "file": str(file_path), - "name": node.name, - "start_line": node.lineno, - "lines": len(func_code.splitlines()), - "args": args, - "returns": returns, - "normalized": self._normalize_function(func_code), - }) - except (SyntaxError, ValueError): - continue - - # Compare all function pairs for similarity - similar_functions = [] - for i, func_a in enumerate(function_signatures): - for func_b in function_signatures[i + 1:]: - # Skip if same file - if func_a["file"] == func_b["file"]: - continue - - # Compare normalized code - similarity = SequenceMatcher( - None, func_a["normalized"], func_b["normalized"] - ).ratio() - - if similarity >= threshold and similarity < 1.0: - similar_functions.append({ - "type": "similar_function", - "similarity": round(similarity, 3), - "functions": [ - {"file": func_a["file"], "name": func_a["name"], "start_line": func_a["start_line"]}, - {"file": func_b["file"], "name": func_b["name"], "start_line": func_b["start_line"]}, - ], - }) - - # Sort by similarity descending - similar_functions.sort(key=lambda x: x["similarity"], reverse=True) - return similar_functions[:50] # Limit to top 50 - - def _normalize_function(self, code: str) -> str: - """Normalize function code for similarity comparison.""" - # Remove variable names, keep structure - normalized = re.sub(r'\b[a-zA-Z_][a-zA-Z0-9_]*\b', 'VAR', code) - # Remove string literals - normalized = re.sub(r'["\'][^"\']*["\']', 'STR', normalized) - # Remove numbers - normalized = re.sub(r'\b\d+\b', 'NUM', normalized) - # Normalize whitespace - return ' '.join(normalized.split()) - - def _find_variable_consolidation( - self, files: list[Path] - ) -> list[dict]: - """Find $VARIABLE patterns and consolidation opportunities.""" - var_to_files: dict[str, list] = defaultdict(list) - env_var_patterns = [ - re.compile(r'\$\{?([A-Z_][A-Z0-9_]*)\}?'), # ${VAR} or $VAR - re.compile(r'os\.environ\.get\(["\']([A-Z_][A-Z0-9_]*)["\']'), # os.environ.get('VAR') - re.compile(r'os\.getenv\(["\']([A-Z_][A-Z0-9_]*)["\']'), # os.getenv('VAR') - re.compile(r'process\.env\.([A-Z_][A-Z0-9_]*)'), # process.env.VAR (JS) - ] - - for file_path in files: - try: - content = file_path.read_text() - for pattern in env_var_patterns: - for match in pattern.finditer(content): - var_name = match.group(1) - var_to_files[var_name].append(str(file_path)) - except Exception: - continue - - # Find variables used in multiple places - consolidation = [] - for var_name, locations in var_to_files.items(): - if len(locations) > 2: # Used in more than 2 files - # Determine scope based on usage patterns - scope = "global" if len(locations) > 5 else "user" - consolidation.append({ - "variable": var_name, - "locations": list(set(locations)), # Unique files - "count": len(locations), - "scope": scope, - "recommendation": f"Consider centralizing ${var_name} in shared config", - }) - - return sorted(consolidation, key=lambda x: x["count"], reverse=True)[:30] - - def _find_script_duplication( - self, target: str - ) -> list[dict]: - """Find duplicated scripts across snacks, scripts, and docs directories.""" - root = Path(__file__).parent.parent.parent.parent - script_patterns = [ - "scripts/**/*.py", - "scripts/**/*.sh", - "backend/tools/**/*.py", - "docs/**/*.md", - "backend/app/menu/snacks/**/*.py", - ] - - all_files = [] - for pattern in script_patterns: - all_files.extend((root / pattern.replace("**", "")).parent.rglob(pattern.split("/")[-1])) - - # Filter out excluded dirs - exclude_dirs = {"node_modules", ".venv", "venv", "__pycache__", ".git", "dist", "build"} - all_files = [f for f in all_files if not any(ex in f.parts for ex in exclude_dirs)] - - # Hash-based detection for scripts - hash_to_files: dict[str, list] = defaultdict(list) - for file_path in all_files: - try: - content = file_path.read_text() - # Normalize for comparison - normalized = '\n'.join( - line.strip() for line in content.splitlines() - if line.strip() and not line.strip().startswith('#') - ) - file_hash = hashlib.sha256(normalized.encode()).hexdigest() - hash_to_files[file_hash].append({ - "file": str(file_path), - "category": self._get_category(file_path), - }) - except Exception: - continue - - # Report duplicates - script_duplicates = [] - for file_hash, locations in hash_to_files.items(): - if len(locations) > 1: - script_duplicates.append({ - "type": "script_duplication", - "locations": locations, - "hash": file_hash[:12], - }) - - return script_duplicates - - def _get_category(self, file_path: Path) -> str: - """Determine category for a file path.""" - path_str = str(file_path) - if "/scripts/" in path_str: - return "script" - if "/snacks/" in path_str: - return "snack" - if "/docs/" in path_str: - return "doc" - if "/tools/" in path_str: - return "tool" - return "other" - - def _find_duplicate_imports(self, files: list[Path]) -> list[dict]: - """Find duplicate import patterns across files.""" - import_to_files: dict[str, list] = defaultdict(list) - - for file_path in files: - try: - content = file_path.read_text() - tree = ast.parse(content) - - for node in ast.walk(tree): - if isinstance(node, ast.Import): - for alias in node.names: - import_to_files[alias.name].append(str(file_path)) - elif isinstance(node, ast.ImportFrom): - if node.module: - import_to_files[node.module].append(str(file_path)) - except (SyntaxError, ValueError): - continue - - duplicates = [] - for imp, locations in import_to_files.items(): - if len(locations) > 3: # Appears in more than 3 files - duplicates.append({ - "import": imp, - "locations": locations, - "count": len(locations), - }) - - return duplicates - - def _find_repeated_literals(self, files: list[Path]) -> list[dict]: - """Find repeated string literals across files.""" - literal_to_files: dict[str, list] = defaultdict(list) - - for file_path in files: - try: - content = file_path.read_text() - # Find string literals (simple approach) - literals = re.findall(r'["\']([^"\']{10,50})["\']', content) - for lit in literals: - literal_to_files[lit].append(str(file_path)) - except Exception: - continue - - repeated = [] - for lit, locations in literal_to_files.items(): - if len(locations) > 3: - repeated.append({ - "literal": lit[:50] + "..." if len(lit) > 50 else lit, - "locations": locations[:5], # Limit to 5 files - "count": len(locations), - }) - - return sorted(repeated, key=lambda x: x["count"], reverse=True)[:20] - - def _generate_recommendations(self, scan_result: dict) -> list[str]: - """Generate recommendations based on scan results.""" - recs = [] - findings = scan_result.get("findings", {}) - - exact = findings.get("exact_duplicates", []) - if exact: - recs.append(f"Found {len(exact)} exact duplicate functions — consider extracting to shared module") - - similar = findings.get("similar_blocks", []) - if similar: - recs.append(f"Found {len(similar)} similar code blocks — review for consolidation") - - func_similar = findings.get("similar_functions", []) - if func_similar: - recs.append(f"Found {len(func_similar)} similar functions with overlapping logic — consider merging") - - imports = findings.get("duplicate_imports", []) - if imports: - top_imports = sorted(imports, key=lambda x: x["count"], reverse=True)[:5] - recs.append(f"Top repeated imports: {', '.join(i['import'] for i in top_imports)}") - - vars_consolidation = findings.get("variable_consolidation", []) - if vars_consolidation: - top_vars = sorted(vars_consolidation, key=lambda x: x["count"], reverse=True)[:5] - recs.append(f"Variables needing consolidation: {', '.join(v['variable'] for v in top_vars)}") - - script_dups = findings.get("script_duplication", []) - if script_dups: - recs.append(f"Found {len(script_dups)} duplicated scripts across snacks/scripts/docs") - - large_files = findings.get("large_files", []) - if large_files: - top_large = sorted(large_files, key=lambda x: x["lines"], reverse=True)[:5] - large_list = [f"{f['file']} ({f['lines']} lines)" for f in top_large] - recs.append(f"Large files needing modularization: {', '.join(large_list)}") - - dup_funcs = findings.get("duplicate_functions_in_file", []) - if dup_funcs: - recs.append(f"Found {len(dup_funcs)} duplicate functions within files — consolidate") - - circular = findings.get("circular_imports", []) - if circular: - recs.append(f"Found {len(circular)} potential circular imports — review imports") - - return recs - - def _archive_duplicates(self, scan_result: dict, dry_run: bool) -> dict: - """Archive duplicate code findings to a report file.""" - archive_dir = Path(__file__).parent.parent.parent.parent / "tmp" / "duplicate_reports" - archive_dir.mkdir(parents=True, exist_ok=True) - - timestamp = int(time.time()) - report_file = archive_dir / f"duplicates_{timestamp}.json" - - if dry_run: - return { - "success": True, - "action": "archive", - "dry_run": True, - "would_write": str(report_file), - "findings_count": scan_result.get("stats", {}).get("total_issues", 0), - } - - report_file.write_text(json.dumps(scan_result, indent=2)) - return { - "success": True, - "action": "archive", - "report_file": str(report_file), - "findings_count": scan_result.get("stats", {}).get("total_issues", 0), - } - - def _find_large_files(self, files: list[Path]) -> list[dict]: - """Find files that need modularization (>1000 lines).""" - large_files = [] - for file_path in files: - try: - line_count = len(file_path.read_text().splitlines()) - if line_count > 1000: - large_files.append({ - "file": str(file_path), - "lines": line_count, - "recommendation": "Split into smaller modules (target: <500 lines per module)", - }) - except Exception: - continue - - return sorted(large_files, key=lambda x: x["lines"], reverse=True)[:20] - - def _find_duplicate_functions_in_file(self, files: list[Path]) -> list[dict]: - """Find duplicate functions within the same file.""" - duplicates = [] - for file_path in files: - try: - content = file_path.read_text() - tree = ast.parse(content) - - functions = [] - for node in ast.walk(tree): - if isinstance(node, ast.FunctionDef): - func_name = node.name - func_start = node.lineno - func_end = node.end_lineno - func_code = "\n".join(content.splitlines()[func_start-1:func_end]) - functions.append({ - "name": func_name, - "start": func_start, - "end": func_end, - "code": func_code, - }) - - # Find exact duplicates within the same file - seen = {} - for func in functions: - func_hash = hashlib.md5(func["code"].encode()).hexdigest() - if func_hash in seen: - duplicates.append({ - "file": str(file_path), - "function_1": seen[func_hash]["name"], - "line_1": seen[func_hash]["start"], - "function_2": func["name"], - "line_2": func["start"], - "reason": "Exact duplicate function within same file", - }) - else: - seen[func_hash] = func - - except (SyntaxError, ValueError): - continue - - return duplicates - - def _find_circular_imports(self, files: list[Path]) -> list[dict]: - """Find potential circular import patterns.""" - circular = [] - import_graph = defaultdict(set) - - for file_path in files: - try: - content = file_path.read_text() - tree = ast.parse(content) - - for node in ast.walk(tree): - if isinstance(node, ast.ImportFrom): - if node.module: - import_graph[str(file_path)].add(node.module) - except (SyntaxError, ValueError): - continue - - # Simple circular import detection (not exhaustive) - for file_a, imports in import_graph.items(): - for file_b in imports: - if file_b in import_graph and file_a in import_graph[file_b]: - circular.append({ - "file_a": file_a, - "file_b": file_b, - "reason": "Potential circular import detected", - }) - - return circular - - def _find_duplicate_install_patterns(self, files: list[Path]) -> list[dict]: - """Find duplicate installation/launchd patterns across files.""" - install_patterns: dict[str, list] = defaultdict(list) - - # Pattern for launchd plist generation - launchd_pattern = re.compile( - r'launchctl\s+(?:bootstrap|load).*com\.udos\.ucore', - re.IGNORECASE - ) - - # Pattern for plist content generation - plist_pattern = re.compile( - r'<\?xml version="1\.0".*plist.*', - re.DOTALL - ) - - for file_path in files: - try: - content = file_path.read_text() - - # Check for launchd bootstrap calls - for match in launchd_pattern.finditer(content): - install_patterns["launchd_bootstrap"].append({ - "file": str(file_path), - "line": content[:match.start()].count('\n') + 1, - }) - - # Check for inline plist generation - for match in plist_pattern.finditer(content): - # Extract the module being installed - module_match = re.search(r'-m\s*([^<]+)', match.group(0)) - module = module_match.group(1) if module_match else "unknown" - install_patterns["plist_generation"].append({ - "file": str(file_path), - "module": module, - "line": content[:match.start()].count('\n') + 1, - }) - except Exception: - continue - - # Report patterns found in multiple files - duplicates = [] - for pattern_type, locations in install_patterns.items(): - if len(locations) > 1: - duplicates.append({ - "type": f"duplicate_{pattern_type}", - "locations": locations, - "count": len(locations), - "recommendation": f"Consider consolidating {pattern_type} into single module", - }) - - return duplicates - - def _remove_duplicate_files(self, duplicates: list[dict], dry_run: bool = True) -> dict: - """Remove or archive duplicate files that have been consolidated.""" - removed = [] - errors = [] - - for dup in duplicates: - if dup.get("type") == "duplicate_plist_generation": - # These are now handled by launchd_manager.py - for loc in dup.get("locations", []): - file_path = Path(loc["file"]) - if "launchd_manager.py" in str(file_path): - continue # Skip the canonical file - - if dry_run: - removed.append({"file": str(file_path), "action": "would_remove"}) - else: - # Move to archive instead of delete - archive_dir = Path(__file__).parent.parent.parent.parent / "tmp" / "archived_duplicates" - archive_dir.mkdir(parents=True, exist_ok=True) - try: - # Just rename with .archived suffix for safety - archived = archive_dir / f"{file_path.stem}.archived{file_path.suffix}" - file_path.rename(archived) - removed.append({"file": str(file_path), "action": "archived", "to": str(archived)}) - except Exception as e: - errors.append({"file": str(file_path), "error": str(e)}) - - return {"removed": removed, "errors": errors} diff --git a/backend/app/skills/builtin/skill_enhancement_planner.py b/backend/app/skills/builtin/skill_enhancement_planner.py deleted file mode 100644 index d333f5ee..00000000 --- a/backend/app/skills/builtin/skill_enhancement_planner.py +++ /dev/null @@ -1,431 +0,0 @@ -"""Enhancement Planner Skill — bridges ecosystem audits to actionable tasks. - -Takes ecosystem audit output and skill audit reports, groups items -by enhancement area, and generates .tasker Markdown items for each gap. - -Actions: - - plan — read audit reports, generate .tasker items - - generate — write ENHANCEMENT_PLAN.md to docs/ -""" -from __future__ import annotations - -import json -import logging -from datetime import datetime, timezone -from pathlib import Path - -from app.skills.base import BaseSkill, SkillMeta, SkillParam - -log = logging.getLogger("ucore.skills.enhancement_planner") - -UCORE_ROOT = Path(__file__).parent.parent.parent.parent.parent -DOCS_DIR = UCORE_ROOT / "docs" -SEEDS_DIR = UCORE_ROOT / "seeds" -TASKER_DIR = UCORE_ROOT / ".tasker" - -CATEGORIES = { - "skills": "Skills & Automation", - "mcp_servers": "MCP Servers", - "runtimes": "Backend Runtimes", - "routes": "API Routes", - "secrets": "Secrets & Config", - "variables": "Variables", - "paths": "File System Paths", -} - - -class EnhancementPlannerSkill(BaseSkill): - """Generate .tasker items from ecosystem audit gaps.""" - - meta = SkillMeta( - id="enhancement-planner", - name="Enhancement Planner", - description=( - "Bridges ecosystem audits to actionable .tasker items." - " Reads audit reports, groups gaps by area, and generates" - " prioritized enhancement tasks." - ), - category="developer", - params=[ - SkillParam( - name="action", - type="string", - description="'plan' (generate tasks) or 'generate' (write to docs)", - required=False, - default="plan", - ), - SkillParam( - name="output", - type="string", - description="Output path for ENHANCEMENT_PLAN.md", - required=False, - default=str(DOCS_DIR / "ENHANCEMENT_PLAN.md"), - ), - ], - timeout=60, - requires_confirmation=False, - ) - - async def run(self, **kwargs) -> dict: - action = kwargs.get("action", "plan") - output_path = kwargs.get("output", str(DOCS_DIR / "ENHANCEMENT_PLAN.md")) - - # Load audit data - skills_audit = self._load_json("skill-audit-report.json") - eco_assess = self._load_json("ecosystem-registry.json") - - # Generate tasks grouped by category - tasks = self._generate_tasks(skills_audit, eco_assess) - - if action == "generate": - md = self._render_markdown(tasks, skills_audit, eco_assess) - out = Path(output_path) - out.parent.mkdir(parents=True, exist_ok=True) - out.write_text(md) - return { - "success": True, - "action": "generate", - "output": str(out), - "task_count": len(tasks), - } - - return { - "success": True, - "action": "plan", - "task_count": len(tasks), - "tasks": tasks, - "recommendations": self._plan_recommendations(tasks), - } - - # ─── Data Loading ───────────────────────────────────────────── - - @staticmethod - def _load_json(filename: str) -> dict: - """Load a JSON report from seeds/.""" - path = SEEDS_DIR / filename - if not path.exists(): - return {} - try: - return json.loads(path.read_text()) - except Exception: - return {} - - # ─── Task Generation ────────────────────────────────────────── - - def _generate_tasks( - self, skills_audit: dict, eco_assess: dict, - ) -> list[dict]: - """Generate prioritized enhancement tasks from audit data.""" - tasks: list[dict] = [] - tid = 0 - - tid = self._tasks_from_skills(skills_audit, tasks, tid) - tid = self._tasks_from_ecosystem(eco_assess, tasks, tid) - tid = self._tasks_from_services(tasks, tid) - - # Sort by priority - priority_order = {"p0": 0, "p1": 1, "p2": 2, "p3": 3} - tasks.sort(key=lambda t: priority_order.get(str(t.get("priority", "p2")), 2)) - - # Assign sequential IDs - for i, t in enumerate(tasks, 1): - t["id"] = f"task.enhance.{i:03d}" - - return tasks - - def _tasks_from_skills( - self, audit: dict, tasks: list[dict], tid: int, - ) -> int: - """Generate tasks for broken/untested skills.""" - if not audit: - return tid - - # Broken skills - broken = [ - s for s in audit.get("skills", []) - if s.get("execute_status") == "failed" - or s.get("import_status") == "failed" - ] - for s in broken[:10]: - tid += 1 - tasks.append({ - "id": f"task.enhance.{tid:03d}", - "title": f"Fix broken skill: {s.get('name', 'unknown')}", - "category": "skills", - "priority": "p0", - "status": "backlog", - "description": ( - f"Skill **{s.get('name')}** fails to " - f"load or execute. Error: {s.get('error', 'unknown')}" - ), - "tags": ["broken", "skills", "audit"], - }) - - # Untested skills - untested = [ - s for s in audit.get("skills", []) - if s.get("execute_status") not in ("success", "failed") - and s.get("import_status") != "failed" - ] - for s in untested[:15]: - tid += 1 - tasks.append({ - "id": f"task.enhance.{tid:03d}", - "title": f"Smoke-test skill: {s.get('name', 'unknown')}", - "category": "skills", - "priority": "p1", - "status": "backlog", - "description": ( - f"Skill **{s.get('name')}** has never been verified." - " Run with dry_run=True and confirm it executes." - ), - "tags": ["untested", "skills", "audit"], - }) - - # Skills without BaseSkill subclass - no_base = [ - s for s in audit.get("skills", []) - if not s.get("is_base_skill") and s.get("has_module_run") - ] - for s in no_base[:10]: - tid += 1 - tasks.append({ - "id": f"task.enhance.{tid:03d}", - "title": f"Convert to BaseSkill: {s.get('name', 'unknown')}", - "category": "skills", - "priority": "p2", - "status": "backlog", - "description": ( - f"**{s.get('name')}** uses module-level run()." - " Convert to BaseSkill subclass for registry integration." - ), - "tags": ["refactor", "skills", "base-skill"], - }) - - return tid - - def _tasks_from_ecosystem( - self, assess: dict, tasks: list[dict], tid: int, - ) -> int: - """Generate tasks from ecosystem assess report.""" - eco = assess.get("ecosystem", {}) - health = assess.get("health", {}) - - # Broken ecosystem items - for cat_key, cat_label in CATEGORIES.items(): - items = eco.get(cat_key, []) - if not isinstance(items, list): - items = items.get("items", []) if isinstance(items, dict) else [] - - for item in items: - if isinstance(item, dict) and item.get("health") == "broken": - tid += 1 - name = item.get("name", item.get("key", item.get("path", "unknown"))) - tasks.append({ - "id": f"task.enhance.{tid:03d}", - "title": f"Fix broken {cat_label[:-1]}: {name}", - "category": cat_key, - "priority": "p0", - "status": "backlog", - "description": ( - f"**{name}** in {cat_label} is marked broken." - f" Issues: {item.get('issues', [])}" - ), - "tags": ["broken", cat_key, "ecosystem"], - }) - - # Orphaned items - for cat_key in ["skills", "mcp_servers", "runtimes"]: - items = eco.get(cat_key, []) - if not isinstance(items, list): - continue - for item in items: - if isinstance(item, dict) and item.get("health") == "orphaned": - tid += 1 - name = item.get("name", "unknown") - tasks.append({ - "id": f"task.enhance.{tid:03d}", - "title": f"Review orphaned {cat_key[:-1]}: {name}", - "category": cat_key, - "priority": "p2", - "status": "backlog", - "description": ( - f"**{name}** is orphaned — no references found." - " Consider archiving or re-wiring." - ), - "tags": ["orphaned", cat_key, "ecosystem"], - }) - - # High untested count - if health.get("untested", 0) > 10: - tid += 1 - tasks.append({ - "id": f"task.enhance.{tid:03d}", - "title": "Smoke-test untested ecosystem items", - "category": "ecosystem", - "priority": "p1", - "status": "backlog", - "description": ( - f"{health['untested']} ecosystem items are untested." - " Run ecosystem-audit assess and smoke-test each." - ), - "tags": ["untested", "ecosystem"], - }) - - return tid - - def _tasks_from_services( - self, tasks: list[dict], tid: int, - ) -> int: - """Generate tasks for service health.""" - # General recommendations - tid += 1 - tasks.append({ - "id": f"task.enhance.{tid:03d}", - "title": "Run ucore-index health-report weekly", - "category": "ecosystem", - "priority": "p2", - "status": "backlog", - "description": ( - "Schedule a weekly ucore-index health-report to track" - " ecosystem health over time. Add to maintenance scheduler." - ), - "tags": ["monitoring", "health", "scheduled"], - }) - - tid += 1 - tasks.append({ - "id": f"task.enhance.{tid:03d}", - "title": "Integrate GH Actions for CI health checks", - "category": "runtimes", - "priority": "p2", - "status": "backlog", - "description": ( - "Add a GitHub Action that runs ucore-index health-report" - " on push and alerts on degraded/critical status." - ), - "tags": ["ci", "health", "github"], - }) - - return tid - - # ─── Markdown Rendering ─────────────────────────────────────── - - def _render_markdown( - self, tasks: list[dict], skills_audit: dict, eco_assess: dict, - ) -> str: - """Render ENHANCEMENT_PLAN.md.""" - now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") - health = eco_assess.get("health", {}) - - lines = [ - "# uCore Enhancement Plan", - f"**Generated:** {now}", - "**Source:** ecosystem-audit assess + skill-audit", - "", - "## Ecosystem Health Summary", - "", - "| Metric | Value |", - "|--------|-------|", - f"| Total Items | {health.get('total_items', '—')} |", - f"| Working | {health.get('working', '—')} |", - f"| Untested | {health.get('untested', '—')} |", - f"| Broken | {health.get('broken', '—')} |", - f"| Health % | {health.get('health_pct', '—')}% |", - "", - "## Skill Health Summary", - "", - ] - - if skills_audit: - lines += [ - "| Metric | Value |", - "|--------|-------|", - f"| Total Skills | {skills_audit.get('total_skills', '—')} |", - f"| Working | {skills_audit.get('working', '—')} |", - f"| Untested | {skills_audit.get('untested', '—')} |", - f"| Broken | {skills_audit.get('broken', '—')} |", - f"| Health % | {skills_audit.get('health_pct', '—')}% |", - "", - ] - - lines += [ - "## Prioritized Tasks", - "", - f"**Total:** {len(tasks)} tasks generated from audit gaps.", - "", - ] - - # Group by priority - for prio_label, prio_key in [ - ("### P0 — Critical (Fix Broken)", "p0"), - ("### P1 — High (Smoke-Test)", "p1"), - ("### P2 — Medium (Improve)", "p2"), - ("### P3 — Low (Nice-to-Have)", "p3"), - ]: - prio_tasks = [t for t in tasks if t.get("priority") == prio_key] - if not prio_tasks: - continue - lines.append(prio_label) - lines.append("") - for t in prio_tasks: - lines.append( - f"- [ ] **{t['id']}** — {t['title']} " - f"(`{t.get('category', '')}`)" - ) - lines.append(f" {t.get('description', '')}") - tags = t.get("tags", []) - if tags: - lines.append(f" Tags: {', '.join('`' + tag + '`' for tag in tags)}") - lines.append("") - lines.append("") - - lines += [ - "## How to Use This Plan", - "", - "1. Import tasks into .tasker: `tasker ingest --from ENHANCEMENT_PLAN.md`", - "2. Assign tasks to agents using kanban", - "3. Run `dev-mode-executor` with task UIDs to auto-execute", - "4. Re-run `ecosystem-audit assess` weekly to track progress", - "", - "## Regenerate", - "", - "To regenerate this plan after fixing items:", - "```bash", - "curl -X POST http://localhost:8484/api/skills/enhancement-planner/run \\", - ' -H "Content-Type: application/json" \\', - ' -d \'{"action": "generate"}\'', - "```", - "", - ] - - return "\n".join(lines) + "\n" - - # ─── Recommendations ────────────────────────────────────────── - - @staticmethod - def _plan_recommendations(tasks: list[dict]) -> list[str]: - """Generate top-level recommendations from the plan.""" - p0 = sum(1 for t in tasks if t.get("priority") == "p0") - p1 = sum(1 for t in tasks if t.get("priority") == "p1") - p2 = sum(1 for t in tasks if t.get("priority") == "p2") - - recs = [] - if p0 > 0: - recs.append( - f"Address {p0} critical (P0) tasks first — " - "these are broken items blocking functionality" - ) - if p1 > 0: - recs.append( - f"Complete {p1} high-priority (P1) tasks — " - "smoke-testing untested items" - ) - if p2 > 0: - recs.append( - f"Schedule {p2} medium-priority (P2) tasks — " - "conversions and improvements" - ) - if not recs: - recs.append("No enhancement tasks needed — ecosystem is healthy") - return recs diff --git a/backend/app/skills/builtin/skill_gh_workflow_bridge.py b/backend/app/skills/builtin/skill_gh_workflow_bridge.py deleted file mode 100644 index 9f3545cb..00000000 --- a/backend/app/skills/builtin/skill_gh_workflow_bridge.py +++ /dev/null @@ -1,318 +0,0 @@ -"""GitHub Workflow Bridge Skill — bridge tasks to GitHub Actions/CLI. - -Bridges uCore tasks to GitHub Actions and CLI workflows: - - trigger-ci: trigger a GitHub Actions workflow run - - create-pr: create a pull request - - run-workflow: execute a specific workflow by name - - status: check CI status for a repo - -Integrates with: gh CLI, GitHub API, tasker. -""" -from __future__ import annotations - -import asyncio -import logging -import subprocess - -from app.skills.base import BaseSkill, SkillMeta, SkillParam - -log = logging.getLogger("ucore.skills.gh_workflow_bridge") - - -def _find_gh_binary() -> str | None: - """Locate the gh CLI binary.""" - try: - result = subprocess.run( - ["which", "gh"], - capture_output=True, text=True, timeout=2, check=False, - ) - if result.returncode == 0: - return result.stdout.strip() - except Exception: - pass - return None - - -def _is_gh_authenticated() -> bool: - """Check if gh is authenticated.""" - try: - result = subprocess.run( - ["gh", "auth", "status"], - capture_output=True, text=True, timeout=3, check=False, - ) - return result.returncode == 0 - except Exception: - return False - - -class GHWorkflowBridgeSkill(BaseSkill): - """Bridge uCore tasks to GitHub Actions and CLI workflows.""" - - meta = SkillMeta( - id="gh-workflow-bridge", - name="GitHub Workflow Bridge", - description=( - "Bridge tasks to GitHub Actions/CLI." - " Trigger CI, create PRs, run workflows." - ), - category="developer", - timeout=180, - params=[ - SkillParam( - name="action", - type="string", - required=True, - description=( - "GitHub action: 'trigger-ci', 'create-pr'," - " 'run-workflow', or 'status'" - ), - ), - SkillParam( - name="repo", - type="string", - required=False, - default="", - description=( - "GitHub repo in owner/repo format" - " (default: uDosGo/uCore)" - ), - ), - SkillParam( - name="workflow", - type="string", - required=False, - default="", - description="Workflow name or ID (for run-workflow)", - ), - SkillParam( - name="branch", - type="string", - required=False, - default="main", - description="Branch name (for create-pr)", - ), - SkillParam( - name="title", - type="string", - required=False, - default="", - description="PR title (for create-pr)", - ), - SkillParam( - name="body", - type="string", - required=False, - default="", - description="PR description body (for create-pr)", - ), - ], - requires_confirmation=True, - ) - - async def run(self, **kwargs) -> dict: - action = kwargs.get("action", "").strip() - repo = kwargs.get("repo", "uDosGo/uCore").strip() - workflow = kwargs.get("workflow", "").strip() - branch = kwargs.get("branch", "main").strip() - title = kwargs.get("title", "").strip() - body = kwargs.get("body", "").strip() - - if not action: - return {"success": False, "error": "action is required"} - - # Check gh CLI availability - gh_bin = _find_gh_binary() - if not gh_bin: - return { - "success": False, - "error": ( - "GitHub CLI (gh) not found." - " Install: brew install gh && gh auth login" - ), - } - - if not _is_gh_authenticated(): - return { - "success": False, - "error": "gh not authenticated. Run: gh auth login", - } - - actions = { - "trigger-ci": self._trigger_ci, - "create-pr": self._create_pr, - "run-workflow": self._run_workflow, - "status": self._check_status, - } - - handler = actions.get(action) - if not handler: - return { - "success": False, - "error": ( - f"Unknown action: {action}." - " Use: trigger-ci, create-pr, run-workflow, status" - ), - } - - result = await handler(repo, workflow, branch, title, body) - return { - "success": True, - "action": action, - "repo": repo, - **result, - } - - async def _trigger_ci( - self, repo: str, workflow: str, branch: str, - title: str, body: str, - ) -> dict: - """Trigger a CI workflow run.""" - wf_ref = workflow or "ci" - try: - proc = await asyncio.create_subprocess_exec( - "gh", "workflow", "run", wf_ref, - "--repo", repo, - "--ref", branch, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) - stdout, stderr = await asyncio.wait_for( - proc.communicate(), timeout=30, - ) - output = stdout.decode().strip() if stdout else "" - err = stderr.decode().strip() if stderr else "" - - if proc.returncode != 0: - return { - "triggered": False, - "error": err or "Unknown error", - "workflow": wf_ref, - } - - # Extract run ID from output - run_id = "" - for line in output.splitlines(): - if "run" in line.lower() and any(c.isdigit() for c in line): - import re - digits = re.findall(r'\d+', line) - if digits: - run_id = digits[0] - break - - return { - "triggered": True, - "workflow": wf_ref, - "run_id": run_id, - "output": output, - } - except asyncio.TimeoutError: - return {"triggered": False, "error": "Timeout after 30s"} - except Exception as exc: - return {"triggered": False, "error": str(exc)} - - async def _create_pr( - self, repo: str, workflow: str, branch: str, - title: str, body: str, - ) -> dict: - """Create a pull request.""" - if not title: - return {"error": "PR title is required"} - - cmd = [ - "gh", "pr", "create", - "--repo", repo, - "--base", branch, - "--title", title, - ] - if body: - cmd.extend(["--body", body]) - - try: - proc = await asyncio.create_subprocess_exec( - *cmd, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) - stdout, stderr = await asyncio.wait_for( - proc.communicate(), timeout=30, - ) - output = stdout.decode().strip() if stdout else "" - err = stderr.decode().strip() if stderr else "" - - if proc.returncode != 0: - return { - "created": False, - "error": err or "Unknown error", - } - - # Extract PR URL - pr_url = "" - for line in output.splitlines(): - if "http" in line and "pull" in line: - pr_url = line.strip() - break - - return { - "created": proc.returncode == 0, - "pr_url": pr_url, - "output": output, - } - except asyncio.TimeoutError: - return {"created": False, "error": "Timeout after 30s"} - except Exception as exc: - return {"created": False, "error": str(exc)} - - async def _run_workflow( - self, repo: str, workflow: str, branch: str, - title: str, body: str, - ) -> dict: - """Run a specific workflow.""" - if not workflow: - return {"error": "workflow name or ID is required"} - return await self._trigger_ci( - repo, workflow, branch, title, body, - ) - - async def _check_status( - self, repo: str, workflow: str, branch: str, - title: str, body: str, - ) -> dict: - """Check CI status for a repo.""" - try: - proc = await asyncio.create_subprocess_exec( - "gh", "run", "list", - "--repo", repo, - "--limit", "5", - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) - stdout, stderr = await asyncio.wait_for( - proc.communicate(), timeout=15, - ) - output = stdout.decode().strip() if stdout else "" - err = stderr.decode().strip() if stderr else "" - - if proc.returncode != 0: - return {"status": "error", "error": err or "Unknown error"} - - # Parse runs - runs = [] - for line in output.splitlines(): - if line.strip(): - runs.append(line.strip()) - - # Determine overall status - status = "healthy" - for line in runs: - if "fail" in line.lower(): - status = "failing" - break - - return { - "status": status, - "recent_runs": runs, - } - except asyncio.TimeoutError: - return {"status": "unknown", "error": "Timeout after 15s"} - except Exception as exc: - return {"status": "error", "error": str(exc)} diff --git a/backend/app/skills/builtin/skill_hardcoded_path_detector.py b/backend/app/skills/builtin/skill_hardcoded_path_detector.py deleted file mode 100644 index 10236c17..00000000 --- a/backend/app/skills/builtin/skill_hardcoded_path_detector.py +++ /dev/null @@ -1,227 +0,0 @@ -#!/usr/bin/env python3 -"""Hardcoded Path Detector Skill — find and report hardcoded paths. - -Scans source files to detect: -- Absolute paths hardcoded in code -- Environment variable references that should be used instead -- Platform-specific paths that should be configurable -- UDOS_CODE, UDOS_VAULT, HOME references - -Reports: file locations, line numbers, path types, and recommendations. -""" -from __future__ import annotations - -import logging -import os -import re -from pathlib import Path -from typing import Any - -from app.skills.base import BaseSkill, SkillMeta, SkillParam - -log = logging.getLogger("ucore.skills.hardcoded_path_detector") - -# Hardcoded path patterns to detect -HARDCODED_PATH_PATTERNS = [ - # Absolute paths - (r'/Users/[a-zA-Z0-9_]+', 'Absolute macOS user path'), - (r'/Users/[a-zA-Z0-9_]+', 'Absolute Linux user path'), - (r'/home/[a-zA-Z0-9_]+', 'Absolute Linux user path'), - (r'/tmp/', 'Temporary directory'), - (r'/var/', 'System directory'), - (r'/etc/', 'Configuration directory'), - - # Project-specific hardcoded paths - (r'/Users/fredbook/Code', 'Hardcoded UDOS_CODE path'), - (r'/Users/fredbook/Vault', 'Hardcoded UDOS_VAULT path'), - - # Common directories without env vars - (r'~/Library/', 'Home directory (should use os.path.expanduser)'), - (r'~/.', 'Home directory (should use os.path.expanduser)'), - (r'node_modules/', 'Node modules (should be in .gitignore)'), - (r'__pycache__/', 'Python cache (should be in .gitignore)'), - (r'\.venv/', 'Virtual environment (should be in .gitignore)'), -] - -# Environment variables that should be used instead of hardcoded paths -ENV_VAR_REPLACEMENTS = { - 'UDOS_CODE': 'Code repository root', - 'UDOS_VAULT': 'Vault directory', - 'UDOS_DATA_HOME': 'Data home directory', - 'UDOS_STATE_HOME': 'State home directory', - 'UDOS_CONFIG_HOME': 'Config home directory', - 'HOME': 'Home directory', - 'USER': 'Username', - 'TMPDIR': 'Temporary directory', -} - - -class HardcodedPathDetectorSkill(BaseSkill): - meta = SkillMeta( - id="hardcoded-path-detector", - name="Hardcoded Path Detector", - description=( - "Find hardcoded paths in Python/JS/TS files. " - "Reports absolute paths, environment variable references, " - "and platform-specific paths that should be configurable." - ), - category="developer", - timeout=60, - params=[ - SkillParam( - name="target", - type="string", - required=False, - default="backend/app", - description="Directory or file pattern to scan", - ), - SkillParam( - name="severity", - type="string", - required=False, - default="high", - description="Minimum severity: low, medium, high", - ), - ], - requires_confirmation=False, - ) - - async def run(self, **kwargs) -> dict: - target = str(kwargs.get("target", "backend/app")) - severity = str(kwargs.get("severity", "high")).lower() - - findings = self._scan_for_hardcoded_paths(target) - - # Filter by severity - filtered = self._filter_by_severity(findings, severity) - - return { - "success": True, - "action": "hardcoded_path_detector", - "findings": filtered, - "stats": { - "total_paths": len(findings), - "filtered": len(filtered), - "files_scanned": len(set(f["file"] for f in findings)), - }, - } - - def _scan_for_hardcoded_paths(self, target: str) -> list[dict]: - """Scan for hardcoded paths in source files.""" - from app.skills.shared_utils import get_source_files - - files = get_source_files(target) - findings = [] - - for file_path in files: - try: - content = file_path.read_text(encoding="utf-8", errors="replace") - lines = content.split("\n") - - for line_num, line in enumerate(lines, 1): - for pattern, description in HARDCODED_PATH_PATTERNS: - if re.search(pattern, line): - # Check if it's using an environment variable - is_env_ref = any( - f'os.environ.get("{var}"' in line or - f'os.getenv("{var}"' in line or - f'process.env.{var}' in line - for var in ENV_VAR_REPLACEMENTS.keys() - ) - - severity = self._determine_severity(pattern, line) - - findings.append({ - "file": str(file_path), - "line": line_num, - "line_content": line.strip(), - "pattern": pattern, - "description": description, - "severity": severity, - "is_env_ref": is_env_ref, - "recommendation": self._get_recommendation(pattern, line), - }) - break # Only report one pattern per line - except Exception as exc: - log.warning("Failed to scan %s: %s", file_path, exc) - continue - - return findings - - def _determine_severity(self, pattern: str, line: str) -> str: - """Determine severity level for a path pattern.""" - if "UDOS_CODE" in pattern or "UDOS_VAULT" in pattern: - return "high" - if "/Users/fredbook" in pattern: - return "high" - if "~/Library" in pattern or "~/." in pattern: - return "medium" - return "low" - - def _get_recommendation(self, pattern: str, line: str) -> str: - """Get recommendation for fixing a hardcoded path.""" - if "UDOS_CODE" in pattern: - return "Use os.environ.get('UDOS_CODE') instead" - if "UDOS_VAULT" in pattern: - return "Use os.environ.get('UDOS_VAULT') instead" - if "~/Library" in pattern or "~/." in pattern: - return "Use os.path.expanduser() instead" - if "/tmp/" in pattern: - return "Use tempfile.gettempdir() or os.environ.get('TMPDIR') instead" - return "Consider using environment variables or configuration files" - - def _filter_by_severity(self, findings: list[dict], severity: str) -> list[dict]: - """Filter findings by severity level.""" - severity_map = {"low": 0, "medium": 1, "high": 2} - min_level = severity_map.get(severity, 1) - - return [ - f for f in findings - if severity_map.get(f["severity"], 1) >= min_level - ] - - def _generate_report(self, findings: list[dict]) -> str: - """Generate markdown report of findings.""" - if not findings: - return "No hardcoded paths found at the specified severity level." - - lines = [ - "# Hardcoded Path Detector Report", - "", - f"**Total findings:** {len(findings)}", - "", - "## Summary by Severity", - "", - ] - - # Count by severity - severity_counts = {} - for f in findings: - severity = f["severity"] - severity_counts[severity] = severity_counts.get(severity, 0) + 1 - - for sev, count in sorted(severity_counts.items(), reverse=True): - lines.append(f"- **{sev.upper()}:** {count} paths") - - lines.extend(["", "## Findings", ""]) - - # Group by file - by_file: dict[str, list[dict]] = {} - for f in findings: - by_file.setdefault(f["file"], []).append(f) - - for file_path, file_findings in sorted(by_file.items()): - lines.append(f"### {file_path}") - lines.append("") - - for f in file_findings: - lines.append(f"- **Line {f['line']}**: {f['description']}") - lines.append(f" - Pattern: `{f['pattern']}`") - lines.append(f" - Severity: {f['severity'].upper()}") - if f['is_env_ref']: - lines.append(" - ⚠️ Already using environment variable") - else: - lines.append(f" - 💡 {f['recommendation']}") - lines.append("") - - return "\n".join(lines) diff --git a/backend/app/skills/builtin/skill_modularisation_planner.py b/backend/app/skills/builtin/skill_modularisation_planner.py deleted file mode 100644 index 33657329..00000000 --- a/backend/app/skills/builtin/skill_modularisation_planner.py +++ /dev/null @@ -1,418 +0,0 @@ -#!/usr/bin/env python3 -"""Modularisation Planner Skill — assess large scripts for improvement opportunities. - -Analyzes scripts longer than 1000 lines to identify: -- Large functions/methods (>100 lines) -- High cyclomatic complexity candidates -- Multiple responsibility violations -- Missing docstrings -- Import clustering opportunities -- Class extraction candidates -- Module splitting recommendations - -Generates: -- Refactoring priority scores -- Module extraction plans -- Complexity metrics -""" -from __future__ import annotations - -import ast -import json -import logging -import time -from collections import defaultdict -from pathlib import Path -from typing import Any - -from app.skills.base import BaseSkill, SkillMeta, SkillParam -from app.skills.shared_utils import get_source_files as _get_source_files - -log = logging.getLogger("ucore.skills.modularisation_planner") - -# Thresholds -MIN_LINES_FOR_ANALYSIS = 1000 -MAX_FUNCTION_LINES = 100 -MAX_CLASS_LINES = 300 -COMPLEXITY_THRESHOLD = 10 - - -class ModularisationPlannerSkill(BaseSkill): - meta = SkillMeta( - id="modularisation-planner", - name="Modularisation Planner", - description=( - "Assess large scripts (>1000 lines) for modularization opportunities. " - "Reports refactoring priorities and extraction plans." - ), - category="developer", - params=[ - SkillParam( - name="action", - type="string", - description="Action: 'scan', 'report', 'plan'", - required=True, - ), - SkillParam( - name="target", - type="string", - description="Optional: specific file or directory", - required=False, - ), - SkillParam( - name="min_lines", - type="int", - description="Minimum lines to consider for analysis", - required=False, - default=1000, - ), - SkillParam( - name="output_format", - type="string", - description="Output format: 'json', 'markdown'", - required=False, - default="json", - ), - ], - timeout=180, - ) - - async def run(self, **kwargs) -> dict: - action = kwargs.get("action", "scan") - target = kwargs.get("target", "") - min_lines = kwargs.get("min_lines", MIN_LINES_FOR_ANALYSIS) - output_format = kwargs.get("output_format", "json") - - if action == "scan": - return self._scan_large_scripts(target, min_lines) - if action == "report": - scan_result = self._scan_large_scripts(target, min_lines) - return { - "success": True, - "action": "report", - "report": scan_result, - "recommendations": self._generate_recommendations(scan_result), - } - if action == "plan": - scan_result = self._scan_large_scripts(target, min_lines) - return self._generate_plan(scan_result, output_format) - return {"success": False, "error": f"Unknown action: {action}"} - - def _scan_large_scripts(self, target: str, min_lines: int) -> dict[str, Any]: - """Scan for large scripts needing modularization.""" - findings: dict[str, list] = { - "large_files": [], - "large_functions": [], - "complex_functions": [], - "missing_docstrings": [], - "import_clusters": [], - "class_extraction_candidates": [], - } - stats = { - "files_scanned": 0, - "files_over_threshold": 0, - "total_issues": 0, - } - - source_files = self._get_source_files(target) - stats["files_scanned"] = len(source_files) - - for file_path in source_files: - try: - line_count = file_path.stat().st_size - # Quick line count - with open(file_path, 'r') as f: - line_count = sum(1 for _ in f) - - if line_count < min_lines: - continue - - stats["files_over_threshold"] += 1 - content = file_path.read_text() - - # Analyze the file - file_analysis = self._analyze_file(file_path, content, line_count) - findings["large_files"].append(file_analysis["summary"]) - - findings["large_functions"].extend(file_analysis["large_functions"]) - findings["complex_functions"].extend(file_analysis["complex_functions"]) - findings["missing_docstrings"].extend(file_analysis["missing_docstrings"]) - findings["import_clusters"].extend(file_analysis["import_clusters"]) - findings["class_extraction_candidates"].extend( - file_analysis["class_extraction_candidates"] - ) - - except Exception as e: - log.warning(f"Failed to analyze {file_path}: {e}") - continue - - stats["total_issues"] = sum(len(v) for v in findings.values()) - - return { - "success": True, - "findings": findings, - "stats": stats, - } - - def _get_source_files(self, target: str) -> list[Path]: - """Get all source files to scan.""" - return _get_source_files(target) - - def _analyze_file( - self, file_path: Path, content: str, line_count: int - ) -> dict[str, Any]: - """Analyze a single file for modularization opportunities.""" - analysis = { - "summary": { - "file": str(file_path), - "lines": line_count, - "priority_score": 0, - }, - "large_functions": [], - "complex_functions": [], - "missing_docstrings": [], - "import_clusters": [], - "class_extraction_candidates": [], - } - - # Parse Python files - if file_path.suffix == ".py": - try: - tree = ast.parse(content) - analysis["summary"]["type"] = "python" - - for node in ast.walk(tree): - if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): - func_lines = self._get_node_line_count(node) - if func_lines > MAX_FUNCTION_LINES: - analysis["large_functions"].append({ - "file": str(file_path), - "function": node.name, - "lines": func_lines, - "line": node.lineno, - }) - - complexity = self._calculate_complexity(node) - if complexity > COMPLEXITY_THRESHOLD: - analysis["complex_functions"].append({ - "file": str(file_path), - "function": node.name, - "complexity": complexity, - "line": node.lineno, - }) - - if not ast.get_docstring(node): - analysis["missing_docstrings"].append({ - "file": str(file_path), - "function": node.name, - "line": node.lineno, - }) - - elif isinstance(node, ast.ClassDef): - class_lines = self._get_node_line_count(node) - if class_lines > MAX_CLASS_LINES: - analysis["class_extraction_candidates"].append({ - "file": str(file_path), - "class": node.name, - "lines": class_lines, - "line": node.lineno, - }) - - # Check import clustering - analysis["import_clusters"] = self._analyze_imports(tree, file_path) - - except (SyntaxError, ValueError): - analysis["summary"]["type"] = "parse_error" - - # Calculate priority score - analysis["summary"]["priority_score"] = self._calculate_priority_score(analysis) - - return analysis - - def _get_node_line_count(self, node: ast.AST) -> int: - """Get the line count for an AST node.""" - if hasattr(node, "end_lineno") and node.end_lineno: - return node.end_lineno - node.lineno + 1 - return 0 - - def _calculate_complexity(self, node: ast.AST) -> int: - """Calculate cyclomatic complexity for a function.""" - complexity = 1 - for child in ast.walk(node): - if isinstance(child, (ast.If, ast.While, ast.For, ast.AsyncFor)): - complexity += 1 - elif isinstance(child, ast.ExceptHandler): - complexity += 1 - elif isinstance(child, (ast.And, ast.Or)): - complexity += 1 - return complexity - - def _analyze_imports(self, tree: ast.AST, file_path: Path) -> list[dict]: - """Analyze import patterns for clustering opportunities.""" - clusters = [] - import_groups: dict[str, list] = defaultdict(list) - - for node in ast.walk(tree): - if isinstance(node, ast.Import): - for alias in node.names: - top_level = alias.name.split(".")[0] - import_groups[top_level].append(alias.name) - elif isinstance(node, ast.ImportFrom): - if node.module: - top_level = node.module.split(".")[0] - import_groups[top_level].append(node.module) - - # Find modules with many imports (potential extraction candidates) - for module, imports in import_groups.items(): - if len(imports) > 5: - clusters.append({ - "file": str(file_path), - "module": module, - "imports": imports, - "count": len(imports), - }) - - return clusters - - def _calculate_priority_score(self, analysis: dict) -> int: - """Calculate refactoring priority score (0-100).""" - score = 0 - - # Large functions add 20 points each (max 40) - score += min(len(analysis["large_functions"]) * 20, 40) - - # Complex functions add 15 points each (max 30) - score += min(len(analysis["complex_functions"]) * 15, 30) - - # Missing docstrings add 5 points each (max 15) - score += min(len(analysis["missing_docstrings"]) * 5, 15) - - # Import clusters add 10 points each (max 10) - score += min(len(analysis["import_clusters"]) * 10, 10) - - # Class extraction candidates add 10 points each (max 10) - score += min(len(analysis["class_extraction_candidates"]) * 10, 10) - - return min(score, 100) - - def _generate_recommendations(self, scan_result: dict) -> list[str]: - """Generate recommendations based on scan results.""" - recs = [] - findings = scan_result.get("findings", {}) - - large_files = findings.get("large_files", []) - if large_files: - sorted_files = sorted( - large_files, key=lambda x: x.get("priority_score", 0), reverse=True - ) - top_files = sorted_files[:3] - recs.append( - f"Top priority files for modularization: " - f"{', '.join(f['file'].split('/')[-1] for f in top_files)}" - ) - - large_funcs = findings.get("large_functions", []) - if large_funcs: - recs.append(f"Found {len(large_funcs)} large functions (>100 lines) — consider splitting") - - complex_funcs = findings.get("complex_functions", []) - if complex_funcs: - recs.append(f"Found {len(complex_funcs)} complex functions — consider refactoring") - - return recs - - def _generate_plan(self, scan_result: dict, output_format: str) -> dict: - """Generate a modularization plan.""" - findings = scan_result.get("findings", {}) - large_files = findings.get("large_files", []) - - # Sort by priority - sorted_files = sorted( - large_files, key=lambda x: x.get("priority_score", 0), reverse=True - ) - - plan = { - "timestamp": int(time.time()), - "files_analyzed": len(large_files), - "priority_order": [], - "extraction_suggestions": [], - } - - for file_info in sorted_files: - plan["priority_order"].append({ - "file": file_info["file"], - "lines": file_info["lines"], - "priority_score": file_info["priority_score"], - }) - - # Generate extraction suggestions - suggestions = self._generate_extraction_suggestions(file_info["file"]) - plan["extraction_suggestions"].extend(suggestions) - - if output_format == "markdown": - plan["markdown"] = self._to_markdown(plan) - - return { - "success": True, - "action": "plan", - "plan": plan, - } - - def _generate_extraction_suggestions(self, file_path: str) -> list[dict]: - """Generate specific extraction suggestions for a file.""" - suggestions = [] - - try: - content = Path(file_path).read_text() - tree = ast.parse(content) - - # Suggest extracting large functions - for node in ast.walk(tree): - if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): - func_lines = self._get_node_line_count(node) - if func_lines > MAX_FUNCTION_LINES: - suggestions.append({ - "file": file_path, - "type": "function_extraction", - "target": node.name, - "lines": func_lines, - "suggestion": f"Extract '{node.name}' to separate module", - }) - - elif isinstance(node, ast.ClassDef): - class_lines = self._get_node_line_count(node) - if class_lines > MAX_CLASS_LINES: - suggestions.append({ - "file": file_path, - "type": "class_extraction", - "target": node.name, - "lines": class_lines, - "suggestion": f"Split '{node.name}' class into smaller modules", - }) - - except Exception: - pass - - return suggestions - - def _to_markdown(self, plan: dict) -> str: - """Convert plan to markdown format.""" - md = ["# Modularization Plan\n"] - md.append(f"Generated: {time.strftime('%Y-%m-%d %H:%M:%S')}\n") - md.append(f"Files analyzed: {plan['files_analyzed']}\n\n") - - md.append("## Priority Order\n") - for i, file_info in enumerate(plan["priority_order"], 1): - md.append( - f"{i}. `{file_info['file'].split('/')[-1]}` " - f"({file_info['lines']} lines, score: {file_info['priority_score']})\n" - ) - - md.append("\n## Extraction Suggestions\n") - for suggestion in plan["extraction_suggestions"]: - md.append( - f"- **{suggestion['type']}**: {suggestion['suggestion']} " - f"({suggestion['lines']} lines)\n" - ) - - return "".join(md) diff --git a/backend/app/skills/builtin/skill_surface_registry.py b/backend/app/skills/builtin/skill_surface_registry.py deleted file mode 100644 index 1e646c95..00000000 --- a/backend/app/skills/builtin/skill_surface_registry.py +++ /dev/null @@ -1,687 +0,0 @@ -"""Surface Registry Skill — discover, validate, scaffold, repair, wire. - -Autonomous maintenance for the uCore surface ecosystem. -Manages frontend Vue surfaces and their backend wiring -(runtimes, variables, commands). - -Actions: - discover — scan filesystem for registered/detached surfaces - validate — check surface compliance (USX, routing, backend wiring) - scaffold — generate new surface from template - repair — auto-fix routing, imports, tab registration gaps - wire — link surface to backend runtime/variables/commands - report — full ecosystem health report -""" -from __future__ import annotations - -import logging -import re -from pathlib import Path - -from app.skills.base import BaseSkill, SkillMeta, SkillParam - -log = logging.getLogger("ucore.skills.surface_registry") - -ROOT_DIR = Path(__file__).parent.parent.parent.parent.parent -FRONTEND_DIR = ROOT_DIR / "frontend-vue" / "src" -SURFACES_DIR = FRONTEND_DIR / "surfaces" -ROUTER_FILE = FRONTEND_DIR / "router" / "index.ts" -DEV_STORE_FILE = FRONTEND_DIR / "stores" / "developer.ts" -DEV_SURFACE_FILE = SURFACES_DIR / "developer" / "DeveloperSurface.vue" -BACKEND_DIR = ROOT_DIR / "backend" / "app" -RUNTIMES_FILE = BACKEND_DIR / "services" / "dev_layer.py" -MANIFEST_FILE = ROOT_DIR / "seeds" / "surface-registry.json" - -EXCLUDES = {"node_modules", ".git", "__pycache__", "dist", "panels", "archive"} - - -class SurfaceRegistrySkill(BaseSkill): - meta = SkillMeta( - id="surface-registry", - name="Surface Registry v1", - description=( - "Discover, validate, scaffold, repair, and wire uCore" - " surfaces. Autonomous maintenance for the surface" - " ecosystem with backend runtime linking." - ), - category="developer", - params=[ - SkillParam( - name="action", - type="string", - description=( - "Action: 'discover', 'validate', 'scaffold'," - " 'repair', 'wire', 'report'" - ), - required=True, - ), - SkillParam( - name="target", - type="string", - description="Surface name (e.g., 'developer', 'workflow')", - required=False, - ), - SkillParam( - name="backend_runtime", - type="string", - description=( - "Backend runtime to wire (e.g., 'dev_layer'," - " 'feed_server')" - ), - required=False, - ), - SkillParam( - name="dry_run", - type="bool", - description="Preview changes without applying", - required=False, - ), - ], - timeout=120, - requires_confirmation=True, - ) - - async def run(self, **kwargs) -> dict: - action = kwargs.get("action", "discover") - target = kwargs.get("target", "") - backend_runtime = kwargs.get("backend_runtime", "") - dry_run = kwargs.get("dry_run", False) - - actions = { - "discover": lambda: self._discover(), - "validate": lambda: self._validate(target), - "scaffold": lambda: self._scaffold(target), - "repair": lambda: self._repair(target, dry_run), - "wire": lambda: self._wire(target, backend_runtime, dry_run), - "report": lambda: self._report(), - } - - handler = actions.get(action) - if handler is None: - return {"success": False, "error": f"Unknown action: {action}"} - return handler() - - # ─── Discover ──────────────────────────────────────────────────── - - def _discover(self) -> dict: - """Scan filesystem for all surfaces and their registration state.""" - registered = self._parse_router_surfaces() - filesystem = self._scan_filesystem_surfaces() - tabs = self._parse_developer_tabs() - - # Detect detached surfaces (in filesystem but not in router) - detached = [s for s in filesystem if s not in registered] - - # Detect phantom surfaces (in router but not in filesystem) - phantom = [s for s in registered if s not in filesystem] - - # Detect untabbed surfaces (in router but no developer tab) - untabbed = [s for s in registered if s not in tabs] - - # Detect backend-wired surfaces - wired = self._detect_backend_wiring(registered) - - return { - "success": True, - "action": "discover", - "total_registered": len(registered), - "total_filesystem": len(filesystem), - "surfaces": registered, - "detached": detached, - "phantom": phantom, - "untabbed": untabbed, - "wired_backend": wired, - "health": ( - "healthy" if not detached and not phantom and not untabbed - else "attention" - ), - } - - # ─── Validate ──────────────────────────────────────────────────── - - def _validate(self, target: str = "") -> dict: - """Validate surface compliance: USX, routing, backend wiring.""" - surfaces = ( - [target] if target - else self._parse_router_surfaces() - ) - results = {} - - for surface in surfaces: - sf_name = f"{self._pascal_case(surface)}Surface.vue" - surface_file = SURFACES_DIR / surface / sf_name - - issues = [] - warnings = [] - - # Check filesystem presence - if not surface_file.exists(): - issues.append("Surface Vue file not found on disk") - results[surface] = { - "exists": False, - "issues": issues, - "clean": False, - } - continue - - # Check router registration - rc = ROUTER_FILE.read_text() if ROUTER_FILE.exists() else "" - if surface.lower() not in rc.lower(): - issues.append("Not registered in router/index.ts") - - # Check USX compliance via surface audit - content = surface_file.read_text() - usx_issues = self._audit_surface_usx(content) - issues.extend(usx_issues) - - # Check panel imports in DeveloperSurface.vue - dev_content = ( - DEV_SURFACE_FILE.read_text() - if DEV_SURFACE_FILE.exists() else "" - ) - if surface.lower() not in dev_content.lower(): - warnings.append( - "No panel registered in DeveloperSurface.vue", - ) - - # Check backend wiring - wiring = self._check_surface_wiring(surface) - if not wiring.get("wired"): - warnings.append("No backend runtime wiring detected") - else: - results.setdefault("_wiring", {})[surface] = wiring - - clean = len(issues) == 0 - results[surface] = { - "exists": True, - "issues": issues, - "warnings": warnings, - "clean": clean, - } - - # Aggregate - all_clean = all( - r.get("clean", False) - for r in results.values() - if isinstance(r, dict) and "clean" in r - ) - - return { - "success": True, - "action": "validate", - "surfaces": results, - "all_clean": all_clean, - "total": len(results), - } - - # ─── Scaffold ──────────────────────────────────────────────────── - - def _scaffold(self, name: str) -> dict: - """Generate surface: Vue file, router entry, dev tab, panel.""" - if not name: - return {"success": False, "error": "surface name required"} - if not re.match(r"^[a-z][a-z0-9-]*[a-z0-9]$", name): - return { - "success": False, - "error": ( - f"Invalid surface name '{name}'. Use lowercase" - " kebab-case (e.g., 'my-feature')" - ), - } - - pascal = self._pascal_case(name) - route_path = f"/{name}" - route_name = name.replace("-", "") - - # Surface Vue template - vue_template = ( - f'\n' - f'\n' - f'\n' - f'\n' - f'\n' - ) - - # Panel template (for DeveloperSurface tab) - panel_template = ( - f'\n' - f'\n' - f'\n' - ) - - # Router entry - router_entry = ( - f" {{\n" - f" path: '{route_path}/:pathMatch(.*)*',\n" - f" name: '{route_name}',\n" - f" component: () => import(\n" - f" '../surfaces/{name}/{pascal}Surface.vue'" - f"\n ),\n" - f" meta: {{ title: '{pascal}', icon: 'widgets' }},\n" - f" }}," - ) - - # Dev store tab entry - dev_tab_entry = ( - f" {{ id: '{name}', label: '{pascal}'," - f" icon: 'widgets' }}," - ) - - # Files to create - files = { - f"frontend-vue/src/surfaces/{name}/" - f"{pascal}Surface.vue": vue_template, - f"frontend-vue/src/surfaces/{name}/" - f"panels/{pascal}Panel.vue": panel_template, - } - - return { - "success": True, - "action": "scaffold", - "name": name, - "component": f"{pascal}Surface", - "panel": f"{pascal}Panel", - "route": route_path, - "files_to_create": files, - "router_entry": router_entry, - "dev_tab_entry": dev_tab_entry, - "instructions": [ - "1. Write the surface Vue file(s) to disk", - "2. Add the router entry to router/index.ts routes array", - ( - "3. Add the dev tab entry to" - " stores/developer.ts DEVELOPER_TABS" - ), - "4. Add to DeveloperTab union type in stores/developer.ts", - "5. Import and wire panel in DeveloperSurface.vue", - "6. Run 'wire' action to link backend runtime", - "7. Run 'validate' action to confirm compliance", - ], - } - - # ─── Repair ────────────────────────────────────────────────────── - - def _repair(self, target: str = "", dry_run: bool = False) -> dict: - """Auto-fix routing, imports, tab registration gaps.""" - repairs = { - "routing_fixed": 0, - "tabs_fixed": 0, - "total_fixed": 0, - } - issues = [] - - if not ROUTER_FILE.exists(): - return { - "success": False, - "error": "Router or dev store file not found", - "repairs": repairs, - } - - registered = self._parse_router_surfaces() - tabs = self._parse_developer_tabs() - filesystem = self._scan_filesystem_surfaces() - - # Fix: surfaces in filesystem but not in router - detached = [s for s in filesystem if s not in registered] - for surface in detached: - if target and surface != target: - continue - issues.append( - f"Surface '{surface}' in filesystem but not in router", - ) - repairs["routing_fixed"] += 1 - - # Fix: surfaces in router but no dev tab - untabbed = [s for s in registered if s not in tabs] - for surface in untabbed: - if target and surface != target: - continue - issues.append( - f"Surface '{surface}' in router but no dev tab", - ) - repairs["tabs_fixed"] += 1 - - repairs["total_fixed"] = ( - repairs["routing_fixed"] + repairs["tabs_fixed"] - ) - - return { - "success": True, - "action": "repair", - "dry_run": dry_run, - "repairs": repairs, - "issues_found": issues, - "note": ( - "Auto-repair provides detection. Manual" - " insertion required for router entries and" - " tab registrations to avoid merge conflicts." - ), - } - - # ─── Wire ──────────────────────────────────────────────────────── - - def _wire( - self, target: str = "", backend_runtime: str = "", - dry_run: bool = False, - ) -> dict: - """Wire a frontend surface to backend runtime/variables/commands.""" - if not target: - return { - "success": False, - "error": "target surface name required", - } - - # Discover available backend handlers - runtimes = self._discover_backend_runtimes() - - # If backend_runtime specified, validate it - if backend_runtime and backend_runtime not in runtimes: - return { - "success": False, - "error": ( - f"Unknown backend runtime '{backend_runtime}'." - f" Available: {list(runtimes.keys())}" - ), - } - - # Build wiring manifest - wiring = { - "surface": target, - "runtime": backend_runtime or "default", - "endpoints": {}, - "variables": {}, - "commands": {}, - } - - if backend_runtime and backend_runtime in runtimes: - rt = runtimes[backend_runtime] - wiring["endpoints"] = rt.get("endpoints", {}) - wiring["variables"] = rt.get("variables", {}) - wiring["commands"] = rt.get("commands", {}) - - return { - "success": True, - "action": "wire", - "target": target, - "wiring": wiring, - "available_runtimes": list(runtimes.keys()), - "instructions": [ - ( - "1. Add API client calls in the surface Vue" - " component to the wired endpoints" - ), - ( - "2. Create a Pinia store for the surface's" - " state if it needs persistent data" - ), - "3. Register the store in the surface's setup()", - ( - "4. Add backend variables to the surface's" - " reactive state" - ), - ( - "5. Expose commands as methods callable from" - " the surface UI" - ), - ( - "6. Update surface-registry.json manifest" - " with the wiring" - ), - ], - } - - # ─── Report ────────────────────────────────────────────────────── - - def _report(self) -> dict: - discover = self._discover() - validate = self._validate() - runtimes = self._discover_backend_runtimes() - - return { - "success": True, - "action": "report", - "ecosystem": { - "total_surfaces": discover["total_registered"], - "health": discover["health"], - "detached": discover["detached"], - "phantom": discover["phantom"], - "untabbed": discover["untabbed"], - "wired_backend": discover["wired_backend"], - }, - "validation": { - "all_clean": validate["all_clean"], - "total_issues": sum( - len(r.get("issues", [])) + len(r.get("warnings", [])) - for r in validate["surfaces"].values() - if isinstance(r, dict) - ), - }, - "backends_available": list(runtimes.keys()), - "recommendations": [ - ( - "Run 'scaffold' to generate new surfaces" - " from templates" - ), - "Run 'validate' after any surface changes", - "Run 'wire' to connect surfaces to backend runtimes", - "Check detached surfaces and register or clean them up", - "Check phantom surfaces and create their Vue files", - ], - } - - # ─── Helpers ───────────────────────────────────────────────────── - - def _parse_router_surfaces(self) -> list[str]: - """Extract surface names from router/index.ts.""" - if not ROUTER_FILE.exists(): - return [] - content = ROUTER_FILE.read_text() - # Match: component: () => import("../surfaces/NAME/NameSurface.vue") - # Supports both single and double quotes - matches = re.findall( - r"""import\(['"]\.\./surfaces/([a-z][a-z0-9-]*)/""", - content, - ) - return sorted(set(matches)) - - def _scan_filesystem_surfaces(self) -> list[str]: - """List surface directories on disk.""" - if not SURFACES_DIR.exists(): - return [] - surfaces = [] - for d in SURFACES_DIR.iterdir(): - if d.is_dir() and d.name not in EXCLUDES: - # Check for a Surface.vue file - vue_files = list(d.glob("*Surface.vue")) - if vue_files: - surfaces.append(d.name) - return sorted(surfaces) - - def _parse_developer_tabs(self) -> list[str]: - """Extract tab IDs from developer store.""" - if not DEV_STORE_FILE.exists(): - return [] - content = DEV_STORE_FILE.read_text() - # Match: { id: 'tabname', ... } - matches = re.findall( - r"\{\s*id:\s*'([a-z][a-z0-9-]*)'", - content, - ) - return sorted(set(matches)) - - def _detect_backend_wiring(self, surfaces: list[str]) -> list[str]: - """Detect surfaces with known backend runtime connections.""" - known_runtimes = { - "developer": ["dev_layer", "control_service"], - "server": ["hivemind_server", "llm_router"], - "workflow": ["task_processor"], - "snackmachine": ["snackmachine"], - "assistui": ["assistui_runtime"], - "documentation": ["documentation_api"], - "terminal": ["terminal_runtime"], - "ucode": ["ucode_runtime"], - "browserui": ["browser_runtime"], - "teletext": ["teletext_runtime"], - "dashboard": ["dashboard_runtime"], - "system": ["system_runtime"], - "feed": ["feed_server", "feed_consumer"], - } - wired = [] - for s in surfaces: - if s in known_runtimes: - wired.append(s) - return wired - - def _discover_backend_runtimes(self) -> dict: - """Scan backend for available runtime services.""" - runtimes = {} - - # Known backend runtime modules - candidate_modules = { - "dev_layer": BACKEND_DIR / "services" / "dev_layer.py", - "feed_server": BACKEND_DIR / "mcp" / "feed" / "feed_server.py", - "feed_consumer": BACKEND_DIR / "services" / "feed_consumer.py", - "hivemind_server": BACKEND_DIR / "mcp" / "hivemind_server.py", - "llm_router": BACKEND_DIR / "mcp" / "llm_router.py", - "model_pricing": BACKEND_DIR / "services" / "model_pricing.py", - "template_manager": ( - BACKEND_DIR / "services" / "template_manager.py" - ), - } - - for name, path in candidate_modules.items(): - if not path.exists(): - continue - content = path.read_text() - - # Extract endpoints (API routes, MCP tools) - endpoints = [] - for match in re.findall( - r'"([a-z_]+)"\s*:\s*self\._[a-z_]+', content, - ): - endpoints.append(match) - - # Extract MCP tool names - for match in re.findall( - r'name="([a-z_]+)"', content, - ): - endpoints.append(match) - - # Extract class variables from __init__ - variables = {} - for match in re.findall( - r"self\.([a-z_]+)\s*=\s*([^#\n]+)", content, - ): - key, val = match - if len(key) > 2 and not key.startswith("_"): - variables[key] = val.strip().rstrip(",") - - # Extract public methods as commands - commands = [] - for match in re.findall( - r"async def ([a-z_]+)\(self", content, - ): - if not match.startswith("_"): - commands.append(match) - for match in re.findall( - r"def ([a-z_]+)\(self", content, - ): - if not match.startswith("_"): - commands.append(match) - - runtimes[name] = { - "file": str(path.relative_to(ROOT_DIR)), - "endpoints": sorted(set(endpoints)), - "variables": variables, - "commands": sorted(set(commands)), - } - - return runtimes - - def _check_surface_wiring(self, surface: str) -> dict: - """Check if a surface has backend wiring.""" - wired = self._detect_backend_wiring([surface]) - return { - "wired": len(wired) > 0, - "runtimes": (self._detect_backend_wiring([surface])), - } - - def _audit_surface_usx(self, content: str) -> list[str]: - """Quick USX compliance check for a surface file.""" - issues = [] - - # Check for hardcoded colors in style blocks - style_blocks = re.findall( - r"]*>(.*?)", content, re.DOTALL, - ) - for i, block in enumerate(style_blocks): - hex_colors = re.findall(r"#[0-9a-fA-F]{3,8}", block) - for h in hex_colors: - issues.append( - f"", content, re.DOTALL, - ) - for i, block in enumerate(style_blocks): - for hw_type, pattern in VARIABLE_MAP.items(): - matches = re.findall(pattern, block) - for m in matches: - issues.append( - f" -""" - - return { - "success": True, - "action": "scaffold-surface", - "name": name, - "component": f"{pascal}Surface", - "path": str(file_path.relative_to(FRONTEND_DIR.parent.parent)), - "template": template, - "note": ( - "This is a preview. Use ACT MODE to write the file" - " to disk." - ), - } - - # ─── Repair ────────────────────────────────────────────────────── - - def _repair(self, target: str = "", dry_run: bool = False) -> dict: - """Apply automatic repairs: replace hardcoded values with variables.""" - repairs = { - "colors_fixed": 0, - "font_sizes_fixed": 0, - "spacing_fixed": 0, - "radius_fixed": 0, - "files_touched": 0, - } - touched = set() - - # Color replacement pairs (heuristic — exact matches only) - color_replacements = [ - # Primary - (r"color:\s*#0d6efd", "color: var(--usx-color-primary)"), - (r"color:\s*#0b5ed7", "color: var(--usx-color-primary-hover)"), - (r"background:\s*#0d6efd", "background: var(--usx-color-primary)"), - (r"color:\s*#ffffff", "color: var(--usx-color-on-primary)"), - # Surface - ( - r"background:\s*#f8f9fa", - "background: var(--usx-color-surface-hover)", - ), - ( - r"background:\s*#e9ecef", - "background: var(--usx-color-surface-active)", - ), - (r"color:\s*#212529", "color: var(--usx-color-on-surface)"), - (r"color:\s*#6c757d", "color: var(--usx-color-on-surface-muted)"), - # Status - (r"color:\s*#198754", "color: var(--usx-color-success)"), - (r"color:\s*#dc3545", "color: var(--usx-color-danger)"), - (r"color:\s*#ffc107", "color: var(--usx-color-warning)"), - (r"color:\s*#0dcaf0", "color: var(--usx-color-info)"), - # Border - (r"border-color:\s*#dee2e6", ( - "border-color: var(--usx-color-border)" - )), - ] - - files = self._get_style_files(target) - for f in files: - content = f.read_text() - original = content - - for pattern, replacement in color_replacements: - new_content = re.sub(pattern, replacement, content) - if new_content != content: - repairs["colors_fixed"] += 1 - content = new_content - - if content != original: - touched.add(str(f)) - if not dry_run: - f.write_text(content) - - repairs["files_touched"] = len(touched) - return { - "success": True, - "action": "repair", - "dry_run": dry_run, - "repairs": repairs, - "touched_files": list(touched), - } - - # ─── Report ────────────────────────────────────────────────────── - - def _report(self, target: str = "") -> dict: - audit = self._deep_audit(target) - token_validation = self._validate_tokens() - - return { - "success": True, - "action": "report", - "audit": audit.get("findings", {}), - "audit_stats": audit.get("stats", {}), - "token_validation": token_validation, - "recommendations": [ - "Zero hardcoded values — use var(--usx-*) for everything", - "Check tokens/tokens-*.css for available variables", - "Import themes in index.html: ", - "Use useTheme() composable for runtime theme switching", - "Run 'validate-tokens' before every push to catch drift", - "Use .surface__* BEM classes from usx-standard.css for layout", - "All interactive elements must use var(--usx-touch-min)", - ], - } - - # ─── Helpers ───────────────────────────────────────────────────── - - def _get_style_files(self, target: str = "") -> list[Path]: - """Get all CSS and Vue files in the frontend source tree.""" - if not FRONTEND_DIR.exists(): - return [] - patterns = ["*.css", "*.vue"] - files: list[Path] = [] - for pattern in patterns: - if target: - files.extend( - FRONTEND_DIR.rglob(f"*{target}*{pattern}"), - ) - else: - files.extend(FRONTEND_DIR.rglob(pattern)) - return [ - f for f in files - if not any(x in str(f) for x in EXCLUDES) - ] - - def _find_hardcoded(self, content: str, issue_type: str) -> list[str]: - """Find hardcoded values of a given type in content.""" - pattern = VARIABLE_MAP.get(issue_type) - if not pattern: - return [] - lines = content.split("\n") - results = [] - in_var_block = False - for i, line in enumerate(lines): - stripped = line.strip() - # Skip variable declaration blocks - if "[data-theme" in stripped or ":root" in stripped: - in_var_block = True - continue - if in_var_block: - if stripped == "}" or stripped.startswith("}"): - in_var_block = False - continue - # Skip lines that already use var() - if "var(" in stripped: - continue - if re.search(pattern, stripped, re.IGNORECASE): - results.append(f"L{i + 1}: {stripped.strip()}") - return results diff --git a/backend/app/skills/catalogue.json b/backend/app/skills/catalogue.json new file mode 100644 index 00000000..13bf2084 --- /dev/null +++ b/backend/app/skills/catalogue.json @@ -0,0 +1,32 @@ +{ + "version": 1, + "modules": [ + {"module": "ask_vault.py", "owner": "uKnowledge", "lifecycle": "active", "risk": "read", "lane": "knowledge", "allowed_roots": ["Vault", "Shared", "Public"]}, + {"module": "attach_context.py", "owner": "uCore", "lifecycle": "active", "risk": "read", "lane": "assist", "allowed_roots": ["Code", "Vault"]}, + {"module": "backup.py", "owner": "uCore", "lifecycle": "active", "risk": "write", "lane": "maintenance", "allowed_roots": ["UDOS_HOME", "Vault", "Shared", "Public"]}, + {"module": "brain_sync.py", "owner": "uCore", "lifecycle": "review", "risk": "write", "lane": "memory", "allowed_roots": ["UDOS_HOME", "Vault"]}, + {"module": "clipboard_maintenance.py", "owner": "uCore", "lifecycle": "active", "risk": "write", "lane": "maintenance", "allowed_roots": ["UDOS_HOME"]}, + {"module": "docs_mirror_sync.py", "owner": "uKnowledge", "lifecycle": "review", "risk": "write", "lane": "knowledge", "allowed_roots": ["UDOS_HOME", "Code"]}, + {"module": "episodic_log.py", "owner": "uCore", "lifecycle": "review", "risk": "write", "lane": "memory", "allowed_roots": ["UDOS_HOME"]}, + {"module": "git_maintenance.py", "owner": "uCore", "lifecycle": "active", "risk": "write", "lane": "developer", "allowed_roots": ["Code"]}, + {"module": "lint_fix.py", "owner": "uCore", "lifecycle": "active", "risk": "write", "lane": "developer", "allowed_roots": ["Code"]}, + {"module": "route_task.py", "owner": "uCore", "lifecycle": "review", "risk": "external", "lane": "assist", "allowed_roots": ["UDOS_HOME"]}, + {"module": "skill_audit.py", "owner": "uCore", "lifecycle": "replace", "risk": "read", "lane": "system", "allowed_roots": ["Code"]}, + {"module": "skill_autostart.py", "owner": "uCore", "lifecycle": "remove", "risk": "write", "lane": "system", "allowed_roots": ["UDOS_HOME"]}, + {"module": "skill_dev_destroy_rebuild.py", "owner": "uCore", "lifecycle": "privileged", "risk": "destructive", "lane": "recovery", "allowed_roots": ["UDOS_HOME", "Code"]}, + {"module": "skill_dev_mode_executor.py", "owner": "uCore", "lifecycle": "replace", "risk": "external", "lane": "developer", "allowed_roots": ["Code", "UDOS_HOME"]}, + {"module": "skill_ecosystem_audit.py", "owner": "uCore", "lifecycle": "replace", "risk": "read", "lane": "system", "allowed_roots": ["Code", "UDOS_HOME"]}, + {"module": "skill_hivemind_consensus.py", "owner": "uCore", "lifecycle": "review", "risk": "external", "lane": "assist", "allowed_roots": ["UDOS_HOME"]}, + {"module": "skill_mcp_self_heal.py", "owner": "uCore", "lifecycle": "merge", "risk": "write", "lane": "recovery", "allowed_roots": ["Code", "UDOS_HOME"]}, + {"module": "skill_nuggets_and_spool.py", "owner": "uCore", "lifecycle": "split", "risk": "destructive", "lane": "recovery", "allowed_roots": ["UDOS_HOME"]}, + {"module": "skill_roundtable_dispatch.py", "owner": "uCore", "lifecycle": "review", "risk": "external", "lane": "assist", "allowed_roots": ["UDOS_HOME"]}, + {"module": "skill_self_heal.py", "owner": "uCore", "lifecycle": "merge", "risk": "destructive", "lane": "recovery", "allowed_roots": ["Code", "UDOS_HOME"]}, + {"module": "skill_ucore_index.py", "owner": "uCore", "lifecycle": "replace", "risk": "read", "lane": "system", "allowed_roots": ["Code", "UDOS_HOME"]}, + {"module": "skill_vault_discovery.py", "owner": "uKnowledge", "lifecycle": "move", "risk": "read", "lane": "knowledge", "allowed_roots": ["Vault", "Shared", "Public"]}, + {"module": "tasker_sync.py", "owner": "uFlow", "lifecycle": "move", "risk": "write", "lane": "workflow", "allowed_roots": ["UDOS_HOME"]}, + {"module": "vault_sync.py", "owner": "uKnowledge", "lifecycle": "move", "risk": "write", "lane": "knowledge", "allowed_roots": ["Vault", "Shared", "Public", "UDOS_HOME"]}, + {"module": "workflow_audit.py", "owner": "uFlow", "lifecycle": "move", "risk": "read", "lane": "workflow", "allowed_roots": ["UDOS_HOME"]}, + {"module": "workflow_guard.py", "owner": "uFlow", "lifecycle": "move", "risk": "write", "lane": "workflow", "allowed_roots": ["UDOS_HOME"]}, + {"module": "workflow_pause.py", "owner": "uFlow", "lifecycle": "move", "risk": "write", "lane": "workflow", "allowed_roots": ["UDOS_HOME"]} + ] +} diff --git a/backend/app/skills/registry.py b/backend/app/skills/registry.py index 128d28d4..165fdaa4 100644 --- a/backend/app/skills/registry.py +++ b/backend/app/skills/registry.py @@ -2,6 +2,7 @@ import importlib.util import inspect +import json import logging import sys from pathlib import Path @@ -12,6 +13,29 @@ _registry: dict[str, BaseSkill] = {} _loaded = False BUILTIN_SKILL_PATH = Path(__file__).parent / "builtin" +CATALOGUE_PATH = Path(__file__).parent / "catalogue.json" + + +def _catalogue_modules() -> list[dict]: + data = json.loads(CATALOGUE_PATH.read_text(encoding="utf-8")) + modules = data.get("modules") + if data.get("version") != 1 or not isinstance(modules, list): + raise RuntimeError("Invalid internal capability catalogue") + required = {"module", "owner", "lifecycle", "risk", "lane", "allowed_roots"} + for item in modules: + if not isinstance(item, dict) or required - set(item): + raise RuntimeError("Incomplete internal capability catalogue entry") + names = [item["module"] for item in modules] + if len(names) != len(set(names)): + raise RuntimeError("Duplicate internal capability catalogue module") + discovered = sorted( + path.name + for path in BUILTIN_SKILL_PATH.glob("*.py") + if not path.name.startswith("_") + ) + if sorted(names) != discovered: + raise RuntimeError("Internal capability catalogue does not match builtin modules") + return modules def _discover(): @@ -20,9 +44,8 @@ def _discover(): if not sd.exists(): return skills sys.path.insert(0, str(sd.parent)) - for f in sd.iterdir(): - if f.suffix != ".py" or f.name.startswith("_"): - continue + for entry in _catalogue_modules(): + f = sd / entry["module"] try: spec = importlib.util.spec_from_file_location(f"skills_{f.stem}", f) if spec and spec.loader: diff --git a/backend/tests/test_skill_registry_discovery_policy.py b/backend/tests/test_skill_registry_discovery_policy.py index e4d96e64..aa81c495 100644 --- a/backend/tests/test_skill_registry_discovery_policy.py +++ b/backend/tests/test_skill_registry_discovery_policy.py @@ -1,5 +1,7 @@ from __future__ import annotations +import json + from app.skills import registry @@ -13,7 +15,19 @@ def test_registry_does_not_scan_runtime_user_python(monkeypatch, tmp_path): ) builtin = tmp_path / "builtin" builtin.mkdir() + catalogue = tmp_path / "catalogue.json" + catalogue.write_text(json.dumps({"version": 1, "modules": []}), encoding="utf-8") monkeypatch.setattr(registry, "BUILTIN_SKILL_PATH", builtin) + monkeypatch.setattr(registry, "CATALOGUE_PATH", catalogue) assert registry._discover() == {} assert not marker.exists() + + +def test_catalogue_covers_every_builtin_module(): + entries = registry._catalogue_modules() + + assert entries + assert all(entry["owner"] for entry in entries) + assert all(entry["risk"] for entry in entries) + assert all(isinstance(entry["allowed_roots"], list) for entry in entries) diff --git a/docs/DEVMODE_CODE_ANALYSIS_SKILLS.md b/docs/DEVMODE_CODE_ANALYSIS_SKILLS.md deleted file mode 100644 index 0fa0eb22..00000000 --- a/docs/DEVMODE_CODE_ANALYSIS_SKILLS.md +++ /dev/null @@ -1,149 +0,0 @@ -> **Canonical version:** `/Users/fredbook/Code/uDocs/guides/devmode-skills.md` -> This repo copy is kept for local reference; edits should be made in uDocs. - -# DevMode Code Analysis Skills - -Three new skills for code quality analysis and modularization planning. - -## Skills Overview - -| Skill ID | Name | Purpose | -|----------|------|---------| -| `duplicate-detector` | Duplicate Code Detector | Find duplicate functions, similar blocks, repeated imports | -| `dead-code-archiver` | Dead Code / Legacy Archiver | Identify unused code, legacy patterns, orphaned imports | -| `modularisation-planner` | Modularisation Planner | Assess large scripts (>1000 lines) for refactoring | - ---- - -## 1. Duplicate Code Detector (`duplicate-detector`) - -### Actions -- `scan` — Scan for duplicate code patterns -- `report` — Scan + generate recommendations -- `archive` — Save findings to timestamped JSON report - -### Parameters -| Param | Type | Default | Description | -|-------|------|---------|-------------| -| `action` | string | required | Action to perform | -| `target` | string | "" | Specific directory or file pattern | -| `min_lines` | int | 5 | Minimum lines for duplicate detection | -| `similarity_threshold` | float | 0.85 | Similarity threshold (0.0-1.0) | -| `dry_run` | bool | true | Preview without archiving | - -### Detection Types -1. **Exact function duplicates** — Hash-based detection across files -2. **Similar code blocks** — Normalized whitespace comparison -3. **Duplicate imports** — Imports appearing in >3 files -4. **Repeated string literals** — Literals appearing in >3 files - -### Example Usage -```bash -# Scan entire workspace -curl -X POST http://localhost:8484/api/skills/duplicate-detector/run \ - -H "Content-Type: application/json" \ - -d '{"action": "scan"}' - -# Generate report -curl -X POST http://localhost:8484/api/skills/duplicate-detector/run \ - -d '{"action": "report", "target": "backend/app"}' -``` - ---- - -## 2. Dead Code / Legacy Archiver (`dead-code-archiver`) - -### Actions -- `scan` — Scan for dead code and legacy patterns -- `report` — Scan + generate recommendations -- `archive` — Save findings to timestamped JSON report - -### Parameters -| Param | Type | Default | Description | -|-------|------|---------|-------------| -| `action` | string | required | Action to perform | -| `target` | string | "" | Specific directory to scan | -| `include_legacy` | bool | true | Include legacy pattern detection | -| `dry_run` | bool | true | Preview without archiving | - -### Detection Types -1. **Unused functions** — Functions defined but never called -2. **Orphaned imports** — Imports that are never used -3. **Dead code paths** — Code after return/break/continue -4. **Legacy patterns** — Deprecated APIs and old patterns -5. **Unreferenced files** — Files not imported anywhere - -### Legacy Patterns Detected -- `typing_extensions` imports (use `typing` instead) -- `requests` imports (consider `httpx` for async) -- `.format()` calls (use f-strings) -- `@asyncio.coroutine` decorator (use `async def`) -- `yield from` (use async/await) -- Old-style `super()` calls - -### Example Usage -```bash -# Scan for dead code -curl -X POST http://localhost:8484/api/skills/dead-code-archiver/run \ - -d '{"action": "scan", "include_legacy": true}' - -# Archive findings -curl -X POST http://localhost:8484/api/skills/dead-code-archiver/run \ - -d '{"action": "archive", "dry_run": false}' -``` - ---- - -## 3. Modularisation Planner (`modularisation-planner`) - -### Actions -- `scan` — Scan for large scripts needing modularization -- `report` — Scan + generate recommendations -- `plan` — Generate detailed extraction plan - -### Parameters -| Param | Type | Default | Description | -|-------|------|---------|-------------| -| `action` | string | required | Action to perform | -| `target` | string | "" | Specific file or directory | -| `min_lines` | int | 1000 | Minimum lines to analyze | -| `output_format` | string | "json" | Output: "json" or "markdown" | - -### Analysis Metrics -1. **Large functions** — Functions >100 lines -2. **Complex functions** — Cyclomatic complexity >10 -3. **Missing docstrings** — Functions without documentation -4. **Import clusters** — Modules with >5 imports -5. **Class extraction candidates** — Classes >300 lines - -### Priority Score (0-100) -- Large functions: +20 each (max 40) -- Complex functions: +15 each (max 30) -- Missing docstrings: +5 each (max 15) -- Import clusters: +10 each (max 10) -- Class extraction: +10 each (max 10) - -### Example Usage -```bash -# Scan for large scripts -curl -X POST http://localhost:8484/api/skills/modularisation-planner/run \ - -d '{"action": "scan"}' - -# Generate markdown plan -curl -X POST http://localhost:8484/api/skills/modularisation-planner/run \ - -d '{"action": "plan", "output_format": "markdown"}' -``` - ---- - -## Output Locations - -- Duplicate reports: `tmp/duplicate_reports/duplicates_{timestamp}.json` -- Dead code archive: `tmp/dead_code_archive/dead_code_{timestamp}.json` - -## Integration - -All skills follow the `BaseSkill` pattern and are auto-discovered by the registry. They integrate with: -- Skill audit system -- DevMode operation logging -- Confirmation requirements for mutating actions \ No newline at end of file