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

Modernize provider support

This commit is contained in:
rob 2026-07-19 20:58:04 -03:00
parent 15943f251c
commit 6908d8ee6f
25 changed files with 1749 additions and 581 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -45,6 +45,8 @@ discussions/
diagrams/
*.puml
.tmp_*
.codex
.envrc
# Documentation symlink (points to project-docs)
docs

2
AGENTS.md
View File

@ -22,7 +22,7 @@ cf # Interactive tool picker
- `cli/` - Routes all subcommands (list, create, run, test, providers, registry, collections, deps, install, etc.)
- `tool.py` - Tool/step dataclasses, YAML loading, wrapper generation
- `runner.py` - Step execution, variable substitution (`{input}`, `{varname}`)
- `providers.py` - AI provider abstraction (calls CLI tools via subprocess)
- `providers.py` - AI provider abstraction (subprocess, API, PTY types; auto-discovery; fallback chains)
- `gui/` - PySide6 desktop GUI with page-based navigation
- `web/` - Flask web UI and forum
- `registry/` - Flask registry API (search, publish, moderation)

31
CLAUDE.md
View File

@ -44,7 +44,7 @@ python -m cmdforge.cli # Alternative CLI invocation
- **runner.py**: Execution engine. Runs tool steps sequentially, handles variable substitution (`{input}`, `{varname}`), executes Python code steps via `exec()`, handles nested tool calls with depth limit (MAX_TOOL_DEPTH=10)
- **resolver.py**: Tool resolution. `resolve_tool()` searches: project manifest → local tools → owner/name → registry. Returns `ResolvedTool` with path info
- **collection.py**: Collection management (`Collection` dataclass, `resolve_tool_references()`, `classify_tool_reference()`), local collection storage in `~/.cmdforge/collections/`
- **providers.py**: Provider abstraction. Calls AI CLI tools via subprocess, reads provider configs from `~/.cmdforge/providers.yaml`
- **providers.py**: Provider abstraction. Supports subprocess CLI tools, OpenAI-compatible HTTP APIs, and experimental PTY wrappers. Auto-discovers installed providers on first run. Config in `~/.cmdforge/providers.yaml` with versioned migration.
- **profiles.py**: AI persona profiles with system prompts, stored in `~/.cmdforge/profiles/`
- **manifest.py**: Project manifest (`cmdforge.yaml`) for declaring tool dependencies with version constraints
- **lockfile.py**: Lock file support for reproducible installs (`cmdforge.lock`)
@ -100,31 +100,48 @@ Variable substitution is handled in `runner.py:substitute_variables()`. Settings
## Provider System
Providers are CLI tools that accept prompts via stdin and output to stdout. Defined in `~/.cmdforge/providers.yaml`:
Providers wrap AI CLIs or compatible HTTP APIs. Defined in `~/.cmdforge/providers.yaml`:
```yaml
version: 2
providers:
- name: claude
command: "claude -p"
description: "Anthropic Claude"
fallback: claude-haiku # Optional: try this provider if primary fails
type: subprocess
- name: openrouter
command: "https://openrouter.ai/api/v1"
description: "OpenRouter API"
type: api
model: openrouter/auto-beta
api_key_env: OPENROUTER_API_KEY
- name: mock
command: "echo '[MOCK]'"
command: "mock"
description: "Mock provider for testing"
```
Provider fields:
- `name`: Provider identifier used in tool configs
- `command`: Shell command to invoke (prompt sent via stdin)
- `command`: Shell command (subprocess/pty) or endpoint URL (api)
- `type`: `subprocess` (default), `api` (OpenAI-compatible HTTP), or `pty` (interactive CLI, experimental)
- `model`: Model ID for api-type providers
- `api_key_env`: Environment variable holding the API key for api-type providers
- `description`: Optional human-readable description
- `fallback`: Optional provider to try if this one fails (prevents infinite loops)
- `fallback`: Optional provider to try if this one fails
- `fallback_chain`: Ordered list of providers for multi-step fallback
- `tags`: List of strings (e.g. `["free", "code", "local"]`)
- `pty_config`: Dict with `prompt_pattern`, `response_pattern`, `exit_command` for pty providers
- `install`: Optional structured install metadata dict
The `mock` provider is built-in for testing without API calls. Use `--provider mock` or `--dry-run` flags when testing tools.
Provider CLI commands:
- `cmdforge providers list` - List all providers and their status
- `cmdforge providers add <name> <command>` - Add/update a provider
- `cmdforge providers check` - Check which providers are available
- `cmdforge providers add <name> <command>` - Add/update a provider (supports `--type`, `--model`, `--api-key-env`, `--tag`, `--fallback`, `--fallback-chain`)
- `cmdforge providers remove <name>` - Remove a provider
- `cmdforge providers test <name>` - Test a provider
- `cmdforge providers discover [--add]` - Scan system for installed CLIs and API keys
- `cmdforge providers install` - Interactive guide to install AI providers
- `cmdforge providers for-tools <tool> [tools...]` - List providers used by tools (with `--warm` to pre-load local models)

11
pyproject.toml
View File

