add one-time migration for legacy global cron store

When upgrading, if jobs.json exists at the old global path and not yet
at the workspace path, move it automatically.  Prevents silent loss of
existing cron jobs.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
MiguelPF
2026-03-18 10:16:06 +01:00
parent c33e01ee62
commit 4e56481f0b
2 changed files with 59 additions and 0 deletions

View File

@@ -449,6 +449,18 @@ def _print_deprecated_memory_window_notice(config: Config) -> None:
) )
def _migrate_cron_store(config: "Config") -> None:
"""One-time migration: move legacy global cron store into the workspace."""
from nanobot.config.paths import get_cron_dir
legacy_path = get_cron_dir() / "jobs.json"
new_path = config.workspace_path / "cron" / "jobs.json"
if legacy_path.is_file() and not new_path.exists():
new_path.parent.mkdir(parents=True, exist_ok=True)
import shutil
shutil.move(str(legacy_path), str(new_path))
# ============================================================================ # ============================================================================
# Gateway / Server # Gateway / Server
# ============================================================================ # ============================================================================
@@ -484,6 +496,9 @@ def gateway(
provider = _make_provider(config) provider = _make_provider(config)
session_manager = SessionManager(config.workspace_path) session_manager = SessionManager(config.workspace_path)
# Migrate legacy global cron store into workspace (one-time)
_migrate_cron_store(config)
# Create cron service with workspace-scoped store # Create cron service with workspace-scoped store
cron_store_path = config.workspace_path / "cron" / "jobs.json" cron_store_path = config.workspace_path / "cron" / "jobs.json"
cron = CronService(cron_store_path) cron = CronService(cron_store_path)
@@ -671,6 +686,9 @@ def agent(
bus = MessageBus() bus = MessageBus()
provider = _make_provider(config) provider = _make_provider(config)
# Migrate legacy global cron store into workspace (one-time)
_migrate_cron_store(config)
# Create cron service with workspace-scoped store # Create cron service with workspace-scoped store
cron_store_path = config.workspace_path / "cron" / "jobs.json" cron_store_path = config.workspace_path / "cron" / "jobs.json"
cron = CronService(cron_store_path) cron = CronService(cron_store_path)

View File

@@ -523,6 +523,47 @@ def test_gateway_uses_config_directory_for_cron_store(monkeypatch, tmp_path: Pat
assert seen["cron_store"] == config.workspace_path / "cron" / "jobs.json" assert seen["cron_store"] == config.workspace_path / "cron" / "jobs.json"
def test_migrate_cron_store_moves_legacy_file(tmp_path: Path) -> None:
"""Legacy global jobs.json is moved into the workspace on first run."""
from nanobot.cli.commands import _migrate_cron_store
legacy_dir = tmp_path / "global" / "cron"
legacy_dir.mkdir(parents=True)
legacy_file = legacy_dir / "jobs.json"
legacy_file.write_text('{"jobs": []}')
config = Config()
config.agents.defaults.workspace = str(tmp_path / "workspace")
workspace_cron = config.workspace_path / "cron" / "jobs.json"
with patch("nanobot.config.paths.get_cron_dir", return_value=legacy_dir):
_migrate_cron_store(config)
assert workspace_cron.exists()
assert workspace_cron.read_text() == '{"jobs": []}'
assert not legacy_file.exists()
def test_migrate_cron_store_skips_when_workspace_file_exists(tmp_path: Path) -> None:
"""Migration does not overwrite an existing workspace cron store."""
from nanobot.cli.commands import _migrate_cron_store
legacy_dir = tmp_path / "global" / "cron"
legacy_dir.mkdir(parents=True)
(legacy_dir / "jobs.json").write_text('{"old": true}')
config = Config()
config.agents.defaults.workspace = str(tmp_path / "workspace")
workspace_cron = config.workspace_path / "cron" / "jobs.json"
workspace_cron.parent.mkdir(parents=True)
workspace_cron.write_text('{"new": true}')
with patch("nanobot.config.paths.get_cron_dir", return_value=legacy_dir):
_migrate_cron_store(config)
assert workspace_cron.read_text() == '{"new": true}'
def test_gateway_uses_configured_port_when_cli_flag_is_missing(monkeypatch, tmp_path: Path) -> None: def test_gateway_uses_configured_port_when_cli_flag_is_missing(monkeypatch, tmp_path: Path) -> None:
config_file = tmp_path / "instance" / "config.json" config_file = tmp_path / "instance" / "config.json"
config_file.parent.mkdir(parents=True) config_file.parent.mkdir(parents=True)