rob/CmdForge
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add admin configuration system and M3 content automation

Browse Source 
Admin Configuration System:
- Add registry_settings table to database schema
- Create settings.py module with 17 configurable settings across 5 categories
  (vetting, similarity, sync, moderation, rate_limits)
- Add admin API endpoints for settings management (list/get/set/reset)
- Create admin settings web UI page with tabbed categories
- Add CLI commands: cmdforge registry config list/get/set
- Wire settings page link into admin index

M3 Content Automation Scripts:
- scrutiny.py: Quality scoring with 5 weighted checks
- similarity.py: TF-IDF based duplicate detection
- vet_pipeline.py: Combined vetting with auto-decision logic
- fabric_sync.py: Scheduled Fabric pattern synchronization
- fabric-sync.service/timer: systemd units for scheduled sync

Settings available:
- Vetting thresholds (approve/reject) and check weights
- Similarity detection thresholds (duplicate/similar/related)
- Fabric sync configuration (enabled, interval, auto-approve)
- Moderation settings (require review, auto-approve private)
- Rate limiting configuration

Co-Authored-By: Claude Opus 4.5 <[email protected]>
This commit is contained in:
rob
2026-01-15 15:20:21 -04:00
co-authored by Claude Opus 4.5
parent d1eb4231a7
commit a5e0948881
14 changed files with 2958 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/fabric-sync.service
+29
View File
@@ -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
scripts/fabric-sync.timer
+14
View File
@@ -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
scripts/fabric_sync.py
Executable
+556
View File
@@ -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-dir>/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())
scripts/scrutiny.py
Executable
+507
View File
@@ -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())
scripts/similarity.py
Executable
+441
View File
@@ -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())
scripts/vet_pipeline.py
Executable
+379
View File
@@ -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())