@ -42,6 +42,7 @@ dependencies = [
dev = [
"pytest>=7.0",
"pytest-cov>=4.0",
"tomli>=1.1; python_version < '3.11'",
]
registry = [
"Flask>=2.3",
@ -53,6 +54,9 @@ flow = [
"NodeGraphQt-QuiltiX-fork[pyside6]>=0.7.0",
"setuptools", # Required for distutils compatibility
]
pty = [
"pexpect>=4.8", # For PTY providers (interactive CLIs like Aider)
]
all = [
"Flask>=2.3",
"argon2-cffi>=21.0",
@ -60,6 +64,7 @@ all = [
"gunicorn>=21.0",
"NodeGraphQt-QuiltiX-fork[pyside6]>=0.7.0",
"setuptools",
"pexpect>=4.8",
]
[project.scripts]
@ -75,6 +80,12 @@ Issues = "https://gitea.brrd.tech/rob/CmdForge/issues"
[tool.setuptools.packages.find]
where = ["src"]
[tool.setuptools.package-data]
"cmdforge.web" = [
"templates/**/*.html",
"static/**/*",
]
[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src"]

23
src/cmdforge/cli/__init__.py
View File

@ -62,17 +62,16 @@ def main():
p_test.set_defaults(func=cmd_test)
# 'run' command
# NOTE: Options like --provider must come BEFORE the tool name due to argparse.REMAINDER
# Example: cmdforge run --provider mock my-tool (not: cmdforge run my-tool --provider mock)
p_run = subparsers.add_parser("run", help="Run a tool (options must come before tool name)")
p_run = subparsers.add_parser("run", help="Run a tool")
p_run.add_argument("name", help="Tool name")
p_run.add_argument("-i", "--input", help="Input file (reads from stdin if piped)")
p_run.add_argument("-o", "--output", help="Output file (writes to stdout if omitted)")
p_run.add_argument("--stdin", action="store_true", help="Read input interactively (type then Ctrl+D)")
p_run.add_argument("-p", "--provider", help="Override provider (must come before tool name)")
p_run.add_argument("-p", "--provider", help="Override provider")
p_run.add_argument("--dry-run", action="store_true", help="Show what would happen without executing")
p_run.add_argument("--show-prompt", action="store_true", help="Show prompts in addition to output")
p_run.add_argument("-v", "--verbose", action="store_true", help="Show debug information")
p_run.add_argument("--auto-install", action="store_true", help="Automatically install missing tool dependencies")
p_run.add_argument("tool_args", nargs=argparse.REMAINDER, help="Additional tool-specific arguments (use -- to separate)")
p_run.set_defaults(func=cmd_run)
@ -111,11 +110,25 @@ def main():
p_prov_install = providers_sub.add_parser("install", help="Interactive guide to install AI providers")
p_prov_install.set_defaults(func=cmd_providers)
# providers discover
p_prov_discover = providers_sub.add_parser("discover", help="Scan system for installed CLIs and API keys")
p_prov_discover.add_argument("--add", action="store_true", help="Add newly discovered providers to the config")
p_prov_discover.set_defaults(func=cmd_providers)
# providers add
p_prov_add = providers_sub.add_parser("add", help="Add or update a provider")
p_prov_add.add_argument("name", help="Provider name")
p_prov_add.add_argument("command", help="Command to run (e.g., 'claude -p')")
p_prov_add.add_argument("command", help="Command to run, or base URL for API providers")
p_prov_add.add_argument("-d", "--description", help="Provider description")
p_prov_add.add_argument("--type", choices=["subprocess", "api", "pty"], help="Provider type")
p_prov_add.add_argument("--model", help="Model ID for API providers")
p_prov_add.add_argument("--api-key-env", help="Environment variable containing the API key")
p_prov_add.add_argument("--tag", action="append", dest="tags", help="Provider tag (repeatable)")
p_prov_add.add_argument("--fallback", help="Fallback provider name; pass an empty value to clear")
p_prov_add.add_argument(
"--fallback-chain",
help="Preset name or comma-separated provider names; pass an empty value to clear",
)
p_prov_add.set_defaults(func=cmd_providers)
# providers remove

295
src/cmdforge/cli/provider_commands.py
View File

@ -1,62 +1,111 @@
"""Provider management commands."""
import os
import shlex
import shutil
import subprocess
from pathlib import Path
from ..providers import load_providers, add_provider, delete_provider, Provider, call_provider
from ..providers import (
PRESET_CHAINS,
Provider,
add_provider,
call_provider,
delete_provider,
get_provider,
load_providers,
save_providers,
)
PROVIDER_INSTALL_INFO = {
"opencode": {
"group": "OpenCode (6+ FREE models)",
"install_cmd": "curl -fsSL https://opencode.ai/install | bash",
"requires": "curl, bash",
"setup": "Run 'opencode' to connect API keys for paid models. Free models work without keys.",
"cost": "6 FREE models included (Big Pickle, DeepSeek V4 Flash, Nemotron, Mimo, etc.)",
"variants": ["opencode-pickle", "opencode-deepseek", "opencode-reasoner", "opencode-free"],
"post_install_note": "Test: cmdforge providers test opencode-pickle",
},
"claude": {
"group": "Anthropic Claude",
"install_cmd": "npm install -g @anthropic-ai/claude-code",
"requires": "Node.js 18+ and npm",
"setup": "Run 'claude' - opens browser for sign-in (auto-saves auth tokens)",
"cost": "Pay-per-use (billed to your Anthropic account)",
"install_cmd": "curl -fsSL https://claude.ai/install.sh | bash",
"requires": "curl, bash (or: brew install --cask claude-code)",
"setup": "Run 'claude' - opens browser for sign-in (subscription or API key required)",
"cost": "Paid subscription or API key. Non-US users may not access frontier models.",
"variants": ["claude", "claude-haiku", "claude-opus", "claude-sonnet"],
},
"codex": {
"group": "OpenAI Codex",
"install_cmd": "npm install -g @openai/codex",
"requires": "Node.js 18+ and npm",
"setup": "Run 'codex' - opens browser for sign-in (auto-saves auth tokens)",
"cost": "Pay-per-use (billed to your OpenAI account)",
"requires": "Node.js 18+ and npm (or: brew install --cask codex)",
"setup": "Run 'codex' - opens browser for sign-in. ChatGPT Plus/Pro includes Codex access.",
"cost": "Free tier available (ChatGPT Free). ChatGPT Plus/Pro includes more usage.",
"variants": ["codex"],
},
"gemini": {
"group": "Google Gemini",
"install_cmd": "npm install -g @google/gemini-cli",
"requires": "Node.js 18+ and npm",
"setup": "Run 'gemini' - opens browser for Google sign-in",
"cost": "Free tier available, pay-per-use for more",
"variants": ["gemini", "gemini-flash"],
"agy": {
"group": "Google Antigravity (replaces Gemini CLI)",
"install_cmd": None,
"requires": "Google Antigravity desktop application",
"setup": "Install from https://antigravity.google/download, then run 'agy install' to configure your PATH.",
"cost": "FREE tier (1,000 req/day). Paid Code Assist license for higher limits.",
"variants": ["agy"],
"binary_name": "agy",
"post_install_note": "Test: cmdforge providers test agy",
},
"opencode": {
"group": "OpenCode (75+ providers)",
"install_cmd": "curl -fsSL https://opencode.ai/install | bash",
"requires": "curl, bash",
"setup": "Run 'opencode' - opens browser to connect more providers",
"cost": "4 FREE models included (Big Pickle, GLM-4.7, Grok Code Fast 1, MiniMax M2.1), 75+ more available",
"variants": ["opencode-pickle", "opencode-deepseek", "opencode-nano", "opencode-reasoner", "opencode-grok"],
"crush": {
"group": "Crush (Charm) - multi-model agent",
"install_cmd": "npm install -g @charmland/crush",
"requires": "Node.js 18+ and npm (or go install, brew, AUR)",
"setup": "Run 'crush' to configure API keys for your preferred providers (OpenAI, Anthropic, Gemini, etc.)",
"cost": "Hyper free tier (limited credits). Use your own API keys for unlimited access via crush config.",
"variants": ["crush"],
"post_install_note": "Test: cmdforge providers test crush\nYou may need to configure API keys in ~/.config/crush/crush.json or subscribe to Hyper.",
},
"ollama": {
"group": "Ollama (Local LLMs)",
"install_cmd": "curl -fsSL https://ollama.ai/install.sh | bash",
"requires": "curl, bash, 8GB+ RAM (GPU recommended)",
"setup": "Run 'ollama pull llama3' to download a model, then add provider",
"setup": "Run 'ollama pull llama3.2' to download a model, then add provider",
"cost": "FREE (runs entirely on your machine)",
"variants": [],
"custom": True,
"post_install_note": "After installing, add the provider:\n cmdforge providers add ollama 'ollama run llama3' -d 'Local Llama 3'",
"post_install_note": "After installing, pull a model and add the provider:\n ollama pull llama3.2\n cmdforge providers add ollama 'ollama run llama3.2' -d 'Local Llama 3.2'",
},
}
def _provider_status(provider):
"""Return (available, status) without trying to call the provider."""
if provider.name.lower() == "mock":
return True, "OK (built-in)"
if provider.type == "api":
env_var = provider.api_key_env or f"{provider.name.upper().replace('-', '_')}_API_KEY"
if os.environ.get(env_var):
return True, f"OK (API key set: {env_var})"
return False, f"API KEY NOT SET ({env_var})"
try:
parts = shlex.split(os.path.expandvars(provider.command))
except ValueError:
return False, "INVALID COMMAND"
if not parts:
return False, "INVALID COMMAND"
executable = os.path.expanduser(parts[0])
if shutil.which(executable) or Path(executable).is_file():
return True, "OK"
return False, f"NOT FOUND ({parts[0]})"
def cmd_providers(args):
"""Manage AI providers."""
if args.providers_cmd == "install":
return _cmd_providers_install(args)
elif args.providers_cmd == "discover":
return _cmd_providers_discover(args)
elif args.providers_cmd == "list":
return _cmd_providers_list(args)
elif args.providers_cmd == "add":
@ -83,11 +132,10 @@ def _cmd_providers_install(args):
providers = load_providers()
installed_groups = set()
for p in providers:
if p.name.lower() == "mock":
if p.name.lower() == "mock" or p.type == "api":
continue
cmd_parts = p.command.split()[0]
cmd_expanded = cmd_parts.replace("$HOME", str(Path.home())).replace("~", str(Path.home()))
if shutil.which(cmd_expanded) or Path(cmd_expanded).exists():
available, _ = _provider_status(p)
if available:
# Find which group this belongs to
for group, info in PROVIDER_INSTALL_INFO.items():
if p.name in info.get("variants", []):
@ -127,10 +175,18 @@ def _cmd_providers_install(args):
print("=" * 60)
print()
print(f"Requirements: {info['requires']}")
print(f"Install command: {info['install_cmd']}")
if info["install_cmd"]:
print(f"Install command: {info['install_cmd']}")
else:
print("Install method: Manual download")
print(f"Post-install: {info['setup']}")
print()
if not info["install_cmd"]:
print("CmdForge cannot install this provider automatically.")
print(info["setup"])
return 0
try:
confirm = input("Run installation command? (y/N): ").strip().lower()
except EOFError:
@ -205,30 +261,107 @@ def _cmd_providers_install(args):
return 0
def _cmd_providers_discover(args):
"""Scan the system for installed AI CLIs and API keys."""
from ..providers import discover_installed_providers
print("=" * 60)
print("CmdForge Provider Discovery")
print("=" * 60)
print()
print("Scanning PATH for installed CLIs and environment for API keys...\n")
found = discover_installed_providers()
if not found:
print("No AI providers detected on this system.")
print("\nTo get started, install one of these free providers:")
print(" opencode: curl -fsSL https://opencode.ai/install | bash")
print(" agy: https://antigravity.google/download (then run 'agy install')")
print(" codex: npm install -g @openai/codex")
print(" crush: npm install -g @charmland/crush")
print(" ollama: curl -fsSL https://ollama.ai/install.sh | bash")
print("\nThen run: cmdforge providers discover")
return 0
# Group by source
cli_providers = [p for p in found if p["source"] == "cli"]
api_providers = [p for p in found if p["source"] == "api-key"]
ollama_models = [p for p in found if p["source"] == "ollama-model"]
if cli_providers:
print(f"Installed CLIs ({len(cli_providers)}):")
for p in cli_providers:
print(f" [+] {p['binary']:12s} {p['path']}")
print(f" -> provider: {p['name']}")
print()
if api_providers:
print(f"API keys detected ({len(api_providers)}):")
for p in api_providers:
print(f" [+] {p['env_var']}")
print(f" -> provider: {p['name']} (model: {p.get('model', 'auto')})")
print()
if ollama_models:
# Only show first 10 to avoid flooding output
shown = ollama_models[:10]
print(f"Local Ollama models ({len(ollama_models)} total, showing first 10):")
for p in shown:
print(f" [+] {p['command']}")
if len(ollama_models) > 10:
print(f" ... and {len(ollama_models) - 10} more")
print()
print(f"Total discovered: {len(found)} provider(s)")
print()
if getattr(args, "add", False):
providers = load_providers()
configured = {provider.name for provider in providers}
added = []
for item in found:
if item["name"] in configured:
continue
providers.append(Provider(
name=item["name"],
command=item["command"],
description=item.get("description", ""),
type=item.get("type", "subprocess"),
model=item.get("model"),
tags=item.get("tags", []),
api_key_env=item.get("env_var"),
))
configured.add(item["name"])
added.append(item["name"])
if added:
save_providers(providers)
print(f"Added {len(added)} provider(s): {', '.join(added)}")
else:
print("All discovered providers are already configured.")
print()
else:
print("To add all newly discovered providers:")
print(" cmdforge providers discover --add")
print()
print("To test a provider:")
print(" cmdforge providers test <name>")
return 0
def _cmd_providers_list(args):
"""List all providers and their status."""
providers = load_providers()
print(f"Configured providers ({len(providers)}):\n")
for p in providers:
# Mock provider is always available
if p.name.lower() == "mock":
print(f" [+] {p.name}")
print(f" Command: (built-in)")
print(f" Status: OK (always available)")
if p.description:
print(f" Info: {p.description}")
print()
continue
# Check if command exists
cmd_parts = p.command.split()[0]
cmd_expanded = cmd_parts.replace("$HOME", str(Path.home())).replace("~", str(Path.home()))
exists = shutil.which(cmd_expanded) is not None or Path(cmd_expanded).exists()
status = "OK" if exists else "NOT FOUND"
exists, status = _provider_status(p)
status_icon = "+" if exists else "-"
print(f" [{status_icon}] {p.name}")
print(f" Command: {p.command}")
label = "Endpoint" if p.type == "api" else "Command"
value = "(built-in)" if p.name.lower() == "mock" else p.command
print(f" {label}: {value}")
print(f" Status: {status}")
if p.description:
print(f" Info: {p.description}")
@ -240,9 +373,40 @@ def _cmd_providers_add(args):
"""Add or update a provider."""
name = args.name
command = args.command
description = args.description or ""
existing = get_provider(name)
description = args.description if args.description is not None else (existing.description if existing else "")
provider_type = args.type or (existing.type if existing else "subprocess")
model = args.model if args.model is not None else (existing.model if existing else None)
api_key_env = args.api_key_env if args.api_key_env is not None else (existing.api_key_env if existing else None)
tags = args.tags if args.tags is not None else (existing.tags if existing else [])
fallback = args.fallback if args.fallback is not None else (existing.fallback if existing else None)
fallback = fallback or None
provider = Provider(name, command, description)
if args.fallback_chain is None:
fallback_chain = existing.fallback_chain if existing else None
elif args.fallback_chain in PRESET_CHAINS:
fallback_chain = list(PRESET_CHAINS[args.fallback_chain])
else:
fallback_chain = [item.strip() for item in args.fallback_chain.split(",") if item.strip()]
fallback_chain = fallback_chain or None
if fallback == name or (fallback_chain and name in fallback_chain):
print("A provider cannot fall back to itself.")
return 1
provider = Provider(
name=name,
command=command,
description=description,
fallback=fallback,
type=provider_type,
model=model,
tags=tags,
install=existing.install if existing else None,
fallback_chain=fallback_chain,
api_key_env=api_key_env,
pty_config=existing.pty_config if existing else None,
)
add_provider(provider)
print(f"Provider '{name}' added/updated.")
return 0
@ -279,10 +443,15 @@ def _cmd_providers_for_tools(args):
from ..resolver import resolve_tool, ToolNotFoundError
# Cloud providers don't need warming - they're always "warm"
CLOUD_PREFIXES = ('claude', 'opencode', 'gemini', 'codex', 'gpt', 'openai')
CLOUD_PREFIXES = ('claude', 'opencode', 'agy', 'codex', 'gpt', 'openai', 'openrouter', 'deepseek', 'crush')
def is_local_provider(name):
"""Check if provider is local (needs warming) vs cloud."""
provider = get_provider(name)
if provider and provider.type == "api":
return False
if provider and "local" in provider.tags:
return True
name_lower = name.lower()
return not any(name_lower.startswith(prefix) for prefix in CLOUD_PREFIXES)
@ -312,14 +481,6 @@ def _cmd_providers_for_tools(args):
# Default to medium size
return 7.0
def get_provider(name):
"""Get provider by name."""
providers = load_providers()
for p in providers:
if p.name == name:
return p
return None
tool_names = args.tools
providers_used = set()
@ -390,29 +551,23 @@ def _cmd_providers_check(args):
missing = []
for p in providers:
# Mock provider is always available (handled specially)
if p.name.lower() == "mock":
available.append(p.name)
print(f" [+] {p.name}: OK (built-in)")
continue
cmd_parts = p.command.split()[0]
cmd_expanded = cmd_parts.replace("$HOME", str(Path.home())).replace("~", str(Path.home()))
exists = shutil.which(cmd_expanded) is not None or Path(cmd_expanded).exists()
exists, status = _provider_status(p)
if exists:
available.append(p.name)
print(f" [+] {p.name}: OK")
print(f" [+] {p.name}: {status}")
else:
missing.append(p.name)
print(f" [-] {p.name}: NOT FOUND ({cmd_parts})")
print(f" [-] {p.name}: {status}")
print(f"\nSummary: {len(available)} available, {len(missing)} missing")
if len(available) == 1 and available[0] == "mock":
print(f"\nNo real AI providers found. Install one of these:")
print(f" - claude: npm install -g @anthropic-ai/claude-cli")
print(f" - codex: pip install openai-codex")
print(f" - gemini: pip install google-generative-ai")
print(f" - opencode: curl -fsSL https://opencode.ai/install | bash")
print(f" - agy: https://antigravity.google/download, then run 'agy install'")
print(f" - codex: npm install -g @openai/codex (FREE tier available)")
print(f" - crush: npm install -g @charmland/crush")
print(f" - ollama: curl -fsSL https://ollama.ai/install.sh | bash (local, FREE)")
print(f"\nMeanwhile, use mock provider for testing:")
print(f" echo 'test' | summarize --provider mock")
elif missing:

114
src/cmdforge/cli/tool_commands.py
View File

@ -181,22 +181,30 @@ def cmd_test(args):
def cmd_run(args):
"""Run a tool."""
from ..runner import run_tool
from ..runner import collect_custom_args, create_argument_parser, run_tool
tool = load_tool(args.name)
if not tool:
print(f"Error: Tool '{args.name}' not found.", file=sys.stderr)
return 1
tool_args = list(args.tool_args or [])
if tool_args and tool_args[0] == '--':
tool_args = tool_args[1:]
tool_parser = create_argument_parser(tool)
parsed_tool_args = tool_parser.parse_args(tool_args)
# Read input
if args.input:
input_file = args.input or parsed_tool_args.input_file
if input_file:
# Read from file
input_path = Path(args.input)
input_path = Path(input_file)
if not input_path.exists():
print(f"Error: Input file not found: {args.input}", file=sys.stderr)
print(f"Error: Input file not found: {input_file}", file=sys.stderr)
return 1
input_text = input_path.read_text()
elif args.stdin:
elif args.stdin or parsed_tool_args.stdin:
# Explicit interactive input requested
print("Reading from stdin (Ctrl+D to end):", file=sys.stderr)
input_text = sys.stdin.read()
@ -207,98 +215,32 @@ def cmd_run(args):
# No input provided - use empty string
input_text = ""
# Collect custom args from remaining arguments
custom_args = {}
# Collect custom args from the same parser used by wrapper scripts.
custom_args = collect_custom_args(tool, parsed_tool_args)
# Build a map from flag names to variable names for proper argument mapping
flag_to_var = {}
for tool_arg in tool.arguments:
# Handle both --flag and -f style flags
flag = tool_arg.flag
if flag.startswith('--'):
flag_key = flag[2:].replace('-', '_')
elif flag.startswith('-'):
flag_key = flag[1:].replace('-', '_')
else:
flag_key = flag.replace('-', '_')
flag_to_var[flag_key] = tool_arg.variable
if args.tool_args:
# Remove leading '--' separator if present (used to separate cmdforge args from tool args)
tool_args = list(args.tool_args)
if tool_args and tool_args[0] == '--':
tool_args = tool_args[1:]
def is_flag(s: str) -> bool:
"""Check if string looks like a flag (not a negative number)."""
if not s.startswith('-'):
return False
# Check if it's a negative number (e.g., -5, -3.14)
rest = s[1:] if s.startswith('--') else s[1:]
if s.startswith('--'):
rest = s[2:]
try:
float(rest)
return False # It's a negative number, not a flag
except ValueError:
return True # It's a flag
# Parse tool-specific arguments
i = 0
while i < len(tool_args):
arg = tool_args[i]
if arg.startswith('--'):
# Handle --flag=value syntax
if '=' in arg:
flag_part, value = arg.split('=', 1)
flag_key = flag_part[2:].replace('-', '_')
var_name = flag_to_var.get(flag_key, flag_key)
custom_args[var_name] = value
i += 1
else:
flag_key = arg[2:].replace('-', '_')
var_name = flag_to_var.get(flag_key, flag_key)
if i + 1 < len(tool_args) and not is_flag(tool_args[i + 1]):
custom_args[var_name] = tool_args[i + 1]
i += 2
else:
custom_args[var_name] = True
i += 1
elif arg.startswith('-') and is_flag(arg):
# Handle -f=value syntax
if '=' in arg:
flag_part, value = arg.split('=', 1)
flag_key = flag_part[1:].replace('-', '_')
var_name = flag_to_var.get(flag_key, flag_key)
custom_args[var_name] = value
i += 1
else:
flag_key = arg[1:].replace('-', '_')
var_name = flag_to_var.get(flag_key, flag_key)
if i + 1 < len(tool_args) and not is_flag(tool_args[i + 1]):
custom_args[var_name] = tool_args[i + 1]
i += 2
else:
custom_args[var_name] = True
i += 1
else:
i += 1
provider_override = args.provider or parsed_tool_args.provider
dry_run = args.dry_run or parsed_tool_args.dry_run
show_prompt = args.show_prompt or parsed_tool_args.show_prompt
verbose = args.verbose or parsed_tool_args.verbose
auto_install = args.auto_install or parsed_tool_args.auto_install
# Run tool
output, code = run_tool(
tool=tool,
input_text=input_text,
custom_args=custom_args,
provider_override=args.provider,
dry_run=args.dry_run,
show_prompt=args.show_prompt,
verbose=args.verbose
provider_override=provider_override,
dry_run=dry_run,
show_prompt=show_prompt,
verbose=verbose,
auto_install=auto_install
)
# Write output
if code == 0 and output:
if args.output:
Path(args.output).write_text(output)
output_file = args.output or parsed_tool_args.output_file
if output_file:
Path(output_file).write_text(output)
else:
print(output)

38
src/cmdforge/collection.py
View File

@ -127,42 +127,22 @@ class ToolResolutionResult:
def _get_tool_visibility_from_yaml(tool_name: str, my_owner: str = None) -> str:
"""
Read visibility directly from tool's config.yaml file.
NOTE: The Tool dataclass doesn't have a visibility field, so we read
the raw YAML to check this. This is a workaround until Tool model is updated.
"""Read visibility from the Tool dataclass.
Args:
tool_name: The tool name (unqualified)
my_owner: Current user's slug for resolving owned tool paths
"""
from .tool import get_tools_dir
from .tool import load_tool
tools_dir = get_tools_dir()
# Try owned path first if my_owner is set
config_path = None
if my_owner:
owned_path = tools_dir / my_owner / tool_name / "config.yaml"
if owned_path.exists():
config_path = owned_path
# Fall back to unqualified path
if not config_path:
unqualified_path = tools_dir / tool_name / "config.yaml"
if unqualified_path.exists():
config_path = unqualified_path
if not config_path:
return "public" # Default if not found
try:
with open(config_path) as f:
config = yaml.safe_load(f) or {}
return config.get("visibility", "public")
except Exception:
return "public"
tool = load_tool(my_owner + "/" + tool_name)
if tool and tool.visibility:
return tool.visibility
tool = load_tool(tool_name)
if tool and tool.visibility:
return tool.visibility
return "public"
def resolve_tool_references(

10
src/cmdforge/config.py
View File

@ -75,7 +75,11 @@ class Config:
def get_config_dir() -> Path:
"""Get the config directory, creating it if needed."""
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
CONFIG_DIR.mkdir(parents=True, exist_ok=True, mode=0o700)
try:
CONFIG_DIR.chmod(0o700)
except OSError:
pass
return CONFIG_DIR
@ -101,6 +105,10 @@ def save_config(config: Config) -> Path:
"""Save configuration to disk."""
config_path = get_config_dir() / "config.yaml"
config_path.write_text(yaml.dump(config.to_dict(), default_flow_style=False, sort_keys=False))
try:
config_path.chmod(0o600)
except OSError:
pass
return config_path

16
src/cmdforge/gui/dialogs/provider_dialog.py
View File

@ -17,6 +17,7 @@ class ProviderDialog(QDialog):
self.setMinimumWidth(450)
self._editing = provider is not None
self._original_name = name
self._provider = provider
self._setup_ui()
if provider:
@ -114,7 +115,20 @@ class ProviderDialog(QDialog):
return
try:
add_provider(Provider(name, command, description, fallback))
existing = self._provider
add_provider(Provider(
name=name,
command=command,
description=description,
fallback=fallback,
type=existing.type if existing else "subprocess",
model=existing.model if existing else None,
tags=existing.tags if existing else [],
install=existing.install if existing else None,
fallback_chain=existing.fallback_chain if existing else None,
api_key_env=existing.api_key_env if existing else None,
pty_config=existing.pty_config if existing else None,
))
self.accept()
except Exception as e:
from PySide6.QtWidgets import QMessageBox

87
src/cmdforge/gui/dialogs/provider_install_dialog.py
View File

@ -19,59 +19,68 @@ from ...providers import load_providers, add_provider, Provider
PROVIDER_INSTALL_INFO = {
"claude": {
"group": "Anthropic Claude",
"install_cmd": "npm install -g @anthropic-ai/claude-code",
"requires": "Node.js 18+ and npm",
"setup": "Run 'claude' - opens browser for sign-in",
"cost": "Pay-per-use (billed to your Anthropic account)",
"install_cmd": "curl -fsSL https://claude.ai/install.sh | bash",
"requires": "curl and bash",
"setup": "Run 'claude' to sign in with a subscription or API key.",
"cost": "Paid subscription or API key",
"variants": [
("claude", "claude -p", "Claude (default model)"),
("claude-haiku", "claude -p --model claude-3-haiku-20240307", "Claude Haiku (fast, cheap)"),
("claude-sonnet", "claude -p --model claude-3-5-sonnet-20241022", "Claude Sonnet (balanced)"),
("claude-opus", "claude -p --model claude-3-opus-20240229", "Claude Opus (most capable)"),
("claude-haiku", "claude -p --model haiku", "Claude Haiku (fast)"),
("claude-sonnet", "claude -p --model sonnet", "Claude Sonnet (balanced)"),
("claude-opus", "claude -p --model opus", "Claude Opus (highest quality)"),
],
},
"codex": {
"group": "OpenAI Codex",
"install_cmd": "npm install -g @openai/codex",
"requires": "Node.js 18+ and npm",
"setup": "Run 'codex' - opens browser for sign-in",
"cost": "Pay-per-use (billed to your OpenAI account)",
"setup": "Run 'codex' to sign in.",
"cost": "Free tier or ChatGPT subscription, subject to account limits",
"variants": [
("codex", "codex", "OpenAI Codex"),
("codex", "codex exec -", "OpenAI Codex"),
],
},
"gemini": {
"group": "Google Gemini",
"install_cmd": "npm install -g @google/gemini-cli",
"requires": "Node.js 18+ and npm",
"setup": "Run 'gemini' - opens browser for Google sign-in",
"cost": "Free tier available, pay-per-use for more",
"agy": {
"group": "Google Antigravity",
"install_cmd": None,
"requires": "Google Antigravity desktop application",
"setup": "Install from https://antigravity.google/download, then run 'agy install' to configure your PATH.",
"cost": "Free tier available, subject to account limits",
"variants": [
("gemini", "gemini", "Gemini Pro (default)"),
("gemini-flash", "gemini --model gemini-1.5-flash", "Gemini Flash (faster)"),
("agy", "agy -p", "Google Antigravity (automatic model selection)"),
],
},
"opencode": {
"group": "OpenCode (75+ providers)",
"group": "OpenCode",
"install_cmd": "curl -fsSL https://opencode.ai/install | bash",
"requires": "curl, bash",
"setup": "Run 'opencode' - opens browser to connect providers",
"cost": "4 FREE models included, 75+ more available",
"setup": "Run 'opencode' and use /connect to configure providers.",
"cost": "Free models available; paid providers require credentials",
"variants": [
("opencode-pickle", "$HOME/.opencode/bin/opencode -p big-pickle", "Big Pickle (FREE, high quality)"),
("opencode-deepseek", "$HOME/.opencode/bin/opencode -p deepseek", "DeepSeek (fast, cheap)"),
("opencode-reasoner", "$HOME/.opencode/bin/opencode -p deepseek-reasoner", "DeepSeek Reasoner (complex tasks)"),
("opencode-grok", "$HOME/.opencode/bin/opencode -p grok", "Grok (xAI)"),
("opencode-pickle", "opencode run --model opencode/big-pickle", "Big Pickle (free)"),
("opencode-free", "opencode run --model opencode/deepseek-v4-flash-free", "DeepSeek V4 Flash (free)"),
("opencode-deepseek", "opencode run --model deepseek/deepseek-chat", "DeepSeek Chat (API key required)"),
("opencode-reasoner", "opencode run --model deepseek/deepseek-reasoner", "DeepSeek Reasoner (API key required)"),
],
},
"crush": {
"group": "Crush (Charm)",
"install_cmd": "npm install -g @charmland/crush",
"requires": "Node.js 18+ and npm",
"setup": "Run 'crush' to configure Hyper or your provider API keys.",
"cost": "Limited Hyper credits or your own provider billing",
"variants": [
("crush", "crush run --quiet", "Crush multi-model agent"),
],
},
"ollama": {
"group": "Ollama (Local LLMs)",
"install_cmd": "curl -fsSL https://ollama.ai/install.sh | bash",
"requires": "curl, bash, 8GB+ RAM (GPU recommended)",
"setup": "Run 'ollama pull llama3' to download a model",
"setup": "Run 'ollama pull llama3.2' to download a model.",
"cost": "FREE (runs entirely on your machine)",
"variants": [
("ollama-llama3", "ollama run llama3", "Llama 3 (general purpose)"),
("ollama-llama3", "ollama run llama3.2", "Llama 3.2 (general purpose)"),
("ollama-codellama", "ollama run codellama", "CodeLlama (coding)"),
("ollama-mistral", "ollama run mistral", "Mistral (fast)"),
],
@ -336,7 +345,7 @@ class ProviderInstallDialog(QDialog):
return opencode_path.exists() or shutil.which("opencode") is not None
elif provider_key == "ollama":
return shutil.which("ollama") is not None
elif provider_key in ("claude", "codex", "gemini"):
elif provider_key in ("claude", "codex", "agy", "crush"):
return shutil.which(provider_key) is not None
return False
@ -351,10 +360,11 @@ class ProviderInstallDialog(QDialog):
self.btn_install.setEnabled(True)
# Update details
install_method = info['install_cmd'] or "Manual download"
details = f"""<b>{info['group']}</b><br><br>
<b>Cost:</b> {info['cost']}<br>
<b>Requirements:</b> {info['requires']}<br>
<b>Install command:</b> <code>{info['install_cmd']}</code><br>
<b>Install method:</b> <code>{install_method}</code><br>
<b>After install:</b> {info['setup']}<br><br>
<b>Available models:</b> {', '.join(v[0] for v in info['variants'])}"""
@ -364,7 +374,10 @@ class ProviderInstallDialog(QDialog):
self.details_label.setText(details)
# Change button text based on installed status
self.btn_install.setText("Reinstall" if is_installed else "Install Selected")
if info["install_cmd"]:
self.btn_install.setText("Reinstall" if is_installed else "Install Selected")
else:
self.btn_install.setText("Add Providers" if is_installed else "Setup Instructions")
def _on_provider_double_clicked(self, item: QListWidgetItem):
"""Handle double-click to start install."""
@ -378,6 +391,14 @@ class ProviderInstallDialog(QDialog):
info = PROVIDER_INSTALL_INFO[self._selected_provider]
if not info["install_cmd"]:
if self._check_installed(self._selected_provider):
self._populate_variants()
self.stack.setCurrentIndex(2)
else:
QMessageBox.information(self, "Manual Setup", info["setup"])
return
# Switch to install page
self.stack.setCurrentIndex(1)
self.install_title.setText(f"Installing {info['group']}...")
@ -450,10 +471,8 @@ class ProviderInstallDialog(QDialog):
# Update next steps
self.next_steps_label.setText(
f"<b>Important:</b> Before using these providers, you need to authenticate:<br><br>"
f"1. Open a terminal<br>"
f"2. Run: <code>source ~/.bashrc</code> (to update PATH)<br>"
f"3. Run: <code>{info['setup'].split(' - ')[0].replace('Run ', '')}</code><br><br>"
f"This will open your browser to sign in."
f"1. Open a new terminal so PATH changes are loaded.<br>"
f"2. {info['setup']}"
)
def _add_selected_variants(self):

4
src/cmdforge/gui/pages/providers_page.py
View File

@ -49,8 +49,8 @@ class ProvidersPage(QWidget):
# Description
desc = QLabel(
"Providers are external AI commands that CmdForge tools can use. "
"Each provider wraps a CLI tool that accepts input on stdin and outputs to stdout."
"Providers connect CmdForge tools to AI CLIs or compatible HTTP APIs. "
"CLI providers accept input on stdin and write responses to stdout."
)
desc.setWordWrap(True)
desc.setStyleSheet("color: #718096;")

635
src/cmdforge/providers.py
View File

@ -26,13 +26,192 @@ def strip_ansi(text: str) -> str:
PROVIDERS_FILE = Path.home() / ".cmdforge" / "providers.yaml"
# Fallback chain templates for callers and manual providers.yaml configuration.
PRESET_CHAINS = {
"premium": ["claude-opus", "claude-sonnet", "claude-haiku", "codex", "crush", "mock"],
"free": ["opencode-free", "agy", "codex", "crush", "ollama", "mock"],
"fast": ["agy", "claude-haiku", "opencode-free", "crush", "mock"],
"reasoning": ["claude-opus", "opencode-reasoner", "deepseek-api", "crush", "mock"],
"balanced": ["opencode-pickle", "codex", "agy", "claude-haiku", "crush", "ollama", "mock"],
}
PROVIDERS_CONFIG_VERSION = 2
V2_DEFAULT_PROVIDER_NAMES = {
"opencode-free",
"agy",
"crush",
"ollama",
"openrouter",
"deepseek-api",
}
# Known CLIs that CmdForge can auto-discover on the user's PATH.
# Maps binary name -> default provider config (used during first-run setup).
KNOWN_PROVIDER_CLIS = {
"opencode": {
"name": "opencode-pickle",
"command": "opencode run --model opencode/big-pickle",
"description": "OpenCode - Big Pickle (free general model)",
"tags": ["free", "code", "general"],
"install_group": "opencode",
},
"agy": {
"name": "agy",
"command": "agy -p",
"description": "Antigravity - Google free tier, Gemini models",
"tags": ["free-tier", "code", "large-context"],
"install_group": "agy",
},
"codex": {
"name": "codex",
"command": "codex exec -",
"description": "Codex CLI - OpenAI free tier available",
"tags": ["free-tier", "code", "general"],
"install_group": "codex",
},
"claude": {
"name": "claude",
"command": "claude -p",
"description": "Claude Code - auto-routes to best model",
"tags": ["paid", "subscription", "code"],
"install_group": "claude",
},
"crush": {
"name": "crush",
"command": "crush run --quiet",
"description": "Crush - multi-model via Hyper credits or API keys",
"tags": ["free-tier", "multi", "code"],
"install_group": "crush",
},
"ollama": {
"name": "ollama",
"command": "ollama run llama3.2",
"description": "Ollama - local, private, free",
"tags": ["free", "local", "private"],
"install_group": "ollama",
},
}
# Known API key environment variables -> provider config
KNOWN_API_KEYS = {
"OPENROUTER_API_KEY": {
"name": "openrouter",
"command": "https://openrouter.ai/api/v1",
"model": "openrouter/auto-beta",
"description": "OpenRouter - 300+ models, one API key",
"tags": ["api", "per-token", "multi"],
},
"DEEPSEEK_API_KEY": {
"name": "deepseek-api",
"command": "https://api.deepseek.com/v1",
"model": "deepseek-chat",
"description": "DeepSeek API - inexpensive per-token access",
"tags": ["api", "per-token", "cheap"],
},
"OPENAI_API_KEY": {
"name": "openai-api",
"command": "https://api.openai.com/v1",
"model": "gpt-4o",
"description": "OpenAI API - direct GPT access",
"tags": ["api", "per-token", "code"],
},
}
def discover_installed_providers() -> List[dict]:
"""Scan the system for installed AI CLIs and configured API keys.
Returns a list of discovery results, each with keys:
- source: "cli" or "api-key"
- name: provider name
- command: provider command
- description, tags, etc.
"""
found = []
# Check PATH for known CLIs
for binary, info in KNOWN_PROVIDER_CLIS.items():
path = shutil.which(binary)
if path:
found.append({
"source": "cli",
"binary": binary,
"path": path,
**{k: v for k, v in info.items() if k != "install_group"},
})
# Check environment for API keys
for env_var, info in KNOWN_API_KEYS.items():
if os.environ.get(env_var):
found.append({
"source": "api-key",
"env_var": env_var,
"type": "api",
**info,
})
# Check Ollama for available local models (if installed).
if shutil.which("ollama"):
try:
result = subprocess.run(
["ollama", "list"],
capture_output=True,
text=True,
timeout=5,
)
if result.returncode == 0:
used_names = {item["name"] for item in found}
# Parse model list (skip header line).
for line in result.stdout.strip().split("\n")[1:]:
if line.strip():
model_name = line.split()[0]
slug = re.sub(r"[^a-z0-9]+", "-", model_name.lower()).strip("-")
provider_name = f"ollama-{slug}"
suffix = 2
while provider_name in used_names:
provider_name = f"ollama-{slug}-{suffix}"
suffix += 1
used_names.add(provider_name)
found.append({
"source": "ollama-model",
"binary": "ollama",
"name": provider_name,
"command": f"ollama run {model_name}",
"description": f"Ollama local model: {model_name}",
"tags": ["free", "local", "private"],
})
except (subprocess.TimeoutExpired, FileNotFoundError):
pass
return found
@dataclass
class Provider:
"""Definition of an AI provider."""
"""Definition of an AI provider.
Types:
- "subprocess" (default): CLI tool invoked via subprocess.run with stdin.
`command` is the shell command (e.g. "claude -p", "opencode run --model X").
- "api": HTTP POST to an OpenAI-compatible endpoint.
`command` is the endpoint URL, `model` is the model ID, `api_key_env`
names the environment variable holding the API key.
- "pty": Interactive CLI wrapped in a pseudo-terminal (experimental).
`command` is the CLI invocation; pty_config dict holds patterns.
"""
name: str
command: str
description: str = ""
fallback: Optional[str] = None # Name of fallback provider if this one fails
fallback: Optional[str] = None
type: str = "subprocess" # "subprocess" | "api" | "pty"
model: Optional[str] = None # Model ID (required for api-type providers)
tags: List[str] = field(default_factory=list) # e.g. ["free", "code", "reasoning"]
install: Optional[dict] = None # Structured install metadata
fallback_chain: Optional[List[str]] = None # Ordered multi-step fallback (new)
api_key_env: Optional[str] = None # Env var name for api-type providers
pty_config: Optional[dict] = None # Patterns for pty-type providers
def to_dict(self) -> dict:
d = {
@ -40,8 +219,23 @@ class Provider:
"command": self.command,
"description": self.description,
}
# Only include type if not default (preserves backward-compat with old YAML)
if self.type and self.type != "subprocess":
d["type"] = self.type
if self.model:
d["model"] = self.model
if self.fallback:
d["fallback"] = self.fallback
if self.tags:
d["tags"] = self.tags
if self.install:
d["install"] = self.install
if self.fallback_chain:
d["fallback_chain"] = self.fallback_chain
if self.api_key_env:
d["api_key_env"] = self.api_key_env
if self.pty_config:
d["pty_config"] = self.pty_config
return d
@classmethod
@ -50,7 +244,14 @@ class Provider:
name=data["name"],
command=data["command"],
description=data.get("description", ""),
type=data.get("type", "subprocess"),
model=data.get("model"),
fallback=data.get("fallback"),
tags=data.get("tags", []) or [],
install=data.get("install"),
fallback_chain=data.get("fallback_chain"),
api_key_env=data.get("api_key_env"),
pty_config=data.get("pty_config"),
)
@ -63,38 +264,141 @@ class ProviderResult:
# Default providers that come pre-configured
# Profiled with 4-task test: Math, Code, Reasoning, Extract (Dec 2025)
# Live-tested July 2026: opencode run, codex exec -, agy -p, ollama run, crush run
DEFAULT_PROVIDERS = [
# TOP PICKS - best value/performance ratio
Provider("opencode-deepseek", "$HOME/.opencode/bin/opencode run --model deepseek/deepseek-chat", "13s 4/4 | BEST VALUE, cheap, fast, accurate"),
Provider("opencode-pickle", "$HOME/.opencode/bin/opencode run --model opencode/big-pickle", "13s 4/4 | BEST FREE, accurate"),
Provider("claude-haiku", "claude -p --model haiku", "14s 4/4 | fast, accurate, best paid option"),
Provider("codex", "codex exec -", "14s 4/4 | reliable, auto-routes"),
# OPENCODE - best free models (binary auto-updates, 6+ free models)
Provider("opencode-deepseek", "opencode run --model deepseek/deepseek-chat",
"DeepSeek V3 - cheap, fast, accurate (paid API key)",
tags=["paid", "code", "reasoning"]),
Provider("opencode-pickle", "opencode run --model opencode/big-pickle",
"Big Pickle - best free general model",
tags=["free", "code", "general"]),
Provider("opencode-reasoner", "opencode run --model deepseek/deepseek-reasoner",
"DeepSeek R1 - complex reasoning, cheap (paid API key)",
tags=["paid", "reasoning"]),
Provider("opencode-free", "opencode run --model opencode/deepseek-v4-flash-free",
"DeepSeek V4 Flash - fast, free",
tags=["free", "fast"]),
# CLAUDE - all accurate, paid, good for code
Provider("claude", "claude -p", "18s 4/4 | auto-routes to best model"),
Provider("claude-opus", "claude -p --model opus", "18s 4/4 | highest quality, expensive"),
Provider("claude-sonnet", "claude -p --model sonnet", "21s 4/4 | balanced quality/speed"),
# ANTHROPIC CLAUDE - paid, high quality (requires subscription or API key)
Provider("claude", "claude -p",
"Claude Code - auto-routes to best model",
tags=["paid", "subscription", "code", "balanced"]),
Provider("claude-haiku", "claude -p --model haiku",
"Claude Haiku - fast, cheapest Claude",
tags=["paid", "subscription", "fast"]),
Provider("claude-sonnet", "claude -p --model sonnet",
"Claude Sonnet - balanced quality/speed",
tags=["paid", "subscription", "code"]),
Provider("claude-opus", "claude -p --model opus",
"Claude Opus - highest quality, expensive",
tags=["paid", "subscription", "reasoning", "quality"]),
# OPENCODE - additional models
Provider("opencode-nano", "$HOME/.opencode/bin/opencode run --model opencode/gpt-5-nano", "24s 4/4 | GPT-5 Nano, reliable"),
Provider("opencode-reasoner", "$HOME/.opencode/bin/opencode run --model deepseek/deepseek-reasoner", "33s 4/4 | complex reasoning, cheap"),
Provider("opencode-grok", "$HOME/.opencode/bin/opencode run --model opencode/grok-code", "11s 2/4 | fastest but unreliable, FREE"),
# OPENAI CODEX - free tier available
Provider("codex", "codex exec -",
"Codex CLI - reliable, auto-routes, free tier available",
tags=["free-tier", "subscription", "code", "general"]),
# GEMINI - slow CLI but good for large docs (1M token context)
Provider("gemini-flash", "gemini --model gemini-2.5-flash", "28s 4/4 | use this for quick tasks"),
Provider("gemini", "gemini --model gemini-2.5-pro", "91s 3/4 | slow CLI, best for large docs/PDFs"),
# GOOGLE ANTIGRAVITY - replaces Gemini CLI (free tier: 1,000 req/day, 60 req/min)
Provider("agy", "agy -p",
"Antigravity (Google) - free tier, Gemini models, large context",
tags=["free-tier", "code", "large-context"]),
# CRUSH - multi-provider agent via Hyper credits or API keys
Provider("crush", "crush run --quiet",
"Crush - multi-model, requires Hyper credits or API keys",
tags=["free-tier", "multi", "code"]),
# LOCAL MODELS
Provider("ollama", "ollama run llama3.2",
"Ollama - local, private, free (GPU recommended)",
tags=["free", "local", "private"]),
# API-TYPE PROVIDERS (pay-per-token, fallback when no CLI covers model)
Provider("openrouter",
"https://openrouter.ai/api/v1",
"OpenRouter - 300+ models, one API key, auto-routing",
type="api",
model="openrouter/auto-beta",
api_key_env="OPENROUTER_API_KEY",
tags=["api", "per-token", "multi", "fallback"]),
Provider("deepseek-api",
"https://api.deepseek.com/v1",
"DeepSeek API - inexpensive per-token access",
type="api",
model="deepseek-chat",
api_key_env="DEEPSEEK_API_KEY",
tags=["api", "per-token", "cheap", "reasoning"]),
# Mock for testing
Provider("mock", "mock", "Mock provider for testing"),
Provider("mock", "mock", "Mock provider for testing",
tags=["testing"]),
]
def get_providers_file() -> Path:
"""Get the providers config file, creating default if needed."""
"""Get the providers config file, creating one on first run."""
if not PROVIDERS_FILE.exists():
PROVIDERS_FILE.parent.mkdir(parents=True, exist_ok=True)
save_providers(DEFAULT_PROVIDERS)
discovered = discover_installed_providers()
selected = []
if discovered:
import sys
print("=" * 60, file=sys.stderr)
print("CmdForge First-Run Provider Setup", file=sys.stderr)
print("=" * 60, file=sys.stderr)
print(file=sys.stderr)
cli_found = [d for d in discovered if d["source"] == "cli"]
api_found = [d for d in discovered if d["source"] == "api-key"]
ollama_found = [d for d in discovered if d["source"] == "ollama-model"]
if cli_found:
print(f"CLIs found on PATH ({len(cli_found)}):", file=sys.stderr)
for d in cli_found:
name = d["name"]
desc = d.get("description", "")
selected.append(Provider(
name=name, command=d["command"], description=desc,
tags=d.get("tags", []),
))
print(f" [+] {name:20s} {desc}", file=sys.stderr)
if api_found:
print(f"\nAPI keys detected ({len(api_found)}):", file=sys.stderr)
for d in api_found:
name = d["name"]
desc = d.get("description", "")
model = d.get("model", "auto")
env_var = d.get("env_var", "")
selected.append(Provider(
name=name, command=d["command"], description=desc,
type="api", model=model, api_key_env=env_var,
tags=d.get("tags", []),
))
print(f" [+] {name:20s} ({model})", file=sys.stderr)
if ollama_found:
print(f"\nLocal Ollama models ({len(ollama_found)}):", file=sys.stderr)
for d in ollama_found[:10]:
name = d["name"]
desc = d.get("description", "")
selected.append(Provider(
name=name, command=d["command"], description=desc,
tags=d.get("tags", []),
))
print(f" [+] {name:30s} {d['command']}", file=sys.stderr)
if len(ollama_found) > 10:
print(f" ... and {len(ollama_found) - 10} more", file=sys.stderr)
if selected:
save_providers(selected)
print(f"\nConfigured {len(selected)} provider(s) from discovery.", file=sys.stderr)
print(f"Run 'cmdforge providers discover' to re-scan at any time.", file=sys.stderr)
else:
save_providers(DEFAULT_PROVIDERS)
print("\nNo AI providers detected. Default providers written.", file=sys.stderr)
print("Run 'cmdforge providers install' for an interactive setup guide.", file=sys.stderr)
return PROVIDERS_FILE
@ -106,16 +410,46 @@ def load_providers() -> List[Provider]:
data = yaml.safe_load(providers_file.read_text())
if not data or "providers" not in data:
return DEFAULT_PROVIDERS.copy()
return [Provider.from_dict(p) for p in data["providers"]]
providers = [Provider.from_dict(p) for p in data["providers"]]
except Exception:
return DEFAULT_PROVIDERS.copy()
config_version = data.get("version", 1)
if not isinstance(config_version, int):
config_version = 1
if config_version < PROVIDERS_CONFIG_VERSION:
providers = _merge_missing_defaults(providers)
try:
save_providers(providers)
except OSError:
# A read-only legacy config should still remain usable.
pass
return providers
def _merge_missing_defaults(providers: List[Provider]) -> List[Provider]:
"""Add defaults introduced since the legacy unversioned config format."""
merged = list(providers)
names = {provider.name for provider in merged}
for default in DEFAULT_PROVIDERS:
if default.name in V2_DEFAULT_PROVIDER_NAMES and default.name not in names:
merged.append(Provider.from_dict(default.to_dict()))
return merged
def save_providers(providers: List[Provider]):
"""Save providers to config file."""
PROVIDERS_FILE.parent.mkdir(parents=True, exist_ok=True)
data = {"providers": [p.to_dict() for p in providers]}
PROVIDERS_FILE.write_text(yaml.dump(data, default_flow_style=False, sort_keys=False))
PROVIDERS_FILE.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
data = {
"version": PROVIDERS_CONFIG_VERSION,
"providers": [p.to_dict() for p in providers],
}
PROVIDERS_FILE.write_text(yaml.safe_dump(data, default_flow_style=False, sort_keys=False))
try:
PROVIDERS_FILE.parent.chmod(0o700)
PROVIDERS_FILE.chmod(0o600)
except OSError:
pass
def get_provider(name: str) -> Optional[Provider]:
@ -159,6 +493,10 @@ def call_provider(provider_name: str, prompt: str, timeout: int = 300, max_token
"""
Call an AI provider with the given prompt.
Dispatches to call_provider_subprocess, call_provider_api, or
call_provider_pty based on the provider's `type` field. Falls back
to the provider's fallback (or fallback_chain) on failure.
Args:
provider_name: Name of the provider to use
prompt: The prompt to send
@ -187,39 +525,83 @@ def call_provider(provider_name: str, prompt: str, timeout: int = 300, max_token
error=f"Provider '{provider_name}' not found. Use 'cmdforge providers' to manage providers."
)
# Parse command (expand environment variables)
if max_tokens is not None:
try:
max_tokens = int(max_tokens)
except (TypeError, ValueError):
return ProviderResult(text="", success=False, error="max_tokens must be an integer")
if max_tokens <= 0 or max_tokens > 1_000_000:
return ProviderResult(
text="", success=False, error="max_tokens must be between 1 and 1000000"
)
# Helper to try fallback provider(s) if available
def try_fallback(error_msg: str) -> ProviderResult:
import sys
last_error = error_msg
# Walk fallback_chain first (ordered, multi-step)
if provider.fallback_chain:
for fb in provider.fallback_chain:
if fb not in _tried:
print(f"[fallback] {provider_name} failed, trying {fb}...", file=sys.stderr)
result = call_provider(fb, prompt, timeout, max_tokens, _tried)
if result.success:
return result
last_error = result.error or last_error
# Fall back to single fallback (backward-compat)
if provider.fallback and provider.fallback not in _tried:
print(f"[fallback] {provider_name} failed, trying {provider.fallback}...", file=sys.stderr)
result = call_provider(provider.fallback, prompt, timeout, max_tokens, _tried)
if result.success:
return result
last_error = result.error or last_error
return ProviderResult(text="", success=False, error=last_error)
# Dispatch by provider type
ptype = getattr(provider, "type", None) or "subprocess"
try:
if ptype == "subprocess":
result = call_provider_subprocess(provider, prompt, timeout, max_tokens)
elif ptype == "api":
result = call_provider_api(provider, prompt, timeout, max_tokens)
elif ptype == "pty":
result = call_provider_pty(provider, prompt, timeout, max_tokens)
else:
return try_fallback(f"Unknown provider type: {ptype}")
if result.success:
return result
return try_fallback(result.error or "Provider call failed")
except Exception as e:
return try_fallback(f"Provider error: {str(e)}")
def call_provider_subprocess(provider: "Provider", prompt: str, timeout: int, max_tokens: Optional[int]) -> ProviderResult:
"""Invoke a CLI provider via subprocess.run with stdin piping."""
cmd = os.path.expandvars(provider.command)
# Append max_tokens flag for known providers
if max_tokens:
name_lower = provider_name.lower()
# Append max_tokens flags only for CLIs whose interfaces support them.
if max_tokens is not None:
name_lower = provider.name.lower()
if name_lower.startswith("claude") or "claude" in cmd.lower():
cmd = f"{cmd} --max-tokens {max_tokens}"
elif name_lower.startswith("gemini") or "gemini" in cmd.lower():
cmd = f"{cmd} --max-output-tokens {max_tokens}"
# opencode and codex use default model limits, no flag available
# opencode, agy, codex, and crush use their configured/default limits.
# Check if base command exists (use shlex for proper quote handling)
try:
cmd_parts = shlex.split(cmd)
base_cmd = cmd_parts[0] if cmd_parts else cmd.split()[0]
except ValueError:
# shlex failed (unbalanced quotes, etc.) - fall back to simple split
base_cmd = cmd.split()[0]
# Helper to try fallback provider if available
def try_fallback(error_msg: str) -> ProviderResult:
if provider.fallback and provider.fallback not in _tried:
import sys
print(f"[fallback] {provider_name} failed, trying {provider.fallback}...", file=sys.stderr)
return call_provider(provider.fallback, prompt, timeout, max_tokens, _tried)
return ProviderResult(text="", success=False, error=error_msg)
# Expand ~ for the which check
base_cmd_expanded = os.path.expanduser(base_cmd)
if not shutil.which(base_cmd_expanded) and not os.path.isfile(base_cmd_expanded):
return try_fallback(
f"Command '{base_cmd}' not found. Is it installed and in PATH?\n\nTo install AI providers, run: cmdforge providers install"
return ProviderResult(
text="",
success=False,
error=f"Command '{base_cmd}' not found. Is it installed and in PATH?\n\nTo install AI providers, run: cmdforge providers install"
)
try:
@ -237,7 +619,7 @@ def call_provider(provider_name: str, prompt: str, timeout: int = 300, max_token
error_msg = f"Provider exited with code {result.returncode}: {stderr_clean}"
if "not found" in stderr_clean.lower() or "not installed" in stderr_clean.lower():
error_msg += "\n\nTo install AI providers, run: cmdforge providers install"
return try_fallback(error_msg)
return ProviderResult(text="", success=False, error=error_msg)
# Warn if output is empty (provider ran but returned nothing)
clean_stdout = strip_ansi(result.stdout)
@ -246,31 +628,172 @@ def call_provider(provider_name: str, prompt: str, timeout: int = 300, max_token
# Check for OpenCode's ProviderModelNotFoundError
if "ProviderModelNotFoundError" in stderr or "ModelNotFoundError" in stderr:
# Extract provider and model info if possible
provider_match = re.search(r'providerID:\s*"([^"]+)"', stderr)
model_match = re.search(r'modelID:\s*"([^"]+)"', stderr)
provider_id = provider_match.group(1) if provider_match else "unknown"
model_id = model_match.group(1) if model_match else "unknown"
return try_fallback(
f"Model '{model_id}' from provider '{provider_id}' is not available.\n\n"
f"To fix this, either:\n"
f" 1. Run 'opencode' to connect the {provider_id} provider\n"
f" 2. Use --provider to pick a different model (e.g., --provider opencode-pickle)\n"
f" 3. Run 'cmdforge ui' to edit the tool's default provider"
return ProviderResult(
text="",
success=False,
error=(
f"Model '{model_id}' from provider '{provider_id}' is not available.\n\n"
f"To fix this, either:\n"
f" 1. Run 'opencode' to connect the {provider_id} provider\n"
f" 2. Use --provider to pick a different model (e.g., --provider opencode-pickle)\n"
f" 3. Run 'cmdforge ui' to edit the tool's default provider"
)
)
stderr_hint = f" (stderr: {stderr[:200]}...)" if len(stderr) > 200 else (f" (stderr: {stderr})" if stderr else "")
return try_fallback(
f"Provider returned empty output{stderr_hint}.\n\nThis may mean the model is not available. Try a different provider or run: cmdforge providers install"
return ProviderResult(
text="",
success=False,
error=f"Provider returned empty output{stderr_hint}.\n\nThis may mean the model is not available. Try a different provider or run: cmdforge providers install"
)
return ProviderResult(text=clean_stdout, success=True)
except subprocess.TimeoutExpired:
return try_fallback(f"Provider timed out after {timeout} seconds")
return ProviderResult(text="", success=False, error=f"Provider timed out after {timeout} seconds")
except Exception as e:
return try_fallback(f"Provider error: {str(e)}")
return ProviderResult(text="", success=False, error=f"Provider error: {str(e)}")
def call_provider_api(provider: "Provider", prompt: str, timeout: int, max_tokens: Optional[int]) -> ProviderResult:
"""Call an OpenAI-compatible HTTP API endpoint.
The provider.command is the endpoint URL (e.g. https://api.openrouter.ai/api/v1/chat/completions).
The provider.model is the model ID (e.g. "deepseek/deepseek-chat").
The provider.api_key_env names the environment variable holding the API key.
"""
if not provider.model:
return ProviderResult(
text="",
success=False,
error=f"API provider '{provider.name}' has no model configured."
)
env_var = provider.api_key_env or _infer_api_key_env(provider.name)
api_key = os.environ.get(env_var) if env_var else None
if not api_key:
return ProviderResult(
text="",
success=False,
error=f"API key not set. Set the {env_var} environment variable to use '{provider.name}'."
)
try:
import requests
except ImportError:
return ProviderResult(
text="",
success=False,
error="The 'requests' package is required for API providers. Install with: pip install requests"
)
endpoint = os.path.expandvars(provider.command)
if not endpoint.endswith("/chat/completions"):
endpoint = endpoint.rstrip("/") + "/chat/completions"
body = {
"model": provider.model,
"messages": [{"role": "user", "content": prompt}],
}
if max_tokens is not None:
body["max_tokens"] = max_tokens
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
}
try:
response = requests.post(endpoint, json=body, headers=headers, timeout=timeout)
if response.status_code != 200:
return ProviderResult(
text="",
success=False,
error=f"API returned HTTP {response.status_code}: {response.text[:500]}"
)
data = response.json()
text = data.get("choices", [{}])[0].get("message", {}).get("content", "")
if not text:
return ProviderResult(text="", success=False, error="API returned empty response")
return ProviderResult(text=text, success=True)
except requests.Timeout:
return ProviderResult(text="", success=False, error=f"API timed out after {timeout} seconds")
except Exception as e:
return ProviderResult(text="", success=False, error=f"API error: {str(e)}")
def _infer_api_key_env(provider_name: str) -> str:
"""Guess the API key env var name from provider name (fallback if api_key_env not set)."""
name_upper = provider_name.upper().replace("-", "_")
return f"{name_upper}_API_KEY"
def call_provider_pty(provider: "Provider", prompt: str, timeout: int, max_tokens: Optional[int]) -> ProviderResult:
"""Wrap an interactive CLI via a pseudo-terminal (experimental).
Requires pty_config on the provider with:
- prompt_pattern: regex/expect pattern for the CLI's ready prompt
- response_pattern: regex/expect pattern marking end of response
- exit_command: command to cleanly exit the CLI (e.g. "/exit", "\\q")
"""
if not provider.pty_config:
return ProviderResult(
text="",
success=False,
error=f"PTY provider '{provider.name}' has no pty_config set. Cannot wrap interactive CLI."
)
try:
import pexpect
except ImportError:
return ProviderResult(
text="",
success=False,
error="The 'pexpect' package is required for PTY providers. Install with: pip install pexpect"
)
cfg = provider.pty_config
prompt_pattern = cfg.get("prompt_pattern")
response_pattern = cfg.get("response_pattern")
exit_command = cfg.get("exit_command", "/exit")
if not prompt_pattern or not response_pattern:
return ProviderResult(
text="",
success=False,
error="pty_config must include prompt_pattern and response_pattern"
)
cmd = os.path.expandvars(provider.command)
child = None
try:
child = pexpect.spawn(cmd, timeout=timeout, encoding='utf-8')
# Wait for the CLI's ready prompt
child.expect(prompt_pattern)
# Send the user's prompt
child.sendline(prompt)
# Wait for the response to complete
child.expect(response_pattern)
text = child.before
# Cleanly exit
child.sendline(exit_command)
return ProviderResult(text=strip_ansi(text), success=True)
except pexpect.TIMEOUT:
return ProviderResult(text="", success=False, error=f"PTY provider timed out after {timeout} seconds")
except pexpect.EOF:
return ProviderResult(text="", success=False, error="PTY provider exited unexpectedly")
except Exception as e:
return ProviderResult(text="", success=False, error=f"PTY provider error: {str(e)}")
finally:
if child is not None and child.isalive():
child.close(force=True)
def mock_provider(prompt: str) -> ProviderResult:

101
src/cmdforge/runner.py
View File

@ -161,32 +161,6 @@ No text before or after the JSON. No markdown fences.{' ' + field_guidance if fi
{prompt}"""
# Legacy function kept for rollback if needed
def _append_schema_instructions_legacy(prompt: str, schema: dict) -> str:
"""
DEPRECATED: Append schema instructions to a prompt.
This approach causes small models to echo the schema. Use
prepend_schema_instructions() instead.
"""
schema_json = json.dumps(schema, indent=2)
return f"""{prompt}
---
RESPONSE FORMAT:
You must respond with ONLY valid JSON matching this exact schema:
{schema_json}
CRITICAL RULES:
- Output ONLY the JSON object, no other text before or after
- Put any thinking, reasoning, or analysis in the "reasoning" field
- Put your actual answer/response in the "output" field
- Do not include markdown code fences
- Do not include any preamble like "Here is my response:"
"""
def check_system_dependencies(tool: Tool) -> list:
"""
Check if system dependencies are satisfied.
@ -506,11 +480,30 @@ def _extract_json(text: str) -> dict | list | None:
return None
def _read_step_file(filename: str, base_dir: Optional[Path], step_type: str) -> str:
"""Read a prompt/code file relative to the tool directory."""
if not base_dir:
raise ValueError(f"{step_type}_file requires the tool to have a filesystem path")
relative_path = Path(filename)
if relative_path.is_absolute():
raise ValueError(f"{step_type}_file must be relative to the tool directory")
base = base_dir.resolve()
path = (base / relative_path).resolve()
if path == base or base not in path.parents:
raise ValueError(f"{step_type}_file cannot reference files outside the tool directory")
if not path.is_file():
raise FileNotFoundError(f"{step_type}_file not found: {filename}")
return path.read_text()
def execute_prompt_step(
step: PromptStep,
variables: dict,
provider_override: str = None,
verbose: bool = False
verbose: bool = False,
base_dir: Optional[Path] = None
) -> tuple[str, bool]:
"""
Execute a prompt step.
@ -529,7 +522,13 @@ def execute_prompt_step(
import re
# Build prompt with variable substitution
prompt = substitute_variables(step.prompt, variables, warn_non_scalar=verbose)
try:
prompt_template = _read_step_file(step.prompt_file, base_dir, "prompt") if step.prompt_file else step.prompt
except (OSError, ValueError) as e:
print(f"Error in prompt step: {e}", file=sys.stderr)
return "", False
prompt = substitute_variables(prompt_template, variables, warn_non_scalar=verbose)
# Inject profile system prompt if specified
if step.profile:
@ -633,7 +632,8 @@ def execute_code_step(
step: CodeStep,
variables: dict,
step_num: int = 0,
verbose: bool = False
verbose: bool = False,
base_dir: Optional[Path] = None
) -> tuple[dict, bool]:
"""
Execute a code step.
@ -647,7 +647,14 @@ def execute_code_step(
Tuple of (output_vars_dict, success)
"""
# Substitute variables in code (like {outputfile} -> actual value)
code = substitute_variables(step.code, variables, warn_non_scalar=verbose)
try:
code_template = _read_step_file(step.code_file, base_dir, "code") if step.code_file else step.code
except (OSError, ValueError) as e:
print(f"Error in code step (step {step_num}):", file=sys.stderr)
print(f" {e}", file=sys.stderr)
return {}, False
code = substitute_variables(code_template, variables, warn_non_scalar=verbose)
# Create execution environment with variables
# IMPORTANT: Use the same dict for both globals and locals.
@ -887,6 +894,8 @@ def run_tool(
print(f"[verbose] Variables: {list(variables.keys())}", file=sys.stderr)
print(f"[verbose] Steps: {len(tool.steps)}", file=sys.stderr)
tool_base_dir = tool.path.parent if tool.path else None
# If no steps, just substitute output template
if not tool.steps:
output = substitute_variables(tool.output, variables, warn_non_scalar=verbose)
@ -916,7 +925,13 @@ def run_tool(
if dry_run:
variables[step.output_var] = f"[DRY RUN - would call {step.provider}]"
else:
output, success = execute_prompt_step(step, variables, provider_override, verbose=verbose)
output, success = execute_prompt_step(
step,
variables,
provider_override,
verbose=verbose,
base_dir=tool_base_dir
)
if not success:
return "", 2
variables[step.output_var] = output
@ -932,7 +947,13 @@ def run_tool(
for var in [v.strip() for v in step.output_var.split(',')]:
variables[var] = "[DRY RUN - would execute code]"
else:
outputs, success = execute_code_step(step, variables, step_num=i+1, verbose=verbose)
outputs, success = execute_code_step(
step,
variables,
step_num=i+1,
verbose=verbose,
base_dir=tool_base_dir
)
if not success:
return "", 1
# Merge all output vars into variables
@ -1011,6 +1032,16 @@ def create_argument_parser(tool: Tool) -> argparse.ArgumentParser:
return parser
def collect_custom_args(tool: Tool, args: argparse.Namespace) -> dict:
"""Collect tool-specific arguments from a parsed argparse namespace."""
custom_args = {}
for arg in tool.arguments:
value = getattr(args, arg.variable, None)
if value is not None:
custom_args[arg.variable] = value
return custom_args
def main():
"""Entry point for tool execution via wrapper script."""
if len(sys.argv) < 2:
@ -1060,11 +1091,7 @@ def main():
input_text = ""
# Collect custom args
custom_args = {}
for arg in tool.arguments:
value = getattr(args, arg.variable, None)
if value is not None:
custom_args[arg.variable] = value
custom_args = collect_custom_args(tool, args)
# Determine provider override (CLI flag takes precedence over manifest)
effective_provider = args.provider or provider_override_from_manifest

250
src/cmdforge/semver.py Normal file
View File

@ -0,0 +1,250 @@
"""Semantic version constraint matching.
Provides version parsing and constraint matching for dependency resolution.
Follows semver 2.0.0 specification with support for common constraint operators.
"""
import re
from dataclasses import dataclass
from typing import Optional, Tuple, List
VERSION_PATTERN = re.compile(
r'^(?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)'
r'(?:-(?P<prerelease>[0-9A-Za-z.-]+))?'
r'(?:\+(?P<build>[0-9A-Za-z.-]+))?$'
)
@dataclass
class Version:
"""Semantic version representation."""
major: int
minor: int
patch: int
prerelease: Optional[str] = None
build: Optional[str] = None
@classmethod
def parse(cls, version_str: str) -> Optional["Version"]:
"""Parse a version string into a Version object.
Args:
version_str: Version string like "1.2.3" or "1.0.0-alpha.1+build.123"
Returns:
Version object, or None if parsing fails
"""
if not version_str:
return None
match = VERSION_PATTERN.match(version_str.strip())
if not match:
return None
return cls(
major=int(match.group("major")),
minor=int(match.group("minor")),
patch=int(match.group("patch")),
prerelease=match.group("prerelease"),
build=match.group("build")
)
def __str__(self) -> str:
s = f"{self.major}.{self.minor}.{self.patch}"
if self.prerelease:
s += f"-{self.prerelease}"
if self.build:
s += f"+{self.build}"
return s
@property
def tuple(self) -> Tuple[int, int, int]:
"""Get (major, minor, patch) tuple for comparison."""
return (self.major, self.minor, self.patch)
def __eq__(self, other: object) -> bool:
if not isinstance(other, Version):
return False
return self.tuple == other.tuple and self.prerelease == other.prerelease
def __lt__(self, other: "Version") -> bool:
if self.tuple != other.tuple:
return self.tuple < other.tuple
# Prerelease versions are less than release versions
# No prerelease = release version
if self.prerelease is None and other.prerelease is not None:
return False # 1.0.0 > 1.0.0-alpha
if self.prerelease is not None and other.prerelease is None:
return True # 1.0.0-alpha < 1.0.0
# Both have prerelease, compare lexicographically
return (self.prerelease or "") < (other.prerelease or "")
def __le__(self, other: "Version") -> bool:
return self == other or self < other
def __gt__(self, other: "Version") -> bool:
return not self <= other
def __ge__(self, other: "Version") -> bool:
return not self < other
def __hash__(self) -> int:
return hash((self.tuple, self.prerelease))
def matches_constraint(version: str, constraint: str) -> bool:
"""
Check if a version satisfies a constraint.
Supported constraints:
- "*" or "latest" or "": any version
- "1.0.0": exact match
- "^1.2.3": compatible (>=1.2.3 and <2.0.0)
- "^0.2.3": for 0.x, means >=0.2.3 and <0.3.0 (semver special case)
- "^0.0.3": for 0.0.x, means exactly 0.0.3 (semver special case)
- "~1.2.3": approximately (>=1.2.3 and <1.3.0)
- ">=1.0.0", "<=1.0.0", ">1.0.0", "<1.0.0": comparisons
- "=1.0.0": explicit exact match
Args:
version: Version string to check (e.g., "1.2.3")
constraint: Constraint string (e.g., "^1.0.0")
Returns:
True if version satisfies constraint
"""
if not constraint or constraint in ("*", "latest"):
return True
v = Version.parse(version)
if not v:
return False
constraint = constraint.strip()
# Standard semver behavior: prerelease versions do NOT satisfy ranges
# unless the constraint itself includes a prerelease.
if v.prerelease and "-" not in constraint:
return False
# Caret: ^1.2.3 means >=1.2.3 and <2.0.0
# Special cases:
# ^0.2.3 -> >=0.2.3 and <0.3.0
# ^0.0.3 -> >=0.0.3 and <0.0.4 (exact match for 0.0.x series)
if constraint.startswith("^"):
c = Version.parse(constraint[1:])
if not c:
return False
# Must be >= constraint version
if v < c:
return False
# Upper bound depends on major version
if c.major == 0:
if c.minor == 0:
# ^0.0.y allows ONLY 0.0.y (exact match for 0.0.x series)
return v.major == 0 and v.minor == 0 and v.patch == c.patch
# ^0.x.y allows >=0.x.y and <0.(x+1).0
return v.major == 0 and v.minor == c.minor
else:
# ^x.y.z allows changes that don't modify major: <(x+1).0.0
return v.major == c.major
# Tilde: ~1.2.3 means >=1.2.3 and <1.3.0
if constraint.startswith("~"):
c = Version.parse(constraint[1:])
if not c:
return False
# Must be >= constraint version
if v < c:
return False
# Must be same major.minor
return v.major == c.major and v.minor == c.minor
# Comparison operators (check longest first)
for op in (">=", "<=", ">", "<", "="):
if constraint.startswith(op):
c = Version.parse(constraint[len(op):].strip())
if not c:
return False
if op == ">=":
return v >= c
elif op == "<=":
return v <= c
elif op == ">":
return v > c
elif op == "<":
return v < c
elif op == "=":
return v == c
# Exact match (no operator)
c = Version.parse(constraint)
if c:
return v == c
return False
def find_best_match(versions: List[str], constraint: str) -> Optional[str]:
"""
Find the best (highest) version that matches a constraint.
Args:
versions: List of available versions
constraint: Version constraint
Returns:
Best matching version, or None if no match
"""
matching = [v for v in versions if matches_constraint(v, constraint)]
if not matching:
return None
# Sort by parsed version, return highest
parsed = [(Version.parse(v), v) for v in matching]
parsed = [(p, v) for p, v in parsed if p is not None]
if not parsed:
return None
parsed.sort(key=lambda x: x[0], reverse=True)
return parsed[0][1]
def is_valid_version(version: str) -> bool:
"""Check if a string is a valid semver version.
Args:
version: Version string to validate
Returns:
True if valid semver format
"""
return Version.parse(version) is not None
def compare_versions(v1: str, v2: str) -> int:
"""Compare two version strings.
Args:
v1: First version
v2: Second version
Returns:
-1 if v1 < v2, 0 if equal, 1 if v1 > v2
Raises:
ValueError: If either version is invalid
"""
parsed_v1 = Version.parse(v1)
parsed_v2 = Version.parse(v2)
if not parsed_v1 or not parsed_v2:
raise ValueError(f"Invalid version: {v1 if not parsed_v1 else v2}")
if parsed_v1 < parsed_v2:
return -1
elif parsed_v1 > parsed_v2:
return 1
return 0

19
src/cmdforge/tool.py
View File

@ -134,6 +134,15 @@ class PromptStep:
@classmethod
def from_dict(cls, data: dict) -> "PromptStep":
max_tokens = data.get("max_tokens")
if max_tokens is not None:
try:
max_tokens = int(max_tokens)
except (TypeError, ValueError):
raise ValueError("max_tokens must be an integer")
if max_tokens <= 0 or max_tokens > 1_000_000:
raise ValueError("max_tokens must be between 1 and 1000000")
return cls(
prompt=data["prompt"],
provider=data["provider"],
@ -145,7 +154,7 @@ class PromptStep:
output_schema=data.get("output_schema"),
max_retries=data.get("max_retries", 1),
plain_text=data.get("plain_text", False),
max_tokens=data.get("max_tokens")
max_tokens=max_tokens
)
@ -292,6 +301,7 @@ class Tool:
system_dependencies: List[SystemDependency] = field(default_factory=list) # System packages (apt, brew, etc.)
source: Optional[ToolSource] = None # Attribution for imported/external tools
version: str = "" # Tool version
visibility: str = "public" # "public", "private", or "unlisted"
path: Optional[Path] = None # Path to config.yaml (set by load_tool)
@classmethod
@ -341,6 +351,7 @@ class Tool:
system_dependencies=system_dependencies,
source=source,
version=data.get("version", ""),
visibility=data.get("visibility", "public"),
)
def to_dict(self) -> dict:
@ -353,6 +364,9 @@ class Tool:
# Only include category if it's not the default
if self.category and self.category != "Other":
d["category"] = self.category
# Only include visibility if it's not the default
if self.visibility and self.visibility != "public":
d["visibility"] = self.visibility
# Include source attribution if present
if self.source:
d["source"] = self.source.to_dict()
@ -493,7 +507,8 @@ def load_tool(name: str) -> Optional[Tool]:
"description": data.get("description", ""),
"arguments": arguments,
"steps": steps,
"output": "{response}" if steps else "{input}"
"output": "{response}" if steps else "{input}",
"visibility": data.get("visibility", "public"),
}
tool = Tool.from_dict(data)

255
src/cmdforge/version.py
View File

@ -1,250 +1,9 @@
"""Semantic version constraint matching.
"""Re-export from semver module for backward compatibility."""
Provides version parsing and constraint matching for dependency resolution.
Follows semver 2.0.0 specification with support for common constraint operators.
"""
import re
from dataclasses import dataclass
from typing import Optional, Tuple, List
VERSION_PATTERN = re.compile(
r'^(?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)'
r'(?:-(?P<prerelease>[0-9A-Za-z.-]+))?'
r'(?:\+(?P<build>[0-9A-Za-z.-]+))?$'
from .semver import ( # noqa: F401
Version,
matches_constraint,
find_best_match,
is_valid_version,
compare_versions,
)
@dataclass
class Version:
"""Semantic version representation."""
major: int
minor: int
patch: int
prerelease: Optional[str] = None
build: Optional[str] = None
@classmethod
def parse(cls, version_str: str) -> Optional["Version"]:
"""Parse a version string into a Version object.
Args:
version_str: Version string like "1.2.3" or "1.0.0-alpha.1+build.123"
Returns:
Version object, or None if parsing fails
"""
if not version_str:
return None
match = VERSION_PATTERN.match(version_str.strip())
if not match:
return None
return cls(
major=int(match.group("major")),
minor=int(match.group("minor")),
patch=int(match.group("patch")),
prerelease=match.group("prerelease"),
build=match.group("build")
)
def __str__(self) -> str:
s = f"{self.major}.{self.minor}.{self.patch}"
if self.prerelease:
s += f"-{self.prerelease}"
if self.build:
s += f"+{self.build}"
return s
@property
def tuple(self) -> Tuple[int, int, int]:
"""Get (major, minor, patch) tuple for comparison."""
return (self.major, self.minor, self.patch)
def __eq__(self, other: object) -> bool:
if not isinstance(other, Version):
return False
return self.tuple == other.tuple and self.prerelease == other.prerelease
def __lt__(self, other: "Version") -> bool:
if self.tuple != other.tuple:
return self.tuple < other.tuple
# Prerelease versions are less than release versions
# No prerelease = release version
if self.prerelease is None and other.prerelease is not None:
return False # 1.0.0 > 1.0.0-alpha
if self.prerelease is not None and other.prerelease is None:
return True # 1.0.0-alpha < 1.0.0
# Both have prerelease, compare lexicographically
return (self.prerelease or "") < (other.prerelease or "")
def __le__(self, other: "Version") -> bool:
return self == other or self < other
def __gt__(self, other: "Version") -> bool:
return not self <= other
def __ge__(self, other: "Version") -> bool:
return not self < other
def __hash__(self) -> int:
return hash((self.tuple, self.prerelease))
def matches_constraint(version: str, constraint: str) -> bool:
"""
Check if a version satisfies a constraint.
Supported constraints:
- "*" or "latest" or "": any version
- "1.0.0": exact match
- "^1.2.3": compatible (>=1.2.3 and <2.0.0)
- "^0.2.3": for 0.x, means >=0.2.3 and <0.3.0 (semver special case)
- "^0.0.3": for 0.0.x, means exactly 0.0.3 (semver special case)
- "~1.2.3": approximately (>=1.2.3 and <1.3.0)
- ">=1.0.0", "<=1.0.0", ">1.0.0", "<1.0.0": comparisons
- "=1.0.0": explicit exact match
Args:
version: Version string to check (e.g., "1.2.3")
constraint: Constraint string (e.g., "^1.0.0")
Returns:
True if version satisfies constraint
"""
if not constraint or constraint in ("*", "latest"):
return True
v = Version.parse(version)
if not v:
return False
constraint = constraint.strip()
# Standard semver behavior: prerelease versions do NOT satisfy ranges
# unless the constraint itself includes a prerelease.
if v.prerelease and "-" not in constraint:
return False
# Caret: ^1.2.3 means >=1.2.3 and <2.0.0
# Special cases:
# ^0.2.3 -> >=0.2.3 and <0.3.0
# ^0.0.3 -> >=0.0.3 and <0.0.4 (exact match for 0.0.x series)
if constraint.startswith("^"):
c = Version.parse(constraint[1:])
if not c:
return False
# Must be >= constraint version
if v < c:
return False
# Upper bound depends on major version
if c.major == 0:
if c.minor == 0:
# ^0.0.y allows ONLY 0.0.y (exact match for 0.0.x series)
return v.major == 0 and v.minor == 0 and v.patch == c.patch
# ^0.x.y allows >=0.x.y and <0.(x+1).0
return v.major == 0 and v.minor == c.minor
else:
# ^x.y.z allows changes that don't modify major: <(x+1).0.0
return v.major == c.major
# Tilde: ~1.2.3 means >=1.2.3 and <1.3.0
if constraint.startswith("~"):
c = Version.parse(constraint[1:])
if not c:
return False
# Must be >= constraint version
if v < c:
return False
# Must be same major.minor
return v.major == c.major and v.minor == c.minor
# Comparison operators (check longest first)
for op in (">=", "<=", ">", "<", "="):
if constraint.startswith(op):
c = Version.parse(constraint[len(op):].strip())
if not c:
return False
if op == ">=":
return v >= c
elif op == "<=":
return v <= c
elif op == ">":
return v > c
elif op == "<":
return v < c
elif op == "=":
return v == c
# Exact match (no operator)
c = Version.parse(constraint)
if c:
return v == c
return False
def find_best_match(versions: List[str], constraint: str) -> Optional[str]:
"""
Find the best (highest) version that matches a constraint.
Args:
versions: List of available versions
constraint: Version constraint
Returns:
Best matching version, or None if no match
"""
matching = [v for v in versions if matches_constraint(v, constraint)]
if not matching:
return None
# Sort by parsed version, return highest
parsed = [(Version.parse(v), v) for v in matching]
parsed = [(p, v) for p, v in parsed if p is not None]
if not parsed:
return None
parsed.sort(key=lambda x: x[0], reverse=True)
return parsed[0][1]
def is_valid_version(version: str) -> bool:
"""Check if a string is a valid semver version.
Args:
version: Version string to validate
Returns:
True if valid semver format
"""
return Version.parse(version) is not None
def compare_versions(v1: str, v2: str) -> int:
"""Compare two version strings.
Args:
v1: First version
v2: Second version
Returns:
-1 if v1 < v2, 0 if equal, 1 if v1 > v2
Raises:
ValueError: If either version is invalid
"""
parsed_v1 = Version.parse(v1)
parsed_v2 = Version.parse(v2)
if not parsed_v1 or not parsed_v2:
raise ValueError(f"Invalid version: {v1 if not parsed_v1 else v2}")
if parsed_v1 < parsed_v2:
return -1
elif parsed_v1 > parsed_v2:
return 1
return 0

87
tests/test_cli.py
View File

@ -199,6 +199,28 @@ class TestRunCommand:
captured = capsys.readouterr()
assert 'MOCK' in captured.out
def test_run_with_tool_specific_args_after_separator(self, temp_tools_dir, capsys):
"""cmdforge run should parse tool args with the wrapper parser."""
from cmdforge.tool import save_tool
tool = Tool(
name="greet",
arguments=[
ToolArgument(flag="--name", variable="name", default="World")
],
output="Hello, {name}!"
)
save_tool(tool)
with patch('sys.argv', ['cmdforge', 'run', 'greet', '--', '--name', 'Alice']):
with patch('sys.stdin', StringIO("")):
with patch('sys.stdin.isatty', return_value=True):
result = main()
assert result == 0
captured = capsys.readouterr()
assert 'Hello, Alice!' in captured.out
def test_run_nonexistent_tool(self, temp_tools_dir, capsys):
"""Running nonexistent tool should fail."""
with patch('sys.argv', ['cmdforge', 'run', 'nonexistent']):
@ -272,6 +294,71 @@ class TestProvidersCommand:
assert provider is not None
assert provider.command == 'my-ai --prompt'
def test_providers_add_api_configuration(self, temp_providers_file, capsys):
with patch('sys.argv', [
'cmdforge', 'providers', 'add', 'custom-api',
'https://example.test/v1',
'--type', 'api',
'--model', 'example/model',
'--api-key-env', 'CUSTOM_API_KEY',
'--tag', 'api',
'--fallback-chain', 'free',
]):
result = main()
assert result == 0
from cmdforge.providers import PRESET_CHAINS, get_provider
provider = get_provider('custom-api')
assert provider.type == 'api'
assert provider.model == 'example/model'
assert provider.api_key_env == 'CUSTOM_API_KEY'
assert provider.tags == ['api']
assert provider.fallback_chain == PRESET_CHAINS['free']
def test_providers_list_reports_missing_api_key(self, temp_providers_file, capsys):
from cmdforge.providers import Provider, save_providers
save_providers([
Provider(
'custom-api',
'https://example.test/v1',
type='api',
model='example/model',
api_key_env='MISSING_CUSTOM_API_KEY',
)
])
with patch.dict('os.environ', {'MISSING_CUSTOM_API_KEY': ''}):
with patch('sys.argv', ['cmdforge', 'providers', 'list']):
result = main()
assert result == 0
captured = capsys.readouterr()
assert 'API KEY NOT SET (MISSING_CUSTOM_API_KEY)' in captured.out
assert 'NOT FOUND (https://' not in captured.out
def test_providers_discover_adds_new_provider(self, temp_providers_file, capsys):
discovered = [{
'source': 'cli',
'binary': 'example-ai',
'path': '/usr/bin/example-ai',
'name': 'example-ai',
'command': 'example-ai --print',
'description': 'Example provider',
'tags': ['test'],
}]
with patch('cmdforge.providers.discover_installed_providers', return_value=discovered):
with patch('sys.argv', ['cmdforge', 'providers', 'discover', '--add']):
result = main()
assert result == 0
from cmdforge.providers import get_provider
provider = get_provider('example-ai')
assert provider is not None
assert provider.command == 'example-ai --print'
assert provider.tags == ['test']
def test_providers_remove(self, temp_providers_file, capsys):
"""Remove a provider."""
from cmdforge.providers import add_provider, Provider

19
tests/test_collection.py
View File

@ -269,6 +269,25 @@ class TestResolveToolReferences:
assert ("official/unapproved", "no approved public version") in result.registry_tool_issues
def test_get_visibility_uses_owner_path(self, tmp_path):
from cmdforge.collection import _get_tool_visibility_from_yaml
from cmdforge.tool import get_tools_dir
with patch('cmdforge.tool.TOOLS_DIR', tmp_path / ".cmdforge"):
tools_dir = get_tools_dir()
owned = tools_dir / "myuser" / "private-tool"
owned.mkdir(parents=True)
(owned / "config.yaml").write_text(
yaml.safe_dump({
"name": "private-tool",
"visibility": "private",
})
)
result = _get_tool_visibility_from_yaml("private-tool", my_owner="myuser")
assert result == "private"
class TestToolResolutionResult:
"""Tests for ToolResolutionResult dataclass."""

20
tests/test_packaging.py Normal file
View File

@ -0,0 +1,20 @@
"""Tests for packaging metadata."""
import sys
import pytest
if sys.version_info >= (3, 11):
import tomllib
else:
tomllib = pytest.importorskip("tomli")
def test_web_templates_and_static_are_declared_as_package_data():
with open("pyproject.toml", "rb") as f:
data = tomllib.load(f)
package_data = data["tool"]["setuptools"]["package-data"]["cmdforge.web"]
assert "templates/**/*.html" in package_data
assert "static/**/*" in package_data

160
tests/test_providers.py
View File

@ -11,7 +11,7 @@ from cmdforge.providers import (
Provider, ProviderResult,
load_providers, save_providers, get_provider,
add_provider, delete_provider,
call_provider, mock_provider,
call_provider, discover_installed_providers, mock_provider,
DEFAULT_PROVIDERS, strip_ansi
)
@ -71,6 +71,29 @@ class TestProvider:
assert restored.command == original.command
assert restored.description == original.description
def test_legacy_positional_fallback_is_preserved(self):
provider = Provider("primary", "primary-cmd", "Primary", "backup")
assert provider.fallback == "backup"
assert provider.type == "subprocess"
def test_extended_fields_roundtrip(self):
original = Provider(
name="api-test",
command="https://example.test/v1",
description="Test API",
fallback="mock",
type="api",
model="test/model",
tags=["api", "test"],
install={"cost": "test"},
fallback_chain=["backup", "mock"],
api_key_env="TEST_API_KEY",
pty_config={"prompt_pattern": ">"},
)
assert Provider.from_dict(original.to_dict()) == original
class TestProviderResult:
"""Tests for ProviderResult dataclass."""
@ -144,6 +167,34 @@ class TestProviderPersistence:
assert loaded[0].name == "test1"
assert loaded[1].name == "test2"
def test_save_providers_uses_private_file_permissions(self, temp_providers_file):
save_providers([Provider("test", "cmd")])
assert oct(temp_providers_file.stat().st_mode & 0o777) == "0o600"
def test_legacy_config_gains_missing_defaults_without_losing_custom_provider(self, temp_providers_file):
temp_providers_file.parent.mkdir(parents=True)
temp_providers_file.write_text(yaml.safe_dump({
"providers": [
{
"name": "custom",
"command": "custom-command",
"description": "Keep me",
"fallback": "mock",
}
]
}))
loaded = load_providers()
custom = next(provider for provider in loaded if provider.name == "custom")
assert custom.command == "custom-command"
assert custom.fallback == "mock"
assert any(provider.name == "opencode-free" for provider in loaded)
assert any(provider.name == "openrouter" for provider in loaded)
saved = yaml.safe_load(temp_providers_file.read_text())
assert saved["version"] == 2
def test_get_provider_exists(self, temp_providers_file):
providers = [
Provider("target", "target-cmd", "Target provider")
@ -246,6 +297,16 @@ class TestCallProvider:
assert result.success is False
assert "not found" in result.error.lower()
@patch('subprocess.run')
def test_call_provider_rejects_non_integer_max_tokens(self, mock_run, temp_providers_file):
save_providers([Provider("claude", "claude -p")])
result = call_provider("claude", "Test", max_tokens="1; touch /tmp/pwned")
assert result.success is False
assert "max_tokens" in result.error
mock_run.assert_not_called()
@patch('subprocess.run')
@patch('shutil.which')
def test_call_real_provider_success(self, mock_which, mock_run, temp_providers_file):
@ -456,6 +517,18 @@ class TestProviderFallback:
assert result.success is True
assert "[MOCK]" in result.text
def test_fallback_chain_continues_after_failed_candidate(self, temp_providers_file):
save_providers([
Provider("primary", "missing-primary", fallback_chain=["secondary", "mock"]),
Provider("secondary", "missing-secondary"),
Provider("mock", "mock"),
])
result = call_provider("primary", "Test prompt")
assert result.success is True
assert "[MOCK]" in result.text
def test_fallback_prevents_infinite_loop(self, temp_providers_file):
"""Circular fallback references should not cause infinite loop."""
save_providers([
@ -626,3 +699,88 @@ class TestMaxTokens:
assert "--max-tokens 4096" in calls[0][0][0]
assert "--max-tokens 4096" in calls[1][0][0]
class TestApiProviders:
@pytest.fixture
def temp_providers_file(self, tmp_path):
providers_file = tmp_path / ".cmdforge" / "providers.yaml"
with patch('cmdforge.providers.PROVIDERS_FILE', providers_file):
yield providers_file
@patch('requests.post')
def test_openai_compatible_api_dispatch(self, mock_post, temp_providers_file):
response = MagicMock(status_code=200)
response.json.return_value = {
"choices": [{"message": {"content": "API response"}}]
}
mock_post.return_value = response
save_providers([
Provider(
"test-api",
"https://example.test/v1",
type="api",
model="example/model",
api_key_env="TEST_API_KEY",
)
])
with patch.dict('os.environ', {"TEST_API_KEY": "secret"}):
result = call_provider("test-api", "Prompt", max_tokens=128)
assert result.success is True
assert result.text == "API response"
mock_post.assert_called_once_with(
"https://example.test/v1/chat/completions",
json={
"model": "example/model",
"messages": [{"role": "user", "content": "Prompt"}],
"max_tokens": 128,
},
headers={
"Authorization": "Bearer secret",
"Content-Type": "application/json",
},
timeout=300,
)
def test_api_provider_requires_configured_key(self, temp_providers_file):
save_providers([
Provider(
"test-api",
"https://example.test/v1",
type="api",
model="example/model",
api_key_env="MISSING_TEST_API_KEY",
)
])
with patch.dict('os.environ', {"MISSING_TEST_API_KEY": ""}):
result = call_provider("test-api", "Prompt")
assert result.success is False
assert "MISSING_TEST_API_KEY" in result.error
class TestProviderDiscovery:
def test_ollama_model_names_include_tags_and_are_unique(self):
ollama_list = MagicMock(
returncode=0,
stdout=(
"NAME ID SIZE MODIFIED\n"
"hermes4.3:latest abc 1 GB now\n"
"hermes4.3:q4_k_m def 1 GB now\n"
),
)
def which(binary):
return "/usr/bin/ollama" if binary == "ollama" else None
with patch('cmdforge.providers.shutil.which', side_effect=which):
with patch('cmdforge.providers.subprocess.run', return_value=ollama_list):
found = discover_installed_providers()
models = [item for item in found if item["source"] == "ollama-model"]
assert len(models) == 2
assert len({item["name"] for item in models}) == 2
assert models[0]["name"] == "ollama-hermes4-3-latest"
assert models[1]["name"] == "ollama-hermes4-3-q4-k-m"

7
tests/test_registry_integration.py
View File

@ -202,6 +202,13 @@ class TestConfig:
assert restored.auto_fetch_from_registry is False
assert restored.default_provider == "claude"
def test_save_config_uses_private_permissions(self, tmp_path):
with patch("cmdforge.config.CONFIG_DIR", tmp_path / ".cmdforge"):
path = save_config(Config(registry=RegistryConfig(token="test_token")))
assert oct(path.parent.stat().st_mode & 0o777) == "0o700"
assert oct(path.stat().st_mode & 0o777) == "0o600"
class TestRegistryClient:
"""Tests for the registry client (mocked)."""

92
tests/test_runner.py
View File

@ -8,7 +8,8 @@ from cmdforge.runner import (
execute_prompt_step,
execute_code_step,
run_tool,
create_argument_parser
create_argument_parser,
collect_custom_args
)
from cmdforge.tool import Tool, ToolArgument, PromptStep, CodeStep
from cmdforge.providers import ProviderResult
@ -246,6 +247,47 @@ class TestExecutePromptStep:
# Should use override, not step's provider
assert mock_call.call_args[0][0] == "gpt4"
@patch('cmdforge.runner.call_provider')
def test_prompt_file_loaded_relative_to_tool_dir(self, mock_call, tmp_path):
mock_call.return_value = ProviderResult(text="response", success=True)
tool_dir = tmp_path / "tool"
tool_dir.mkdir()
(tool_dir / "prompt.txt").write_text("Summarize: {input}")
step = PromptStep(
prompt="",
provider="claude",
output_var="out",
prompt_file="prompt.txt",
plain_text=True,
)
output, success = execute_prompt_step(
step,
{"input": "file text"},
base_dir=tool_dir,
)
assert success is True
assert output == "response"
assert mock_call.call_args[0][1] == "Summarize: file text"
def test_prompt_file_rejects_path_traversal(self, tmp_path):
tool_dir = tmp_path / "tool"
tool_dir.mkdir()
(tmp_path / "outside.txt").write_text("Nope")
step = PromptStep(
prompt="",
provider="mock",
output_var="out",
prompt_file="../outside.txt",
plain_text=True,
)
output, success = execute_prompt_step(step, {"input": ""}, base_dir=tool_dir)
assert success is False
assert output == ""
class TestStructuredOutput:
"""Tests for structured output enforcement."""
@ -685,6 +727,38 @@ result = multiply_all([1, 2, 3])
assert success is True
assert outputs["result"] == [10, 20, 30]
def test_code_file_loaded_relative_to_tool_dir(self, tmp_path):
tool_dir = tmp_path / "tool"
tool_dir.mkdir()
(tool_dir / "process.py").write_text("result = input.upper()")
step = CodeStep(code="", code_file="process.py", output_var="result")
outputs, success = execute_code_step(
step,
{"input": "hello"},
step_num=1,
base_dir=tool_dir,
)
assert success is True
assert outputs["result"] == "HELLO"
def test_code_file_rejects_path_traversal(self, tmp_path):
tool_dir = tmp_path / "tool"
tool_dir.mkdir()
(tmp_path / "outside.py").write_text("result = 'bad'")
step = CodeStep(code="", code_file="../outside.py", output_var="result")
outputs, success = execute_code_step(
step,
{"input": ""},
step_num=1,
base_dir=tool_dir,
)
assert success is False
assert outputs == {}
class TestRunTool:
"""Tests for run_tool function."""
@ -920,3 +994,19 @@ class TestCreateArgumentParser:
assert args.input_file == "input.txt"
assert args.output_file == "output.txt"
def test_collect_custom_args_uses_parser_namespace(self):
tool = Tool(
name="test",
arguments=[
ToolArgument(flag="--max-size", variable="max_size", default="100"),
ToolArgument(flag="-f", variable="format"),
]
)
parser = create_argument_parser(tool)
args = parser.parse_args(["--max-size", "50", "-f", "json"])
assert collect_custom_args(tool, args) == {
"max_size": "50",
"format": "json",
}

38
tests/test_tool.py
View File

@ -142,6 +142,28 @@ class TestPromptStep:
assert restored.output_var == original.output_var
assert restored.prompt_file == original.prompt_file
def test_from_dict_rejects_non_integer_max_tokens(self):
data = {
"type": "prompt",
"prompt": "Test",
"provider": "claude",
"output_var": "result",
"max_tokens": "100; touch /tmp/pwned",
}
with pytest.raises(ValueError, match="max_tokens"):
PromptStep.from_dict(data)
def test_from_dict_rejects_out_of_range_max_tokens(self):
data = {
"type": "prompt",
"prompt": "Test",
"provider": "claude",
"output_var": "result",
"max_tokens": 0,
}
with pytest.raises(ValueError, match="max_tokens"):
PromptStep.from_dict(data)
class TestCodeStep:
"""Tests for CodeStep dataclass."""
@ -507,6 +529,22 @@ class TestLegacyFormat:
assert len(tool.arguments) == 1
assert tool.arguments[0].variable == "max_size"
def test_legacy_format_preserves_visibility(self, temp_tools_dir):
tool_dir = temp_tools_dir / "legacy-private"
tool_dir.mkdir(parents=True)
legacy_config = {
"name": "legacy-private",
"prompt": "Process: {input}",
"provider": "mock",
"visibility": "private",
}
(tool_dir / "config.yaml").write_text(yaml.dump(legacy_config))
tool = load_tool("legacy-private")
assert tool is not None
assert tool.visibility == "private"
class TestDefaultCategories:
"""Tests for default categories."""

14
tests/test_version.py
View File

@ -216,3 +216,17 @@ class TestCompareVersions:
compare_versions("invalid", "1.0.0")
with pytest.raises(ValueError):
compare_versions("1.0.0", "invalid")
class TestRegexEdgeCases:
"""Tests for semver regex edge cases."""
def test_dot_separators_required(self):
assert Version.parse("1x2x3") is None
assert Version.parse("1-2-3") is None
def test_double_dot_rejected(self):
assert Version.parse("1..2.3") is None
def test_trailing_dot_rejected(self):
assert Version.parse("1.2.3.") is None