diff --git a/scripts/fabric-sync.service b/scripts/fabric-sync.service new file mode 100644 index 0000000..a19029d --- /dev/null +++ b/scripts/fabric-sync.service @@ -0,0 +1,29 @@ +[Unit] +Description=CmdForge Fabric Pattern Sync +Documentation=https://pages.brrd.tech/rob/cmdforge/ +After=network-online.target +Wants=network-online.target + +[Service] +Type=oneshot +User=rob +Group=rob +WorkingDirectory=/home/rob/PycharmProjects/CmdForge + +# Use the project's virtual environment +ExecStart=/home/rob/PycharmProjects/CmdForge/.venv/bin/python scripts/fabric_sync.py --sync + +# Logging +StandardOutput=journal +StandardError=journal +SyslogIdentifier=fabric-sync + +# Security hardening +NoNewPrivileges=yes +ProtectSystem=strict +ProtectHome=read-only +ReadWritePaths=/var/lib/cmdforge /home/rob/.cmdforge +PrivateTmp=yes + +[Install] +WantedBy=multi-user.target diff --git a/scripts/fabric-sync.timer b/scripts/fabric-sync.timer new file mode 100644 index 0000000..f5a543e --- /dev/null +++ b/scripts/fabric-sync.timer @@ -0,0 +1,14 @@ +[Unit] +Description=Daily Fabric Pattern Sync for CmdForge +Documentation=https://pages.brrd.tech/rob/cmdforge/ + +[Timer] +# Run daily at 3 AM +OnCalendar=*-*-* 03:00:00 +# Randomize start time by up to 1 hour to avoid thundering herd +RandomizedDelaySec=3600 +# Run immediately if we missed a scheduled run +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/scripts/fabric_sync.py b/scripts/fabric_sync.py new file mode 100755 index 0000000..a960e46 --- /dev/null +++ b/scripts/fabric_sync.py @@ -0,0 +1,556 @@ +#!/usr/bin/env python3 +""" +Scheduled Fabric pattern sync for CmdForge. + +Monitors the Fabric repository for new/updated patterns and syncs them +to the CmdForge registry through the vetting pipeline. + +Usage: + # Check for new patterns (dry run) + python scripts/fabric_sync.py --dry-run + + # Sync new patterns to registry + python scripts/fabric_sync.py --sync + + # Check status of tracked patterns + python scripts/fabric_sync.py --status + + # Force resync of specific patterns + python scripts/fabric_sync.py --force summarize extract_wisdom + + # Run as daemon with interval + python scripts/fabric_sync.py --daemon --interval 3600 + +Setup for cron (daily sync): + 0 3 * * * /path/to/venv/bin/python /path/to/scripts/fabric_sync.py --sync >> /var/log/fabric_sync.log 2>&1 + +Setup for systemd timer: + See scripts/fabric-sync.service and scripts/fabric-sync.timer +""" + +import argparse +import hashlib +import json +import logging +import os +import subprocess +import sys +import time +from dataclasses import dataclass, field, asdict +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + +import yaml + +# Configure logging +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s [%(levelname)s] %(message)s', + datefmt='%Y-%m-%d %H:%M:%S' +) +logger = logging.getLogger(__name__) + +# Constants +FABRIC_REPO = "https://github.com/danielmiessler/fabric.git" +DEFAULT_SYNC_DIR = Path("/var/lib/cmdforge/fabric-sync") +DEFAULT_STATE_FILE = DEFAULT_SYNC_DIR / "sync_state.json" +DEFAULT_PROVIDER = "opencode-pickle" + + +@dataclass +class PatternState: + """State of a single pattern.""" + name: str + hash: str # SHA256 of system.md content + synced_at: Optional[str] = None + version: str = "1.0.0" + status: str = "pending" # pending, synced, failed, skipped + + +@dataclass +class SyncState: + """Overall sync state.""" + last_check: Optional[str] = None + last_sync: Optional[str] = None + repo_commit: Optional[str] = None + patterns: dict = field(default_factory=dict) # name -> PatternState as dict + + def to_dict(self) -> dict: + return { + "last_check": self.last_check, + "last_sync": self.last_sync, + "repo_commit": self.repo_commit, + "patterns": self.patterns, + } + + @classmethod + def from_dict(cls, data: dict) -> "SyncState": + return cls( + last_check=data.get("last_check"), + last_sync=data.get("last_sync"), + repo_commit=data.get("repo_commit"), + patterns=data.get("patterns", {}), + ) + + +def load_state(state_file: Path) -> SyncState: + """Load sync state from file.""" + if state_file.exists(): + with open(state_file) as f: + data = json.load(f) + return SyncState.from_dict(data) + return SyncState() + + +def save_state(state: SyncState, state_file: Path): + """Save sync state to file.""" + state_file.parent.mkdir(parents=True, exist_ok=True) + with open(state_file, "w") as f: + json.dump(state.to_dict(), f, indent=2) + + +def clone_or_update_repo(sync_dir: Path) -> tuple[Path, str]: + """Clone or update the Fabric repository. + + Returns: + Tuple of (patterns_dir, commit_hash) + """ + fabric_dir = sync_dir / "fabric" + patterns_dir = fabric_dir / "data" / "patterns" + + if fabric_dir.exists(): + logger.info("Updating existing Fabric clone...") + subprocess.run( + ["git", "-C", str(fabric_dir), "fetch", "--quiet"], + check=True, + capture_output=True + ) + subprocess.run( + ["git", "-C", str(fabric_dir), "reset", "--hard", "origin/main", "--quiet"], + check=True, + capture_output=True + ) + else: + logger.info("Cloning Fabric repository...") + sync_dir.mkdir(parents=True, exist_ok=True) + subprocess.run( + ["git", "clone", "--depth", "1", FABRIC_REPO, str(fabric_dir)], + check=True, + capture_output=True + ) + + # Get current commit hash + result = subprocess.run( + ["git", "-C", str(fabric_dir), "rev-parse", "HEAD"], + capture_output=True, + text=True + ) + commit_hash = result.stdout.strip()[:12] + + return patterns_dir, commit_hash + + +def hash_pattern(pattern_dir: Path) -> Optional[str]: + """Calculate hash of pattern content.""" + system_md = pattern_dir / "system.md" + if not system_md.exists(): + return None + + content = system_md.read_bytes() + return hashlib.sha256(content).hexdigest()[:16] + + +def scan_patterns(patterns_dir: Path) -> dict[str, str]: + """Scan all patterns and return name -> hash mapping.""" + patterns = {} + for entry in sorted(patterns_dir.iterdir()): + if entry.is_dir(): + pattern_hash = hash_pattern(entry) + if pattern_hash: + patterns[entry.name] = pattern_hash + return patterns + + +def find_changes( + current_patterns: dict[str, str], + state: SyncState +) -> tuple[list[str], list[str], list[str]]: + """Find new, updated, and removed patterns. + + Returns: + Tuple of (new_patterns, updated_patterns, removed_patterns) + """ + new_patterns = [] + updated_patterns = [] + removed_patterns = [] + + # Check for new and updated + for name, current_hash in current_patterns.items(): + if name not in state.patterns: + new_patterns.append(name) + elif state.patterns[name].get("hash") != current_hash: + updated_patterns.append(name) + + # Check for removed + for name in state.patterns: + if name not in current_patterns: + removed_patterns.append(name) + + return new_patterns, updated_patterns, removed_patterns + + +def vet_pattern(pattern_dir: Path, provider: str = DEFAULT_PROVIDER) -> tuple[bool, str]: + """Run vetting pipeline on a pattern. + + Returns: + Tuple of (passed, reason) + """ + try: + # Try to import the vetting pipeline + script_dir = Path(__file__).parent + sys.path.insert(0, str(script_dir)) + + from import_fabric import create_tool_config, clean_prompt, get_category, pattern_to_display_name + + # Read pattern + system_md = pattern_dir / "system.md" + system_prompt = system_md.read_text() + + # Create config + config = create_tool_config(pattern_dir.name, system_prompt, provider) + + # Run scrutiny + try: + from scrutiny import vet_tool, VetResult + report = vet_tool(config, str(pattern_dir)) + + if report.result == VetResult.REJECT: + return False, f"Rejected: {report.suggestions[0] if report.suggestions else 'quality too low'}" + elif report.result == VetResult.REVIEW: + return True, f"Approved (needs review): score {report.overall_score:.2f}" + else: + return True, f"Approved: score {report.overall_score:.2f}" + except ImportError: + # Scrutiny not available - basic validation + if len(system_prompt.strip()) < 50: + return False, "Pattern too short" + return True, "Basic validation passed" + + except Exception as e: + return False, f"Vetting error: {e}" + + +def sync_pattern( + pattern_dir: Path, + output_dir: Path, + provider: str, + state: SyncState, + dry_run: bool = False +) -> bool: + """Sync a single pattern. + + Returns: + True if successful + """ + name = pattern_dir.name + pattern_hash = hash_pattern(pattern_dir) + + # Vet the pattern + passed, reason = vet_pattern(pattern_dir, provider) + + if not passed: + logger.warning(f" ✗ {name}: {reason}") + state.patterns[name] = { + "name": name, + "hash": pattern_hash, + "status": "failed", + "reason": reason, + "synced_at": datetime.now(timezone.utc).isoformat(), + } + return False + + if dry_run: + logger.info(f" [DRY RUN] Would sync: {name} ({reason})") + return True + + # Import the pattern + try: + script_dir = Path(__file__).parent + sys.path.insert(0, str(script_dir)) + + from import_fabric import import_pattern + + success = import_pattern( + name, + pattern_dir.parent, + output_dir, + provider, + dry_run=False, + registry_format=False, + ) + + if success: + logger.info(f" ✓ {name}: {reason}") + state.patterns[name] = { + "name": name, + "hash": pattern_hash, + "status": "synced", + "synced_at": datetime.now(timezone.utc).isoformat(), + } + return True + else: + logger.error(f" ✗ {name}: Import failed") + state.patterns[name] = { + "name": name, + "hash": pattern_hash, + "status": "failed", + "reason": "Import failed", + "synced_at": datetime.now(timezone.utc).isoformat(), + } + return False + + except Exception as e: + logger.error(f" ✗ {name}: {e}") + state.patterns[name] = { + "name": name, + "hash": pattern_hash, + "status": "failed", + "reason": str(e), + "synced_at": datetime.now(timezone.utc).isoformat(), + } + return False + + +def run_sync( + sync_dir: Path, + output_dir: Path, + state_file: Path, + provider: str, + dry_run: bool = False, + force_patterns: list[str] = None +) -> dict: + """Run the sync process. + + Returns: + Summary dict with counts + """ + # Load state + state = load_state(state_file) + + # Clone/update repo + patterns_dir, commit_hash = clone_or_update_repo(sync_dir) + + # Scan patterns + current_patterns = scan_patterns(patterns_dir) + logger.info(f"Found {len(current_patterns)} patterns in Fabric repo (commit {commit_hash})") + + # Find changes + if force_patterns: + new_patterns = [p for p in force_patterns if p in current_patterns] + updated_patterns = [] + removed_patterns = [] + else: + new_patterns, updated_patterns, removed_patterns = find_changes(current_patterns, state) + + logger.info(f"Changes: {len(new_patterns)} new, {len(updated_patterns)} updated, {len(removed_patterns)} removed") + + # Update state timestamp + state.last_check = datetime.now(timezone.utc).isoformat() + state.repo_commit = commit_hash + + # Process new and updated patterns + to_sync = new_patterns + updated_patterns + synced = 0 + failed = 0 + + if to_sync: + logger.info(f"\nSyncing {len(to_sync)} patterns...") + for name in to_sync: + pattern_dir = patterns_dir / name + if sync_pattern(pattern_dir, output_dir, provider, state, dry_run): + synced += 1 + else: + failed += 1 + + # Mark removed patterns + for name in removed_patterns: + if name in state.patterns: + state.patterns[name]["status"] = "removed" + + # Save state + if not dry_run: + state.last_sync = datetime.now(timezone.utc).isoformat() + save_state(state, state_file) + logger.info(f"\nState saved to {state_file}") + + # Summary + summary = { + "total_patterns": len(current_patterns), + "new": len(new_patterns), + "updated": len(updated_patterns), + "removed": len(removed_patterns), + "synced": synced, + "failed": failed, + "commit": commit_hash, + } + + logger.info(f"\nSync complete: {synced} synced, {failed} failed") + + return summary + + +def print_status(state_file: Path): + """Print current sync status.""" + state = load_state(state_file) + + print(f"\nFabric Sync Status") + print(f"{'=' * 50}") + print(f"Last check: {state.last_check or 'Never'}") + print(f"Last sync: {state.last_sync or 'Never'}") + print(f"Repo commit: {state.repo_commit or 'Unknown'}") + + if state.patterns: + # Count by status + by_status = {} + for p in state.patterns.values(): + status = p.get("status", "unknown") + by_status[status] = by_status.get(status, 0) + 1 + + print(f"\nPatterns: {len(state.patterns)} total") + for status, count in sorted(by_status.items()): + print(f" {status}: {count}") + + # Show failed patterns + failed = [p for p in state.patterns.values() if p.get("status") == "failed"] + if failed: + print(f"\nFailed patterns:") + for p in failed[:10]: + print(f" - {p['name']}: {p.get('reason', 'Unknown error')}") + if len(failed) > 10: + print(f" ... and {len(failed) - 10} more") + else: + print("\nNo patterns tracked yet. Run --sync to start.") + + +def daemon_loop( + sync_dir: Path, + output_dir: Path, + state_file: Path, + provider: str, + interval: int +): + """Run sync in a loop.""" + logger.info(f"Starting daemon mode with {interval}s interval") + + while True: + try: + run_sync(sync_dir, output_dir, state_file, provider) + except Exception as e: + logger.error(f"Sync failed: {e}") + + logger.info(f"Sleeping for {interval}s...") + time.sleep(interval) + + +def main(): + parser = argparse.ArgumentParser( + description="Scheduled Fabric pattern sync for CmdForge", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__ + ) + + parser.add_argument( + "--sync", + action="store_true", + help="Run sync process" + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Show what would be synced without making changes" + ) + parser.add_argument( + "--status", + action="store_true", + help="Show current sync status" + ) + parser.add_argument( + "--force", + nargs="+", + metavar="PATTERN", + help="Force resync of specific patterns" + ) + parser.add_argument( + "--daemon", + action="store_true", + help="Run in daemon mode" + ) + parser.add_argument( + "--interval", + type=int, + default=3600, + help="Sync interval in seconds for daemon mode (default: 3600)" + ) + parser.add_argument( + "--sync-dir", + type=Path, + default=DEFAULT_SYNC_DIR, + help=f"Directory for sync data (default: {DEFAULT_SYNC_DIR})" + ) + parser.add_argument( + "--output", + type=Path, + default=Path.home() / ".cmdforge", + help="Output directory for synced tools (default: ~/.cmdforge)" + ) + parser.add_argument( + "--state-file", + type=Path, + help="State file path (default: /sync_state.json)" + ) + parser.add_argument( + "--provider", + default=DEFAULT_PROVIDER, + help=f"Default provider for tools (default: {DEFAULT_PROVIDER})" + ) + + args = parser.parse_args() + + # Set state file default + state_file = args.state_file or (args.sync_dir / "sync_state.json") + + if args.status: + print_status(state_file) + return 0 + + if args.daemon: + daemon_loop( + args.sync_dir, + args.output, + state_file, + args.provider, + args.interval + ) + return 0 + + if args.sync or args.dry_run or args.force: + summary = run_sync( + args.sync_dir, + args.output, + state_file, + args.provider, + dry_run=args.dry_run, + force_patterns=args.force + ) + + if summary["failed"] > 0: + return 1 + return 0 + + parser.print_help() + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/scrutiny.py b/scripts/scrutiny.py new file mode 100755 index 0000000..5649eae --- /dev/null +++ b/scripts/scrutiny.py @@ -0,0 +1,507 @@ +#!/usr/bin/env python3 +""" +Tool vetting/scrutiny module for CmdForge. + +Performs AI-powered analysis of tools to assess quality and safety: +- Honesty: Does the tool do what it claims? +- Transparency: Is the prompt clear and understandable? +- Scope: Is the tool appropriately scoped? +- Efficiency: Is the prompt well-structured? +- Safety: Are there any concerning patterns? + +Usage: + # Vet a single tool + python scripts/scrutiny.py path/to/tool/config.yaml + + # Vet all tools in directory + python scripts/scrutiny.py --all ~/.cmdforge/ + + # Output as JSON + python scripts/scrutiny.py --json path/to/tool/config.yaml + + # Use specific provider for analysis + python scripts/scrutiny.py --provider claude path/to/tool/config.yaml +""" + +import argparse +import json +import sys +from dataclasses import dataclass, field, asdict +from enum import Enum +from pathlib import Path +from typing import Optional + +import yaml + + +class VetResult(Enum): + """Vetting decision.""" + APPROVE = "approve" # Auto-approve - meets all criteria + REVIEW = "review" # Needs human review - some concerns + REJECT = "reject" # Auto-reject - fails criteria + ERROR = "error" # Could not vet + + +@dataclass +class VetScore: + """Individual score for a vetting criterion.""" + criterion: str + score: float # 0.0 to 1.0 + max_score: float = 1.0 + notes: str = "" + concerns: list[str] = field(default_factory=list) + + +@dataclass +class VetReport: + """Complete vetting report for a tool.""" + tool_name: str + tool_path: str + result: VetResult + overall_score: float # 0.0 to 1.0 + scores: list[VetScore] = field(default_factory=list) + suggestions: list[str] = field(default_factory=list) + error: Optional[str] = None + + def to_dict(self) -> dict: + """Convert to dictionary for JSON serialization.""" + d = asdict(self) + d['result'] = self.result.value + return d + + +# Thresholds for auto-approve/reject +APPROVE_THRESHOLD = 0.8 # Score >= 0.8 -> auto-approve +REJECT_THRESHOLD = 0.3 # Score < 0.3 -> auto-reject + + +def load_tool_config(path: Path) -> Optional[dict]: + """Load tool configuration from YAML file.""" + if path.is_dir(): + config_file = path / "config.yaml" + else: + config_file = path + + if not config_file.exists(): + return None + + with open(config_file) as f: + return yaml.safe_load(f) + + +def vet_honesty(config: dict) -> VetScore: + """Check if tool description matches what it actually does.""" + score = VetScore(criterion="honesty", score=0.0, notes="") + concerns = [] + + name = config.get("name", "") + description = config.get("description", "") + steps = config.get("steps", []) + + # Check that description exists + if not description: + concerns.append("Missing description") + score.score = 0.3 + else: + score.score = 0.6 + + # Check that steps exist + if not steps: + concerns.append("No execution steps defined") + score.score = min(score.score, 0.2) + else: + # Check if description keywords appear in prompts + desc_words = set(description.lower().split()) + prompt_text = "" + for step in steps: + if step.get("type") == "prompt": + prompt_text += step.get("prompt", "").lower() + " " + + # Simple keyword overlap check + prompt_words = set(prompt_text.split()) + overlap = desc_words & prompt_words + meaningful_overlap = overlap - {"the", "a", "an", "and", "or", "is", "to", "for", "of", "in"} + + if len(meaningful_overlap) >= 2: + score.score = min(1.0, score.score + 0.3) + score.notes = f"Description matches prompt content ({len(meaningful_overlap)} keywords)" + else: + concerns.append("Description may not match actual behavior") + + score.concerns = concerns + return score + + +def vet_transparency(config: dict) -> VetScore: + """Check if the tool's behavior is clear and understandable.""" + score = VetScore(criterion="transparency", score=0.0, notes="") + concerns = [] + + steps = config.get("steps", []) + + if not steps: + concerns.append("No steps to analyze") + score.concerns = concerns + return score + + # Analyze each step + total_prompt_length = 0 + has_clear_instructions = False + + for step in steps: + if step.get("type") == "prompt": + prompt = step.get("prompt", "") + total_prompt_length += len(prompt) + + # Check for clear instruction patterns + instruction_patterns = [ + "you are", "your task", "please", "analyze", "extract", + "summarize", "create", "write", "explain", "review" + ] + prompt_lower = prompt.lower() + if any(p in prompt_lower for p in instruction_patterns): + has_clear_instructions = True + + # Score based on findings + if has_clear_instructions: + score.score += 0.5 + score.notes = "Contains clear instructions" + + if total_prompt_length > 50: + score.score += 0.3 + score.notes += "; Substantial prompt content" + elif total_prompt_length > 0: + score.score += 0.1 + concerns.append("Very short prompt - may lack clarity") + + # Check for output variable naming + for step in steps: + output_var = step.get("output_var", "") + if output_var and output_var != "response": + score.score += 0.2 + score.notes += "; Descriptive output variable" + break + + score.score = min(1.0, score.score) + score.concerns = concerns + return score + + +def vet_scope(config: dict) -> VetScore: + """Check if tool is appropriately scoped (not too broad/narrow).""" + score = VetScore(criterion="scope", score=0.0, notes="") + concerns = [] + + description = config.get("description", "") + steps = config.get("steps", []) + arguments = config.get("arguments", []) + + # Start with base score + score.score = 0.5 + + # Single-step tools are well-scoped + if len(steps) == 1: + score.score += 0.2 + score.notes = "Single-step tool - focused scope" + elif len(steps) <= 3: + score.score += 0.1 + score.notes = "Multi-step tool with reasonable complexity" + else: + concerns.append(f"Complex tool with {len(steps)} steps - may be over-scoped") + score.score -= 0.1 + + # Check for overly generic descriptions + generic_terms = ["everything", "anything", "all", "any task", "general purpose"] + desc_lower = description.lower() + if any(term in desc_lower for term in generic_terms): + concerns.append("Description suggests overly broad scope") + score.score -= 0.2 + + # Arguments indicate configurable scope (good) + if arguments: + score.score += 0.1 + score.notes += "; Configurable via arguments" + + score.score = max(0.0, min(1.0, score.score)) + score.concerns = concerns + return score + + +def vet_efficiency(config: dict) -> VetScore: + """Check if prompt is well-structured and efficient.""" + score = VetScore(criterion="efficiency", score=0.0, notes="") + concerns = [] + + steps = config.get("steps", []) + + # Analyze prompts + for step in steps: + if step.get("type") == "prompt": + prompt = step.get("prompt", "") + + # Check for excessive repetition + words = prompt.lower().split() + word_counts = {} + for word in words: + if len(word) > 4: # Only check meaningful words + word_counts[word] = word_counts.get(word, 0) + 1 + + max_repetition = max(word_counts.values()) if word_counts else 0 + if max_repetition > 5: + concerns.append(f"Repetitive language detected ({max_repetition}x)") + score.score = max(0.0, score.score - 0.2) + + # Check for structured output hints + structure_patterns = [ + "markdown", "json", "format", "structure", "sections", + "bullet", "numbered", "list", "table" + ] + if any(p in prompt.lower() for p in structure_patterns): + score.score += 0.3 + score.notes = "Specifies output structure" + + # Reasonable length (not too short, not excessive) + if 100 <= len(prompt) <= 5000: + score.score += 0.4 + elif len(prompt) < 100: + concerns.append("Very short prompt - may lack guidance") + score.score += 0.2 + else: + concerns.append("Very long prompt - may be inefficient") + score.score += 0.2 + + # Base score if steps exist + if steps: + score.score += 0.3 + + score.score = min(1.0, score.score) + score.concerns = concerns + return score + + +def vet_safety(config: dict) -> VetScore: + """Check for concerning patterns in the tool.""" + score = VetScore(criterion="safety", score=1.0, notes="No safety concerns") + concerns = [] + + steps = config.get("steps", []) + + # Check for code steps + code_step_count = 0 + for step in steps: + if step.get("type") == "code": + code_step_count += 1 + code = step.get("code", "") + + # Check for potentially dangerous patterns + dangerous_patterns = [ + ("subprocess", "Executes shell commands"), + ("os.system", "Executes shell commands"), + ("eval(", "Dynamic code execution"), + ("exec(", "Dynamic code execution"), + ("open(", "File operations"), + ("requests.", "Network requests"), + ("urllib", "Network requests"), + ("shutil.rmtree", "Recursive deletion"), + ] + + for pattern, concern in dangerous_patterns: + if pattern in code: + concerns.append(f"Code contains {concern.lower()}") + score.score -= 0.15 + + if code_step_count > 0: + score.notes = f"Contains {code_step_count} code step(s)" + if not concerns: + score.notes += " - no dangerous patterns detected" + + score.score = max(0.0, score.score) + score.concerns = concerns + return score + + +def vet_tool(config: dict, tool_path: str) -> VetReport: + """Perform complete vetting of a tool.""" + name = config.get("name", "unknown") + + # Run all checks + scores = [ + vet_honesty(config), + vet_transparency(config), + vet_scope(config), + vet_efficiency(config), + vet_safety(config), + ] + + # Calculate overall score (weighted average) + weights = { + "honesty": 0.25, + "transparency": 0.20, + "scope": 0.15, + "efficiency": 0.15, + "safety": 0.25, + } + + total_weight = sum(weights.values()) + weighted_sum = sum(s.score * weights.get(s.criterion, 0.1) for s in scores) + overall_score = weighted_sum / total_weight + + # Determine result + if overall_score >= APPROVE_THRESHOLD: + result = VetResult.APPROVE + elif overall_score < REJECT_THRESHOLD: + result = VetResult.REJECT + else: + result = VetResult.REVIEW + + # Collect all concerns for suggestions + suggestions = [] + for s in scores: + for concern in s.concerns: + suggestions.append(f"[{s.criterion}] {concern}") + + return VetReport( + tool_name=name, + tool_path=tool_path, + result=result, + overall_score=overall_score, + scores=scores, + suggestions=suggestions, + ) + + +def vet_directory(directory: Path, provider: Optional[str] = None) -> list[VetReport]: + """Vet all tools in a directory.""" + reports = [] + + for entry in directory.iterdir(): + config_file = None + if entry.is_dir(): + config_file = entry / "config.yaml" + elif entry.suffix in [".yaml", ".yml"]: + config_file = entry + + if config_file and config_file.exists(): + config = load_tool_config(config_file) + if config: + report = vet_tool(config, str(entry)) + reports.append(report) + + return reports + + +def print_report(report: VetReport, verbose: bool = False): + """Print a vetting report to console.""" + # Result emoji + result_emoji = { + VetResult.APPROVE: "✅", + VetResult.REVIEW: "⚠️", + VetResult.REJECT: "❌", + VetResult.ERROR: "💥", + } + + emoji = result_emoji.get(report.result, "❓") + print(f"\n{emoji} {report.tool_name}: {report.result.value.upper()} (score: {report.overall_score:.2f})") + + if verbose or report.result != VetResult.APPROVE: + print(f" Path: {report.tool_path}") + + # Print individual scores + for score in report.scores: + bar = "█" * int(score.score * 10) + "░" * (10 - int(score.score * 10)) + print(f" {score.criterion:12} [{bar}] {score.score:.2f}") + if score.concerns: + for concern in score.concerns: + print(f" ⚠ {concern}") + + # Print suggestions + if report.suggestions and verbose: + print(" Suggestions:") + for suggestion in report.suggestions: + print(f" • {suggestion}") + + +def main(): + parser = argparse.ArgumentParser( + description="Vet CmdForge tools for quality and safety", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__ + ) + + parser.add_argument( + "path", + type=Path, + nargs="?", + help="Tool config file or directory to vet" + ) + parser.add_argument( + "--all", + action="store_true", + help="Vet all tools in directory" + ) + parser.add_argument( + "--json", + action="store_true", + help="Output as JSON" + ) + parser.add_argument( + "--verbose", "-v", + action="store_true", + help="Show detailed output" + ) + parser.add_argument( + "--provider", + default=None, + help="AI provider for enhanced analysis (future feature)" + ) + + args = parser.parse_args() + + if not args.path: + parser.error("Please specify a path to vet") + + # Collect reports + reports = [] + + if args.all or args.path.is_dir(): + reports = vet_directory(args.path, args.provider) + else: + config = load_tool_config(args.path) + if not config: + print(f"Error: Could not load tool config from {args.path}", file=sys.stderr) + return 1 + report = vet_tool(config, str(args.path)) + reports.append(report) + + if not reports: + print("No tools found to vet", file=sys.stderr) + return 1 + + # Output + if args.json: + output = [r.to_dict() for r in reports] + print(json.dumps(output, indent=2)) + else: + # Summary + approved = sum(1 for r in reports if r.result == VetResult.APPROVE) + review = sum(1 for r in reports if r.result == VetResult.REVIEW) + rejected = sum(1 for r in reports if r.result == VetResult.REJECT) + + print(f"Vetting {len(reports)} tool(s)...") + + for report in reports: + print_report(report, args.verbose) + + print(f"\n{'─' * 40}") + print(f"Summary: {approved} approved, {review} need review, {rejected} rejected") + + # Return code based on results + if rejected > 0: + return 2 + elif review > 0: + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/similarity.py b/scripts/similarity.py new file mode 100755 index 0000000..ff25ffc --- /dev/null +++ b/scripts/similarity.py @@ -0,0 +1,441 @@ +#!/usr/bin/env python3 +""" +Duplicate detection for CmdForge tools using text similarity. + +Finds tools that may be duplicates or very similar to existing tools. +Uses TF-IDF vectorization and cosine similarity for comparison. + +Usage: + # Check a tool against all existing tools + python scripts/similarity.py path/to/tool/config.yaml + + # Check against tools in a specific directory + python scripts/similarity.py path/to/tool/config.yaml --against ~/.cmdforge/ + + # Find all similar pairs in a directory + python scripts/similarity.py --scan ~/.cmdforge/ + + # Set similarity threshold (default: 0.7) + python scripts/similarity.py --threshold 0.8 path/to/tool/config.yaml + + # Output as JSON + python scripts/similarity.py --json path/to/tool/config.yaml +""" + +import argparse +import json +import math +import re +import sys +from collections import Counter +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional + +import yaml + + +@dataclass +class SimilarityMatch: + """A similarity match between two tools.""" + tool1_name: str + tool1_path: str + tool2_name: str + tool2_path: str + similarity: float # 0.0 to 1.0 + match_type: str # "duplicate", "similar", "related" + + def to_dict(self) -> dict: + return { + "tool1": {"name": self.tool1_name, "path": self.tool1_path}, + "tool2": {"name": self.tool2_name, "path": self.tool2_path}, + "similarity": self.similarity, + "match_type": self.match_type, + } + + +@dataclass +class ToolText: + """Extracted text from a tool for similarity comparison.""" + name: str + path: str + text: str + tokens: list[str] = field(default_factory=list) + tfidf: dict = field(default_factory=dict) + + +# Similarity thresholds +DUPLICATE_THRESHOLD = 0.9 # >= 0.9 is likely a duplicate +SIMILAR_THRESHOLD = 0.7 # >= 0.7 is very similar +RELATED_THRESHOLD = 0.5 # >= 0.5 is related + + +def load_tool_config(path: Path) -> Optional[dict]: + """Load tool configuration from YAML file.""" + if path.is_dir(): + config_file = path / "config.yaml" + else: + config_file = path + + if not config_file.exists(): + return None + + with open(config_file) as f: + return yaml.safe_load(f) + + +def extract_tool_text(config: dict, path: str) -> ToolText: + """Extract all meaningful text from a tool config.""" + texts = [] + + # Name and description + name = config.get("name", "") + texts.append(name) + texts.append(config.get("description", "")) + + # Category and tags + texts.append(config.get("category", "")) + tags = config.get("tags", []) + if isinstance(tags, list): + texts.extend(tags) + + # Steps - extract prompts and code + for step in config.get("steps", []): + if step.get("type") == "prompt": + texts.append(step.get("prompt", "")) + elif step.get("type") == "code": + # Extract meaningful parts from code (comments, strings) + code = step.get("code", "") + # Add variable names + texts.append(code) + + # Arguments + for arg in config.get("arguments", []): + texts.append(arg.get("flag", "")) + texts.append(arg.get("description", "")) + texts.append(arg.get("variable", "")) + + # Combine and clean + combined = " ".join(texts) + return ToolText(name=name, path=path, text=combined) + + +def tokenize(text: str) -> list[str]: + """Tokenize text into words, removing stopwords.""" + # Convert to lowercase and extract words + words = re.findall(r'\b[a-z]{2,}\b', text.lower()) + + # Remove common stopwords + stopwords = { + "the", "a", "an", "and", "or", "but", "in", "on", "at", "to", "for", + "of", "with", "by", "from", "as", "is", "was", "are", "were", "been", + "be", "have", "has", "had", "do", "does", "did", "will", "would", + "could", "should", "may", "might", "must", "shall", "can", "this", + "that", "these", "those", "it", "its", "you", "your", "we", "our", + "they", "their", "he", "his", "she", "her", "if", "then", "else", + "when", "where", "which", "who", "what", "how", "all", "each", + "every", "both", "few", "more", "most", "other", "some", "such", + "no", "not", "only", "same", "so", "than", "too", "very", "just", + "also", "now", "here", "there", "any", "into", "out", "up", "down", + } + + return [w for w in words if w not in stopwords] + + +def compute_tfidf(documents: list[ToolText]) -> None: + """Compute TF-IDF vectors for all documents (modifies in place).""" + # Tokenize all documents + for doc in documents: + doc.tokens = tokenize(doc.text) + + # Compute document frequencies + doc_freq = Counter() + for doc in documents: + unique_tokens = set(doc.tokens) + doc_freq.update(unique_tokens) + + num_docs = len(documents) + + # Compute TF-IDF for each document + for doc in documents: + term_freq = Counter(doc.tokens) + total_terms = len(doc.tokens) or 1 + + tfidf = {} + for term, count in term_freq.items(): + tf = count / total_terms + # Add 1 to avoid division by zero + idf = math.log((num_docs + 1) / (doc_freq[term] + 1)) + 1 + tfidf[term] = tf * idf + + doc.tfidf = tfidf + + +def cosine_similarity(vec1: dict, vec2: dict) -> float: + """Compute cosine similarity between two TF-IDF vectors.""" + # Get all terms + all_terms = set(vec1.keys()) | set(vec2.keys()) + + if not all_terms: + return 0.0 + + # Compute dot product and magnitudes + dot_product = 0.0 + mag1 = 0.0 + mag2 = 0.0 + + for term in all_terms: + v1 = vec1.get(term, 0.0) + v2 = vec2.get(term, 0.0) + dot_product += v1 * v2 + mag1 += v1 * v1 + mag2 += v2 * v2 + + mag1 = math.sqrt(mag1) + mag2 = math.sqrt(mag2) + + if mag1 == 0 or mag2 == 0: + return 0.0 + + return dot_product / (mag1 * mag2) + + +def classify_similarity(score: float) -> str: + """Classify similarity score into a match type.""" + if score >= DUPLICATE_THRESHOLD: + return "duplicate" + elif score >= SIMILAR_THRESHOLD: + return "similar" + elif score >= RELATED_THRESHOLD: + return "related" + return "different" + + +def find_similar(tool: ToolText, corpus: list[ToolText], threshold: float = RELATED_THRESHOLD) -> list[SimilarityMatch]: + """Find tools in corpus similar to the given tool.""" + matches = [] + + for other in corpus: + if other.path == tool.path: + continue + + similarity = cosine_similarity(tool.tfidf, other.tfidf) + + if similarity >= threshold: + match_type = classify_similarity(similarity) + matches.append(SimilarityMatch( + tool1_name=tool.name, + tool1_path=tool.path, + tool2_name=other.name, + tool2_path=other.path, + similarity=similarity, + match_type=match_type, + )) + + # Sort by similarity descending + matches.sort(key=lambda m: m.similarity, reverse=True) + return matches + + +def scan_directory(directory: Path, threshold: float = RELATED_THRESHOLD) -> list[SimilarityMatch]: + """Scan a directory for similar tool pairs.""" + # Load all tools + tools = [] + for entry in directory.iterdir(): + config_file = None + if entry.is_dir(): + config_file = entry / "config.yaml" + elif entry.suffix in [".yaml", ".yml"]: + config_file = entry + + if config_file and config_file.exists(): + config = load_tool_config(config_file) + if config: + tool_text = extract_tool_text(config, str(entry)) + tools.append(tool_text) + + if len(tools) < 2: + return [] + + # Compute TF-IDF + compute_tfidf(tools) + + # Find all similar pairs + matches = [] + seen_pairs = set() + + for tool in tools: + for other in tools: + if tool.path >= other.path: # Avoid duplicates + continue + + pair_key = (tool.path, other.path) + if pair_key in seen_pairs: + continue + seen_pairs.add(pair_key) + + similarity = cosine_similarity(tool.tfidf, other.tfidf) + + if similarity >= threshold: + match_type = classify_similarity(similarity) + matches.append(SimilarityMatch( + tool1_name=tool.name, + tool1_path=tool.path, + tool2_name=other.name, + tool2_path=other.path, + similarity=similarity, + match_type=match_type, + )) + + # Sort by similarity descending + matches.sort(key=lambda m: m.similarity, reverse=True) + return matches + + +def load_corpus(directory: Path) -> list[ToolText]: + """Load all tools from a directory as ToolText objects.""" + tools = [] + + for entry in directory.iterdir(): + config_file = None + if entry.is_dir(): + config_file = entry / "config.yaml" + elif entry.suffix in [".yaml", ".yml"]: + config_file = entry + + if config_file and config_file.exists(): + config = load_tool_config(config_file) + if config: + tool_text = extract_tool_text(config, str(entry)) + tools.append(tool_text) + + return tools + + +def print_match(match: SimilarityMatch, verbose: bool = False): + """Print a similarity match to console.""" + # Match type emoji + type_emoji = { + "duplicate": "🔴", + "similar": "🟠", + "related": "🟡", + } + + emoji = type_emoji.get(match.match_type, "⚪") + bar = "█" * int(match.similarity * 20) + "░" * (20 - int(match.similarity * 20)) + + print(f" {emoji} {match.match_type.upper()} [{bar}] {match.similarity:.2%}") + print(f" {match.tool1_name} <-> {match.tool2_name}") + + if verbose: + print(f" Paths: {match.tool1_path}") + print(f" {match.tool2_path}") + + +def main(): + parser = argparse.ArgumentParser( + description="Find similar/duplicate CmdForge tools", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__ + ) + + parser.add_argument( + "path", + type=Path, + nargs="?", + help="Tool config file to check" + ) + parser.add_argument( + "--against", + type=Path, + default=Path.home() / ".cmdforge", + help="Directory to compare against (default: ~/.cmdforge)" + ) + parser.add_argument( + "--scan", + type=Path, + metavar="DIR", + help="Scan directory for all similar pairs" + ) + parser.add_argument( + "--threshold", "-t", + type=float, + default=RELATED_THRESHOLD, + help=f"Similarity threshold (default: {RELATED_THRESHOLD})" + ) + parser.add_argument( + "--json", + action="store_true", + help="Output as JSON" + ) + parser.add_argument( + "--verbose", "-v", + action="store_true", + help="Show detailed output" + ) + + args = parser.parse_args() + + matches = [] + + if args.scan: + # Scan mode - find all similar pairs + print(f"Scanning {args.scan} for similar tools...") + matches = scan_directory(args.scan, args.threshold) + elif args.path: + # Check single tool against corpus + config = load_tool_config(args.path) + if not config: + print(f"Error: Could not load tool config from {args.path}", file=sys.stderr) + return 1 + + tool_text = extract_tool_text(config, str(args.path)) + + # Load corpus + corpus = load_corpus(args.against) + if not corpus: + print(f"No tools found in {args.against}", file=sys.stderr) + return 1 + + # Add the new tool to corpus for TF-IDF computation + all_tools = corpus + [tool_text] + compute_tfidf(all_tools) + + print(f"Checking {tool_text.name} against {len(corpus)} existing tools...") + matches = find_similar(tool_text, corpus, args.threshold) + else: + parser.error("Specify a tool path or use --scan") + + # Output + if args.json: + output = [m.to_dict() for m in matches] + print(json.dumps(output, indent=2)) + else: + if not matches: + print("\n✅ No similar tools found above threshold") + else: + print(f"\nFound {len(matches)} match(es):\n") + for match in matches: + print_match(match, args.verbose) + + # Summary by type + duplicates = sum(1 for m in matches if m.match_type == "duplicate") + similar = sum(1 for m in matches if m.match_type == "similar") + related = sum(1 for m in matches if m.match_type == "related") + + print(f"\n{'─' * 40}") + print(f"Summary: {duplicates} duplicates, {similar} similar, {related} related") + + # Return code + has_duplicates = any(m.match_type == "duplicate" for m in matches) + has_similar = any(m.match_type == "similar" for m in matches) + + if has_duplicates: + return 2 + elif has_similar: + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/vet_pipeline.py b/scripts/vet_pipeline.py new file mode 100755 index 0000000..7512316 --- /dev/null +++ b/scripts/vet_pipeline.py @@ -0,0 +1,379 @@ +#!/usr/bin/env python3 +""" +Auto-vetting pipeline for CmdForge tool submissions. + +Combines scrutiny (quality checks) and similarity (duplicate detection) +to automatically triage incoming tools: +- Auto-approve: High quality, no duplicates +- Auto-reject: Low quality or exact duplicates +- Review queue: Needs human review + +Usage: + # Vet a single tool + python scripts/vet_pipeline.py path/to/tool/config.yaml + + # Vet with custom thresholds + python scripts/vet_pipeline.py --approve-threshold 0.85 path/to/tool/config.yaml + + # Process all tools in import directory + python scripts/vet_pipeline.py --batch /tmp/fabric-import/ + + # Output detailed JSON report + python scripts/vet_pipeline.py --json path/to/tool/config.yaml +""" + +import argparse +import json +import sys +from dataclasses import dataclass, asdict +from enum import Enum +from pathlib import Path +from typing import Optional + +# Import our vetting modules +from scrutiny import ( + VetResult as ScrutinyResult, + VetReport, + load_tool_config, + vet_tool, +) +from similarity import ( + ToolText, + SimilarityMatch, + extract_tool_text, + load_corpus, + compute_tfidf, + find_similar, + DUPLICATE_THRESHOLD, + SIMILAR_THRESHOLD, +) + + +class PipelineDecision(Enum): + """Final pipeline decision.""" + AUTO_APPROVE = "auto_approve" + AUTO_REJECT = "auto_reject" + NEEDS_REVIEW = "needs_review" + ERROR = "error" + + +@dataclass +class PipelineResult: + """Complete pipeline result for a tool.""" + tool_name: str + tool_path: str + decision: PipelineDecision + reason: str + scrutiny_report: Optional[VetReport] = None + similarity_matches: list = None + suggestions: list = None + + def __post_init__(self): + if self.similarity_matches is None: + self.similarity_matches = [] + if self.suggestions is None: + self.suggestions = [] + + def to_dict(self) -> dict: + d = { + "tool_name": self.tool_name, + "tool_path": self.tool_path, + "decision": self.decision.value, + "reason": self.reason, + "suggestions": self.suggestions, + } + if self.scrutiny_report: + d["scrutiny"] = self.scrutiny_report.to_dict() + if self.similarity_matches: + d["similar_tools"] = [m.to_dict() for m in self.similarity_matches] + return d + + +# Pipeline thresholds +DEFAULT_APPROVE_THRESHOLD = 0.8 +DEFAULT_REJECT_THRESHOLD = 0.3 + + +def run_pipeline( + tool_path: Path, + corpus_dir: Path, + approve_threshold: float = DEFAULT_APPROVE_THRESHOLD, + reject_threshold: float = DEFAULT_REJECT_THRESHOLD, +) -> PipelineResult: + """Run the complete vetting pipeline on a tool.""" + + # Load tool config + config = load_tool_config(tool_path) + if not config: + return PipelineResult( + tool_name="unknown", + tool_path=str(tool_path), + decision=PipelineDecision.ERROR, + reason=f"Could not load tool config from {tool_path}", + ) + + tool_name = config.get("name", "unknown") + + # Phase 1: Scrutiny (quality checks) + scrutiny_report = vet_tool(config, str(tool_path)) + + # Phase 2: Similarity check + tool_text = extract_tool_text(config, str(tool_path)) + corpus = load_corpus(corpus_dir) if corpus_dir.exists() else [] + + similarity_matches = [] + if corpus: + all_tools = corpus + [tool_text] + compute_tfidf(all_tools) + similarity_matches = find_similar(tool_text, corpus, threshold=0.5) + + # Phase 3: Decision logic + decision, reason, suggestions = make_decision( + scrutiny_report, + similarity_matches, + approve_threshold, + reject_threshold, + ) + + return PipelineResult( + tool_name=tool_name, + tool_path=str(tool_path), + decision=decision, + reason=reason, + scrutiny_report=scrutiny_report, + similarity_matches=similarity_matches, + suggestions=suggestions, + ) + + +def make_decision( + scrutiny: VetReport, + similarity_matches: list[SimilarityMatch], + approve_threshold: float, + reject_threshold: float, +) -> tuple[PipelineDecision, str, list[str]]: + """Make final decision based on scrutiny and similarity results.""" + + suggestions = list(scrutiny.suggestions) if scrutiny.suggestions else [] + score = scrutiny.overall_score + + # Check for exact duplicates first + duplicates = [m for m in similarity_matches if m.match_type == "duplicate"] + if duplicates: + dupe = duplicates[0] + return ( + PipelineDecision.AUTO_REJECT, + f"Duplicate of existing tool '{dupe.tool2_name}' ({dupe.similarity:.0%} match)", + suggestions + [f"Consider updating {dupe.tool2_name} instead of creating a new tool"], + ) + + # Check for very similar tools + similar = [m for m in similarity_matches if m.match_type == "similar"] + if similar: + # If high quality + similar, might be an improved version - needs review + if score >= approve_threshold: + sim = similar[0] + return ( + PipelineDecision.NEEDS_REVIEW, + f"High quality but similar to '{sim.tool2_name}' ({sim.similarity:.0%})", + suggestions + [f"Review whether this improves on {sim.tool2_name}"], + ) + else: + sim = similar[0] + return ( + PipelineDecision.NEEDS_REVIEW, + f"Similar to existing tool '{sim.tool2_name}' ({sim.similarity:.0%})", + suggestions + [f"Consider if this duplicates {sim.tool2_name}"], + ) + + # No duplicates or very similar tools - decide based on quality + if scrutiny.result == ScrutinyResult.REJECT or score < reject_threshold: + return ( + PipelineDecision.AUTO_REJECT, + f"Quality score too low ({score:.2f} < {reject_threshold})", + suggestions, + ) + + if scrutiny.result == ScrutinyResult.APPROVE and score >= approve_threshold: + return ( + PipelineDecision.AUTO_APPROVE, + f"High quality score ({score:.2f}) with no similar tools", + suggestions, + ) + + # Middle ground - needs review + return ( + PipelineDecision.NEEDS_REVIEW, + f"Quality score {score:.2f} - needs human review", + suggestions, + ) + + +def print_result(result: PipelineResult, verbose: bool = False): + """Print pipeline result to console.""" + decision_emoji = { + PipelineDecision.AUTO_APPROVE: "✅", + PipelineDecision.AUTO_REJECT: "❌", + PipelineDecision.NEEDS_REVIEW: "⚠️", + PipelineDecision.ERROR: "💥", + } + + decision_color = { + PipelineDecision.AUTO_APPROVE: "\033[92m", # Green + PipelineDecision.AUTO_REJECT: "\033[91m", # Red + PipelineDecision.NEEDS_REVIEW: "\033[93m", # Yellow + PipelineDecision.ERROR: "\033[91m", # Red + } + + reset = "\033[0m" + emoji = decision_emoji.get(result.decision, "❓") + color = decision_color.get(result.decision, "") + + print(f"\n{emoji} {color}{result.tool_name}: {result.decision.value.upper()}{reset}") + print(f" {result.reason}") + + if verbose: + print(f" Path: {result.tool_path}") + + if result.scrutiny_report: + print(f"\n Quality Scores:") + for score in result.scrutiny_report.scores: + bar = "█" * int(score.score * 10) + "░" * (10 - int(score.score * 10)) + print(f" {score.criterion:12} [{bar}] {score.score:.2f}") + + if result.similarity_matches: + print(f"\n Similar Tools:") + for match in result.similarity_matches[:3]: + print(f" • {match.tool2_name} ({match.similarity:.0%} {match.match_type})") + + if result.suggestions: + print(f"\n Suggestions:") + for suggestion in result.suggestions[:5]: + print(f" • {suggestion}") + + +def main(): + parser = argparse.ArgumentParser( + description="Run auto-vetting pipeline on CmdForge tools", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__ + ) + + parser.add_argument( + "path", + type=Path, + nargs="?", + help="Tool config file or directory to vet" + ) + parser.add_argument( + "--batch", + type=Path, + metavar="DIR", + help="Process all tools in directory" + ) + parser.add_argument( + "--corpus", + type=Path, + default=Path.home() / ".cmdforge", + help="Directory of existing tools for comparison (default: ~/.cmdforge)" + ) + parser.add_argument( + "--approve-threshold", + type=float, + default=DEFAULT_APPROVE_THRESHOLD, + help=f"Score threshold for auto-approve (default: {DEFAULT_APPROVE_THRESHOLD})" + ) + parser.add_argument( + "--reject-threshold", + type=float, + default=DEFAULT_REJECT_THRESHOLD, + help=f"Score threshold for auto-reject (default: {DEFAULT_REJECT_THRESHOLD})" + ) + parser.add_argument( + "--json", + action="store_true", + help="Output as JSON" + ) + parser.add_argument( + "--verbose", "-v", + action="store_true", + help="Show detailed output" + ) + + args = parser.parse_args() + + if not args.path and not args.batch: + parser.error("Specify a tool path or use --batch") + + results = [] + + if args.batch: + # Batch mode + batch_dir = args.batch + if not batch_dir.exists(): + print(f"Error: Directory {batch_dir} does not exist", file=sys.stderr) + return 1 + + print(f"Processing tools in {batch_dir}...") + + for entry in sorted(batch_dir.iterdir()): + config_file = None + if entry.is_dir(): + config_file = entry / "config.yaml" + elif entry.suffix in [".yaml", ".yml"]: + config_file = entry + + if config_file and config_file.exists(): + result = run_pipeline( + config_file, + args.corpus, + args.approve_threshold, + args.reject_threshold, + ) + results.append(result) + else: + # Single tool mode + result = run_pipeline( + args.path, + args.corpus, + args.approve_threshold, + args.reject_threshold, + ) + results.append(result) + + # Output + if args.json: + output = [r.to_dict() for r in results] + print(json.dumps(output, indent=2)) + else: + for result in results: + print_result(result, args.verbose) + + # Summary + if len(results) > 1: + approved = sum(1 for r in results if r.decision == PipelineDecision.AUTO_APPROVE) + rejected = sum(1 for r in results if r.decision == PipelineDecision.AUTO_REJECT) + review = sum(1 for r in results if r.decision == PipelineDecision.NEEDS_REVIEW) + errors = sum(1 for r in results if r.decision == PipelineDecision.ERROR) + + print(f"\n{'═' * 50}") + print(f"Pipeline Summary: {len(results)} tools processed") + print(f" ✅ Auto-approved: {approved}") + print(f" ⚠️ Needs review: {review}") + print(f" ❌ Auto-rejected: {rejected}") + if errors: + print(f" 💥 Errors: {errors}") + + # Return codes + has_rejected = any(r.decision == PipelineDecision.AUTO_REJECT for r in results) + has_review = any(r.decision == PipelineDecision.NEEDS_REVIEW for r in results) + + if has_rejected: + return 2 + elif has_review: + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/cmdforge/cli/__init__.py b/src/cmdforge/cli/__init__.py index 6177984..4abcb1c 100644 --- a/src/cmdforge/cli/__init__.py +++ b/src/cmdforge/cli/__init__.py @@ -190,6 +190,16 @@ def main(): p_reg_browse = registry_sub.add_parser("browse", help="Browse tools (TUI)") p_reg_browse.set_defaults(func=cmd_registry) + # registry config (admin settings management) + p_reg_config = registry_sub.add_parser("config", help="Manage registry settings (admin)") + p_reg_config.add_argument("action", nargs="?", choices=["list", "get", "set"], default="list", + help="Action to perform (default: list)") + p_reg_config.add_argument("key", nargs="?", help="Setting key (for get/set)") + p_reg_config.add_argument("value", nargs="?", help="Setting value (for set)") + p_reg_config.add_argument("--json", action="store_true", help="Output as JSON") + p_reg_config.add_argument("--category", "-c", help="Filter by category (for list)") + p_reg_config.set_defaults(func=cmd_registry) + # Default for registry with no subcommand p_registry.set_defaults(func=lambda args: cmd_registry(args) if args.registry_cmd else (setattr(args, 'registry_cmd', None) or cmd_registry(args))) diff --git a/src/cmdforge/cli/registry_commands.py b/src/cmdforge/cli/registry_commands.py index 685778e..365af39 100644 --- a/src/cmdforge/cli/registry_commands.py +++ b/src/cmdforge/cli/registry_commands.py @@ -32,6 +32,8 @@ def cmd_registry(args): return _cmd_registry_my_tools(args) elif args.registry_cmd == "browse": return _cmd_registry_browse(args) + elif args.registry_cmd == "config": + return _cmd_registry_config(args) else: # Default: show registry help print("Registry commands:") @@ -44,6 +46,7 @@ def cmd_registry(args): print(" publish [path] Publish a tool") print(" my-tools List your published tools") print(" browse Browse tools (TUI)") + print(" config [action] Manage registry settings (admin)") return 0 @@ -505,3 +508,158 @@ def _cmd_registry_browse(args): from ..gui import run_gui # Launch GUI - it will open to Registry page return run_gui() + + +def _cmd_registry_config(args): + """Manage registry settings (admin only).""" + from ..registry_client import RegistryError, get_client + + action = getattr(args, 'action', 'list') + key = getattr(args, 'key', None) + value = getattr(args, 'value', None) + as_json = getattr(args, 'json', False) + category = getattr(args, 'category', None) + + try: + client = get_client() + + if action == "list": + return _config_list(client, as_json, category) + elif action == "get": + if not key: + print("Error: key is required for 'get' action", file=sys.stderr) + print("Usage: cmdforge registry config get ", file=sys.stderr) + return 1 + return _config_get(client, key, as_json) + elif action == "set": + if not key or value is None: + print("Error: key and value are required for 'set' action", file=sys.stderr) + print("Usage: cmdforge registry config set ", file=sys.stderr) + return 1 + return _config_set(client, key, value) + + except RegistryError as e: + if e.code == "UNAUTHORIZED": + print("Authentication failed.", file=sys.stderr) + print("This command requires admin privileges.", file=sys.stderr) + print("Set your admin token with: cmdforge config set-token ", file=sys.stderr) + elif e.code == "FORBIDDEN": + print("Access denied. Admin privileges required.", file=sys.stderr) + elif e.code == "CONNECTION_ERROR": + print("Could not connect to the registry.", file=sys.stderr) + else: + print(f"Error: {e.message}", file=sys.stderr) + return 1 + except Exception as e: + print(f"Error: {e}", file=sys.stderr) + return 1 + + return 0 + + +def _config_list(client, as_json, category=None): + """List all settings.""" + # Use the admin settings endpoint + response = client._request("GET", "/admin/settings") + settings = response.get("settings", []) + + # Filter by category if specified + if category: + settings = [s for s in settings if s.get("category") == category] + + if as_json: + print(json.dumps({"settings": settings}, indent=2)) + return 0 + + if not settings: + print("No settings found.") + return 0 + + # Group by category + by_category = {} + for s in settings: + cat = s.get("category", "general") + if cat not in by_category: + by_category[cat] = [] + by_category[cat].append(s) + + print("Registry Settings") + print("=" * 60) + + for cat, cat_settings in sorted(by_category.items()): + print(f"\n[{cat.upper()}]") + for s in cat_settings: + key = s.get("key", "") + value = s.get("value") + value_type = s.get("value_type", "string") + desc = s.get("description", "") + is_default = s.get("is_default", True) + + # Format value display + if value_type == "bool": + value_str = "true" if value else "false" + else: + value_str = str(value) + + status = "" if is_default else " (modified)" + print(f" {key}") + print(f" Value: {value_str}{status}") + if desc: + print(f" {desc}") + + print() + print("Use 'cmdforge registry config get ' to see a setting's value") + print("Use 'cmdforge registry config set ' to change a setting") + return 0 + + +def _config_get(client, key, as_json): + """Get a specific setting.""" + response = client._request("GET", f"/admin/settings/{key}") + + if as_json: + print(json.dumps(response, indent=2)) + return 0 + + setting = response.get("setting", {}) + print(f"Key: {setting.get('key', key)}") + print(f"Value: {setting.get('value')}") + print(f"Type: {setting.get('value_type', 'string')}") + print(f"Category: {setting.get('category', 'general')}") + if setting.get('description'): + print(f"Description: {setting['description']}") + if setting.get('updated_at'): + print(f"Last updated: {setting['updated_at'][:19]} by {setting.get('updated_by', 'system')}") + + return 0 + + +def _config_set(client, key, value): + """Set a setting value.""" + # Try to parse value as appropriate type + parsed_value = value + + # Try to parse as bool + if value.lower() in ("true", "false"): + parsed_value = value.lower() == "true" + # Try to parse as number + else: + try: + if "." in value: + parsed_value = float(value) + else: + parsed_value = int(value) + except ValueError: + # Keep as string + pass + + response = client._request("PUT", f"/admin/settings/{key}", json={"value": parsed_value}) + + if response.get("success"): + print(f"Setting '{key}' updated successfully.") + print(f"New value: {response.get('setting', {}).get('value', parsed_value)}") + else: + print(f"Failed to update setting: {response.get('error', 'Unknown error')}", file=sys.stderr) + return 1 + + return 0 diff --git a/src/cmdforge/registry/app.py b/src/cmdforge/registry/app.py index e469b91..c6f1caf 100644 --- a/src/cmdforge/registry/app.py +++ b/src/cmdforge/registry/app.py @@ -2932,6 +2932,163 @@ def create_app() -> Flask: "meta": paginate(page, per_page, total), }) + # ─── Admin Settings API ─────────────────────────────────────────────────────── + + @app.route("/api/v1/admin/settings", methods=["GET"]) + @require_admin + def admin_list_settings() -> Response: + """List all configurable settings with current values.""" + from .settings import get_all_settings + + category = request.args.get("category") + settings = get_all_settings(g.db) + + if category: + settings = [s for s in settings if s["category"] == category] + + # Group by category + categories = {} + for s in settings: + cat = s["category"] + if cat not in categories: + categories[cat] = [] + categories[cat].append(s) + + return jsonify({ + "data": settings, + "categories": categories, + "available_categories": list(categories.keys()), + }) + + @app.route("/api/v1/admin/settings/", methods=["GET"]) + @require_admin + def admin_get_setting(key: str) -> Response: + """Get a single setting value.""" + from .settings import get_setting, DEFAULT_SETTINGS + + value = get_setting(g.db, key) + if value is None: + return error_response("NOT_FOUND", f"Setting '{key}' not found", 404) + + # Find metadata + setting_meta = None + for s in DEFAULT_SETTINGS: + if s.key == key: + setting_meta = s + break + + return jsonify({ + "key": key, + "value": value, + "value_type": setting_meta.value_type if setting_meta else "string", + "description": setting_meta.description if setting_meta else "", + "category": setting_meta.category if setting_meta else "general", + }) + + @app.route("/api/v1/admin/settings/", methods=["PUT"]) + @require_admin + def admin_update_setting(key: str) -> Response: + """Update a setting value.""" + from .settings import set_setting, get_setting + + data = request.get_json() + if not data or "value" not in data: + return error_response("VALIDATION_ERROR", "Missing 'value' in request body") + + value = data["value"] + success = set_setting(g.db, key, value, updated_by=g.user_slug) + + if not success: + return error_response( + "VALIDATION_ERROR", + f"Invalid setting key '{key}' or invalid value", + 400, + ) + + # Log the change + log_audit( + "update_setting", + "setting", + key, + {"old_value": get_setting(g.db, key), "new_value": value}, + ) + + return jsonify({ + "success": True, + "key": key, + "value": get_setting(g.db, key), + }) + + @app.route("/api/v1/admin/settings/", methods=["DELETE"]) + @require_admin + def admin_reset_setting(key: str) -> Response: + """Reset a setting to its default value.""" + from .settings import reset_setting, get_setting, DEFAULT_SETTINGS + + # Check if it's a valid setting + valid = any(s.key == key for s in DEFAULT_SETTINGS) + if not valid: + return error_response("NOT_FOUND", f"Setting '{key}' not found", 404) + + reset_setting(g.db, key) + + log_audit("reset_setting", "setting", key, {}) + + return jsonify({ + "success": True, + "key": key, + "value": get_setting(g.db, key), + "message": "Setting reset to default", + }) + + @app.route("/api/v1/admin/settings/reset-all", methods=["POST"]) + @require_admin + def admin_reset_all_settings() -> Response: + """Reset all settings to defaults.""" + from .settings import reset_all_settings + + count = reset_all_settings(g.db) + + log_audit("reset_all_settings", "settings", "all", {"count": count}) + + return jsonify({ + "success": True, + "reset_count": count, + "message": f"Reset {count} settings to defaults", + }) + + @app.route("/api/v1/admin/settings/vetting", methods=["GET"]) + @require_moderator + def admin_get_vetting_config() -> Response: + """Get current vetting configuration (for moderators).""" + from .settings import get_vetting_config + + return jsonify(get_vetting_config(g.db)) + + @app.route("/api/v1/admin/settings/similarity", methods=["GET"]) + @require_moderator + def admin_get_similarity_config() -> Response: + """Get current similarity detection configuration.""" + from .settings import get_similarity_config + + return jsonify(get_similarity_config(g.db)) + + @app.route("/api/v1/admin/settings/sync", methods=["GET"]) + @require_admin + def admin_get_sync_config() -> Response: + """Get Fabric sync configuration.""" + from .settings import get_sync_config + + return jsonify(get_sync_config(g.db)) + + @app.route("/api/v1/admin/settings/moderation", methods=["GET"]) + @require_moderator + def admin_get_moderation_config() -> Response: + """Get moderation configuration.""" + from .settings import get_moderation_config + + return jsonify(get_moderation_config(g.db)) + # ─── Reviews & Ratings API ──────────────────────────────────────────────────── @app.route("/api/v1/tools///reviews", methods=["POST"]) diff --git a/src/cmdforge/registry/db.py b/src/cmdforge/registry/db.py index 9541173..52524dc 100644 --- a/src/cmdforge/registry/db.py +++ b/src/cmdforge/registry/db.py @@ -383,6 +383,19 @@ CREATE TABLE IF NOT EXISTS tool_usage ( ); CREATE INDEX IF NOT EXISTS idx_usage_tool ON tool_usage(tool_id); + +-- Admin Configuration Settings +CREATE TABLE IF NOT EXISTS registry_settings ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL, + value_type TEXT DEFAULT 'string', + description TEXT, + category TEXT DEFAULT 'general', + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_by TEXT +); + +CREATE INDEX IF NOT EXISTS idx_settings_category ON registry_settings(category); """ diff --git a/src/cmdforge/registry/settings.py b/src/cmdforge/registry/settings.py new file mode 100644 index 0000000..960b84a --- /dev/null +++ b/src/cmdforge/registry/settings.py @@ -0,0 +1,399 @@ +"""Registry settings management. + +Provides configurable settings for vetting, moderation, and sync. +Settings are stored in the database and can be modified via admin API. +""" + +from __future__ import annotations + +import json +import sqlite3 +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from .db import query_one, query_all + + +@dataclass +class Setting: + """A single configuration setting.""" + key: str + value: Any + value_type: str # string, int, float, bool, json + description: str + category: str + updated_at: Optional[str] = None + updated_by: Optional[str] = None + + +# Default settings with descriptions +DEFAULT_SETTINGS: List[Setting] = [ + # Vetting thresholds + Setting( + key="vetting.approve_threshold", + value=0.8, + value_type="float", + description="Quality score threshold for auto-approval (0.0-1.0)", + category="vetting", + ), + Setting( + key="vetting.reject_threshold", + value=0.3, + value_type="float", + description="Quality score threshold for auto-rejection (0.0-1.0)", + category="vetting", + ), + + # Vetting weights + Setting( + key="vetting.weight.honesty", + value=0.25, + value_type="float", + description="Weight for honesty check (description matches behavior)", + category="vetting", + ), + Setting( + key="vetting.weight.transparency", + value=0.20, + value_type="float", + description="Weight for transparency check (no obfuscation)", + category="vetting", + ), + Setting( + key="vetting.weight.scope", + value=0.15, + value_type="float", + description="Weight for scope check (appropriate boundaries)", + category="vetting", + ), + Setting( + key="vetting.weight.efficiency", + value=0.15, + value_type="float", + description="Weight for efficiency check (prompt quality)", + category="vetting", + ), + Setting( + key="vetting.weight.safety", + value=0.25, + value_type="float", + description="Weight for safety check (no dangerous patterns)", + category="vetting", + ), + + # Similarity detection + Setting( + key="similarity.duplicate_threshold", + value=0.9, + value_type="float", + description="Similarity score to consider tools as duplicates (0.0-1.0)", + category="similarity", + ), + Setting( + key="similarity.similar_threshold", + value=0.7, + value_type="float", + description="Similarity score to flag tools as similar (0.0-1.0)", + category="similarity", + ), + Setting( + key="similarity.related_threshold", + value=0.5, + value_type="float", + description="Similarity score to flag tools as related (0.0-1.0)", + category="similarity", + ), + + # Fabric sync + Setting( + key="sync.enabled", + value=False, + value_type="bool", + description="Enable scheduled Fabric pattern sync", + category="sync", + ), + Setting( + key="sync.interval_hours", + value=24, + value_type="int", + description="Hours between Fabric sync runs", + category="sync", + ), + Setting( + key="sync.auto_approve", + value=False, + value_type="bool", + description="Auto-approve synced Fabric patterns (skip moderation)", + category="sync", + ), + Setting( + key="sync.default_provider", + value="claude", + value_type="string", + description="Default AI provider for synced patterns", + category="sync", + ), + + # Moderation + Setting( + key="moderation.require_review", + value=True, + value_type="bool", + description="Require manual review for public tools", + category="moderation", + ), + Setting( + key="moderation.auto_approve_private", + value=True, + value_type="bool", + description="Auto-approve private/unlisted tools", + category="moderation", + ), + Setting( + key="moderation.max_pending_per_user", + value=10, + value_type="int", + description="Maximum pending tools per user before rate limiting", + category="moderation", + ), + + # Rate limits (can override defaults) + Setting( + key="rate_limit.publish.limit", + value=20, + value_type="int", + description="Max publish requests per window", + category="rate_limits", + ), + Setting( + key="rate_limit.publish.window", + value=3600, + value_type="int", + description="Publish rate limit window in seconds", + category="rate_limits", + ), +] + + +def _convert_value(value: str, value_type: str) -> Any: + """Convert stored string value to appropriate type.""" + if value_type == "int": + return int(value) + elif value_type == "float": + return float(value) + elif value_type == "bool": + return value.lower() in ("true", "1", "yes") + elif value_type == "json": + return json.loads(value) + return value + + +def _serialize_value(value: Any, value_type: str) -> str: + """Serialize value for storage.""" + if value_type == "bool": + return "true" if value else "false" + elif value_type == "json": + return json.dumps(value) + return str(value) + + +def get_setting(conn: sqlite3.Connection, key: str) -> Optional[Any]: + """Get a single setting value. + + Returns the default if not set in database. + """ + row = query_one( + conn, + "SELECT value, value_type FROM registry_settings WHERE key = ?", + [key], + ) + + if row: + return _convert_value(row["value"], row["value_type"]) + + # Return default if exists + for setting in DEFAULT_SETTINGS: + if setting.key == key: + return setting.value + + return None + + +def get_settings_by_category(conn: sqlite3.Connection, category: str) -> Dict[str, Any]: + """Get all settings for a category as a dict.""" + result = {} + + # Start with defaults + for setting in DEFAULT_SETTINGS: + if setting.category == category: + result[setting.key] = setting.value + + # Override with DB values + rows = query_all( + conn, + "SELECT key, value, value_type FROM registry_settings WHERE category = ?", + [category], + ) + for row in rows: + result[row["key"]] = _convert_value(row["value"], row["value_type"]) + + return result + + +def get_all_settings(conn: sqlite3.Connection) -> List[Dict[str, Any]]: + """Get all settings with their current values and metadata.""" + result = [] + + # Get all DB settings + db_settings = {} + rows = query_all(conn, "SELECT * FROM registry_settings") + for row in rows: + db_settings[row["key"]] = dict(row) + + # Build combined list + for default in DEFAULT_SETTINGS: + if default.key in db_settings: + row = db_settings[default.key] + result.append({ + "key": default.key, + "value": _convert_value(row["value"], row["value_type"]), + "value_type": row["value_type"], + "description": default.description, + "category": default.category, + "updated_at": row["updated_at"], + "updated_by": row["updated_by"], + "is_default": False, + }) + else: + result.append({ + "key": default.key, + "value": default.value, + "value_type": default.value_type, + "description": default.description, + "category": default.category, + "updated_at": None, + "updated_by": None, + "is_default": True, + }) + + return result + + +def set_setting( + conn: sqlite3.Connection, + key: str, + value: Any, + updated_by: Optional[str] = None, +) -> bool: + """Set a setting value. + + Returns True if successful, False if key is not a valid setting. + """ + # Find the setting definition + setting_def = None + for s in DEFAULT_SETTINGS: + if s.key == key: + setting_def = s + break + + if not setting_def: + return False + + # Validate and convert value + try: + if setting_def.value_type == "int": + value = int(value) + elif setting_def.value_type == "float": + value = float(value) + # Validate range for thresholds + if "threshold" in key or "weight" in key: + if not 0.0 <= value <= 1.0: + return False + elif setting_def.value_type == "bool": + if isinstance(value, str): + value = value.lower() in ("true", "1", "yes") + else: + value = bool(value) + except (ValueError, TypeError): + return False + + serialized = _serialize_value(value, setting_def.value_type) + now = datetime.now(timezone.utc).isoformat() + + conn.execute( + """ + INSERT INTO registry_settings (key, value, value_type, description, category, updated_at, updated_by) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(key) DO UPDATE SET + value = excluded.value, + updated_at = excluded.updated_at, + updated_by = excluded.updated_by + """, + [key, serialized, setting_def.value_type, setting_def.description, + setting_def.category, now, updated_by], + ) + conn.commit() + return True + + +def reset_setting(conn: sqlite3.Connection, key: str) -> bool: + """Reset a setting to its default value (remove from DB).""" + conn.execute("DELETE FROM registry_settings WHERE key = ?", [key]) + conn.commit() + return True + + +def reset_all_settings(conn: sqlite3.Connection) -> int: + """Reset all settings to defaults. Returns count of reset settings.""" + cursor = conn.execute("SELECT COUNT(*) FROM registry_settings") + count = cursor.fetchone()[0] + conn.execute("DELETE FROM registry_settings") + conn.commit() + return count + + +def get_vetting_config(conn: sqlite3.Connection) -> Dict[str, Any]: + """Get vetting configuration as a structured dict. + + This is the main entry point for the scrutiny system to get config. + """ + return { + "approve_threshold": get_setting(conn, "vetting.approve_threshold"), + "reject_threshold": get_setting(conn, "vetting.reject_threshold"), + "weights": { + "honesty": get_setting(conn, "vetting.weight.honesty"), + "transparency": get_setting(conn, "vetting.weight.transparency"), + "scope": get_setting(conn, "vetting.weight.scope"), + "efficiency": get_setting(conn, "vetting.weight.efficiency"), + "safety": get_setting(conn, "vetting.weight.safety"), + }, + } + + +def get_similarity_config(conn: sqlite3.Connection) -> Dict[str, float]: + """Get similarity detection configuration.""" + return { + "duplicate_threshold": get_setting(conn, "similarity.duplicate_threshold"), + "similar_threshold": get_setting(conn, "similarity.similar_threshold"), + "related_threshold": get_setting(conn, "similarity.related_threshold"), + } + + +def get_sync_config(conn: sqlite3.Connection) -> Dict[str, Any]: + """Get Fabric sync configuration.""" + return { + "enabled": get_setting(conn, "sync.enabled"), + "interval_hours": get_setting(conn, "sync.interval_hours"), + "auto_approve": get_setting(conn, "sync.auto_approve"), + "default_provider": get_setting(conn, "sync.default_provider"), + } + + +def get_moderation_config(conn: sqlite3.Connection) -> Dict[str, Any]: + """Get moderation configuration.""" + return { + "require_review": get_setting(conn, "moderation.require_review"), + "auto_approve_private": get_setting(conn, "moderation.auto_approve_private"), + "max_pending_per_user": get_setting(conn, "moderation.max_pending_per_user"), + } diff --git a/src/cmdforge/web/routes.py b/src/cmdforge/web/routes.py index fb274c5..1b4ab26 100644 --- a/src/cmdforge/web/routes.py +++ b/src/cmdforge/web/routes.py @@ -973,6 +973,17 @@ def _require_moderator_role(): return None +def _require_admin_role(): + """Check if current user has admin role.""" + redirect_response = _require_login() + if redirect_response: + return redirect_response + user = _load_current_publisher() + if not user or user.get("role") != "admin": + return render_template("errors/403.html"), 403 + return None + + @web_bp.route("/dashboard/admin", endpoint="admin_dashboard") def admin_dashboard(): """Admin dashboard overview.""" @@ -1077,3 +1088,47 @@ def admin_reports(): meta=meta, status_filter=status_filter, ) + + +@web_bp.route("/dashboard/admin/settings", endpoint="admin_settings") +def admin_settings(): + """Admin settings configuration page.""" + forbidden = _require_admin_role() + if forbidden: + return forbidden + + user = _load_current_publisher() + token = session.get("auth_token") + + status, payload = _api_get("/api/v1/admin/settings", token=token) + + if status != 200: + return render_template( + "admin/settings.html", + user=user, + active_page="admin_settings", + settings_by_category={}, + categories=[], + token=token, + error=payload.get("error", "Failed to load settings"), + ) + + settings = payload.get("data", []) + categories = payload.get("available_categories", []) + + # Group settings by category + settings_by_category = {} + for s in settings: + cat = s.get("category", "general") + if cat not in settings_by_category: + settings_by_category[cat] = [] + settings_by_category[cat].append(s) + + return render_template( + "admin/settings.html", + user=user, + active_page="admin_settings", + settings_by_category=settings_by_category, + categories=categories, + token=token, + ) diff --git a/src/cmdforge/web/templates/admin/index.html b/src/cmdforge/web/templates/admin/index.html index 988fa75..8dd5734 100644 --- a/src/cmdforge/web/templates/admin/index.html +++ b/src/cmdforge/web/templates/admin/index.html @@ -83,6 +83,15 @@ Manage publishers + {% if user.role == 'admin' %} + + + + + + Registry settings + + {% endif %} diff --git a/src/cmdforge/web/templates/admin/settings.html b/src/cmdforge/web/templates/admin/settings.html new file mode 100644 index 0000000..4a38a58 --- /dev/null +++ b/src/cmdforge/web/templates/admin/settings.html @@ -0,0 +1,231 @@ +{% extends "dashboard/base.html" %} + +{% block dashboard_header %} +
+
+

