fix(classic): resolve CI lint, type, and test failures

- Fix line-too-long in test_permissions.py docstring
- Fix type annotation in validators.py (callable -> Callable)
- Add --fresh flag to benchmark tests to prevent state resumption
- Exclude direct_benchmark/adapters from pyright (optional deps)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Nicholas Tindle
2026-01-29 14:31:11 -06:00
parent 0040636948
commit 791e1d8982
11 changed files with 1772 additions and 1 deletions

View File

@@ -66,6 +66,7 @@ jobs:
run: |
echo "Testing ReadFile challenge with one_shot strategy..."
poetry run direct-benchmark run \
--fresh \
--strategies one_shot \
--models claude \
--tests ReadFile \
@@ -73,6 +74,7 @@ jobs:
echo "Testing WriteFile challenge..."
poetry run direct-benchmark run \
--fresh \
--strategies one_shot \
--models claude \
--tests WriteFile \
@@ -87,6 +89,7 @@ jobs:
run: |
echo "Testing coding category..."
poetry run direct-benchmark run \
--fresh \
--strategies one_shot \
--models claude \
--categories coding \
@@ -102,6 +105,7 @@ jobs:
run: |
echo "Testing multiple strategies..."
poetry run direct-benchmark run \
--fresh \
--strategies one_shot,plan_execute \
--models claude \
--tests ReadFile \
@@ -145,6 +149,7 @@ jobs:
run: |
echo "Running regression tests (previously beaten challenges)..."
poetry run direct-benchmark run \
--fresh \
--strategies one_shot \
--models claude \
--maintain \

View File

@@ -640,7 +640,7 @@ class TestApprovalScopes:
def test_agent_approval_auto_approves_subsequent_calls(
self, workspace: Path, agent_dir: Path
):
"""After AGENT approval, subsequent calls to same command type should auto-approve.
"""After AGENT approval, subsequent calls should auto-approve.
This tests the scenario where multiple tools are executed in sequence -
after approving the first one with 'Always (this agent)', subsequent

View File

@@ -0,0 +1,21 @@
"""Settings UI package for AutoGPT configuration."""
from .categories import CATEGORIES, Category, get_categories_for_display
from .env_file import find_env_file, get_default_env_path, load_env_file, save_env_file
from .introspection import SettingInfo, get_complete_settings
from .ui import SettingsUI
from .validators import validate_setting
__all__ = [
"CATEGORIES",
"Category",
"SettingsUI",
"SettingInfo",
"find_env_file",
"get_categories_for_display",
"get_complete_settings",
"get_default_env_path",
"load_env_file",
"save_env_file",
"validate_setting",
]

View File

