feat(config): add actionable startup diagnostics and WebUI recovery (#5110)

This commit is contained in:
chengyongru 2026-07-28 18:52:05 +08:00 committed by GitHub
parent 76ab04ac48
commit 0c6c0438d4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
19 changed files with 1425 additions and 64 deletions

View File

@ -11,7 +11,7 @@ Use this page when you know what you want to run and need the command shape. For
| Refresh config non-interactively | `nanobot onboard --refresh` | Preserves existing values and adds missing default fields without prompting |
| Use guided setup | `nanobot onboard --wizard` | Best when you prefer prompts over hand-editing JSON |
| Open the browser workbench | `nanobot webui` | Prepares local WebUI settings, starts the gateway, and opens the browser |
| Check config without calling a model | `nanobot status` | Summarizes the selected config, workspace, active model, and providers |
| Check readiness without calling a model | `nanobot status` | Summarizes config/workspace and validates the active provider/model configuration |
| Send one test message | `nanobot agent -m "Hello!"` | First proof that install, config, provider, model, and workspace all work |
| Chat in the terminal | `nanobot agent` | Interactive local chat; exit with `exit`, `/exit`, `:q`, or `Ctrl+D` |
| Run the gateway directly | `nanobot gateway` | Service/ops command for WebUI, chat apps, cron, and heartbeat |
@ -70,6 +70,18 @@ Default paths:
| Config | `~/.nanobot/config.json` |
| Workspace | `~/.nanobot/workspace/` |
## Status
| Command | Description |
|---|---|
| `nanobot status` | Summarize the default config/workspace and check Agent provider/model readiness |
| `nanobot status --config <path>` | Check a specific config file |
| `nanobot status --workspace <path>` | Show status with a workspace override |
Status does not send a model request. On success, run the printed
`nanobot agent -m "Hello!"` command to verify network access and credentials. On failure,
follow the printed WebUI **Settings → Models** or `nanobot onboard --wizard` route.
## Agent CLI
| Command | Description |

View File

@ -90,7 +90,9 @@ Instead of storing secrets directly in `config.json`, you can use `${VAR_NAME}`
Any string value in `config.json` can use `${VAR_NAME}`. Resolution runs once at startup, in memory only — resolved values are never written back to disk, so editing config through `nanobot onboard` or the WebUI preserves the placeholder.
If a referenced variable is unset, nanobot fails fast at startup with `ValueError: Environment variable 'NAME' referenced in config is not set`.
If a referenced variable is unset, nanobot fails fast and reports the exact config field
and variable name without echoing the field value. Run `nanobot status` with the same
`--config` path to inspect the problem.
### More examples

View File

@ -23,15 +23,20 @@ This separates failures into layers:
| Layer | What it proves |
|---|---|
| `nanobot --version` | Install and shell command discovery |
| `nanobot status` | Config path, workspace path, active model, and provider summary |
| `nanobot status` | Config path, workspace, environment references, and active provider/model configuration |
| `nanobot agent -m "Hello!"` | Config loading, provider/model access, workspace writes, and agent loop |
| `nanobot gateway` | Channel startup, cron system jobs, heartbeat, WebUI/WebSocket, and health endpoint |
If `nanobot agent -m "Hello!"` fails, fix that before debugging WebUI, Telegram, Discord, Docker, systemd, or any chat app.
`nanobot status` does not call the model. If provider/model setup is incomplete, it points to
WebUI **Settings → Models** or the CLI setup wizard, then prints the command to check again.
## How to Read `nanobot status`
`nanobot status` does not call a model. It only checks whether nanobot can find the selected config, selected workspace, active model or preset, and provider setup summary.
`nanobot status` does not call a model. It checks the selected config and workspace,
resolves environment references, and validates the local settings required by the active
provider/model without constructing a provider client.
The output has this shape:
@ -41,6 +46,7 @@ nanobot Status
Config: /path/to/config.json ✓
Workspace: /path/to/workspace ✓
Model: provider/model-name (preset: primary)
Agent: ✓ provider/model configuration is ready
Provider A: not set
Provider B: ✓
Local Provider: ✓ http://localhost:11434/v1
@ -54,6 +60,7 @@ Read it like this:
| `Config` | It points to the config file you meant to use and shows `✓`. | Run `nanobot onboard`, or pass `--config` to `nanobot agent`, `gateway`, or `serve` when testing a non-default instance. |
| `Workspace` | It points to the workspace you meant to use and shows `✓`. | Run `nanobot onboard`, create the folder, fix permissions, or pass `--workspace` on commands that support it. |
| `Model` | It shows the active model or the preset name you expect. | Set `agents.defaults.modelPreset` to the intended preset, or check `/model` if you changed models during a chat session. |
| `Agent` | It says `provider/model configuration is ready`. | Follow the printed WebUI or CLI setup route, then run `nanobot status` again. |
| Provider rows | The provider used by the active preset shows `✓`, an OAuth marker, or a local URL. | Configure only the active provider first. It is normal for unused providers to say `not set`. |
If `nanobot status` looks right but `nanobot agent -m "Hello!"` fails, the install and config paths are probably fine. Continue with [Provider and Model Problems](#provider-and-model-problems).
@ -108,6 +115,12 @@ Common config mistakes:
| Environment variable error | `${VAR_NAME}` references are resolved at startup. Set the variable before running nanobot. |
| Edited config but behavior did not change | Restart `nanobot gateway`; long-running processes read config at startup. |
After editing config, check the shortest path to an Agent reply:
```bash
nanobot status
```
To refresh missing defaults without overwriting existing settings, run:
```bash

View File

@ -33,7 +33,10 @@ def load_model_preset_catalog(
from nanobot.config.loader import load_config, resolve_config_env_vars
return configured_model_presets(
resolve_config_env_vars(load_config(config_path)),
resolve_config_env_vars(
load_config(config_path),
config_path=config_path,
),
)

View File

@ -54,6 +54,7 @@ from prompt_toolkit.history import FileHistory # noqa: E402
from prompt_toolkit.key_binding import KeyBindings # noqa: E402
from prompt_toolkit.keys import Keys # noqa: E402
from prompt_toolkit.patch_stdout import patch_stdout # noqa: E402
from pydantic import ValidationError # noqa: E402
from rich.console import Console # noqa: E402
from rich.markdown import Markdown # noqa: E402
from rich.markup import escape # noqa: E402
@ -797,9 +798,89 @@ def _model_display(config: Config) -> tuple[str, str]:
return resolved.model, tag
def _print_config_error(error: Exception) -> None:
"""Render a configuration failure without exposing traceback internals."""
from nanobot.config.errors import ConfigLoadError
console.print(Text(str(error), style="red"))
if isinstance(error, ConfigLoadError):
command = _status_command(error.path)
console.print(f"[dim]Check again after editing: {escape(command)}[/dim]")
def _print_runtime_config_validation_error(
error: ValidationError,
*,
config_path: Path,
summary: str,
path_prefix: tuple[str | int, ...],
retry_command: str,
) -> None:
"""Render a runtime-owned Pydantic config error without exposing input values."""
from nanobot.config.errors import ConfigIssue, ConfigLoadError, validation_issues
issues = tuple(
ConfigIssue(
path=(*path_prefix, *issue.path),
message=issue.message,
)
for issue in validation_issues(error)
)
diagnostic = ConfigLoadError(
config_path,
kind="invalid_schema",
summary=summary,
issues=issues,
)
console.print(Text(str(diagnostic), style="red"))
console.print(f"[dim]Fix the listed setting, then retry: {escape(retry_command)}[/dim]")
def _status_command(config_path: Path) -> str:
return f'nanobot status --config "{config_path}"'
def _print_model_setup_steps(config_path: Path) -> None:
"""Show the shortest setup routes shared by Status and Agent startup."""
config_arg = f'--config "{config_path}"'
console.print(
f" WebUI: run [cyan]nanobot webui {escape(config_arg)}[/cyan], "
"then open Settings → Models"
)
console.print(f" CLI: run [cyan]nanobot onboard --wizard {escape(config_arg)}[/cyan]")
console.print(f" Check: [cyan]{escape(_status_command(config_path))}[/cyan]")
def _print_agent_start_error(error: ValueError) -> None:
from nanobot.config.loader import get_config_path
console.print(Text(f"Agent cannot start: {error}", style="red"))
console.print("Complete provider/model setup:")
_print_model_setup_steps(get_config_path())
def _load_config_for_cli(
config_path: Path | None = None,
*,
resolve_env: bool = False,
) -> Config:
"""Load CLI configuration and turn expected failures into a clean exit."""
from nanobot.config.errors import ConfigLoadError
from nanobot.config.loader import load_config, resolve_config_env_vars
try:
loaded = load_config(config_path)
if resolve_env:
loaded = resolve_config_env_vars(loaded)
return loaded
except ConfigLoadError as exc:
_print_config_error(exc)
raise typer.Exit(1) from exc
def _load_runtime_config(config: str | None = None, workspace: str | None = None) -> Config:
"""Load config and optionally override the active workspace."""
from nanobot.config.loader import load_config, resolve_config_env_vars, set_config_path
from nanobot.config.loader import set_config_path
config_path = None
if config:
@ -810,11 +891,7 @@ def _load_runtime_config(config: str | None = None, workspace: str | None = None
set_config_path(config_path)
console.print(f"[dim]Using config: {config_path}[/dim]")
try:
loaded = resolve_config_env_vars(load_config(config_path))
except ValueError as e:
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(1)
loaded = _load_config_for_cli(config_path, resolve_env=True)
if workspace:
loaded.agents.defaults.workspace = workspace
return loaded
@ -840,6 +917,7 @@ def _load_inspection_config(
workspace: str | None = None,
) -> tuple[Path, Config]:
"""Load config for diagnostic commands without resolving secret env refs."""
from nanobot.config.errors import ConfigLoadError
from nanobot.config.loader import get_config_path, load_config, set_config_path
config_path = None
@ -851,6 +929,9 @@ def _load_inspection_config(
display_path = config_path or get_config_path()
try:
loaded = load_config(config_path)
except ConfigLoadError as exc:
_print_config_error(exc)
raise typer.Exit(1) from exc
except ValueError as exc:
console.print(f"[red]Error: {exc}[/red]")
raise typer.Exit(1) from exc
@ -901,21 +982,15 @@ def _resolve_webui_config_path(config: str | None) -> Path:
def _load_webui_setup_config(config_path: Path) -> Config:
"""Load config for first-run mutation without resolving env-var placeholders."""
from nanobot.config.loader import load_config
try:
return load_config(config_path)
except ValueError as e:
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(1) from e
return _load_config_for_cli(config_path)
def _provider_setup_error(config: Config) -> str | None:
"""Return the provider setup error, or None when the current model can start."""
from nanobot.providers.factory import build_provider_snapshot
"""Return a local provider/model configuration error, or None."""
from nanobot.providers.factory import validate_provider_setup
try:
build_provider_snapshot(config)
validate_provider_setup(config)
except ValueError as exc:
return str(exc)
return None
@ -937,6 +1012,60 @@ def _webui_channel_enabled(config: Config) -> bool:
return bool(WebSocketConfig.model_validate(current).enabled)
def _validate_gateway_startup(config: Config) -> str | None:
"""Validate gateway startup and return a provider error recoverable through WebUI."""
from nanobot.config.loader import get_config_path
config_path = get_config_path()
try:
webui_config = _webui_config_dict(config)
except ValidationError as exc:
retry_command = f'nanobot gateway --config "{config_path}"'
_print_runtime_config_validation_error(
exc,
config_path=config_path,
summary="Gateway configuration is invalid.",
path_prefix=("channels", "websocket"),
retry_command=retry_command,
)
raise typer.Exit(1) from exc
provider_error = _provider_setup_error(config)
if not provider_error:
return None
if bool(webui_config["enabled"]):
console.print(
Text(f"Provider/model setup is incomplete: {provider_error}", style="yellow")
)
console.print(
"Gateway will start so you can configure a provider and model "
"in WebUI Settings → Models."
)
browser_url = _webui_browser_url(config)
webui_url = browser_url.split("/#/", 1)[0]
console.print(Text(f"WebUI: {webui_url}", style="cyan"))
if browser_url != webui_url:
secret_key = (
"tokenIssueSecret"
if str(webui_config.get("tokenIssueSecret") or "").strip()
else "token"
)
console.print(
Text(
f"If prompted, enter the configured channels.websocket.{secret_key} "
f"value (see {config_path}).",
style="dim",
)
)
return provider_error
console.print(Text(f"Gateway cannot start: {provider_error}", style="red"))
console.print("Complete provider/model setup:")
_print_model_setup_steps(config_path)
raise typer.Exit(1)
def _prepare_webui_bundle_for_gateway(
config: Config,
*,
@ -1234,14 +1363,20 @@ def _gateway_instance_command(
return " ".join(shlex.quote(part) for part in parts)
def _run_quick_start_for_webui(config: Config, *, yes: bool) -> Config:
def _run_quick_start_for_webui(
config: Config,
*,
yes: bool,
config_path: Path,
) -> Config:
"""Offer the existing Quick Start flow when provider setup is missing."""
if yes:
console.print(
"[red]Error: provider/model setup is incomplete, and --yes cannot answer "
"provider credentials. Run `nanobot webui` interactively or "
"`nanobot onboard --wizard`.[/red]"
"provider credentials.[/red]"
)
console.print("Complete provider/model setup:")
_print_model_setup_steps(config_path)
raise typer.Exit(1)
console.print()
@ -1433,9 +1568,12 @@ def webui(
setup_config.agents.defaults.workspace = workspace
try:
resolved_setup_config = resolve_config_env_vars(setup_config.model_copy(deep=True))
resolved_setup_config = resolve_config_env_vars(
setup_config.model_copy(deep=True),
config_path=config_path,
)
except ValueError as exc:
console.print(f"[red]Error: {exc}[/red]")
_print_config_error(exc)
raise typer.Exit(1) from exc
provider_error = _provider_setup_error(resolved_setup_config)
@ -1451,7 +1589,11 @@ def webui(
raise typer.Exit(1)
elif provider_error:
console.print(f"[dim]Provider check: {provider_error}[/dim]")
setup_config = _run_quick_start_for_webui(setup_config, yes=yes)
setup_config = _run_quick_start_for_webui(
setup_config,
yes=yes,
config_path=config_path,
)
if workspace:
setup_config.agents.defaults.workspace = workspace
@ -1463,6 +1605,16 @@ def webui(
)
_warn_webui_bind_scope(setup_config)
webui_url = _webui_browser_url(setup_config)
except ValidationError as exc:
retry_command = f'nanobot webui --config "{config_path}"'
_print_runtime_config_validation_error(
exc,
config_path=config_path,
summary="WebUI configuration is invalid.",
path_prefix=("channels", "websocket"),
retry_command=retry_command,
)
raise typer.Exit(1) from exc
except ValueError as exc:
console.print(f"[red]Error: invalid WebUI channel config: {exc}[/red]")
raise typer.Exit(1) from exc
@ -2228,6 +2380,7 @@ app.add_typer(
log_handler_id=_log_handler_id,
load_runtime_config=_load_runtime_config,
run_gateway=_run_gateway,
validate_startup_config=_validate_gateway_startup,
prepare_webui_bundle=lambda config, mode: _prepare_webui_bundle_for_gateway(
config,
mode=mode,
@ -2254,9 +2407,16 @@ def agent(
"""Interact with the agent directly."""
from nanobot.bus.queue import MessageBus
from nanobot.cron.service import CronService
from nanobot.providers.factory import make_provider
from nanobot.providers.image_generation import image_gen_provider_configs
config = _load_runtime_config(config, workspace)
try:
provider = make_provider(config)
except ValueError as exc:
_print_agent_start_error(exc)
raise typer.Exit(1) from exc
sync_workspace_templates(config.workspace_path)
bus = MessageBus()
@ -2274,12 +2434,13 @@ def agent(
try:
agent_loop = AgentLoop.from_config(
config, bus,
provider=provider,
cron_service=cron,
image_generation_provider_configs=image_gen_provider_configs(config),
hook_factories=[create_file_edit_activity_hook],
)
except ValueError as exc:
console.print(f"[red]Error: {exc}[/red]")
_print_agent_start_error(exc)
raise typer.Exit(1) from exc
restart_notice = consume_restart_notice_from_env()
if restart_notice and should_show_cli_restart_notice(restart_notice, session_id):
@ -2683,11 +2844,32 @@ def status(
)
if config_path.exists():
from nanobot.config.errors import ConfigLoadError
from nanobot.config.loader import resolve_config_env_vars, resolve_env_refs
from nanobot.providers.registry import PROVIDERS
_model, _preset_tag = _model_display(loaded)
console.print(f"Model: {_model}{_preset_tag}")
provider_ready = False
try:
resolved = resolve_config_env_vars(
loaded.model_copy(deep=True),
config_path=config_path,
)
except ConfigLoadError as exc:
console.print("Agent: [red]✗ configuration is not ready[/red]")
_print_config_error(exc)
else:
provider_error = _provider_setup_error(resolved)
if provider_error:
console.print(Text(f"Agent: ✗ {provider_error}", style="red"))
console.print("Complete provider/model setup:")
_print_model_setup_steps(config_path)
else:
provider_ready = True
console.print("Agent: [green]✓ provider/model configuration is ready[/green]")
# Check API keys from registry
for spec in PROVIDERS:
p = getattr(loaded.providers, spec.name, None)
@ -2697,14 +2879,25 @@ def status(
console.print(f"{spec.label}: [green]✓ (OAuth)[/green]")
elif spec.is_local:
# Local deployments show api_base instead of api_key
if p.api_base:
if resolve_env_refs(p.api_base or ""):
console.print(f"{spec.label}: [green]✓ {p.api_base}[/green]")
else:
console.print(f"{spec.label}: [dim]not set[/dim]")
else:
has_key = bool(p.api_key)
has_key = bool(resolve_env_refs(p.api_key or ""))
console.print(f"{spec.label}: {'[green]✓[/green]' if has_key else '[dim]not set[/dim]'}")
if provider_ready:
console.print()
console.print('Next: [cyan]nanobot agent -m "Hello!"[/cyan]')
console.print(
"[dim]Status does not call the model or verify network access and credentials.[/dim]"
)
else:
console.print("Agent: [red]✗ configuration file not found[/red]")
console.print("Create the provider/model configuration:")
_print_model_setup_steps(config_path)
# ============================================================================
# OAuth Login

View File

@ -29,6 +29,7 @@ from nanobot.webui.build import BuildMode
RuntimeConfigLoader = Callable[[str | None, str | None], Config]
GatewayRunner = Callable[..., None]
GatewayConfigValidator = Callable[[Config], str | None]
GatewayRuntimeFactory = Callable[..., Any]
GatewayServiceFactory = Callable[[], Any]
WebUIBundlePreparer = Callable[[Config, BuildMode], None]
@ -40,6 +41,7 @@ def create_gateway_app(
log_handler_id: int,
load_runtime_config: RuntimeConfigLoader,
run_gateway: GatewayRunner,
validate_startup_config: GatewayConfigValidator | None = None,
runtime_factory: GatewayRuntimeFactory | None = None,
service_factory: GatewayServiceFactory | None = None,
prepare_webui_bundle: WebUIBundlePreparer | None = None,
@ -149,6 +151,8 @@ def create_gateway_app(
raise typer.Exit(1)
if background:
cfg = load_runtime_config(config, workspace)
if validate_startup_config is not None:
validate_startup_config(cfg)
if prepare_webui_bundle is not None:
prepare_webui_bundle(cfg, interactive_build_mode())
runtime = runtime_for_instance(workspace=workspace, config=config)
@ -171,7 +175,18 @@ def create_gateway_app(
configure_logging(verbose)
cfg = load_runtime_config(config, workspace)
run_gateway(cfg, port=port, webui_bundle_mode=interactive_build_mode())
unconfigured_provider_error = None
if validate_startup_config is not None:
unconfigured_provider_error = validate_startup_config(cfg)
if unconfigured_provider_error is None:
run_gateway(cfg, port=port, webui_bundle_mode=interactive_build_mode())
else:
run_gateway(
cfg,
port=port,
webui_bundle_mode=interactive_build_mode(),
unconfigured_provider_error=unconfigured_provider_error,
)
@gateway_app.command("status")
def gateway_status(
@ -225,6 +240,8 @@ def create_gateway_app(
) -> None:
"""Restart the background gateway."""
cfg = load_runtime_config(config, workspace)
if validate_startup_config is not None:
validate_startup_config(cfg)
if prepare_webui_bundle is not None:
prepare_webui_bundle(cfg, interactive_build_mode())
runtime = runtime_for_instance(workspace=workspace, config=config)

View File

@ -1,5 +1,6 @@
"""Configuration module for nanobot."""
from nanobot.config.errors import ConfigIssue, ConfigLoadError
from nanobot.config.loader import get_config_path, load_config
from nanobot.config.paths import (
get_cli_history_path,
@ -17,6 +18,8 @@ from nanobot.config.schema import Config
__all__ = [
"Config",
"ConfigIssue",
"ConfigLoadError",
"load_config",
"get_config_path",
"get_data_dir",

112
nanobot/config/errors.py Normal file
View File

@ -0,0 +1,112 @@
"""User-safe configuration diagnostics."""
from __future__ import annotations
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Literal
from pydantic import ValidationError
ConfigErrorKind = Literal[
"invalid_json",
"invalid_root",
"invalid_schema",
"missing_env",
"io_error",
]
ConfigPathPart = str | int
_SAFE_LOCATION_PART = re.compile(r"[A-Za-z_][A-Za-z0-9_-]{0,63}")
def _display_location_part(part: ConfigPathPart) -> str:
if isinstance(part, int):
return str(part)
return part if _SAFE_LOCATION_PART.fullmatch(part) else "<redacted>"
@dataclass(frozen=True)
class ConfigIssue:
"""One actionable configuration problem."""
path: tuple[ConfigPathPart, ...]
message: str
@property
def location(self) -> str:
# Pydantic locations can contain user-controlled mapping keys. Only
# render conventional config identifiers so credential-bearing URLs
# and other free-form values cannot leak through a redacted error.
if not self.path:
return "<root>"
return ".".join(_display_location_part(part) for part in self.path)
class ConfigLoadError(ValueError):
"""A structured, user-safe configuration loading failure."""
def __init__(
self,
path: Path,
*,
kind: ConfigErrorKind,
summary: str,
issues: tuple[ConfigIssue, ...] = (),
) -> None:
self.path = path
self.kind = kind
self.summary = summary
self.issues = issues
super().__init__(summary)
def __str__(self) -> str:
lines = [f"Invalid configuration: {self.path}", "", self.summary]
for issue in self.issues[:10]:
lines.extend(("", f" {issue.location}", f" {issue.message}"))
remaining = len(self.issues) - 10
if remaining > 0:
lines.extend(("", f" … and {remaining} more issue(s)"))
return "\n".join(lines)
def validation_issues(
error: ValidationError,
) -> tuple[ConfigIssue, ...]:
"""Convert Pydantic details to actionable messages without exposing input values."""
issues: list[ConfigIssue] = []
for detail in error.errors(
include_url=False,
include_context=False,
include_input=False,
):
location = tuple(detail.get("loc", ()))
code = str(detail.get("type") or "")
message = _friendly_validation_message(
str(detail.get("msg") or "Invalid value"),
code,
)
issues.append(ConfigIssue(path=location, message=message))
return tuple(issues)
def _friendly_validation_message(message: str, code: str) -> str:
if code == "extra_forbidden":
return "Unknown setting."
if code == "missing":
return "This setting is required."
if code in {"assertion_error", "value_error"}:
# Custom validators control these messages and may interpolate the
# rejected value. Keep the field location, but never render that text.
return "Value does not satisfy this setting's requirements."
if message.startswith("Value error, "):
message = message.removeprefix("Value error, ")
elif message.startswith("Input should be "):
message = "Must be " + message.removeprefix("Input should be ")
elif message.startswith("Input should have "):
message = "Must have " + message.removeprefix("Input should have ")
if message:
message = message[:1].upper() + message[1:]
if message and message[-1] not in ".!?":
message += "."
return message or "Invalid value."

View File

@ -6,9 +6,10 @@ import re
from pathlib import Path
from typing import Any
import pydantic
from pydantic import BaseModel
from pydantic import BaseModel, ValidationError
from pydantic_settings import SettingsError
from nanobot.config.errors import ConfigIssue, ConfigLoadError, validation_issues
from nanobot.config.schema import Config, _resolve_tool_config_refs
from nanobot.utils.helpers import _write_text_atomic
@ -47,15 +48,79 @@ def load_config(config_path: Path | None = None) -> Config:
path = config_path or get_config_path()
config = Config()
if path.exists():
if not path.exists():
try:
with open(path, encoding="utf-8") as f:
data = json.load(f)
data = _migrate_config(data)
config = Config.model_validate(data)
except (json.JSONDecodeError, ValueError, pydantic.ValidationError) as e:
raise ValueError(f"Failed to load config from {path}: {e}") from e
config = Config()
except SettingsError as exc:
raise ConfigLoadError(
path,
kind="invalid_schema",
summary=(
"Environment-based configuration could not be parsed. "
"Check that complex NANOBOT_* values use valid JSON."
),
) from exc
except ValidationError as exc:
raise ConfigLoadError(
path,
kind="invalid_schema",
summary="Environment-based configuration is invalid.",
issues=validation_issues(exc),
) from exc
_apply_ssrf_whitelist(config)
return config
try:
with path.open(encoding="utf-8") as handle:
data = json.load(handle)
except json.JSONDecodeError as exc:
raise ConfigLoadError(
path,
kind="invalid_json",
summary=(
f"JSON syntax error at line {exc.lineno}, column {exc.colno}: "
f"{_sentence(exc.msg)}"
),
) from exc
except UnicodeDecodeError as exc:
raise ConfigLoadError(
path,
kind="io_error",
summary="The file is not valid UTF-8.",
) from exc
except OSError as exc:
detail = exc.strerror or type(exc).__name__
raise ConfigLoadError(
path,
kind="io_error",
summary=f"Unable to read the file: {_sentence(detail)}",
) from exc
if not isinstance(data, dict):
root_type = type(data).__name__
raise ConfigLoadError(
path,
kind="invalid_root",
summary="The top level of config.json must be a JSON object.",
issues=(
ConfigIssue(
path=(),
message=f"Expected an object, but found {root_type}.",
),
),
)
data = _migrate_config(data)
try:
config = Config.model_validate(data)
except ValidationError as exc:
issues = validation_issues(exc)
raise ConfigLoadError(
path,
kind="invalid_schema",
summary=f"Found {len(issues)} invalid setting(s).",
issues=issues,
) from exc
_apply_ssrf_whitelist(config)
return config
@ -116,13 +181,25 @@ def merge_missing_defaults(existing: Any, defaults: Any) -> Any:
_ENV_REF_PATTERN = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
def resolve_config_env_vars(config: Config) -> Config:
def resolve_config_env_vars(
config: Config,
*,
config_path: Path | None = None,
) -> Config:
"""Return *config* with ``${VAR}`` env-var references resolved.
Walks in place so fields declared with ``exclude=True`` survive;
returns the same instance when no references are present.
Raises ``ValueError`` if a referenced variable is not set.
Raises ``ConfigLoadError`` if a referenced variable is not set.
"""
missing = tuple(_missing_env_issues(config))
if missing:
raise ConfigLoadError(
config_path or get_config_path(),
kind="missing_env",
summary=f"Found {len(missing)} missing environment variable reference(s).",
issues=missing,
)
return _resolve_in_place(config)
@ -176,6 +253,42 @@ def _resolve_in_place(obj: Any) -> Any:
return obj
def _missing_env_issues(
obj: Any,
path: tuple[str | int, ...] = (),
) -> list[ConfigIssue]:
if isinstance(obj, str):
return [
ConfigIssue(
path=path,
message=f"Environment variable '{name}' is not set.",
)
for name in dict.fromkeys(_ENV_REF_PATTERN.findall(obj))
if name not in os.environ
]
if isinstance(obj, BaseModel):
issues: list[ConfigIssue] = []
for name, field in type(obj).model_fields.items():
alias = field.serialization_alias or field.alias or name
part = alias if isinstance(alias, str) else name
issues.extend(_missing_env_issues(getattr(obj, name), (*path, part)))
for name, value in (obj.__pydantic_extra__ or {}).items():
issues.extend(_missing_env_issues(value, (*path, name)))
return issues
if isinstance(obj, dict):
issues = []
for name, value in obj.items():
part = name if isinstance(name, (str, int)) else str(name)
issues.extend(_missing_env_issues(value, (*path, part)))
return issues
if isinstance(obj, list):
issues = []
for index, value in enumerate(obj):
issues.extend(_missing_env_issues(value, (*path, index)))
return issues
return []
def _resolve_env_vars(obj: object) -> object:
"""Recursively resolve ``${VAR}`` patterns in plain strings/dicts/lists."""
if isinstance(obj, str):
@ -201,15 +314,26 @@ def _migrate_config(data: dict) -> dict:
"""Migrate old config formats to current."""
# Move tools.exec.restrictToWorkspace → tools.restrictToWorkspace
tools = data.get("tools", {})
if not isinstance(tools, dict):
return data
exec_cfg = tools.get("exec", {})
if "restrictToWorkspace" in exec_cfg and "restrictToWorkspace" not in tools:
if (
isinstance(exec_cfg, dict)
and "restrictToWorkspace" in exec_cfg
and "restrictToWorkspace" not in tools
):
tools["restrictToWorkspace"] = exec_cfg.pop("restrictToWorkspace")
# Move tools.myEnabled / tools.mySet → tools.my.{enable, allowSet}.
# The old flat keys shipped in the initial MyTool landing; wrapping them in a
# sub-config keeps `web` / `exec` / `my` symmetric and gives room to grow.
if "myEnabled" in tools or "mySet" in tools:
my_cfg = tools.setdefault("my", {})
my_cfg = tools.get("my")
if my_cfg is None:
my_cfg = {}
tools["my"] = my_cfg
if not isinstance(my_cfg, dict):
return data
if "myEnabled" in tools and "enable" not in my_cfg:
my_cfg["enable"] = tools.pop("myEnabled")
else:
@ -220,3 +344,10 @@ def _migrate_config(data: dict) -> dict:
tools.pop("mySet", None)
return data
def _sentence(message: str) -> str:
message = message.strip()
if message and message[-1] not in ".!?":
message += "."
return message

View File

@ -105,7 +105,10 @@ class Nanobot:
if not resolved.exists():
raise FileNotFoundError(f"Config not found: {resolved}")
config: Config = resolve_config_env_vars(load_config(resolved))
config: Config = resolve_config_env_vars(
load_config(resolved),
config_path=resolved,
)
if workspace is not None:
config.agents.defaults.workspace = str(
Path(workspace).expanduser().resolve()

View File

@ -21,6 +21,15 @@ class ProviderSnapshot:
model_preset: str | None = None
@dataclass(frozen=True)
class _ProviderSetup:
model: str
provider_name: str
provider_config: ProviderConfig | None
spec: ProviderSpec | None
backend: str
def _resolve_model_preset(
config: Config,
*,
@ -40,18 +49,20 @@ def _provider_extra_headers(
return headers or None
def _make_provider_core(
def _resolve_provider_setup(
config: Config,
*,
preset: ModelPresetConfig,
model: str | None = None,
) -> LLMProvider:
"""Create a plain LLM provider without failover wrapping."""
) -> _ProviderSetup:
"""Resolve and validate provider configuration without constructing a client."""
model = model or preset.model
provider_name = config.get_provider_name(model, preset=preset)
p = config.get_provider(model, preset=preset)
spec = find_by_name(provider_name) if provider_name else None
if provider_name and not spec and p:
if not provider_name:
raise ValueError(f"No provider is configured for model '{model}'.")
spec = find_by_name(provider_name)
if not spec and p:
if not p.api_base:
raise ValueError(f"Provider '{provider_name}' requires api_base in config.")
spec = create_dynamic_spec(
@ -79,12 +90,57 @@ def _make_provider_core(
and not (p and p.api_base)
):
raise ValueError(f"Provider '{provider_name}' requires api_base in config.")
elif backend == "openai_compat" and not model.startswith("bedrock/"):
elif backend in {"anthropic", "openai_compat"} and not (
backend == "openai_compat" and model.startswith("bedrock/")
):
needs_key = not (p and p.api_key)
exempt = spec and (spec.is_oauth or spec.is_local or spec.is_direct)
if needs_key and not exempt:
raise ValueError(f"No API key configured for provider '{provider_name}'.")
return _ProviderSetup(
model=model,
provider_name=provider_name,
provider_config=p,
spec=spec,
backend=backend,
)
def validate_provider_setup(
config: Config,
*,
preset_name: str | None = None,
preset: ModelPresetConfig | None = None,
model: str | None = None,
) -> None:
"""Validate local provider/model settings without loading a provider client."""
resolved = _resolve_model_preset(config, preset_name=preset_name, preset=preset)
_resolve_provider_setup(
config,
preset=resolved,
model=model,
)
def _make_provider_core(
config: Config,
*,
preset: ModelPresetConfig,
model: str | None = None,
) -> LLMProvider:
"""Create a plain LLM provider without failover wrapping."""
setup = _resolve_provider_setup(
config,
preset=preset,
model=model,
)
model = setup.model
provider_name = setup.provider_name
p = setup.provider_config
spec = setup.spec
backend = setup.backend
if backend == "openai_codex":
from nanobot.providers.openai_codex_provider import OpenAICodexProvider
@ -315,6 +371,9 @@ def load_provider_snapshot(
from nanobot.config.loader import load_config, resolve_config_env_vars
return build_provider_snapshot(
resolve_config_env_vars(load_config(config_path)),
resolve_config_env_vars(
load_config(config_path),
config_path=config_path,
),
preset_name=preset_name,
)

View File

@ -1,4 +1,5 @@
import asyncio
import json
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock
@ -8,6 +9,7 @@ import pytest
from nanobot.agent.loop import AgentLoop
from nanobot.bus.queue import MessageBus
from nanobot.bus.runtime_events import RuntimeModelChanged
from nanobot.config.errors import ConfigLoadError
from nanobot.config.loader import save_config
from nanobot.config.schema import Config, ModelPresetConfig
from nanobot.providers.base import GenerationSettings
@ -127,6 +129,24 @@ def test_llm_runtime_surfaces_invalidated_config_errors(tmp_path: Path) -> None:
loop.llm_runtime()
def test_provider_snapshot_missing_env_reports_explicit_config_path(
tmp_path: Path,
monkeypatch,
) -> None:
name = "NANOBOT_TEST_REFRESH_MISSING_KEY"
monkeypatch.delenv(name, raising=False)
config_path = tmp_path / "custom.json"
config_path.write_text(
json.dumps({"providers": {"openrouter": {"apiKey": f"${{{name}}}"}}}),
encoding="utf-8",
)
with pytest.raises(ConfigLoadError) as exc_info:
load_provider_snapshot(config_path)
assert exc_info.value.path == config_path
def test_same_snapshot_default_clears_preset_and_publishes_update(tmp_path: Path) -> None:
base_provider = _provider("base-model")
fast_provider = _provider("fast-model")

View File

@ -1833,12 +1833,10 @@ def _test_provider_snapshot(provider: object, config: Config) -> ProviderSnapsho
def _patch_webui_provider_ready(monkeypatch) -> None:
provider = _fake_provider()
def _snapshot(config: Config, **_kwargs) -> ProviderSnapshot:
return _test_provider_snapshot(provider, config)
monkeypatch.setattr("nanobot.providers.factory.build_provider_snapshot", _snapshot)
monkeypatch.setattr(
"nanobot.providers.factory.validate_provider_setup",
lambda _config: None,
)
def _patch_gateway_ports_free(monkeypatch) -> None:
@ -1883,6 +1881,10 @@ def _patch_cli_command_runtime(
"nanobot.providers.factory.load_provider_snapshot",
lambda _config_path=None: _test_provider_snapshot(provider_factory(config), config),
)
monkeypatch.setattr(
"nanobot.cli.commands._provider_setup_error",
lambda _config: None,
)
_patch_gateway_ports_free(monkeypatch)
if message_bus is not None:
@ -2116,6 +2118,9 @@ def test_webui_missing_runtime_env_fails_before_starting_gateway(
assert result.exit_code == 1
assert missing_env in result.stdout
assert "nanobot status --config" in result.stdout
assert config_file.name in result.stdout
assert "Traceback" not in result.stdout
assert f"${{{missing_env}}}" in config_file.read_text(encoding="utf-8")
@ -2145,6 +2150,10 @@ def test_webui_yes_still_refuses_invalid_custom_model_setup(
assert result.exit_code == 1
assert "provider/model setup is incomplete" in result.stdout
assert "Settings → Models" in result.stdout
assert "nanobot onboard --wizard" in result.stdout
assert "nanobot status --config" in result.stdout
assert config_file.name in result.stdout
def test_webui_background_starts_runtime_and_opens_browser(monkeypatch, tmp_path: Path) -> None:
@ -2564,6 +2573,7 @@ def test_gateway_unbound_agent_cron_is_skipped(
monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: provider)
monkeypatch.setattr("nanobot.cli.commands._provider_setup_error", lambda _config: None)
_patch_gateway_ports_free(monkeypatch)
monkeypatch.setattr(
"nanobot.providers.factory.build_provider_snapshot",
@ -2691,6 +2701,7 @@ def test_gateway_bound_cron_runs_as_session_turn(
monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: provider)
monkeypatch.setattr("nanobot.cli.commands._provider_setup_error", lambda _config: None)
_patch_gateway_ports_free(monkeypatch)
monkeypatch.setattr(
"nanobot.providers.factory.build_provider_snapshot",
@ -2997,6 +3008,8 @@ def test_gateway_local_trigger_queue_submits_agent_turns(
agent_kwargs = seen["agent_from_config_kwargs"]
kwargs = seen["local_trigger_queue_kwargs"]
assert isinstance(agent_kwargs["provider"], UnconfiguredProvider) is bool(setup_error)
refreshed_snapshot = agent_kwargs["provider_snapshot_loader"]()
assert not isinstance(refreshed_snapshot.provider, UnconfiguredProvider)
assert "local_trigger_store" in agent_kwargs
assert kwargs["store"] is agent_kwargs["local_trigger_store"]
assert "bus" not in kwargs

View File

@ -0,0 +1,539 @@
import json
import pytest
from typer.testing import CliRunner
from nanobot.cli.commands import app
from nanobot.gateway import GatewayRuntime, GatewayStartOptions, GatewayStatus, RuntimeResult
runner = CliRunner()
_ANTHROPIC_BACKEND_CASES = (
("anthropic", "anthropic", "claude-sonnet-4-5", "ANTHROPIC_API_KEY", "Anthropic"),
("kimi_coding", "kimiCoding", "kimi-for-coding", "KIMI_CODING_API_KEY", "Kimi Coding"),
(
"minimax_anthropic",
"minimaxAnthropic",
"MiniMax-M2.7-highspeed",
"MINIMAX_API_KEY",
"MiniMax (Anthropic)",
),
)
def _without_rendered_line_breaks(output: str) -> str:
return "".join(output.splitlines())
def _write_ready_config(config_path, *, channels: dict | None = None) -> None:
config_path.write_text(
json.dumps(
{
"agents": {
"defaults": {
"model": "ollama/llama3.2",
"provider": "ollama",
}
},
"providers": {
"ollama": {
"apiBase": "http://localhost:11434/v1",
}
},
"channels": channels or {},
}
),
encoding="utf-8",
)
def test_status_reports_ready_provider_and_next_step(tmp_path) -> None:
config_path = tmp_path / "config.json"
_write_ready_config(config_path)
result = runner.invoke(app, ["status", "--config", str(config_path)])
assert result.exit_code == 0
assert "Agent: ✓ provider/model configuration is ready" in result.stdout
assert "Ollama:" in result.stdout
assert "Model: ollama/llama3.2" in result.stdout
assert 'nanobot agent -m "Hello!"' in result.stdout
assert "Status does not call the model" in result.stdout
def test_status_validates_bedrock_without_constructing_provider(
tmp_path,
monkeypatch,
) -> None:
from nanobot.providers.bedrock_provider import BedrockProvider
config_path = tmp_path / "config.json"
config_path.write_text(
json.dumps(
{
"agents": {
"defaults": {
"model": "bedrock/amazon.nova-lite-v1:0",
"provider": "bedrock",
}
},
"providers": {"bedrock": {"region": "us-east-1"}},
}
),
encoding="utf-8",
)
def _unexpected_init(*_args, **_kwargs) -> None:
pytest.fail("status must not construct a provider client")
monkeypatch.setattr(BedrockProvider, "__init__", _unexpected_init)
result = runner.invoke(app, ["status", "--config", str(config_path)])
assert result.exit_code == 0
assert "Agent: ✓ provider/model configuration is ready" in result.stdout
assert "Status does not call the model or verify network access" in result.stdout
@pytest.mark.parametrize(
("provider", "provider_key", "model", "env_name", "label"),
_ANTHROPIC_BACKEND_CASES,
)
def test_status_reports_missing_key_for_anthropic_backends(
tmp_path,
monkeypatch,
provider: str,
provider_key: str,
model: str,
env_name: str,
label: str,
) -> None:
monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
monkeypatch.delenv(env_name, raising=False)
config_path = tmp_path / "config.json"
config_path.write_text(
json.dumps(
{
"agents": {"defaults": {"model": model, "provider": provider}},
"providers": {provider_key: {}},
}
),
encoding="utf-8",
)
result = runner.invoke(app, ["status", "--config", str(config_path)])
output = _without_rendered_line_breaks(result.stdout)
assert result.exit_code == 0
assert f"Agent: ✗ No API key configured for provider '{provider}'." in output
assert f"{label}: not set" in output
assert "provider/model configuration is ready" not in output
assert 'Next: nanobot agent -m "Hello!"' not in output
assert "Settings → Models" in output
@pytest.mark.parametrize(
("provider", "provider_key", "model", "env_name", "label"),
_ANTHROPIC_BACKEND_CASES,
)
def test_status_accepts_resolved_key_for_anthropic_backends(
tmp_path,
monkeypatch,
provider: str,
provider_key: str,
model: str,
env_name: str,
label: str,
) -> None:
monkeypatch.setenv(env_name, "test-api-key")
config_path = tmp_path / "config.json"
config_path.write_text(
json.dumps(
{
"agents": {"defaults": {"model": model, "provider": provider}},
"providers": {provider_key: {"apiKey": f"${{{env_name}}}"}},
}
),
encoding="utf-8",
)
result = runner.invoke(app, ["status", "--config", str(config_path)])
assert result.exit_code == 0
assert "Agent: ✓ provider/model configuration is ready" in result.stdout
assert f"{label}: ✓" in result.stdout
assert 'nanobot agent -m "Hello!"' in result.stdout
def test_status_reports_missing_provider_with_shortest_setup_routes(tmp_path) -> None:
config_path = tmp_path / "config.json"
config_path.write_text("{}", encoding="utf-8")
result = runner.invoke(app, ["status", "--config", str(config_path)])
assert result.exit_code == 0
assert "Agent: ✗" in result.stdout
assert "No provider is configured for model" in result.stdout
assert "Settings → Models" in result.stdout
assert "nanobot onboard --wizard" in result.stdout
assert "nanobot status --config" in result.stdout
def test_status_readiness_does_not_validate_channel_configuration(tmp_path) -> None:
config_path = tmp_path / "config.json"
_write_ready_config(
config_path,
channels={"websocket": {"enabled": False, "path": "missing-slash"}},
)
result = runner.invoke(app, ["status", "--config", str(config_path)])
assert result.exit_code == 0
assert "Agent: ✓ provider/model configuration is ready" in result.stdout
assert "channels.websocket" not in result.stdout
def test_status_reports_json_location_without_traceback(tmp_path) -> None:
config_path = tmp_path / "config.json"
config_path.write_text("{broken", encoding="utf-8")
result = runner.invoke(app, ["status", "--config", str(config_path)])
assert result.exit_code == 1
assert "Invalid configuration" in result.stdout
assert "JSON syntax error at line 1, column 2" in result.stdout
assert "Traceback" not in result.stdout
def test_status_reports_field_without_exposing_secret(tmp_path) -> None:
config_path = tmp_path / "config.json"
secret = "should-never-appear"
config_path.write_text(
json.dumps({"providers": {"openrouter": {"apiKey": [secret]}}}),
encoding="utf-8",
)
result = runner.invoke(app, ["status", "--config", str(config_path)])
assert result.exit_code == 1
assert "providers.openrouter.apiKey" in result.stdout
assert secret not in result.stdout
assert "input_value" not in result.stdout
assert "errors.pydantic.dev" not in result.stdout
def test_status_reports_missing_env_var_at_field(tmp_path, monkeypatch) -> None:
name = "NANOBOT_TEST_STATUS_MISSING"
monkeypatch.delenv(name, raising=False)
config_path = tmp_path / "config.json"
config_path.write_text(
json.dumps({"providers": {"openrouter": {"apiKey": f"${{{name}}}"}}}),
encoding="utf-8",
)
result = runner.invoke(app, ["status", "--config", str(config_path)])
assert result.exit_code == 0
assert "providers.openrouter.apiKey" in result.stdout
assert name in result.stdout
assert "OpenRouter: not set" in result.stdout
assert "OpenRouter: ✓" not in result.stdout
def test_webui_reports_malformed_environment_config_without_traceback(
tmp_path,
monkeypatch,
) -> None:
config_path = tmp_path / "missing.json"
invalid_value = "sensitive-not-json"
monkeypatch.setenv("NANOBOT_PROVIDERS", invalid_value)
result = runner.invoke(
app,
["webui", "--config", str(config_path), "--yes", "--no-open"],
)
assert result.exit_code == 1
assert isinstance(result.exception, SystemExit)
assert "Environment-based configuration could not be parsed" in result.stdout
assert "nanobot status --config" in result.stdout
assert invalid_value not in result.stdout
assert not config_path.exists()
@pytest.mark.parametrize(
"args",
[
["webui", "--yes", "--no-open"],
["agent", "--message", "hello"],
],
)
def test_agent_entrypoints_point_invalid_config_to_status(tmp_path, args: list[str]) -> None:
config_path = tmp_path / "config.json"
config_path.write_text("{broken", encoding="utf-8")
result = runner.invoke(app, [*args, "--config", str(config_path)])
assert result.exit_code == 1
assert "Invalid configuration" in result.stdout
assert "nanobot status --config" in result.stdout
assert "Traceback" not in result.stdout
def test_agent_provider_setup_failure_points_to_shortest_routes(tmp_path) -> None:
config_path = tmp_path / "config.json"
workspace = tmp_path / "workspace"
config_path.write_text(
json.dumps({"agents": {"defaults": {"workspace": str(workspace)}}}),
encoding="utf-8",
)
result = runner.invoke(
app,
["agent", "--message", "hello", "--config", str(config_path)],
)
output = _without_rendered_line_breaks(result.stdout)
assert result.exit_code == 1
assert "Agent cannot start: No provider is configured for model" in output
assert "Settings → Models" in output
assert "nanobot onboard --wizard" in output
assert "nanobot status --config" in output
assert "Traceback" not in output
assert not workspace.exists()
@pytest.mark.parametrize(
"args",
[
["gateway"],
["gateway", "--background"],
["gateway", "restart"],
],
)
def test_gateway_provider_setup_failure_points_to_shortest_routes_when_webui_disabled(
tmp_path,
monkeypatch,
args: list[str],
) -> None:
config_path = tmp_path / "explicit-gateway-config.json"
workspace = tmp_path / "workspace"
config_path.write_text(
json.dumps(
{
"agents": {"defaults": {"workspace": str(workspace)}},
"channels": {"websocket": {"enabled": False}},
}
),
encoding="utf-8",
)
def unexpected_managed_start(*_args, **_kwargs) -> RuntimeResult:
pytest.fail("provider validation must fail before a managed gateway start")
monkeypatch.setattr(GatewayRuntime, "start_background", unexpected_managed_start)
monkeypatch.setattr(GatewayRuntime, "restart", unexpected_managed_start)
result = runner.invoke(app, [*args, "--config", str(config_path)])
output = _without_rendered_line_breaks(result.stdout)
assert result.exit_code == 1
assert "Gateway cannot start: No provider is configured for model" in output
assert "Settings → Models" in output
assert "nanobot onboard --wizard" in output
assert "nanobot status --config" in output
assert config_path.name in output
assert "Traceback" not in output
assert not workspace.exists()
@pytest.mark.parametrize(
("args", "start_mode"),
[
(["gateway", "--background"], "background"),
(["gateway", "restart"], "restart"),
],
)
@pytest.mark.parametrize("secret_field", ["tokenIssueSecret", "token"])
def test_gateway_missing_provider_managed_start_for_webui_setup(
tmp_path,
monkeypatch,
args: list[str],
start_mode: str,
secret_field: str,
) -> None:
config_path = tmp_path / "explicit-gateway-config.json"
workspace = tmp_path / "workspace"
webui_port = 18776
bootstrap_secret = "must-not-appear-in-gateway-output"
config_path.write_text(
json.dumps(
{
"agents": {"defaults": {"workspace": str(workspace)}},
"channels": {
"websocket": {
"enabled": True,
"host": "127.0.0.1",
"port": webui_port,
secret_field: bootstrap_secret,
}
},
}
),
encoding="utf-8",
)
started_options: list[tuple[str, GatewayStartOptions]] = []
status = GatewayStatus(
running=True,
pid=12345,
state_path=tmp_path / "gateway.json",
log_path=tmp_path / "gateway.log",
started_at="2026-07-28T00:00:00Z",
port=18790,
reason="running",
)
def fake_start_background(
_runtime: GatewayRuntime,
options: GatewayStartOptions,
) -> RuntimeResult:
started_options.append(("background", options))
return RuntimeResult(True, "gateway_started_background", status)
def fake_restart(
_runtime: GatewayRuntime,
options: GatewayStartOptions,
*,
timeout_s: int,
) -> RuntimeResult:
assert timeout_s == 20
started_options.append(("restart", options))
return RuntimeResult(True, "gateway_started_background", status)
monkeypatch.setattr(GatewayRuntime, "start_background", fake_start_background)
monkeypatch.setattr(GatewayRuntime, "restart", fake_restart)
monkeypatch.setattr(
"nanobot.cli.commands.ensure_webui_bundle",
lambda **_kwargs: None,
)
result = runner.invoke(
app,
[*args, "--config", str(config_path)],
)
output = _without_rendered_line_breaks(result.stdout)
assert result.exit_code == 0
assert "Provider/model setup is incomplete: No provider is configured for model" in output
assert "Gateway will start so you can configure a provider and model" in output
assert "WebUI Settings" in output
assert "Models." in output
assert f"WebUI: http://127.0.0.1:{webui_port}" in output
assert f"channels.websocket.{secret_field}" in output
if secret_field == "token":
assert "channels.websocket.tokenIssueSecret" not in output
assert "bootstrapSecret" not in output
assert bootstrap_secret not in output
assert "Gateway cannot start" not in output
assert started_options == [
(
start_mode,
GatewayStartOptions(
port=18790,
config_path=str(config_path.resolve()),
),
)
]
assert not workspace.exists()
def test_gateway_invalid_webui_config_blocks_unconfigured_setup_mode(tmp_path) -> None:
config_path = tmp_path / "invalid-webui-config.json"
workspace = tmp_path / "workspace"
config_path.write_text(
json.dumps(
{
"agents": {"defaults": {"workspace": str(workspace)}},
"channels": {
"websocket": {
"enabled": True,
"port": "not-a-port",
}
},
}
),
encoding="utf-8",
)
result = runner.invoke(app, ["gateway", "--config", str(config_path)])
output = _without_rendered_line_breaks(result.stdout)
assert result.exit_code == 1
assert "Gateway configuration is invalid." in output
assert "channels.websocket.port" in output
assert "Provider/model setup is incomplete" not in output
assert "Traceback" not in output
assert not workspace.exists()
@pytest.mark.parametrize(
("args", "summary", "retry_command"),
[
(
["webui", "--yes", "--no-open"],
"WebUI configuration is invalid.",
"nanobot webui --config",
),
(
["gateway"],
"Gateway configuration is invalid.",
"nanobot gateway --config",
),
],
)
def test_runtime_config_validation_is_redacted_and_actionable(
tmp_path,
args: list[str],
summary: str,
retry_command: str,
) -> None:
config_path = tmp_path / "explicit-runtime-config.json"
workspace = tmp_path / "workspace"
invalid_value = "sensitive-not-a-port"
_write_ready_config(
config_path,
channels={
"websocket": {
"enabled": True,
"port": invalid_value,
}
},
)
data = json.loads(config_path.read_text(encoding="utf-8"))
data["agents"]["defaults"]["workspace"] = str(workspace)
config_path.write_text(json.dumps(data), encoding="utf-8")
result = runner.invoke(app, [*args, "--config", str(config_path)])
output = _without_rendered_line_breaks(result.stdout)
assert result.exit_code == 1
assert summary in output
assert "channels.websocket.port" in output
assert retry_command in output
assert config_path.name in output
assert invalid_value not in output
assert "input_value" not in output
assert "errors.pydantic.dev" not in output
assert "Traceback" not in output
assert not workspace.exists()
def test_status_missing_file_points_to_setup_without_changing_exit_contract(tmp_path) -> None:
config_path = tmp_path / "missing.json"
result = runner.invoke(app, ["status", "--config", str(config_path)])
assert result.exit_code == 0
assert "configuration file not found" in result.stdout
assert "nanobot webui" in result.stdout
assert "nanobot onboard --wizard" in result.stdout

View File

@ -1,5 +1,6 @@
from pathlib import Path
import pytest
import typer
from rich.console import Console
from typer.testing import CliRunner
@ -27,6 +28,7 @@ class FakeRuntime:
self.restarted_options: GatewayStartOptions | None = None
self.stop_timeout: int | None = None
self.follow_tail: int | None = None
self.validated_configs: list[Config] = []
def start_background(self, options: GatewayStartOptions) -> RuntimeResult:
self.started_options = options
@ -84,11 +86,15 @@ class FakeServiceInstaller:
)
def _test_app(tmp_path: Path, config: Config | None = None):
def _test_app(
tmp_path: Path,
config: Config | None = None,
startup_error: str | None = None,
):
app = typer.Typer()
fake_runtime = FakeRuntime(tmp_path)
fake_service = FakeServiceInstaller(tmp_path)
run_calls: list[tuple[Config, int | None, str | None]] = []
run_calls: list[tuple[Config, int | None, str | None, str | None]] = []
prepare_calls: list[tuple[Config, str]] = []
def load_runtime_config(_config_path: str | None, _workspace: str | None) -> Config:
@ -99,18 +105,28 @@ def _test_app(tmp_path: Path, config: Config | None = None):
*,
port: int | None = None,
webui_bundle_mode: str | None = None,
unconfigured_provider_error: str | None = None,
) -> None:
run_calls.append((config, port, webui_bundle_mode))
run_calls.append(
(config, port, webui_bundle_mode, unconfigured_provider_error)
)
def prepare_webui_bundle(config: Config, mode: str) -> None:
prepare_calls.append((config, mode))
def validate_startup_config(config: Config) -> str | None:
fake_runtime.validated_configs.append(config)
return startup_error
app.add_typer(
create_gateway_app(
console=Console(),
log_handler_id=0,
load_runtime_config=load_runtime_config,
run_gateway=run_gateway,
validate_startup_config=(
validate_startup_config if startup_error is not None else None
),
runtime_factory=lambda **_kwargs: fake_runtime,
service_factory=lambda: fake_service,
prepare_webui_bundle=prepare_webui_bundle,
@ -131,6 +147,46 @@ def test_gateway_default_still_runs_foreground(tmp_path):
assert calls[0][2] == "warn"
def test_gateway_foreground_passes_recoverable_provider_error_to_runner(tmp_path):
setup_error = "No provider is configured."
app, _runtime, _service, calls, _prepare_calls = _test_app(
tmp_path,
startup_error=setup_error,
)
result = runner.invoke(app, ["gateway"])
assert result.exit_code == 0
assert len(calls) == 1
assert calls[0][3] == setup_error
assert len(_runtime.validated_configs) == 1
@pytest.mark.parametrize(
("args", "runtime_attribute"),
[
(["gateway", "--background"], "started_options"),
(["gateway", "restart"], "restarted_options"),
],
)
def test_gateway_managed_start_allows_recoverable_provider_error(
tmp_path,
args: list[str],
runtime_attribute: str,
) -> None:
app, fake_runtime, _service, calls, _prepare_calls = _test_app(
tmp_path,
startup_error="No provider is configured.",
)
result = runner.invoke(app, args)
assert result.exit_code == 0
assert calls == []
assert len(fake_runtime.validated_configs) == 1
assert getattr(fake_runtime, runtime_attribute) is not None
def test_gateway_background_starts_detached_runtime(tmp_path):
config = Config()
config.gateway.port = 18792

View File

@ -2,6 +2,7 @@ import json
import pytest
from nanobot.config.errors import ConfigLoadError
from nanobot.config.loader import load_config
from nanobot.config.schema import ApiConfig
@ -12,13 +13,35 @@ def test_load_config_missing_file_uses_defaults(tmp_path) -> None:
assert config.agents.defaults.model
def test_load_config_reports_malformed_environment_safely(
tmp_path,
monkeypatch,
) -> None:
config_path = tmp_path / "missing.json"
invalid_value = "sensitive-not-json"
monkeypatch.setenv("NANOBOT_PROVIDERS", invalid_value)
with pytest.raises(ConfigLoadError) as exc_info:
load_config(config_path)
error = exc_info.value
assert error.kind == "invalid_schema"
assert error.path == config_path
assert "complex NANOBOT_* values use valid JSON" in str(error)
assert invalid_value not in str(error)
def test_load_config_invalid_json_fails_fast(tmp_path) -> None:
config_path = tmp_path / "config.json"
config_path.write_text("{broken json", encoding="utf-8")
with pytest.raises(ValueError, match="Failed to load config"):
with pytest.raises(ConfigLoadError) as exc_info:
load_config(config_path)
error = exc_info.value
assert error.kind == "invalid_json"
assert "line 1, column 2" in str(error)
def test_load_config_invalid_schema_fails_fast(tmp_path) -> None:
config_path = tmp_path / "config.json"
@ -27,9 +50,113 @@ def test_load_config_invalid_schema_fails_fast(tmp_path) -> None:
encoding="utf-8",
)
with pytest.raises(ValueError, match="Failed to load config"):
with pytest.raises(ConfigLoadError) as exc_info:
load_config(config_path)
error = exc_info.value
message = str(error)
assert error.kind == "invalid_schema"
assert "tools.exec.timeout" in message
assert "Must be greater than or equal to 0." in message
assert "input_value" not in message
assert "errors.pydantic.dev" not in message
@pytest.mark.parametrize(
("content", "root_type"),
[("[]", "list"), ("null", "NoneType"), ('"value"', "str")],
)
def test_load_config_rejects_non_object_root(tmp_path, content: str, root_type: str) -> None:
config_path = tmp_path / "config.json"
config_path.write_text(content, encoding="utf-8")
with pytest.raises(ConfigLoadError) as exc_info:
load_config(config_path)
error = exc_info.value
assert error.kind == "invalid_root"
assert f"Expected an object, but found {root_type}." in str(error)
def test_load_config_error_does_not_expose_invalid_secret_value(tmp_path) -> None:
config_path = tmp_path / "config.json"
secret = "should-never-appear"
config_path.write_text(
json.dumps({"providers": {"openrouter": {"apiKey": [secret]}}}),
encoding="utf-8",
)
with pytest.raises(ConfigLoadError) as exc_info:
load_config(config_path)
assert secret not in str(exc_info.value)
def test_load_config_error_redacts_untrusted_location_parts(tmp_path) -> None:
config_path = tmp_path / "config.json"
secret = "should-never-appear-in-location"
server_name = f"https://user:{secret}@example.test"
config_path.write_text(
json.dumps(
{
"tools": {
"mcpServers": {
server_name: {"toolTimeout": "not-a-number"},
}
}
}
),
encoding="utf-8",
)
with pytest.raises(ConfigLoadError) as exc_info:
load_config(config_path)
message = str(exc_info.value)
assert "tools.mcpServers.<redacted>.toolTimeout" in message
assert server_name not in message
assert secret not in message
def test_load_config_error_does_not_trust_custom_validator_message(tmp_path) -> None:
config_path = tmp_path / "config.json"
secret = "diagnostic-secret-should-not-print"
config_path.write_text(
json.dumps({"providers": {"openrouter": {"thinkingStyle": secret}}}),
encoding="utf-8",
)
with pytest.raises(ConfigLoadError) as exc_info:
load_config(config_path)
message = str(exc_info.value)
assert "providers.openrouter.thinkingStyle" in message
assert "Value does not satisfy this setting's requirements." in message
assert secret not in message
@pytest.mark.parametrize(
"tools",
[
[],
{"exec": []},
{"my": 1, "myEnabled": True},
],
)
def test_load_config_malformed_legacy_sections_use_structured_error(
tmp_path,
tools: object,
) -> None:
config_path = tmp_path / "config.json"
config_path.write_text(json.dumps({"tools": tools}), encoding="utf-8")
with pytest.raises(ConfigLoadError) as exc_info:
load_config(config_path)
error = exc_info.value
assert error.kind == "invalid_schema"
assert "tools" in str(error)
@pytest.mark.parametrize("host", ["0.0.0.0", "::"])
def test_api_config_requires_key_for_wildcard_hosts(host: str) -> None:

View File

@ -2,6 +2,7 @@ import json
import pytest
from nanobot.config.errors import ConfigLoadError
from nanobot.config.loader import (
_resolve_env_vars,
load_config,
@ -66,6 +67,22 @@ class TestResolveConfig:
resolved = resolve_config_env_vars(raw)
assert resolved.providers.groq.api_key == "resolved-key"
def test_missing_env_var_reports_config_field(self, tmp_path, monkeypatch):
name = "NANOBOT_TEST_MISSING_PROVIDER_KEY"
monkeypatch.delenv(name, raising=False)
config_path = tmp_path / "config.json"
config = Config.model_validate(
{"providers": {"openrouter": {"apiKey": f"${{{name}}}"}}}
)
with pytest.raises(ConfigLoadError) as exc_info:
resolve_config_env_vars(config, config_path=config_path)
error = exc_info.value
assert error.kind == "missing_env"
assert "providers.openrouter.apiKey" in str(error)
assert name in str(error)
def test_save_preserves_templates(self, tmp_path, monkeypatch):
monkeypatch.setenv("MY_TOKEN", "real-token")
config_path = tmp_path / "config.json"

View File

@ -1,7 +1,10 @@
import json
import warnings
import pytest
from nanobot.agent.model_presets import load_model_preset_catalog
from nanobot.config.errors import ConfigLoadError
from nanobot.config.schema import Config
@ -16,6 +19,24 @@ def test_resolve_preset_returns_defaults_when_no_preset() -> None:
assert resolved.reasoning_effort == config.agents.defaults.reasoning_effort
def test_model_preset_catalog_missing_env_reports_explicit_config_path(
tmp_path,
monkeypatch,
) -> None:
name = "NANOBOT_TEST_CATALOG_MISSING_KEY"
monkeypatch.delenv(name, raising=False)
config_path = tmp_path / "custom.json"
config_path.write_text(
json.dumps({"providers": {"openrouter": {"apiKey": f"${{{name}}}"}}}),
encoding="utf-8",
)
with pytest.raises(ConfigLoadError) as exc_info:
load_model_preset_catalog(config_path)
assert exc_info.value.path == config_path
def test_agent_timezone_rejects_unknown_iana_name() -> None:
with pytest.raises(ValueError, match="unknown timezone"):
Config.model_validate({"agents": {"defaults": {"timezone": "Not/AZone"}}})

View File

@ -74,6 +74,26 @@ def test_from_config_missing_file():
Nanobot.from_config("/nonexistent/config.json")
def test_from_config_missing_env_reports_explicit_config_path(
tmp_path,
monkeypatch,
) -> None:
from nanobot.config.errors import ConfigLoadError
name = "NANOBOT_TEST_SDK_MISSING_KEY"
monkeypatch.delenv(name, raising=False)
config_path = tmp_path / "custom.json"
config_path.write_text(
json.dumps({"providers": {"openrouter": {"apiKey": f"${{{name}}}"}}}),
encoding="utf-8",
)
with pytest.raises(ConfigLoadError) as exc_info:
Nanobot.from_config(config_path)
assert exc_info.value.path == config_path.resolve()
def test_from_config_creates_instance(tmp_path):
config_path = _write_config(tmp_path)
bot = Nanobot.from_config(config_path, workspace=tmp_path)