Registry Settings

+

Configure vetting, moderation, and sync settings

+
+
+ +
+
+{% endblock %} + +{% block dashboard_content %} +
+ +
+ +
+ + + {% for cat, cat_settings in settings_by_category.items() %} +
+
+
+

{{ cat.replace('_', ' ') }} Settings

+
+
+ {% for setting in cat_settings %} +
+
+

{{ setting.key.split('.')[-1].replace('_', ' ').title() }}

+

{{ setting.description }}

+

+ Key: {{ setting.key }} + {% if setting.updated_at %} + Last updated: {{ setting.updated_at[:10] }} by {{ setting.updated_by or 'system' }} + {% endif %} +

+
+
+ {% if setting.value_type == 'bool' %} + + {% elif setting.value_type == 'float' %} +
+ + +
+ {% elif setting.value_type == 'int' %} + + {% else %} + + {% endif %} + +
+
+ {% endfor %} +
+
+
+ {% endfor %} + + +
+
+
+ + + +
+
+

About these settings

+
+
    +
  • Vetting: Controls how tools are automatically scored and approved/rejected
  • +
  • Similarity: Thresholds for detecting duplicate or similar tools
  • +
  • Sync: Fabric pattern synchronization settings
  • +
  • Moderation: Rules for tool review workflow
  • +
  • Rate Limits: API rate limiting configuration
  • +
+
+
+
+
+
+ + +{% endblock %}