@@ -0,0 +1,225 @@
"""Settings category definitions."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .introspection import SettingInfo
@dataclass
class Category:
"""A category of settings."""
id: str
name: str
description: str
env_vars: list[str] = field(default_factory=list)
icon: str = ""
def get_settings(
self, all_settings: dict[str, "SettingInfo"]
) -> list["SettingInfo"]:
"""Get the settings in this category."""
return [all_settings[env] for env in self.env_vars if env in all_settings]
# Define all categories with their env vars
CATEGORIES: list[Category] = [
Category(
id="api_keys",
name="API Keys",
description="LLM provider API keys",
icon="🔑",
env_vars=[
"OPENAI_API_KEY",
"ANTHROPIC_API_KEY",
"GROQ_API_KEY",
],
),
Category(
id="models",
name="Models",
description="LLM model configuration",
icon="🤖",
env_vars=[
"SMART_LLM",
"FAST_LLM",
"TEMPERATURE",
"EMBEDDING_MODEL",
"PROMPT_STRATEGY",
"THINKING_BUDGET_TOKENS",
"REASONING_EFFORT",
],
),
Category(
id="search",
name="Web Search",
description="Search provider configuration",
icon="🔍",
env_vars=[
"TAVILY_API_KEY",
"SERPER_API_KEY",
"GOOGLE_API_KEY",
"GOOGLE_CUSTOM_SEARCH_ENGINE_ID",
],
),
Category(
id="storage",
name="Storage",
description="File storage configuration",
icon="💾",
env_vars=[
"FILE_STORAGE_BACKEND",
"STORAGE_BUCKET",
"S3_ENDPOINT_URL",
"RESTRICT_TO_WORKSPACE",
],
),
Category(
id="image",
name="Image Gen",
description="Image generation configuration",
icon="🎨",
env_vars=[
"HUGGINGFACE_API_TOKEN",
"SD_WEBUI_AUTH",
],
),
Category(
id="github",
name="GitHub",
description="GitHub integration",
icon="🐙",
env_vars=[
"GITHUB_API_KEY",
"GITHUB_USERNAME",
],
),
Category(
id="tts",
name="Text-to-Speech",
description="Text-to-speech configuration",
icon="🔊",
env_vars=[
"TEXT_TO_SPEECH_PROVIDER",
"ELEVENLABS_API_KEY",
"ELEVENLABS_VOICE_ID",
"STREAMELEMENTS_VOICE",
],
),
Category(
id="logging",
name="Logging",
description="Logging configuration",
icon="📝",
env_vars=[
"LOG_LEVEL",
"LOG_FORMAT",
"LOG_FILE_FORMAT",
"PLAIN_OUTPUT",
],
),
Category(
id="server",
name="Server",
description="Agent Protocol server settings",
icon="🌐",
env_vars=[
"AP_SERVER_PORT",
"AP_SERVER_DB_URL",
"AP_SERVER_CORS_ALLOWED_ORIGINS",
],
),
Category(
id="app",
name="Application",
description="Application settings",
icon="⚙️",
env_vars=[
"AUTHORISE_COMMAND_KEY",
"EXIT_KEY",
"NONINTERACTIVE_MODE",
"DISABLED_COMMANDS",
"TELEMETRY_OPT_IN",
"COMPONENT_CONFIG_FILE",
],
),
Category(
id="openai",
name="OpenAI",
description="OpenAI-specific settings",
icon="🟢",
env_vars=[
"OPENAI_API_BASE_URL",
"OPENAI_ORGANIZATION",
"OPENAI_API_TYPE",
"OPENAI_API_VERSION",
"AZURE_CONFIG_FILE",
],
),
Category(
id="anthropic",
name="Anthropic",
description="Anthropic-specific settings",
icon="🟠",
env_vars=[
"ANTHROPIC_API_BASE_URL",
],
),
Category(
id="groq",
name="Groq",
description="Groq-specific settings",
icon="🟣",
env_vars=[
"GROQ_API_BASE_URL",
],
),
Category(
id="platform",
name="Platform",
description="AutoGPT Platform integration",
icon="🚀",
env_vars=[
"PLATFORM_API_KEY",
"PLATFORM_BLOCKS_ENABLED",
"PLATFORM_URL",
"PLATFORM_TIMEOUT",
],
),
Category(
id="local_llm",
name="Local LLM",
description="Local LLM configuration (Llamafile, etc.)",
icon="💻",
env_vars=[
"LLAMAFILE_API_BASE",
],
),
]
def get_category_by_id(category_id: str) -> Category | None:
"""Get a category by its ID."""
for cat in CATEGORIES:
if cat.id == category_id:
return cat
return None
def get_categories_for_display() -> list[Category]:
"""Get categories in display order, filtered to those with settings."""
# Return categories that have at least one env var defined
return [cat for cat in CATEGORIES if cat.env_vars]
def categorize_env_vars() -> dict[str, str]:
"""Create a mapping of env var to category id."""
mapping: dict[str, str] = {}
for cat in CATEGORIES:
for env_var in cat.env_vars:
mapping[env_var] = cat.id
return mapping

View File

@@ -0,0 +1,192 @@
"""Utilities for reading and writing .env files."""
from __future__ import annotations
import re
from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .categories import Category
def load_env_file(path: Path) -> dict[str, str]:
"""Load environment variables from a .env file.
Handles:
- KEY=VALUE format
- Quoted values (single and double quotes)
- Comments (lines starting with #)
- Empty lines
Args:
path: Path to the .env file
Returns:
Dict mapping variable names to values
"""
settings: dict[str, str] = {}
if not path.exists():
return settings
with open(path, "r") as f:
for line in f:
line = line.strip()
# Skip empty lines and comments
if not line or line.startswith("#"):
continue
# Parse KEY=VALUE
match = re.match(r"^([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(.*)$", line)
if not match:
continue
key = match.group(1)
value = match.group(2).strip()
# Handle quoted values
if len(value) >= 2:
if (value.startswith('"') and value.endswith('"')) or (
value.startswith("'") and value.endswith("'")
):
value = value[1:-1]
settings[key] = value
return settings
def save_env_file(
path: Path,
settings: dict[str, str],
categories: list["Category"],
) -> None:
"""Save environment variables to a .env file.
Organizes settings by category with headers and preserves any
existing comments or settings not in our category list.
Args:
path: Path to the .env file
settings: Dict mapping variable names to values
categories: List of categories for organization
"""
# Ensure parent directory exists
path.parent.mkdir(parents=True, exist_ok=True)
lines: list[str] = []
# Header
lines.append("#" * 80)
lines.append("### AutoGPT - CONFIGURATION FILE")
lines.append("#" * 80)
lines.append("#")
lines.append("# Generated by `autogpt config`")
lines.append("#")
lines.append("#" * 80)
lines.append("")
# Track which settings we've written
written_keys: set[str] = set()
# Write settings by category
for category in categories:
category_settings = []
for env_var in category.env_vars:
if env_var in settings and settings[env_var]:
category_settings.append((env_var, settings[env_var]))
written_keys.add(env_var)
# Only write category if it has settings
if category_settings:
lines.append("#" * 80)
lines.append(f"### {category.name.upper()}")
lines.append("#" * 80)
lines.append("")
for key, value in category_settings:
# Quote values that need it
if _needs_quoting(value):
value = f'"{_escape_value(value)}"'
lines.append(f"{key}={value}")
lines.append("")
# Write any remaining settings not in categories
remaining = {k: v for k, v in settings.items() if k not in written_keys and v}
if remaining:
lines.append("#" * 80)
lines.append("### OTHER SETTINGS")
lines.append("#" * 80)
lines.append("")
for key, value in sorted(remaining.items()):
if _needs_quoting(value):
value = f'"{_escape_value(value)}"'
lines.append(f"{key}={value}")
lines.append("")
# Write to file
with open(path, "w") as f:
f.write("\n".join(lines))
def _needs_quoting(value: str) -> bool:
"""Check if a value needs to be quoted in .env format."""
if not value:
return False
# Quote if contains spaces, special chars, or starts/ends with whitespace
if " " in value or "\t" in value:
return True
if value[0].isspace() or value[-1].isspace():
return True
if any(c in value for c in ["#", "'", '"', "\\", "\n", "\r"]):
return True
return False
def _escape_value(value: str) -> str:
"""Escape special characters in a value for .env format."""
# Escape backslashes first
value = value.replace("\\", "\\\\")
# Escape double quotes
value = value.replace('"', '\\"')
# Escape newlines
value = value.replace("\n", "\\n")
value = value.replace("\r", "\\r")
return value
def get_default_env_path() -> Path:
"""Get the default .env file path.
Returns ~/.autogpt/.env for user-level configuration.
"""
return Path.home() / ".autogpt" / ".env"
def find_env_file() -> Path | None:
"""Find an existing .env file in standard locations.
Searches in order:
1. Current working directory (./.env)
2. User config directory (~/.autogpt/.env)
3. XDG config directory (~/.config/autogpt/.env)
Returns:
Path to the first found .env file, or None if not found
"""
search_paths = [
Path.cwd() / ".env",
Path.home() / ".autogpt" / ".env",
Path.home() / ".config" / "autogpt" / ".env",
]
for path in search_paths:
if path.exists():
return path
return None

View File

@@ -0,0 +1,359 @@
"""Introspect Pydantic models to extract UserConfigurable fields."""
from __future__ import annotations
import enum
from dataclasses import dataclass, field
from typing import Any, Literal, Type, Union, get_args, get_origin
from pydantic import BaseModel, SecretStr
from pydantic.fields import FieldInfo
from forge.models.config import _get_field_metadata
@dataclass
class SettingInfo:
"""Information about a configurable setting."""
name: str
env_var: str
description: str
field_type: str # "str", "int", "float", "bool", "secret", "choice"
choices: list[str] = field(default_factory=list)
default: Any = None
required: bool = False
def get_display_value(self, value: Any) -> str:
"""Get display-friendly representation of a value."""
if value is None:
return "[not set]"
if self.field_type == "secret":
secret_val = (
value.get_secret_value() if isinstance(value, SecretStr) else str(value)
)
if not secret_val:
return "[not set]"
# Mask all but first 3 and last 4 characters
if len(secret_val) > 10:
return f"{secret_val[:3]}...{secret_val[-4:]}"
return "***"
if self.field_type == "bool":
return "true" if value else "false"
return str(value)
def _extract_field_type(field_info: FieldInfo) -> tuple[str, list[str]]:
"""Extract the field type and choices from a Pydantic FieldInfo.
Returns:
Tuple of (field_type, choices) where field_type is one of:
"str", "int", "float", "bool", "secret", "choice"
"""
annotation = field_info.annotation
choices: list[str] = []
# Unwrap Optional
origin = get_origin(annotation)
if origin is Union:
args = get_args(annotation)
# Filter out NoneType
non_none_args = [a for a in args if a is not type(None)]
if len(non_none_args) == 1:
annotation = non_none_args[0]
origin = get_origin(annotation)
# Check for SecretStr
if annotation is SecretStr:
return "secret", []
# Check for Literal (choices)
if origin is Literal:
choices = list(get_args(annotation))
return "choice", [str(c) for c in choices]
# Check for Enum
if isinstance(annotation, type) and issubclass(annotation, enum.Enum):
choices = [e.value for e in annotation]
return "choice", choices
# Check basic types
if annotation is bool:
return "bool", []
if annotation is int:
return "int", []
if annotation is float:
return "float", []
if annotation is str:
return "str", []
# Default to string
return "str", []
def extract_configurable_fields(
model_class: Type[BaseModel],
) -> list[SettingInfo]:
"""Extract all UserConfigurable fields from a Pydantic model.
Args:
model_class: A Pydantic BaseModel class
Returns:
List of SettingInfo objects for each configurable field
"""
settings: list[SettingInfo] = []
for name, field_info in model_class.model_fields.items():
# Check if this field is user configurable
if not _get_field_metadata(field_info, "user_configurable"):
continue
# Get the environment variable name
from_env = _get_field_metadata(field_info, "from_env")
if from_env is None:
continue
# Handle callable from_env (skip these - they're complex)
if callable(from_env):
continue
env_var = from_env
field_type, choices = _extract_field_type(field_info)
# Get default value
default = field_info.default
if default is not None and hasattr(default, "__class__"):
# Handle PydanticUndefined
if "PydanticUndefined" in str(type(default)):
default = None
settings.append(
SettingInfo(
name=name,
env_var=env_var,
description=field_info.description or "",
field_type=field_type,
choices=choices,
default=default,
required=field_info.is_required(),
)
)
return settings
def get_all_configurable_settings() -> dict[str, SettingInfo]:
"""Get all configurable settings from known models.
Returns:
Dict mapping environment variable names to SettingInfo
"""
from autogpt.app.config import AppConfig
from forge.llm.providers.anthropic import AnthropicCredentials
from forge.llm.providers.groq import GroqCredentials
from forge.llm.providers.openai import OpenAICredentials
from forge.logging.config import LoggingConfig
settings: dict[str, SettingInfo] = {}
# Extract from all known models
models = [
AppConfig,
OpenAICredentials,
AnthropicCredentials,
GroqCredentials,
LoggingConfig,
]
for model in models:
for setting in extract_configurable_fields(model):
# Use env_var as key to deduplicate
if setting.env_var not in settings:
settings[setting.env_var] = setting
return settings
# Additional env vars from .env.template that aren't in models
ADDITIONAL_ENV_VARS: dict[str, SettingInfo] = {
"TAVILY_API_KEY": SettingInfo(
name="tavily_api_key",
env_var="TAVILY_API_KEY",
description="Tavily API key for AI-optimized search",
field_type="secret",
),
"SERPER_API_KEY": SettingInfo(
name="serper_api_key",
env_var="SERPER_API_KEY",
description="Serper.dev API key for Google SERP results",
field_type="secret",
),
"GOOGLE_API_KEY": SettingInfo(
name="google_api_key",
env_var="GOOGLE_API_KEY",
description="Google API key (deprecated, use Serper)",
field_type="secret",
),
"GOOGLE_CUSTOM_SEARCH_ENGINE_ID": SettingInfo(
name="google_cse_id",
env_var="GOOGLE_CUSTOM_SEARCH_ENGINE_ID",
description="Google Custom Search Engine ID (deprecated)",
field_type="str",
),
"HUGGINGFACE_API_TOKEN": SettingInfo(
name="huggingface_api_token",
env_var="HUGGINGFACE_API_TOKEN",
description="HuggingFace API token for image generation",
field_type="secret",
),
"SD_WEBUI_AUTH": SettingInfo(
name="sd_webui_auth",
env_var="SD_WEBUI_AUTH",
description="Stable Diffusion Web UI username:password",
field_type="secret",
),
"GITHUB_API_KEY": SettingInfo(
name="github_api_key",
env_var="GITHUB_API_KEY",
description="GitHub API key / PAT",
field_type="secret",
),
"GITHUB_USERNAME": SettingInfo(
name="github_username",
env_var="GITHUB_USERNAME",
description="GitHub username",
field_type="str",
),
"TEXT_TO_SPEECH_PROVIDER": SettingInfo(
name="tts_provider",
env_var="TEXT_TO_SPEECH_PROVIDER",
description="Text-to-speech provider",
field_type="choice",
choices=["gtts", "streamelements", "elevenlabs", "macos"],
default="gtts",
),
"ELEVENLABS_API_KEY": SettingInfo(
name="elevenlabs_api_key",
env_var="ELEVENLABS_API_KEY",
description="Eleven Labs API key",
field_type="secret",
),
"ELEVENLABS_VOICE_ID": SettingInfo(
name="elevenlabs_voice_id",
env_var="ELEVENLABS_VOICE_ID",
description="Eleven Labs voice ID",
field_type="str",
),
"STREAMELEMENTS_VOICE": SettingInfo(
name="streamelements_voice",
env_var="STREAMELEMENTS_VOICE",
description="StreamElements voice name",
field_type="str",
default="Brian",
),
"FILE_STORAGE_BACKEND": SettingInfo(
name="file_storage_backend",
env_var="FILE_STORAGE_BACKEND",
description="Storage backend for file operations",
field_type="choice",
choices=["local", "gcs", "s3"],
default="local",
),
"STORAGE_BUCKET": SettingInfo(
name="storage_bucket",
env_var="STORAGE_BUCKET",
description="GCS/S3 bucket name",
field_type="str",
),
"S3_ENDPOINT_URL": SettingInfo(
name="s3_endpoint_url",
env_var="S3_ENDPOINT_URL",
description="S3 endpoint URL (for non-AWS S3)",
field_type="str",
),
"AP_SERVER_PORT": SettingInfo(
name="ap_server_port",
env_var="AP_SERVER_PORT",
description="Agent Protocol server port",
field_type="int",
default=8000,
),
"AP_SERVER_DB_URL": SettingInfo(
name="ap_server_db_url",
env_var="AP_SERVER_DB_URL",
description="Agent Protocol database URL",
field_type="str",
),
"AP_SERVER_CORS_ALLOWED_ORIGINS": SettingInfo(
name="ap_server_cors_origins",
env_var="AP_SERVER_CORS_ALLOWED_ORIGINS",
description="CORS allowed origins (comma-separated)",
field_type="str",
),
"TELEMETRY_OPT_IN": SettingInfo(
name="telemetry_opt_in",
env_var="TELEMETRY_OPT_IN",
description="Share telemetry with AutoGPT team",
field_type="bool",
default=False,
),
"PLAIN_OUTPUT": SettingInfo(
name="plain_output",
env_var="PLAIN_OUTPUT",
description="Disable animated typing and spinner",
field_type="bool",
default=False,
),
# Platform integration
"PLATFORM_API_KEY": SettingInfo(
name="platform_api_key",
env_var="PLATFORM_API_KEY",
description="AutoGPT Platform API key for blocks integration",
field_type="secret",
),
"PLATFORM_BLOCKS_ENABLED": SettingInfo(
name="platform_blocks_enabled",
env_var="PLATFORM_BLOCKS_ENABLED",
description="Enable platform blocks integration",
field_type="bool",
default=True,
),
"PLATFORM_URL": SettingInfo(
name="platform_url",
env_var="PLATFORM_URL",
description="AutoGPT Platform URL",
field_type="str",
default="https://platform.agpt.co",
),
"PLATFORM_TIMEOUT": SettingInfo(
name="platform_timeout",
env_var="PLATFORM_TIMEOUT",
description="Platform API timeout in seconds",
field_type="int",
default=60,
),
# Groq settings
"GROQ_API_BASE_URL": SettingInfo(
name="groq_api_base_url",
env_var="GROQ_API_BASE_URL",
description="Groq API base URL (for custom endpoints)",
field_type="str",
),
# Llamafile settings
"LLAMAFILE_API_BASE": SettingInfo(
name="llamafile_api_base",
env_var="LLAMAFILE_API_BASE",
description="Llamafile API base URL",
field_type="str",
default="http://localhost:8080/v1",
),
}
def get_complete_settings() -> dict[str, SettingInfo]:
"""Get all settings including additional env vars."""
settings = get_all_configurable_settings()
settings.update(ADDITIONAL_ENV_VARS)
return settings

View File

@@ -0,0 +1,454 @@
"""Main Settings UI class - tabbed settings browser."""
from __future__ import annotations
import sys
import termios
import tty
from pathlib import Path
from typing import Any
from rich.console import Console
from rich.panel import Panel
from rich.text import Text
from .categories import CATEGORIES
from .env_file import get_default_env_path, load_env_file, save_env_file
from .introspection import get_complete_settings
from .validators import validate_setting
from .widgets import (
prompt_boolean,
prompt_float,
prompt_numeric,
prompt_secret_input,
prompt_selection,
prompt_text_input,
)
def _getch() -> str:
"""Read a single character from stdin without echo."""
fd = sys.stdin.fileno()
old_settings = termios.tcgetattr(fd)
try:
tty.setraw(fd)
ch = sys.stdin.read(1)
# Handle escape sequences (arrow keys, etc.)
if ch == "\x1b":
ch2 = sys.stdin.read(1)
if ch2 == "[":
ch3 = sys.stdin.read(1)
# Handle Shift+Tab (reverse tab)
if ch3 == "Z":
return "shift_tab"
return f"\x1b[{ch3}"
return ch
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
class SettingsUI:
"""Interactive tabbed settings browser using Rich."""
def __init__(self):
self.console = Console()
self.all_settings = get_complete_settings()
self.categories = [
cat for cat in CATEGORIES if cat.get_settings(self.all_settings)
]
self.current_tab = 0
self.selected_index = 0
self.values: dict[str, str] = {}
self.original_values: dict[str, str] = {}
self.env_path: Path = get_default_env_path()
self.has_unsaved_changes = False
def run(self, env_path: Path | None = None) -> None:
"""Run the interactive settings browser.
Args:
env_path: Optional path to .env file. Uses default if not specified.
"""
if env_path:
self.env_path = env_path
# Load existing values
self.values = load_env_file(self.env_path)
self.original_values = self.values.copy()
# Main loop
try:
while True:
self._render()
if not self._handle_input():
break
except KeyboardInterrupt:
self._cleanup()
self.console.print("\n[yellow]Cancelled[/yellow]")
def _render(self) -> None:
"""Render the UI."""
# Clear screen
self.console.clear()
# Header with file path
header = Text()
header.append("AutoGPT Config", style="bold cyan")
header.append(f" ({self.env_path})", style="dim")
if self.has_unsaved_changes:
header.append(" *", style="bold yellow")
self.console.print()
self.console.print(Panel(header, border_style="cyan", padding=(0, 1)))
self.console.print()
# Tab bar
self._render_tabs()
self.console.print()
# Current category settings
self._render_settings()
self.console.print()
# Help text
self._render_help()
def _render_tabs(self) -> None:
"""Render the tab bar."""
tabs = Text()
tabs.append(" ")
for i, cat in enumerate(self.categories):
is_active = i == self.current_tab
label = f"{i + 1} {cat.name}"
if is_active:
tabs.append("[", style="bold cyan")
tabs.append(label, style="bold cyan")
tabs.append("]", style="bold cyan")
else:
tabs.append(f" {label} ", style="dim")
tabs.append(" ")
self.console.print(tabs)
# Underline for active tab
underline = Text()
underline.append(" ")
for i, cat in enumerate(self.categories):
label = f"{i + 1} {cat.name}"
if i == self.current_tab:
underline.append("" * (len(label) + 2), style="bold cyan")
else:
underline.append(" " * (len(label) + 2), style="dim")
underline.append(" ")
self.console.print(underline)
def _render_settings(self) -> None:
"""Render settings for the current category."""
if not self.categories:
self.console.print(" [dim]No settings available[/dim]")
return
category = self.categories[self.current_tab]
settings = category.get_settings(self.all_settings)
if not settings:
self.console.print(f" [dim]No settings in {category.name}[/dim]")
return
for i, setting in enumerate(settings):
is_selected = i == self.selected_index
value = self.values.get(setting.env_var, "")
display_value = setting.get_display_value(value or None)
# Determine if value has changed from original
changed = value != self.original_values.get(setting.env_var, "")
line = Text()
if is_selected:
line.append(" ", style="bold green")
line.append(setting.env_var, style="bold green")
else:
line.append(" ", style="dim")
line.append(setting.env_var, style="dim")
# Pad to align values
padding = 30 - len(setting.env_var)
line.append(" " * max(padding, 1))
# Value
if display_value == "[not set]":
line.append(display_value, style="dim italic")
elif changed:
line.append(display_value, style="yellow")
else:
line.append(display_value, style="white")
self.console.print(line)
def _render_help(self) -> None:
"""Render help text at the bottom."""
help_text = Text()
help_text.append(" ")
help_text.append("←→", style="bold cyan")
help_text.append("/", style="dim")
help_text.append("Tab", style="bold cyan")
help_text.append("/", style="dim")
help_text.append("1-9", style="bold cyan")
help_text.append(" category ", style="dim")
help_text.append("↑↓", style="bold cyan")
help_text.append(" navigate ", style="dim")
help_text.append("Enter", style="bold cyan")
help_text.append(" edit ", style="dim")
help_text.append("S", style="bold cyan")
help_text.append(" save ", style="dim")
help_text.append("Q", style="bold cyan")
help_text.append(" quit", style="dim")
self.console.print(help_text)
def _handle_input(self) -> bool:
"""Handle keyboard input.
Returns:
True to continue, False to exit
"""
ch = _getch()
# Tab - next category
if ch == "\t":
self.current_tab = (self.current_tab + 1) % len(self.categories)
self.selected_index = 0
return True
# Shift+Tab - previous category
if ch == "shift_tab":
self.current_tab = (self.current_tab - 1) % len(self.categories)
self.selected_index = 0
return True
# Number keys 1-9 - jump to category
if ch in "123456789":
idx = int(ch) - 1
if idx < len(self.categories):
self.current_tab = idx
self.selected_index = 0
return True
# Arrow up
if ch == "\x1b[A":
category = self.categories[self.current_tab]
settings = category.get_settings(self.all_settings)
if settings:
self.selected_index = (self.selected_index - 1) % len(settings)
return True
# Arrow down
if ch == "\x1b[B":
category = self.categories[self.current_tab]
settings = category.get_settings(self.all_settings)
if settings:
self.selected_index = (self.selected_index + 1) % len(settings)
return True
# Arrow left - previous category
if ch == "\x1b[D":
self.current_tab = (self.current_tab - 1) % len(self.categories)
self.selected_index = 0
return True
# Arrow right - next category
if ch == "\x1b[C":
self.current_tab = (self.current_tab + 1) % len(self.categories)
self.selected_index = 0
return True
# Enter - edit selected setting
if ch in ("\r", "\n"):
self._edit_current_setting()
return True
# S - save
if ch in ("s", "S"):
self._save_settings()
return True
# Q - quit
if ch in ("q", "Q"):
if self.has_unsaved_changes:
return self._confirm_quit()
return False
# Ctrl+C
if ch == "\x03":
raise KeyboardInterrupt()
return True
def _edit_current_setting(self) -> None:
"""Edit the currently selected setting."""
if not self.categories:
return
category = self.categories[self.current_tab]
settings = category.get_settings(self.all_settings)
if not settings or self.selected_index >= len(settings):
return
setting = settings[self.selected_index]
current_value = self.values.get(setting.env_var, "")
# Clear screen for edit mode
self.console.clear()
self.console.print()
new_value: Any = None
if setting.field_type == "secret":
masked = setting.get_display_value(current_value or None)
new_value = prompt_secret_input(
self.console,
label=setting.env_var,
description=setting.description,
current_masked=masked if masked != "[not set]" else "",
env_var=setting.env_var,
)
# Keep current value if empty input
if not new_value and current_value:
return
elif setting.field_type == "choice":
default_idx = 0
if current_value and current_value in setting.choices:
default_idx = setting.choices.index(current_value)
new_value = prompt_selection(
label=setting.env_var,
choices=setting.choices,
description=setting.description,
default_index=default_idx,
env_var=setting.env_var,
)
elif setting.field_type == "bool":
current_bool = (
current_value.lower() in ("true", "1", "yes")
if current_value
else False
)
result = prompt_boolean(
self.console,
label=setting.env_var,
description=setting.description,
default=current_bool,
env_var=setting.env_var,
)
new_value = "true" if result else "false"
elif setting.field_type == "int":
current_int = (
int(current_value)
if current_value and current_value.isdigit()
else None
)
result = prompt_numeric(
self.console,
label=setting.env_var,
description=setting.description,
default=current_int,
env_var=setting.env_var,
)
new_value = str(result) if result is not None else ""
elif setting.field_type == "float":
try:
current_float = float(current_value) if current_value else None
except ValueError:
current_float = None
result = prompt_float(
self.console,
label=setting.env_var,
description=setting.description,
default=current_float,
env_var=setting.env_var,
)
new_value = str(result) if result is not None else ""
else: # str
new_value = prompt_text_input(
self.console,
label=setting.env_var,
description=setting.description,
default=current_value,
env_var=setting.env_var,
)
# Validate the new value
if new_value:
is_valid, error = validate_setting(setting.env_var, new_value)
if not is_valid:
self.console.print(f"\n[red]Validation error: {error}[/red]")
self.console.print("[dim]Press any key to continue...[/dim]")
_getch()
return
elif error: # Warning
self.console.print(f"\n[yellow]{error}[/yellow]")
# Update value
if new_value != current_value:
self.values[setting.env_var] = new_value
self.has_unsaved_changes = True
def _save_settings(self) -> None:
"""Save settings to .env file."""
try:
save_env_file(self.env_path, self.values, CATEGORIES)
self.original_values = self.values.copy()
self.has_unsaved_changes = False
self.console.clear()
self.console.print()
self.console.print(
Panel(
f"[green]Settings saved to {self.env_path}[/green]",
border_style="green",
)
)
self.console.print("\n[dim]Press any key to continue...[/dim]")
_getch()
except Exception as e:
self.console.print(f"\n[red]Error saving settings: {e}[/red]")
self.console.print("[dim]Press any key to continue...[/dim]")
_getch()
def _confirm_quit(self) -> bool:
"""Confirm quitting with unsaved changes.
Returns:
True to continue (not quit), False to quit
"""
self.console.clear()
self.console.print()
self.console.print(
Panel(
"[yellow]You have unsaved changes![/yellow]\n\n"
"Press [bold]S[/bold] to save, [bold]Q[/bold] to quit without saving, "
"or any other key to cancel",
border_style="yellow",
)
)
ch = _getch()
if ch in ("s", "S"):
self._save_settings()
return False
elif ch in ("q", "Q"):
return False
return True
def _cleanup(self) -> None:
"""Clean up terminal state."""
# Terminal should be restored by _getch's finally block
pass

View File

@@ -0,0 +1,220 @@
"""Input validation functions for settings."""
from __future__ import annotations
import re
from typing import Any, Callable
def validate_api_key_format(key_name: str, value: str) -> tuple[bool, str]:
"""Validate API key format based on known patterns.
Args:
key_name: The name of the key (e.g., "OPENAI_API_KEY")
value: The key value to validate
Returns:
Tuple of (is_valid, error_message)
"""
if not value:
return True, "" # Empty is allowed (will be skipped)
patterns = {
"OPENAI_API_KEY": (
r"^sk-[a-zA-Z0-9-_]{20,}$",
"OpenAI keys should start with 'sk-' followed by alphanumeric characters",
),
"ANTHROPIC_API_KEY": (
r"^sk-ant-api03-[a-zA-Z0-9-_]{80,}$",
"Anthropic keys should start with 'sk-ant-api03-'",
),
"GROQ_API_KEY": (
r"^gsk_[a-zA-Z0-9]{48,}$",
"Groq keys should start with 'gsk_'",
),
"TAVILY_API_KEY": (
r"^tvly-[a-zA-Z0-9-_]{20,}$",
"Tavily keys should start with 'tvly-'",
),
"GITHUB_API_KEY": (
r"^(ghp_[a-zA-Z0-9]{36}|github_pat_[a-zA-Z0-9_]{80,})$",
"GitHub tokens should start with 'ghp_' or 'github_pat_'",
),
}
if key_name not in patterns:
return True, "" # No pattern to validate against
pattern, error_msg = patterns[key_name]
if re.match(pattern, value):
return True, ""
return False, error_msg
def validate_model_name(model_name: str) -> tuple[bool, str]:
"""Validate that a model name is in a known format.
Args:
model_name: The model name to validate
Returns:
Tuple of (is_valid, error_message)
"""
if not model_name:
return True, ""
# Known model prefixes
valid_prefixes = [
"gpt-3.5",
"gpt-4",
"gpt-5",
"o1",
"o3",
"o4",
"claude-",
"mixtral",
"gemma",
"llama",
]
model_lower = model_name.lower()
for prefix in valid_prefixes:
if model_lower.startswith(prefix):
return True, ""
# Also allow full model names from enums
# Just warn, don't block
return True, f"Note: '{model_name}' is not a recognized model name"
def validate_port(port: int | str) -> tuple[bool, str]:
"""Validate a port number.
Args:
port: The port number to validate
Returns:
Tuple of (is_valid, error_message)
"""
try:
port_num = int(port)
except (ValueError, TypeError):
return False, "Port must be a number"
if port_num < 1 or port_num > 65535:
return False, "Port must be between 1 and 65535"
if port_num < 1024:
return True, "Note: Ports below 1024 typically require root privileges"
return True, ""
def validate_url(url: str) -> tuple[bool, str]:
"""Validate a URL format.
Args:
url: The URL to validate
Returns:
Tuple of (is_valid, error_message)
"""
if not url:
return True, ""
# Basic URL pattern
pattern = r"^https?://[a-zA-Z0-9.-]+(:[0-9]+)?(/.*)?$"
if re.match(pattern, url):
return True, ""
return False, "Invalid URL format (should start with http:// or https://)"
def validate_log_level(level: str) -> tuple[bool, str]:
"""Validate a log level.
Args:
level: The log level to validate
Returns:
Tuple of (is_valid, error_message)
"""
valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
if level.upper() in valid_levels:
return True, ""
return False, f"Log level must be one of: {', '.join(valid_levels)}"
def validate_storage_backend(backend: str) -> tuple[bool, str]:
"""Validate a storage backend.
Args:
backend: The storage backend to validate
Returns:
Tuple of (is_valid, error_message)
"""
valid_backends = ["local", "gcs", "s3"]
if backend.lower() in valid_backends:
return True, ""
return False, f"Storage backend must be one of: {', '.join(valid_backends)}"
def validate_temperature(temp: float | str) -> tuple[bool, str]:
"""Validate a temperature value.
Args:
temp: The temperature to validate
Returns:
Tuple of (is_valid, error_message)
"""
try:
temp_num = float(temp)
except (ValueError, TypeError):
return False, "Temperature must be a number"
if temp_num < 0 or temp_num > 2:
return False, "Temperature should be between 0 and 2"
return True, ""
# Mapping of env var names to validator functions
VALIDATORS: dict[str, Callable[[Any], tuple[bool, str]]] = {
"OPENAI_API_KEY": lambda v: validate_api_key_format("OPENAI_API_KEY", v),
"ANTHROPIC_API_KEY": lambda v: validate_api_key_format("ANTHROPIC_API_KEY", v),
"GROQ_API_KEY": lambda v: validate_api_key_format("GROQ_API_KEY", v),
"TAVILY_API_KEY": lambda v: validate_api_key_format("TAVILY_API_KEY", v),
"GITHUB_API_KEY": lambda v: validate_api_key_format("GITHUB_API_KEY", v),
"SMART_LLM": validate_model_name,
"FAST_LLM": validate_model_name,
"AP_SERVER_PORT": validate_port,
"OPENAI_API_BASE_URL": validate_url,
"ANTHROPIC_API_BASE_URL": validate_url,
"GROQ_API_BASE_URL": validate_url,
"S3_ENDPOINT_URL": validate_url,
"LOG_LEVEL": validate_log_level,
"FILE_STORAGE_BACKEND": validate_storage_backend,
"TEMPERATURE": validate_temperature,
}
def validate_setting(env_var: str, value: str) -> tuple[bool, str]:
"""Validate a setting value.
Args:
env_var: The environment variable name
value: The value to validate
Returns:
Tuple of (is_valid, error_message)
"""
if env_var in VALIDATORS:
return VALIDATORS[env_var](value)
return True, ""

View File

@@ -0,0 +1,292 @@
"""Reusable Rich input widgets for settings UI."""
from __future__ import annotations
from autogpt.app.ui.rich_select import RichSelect
from rich.console import Console
from rich.panel import Panel
from rich.prompt import Confirm, Prompt
from rich.text import Text
def prompt_text_input(
console: Console,
label: str,
description: str = "",
default: str = "",
env_var: str = "",
) -> str:
"""Prompt for text input with Rich styling.
Args:
console: Rich Console instance
label: Setting name/label
description: Help text for the setting
default: Default value
env_var: Environment variable name
Returns:
User input string
"""
header = Text()
header.append(f"{label}\n", style="bold cyan")
if description:
header.append(f"{description}\n", style="dim")
if env_var:
header.append(f"ENV: {env_var}", style="dim italic")
console.print()
console.print(Panel(header, border_style="cyan", padding=(0, 1)))
prompt_text = "Value"
if default:
prompt_text += f" [{default}]"
result = Prompt.ask(f" {prompt_text}", default=default, console=console)
return result
def prompt_secret_input(
console: Console,
label: str,
description: str = "",
current_masked: str = "",
env_var: str = "",
) -> str:
"""Prompt for secret/password input with Rich styling.
Args:
console: Rich Console instance
label: Setting name/label
description: Help text for the setting
current_masked: Current masked value for display
env_var: Environment variable name
Returns:
User input string (unmasked)
"""
header = Text()
header.append(f"{label}\n", style="bold cyan")
if description:
header.append(f"{description}\n", style="dim")
if env_var:
header.append(f"ENV: {env_var}\n", style="dim italic")
if current_masked:
header.append(f"Current: {current_masked}", style="yellow")
console.print()
console.print(Panel(header, border_style="cyan", padding=(0, 1)))
console.print(" [dim](Leave empty to keep current value)[/dim]")
result = Prompt.ask(" Value", password=True, console=console)
return result
def prompt_selection(
label: str,
choices: list[str],
description: str = "",
default_index: int = 0,
env_var: str = "",
) -> str:
"""Prompt for selection from choices using RichSelect.
Args:
label: Setting name/label
choices: List of choices
description: Help text for the setting
default_index: Index of default choice
env_var: Environment variable name
Returns:
Selected choice string
"""
subtitle = description
if env_var:
subtitle = (
f"{description}\nENV: {env_var}" if description else f"ENV: {env_var}"
)
selector = RichSelect(
choices=choices,
title=label,
subtitle=subtitle,
default_index=default_index,
show_feedback_option=False,
)
result = selector.run()
return result.choice
def prompt_boolean(
console: Console,
label: str,
description: str = "",
default: bool = False,
env_var: str = "",
) -> bool:
"""Prompt for boolean input with Rich styling.
Args:
console: Rich Console instance
label: Setting name/label
description: Help text for the setting
default: Default value
env_var: Environment variable name
Returns:
Boolean value
"""
header = Text()
header.append(f"{label}\n", style="bold cyan")
if description:
header.append(f"{description}\n", style="dim")
if env_var:
header.append(f"ENV: {env_var}", style="dim italic")
console.print()
console.print(Panel(header, border_style="cyan", padding=(0, 1)))
return Confirm.ask(" Enable", default=default, console=console)
def prompt_numeric(
console: Console,
label: str,
description: str = "",
default: int | None = None,
env_var: str = "",
min_value: int | None = None,
max_value: int | None = None,
) -> int | None:
"""Prompt for numeric input with Rich styling.
Args:
console: Rich Console instance
label: Setting name/label
description: Help text for the setting
default: Default value
env_var: Environment variable name
min_value: Minimum allowed value
max_value: Maximum allowed value
Returns:
Integer value or None if empty
"""
header = Text()
header.append(f"{label}\n", style="bold cyan")
if description:
header.append(f"{description}\n", style="dim")
if env_var:
header.append(f"ENV: {env_var}\n", style="dim italic")
constraints = []
if min_value is not None:
constraints.append(f"min: {min_value}")
if max_value is not None:
constraints.append(f"max: {max_value}")
if constraints:
header.append(f"({', '.join(constraints)})", style="dim")
console.print()
console.print(Panel(header, border_style="cyan", padding=(0, 1)))
prompt_text = "Value"
if default is not None:
prompt_text += f" [{default}]"
else:
prompt_text += " [empty to skip]"
while True:
result = Prompt.ask(f" {prompt_text}", console=console)
# Handle empty input
if not result.strip():
return default
# Try to parse as integer
try:
value = int(result)
# Validate range
if min_value is not None and value < min_value:
console.print(f" [red]Value must be at least {min_value}[/red]")
continue
if max_value is not None and value > max_value:
console.print(f" [red]Value must be at most {max_value}[/red]")
continue
return value
except ValueError:
console.print(" [red]Please enter a valid number[/red]")
def prompt_float(
console: Console,
label: str,
description: str = "",
default: float | None = None,
env_var: str = "",
min_value: float | None = None,
max_value: float | None = None,
) -> float | None:
"""Prompt for float input with Rich styling.
Args:
console: Rich Console instance
label: Setting name/label
description: Help text for the setting
default: Default value
env_var: Environment variable name
min_value: Minimum allowed value
max_value: Maximum allowed value
Returns:
Float value or None if empty
"""
header = Text()
header.append(f"{label}\n", style="bold cyan")
if description:
header.append(f"{description}\n", style="dim")
if env_var:
header.append(f"ENV: {env_var}\n", style="dim italic")
constraints = []
if min_value is not None:
constraints.append(f"min: {min_value}")
if max_value is not None:
constraints.append(f"max: {max_value}")
if constraints:
header.append(f"({', '.join(constraints)})", style="dim")
console.print()
console.print(Panel(header, border_style="cyan", padding=(0, 1)))
prompt_text = "Value"
if default is not None:
prompt_text += f" [{default}]"
else:
prompt_text += " [empty to skip]"
while True:
result = Prompt.ask(f" {prompt_text}", console=console)
# Handle empty input
if not result.strip():
return default
# Try to parse as float
try:
value = float(result)
# Validate range
if min_value is not None and value < min_value:
console.print(f" [red]Value must be at least {min_value}[/red]")
continue
if max_value is not None and value > max_value:
console.print(f" [red]Value must be at most {max_value}[/red]")
continue
return value
except ValueError:
console.print(" [red]Please enter a valid number[/red]")

View File

@@ -86,6 +86,7 @@ def test_strategy_comparison_quick():
Note: Requires API keys to be configured in environment.
"""
result = run_harness(
"--fresh", # Don't resume from previous runs
"--strategies",
"one_shot",
"--categories",
@@ -116,6 +117,7 @@ def test_single_strategy():
to verify basic functionality without testing all strategies.
"""
result = run_harness(
"--fresh", # Don't resume from previous runs
"--strategies",
"one_shot",
"--categories",

View File

@@ -179,6 +179,7 @@ exclude = [
"**/__pycache__",
"**/.*",
"direct_benchmark/challenges/**", # Legacy code with unavailable imports
"direct_benchmark/direct_benchmark/adapters/**", # Optional deps (datasets, swebench, modal)
]