merge: integrate pr-1581 multi-instance path cleanup

This commit is contained in:
Re-bin
2026-03-08 02:58:28 +00:00
16 changed files with 351 additions and 65 deletions

View File

@@ -905,30 +905,92 @@ MCP tools are automatically discovered and registered on startup. The LLM can us
## Multiple Instances ## Multiple Instances
Run multiple nanobot instances simultaneously, each with its own workspace and configuration. Run multiple nanobot instances simultaneously with separate configs and runtime data. Use `--config` as the main entrypoint, and optionally use `--workspace` to override the workspace for a specific run.
### Quick Start
```bash ```bash
# Instance A - Telegram bot # Instance A - Telegram bot
nanobot gateway -w ~/.nanobot/botA -p 18791 nanobot gateway --config ~/.nanobot-telegram/config.json
# Instance B - Discord bot # Instance B - Discord bot
nanobot gateway -w ~/.nanobot/botB -p 18792 nanobot gateway --config ~/.nanobot-discord/config.json
# Instance C - Using custom config file # Instance C - Feishu bot with custom port
nanobot gateway -w ~/.nanobot/botC -c ~/.nanobot/botC/config.json -p 18793 nanobot gateway --config ~/.nanobot-feishu/config.json --port 18792
``` ```
| Option | Short | Description | ### Path Resolution
|--------|-------|-------------|
| `--workspace` | `-w` | Workspace directory (default: `~/.nanobot/workspace`) |
| `--config` | `-c` | Config file path (default: `~/.nanobot/config.json`) |
| `--port` | `-p` | Gateway port (default: `18790`) |
Each instance has its own: When using `--config`, nanobot derives its runtime data directory from the config file location. The workspace still comes from `agents.defaults.workspace` unless you override it with `--workspace`.
- Workspace directory (MEMORY.md, HEARTBEAT.md, session files)
- Cron jobs storage (`workspace/cron/jobs.json`)
- Configuration (if using `--config`)
| Component | Resolved From | Example |
|-----------|---------------|---------|
| **Config** | `--config` path | `~/.nanobot-A/config.json` |
| **Workspace** | `--workspace` or config | `~/.nanobot-A/workspace/` |
| **Cron Jobs** | config directory | `~/.nanobot-A/cron/` |
| **Media / runtime state** | config directory | `~/.nanobot-A/media/` |
### How It Works
- `--config` selects which config file to load
- By default, the workspace comes from `agents.defaults.workspace` in that config
- If you pass `--workspace`, it overrides the workspace from the config file
### Minimal Setup
1. Copy your base config into a new instance directory.
2. Set a different `agents.defaults.workspace` for that instance.
3. Start the instance with `--config`.
Example config:
```json
{
"agents": {
"defaults": {
"workspace": "~/.nanobot-telegram/workspace",
"model": "anthropic/claude-sonnet-4-6"
}
},
"channels": {
"telegram": {
"enabled": true,
"token": "YOUR_TELEGRAM_BOT_TOKEN"
}
},
"gateway": {
"port": 18790
}
}
```
Start separate instances:
```bash
nanobot gateway --config ~/.nanobot-telegram/config.json
nanobot gateway --config ~/.nanobot-discord/config.json
```
Override workspace for one-off runs when needed:
```bash
nanobot gateway --config ~/.nanobot-telegram/config.json --workspace /tmp/nanobot-telegram-test
```
### Common Use Cases
- Run separate bots for Telegram, Discord, Feishu, and other platforms
- Keep testing and production instances isolated
- Use different models or providers for different teams
- Serve multiple tenants with separate configs and runtime data
### Notes
- Each instance must use a different port if they run at the same time
- Use a different workspace per instance if you want isolated memory, sessions, and skills
- `--workspace` overrides the workspace defined in the config file
- Cron jobs and runtime media/state are derived from the config directory
## CLI Reference ## CLI Reference

View File

