feat(providers): implement Packet S4 agent-harness-setup CLI wrapper and provider handoff

This commit is contained in:
rob
2026-09-11 22:31:08 -03:00
parent 1e698996b2
commit 60cd44bcfb
3 changed files with 442 additions and 128 deletions
+2
View File
@@ -190,6 +190,8 @@ def main():
# providers install
p_prov_install = providers_sub.add_parser("install", help="Interactive guide to install AI providers")
p_prov_install.add_argument("--harness", help="Specific harness ID to install")
p_prov_install.add_argument("--dry-run", action="store_true", help="Preview installation without executing")
p_prov_install.set_defaults(func=cmd_providers)
# providers discover
+224 -128
View File
@@ -1,12 +1,15 @@
"""Provider management commands."""
import json
import os
import shlex
import shutil
import subprocess
import sys
import tempfile
import uuid
from pathlib import Path
from ..providers import (
KNOWN_PROVIDER_CLIS,
PRESET_CHAINS,
Provider,
add_provider,
@@ -121,142 +124,235 @@ def cmd_providers(args):
return 0
def _cmd_providers_install(args):
"""Interactive guide to install AI providers."""
print("=" * 60)
print("CmdForge Provider Installation Guide")
print("=" * 60)
print()
def _find_setup_helper() -> list[str] | None:
"""Locate the agent-harness-setup CLI executable or python module."""
# 1. System/user PATH
path_bin = shutil.which("agent-harness-setup")
if path_bin:
return [path_bin]
# Check what's already installed
providers = load_providers()
installed_groups = set()
for p in providers:
if p.name.lower() == "mock" or p.type == "api":
continue
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", []):
installed_groups.add(group)
# 2. Current environment virtualenv bin
venv_bin = Path(sys.prefix) / "bin" / "agent-harness-setup"
if venv_bin.is_file() and os.access(venv_bin, os.X_OK):
return [str(venv_bin)]
# Show available provider groups
print("Available AI Provider Groups:\n")
options = []
for i, (key, info) in enumerate(PROVIDER_INSTALL_INFO.items(), 1):
status = "[INSTALLED]" if key in installed_groups else ""
print(f" {i}. {info['group']} {status}")
print(f" Cost: {info['cost']}")
print(f" Models: {', '.join(info['variants']) if info['variants'] else 'Custom'}")
print()
options.append(key)
# 3. User local bin
local_bin = Path.home() / ".local" / "bin" / "agent-harness-setup"
if local_bin.is_file() and os.access(local_bin, os.X_OK):
return [str(local_bin)]
print(" 0. Cancel")
print()
# 4. Workspace sibling repository
workspace_candidates = [
Path(__file__).resolve().parents[4] / "agent-harness-setup" / ".venv" / "bin" / "agent-harness-setup",
Path("/home/rob/PycharmProjects/agent-harness-setup/.venv/bin/agent-harness-setup"),
]
for cand in workspace_candidates:
if cand.is_file() and os.access(cand, os.X_OK):
return [str(cand)]
# 5. Check if agent_harness_setup is importable via python
try:
choice = input("Select a provider to install (1-{}, or 0 to cancel): ".format(len(options)))
choice = int(choice)
except (ValueError, EOFError):
print("Cancelled.")
return 0
import agent_harness_setup # type: ignore # noqa: F401
return [sys.executable, "-m", "agent_harness_setup"]
except ImportError:
pass
if choice == 0 or choice > len(options):
print("Cancelled.")
return 0
return None
selected = options[choice - 1]
info = PROVIDER_INSTALL_INFO[selected]
print()
print("=" * 60)
print(f"Installing: {info['group']}")
print("=" * 60)
print()
print(f"Requirements: {info['requires']}")
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:
confirm = "n"
if confirm == "y":
print()
print(f"Running: {info['install_cmd']}")
print("-" * 40)
result = subprocess.run(info['install_cmd'], shell=True)
print("-" * 40)
if result.returncode == 0:
# Refresh PATH to pick up newly installed tools
import os
new_paths = []
# Common install locations that might have been added
potential_paths = [
Path.home() / ".opencode" / "bin", # OpenCode
Path.home() / ".local" / "bin", # pip/pipx installs
Path("/usr/local/bin"), # Ollama, system installs
]
# Also try to get npm global bin path
try:
npm_result = subprocess.run(
["npm", "bin", "-g"],
capture_output=True, text=True, timeout=5
)
if npm_result.returncode == 0:
npm_bin = npm_result.stdout.strip()
if npm_bin:
potential_paths.append(Path(npm_bin))
except (subprocess.TimeoutExpired, FileNotFoundError):
pass
current_path = os.environ.get("PATH", "")
for p in potential_paths:
if p.exists() and str(p) not in current_path:
new_paths.append(str(p))
if new_paths:
os.environ["PATH"] = ":".join(new_paths) + ":" + current_path
print()
print("Installation completed!")
print()
print("IMPORTANT: Refresh your shell PATH before continuing:")
print(" source ~/.bashrc")
print()
print(f"Next steps:")
print(f" 1. source ~/.bashrc (required!)")
print(f" 2. {info['setup']}")
if info.get('post_install_note'):
print(f" 3. {info['post_install_note']}")
print(f" 4. Test with: cmdforge providers test {selected}")
def _get_provider_defaults_for_harness(harness_id: str, executable_path: str | None = None) -> dict:
"""Derive CmdForge provider configuration defaults for an installed harness."""
harness_map = {
"claude-code": "claude",
"claude": "claude",
"openai-codex": "codex",
"codex": "codex",
"agy": "agy",
"gemini-cli": "agy",
"opencode": "opencode",
"crush": "crush",
"ollama": "ollama",
}
cli_key = harness_map.get(harness_id, harness_id)
if cli_key in KNOWN_PROVIDER_CLIS:
base = dict(KNOWN_PROVIDER_CLIS[cli_key])
if executable_path:
old_cmd = base.get("command", "")
parts = shlex.split(old_cmd)
if parts:
base["command"] = " ".join([executable_path] + parts[1:])
else:
print(f" 3. Test with: cmdforge providers test {info['variants'][0] if info['variants'] else selected}")
base["command"] = executable_path
return base
cmd = executable_path if executable_path else harness_id
return {
"name": harness_id,
"command": cmd,
"description": f"{harness_id} CLI provider",
"tags": ["cli"],
"locality": "remote",
"capabilities": ["text"],
"data_policy": "public",
}
def _cmd_providers_install(args):
"""Interactive guide to install AI harnesses and configure providers."""
helper_argv = _find_setup_helper()
if not helper_argv:
print("=" * 60)
print("CmdForge Provider Installation")
print("=" * 60)
print()
print("Notice: Agent Harness Setup Helper ('agent-harness-setup') was not")
print("found on your PATH or in active virtual environments.")
print()
print("To install the setup helper, run:")
print(" pip install -e /home/rob/PycharmProjects/agent-harness-setup")
print()
print("Or run the helper directly:")
print(" PYTHONPATH=/home/rob/PycharmProjects/agent-harness-setup/src python3 -m agent_harness_setup")
print()
print("Manual installation instructions are available in documentation:")
print(" https://docs.anthropic.com/en/docs/agents-and-tools/claude-code/overview")
return 0
# Prepare secure temporary result file and correlation request_id
temp_dir = tempfile.mkdtemp(prefix="cmdforge-harness-setup-")
try:
os.chmod(temp_dir, 0o700)
except OSError:
pass
result_file = Path(temp_dir) / "result.json"
req_id = str(uuid.uuid4())
cmd = list(helper_argv) + ["--result-file", str(result_file), "--request-id", req_id]
harness_arg = getattr(args, "harness", None)
if harness_arg:
cmd.extend(["--harness", harness_arg])
if getattr(args, "dry_run", False):
cmd.append("--dry-run")
try:
proc = subprocess.run(cmd)
helper_exit_code = proc.returncode
except Exception as e:
print(f"\nError executing agent-harness-setup: {e}")
shutil.rmtree(temp_dir, ignore_errors=True)
return 1
# Read and parse completion result
result_data = None
if result_file.is_file():
try:
result_data = json.loads(result_file.read_text(encoding="utf-8"))
except Exception:
result_data = None
# Clean up temporary directory
shutil.rmtree(temp_dir, ignore_errors=True)
# Validate correlation request ID
if result_data is not None and result_data.get("request_id") != req_id:
print("\nWarning: Correlation request ID mismatch in setup result. Rejecting result.")
return 1
# Incoherence handling (Contract §C.5)
if result_data is not None:
status = result_data.get("status")
if helper_exit_code != 0 and status == "success":
print("\nWarning: Helper exited with non-zero code but result reported success. Treating as failure.")
return helper_exit_code
if helper_exit_code == 0 and status in ("failed", "cancelled"):
print("\nWarning: Helper exited with code 0 but result reported failure/cancellation. Treating as failure.")
return 1
# Dry-run completion
if getattr(args, "dry_run", False) or (result_data and result_data.get("status") == "dry_run"):
return 0
# Cancellation
if helper_exit_code in (10, 130, 143) or (result_data and result_data.get("status") == "cancelled"):
print("\nSetup cancelled. No provider configuration was changed.")
return 0
# Failure
if helper_exit_code != 0 or (result_data and result_data.get("status") == "failed"):
diag = result_data.get("error_diagnostics") if result_data else None
if diag:
print(f"\nSetup failed: {diag}")
return helper_exit_code if helper_exit_code != 0 else 1
# Verified Success
if helper_exit_code == 0 and result_data and result_data.get("status") == "success":
harness_id = result_data.get("harness_id") or "unknown"
verification = result_data.get("verification") or {}
executable_path = verification.get("executable_path")
path_updates = result_data.get("path_updates") or []
# Advisory PATH updates
if path_updates:
cur_path = os.environ.get("PATH", "")
new_dirs = [
str(Path(os.path.expanduser(p)))
for p in path_updates
if Path(os.path.expanduser(p)).is_dir() and str(Path(os.path.expanduser(p))) not in cur_path
]
if new_dirs:
os.environ["PATH"] = ":".join(new_dirs) + ":" + cur_path
provider_defaults = _get_provider_defaults_for_harness(harness_id, executable_path)
provider_name = provider_defaults["name"]
print()
print("=" * 60)
print(f"Harness Setup Verified: {harness_id}")
print("=" * 60)
if executable_path:
print(f"Verified executable: {executable_path}")
if result_data.get("manual_steps_remaining"):
print("\nRemaining setup steps:")
for step in result_data["manual_steps_remaining"]:
print(f" * {step}")
existing = get_provider(provider_name)
if existing:
print(f"\nProvider '{provider_name}' is already configured in ~/.cmdforge/providers.yaml.")
print(f"Command: {existing.command}")
print(f"\nTest with: cmdforge providers test {provider_name}")
return 0
# Interactive confirmation prompt for provider configuration
print()
try:
prompt_resp = input(
f"Would you like to register '{provider_name}' as a model provider in ~/.cmdforge/providers.yaml? (Y/n): "
).strip().lower()
except (EOFError, KeyboardInterrupt):
prompt_resp = "n"
if prompt_resp in ("", "y", "yes"):
new_prov = Provider(
name=provider_name,
command=provider_defaults["command"],
description=provider_defaults.get("description", ""),
tags=provider_defaults.get("tags", []),
locality=provider_defaults.get("locality", "remote"),
capabilities=provider_defaults.get("capabilities", ["text"]),
data_policy=provider_defaults.get("data_policy", "public"),
)
add_provider(new_prov)
print(f"\nProvider '{provider_name}' successfully added to ~/.cmdforge/providers.yaml.")
print(f"Command: {new_prov.command}")
print(f"\nTest with: cmdforge providers test {provider_name}")
else:
print()
print(f"Installation failed (exit code {result.returncode})")
print("Try running the command manually:")
print(f" {info['install_cmd']}")
else:
print()
print("To install manually, run:")
print(f" {info['install_cmd']}")
print()
print(f"Then: {info['setup']}")
print("\nSkipped provider registration. You can add it later with:")
print(f" cmdforge providers add {provider_name} '{provider_defaults['command']}'")
return 0
return 0
+216
View File
@@ -0,0 +1,216 @@
"""Tests for Packet S4: CmdForge CLI integration with agent-harness-setup."""
import json
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
from cmdforge.cli import main
from cmdforge.cli.provider_commands import _find_setup_helper, _get_provider_defaults_for_harness
from cmdforge.providers import Provider, get_provider, load_providers, save_providers
@pytest.fixture
def temp_providers_file(tmp_path):
providers_file = tmp_path / ".cmdforge" / "providers.yaml"
with patch("cmdforge.providers.PROVIDERS_FILE", providers_file):
save_providers([])
yield providers_file
def _mock_helper_run(status: str, harness_id: str = "claude-code", exit_code: int = 0, diagnostics: str | None = None, executable_path: str = "/usr/local/bin/claude"):
"""Create a mock subprocess.run that writes a valid completion result to --result-file."""
def _run(cmd, *args, **kwargs):
# Extract --result-file and --request-id from cmd
res_idx = cmd.index("--result-file")
result_file_path = Path(cmd[res_idx + 1])
req_idx = cmd.index("--request-id")
request_id = cmd[req_idx + 1]
data = {
"schema_version": 1,
"helper_version": "0.1.0",
"request_id": request_id,
"timestamp": "2026-09-11T22:00:00Z",
"status": status,
"harness_id": harness_id,
"recipe_id": harness_id,
"recipe_version": "2026-09-10",
"install_method": "argv",
"installer_exit_code": 0 if status == "success" else None,
"verification": {
"verified": True,
"executable_path": executable_path,
"version_output": f"{harness_id} 1.0.0",
} if status == "success" else None,
"path_updates": ["/usr/local/bin"],
"auth_status": "setup_required",
"manual_steps_remaining": ["Run 'claude' to complete browser sign-in."],
"partial_changes": {
"observed_created": [executable_path] if status == "success" else [],
"observed_modified": [],
"unknown_changes": True,
},
"error_diagnostics": diagnostics,
}
result_file_path.write_text(json.dumps(data), encoding="utf-8")
return SimpleNamespace(returncode=exit_code)
return _run
def test_helper_discovery_finds_executable():
"""Verify _find_setup_helper locates agent-harness-setup."""
with patch("shutil.which", return_value="/usr/local/bin/agent-harness-setup"):
helper = _find_setup_helper()
assert helper == ["/usr/local/bin/agent-harness-setup"]
def test_get_provider_defaults_for_claude():
"""Verify provider defaults derived for claude-code."""
defaults = _get_provider_defaults_for_harness("claude-code", "/custom/bin/claude")
assert defaults["name"] == "claude"
assert defaults["command"] == "/custom/bin/claude -p"
assert "Claude" in defaults["description"]
def test_providers_install_verified_success_registers_provider(temp_providers_file, capsys):
"""Test verified installation prompts and registers provider in providers.yaml."""
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["agent-harness-setup"]),
patch("subprocess.run", side_effect=_mock_helper_run("success", "claude-code", 0)) as mock_run,
patch("builtins.input", return_value="y"),
patch("sys.argv", ["cmdforge", "providers", "install", "--harness", "claude-code"]),
):
exit_code = main()
assert exit_code == 0
captured = capsys.readouterr()
assert "Harness Setup Verified: claude-code" in captured.out
assert "Provider 'claude' successfully added" in captured.out
# Verify provider stored in providers.yaml
provider = get_provider("claude")
assert provider is not None
assert provider.command == "/usr/local/bin/claude -p"
assert "code" in provider.tags
# Verify subprocess invocation arguments
call_args = mock_run.call_args[0][0]
assert call_args[0] == "agent-harness-setup"
assert "--result-file" in call_args
assert "--request-id" in call_args
assert "--harness" in call_args
assert "claude-code" in call_args
def test_providers_install_declined_registration(temp_providers_file, capsys):
"""Test declining the provider registration prompt leaves providers.yaml unchanged."""
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["agent-harness-setup"]),
patch("subprocess.run", side_effect=_mock_helper_run("success", "claude-code", 0)),
patch("builtins.input", return_value="n"),
patch("sys.argv", ["cmdforge", "providers", "install"]),
):
exit_code = main()
assert exit_code == 0
captured = capsys.readouterr()
assert "Harness Setup Verified: claude-code" in captured.out
assert "Skipped provider registration" in captured.out
assert get_provider("claude") is None
def test_providers_install_already_configured(temp_providers_file, capsys):
"""Test that existing provider is recognized and not re-registered."""
save_providers([Provider("claude", "existing-claude -p")])
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["agent-harness-setup"]),
patch("subprocess.run", side_effect=_mock_helper_run("success", "claude-code", 0)),
patch("sys.argv", ["cmdforge", "providers", "install"]),
):
exit_code = main()
assert exit_code == 0
captured = capsys.readouterr()
assert "already configured in ~/.cmdforge/providers.yaml" in captured.out
assert get_provider("claude").command == "existing-claude -p"
def test_providers_install_cancelled(temp_providers_file, capsys):
"""Test cancellation exits cleanly without changing providers."""
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["agent-harness-setup"]),
patch("subprocess.run", side_effect=_mock_helper_run("cancelled", "claude-code", 10)),
patch("sys.argv", ["cmdforge", "providers", "install"]),
):
exit_code = main()
assert exit_code == 0
captured = capsys.readouterr()
assert "Setup cancelled. No provider configuration was changed." in captured.out
assert get_provider("claude") is None
def test_providers_install_execution_failure(temp_providers_file, capsys):
"""Test helper failure reports error diagnostics and does not register provider."""
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["agent-harness-setup"]),
patch("subprocess.run", side_effect=_mock_helper_run("failed", "claude-code", 1, "Missing node.js")),
patch("sys.argv", ["cmdforge", "providers", "install"]),
):
exit_code = main()
assert exit_code == 1
captured = capsys.readouterr()
assert "Setup failed: Missing node.js" in captured.out
assert get_provider("claude") is None
def test_providers_install_incoherent_result_rejected(temp_providers_file, capsys):
"""Test exit code 1 with status 'success' is rejected as incoherent."""
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["agent-harness-setup"]),
patch("subprocess.run", side_effect=_mock_helper_run("success", "claude-code", 1)),
patch("sys.argv", ["cmdforge", "providers", "install"]),
):
exit_code = main()
assert exit_code == 1
captured = capsys.readouterr()
assert "Treating as failure" in captured.out
assert get_provider("claude") is None
def test_providers_install_helper_not_found(temp_providers_file, capsys):
"""Test that missing helper prints actionable instructions without crashing."""
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=None),
patch("sys.argv", ["cmdforge", "providers", "install"]),
):
exit_code = main()
assert exit_code == 0
captured = capsys.readouterr()
assert "agent-harness-setup" in captured.out
assert "pip install" in captured.out
assert "PYTHONPATH=" in captured.out
def test_providers_install_dry_run(temp_providers_file, capsys):
"""Test --dry-run passes flag to helper and makes no provider changes."""
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["agent-harness-setup"]),
patch("subprocess.run", side_effect=_mock_helper_run("dry_run", "claude-code", 0)) as mock_run,
patch("sys.argv", ["cmdforge", "providers", "install", "--dry-run"]),
):
exit_code = main()
assert exit_code == 0
call_args = mock_run.call_args[0][0]
assert "--dry-run" in call_args
assert get_provider("claude") is None