@@ -18,7 +18,6 @@ import qrcode from 'qrcode-terminal';
import pino from 'pino'; import pino from 'pino';
import { writeFile, mkdir } from 'fs/promises'; import { writeFile, mkdir } from 'fs/promises';
import { join } from 'path'; import { join } from 'path';
import { homedir } from 'os';
import { randomBytes } from 'crypto'; import { randomBytes } from 'crypto';
const VERSION = '0.1.0'; const VERSION = '0.1.0';
@@ -162,7 +161,7 @@ export class WhatsAppClient {
private async downloadMedia(msg: any, mimetype?: string, fileName?: string): Promise<string | null> { private async downloadMedia(msg: any, mimetype?: string, fileName?: string): Promise<string | null> {
try { try {
const mediaDir = join(homedir(), '.nanobot', 'media'); const mediaDir = join(this.options.authDir, '..', 'media');
await mkdir(mediaDir, { recursive: true }); await mkdir(mediaDir, { recursive: true });
const buffer = await downloadMediaMessage(msg, 'buffer', {}) as Buffer; const buffer = await downloadMediaMessage(msg, 'buffer', {}) as Buffer;

View File

@@ -12,6 +12,7 @@ from loguru import logger
from nanobot.bus.events import OutboundMessage from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel from nanobot.channels.base import BaseChannel
from nanobot.config.paths import get_media_dir
from nanobot.config.schema import DiscordConfig from nanobot.config.schema import DiscordConfig
from nanobot.utils.helpers import split_message from nanobot.utils.helpers import split_message
@@ -289,7 +290,7 @@ class DiscordChannel(BaseChannel):
content_parts = [content] if content else [] content_parts = [content] if content else []
media_paths: list[str] = [] media_paths: list[str] = []
media_dir = Path.home() / ".nanobot" / "media" media_dir = get_media_dir("discord")
for attachment in payload.get("attachments") or []: for attachment in payload.get("attachments") or []:
url = attachment.get("url") url = attachment.get("url")

View File

@@ -14,6 +14,7 @@ from loguru import logger
from nanobot.bus.events import OutboundMessage from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel from nanobot.channels.base import BaseChannel
from nanobot.config.paths import get_media_dir
from nanobot.config.schema import FeishuConfig from nanobot.config.schema import FeishuConfig
import importlib.util import importlib.util
@@ -732,8 +733,7 @@ class FeishuChannel(BaseChannel):
(file_path, content_text) - file_path is None if download failed (file_path, content_text) - file_path is None if download failed
""" """
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
media_dir = Path.home() / ".nanobot" / "media" media_dir = get_media_dir("feishu")
media_dir.mkdir(parents=True, exist_ok=True)
data, filename = None, None data, filename = None, None

View File

@@ -38,7 +38,7 @@ except ImportError as e:
from nanobot.bus.events import OutboundMessage from nanobot.bus.events import OutboundMessage
from nanobot.channels.base import BaseChannel from nanobot.channels.base import BaseChannel
from nanobot.config.loader import get_data_dir from nanobot.config.paths import get_data_dir, get_media_dir
from nanobot.utils.helpers import safe_filename from nanobot.utils.helpers import safe_filename
TYPING_NOTICE_TIMEOUT_MS = 30_000 TYPING_NOTICE_TIMEOUT_MS = 30_000
@@ -490,9 +490,7 @@ class MatrixChannel(BaseChannel):
return False return False
def _media_dir(self) -> Path: def _media_dir(self) -> Path:
d = get_data_dir() / "media" / "matrix" return get_media_dir("matrix")
d.mkdir(parents=True, exist_ok=True)
return d
@staticmethod @staticmethod
def _event_source_content(event: RoomMessage) -> dict[str, Any]: def _event_source_content(event: RoomMessage) -> dict[str, Any]:

View File

@@ -15,8 +15,8 @@ from loguru import logger
from nanobot.bus.events import OutboundMessage from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel from nanobot.channels.base import BaseChannel
from nanobot.config.paths import get_runtime_subdir
from nanobot.config.schema import MochatConfig from nanobot.config.schema import MochatConfig
from nanobot.utils.helpers import get_data_path
try: try:
import socketio import socketio
@@ -224,7 +224,7 @@ class MochatChannel(BaseChannel):
self._socket: Any = None self._socket: Any = None
self._ws_connected = self._ws_ready = False self._ws_connected = self._ws_ready = False
self._state_dir = get_data_path() / "mochat" self._state_dir = get_runtime_subdir("mochat")
self._cursor_path = self._state_dir / "session_cursors.json" self._cursor_path = self._state_dir / "session_cursors.json"
self._session_cursor: dict[str, int] = {} self._session_cursor: dict[str, int] = {}
self._cursor_save_task: asyncio.Task | None = None self._cursor_save_task: asyncio.Task | None = None

View File

@@ -15,6 +15,7 @@ from telegram.request import HTTPXRequest
from nanobot.bus.events import OutboundMessage from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel from nanobot.channels.base import BaseChannel
from nanobot.config.paths import get_media_dir
from nanobot.config.schema import TelegramConfig from nanobot.config.schema import TelegramConfig
from nanobot.utils.helpers import split_message from nanobot.utils.helpers import split_message
@@ -536,10 +537,7 @@ class TelegramChannel(BaseChannel):
getattr(media_file, 'mime_type', None), getattr(media_file, 'mime_type', None),
getattr(media_file, 'file_name', None), getattr(media_file, 'file_name', None),
) )
# Save to workspace/media/ media_dir = get_media_dir("telegram")
from pathlib import Path
media_dir = Path.home() / ".nanobot" / "media"
media_dir.mkdir(parents=True, exist_ok=True)
file_path = media_dir / f"{media_file.file_id[:16]}{ext}" file_path = media_dir / f"{media_file.file_id[:16]}{ext}"
await file.download_to_drive(str(file_path)) await file.download_to_drive(str(file_path))

View File

@@ -30,6 +30,7 @@ from rich.table import Table
from rich.text import Text from rich.text import Text
from nanobot import __logo__, __version__ from nanobot import __logo__, __version__
from nanobot.config.paths import get_workspace_path
from nanobot.config.schema import Config from nanobot.config.schema import Config
from nanobot.utils.helpers import sync_workspace_templates from nanobot.utils.helpers import sync_workspace_templates
@@ -99,7 +100,9 @@ def _init_prompt_session() -> None:
except Exception: except Exception:
pass pass
history_file = Path.home() / ".nanobot" / "history" / "cli_history" from nanobot.config.paths import get_cli_history_path
history_file = get_cli_history_path()
history_file.parent.mkdir(parents=True, exist_ok=True) history_file.parent.mkdir(parents=True, exist_ok=True)
_PROMPT_SESSION = PromptSession( _PROMPT_SESSION = PromptSession(
@@ -170,7 +173,6 @@ def onboard():
"""Initialize nanobot configuration and workspace.""" """Initialize nanobot configuration and workspace."""
from nanobot.config.loader import get_config_path, load_config, save_config from nanobot.config.loader import get_config_path, load_config, save_config
from nanobot.config.schema import Config from nanobot.config.schema import Config
from nanobot.utils.helpers import get_workspace_path
config_path = get_config_path() config_path = get_config_path()
@@ -272,14 +274,25 @@ def _make_provider(config: Config):
def gateway( def gateway(
port: int = typer.Option(18790, "--port", "-p", help="Gateway port"), port: int = typer.Option(18790, "--port", "-p", help="Gateway port"),
workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"), workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
config: str | None = typer.Option(None, "--config", "-c", help="Config file path"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose output"), verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose output"),
config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
): ):
"""Start the nanobot gateway.""" """Start the nanobot gateway."""
# Set config path if provided (must be done before any imports that use get_data_dir)
if config:
from nanobot.config.loader import set_config_path
config_path = Path(config).expanduser().resolve()
if not config_path.exists():
console.print(f"[red]Error: Config file not found: {config_path}[/red]")
raise typer.Exit(1)
set_config_path(config_path)
console.print(f"[dim]Using config: {config_path}[/dim]")
from nanobot.agent.loop import AgentLoop from nanobot.agent.loop import AgentLoop
from nanobot.bus.queue import MessageBus from nanobot.bus.queue import MessageBus
from nanobot.channels.manager import ChannelManager from nanobot.channels.manager import ChannelManager
from nanobot.config.loader import load_config from nanobot.config.loader import load_config
from nanobot.config.paths import get_cron_dir
from nanobot.cron.service import CronService from nanobot.cron.service import CronService
from nanobot.cron.types import CronJob from nanobot.cron.types import CronJob
from nanobot.heartbeat.service import HeartbeatService from nanobot.heartbeat.service import HeartbeatService
@@ -289,20 +302,18 @@ def gateway(
import logging import logging
logging.basicConfig(level=logging.DEBUG) logging.basicConfig(level=logging.DEBUG)
config_path = Path(config) if config else None console.print(f"{__logo__} Starting nanobot gateway on port {port}...")
config = load_config(config_path)
config = load_config()
if workspace: if workspace:
config.agents.defaults.workspace = workspace config.agents.defaults.workspace = workspace
console.print(f"{__logo__} Starting nanobot gateway on port {port}...")
sync_workspace_templates(config.workspace_path) sync_workspace_templates(config.workspace_path)
bus = MessageBus() bus = MessageBus()
provider = _make_provider(config) provider = _make_provider(config)
session_manager = SessionManager(config.workspace_path) session_manager = SessionManager(config.workspace_path)
# Create cron service first (callback set after agent creation) # Create cron service first (callback set after agent creation)
# Use workspace path for per-instance cron store cron_store_path = get_cron_dir() / "jobs.json"
cron_store_path = config.workspace_path / "cron" / "jobs.json"
cron = CronService(cron_store_path) cron = CronService(cron_store_path)
# Create agent with cron service # Create agent with cron service
@@ -471,7 +482,8 @@ def agent(
from nanobot.agent.loop import AgentLoop from nanobot.agent.loop import AgentLoop
from nanobot.bus.queue import MessageBus from nanobot.bus.queue import MessageBus
from nanobot.config.loader import get_data_dir, load_config from nanobot.config.loader import load_config
from nanobot.config.paths import get_cron_dir
from nanobot.cron.service import CronService from nanobot.cron.service import CronService
config = load_config() config = load_config()
@@ -481,7 +493,7 @@ def agent(
provider = _make_provider(config) provider = _make_provider(config)
# Create cron service for tool usage (no callback needed for CLI unless running) # Create cron service for tool usage (no callback needed for CLI unless running)
cron_store_path = get_data_dir() / "cron" / "jobs.json" cron_store_path = get_cron_dir() / "jobs.json"
cron = CronService(cron_store_path) cron = CronService(cron_store_path)
if logs: if logs:
@@ -747,7 +759,9 @@ def _get_bridge_dir() -> Path:
import subprocess import subprocess
# User's bridge location # User's bridge location
user_bridge = Path.home() / ".nanobot" / "bridge" from nanobot.config.paths import get_bridge_install_dir
user_bridge = get_bridge_install_dir()
# Check if already built # Check if already built
if (user_bridge / "dist" / "index.js").exists(): if (user_bridge / "dist" / "index.js").exists():
@@ -805,6 +819,7 @@ def channels_login():
import subprocess import subprocess
from nanobot.config.loader import load_config from nanobot.config.loader import load_config
from nanobot.config.paths import get_runtime_subdir
config = load_config() config = load_config()
bridge_dir = _get_bridge_dir() bridge_dir = _get_bridge_dir()
@@ -815,6 +830,7 @@ def channels_login():
env = {**os.environ} env = {**os.environ}
if config.channels.whatsapp.bridge_token: if config.channels.whatsapp.bridge_token:
env["BRIDGE_TOKEN"] = config.channels.whatsapp.bridge_token env["BRIDGE_TOKEN"] = config.channels.whatsapp.bridge_token
env["AUTH_DIR"] = str(get_runtime_subdir("whatsapp-auth"))
try: try:
subprocess.run(["npm", "start"], cwd=bridge_dir, check=True, env=env) subprocess.run(["npm", "start"], cwd=bridge_dir, check=True, env=env)

View File

@@ -1,6 +1,30 @@
"""Configuration module for nanobot.""" """Configuration module for nanobot."""
from nanobot.config.loader import get_config_path, load_config from nanobot.config.loader import get_config_path, load_config
from nanobot.config.paths import (
get_bridge_install_dir,
get_cli_history_path,
get_cron_dir,
get_data_dir,
get_legacy_sessions_dir,
get_logs_dir,
get_media_dir,
get_runtime_subdir,
get_workspace_path,
)
from nanobot.config.schema import Config from nanobot.config.schema import Config
__all__ = ["Config", "load_config", "get_config_path"] __all__ = [
"Config",
"load_config",
"get_config_path",
"get_data_dir",
"get_runtime_subdir",
"get_media_dir",
"get_cron_dir",
"get_logs_dir",
"get_workspace_path",
"get_cli_history_path",
"get_bridge_install_dir",
"get_legacy_sessions_dir",
]

View File

@@ -6,17 +6,23 @@ from pathlib import Path
from nanobot.config.schema import Config from nanobot.config.schema import Config
# Global variable to store current config path (for multi-instance support)
_current_config_path: Path | None = None
def set_config_path(path: Path) -> None:
"""Set the current config path (used to derive data directory)."""
global _current_config_path
_current_config_path = path
def get_config_path() -> Path: def get_config_path() -> Path:
"""Get the default configuration file path.""" """Get the configuration file path."""
if _current_config_path:
return _current_config_path
return Path.home() / ".nanobot" / "config.json" return Path.home() / ".nanobot" / "config.json"
def get_data_dir() -> Path:
"""Get the nanobot data directory."""
from nanobot.utils.helpers import get_data_path
return get_data_path()
def load_config(config_path: Path | None = None) -> Config: def load_config(config_path: Path | None = None) -> Config:
""" """
Load configuration from file or create default. Load configuration from file or create default.

55
nanobot/config/paths.py Normal file
View File

@@ -0,0 +1,55 @@
"""Runtime path helpers derived from the active config context."""
from __future__ import annotations
from pathlib import Path
from nanobot.config.loader import get_config_path
from nanobot.utils.helpers import ensure_dir
def get_data_dir() -> Path:
"""Return the instance-level runtime data directory."""
return ensure_dir(get_config_path().parent)
def get_runtime_subdir(name: str) -> Path:
"""Return a named runtime subdirectory under the instance data dir."""
return ensure_dir(get_data_dir() / name)
def get_media_dir(channel: str | None = None) -> Path:
"""Return the media directory, optionally namespaced per channel."""
base = get_runtime_subdir("media")
return ensure_dir(base / channel) if channel else base
def get_cron_dir() -> Path:
"""Return the cron storage directory."""
return get_runtime_subdir("cron")
def get_logs_dir() -> Path:
"""Return the logs directory."""
return get_runtime_subdir("logs")
def get_workspace_path(workspace: str | None = None) -> Path:
"""Resolve and ensure the agent workspace path."""
path = Path(workspace).expanduser() if workspace else Path.home() / ".nanobot" / "workspace"
return ensure_dir(path)
def get_cli_history_path() -> Path:
"""Return the shared CLI history file path."""
return Path.home() / ".nanobot" / "history" / "cli_history"
def get_bridge_install_dir() -> Path:
"""Return the shared WhatsApp bridge installation directory."""
return Path.home() / ".nanobot" / "bridge"
def get_legacy_sessions_dir() -> Path:
"""Return the legacy global session directory used for migration fallback."""
return Path.home() / ".nanobot" / "sessions"

View File

@@ -9,6 +9,7 @@ from typing import Any
from loguru import logger from loguru import logger
from nanobot.config.paths import get_legacy_sessions_dir
from nanobot.utils.helpers import ensure_dir, safe_filename from nanobot.utils.helpers import ensure_dir, safe_filename
@@ -79,7 +80,7 @@ class SessionManager:
def __init__(self, workspace: Path): def __init__(self, workspace: Path):
self.workspace = workspace self.workspace = workspace
self.sessions_dir = ensure_dir(self.workspace / "sessions") self.sessions_dir = ensure_dir(self.workspace / "sessions")
self.legacy_sessions_dir = Path.home() / ".nanobot" / "sessions" self.legacy_sessions_dir = get_legacy_sessions_dir()
self._cache: dict[str, Session] = {} self._cache: dict[str, Session] = {}
def _get_session_path(self, key: str) -> Path: def _get_session_path(self, key: str) -> Path:

View File

@@ -1,5 +1,5 @@
"""Utility functions for nanobot.""" """Utility functions for nanobot."""
from nanobot.utils.helpers import ensure_dir, get_data_path, get_workspace_path from nanobot.utils.helpers import ensure_dir
__all__ = ["ensure_dir", "get_workspace_path", "get_data_path"] __all__ = ["ensure_dir"]

View File

@@ -24,17 +24,6 @@ def ensure_dir(path: Path) -> Path:
return path return path
def get_data_path() -> Path:
"""~/.nanobot data directory."""
return ensure_dir(Path.home() / ".nanobot")
def get_workspace_path(workspace: str | None = None) -> Path:
"""Resolve and ensure workspace path. Defaults to ~/.nanobot/workspace."""
path = Path(workspace).expanduser() if workspace else Path.home() / ".nanobot" / "workspace"
return ensure_dir(path)
def timestamp() -> str: def timestamp() -> str:
"""Current ISO timestamp.""" """Current ISO timestamp."""
return datetime.now().isoformat() return datetime.now().isoformat()

View File

@@ -14,13 +14,17 @@ from nanobot.providers.registry import find_by_model
runner = CliRunner() runner = CliRunner()
class _StopGateway(RuntimeError):
pass
@pytest.fixture @pytest.fixture
def mock_paths(): def mock_paths():
"""Mock config/workspace paths for test isolation.""" """Mock config/workspace paths for test isolation."""
with patch("nanobot.config.loader.get_config_path") as mock_cp, \ with patch("nanobot.config.loader.get_config_path") as mock_cp, \
patch("nanobot.config.loader.save_config") as mock_sc, \ patch("nanobot.config.loader.save_config") as mock_sc, \
patch("nanobot.config.loader.load_config") as mock_lc, \ patch("nanobot.config.loader.load_config") as mock_lc, \
patch("nanobot.utils.helpers.get_workspace_path") as mock_ws: patch("nanobot.cli.commands.get_workspace_path") as mock_ws:
base_dir = Path("./test_onboard_data") base_dir = Path("./test_onboard_data")
if base_dir.exists(): if base_dir.exists():
@@ -128,3 +132,94 @@ def test_litellm_provider_canonicalizes_github_copilot_hyphen_prefix():
def test_openai_codex_strip_prefix_supports_hyphen_and_underscore(): def test_openai_codex_strip_prefix_supports_hyphen_and_underscore():
assert _strip_model_prefix("openai-codex/gpt-5.1-codex") == "gpt-5.1-codex" assert _strip_model_prefix("openai-codex/gpt-5.1-codex") == "gpt-5.1-codex"
assert _strip_model_prefix("openai_codex/gpt-5.1-codex") == "gpt-5.1-codex" assert _strip_model_prefix("openai_codex/gpt-5.1-codex") == "gpt-5.1-codex"
def test_gateway_uses_workspace_from_config_by_default(monkeypatch, tmp_path: Path) -> None:
config_file = tmp_path / "instance" / "config.json"
config_file.parent.mkdir(parents=True)
config_file.write_text("{}")
config = Config()
config.agents.defaults.workspace = str(tmp_path / "config-workspace")
seen: dict[str, Path] = {}
monkeypatch.setattr(
"nanobot.config.loader.set_config_path",
lambda path: seen.__setitem__("config_path", path),
)
monkeypatch.setattr("nanobot.config.loader.load_config", lambda: config)
monkeypatch.setattr(
"nanobot.cli.commands.sync_workspace_templates",
lambda path: seen.__setitem__("workspace", path),
)
monkeypatch.setattr(
"nanobot.cli.commands._make_provider",
lambda _config: (_ for _ in ()).throw(_StopGateway("stop")),
)
result = runner.invoke(app, ["gateway", "--config", str(config_file)])
assert isinstance(result.exception, _StopGateway)
assert seen["config_path"] == config_file.resolve()
assert seen["workspace"] == Path(config.agents.defaults.workspace)
def test_gateway_workspace_option_overrides_config(monkeypatch, tmp_path: Path) -> None:
config_file = tmp_path / "instance" / "config.json"
config_file.parent.mkdir(parents=True)
config_file.write_text("{}")
config = Config()
config.agents.defaults.workspace = str(tmp_path / "config-workspace")
override = tmp_path / "override-workspace"
seen: dict[str, Path] = {}
monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None)
monkeypatch.setattr("nanobot.config.loader.load_config", lambda: config)
monkeypatch.setattr(
"nanobot.cli.commands.sync_workspace_templates",
lambda path: seen.__setitem__("workspace", path),
)
monkeypatch.setattr(
"nanobot.cli.commands._make_provider",
lambda _config: (_ for _ in ()).throw(_StopGateway("stop")),
)
result = runner.invoke(
app,
["gateway", "--config", str(config_file), "--workspace", str(override)],
)
assert isinstance(result.exception, _StopGateway)
assert seen["workspace"] == override
assert config.workspace_path == override
def test_gateway_uses_config_directory_for_cron_store(monkeypatch, tmp_path: Path) -> None:
config_file = tmp_path / "instance" / "config.json"
config_file.parent.mkdir(parents=True)
config_file.write_text("{}")
config = Config()
config.agents.defaults.workspace = str(tmp_path / "config-workspace")
seen: dict[str, Path] = {}
monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None)
monkeypatch.setattr("nanobot.config.loader.load_config", lambda: config)
monkeypatch.setattr("nanobot.config.paths.get_cron_dir", lambda: config_file.parent / "cron")
monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.commands._make_provider", lambda _config: object())
monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object())
monkeypatch.setattr("nanobot.session.manager.SessionManager", lambda _workspace: object())
class _StopCron:
def __init__(self, store_path: Path) -> None:
seen["cron_store"] = store_path
raise _StopGateway("stop")
monkeypatch.setattr("nanobot.cron.service.CronService", _StopCron)
result = runner.invoke(app, ["gateway", "--config", str(config_file)])
assert isinstance(result.exception, _StopGateway)
assert seen["cron_store"] == config_file.parent / "cron" / "jobs.json"

View File

@@ -0,0 +1,42 @@
from pathlib import Path
from nanobot.config.paths import (
get_bridge_install_dir,
get_cli_history_path,
get_cron_dir,
get_data_dir,
get_legacy_sessions_dir,
get_logs_dir,
get_media_dir,
get_runtime_subdir,
get_workspace_path,
)
def test_runtime_dirs_follow_config_path(monkeypatch, tmp_path: Path) -> None:
config_file = tmp_path / "instance-a" / "config.json"
monkeypatch.setattr("nanobot.config.paths.get_config_path", lambda: config_file)
assert get_data_dir() == config_file.parent
assert get_runtime_subdir("cron") == config_file.parent / "cron"
assert get_cron_dir() == config_file.parent / "cron"
assert get_logs_dir() == config_file.parent / "logs"
def test_media_dir_supports_channel_namespace(monkeypatch, tmp_path: Path) -> None:
config_file = tmp_path / "instance-b" / "config.json"
monkeypatch.setattr("nanobot.config.paths.get_config_path", lambda: config_file)
assert get_media_dir() == config_file.parent / "media"
assert get_media_dir("telegram") == config_file.parent / "media" / "telegram"
def test_shared_and_legacy_paths_remain_global() -> None:
assert get_cli_history_path() == Path.home() / ".nanobot" / "history" / "cli_history"
assert get_bridge_install_dir() == Path.home() / ".nanobot" / "bridge"
assert get_legacy_sessions_dir() == Path.home() / ".nanobot" / "sessions"
def test_workspace_path_is_explicitly_resolved() -> None:
assert get_workspace_path() == Path.home() / ".nanobot" / "workspace"
assert get_workspace_path("~/custom-workspace") == Path.home() / "custom-workspace"