From 8e4bcd7c2cea409335bcafaa85d1c262f33130aa Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 28 Jul 2026 11:12:44 +0200 Subject: [PATCH] Git-backed workspace sync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The workspace is treated as a git repo synced with a remote: a manual command, a configurable periodic loop, and a trigger route. The order is fetch, validate, then merge, so a bundle that fails validation never lands on disk — validating after the merge would leave the bad config in the working tree with nothing but a log line to say so. Fast-forward only, and a dirty config subtree refuses the merge rather than overwriting local edits. Sync validates with strict env, because the daemon will need those variables at load time and a lenient pass would wave through a bundle it cannot actually run. It honours its never-raises contract, so a workspace value that is null or a list reports a 400 rather than a stack trace, and the periodic loop reads config fresh each pass instead of holding a reference from startup. An unresolved variable the bundle itself needs is not blamed on the machine layers, and a path is reported by its real name rather than C-escaped. Co-Authored-By: Claude --- docs/config.md | 60 ++ nerve/cli.py | 44 ++ nerve/config.py | 43 +- nerve/config_validate.py | 32 + nerve/cron/service.py | 36 +- nerve/gateway/routes/__init__.py | 2 + nerve/gateway/routes/config.py | 56 ++ nerve/gateway/server.py | 19 + nerve/sync_service.py | 501 ++++++++++++++++ tests/test_config_validate.py | 44 ++ tests/test_sync_service.py | 980 +++++++++++++++++++++++++++++++ 11 files changed, 1804 insertions(+), 13 deletions(-) create mode 100644 nerve/gateway/routes/config.py create mode 100644 nerve/sync_service.py create mode 100644 tests/test_sync_service.py diff --git a/docs/config.md b/docs/config.md index b248c301..9ed99c88 100644 --- a/docs/config.md +++ b/docs/config.md @@ -179,6 +179,66 @@ is too old to read shows up as a startup warning on the instance and in `nerve doctor`. If a key is ever retired upstream, `--strict-keys` flags it here first; drop the key, or install the ref you deploy (`nerve @ git+...@v1.2.3`) instead. +## Git-Backed Workspace Sync + +The workspace can be a git repository whose remote is a shared **config repo**. +Config changes are proposed as PRs, reviewed and merged there; the instance pulls +the merged result and reloads, with no restart and no editing on the box. + +```bash +nerve config sync # git pull --ff-only the workspace, then validate +nerve config sync --branch main +nerve config sync --no-validate +nerve config sync --no-strict-env # tolerate ${VAR}s your shell doesn't have +``` + +Enable periodic pulls in the daemon (opt-in): + +```yaml +workspace_sync: + enabled: true # off by default + branch: main # empty = current tracking branch + interval_minutes: 1 # also the upper bound on how stale a box can be + validate: true # validate the pulled bundle before applying + strict_env: true # unset required ${VAR} in the bundle blocks the merge +``` + +Each sync is fetch, then validate, then fast-forward merge. The fetched bundle is +validated in a throwaway git worktree and the live working tree is fast-forwarded +only if that passes, so an invalid bundle never lands on disk and there is nothing +for a later reload or restart to pick up (`POST /api/config/sync` returns 400 and +leaves the workspace untouched). A pull that changed something reloads cron and MCP +config, so the merged change takes effect immediately. CI on the PR is still the +first line of defense. The remote and credentials come from git itself, so +configure `git remote` and auth in the workspace as usual. + +**Keep the config subtree clean.** Sync refuses to merge while +`/config/` has local changes: an edited or deleted tracked file, a +staged change, or an untracked file. Validation judges a clean checkout of the +fetched commit, but the merge lands in your working tree, and `--ff-only` only +refuses when the incoming commit touches the same path. Anything else would survive +the merge unchecked, leaving a bundle on disk that is not the one that passed. The +case that matters most is an untracked `config/cron/gates/*.py`: the daemon imports +and runs gate plugins and validation never loads them, so a box meant to run only +reviewed config would be running local code. Commit, discard or push local edits; +the failure message names the paths. Files matched by `.gitignore` inside `config/` +are warnings rather than refusals, since the shared repo can never carry them. + +Sync validates more strictly than CI: an unset required `${VAR}` blocks the merge. +CI has no secrets, so it reports those as info, but the daemon does have them, and a +bundle with an unresolved required variable is one it will refuse to load on its +next restart. If a shared change adds a `${VAR}` this box legitimately does not set, +use `workspace_sync.strict_env: false` rather than letting every sync fail. +`nerve config sync` runs in your shell, which may not carry the daemon's environment +(systemd `EnvironmentFile`, docker `--env-file`), so pass `--no-strict-env` there +for a one-off. Warnings never block a merge: an unrecognized cron gate type, an +unknown key, or a skipped validation. + +`workspace_sync` changes need a daemon restart. The sync loop reads the current +config object every cycle, so it adds no staleness of its own, but nothing refreshes +that object while the process runs. Turning `enabled` on needs a restart in any +case, because the sync task is only created at startup. + ## Migrating an Existing Install Installs from before the workspace-config layout are migrated automatically and diff --git a/nerve/cli.py b/nerve/cli.py index 177ff7c4..3d398606 100644 --- a/nerve/cli.py +++ b/nerve/cli.py @@ -1195,6 +1195,50 @@ def config_group() -> None: """Configuration commands.""" +@config_group.command("sync") +@click.option("--branch", default="", help="Branch to pull (default: current tracking branch).") +@click.option("--no-validate", is_flag=True, help="Skip validating the pulled bundle.") +@click.option( + "--no-strict-env", is_flag=True, + help="Allow unset ${ENV_VAR} references in the pulled bundle. Normally they " + "block the merge, because the daemon refuses to load a config with an " + "unresolved required variable. Use this when your shell doesn't carry " + "the daemon's environment (systemd/docker). Otherwise follows " + "workspace_sync.strict_env.", +) +@click.pass_context +def config_sync( + ctx: click.Context, branch: str, no_validate: bool, no_strict_env: bool, +) -> None: + """Pull the workspace from its git remote (the shared config repo). + + Fast-forward only. This moves the files; it does not tell a running daemon + about them. The daemon applies them on its next sync cycle, or immediately + via POST /api/config/sync. + """ + from nerve.sync_service import sync_workspace + + config = ctx.obj["config"] + if config is None: + raise click.ClickException("Config could not be loaded; run 'nerve doctor'.") + # Raw values, not Path(...): sync_workspace coerces inside its own + # never-raises guard, so a bad `workspace` reports instead of traceback. + result = sync_workspace( + config.workspace, ctx.obj["config_dir"], branch=branch, + validate=not no_validate, + strict_env=config.workspace_sync.strict_env and not no_strict_env, + ) + for warning in result.validation_warnings: + click.secho(f" [WARN] {warning}", fg="yellow") + if result.ok: + click.secho(result.message, fg="green") + else: + for err in result.validation_errors: + click.secho(f" [ERR] {err}", fg="red") + click.secho(result.message, fg="red") + ctx.exit(1) + + @config_group.command("validate") @click.option( "--workspace", "workspace", type=click.Path(), default=None, diff --git a/nerve/config.py b/nerve/config.py index 5cf06747..72473963 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -176,8 +176,7 @@ def _resolve_env_refs(merged: dict[str, Any]) -> dict[str, Any]: # # ${ENV_VAR} interpolation is applied once at the end. This keeps existing # single-file installs working unchanged (layer 1 absent → prior behavior) while -# letting shared settings live in the workspace. Lockdown mode (a later story) -# will drop layers 2/3 so only the tracked workspace layer applies. +# letting shared settings live in the workspace. def _read_yaml_mapping(path: Path, *, strict: bool = False) -> dict[str, Any]: @@ -947,6 +946,44 @@ def from_dict(cls, d: dict, workspace: Path | None = None) -> CronConfig: ) +@dataclass +class WorkspaceSyncConfig: + """Git-backed sync of the workspace from a remote (the config repo). + + Opt-in. When enabled, the daemon periodically ``git pull --ff-only``s the + workspace so config changes reviewed & merged via PR on the remote land on + the instance and hot-reload. The remote/branch come from git itself; only the + cadence and enablement live here. + """ + + enabled: bool = False + branch: str = "" # empty → git's default (current tracking branch) + # How often the daemon pulls. This is also the upper bound on how long a + # merged config change takes to reach the instance, since the pull is what + # applies it, so it is deliberately short; a fetch that finds nothing new is + # one round trip. Raise it for a large fleet pointed at one repo. + interval_minutes: int = 1 + validate: bool = True # validate the pulled bundle before applying + # Treat an unset required ${VAR} in the pulled bundle as invalid. On by + # default: the daemon refuses to load such a config, so merging it only + # defers the failure to the next restart. Configurable because a shared repo + # can introduce a ${VAR} that this particular box legitimately does not set, + # and the alternative is a machine whose every sync fails until someone + # notices. + strict_env: bool = True + + @classmethod + @_coerced + def from_dict(cls, d: dict) -> WorkspaceSyncConfig: + return cls( + enabled=d.get("enabled", False), + branch=str(d.get("branch", "") or ""), + interval_minutes=d.get("interval_minutes", 1), + validate=d.get("validate", True), + strict_env=d.get("strict_env", True), + ) + + @dataclass class BackupConfig: """Scheduled backup of Nerve state to a local directory. @@ -1863,6 +1900,7 @@ class NerveConfig: sync: SyncConfig = field(default_factory=SyncConfig) memory: MemoryConfig = field(default_factory=MemoryConfig) cron: CronConfig = field(default_factory=CronConfig) + workspace_sync: WorkspaceSyncConfig = field(default_factory=WorkspaceSyncConfig) backup: BackupConfig = field(default_factory=BackupConfig) sessions: SessionsConfig = field(default_factory=SessionsConfig) retention: RetentionConfig = field(default_factory=RetentionConfig) @@ -2068,6 +2106,7 @@ def _build_from_dict(cls, d: dict) -> NerveConfig: sync=SyncConfig.from_dict(d.get("sync", {})), memory=MemoryConfig.from_dict(d.get("memory", {})), cron=CronConfig.from_dict(d.get("cron", {}), workspace=workspace), + workspace_sync=WorkspaceSyncConfig.from_dict(d.get("workspace_sync", {})), backup=BackupConfig.from_dict(d.get("backup", {})), sessions=SessionsConfig.from_dict(d.get("sessions", {})), retention=RetentionConfig.from_dict(d.get("retention", {})), diff --git a/nerve/config_validate.py b/nerve/config_validate.py index e3405bee..5948b0c0 100644 --- a/nerve/config_validate.py +++ b/nerve/config_validate.py @@ -118,6 +118,38 @@ def validate_config_bundle( env_names = ", ".join(sorted(set(missing))) if missing: msg = f"references unset environment variable(s): {env_names}" + # Say which layer asked for them. The refs are collected from the merged + # config, so a variable named only by this host's config.yaml / + # config.local.yaml is otherwise indistinguishable from one the portable + # bundle needs — and when this runs as sync's gate, that reads as a defect + # in an incoming change that has nothing to do with it. + # + # Which is why the workspace layer is scanned separately rather than + # inferred from the machine one: a var both layers name would otherwise be + # reported as the machine's alone, sending the reader to a file whose only + # fault is that it also needs the variable, while the portable config that + # equally needs it goes unmentioned. Only claim exclusivity when the + # workspace config genuinely does not ask for the variable. + from_machine: list[str] = [] + cfg._interpolate_env(machine, from_machine) + from_workspace: list[str] = [] + cfg._interpolate_env(ws_settings, from_workspace) + unset = set(missing) + machine_only = sorted((set(from_machine) - set(from_workspace)) & unset) + both_layers = sorted(set(from_machine) & set(from_workspace) & unset) + clauses = [] + if machine_only: + clauses.append( + f"{', '.join(machine_only)} referenced only by this machine's " + f"config.yaml/config.local.yaml, not by the workspace config" + ) + if both_layers: + clauses.append( + f"{', '.join(both_layers)} referenced by both the workspace config " + f"and this machine's config.yaml/config.local.yaml" + ) + if clauses: + msg += " — " + "; ".join(clauses) (result.errors if strict_env else result.info).append(msg) # Unknown / misspelled keys: warnings by default (forward-compat / example diff --git a/nerve/cron/service.py b/nerve/cron/service.py index c3fb0352..9e67f41f 100644 --- a/nerve/cron/service.py +++ b/nerve/cron/service.py @@ -252,6 +252,9 @@ def __init__(self, config: NerveConfig, engine: AgentEngine, db: Database): self._jobs: list[CronJob] = [] self._source_runners: list[SourceRunner] = [] self._job_locks: dict[str, asyncio.Lock] = {} + # Serialize reload() so the file watcher, the sync loop, and the HTTP + # route can't interleave scheduler mutations. + self._reload_lock = asyncio.Lock() async def start(self) -> None: """Load jobs and start the scheduler.""" @@ -405,23 +408,34 @@ async def reload(self) -> dict: All-or-nothing: the complete change set — every trigger included — is built before the scheduler is touched, so a reload the daemon cannot carry out raises with the running schedule exactly as it was. That - covers the gate registry too; see :meth:`_reload_from_disk`. + covers the gate registry too; see :meth:`_reload_locked`. + + Serialized via a lock so concurrent callers (file watcher, sync loop, + HTTP route) can't interleave scheduler mutations. The two properties are + separate: the lock keeps two reloads from overlapping, and the planning + pass keeps a single failing one from applying half of itself. Returns a summary dict: ``{"added", "removed", "updated", "enabled"}``, where "updated" is every enabled job that already had a trigger. It says what the reload rescheduled, not which jobs the file changed. """ + async with self._reload_lock: + return await self._reload_locked() + + async def _reload_locked(self) -> dict: + """Reload under the lock, undoing the one change that isn't local. + + GATE_REGISTRY is process-global and has to be replaced *before* jobs are + rebuilt, because a CronJob builds its gates from it at construction time. + That puts it ahead of every check that can still refuse the reload, so it + needs undoing by hand — leaving the scheduler untouched is not enough when + a deleted plugin's gate has already been unregistered. The next CronJob + built from disk (run_job, rotate_session) would then fail to build for + want of a gate type, off a reload that returned 400: the instance would + lose jobs it had never agreed to change. + """ from nerve.cron.gates import GATE_REGISTRY - # The one piece of reload state that isn't local: GATE_REGISTRY is - # process-global and has to be replaced *before* jobs are rebuilt, - # because a CronJob builds its gates from it at construction time. That - # puts it ahead of every check that can still refuse the reload, so it - # needs undoing by hand — leaving the scheduler untouched is not enough - # when a deleted plugin's gate has already been unregistered. The next - # CronJob built from disk (run_job, rotate_session) would then fail to - # build for want of a gate type, off a reload that returned 400: the - # instance would lose jobs it had never agreed to change. gates_before = dict(GATE_REGISTRY) try: return await self._reload_from_disk(gates_before) @@ -435,7 +449,7 @@ async def reload(self) -> dict: async def _reload_from_disk( self, gates_before: dict[str, type["CronGate"]], ) -> dict: - """The body of :meth:`reload`. Only call it through there. + """The body of :meth:`reload`. Only call it through :meth:`_reload_locked`. Takes the pre-reload registry snapshot so it can report which gates vanished once the change is committed, and so its caller can restore diff --git a/nerve/gateway/routes/__init__.py b/nerve/gateway/routes/__init__.py index 1273cb8c..55436e44 100644 --- a/nerve/gateway/routes/__init__.py +++ b/nerve/gateway/routes/__init__.py @@ -35,6 +35,7 @@ codex, workflow_runs, review_loops, + config, ) __all__ = [ @@ -67,4 +68,5 @@ def register_all_routes() -> APIRouter: router.include_router(codex.router) router.include_router(workflow_runs.router) router.include_router(review_loops.router) + router.include_router(config.router) return router diff --git a/nerve/gateway/routes/config.py b/nerve/gateway/routes/config.py new file mode 100644 index 00000000..36bb9657 --- /dev/null +++ b/nerve/gateway/routes/config.py @@ -0,0 +1,56 @@ +"""Config / workspace-sync routes.""" + +from __future__ import annotations + +from fastapi import APIRouter, Depends, HTTPException + +from nerve.gateway.auth import require_auth +from nerve.gateway.routes._deps import get_deps + +router = APIRouter() + + +@router.post("/api/config/sync") +async def sync_workspace_route(user: dict = Depends(require_auth)): + """Pull the workspace from its git remote and apply changes (cron + MCP). + + Fast-forward only; validates the pulled bundle before applying. + """ + import asyncio + + from nerve.config import get_config + from nerve.gateway.server import _cron_service + from nerve.sync_service import _apply_sync, sync_workspace + + config = get_config() + # Hand over the raw configured values: sync_workspace coerces them inside its + # own never-raises guard, so a `workspace` that is null or a list reports a + # 400 rather than a 500 with a stack trace. + result = await asyncio.to_thread( + sync_workspace, + config.workspace, + config.config_dir or config.workspace, + branch=config.workspace_sync.branch, + validate=config.workspace_sync.validate, + strict_env=config.workspace_sync.strict_env, + ) + if not result.ok: + raise HTTPException( + status_code=400, + detail={ + "message": result.message, + "errors": result.validation_errors, + "warnings": result.validation_warnings, + }, + ) + if result.changed: + deps = get_deps() + await _apply_sync(deps.engine, _cron_service) + return { + "ok": True, + "changed": result.changed, + "message": result.message, + "old_rev": result.old_rev, + "new_rev": result.new_rev, + "warnings": result.validation_warnings, + } diff --git a/nerve/gateway/server.py b/nerve/gateway/server.py index 037b1afa..717ded36 100644 --- a/nerve/gateway/server.py +++ b/nerve/gateway/server.py @@ -38,6 +38,7 @@ flush as langfuse_flush, init_langfuse, ) +from nerve.utils.aio import stop_background_task logger = logging.getLogger(__name__) @@ -221,6 +222,8 @@ async def lifespan(app: FastAPI): # Start cron service global _cron_service cron_task = None + ws_sync_task = None + ws_sync_stop = None try: from nerve.cron.service import CronService cron = CronService(config, _engine, db) @@ -339,6 +342,14 @@ async def lifespan(app: FastAPI): except Exception as e: logger.warning("houseofagents artifact cleanup failed: %s", e) + # Periodically pull the workspace from its git remote and apply (opt-in). + if config.workspace_sync.enabled: + from nerve.sync_service import run_periodic_sync + ws_sync_stop = asyncio.Event() + ws_sync_task = asyncio.create_task( + run_periodic_sync(config, _engine, _cron_service, ws_sync_stop) + ) + # Periodic session cleanup. Default cadence is every 6 hours (unchanged); # it tightens to hourly only when the opt-in interactive idle auto-close # (sessions.interactive_archive_after_hours > 0) is enabled and needs finer resolution. @@ -627,6 +638,14 @@ async def _periodic_backup(): # the telegram polling task before we get a chance to stop it cleanly. if telegram_channel: await telegram_channel.stop() + if ws_sync_task: + # Exit through the loop's own stop path rather than cancelling it where + # it stands: a cycle interrupted between the merge and the reload leaves + # the workspace on the new commit with the daemon still running the old + # config. The git phase runs in a worker thread and so is out of reach of + # cancellation either way; the bounded wait is for the reload that + # follows it. Cancellation is the backstop, not the mechanism. + await stop_background_task(ws_sync_task, ws_sync_stop, "Workspace sync") if cron_task: await cron_task.stop() if _review_loop_service is not None: diff --git a/nerve/sync_service.py b/nerve/sync_service.py new file mode 100644 index 00000000..de4c54ec --- /dev/null +++ b/nerve/sync_service.py @@ -0,0 +1,501 @@ +"""Git-backed workspace sync. + +The workspace is a git repository whose remote (on GitHub) is the shared config +repo. Config changes are proposed as PRs, reviewed, and merged there; this module +pulls the merged result onto the instance so it can hot-reload — no restart, no +hand-editing on the box. + +``sync_workspace`` does a guarded ``git pull --ff-only`` and (optionally) +validates the pulled bundle. Applying the change (reloading cron / MCP) is the +caller's job: the in-daemon periodic loop reloads explicitly, while the CLI only +moves the files and leaves a running daemon to pick them up on its next cycle. +""" + +from __future__ import annotations + +import asyncio +import logging +import shutil +import subprocess +import tempfile +import threading +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from nerve.config_validate import ValidationResult + +logger = logging.getLogger(__name__) + +# Bound network git operations so a slow/hung remote can't stall the loop or a +# graceful shutdown indefinitely. +_GIT_TIMEOUT_SECONDS = 120 + +# Two callers drive syncs concurrently — the periodic loop and POST +# /api/config/sync — and git's own ref locks are per-process-pair, not per-caller. +# Overlapping runs produce "cannot lock ref 'HEAD'" reported as an ff-only merge +# failure, which blames fast-forwardability for what is really contention. Serialize +# instead: a manual sync waits for the cycle in flight and then does its own. +_sync_lock = threading.Lock() + +# Prefix for the throwaway validation worktrees, so leftovers from a process that +# died mid-validation can be recognized and swept. +_TMP_WORKTREE_PREFIX = ".nerve-sync-" + +# How old a leftover validation worktree must be before it is assumed abandoned. +# Generous on purpose: a sync in another process (`nerve config sync` while the +# daemon loop runs) owns a directory this sweep must not delete out from under it. +_ABANDONED_WORKTREE_AGE_SECONDS = 3600 + +# Cap on how many paths a diagnostic message lists before summarizing. +_MAX_LISTED_PATHS = 10 + +# Prefix for any git command whose output we quote paths out of. By default git +# C-escapes every non-ASCII byte in a path, so `config/sübmodül` reaches the +# operator as `"config/s\303\274bmod\303\274l"` — and naming the offending path is +# the entire point of the messages below. Shared so a third such call cannot +# quietly forget it. +_QUOTEPATH_OFF = ("-c", "core.quotepath=false") + + +@dataclass +class SyncResult: + ok: bool + changed: bool = False + message: str = "" + old_rev: str = "" + new_rev: str = "" + validation_errors: list[str] = field(default_factory=list) + validation_warnings: list[str] = field(default_factory=list) + + +def _git(args: list[str], cwd: Path) -> subprocess.CompletedProcess: + """Run a git command in ``cwd``; captured, never raises. + + A failed git invocation comes back as a non-zero ``CompletedProcess`` with + the reason on stderr, whatever the cause: git exiting non-zero, git not + being installed at all, ``cwd`` having been deleted underneath us, or the + remote hanging past the timeout. Callers already branch on ``returncode``, + and the one thing they must never have to handle is an exception — the + whole module's contract is that a sync reports failure rather than raising. + + Output is decoded as UTF-8, leniently. git echoes bytes it was given (branch + names, paths, config values, remote error text) and has no obligation to make + them UTF-8, so a strict decode would turn someone else's mojibake into a crash + here. The encoding is pinned rather than left to the locale because the daemon + usually runs under a service manager with ``LC_ALL=C``, where the default would + mangle a remote's perfectly valid UTF-8 error message into unreadable + replacement characters — losing the one thing the operator needs. + """ + try: + return subprocess.run( + ["git", *args], cwd=str(cwd), capture_output=True, text=True, + encoding="utf-8", errors="replace", timeout=_GIT_TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired: + return subprocess.CompletedProcess( + args, 1, "", f"git command timed out after {_GIT_TIMEOUT_SECONDS}s", + ) + except Exception as e: # noqa: BLE001 — git missing, bad cwd, OS refusal… + return subprocess.CompletedProcess(args, 1, "", f"could not run git: {e}") + + +def is_git_repo(path: Path) -> bool: + return (Path(path) / ".git").exists() + + +def _rev(ref: str, cwd: Path) -> str: + r = _git(["rev-parse", ref], cwd) + return r.stdout.strip() if r.returncode == 0 else "" + + +def _config_pathspec(workspace: Path) -> str: + """The git pathspec for the portable config subtree of ``workspace``.""" + from nerve.config import workspace_config_dir + + return str(workspace_config_dir(workspace)) + + +def _local_config_divergence( + workspace: Path, rev: str, +) -> tuple[list[str], list[str]]: + """Local state in the config subtree that the validated rev does not contain. + + Validation runs against a clean detached checkout of ``rev``; the merge lands + in the live working tree. ``git merge --ff-only`` only refuses when the + incoming commit touches a path that is locally modified, so everything else + survives the merge untouched — and was never part of what was checked. A + locally edited or deleted tracked file, a staged-but-uncommitted change, a + tracked file swapped for a symlink pointing out of the repo, an untracked + file: each one makes the bundle on disk something other than the bundle that + passed. + + The sharp end is ``cron/gates/*.py``. Those are imported and executed by the + daemon, the validator deliberately never loads them, and an untracked one is + invisible to the throwaway worktree — so a box whose whole premise is "only + reviewed remote config runs here" would run local unreviewed code while every + sync reported success. Refusing is the only honest answer: sync's guarantee is + about what ends up on disk, and it cannot make that promise about a tree + somebody else is also editing. + + Scoped to the config subtree on purpose. A nerve workspace is also the + agent's working directory, so uncommitted notes and scratch files elsewhere + in it are normal and none of sync's business — and where they *would* break + the fast-forward, git says so itself. + + Returns ``(blocking, warnings)``. + """ + blocking: list[str] = [] + warnings: list[str] = [] + pathspec = _config_pathspec(workspace) + + status = _git([ + *_QUOTEPATH_OFF, "status", "--porcelain", + "--untracked-files=all", "--ignored=matching", "--", pathspec, + ], workspace) + if status.returncode != 0: + # Fail closed: unable to establish that the tree is clean is not the + # same as clean. + return ([ + f"could not check the workspace for local changes: " + f"{status.stderr.strip() or status.stdout.strip()}" + ], warnings) + + for line in status.stdout.splitlines(): + if len(line) < 4: + continue + code, path = line[:2], line[3:] + if code == "!!": + # An ignored file inside a *tracked config subtree* is a layout + # mistake — it is config the shared repo can never carry and no + # reviewer will ever see. Worth saying; not worth refusing a merge + # over, since refusing would be a policy decision about what the + # box is allowed to have locally rather than a statement about + # whether the merge is sound. + warnings.append(f"ignored file inside the tracked config subtree: {path}") + else: + blocking.append(f"{code.strip() or '??'} {path}") + + # A submodule under the config subtree is a third case: `git worktree add` + # does not initialize submodules, so validation saw an empty directory, and + # the fast-forward leaves the live checkout on its old commit. Neither the + # old contents nor the new ones were checked. + tree = _git([*_QUOTEPATH_OFF, "ls-tree", "-r", rev, "--", pathspec], workspace) + if tree.returncode == 0: + for line in tree.stdout.splitlines(): + if line.startswith("160000 "): + sub = line.split("\t", 1)[-1] + warnings.append( + f"submodule {sub!r} in the config subtree was not validated " + f"(a validation checkout does not initialize submodules) and " + f"a fast-forward does not update it" + ) + return blocking, warnings + + +def _describe_paths(paths: list[str]) -> str: + """Join paths for a message, summarizing once the list stops being readable.""" + if len(paths) <= _MAX_LISTED_PATHS: + return ", ".join(paths) + shown = ", ".join(paths[:_MAX_LISTED_PATHS]) + return f"{shown}, +{len(paths) - _MAX_LISTED_PATHS} more" + + +def _sweep_abandoned_worktrees(workspace: Path) -> None: + """Drop validation worktrees left behind by a process that died mid-check. + + ``_validate_rev`` unregisters its worktree in a ``finally``, which covers + every exception but not SIGKILL, a power loss, or a container stop. What + survives is both a directory and a live entry in ``.git/worktrees``, so + ``git worktree prune`` (which only forgets worktrees whose directory is + already gone) will never clear it. On a five-minute sync cadence that grows + without bound, and nothing else ever looks. + + Only clearly abandoned directories are touched — see + ``_ABANDONED_WORKTREE_AGE_SECONDS``. Best-effort throughout: failing to tidy + up is not a reason to fail a sync. + """ + cutoff = time.time() - _ABANDONED_WORKTREE_AGE_SECONDS + try: + candidates = [ + p for p in workspace.parent.iterdir() + if p.name.startswith(_TMP_WORKTREE_PREFIX) and p.is_dir() + and p.stat().st_mtime < cutoff + ] + except OSError: + return + for stale in candidates: + _git(["worktree", "remove", "--force", str(stale / "wt")], workspace) + shutil.rmtree(stale, ignore_errors=True) + if candidates: + # Clears registrations whose directory the loop above just removed. + _git(["worktree", "prune"], workspace) + logger.info( + "Workspace sync: removed %d abandoned validation worktree(s)", + len(candidates), + ) + + +def _validate_rev( + workspace: Path, rev: str, config_dir: Path, strict_env: bool = True, +) -> ValidationResult: + """Validate the config bundle *at a fetched rev* without touching the live + working tree, by checking it out into a throwaway git worktree. + + Returns a :class:`~nerve.config_validate.ValidationResult`. Anything that + goes wrong on the way to a verdict — no room for the worktree, an + unreadable file, a validator that raises — becomes an *error* in that + result rather than an exception: sync fails closed, and never by crashing. + + ``strict_env`` is on by default because this asks a narrower question than + ``nerve config validate`` does. CI is lenient about ``${VAR}`` references it + has no secrets for; here the answer that matters is "can *this* process load + the bundle", and this process is the daemon, with the daemon's environment. + An unset required reference means ``load_config`` will raise on the next + restart, so it has to block the merge. + """ + from nerve.config_validate import ValidationResult, validate_config_bundle + + tmp = wt = None + try: + _sweep_abandoned_worktrees(workspace) + # Keep the temp worktree on the same filesystem as the repo. + tmp = Path(tempfile.mkdtemp( + prefix=_TMP_WORKTREE_PREFIX, dir=str(workspace.parent), + )) + wt = tmp / "wt" + add = _git(["worktree", "add", "--detach", str(wt), rev], workspace) + if add.returncode != 0: + return ValidationResult( + errors=[ + "could not create validation worktree: " + f"{add.stderr.strip() or add.stdout.strip()}" + ] + ) + return validate_config_bundle( + config_dir, workspace_override=wt, strict_env=strict_env, + ) + except Exception as e: # noqa: BLE001 — an unreachable verdict is a failed one + return ValidationResult(errors=[f"validation failed: {type(e).__name__}: {e}"]) + finally: + if wt is not None: + _git(["worktree", "remove", "--force", str(wt)], workspace) + if tmp is not None: + shutil.rmtree(tmp, ignore_errors=True) + + +def sync_workspace( + workspace: Path, + config_dir: Path, + branch: str = "", + validate: bool = True, + strict_env: bool = True, +) -> SyncResult: + """Fetch the workspace remote, validate the fetched bundle, then ff-merge it. + + Crucially the live working tree is only fast-forwarded **after** validation + passes, so an invalid bundle never lands on disk (nothing to be picked up by + a later reload or the next restart). Never raises; returns a + :class:`SyncResult`. + + That guarantee is about the tree the daemon will read, not merely about the + commit that was checked, so a merge is also refused when the live config + subtree has local changes of its own — see + :func:`_local_config_divergence`. ``changed`` reports whether the live tree + actually moved; a refusal leaves it exactly where it was. + + ``strict_env`` treats an unset required ``${VAR}`` reference in the fetched + bundle as invalid — the merge would otherwise leave a checkout the daemon + refuses to load on its next restart. Turn it off only when running somewhere + that legitimately lacks the daemon's environment, e.g. an operator's shell. + + "Never raises" is the whole contract, not an aspiration: the HTTP route + turns anything that escapes into a 500 and the daemon's loop would report a + stack trace instead of a config problem. The guard here is deliberately + broader than the failures currently known — every caller handles a + ``SyncResult``, and none handles an exception. Callers should hand over the + raw configured values and let this function coerce them: doing ``Path(...)`` + on the caller's side puts that conversion outside the guard, where a + ``workspace`` that is ``None`` or a list (a merge conflict resolved badly) + becomes the traceback the contract exists to prevent. + """ + try: + with _sync_lock: + return _sync_workspace( + Path(workspace), Path(config_dir), branch, validate, strict_env, + ) + except Exception as e: # noqa: BLE001 — see the contract above + logger.warning("Workspace sync failed unexpectedly: %s", e, exc_info=True) + return SyncResult(ok=False, message=f"sync failed: {type(e).__name__}: {e}") + + +def _sync_workspace( + workspace: Path, + config_dir: Path, + branch: str, + validate: bool, + strict_env: bool, +) -> SyncResult: + """The fetch → validate → merge sequence. See :func:`sync_workspace`.""" + if not is_git_repo(workspace): + return SyncResult(ok=False, message=f"{workspace} is not a git repository") + + old = _rev("HEAD", workspace) + + fetch = _git(["fetch", "origin", branch] if branch else ["fetch"], workspace) + if fetch.returncode != 0: + return SyncResult( + ok=False, old_rev=old, + message=f"git fetch failed: {fetch.stderr.strip() or fetch.stdout.strip()}", + ) + + target_ref = f"origin/{branch}" if branch else "@{u}" + new = _rev(target_ref, workspace) + if not new: + return SyncResult( + ok=False, old_rev=old, + message=f"could not resolve upstream {target_ref!r} (no tracking branch?)", + ) + + if new == old: + return SyncResult(ok=True, changed=False, old_rev=old, new_rev=new, message="up to date") + + blocking, warnings = _local_config_divergence(workspace, new) + if blocking: + return SyncResult( + ok=False, changed=False, old_rev=old, new_rev=new, + validation_warnings=warnings, + message=( + f"fetched {new[:8]} but the workspace config subtree has local " + f"changes — not applying, because they would survive the " + f"fast-forward without ever having been validated: " + f"{_describe_paths(blocking)}. Commit, discard or push them." + ), + ) + + if validate: + report = _validate_rev(workspace, new, config_dir, strict_env) + # Warnings ride along on the result even when the merge goes ahead. + # Validation can only confirm what it is allowed to look at — it does + # not load the bundle's gate plugins, and unknown keys are tolerated — + # so "no errors" is not "nothing to know about". Applying a bundle whose + # gate type nothing recognizes is how a cron job quietly starts running + # unconditionally, and the only place that is visible is here. + warnings += report.warnings + if report.errors: + return SyncResult( + ok=False, changed=False, old_rev=old, new_rev=new, + validation_errors=report.errors, validation_warnings=warnings, + message=( + f"fetched {new[:8]} but the config bundle is INVALID — not " + f"applying ({len(report.errors)} error(s))" + ), + ) + merge = _git(["merge", "--ff-only", new], workspace) + if merge.returncode != 0: + return SyncResult( + ok=False, old_rev=old, new_rev=new, validation_warnings=warnings, + message=f"ff-only merge failed: {merge.stderr.strip() or merge.stdout.strip()}", + ) + if not validate: + # Reported on every merge, not once at start-up. Validation can be + # switched off in ways that leave no trace — a config key, a CLI flag, an + # env reference exported empty (an empty value means off) — and the + # result is a fast-forward to whatever the remote happens to carry, + # including a bundle the daemon will refuse to load. Each occurrence is + # worth a line. + warnings.append( + "validation is disabled (workspace_sync.validate is off): the bundle " + "now on disk has not been checked" + ) + return SyncResult( + ok=True, changed=True, old_rev=old, new_rev=new, + validation_warnings=warnings, + message=f"updated {old[:8]}→{new[:8]}", + ) + + +async def run_periodic_sync(config, engine, cron_service, stop_event: asyncio.Event) -> None: + """Daemon loop: pull the workspace every ``interval_minutes`` and apply. + + On a successful, changed, valid pull it reloads MCP config and cron so the + merged changes take effect without a restart. Best-effort — a failed cycle is + logged and the loop continues. + + This loop is the only thing that applies a merged change on its own, so + ``interval_minutes`` is the upper bound on how stale an instance can be. + ``POST /api/config/sync`` does the same pull-and-apply on demand. + + Every cycle re-reads the process-wide config object rather than working from + a reference captured before the loop, so the loop can never be the reason a + setting is stuck. **On its own that changes nothing today:** nothing refreshes + that object after start-up — it is loaded once and only replaced by the CLI + at process start — so editing ``workspace_sync`` still requires a restart to + take effect. Once something does refresh it, ``branch``, ``validate``, + ``strict_env``, ``interval_minutes``, the workspace location and turning sync + off will apply from the following cycle with no further work here. Turning + sync *on* will still need a restart regardless: this task is created at + start-up only when sync is already enabled, and nothing creates it later. + """ + from nerve.config import get_config + + cfg = config.workspace_sync + interval = max(1, cfg.interval_minutes) * 60 + enabled = True + logger.info( + "Workspace sync enabled: pulling %s every %d min", + config.workspace, cfg.interval_minutes, + ) + while not stop_event.is_set(): + try: + await asyncio.wait_for(stop_event.wait(), timeout=interval) + break # stop_event set → exit + except asyncio.TimeoutError: + pass # interval elapsed → run a sync + # One guard around the whole cycle, applying included: a loop that dies + # here never syncs again, and nothing restarts it. + try: + config = get_config() + cfg = config.workspace_sync + interval = max(1, cfg.interval_minutes) * 60 + if cfg.enabled != enabled: + enabled = cfg.enabled + logger.info( + "Workspace sync %s by config", "enabled" if enabled else "disabled", + ) + if not enabled: + continue + result = await asyncio.to_thread( + sync_workspace, config.workspace, + config.config_dir or config.workspace, + branch=cfg.branch, validate=cfg.validate, strict_env=cfg.strict_env, + ) + for warning in result.validation_warnings: + logger.warning("Workspace sync: config warning: %s", warning) + if not result.ok: + logger.warning("Workspace sync: %s", result.message) + for err in result.validation_errors: + logger.warning(" config error: %s", err) + continue + if result.changed: + logger.info("Workspace sync: %s — applying", result.message) + await _apply_sync(engine, cron_service) + except Exception as e: # noqa: BLE001 — never let the loop die + logger.warning("Workspace sync cycle failed: %s", e) + continue + + +async def _apply_sync(engine, cron_service) -> None: + """Hot-reload the subsystems affected by a workspace pull.""" + if cron_service is not None: + try: + await cron_service.reload() + except Exception as e: # noqa: BLE001 + logger.warning("cron reload after sync failed: %s", e) + if engine is not None: + try: + await engine.reload_mcp_config() + except Exception as e: # noqa: BLE001 + logger.warning("MCP reload after sync failed: %s", e) diff --git a/tests/test_config_validate.py b/tests/test_config_validate.py index 9018523b..cf6b6564 100644 --- a/tests/test_config_validate.py +++ b/tests/test_config_validate.py @@ -151,6 +151,50 @@ def test_unset_env_is_error_when_strict(self, tmp_path, monkeypatch): assert not result.ok assert any("SECRET_X" in e for e in result.errors) + def test_unset_env_from_machine_layer_only_is_named_as_such( + self, tmp_path, monkeypatch + ): + """A var only this host asks for must be pinned on this host's config — + otherwise a sync refusal reads as a defect in the incoming bundle.""" + monkeypatch.delenv("MACHINE_ONLY_TOKEN", raising=False) + ws = tmp_path / "ws" + ws.mkdir() + _settings(ws, "timezone: UTC\n") + result = validate_config_bundle( + _cfg(tmp_path, "anthropic_api_key: ${MACHINE_ONLY_TOKEN}\n", workspace=ws) + ) + msg = next(i for i in result.info if "MACHINE_ONLY_TOKEN" in i) + assert "only by this machine's" in msg + assert "not by the workspace config" in msg + + def test_unset_env_from_both_layers_is_not_blamed_on_the_machine( + self, tmp_path, monkeypatch + ): + """A var *both* layers reference is not the machine's alone: claiming it + is sends the reader to config.yaml for a variable the portable bundle + needs just as much.""" + monkeypatch.delenv("SHARED_TOKEN", raising=False) + ws = tmp_path / "ws" + ws.mkdir() + _settings(ws, "openai_api_key: ${SHARED_TOKEN}\n") + result = validate_config_bundle( + _cfg(tmp_path, "anthropic_api_key: ${SHARED_TOKEN}\n", workspace=ws) + ) + msg = next(i for i in result.info if "SHARED_TOKEN" in i) + assert "not by the workspace config" not in msg + assert "both the workspace config" in msg + + def test_unset_env_from_workspace_only_is_not_attributed_to_the_machine( + self, tmp_path, monkeypatch + ): + monkeypatch.delenv("WS_ONLY_TOKEN", raising=False) + ws = tmp_path / "ws" + ws.mkdir() + _settings(ws, "anthropic_api_key: ${WS_ONLY_TOKEN}\n") + result = validate_config_bundle(_cfg(tmp_path, "timezone: UTC\n", workspace=ws)) + msg = next(i for i in result.info if "WS_ONLY_TOKEN" in i) + assert "this machine's" not in msg + def test_malformed_cron_is_error(self, tmp_path): ws = tmp_path / "ws" cron = ws / "config" / "cron" diff --git a/tests/test_sync_service.py b/tests/test_sync_service.py new file mode 100644 index 00000000..1c4be711 --- /dev/null +++ b/tests/test_sync_service.py @@ -0,0 +1,980 @@ +"""Tests for git-backed workspace sync.""" + +from __future__ import annotations + +import asyncio +import os +import shutil +import subprocess +import threading +import time +from pathlib import Path +from unittest.mock import AsyncMock + +import pytest + +import nerve.sync_service as sync +from nerve.config import WorkspaceSyncConfig +from nerve.config_validate import ValidationResult +from nerve.sync_service import sync_workspace + + +def _cp(returncode=0, stdout="", stderr=""): + return subprocess.CompletedProcess([], returncode, stdout, stderr) + + +def _valid(*_a, **_k): + return ValidationResult() + + +def _invalid(*errors): + return lambda *_a, **_k: ValidationResult(errors=list(errors)) + + +class _FakeGit: + """Scripts git responses for the fetch → rev-parse → merge flow.""" + + def __init__(self, head, upstream, *, fetch_rc=0, merge_rc=0): + self.head = head + self.upstream = upstream + self.fetch_rc = fetch_rc + self.merge_rc = merge_rc + self.calls = [] + self.merged = False + + def __call__(self, args, cwd): + self.calls.append(args) + if args[0] == "rev-parse": + ref = args[1] + if ref == "HEAD": + return _cp(stdout=self.upstream if self.merged else self.head) + return _cp(stdout=self.upstream) # origin/ or @{u} + if args[0] == "fetch": + return _cp(returncode=self.fetch_rc, stderr="" if not self.fetch_rc else "fetch err") + if args[0] == "merge": + self.merged = True + return _cp(returncode=self.merge_rc, stderr="" if not self.merge_rc else "not ff") + return _cp() + + def did(self, verb): + return any(a and a[0] == verb for a in self.calls) + + +class TestSyncWorkspaceOrchestration: + def _repo(self, tmp_path): + ws = tmp_path / "ws" + (ws / ".git").mkdir(parents=True) + return ws + + def test_not_a_git_repo(self, tmp_path): + result = sync_workspace(tmp_path / "ws", tmp_path / "cfg") + assert not result.ok and "not a git repository" in result.message + + def test_up_to_date(self, tmp_path, monkeypatch): + ws = self._repo(tmp_path) + monkeypatch.setattr(sync, "_git", _FakeGit(head="abc", upstream="abc")) + result = sync_workspace(ws, tmp_path / "cfg") + assert result.ok and not result.changed and result.message == "up to date" + + def test_fetch_failure(self, tmp_path, monkeypatch): + ws = self._repo(tmp_path) + monkeypatch.setattr(sync, "_git", _FakeGit(head="a", upstream="a", fetch_rc=1)) + result = sync_workspace(ws, tmp_path / "cfg") + assert not result.ok and "git fetch failed" in result.message + + def test_no_upstream(self, tmp_path, monkeypatch): + ws = self._repo(tmp_path) + monkeypatch.setattr(sync, "_git", _FakeGit(head="a", upstream="")) + result = sync_workspace(ws, tmp_path / "cfg") + assert not result.ok and "upstream" in result.message + + def test_changed_valid_merges(self, tmp_path, monkeypatch): + ws = self._repo(tmp_path) + fake = _FakeGit(head="aaaaaaaa", upstream="bbbbbbbb") + monkeypatch.setattr(sync, "_git", fake) + monkeypatch.setattr(sync, "_validate_rev", _valid) + result = sync_workspace(ws, tmp_path / "cfg", validate=True) + assert result.ok and result.changed and "updated" in result.message + assert fake.did("merge") # working tree fast-forwarded + + def test_changed_invalid_does_not_merge(self, tmp_path, monkeypatch): + """The core guarantee: an invalid fetched bundle is never merged into the + live working tree.""" + ws = self._repo(tmp_path) + fake = _FakeGit(head="aaaaaaaa", upstream="bbbbbbbb") + monkeypatch.setattr(sync, "_git", fake) + monkeypatch.setattr(sync, "_validate_rev", _invalid("bad backend")) + result = sync_workspace(ws, tmp_path / "cfg", validate=True) + assert not result.ok + # `changed` reports whether the live tree moved, and it did not. + assert not result.changed + assert result.new_rev != result.old_rev # the remote did have something new + assert result.validation_errors == ["bad backend"] + assert not fake.did("merge") # NOT applied — tree untouched + + def test_branch_passed_through(self, tmp_path, monkeypatch): + ws = self._repo(tmp_path) + fake = _FakeGit(head="a", upstream="b") + monkeypatch.setattr(sync, "_git", fake) + monkeypatch.setattr(sync, "_validate_rev", _valid) + sync_workspace(ws, tmp_path / "cfg", branch="main") + assert ["fetch", "origin", "main"] in fake.calls + assert ["rev-parse", "origin/main"] in fake.calls + + def test_ff_merge_failure(self, tmp_path, monkeypatch): + ws = self._repo(tmp_path) + monkeypatch.setattr(sync, "_git", _FakeGit(head="a", upstream="b", merge_rc=1)) + monkeypatch.setattr(sync, "_validate_rev", _valid) + result = sync_workspace(ws, tmp_path / "cfg") + assert not result.ok and "ff-only merge failed" in result.message + + +class _RealGit: + """Drives real ``git`` against local repositories only — never a network + remote. ``_pair`` builds an origin and a clone of it under ``tmp_path``. + """ + + def _git(self, *args, cwd): + return subprocess.run( + ["git", *args], cwd=str(cwd), check=True, + capture_output=True, text=True, + env={"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", + "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t", + "HOME": str(cwd), "PATH": os.environ["PATH"]}, + ) + + def _pair(self, tmp_path, settings="timezone: UTC\n"): + """A local origin repo with a config bundle, plus a clone of it.""" + origin = tmp_path / "origin" + origin.mkdir() + self._git("init", "-b", "main", cwd=origin) + (origin / "config").mkdir() + (origin / "config" / "settings.yaml").write_text(settings) + self._git("add", "-A", cwd=origin) + self._git("commit", "-m", "init", cwd=origin) + ws = tmp_path / "ws" + self._git("clone", str(origin), str(ws), cwd=tmp_path) + return origin, ws + + +@pytest.mark.skipif(not shutil.which("git"), reason="git not available") +class TestSyncWorkspaceRealGit(_RealGit): + def test_end_to_end_pull_valid(self, tmp_path): + origin = tmp_path / "origin" + origin.mkdir() + self._git("init", "-b", "main", cwd=origin) + (origin / "config").mkdir() + (origin / "config" / "settings.yaml").write_text("timezone: UTC\n") + self._git("add", "-A", cwd=origin) + self._git("commit", "-m", "init", cwd=origin) + + ws = tmp_path / "ws" + self._git("clone", str(origin), str(ws), cwd=tmp_path) + + # New commit lands on the remote. + (origin / "config" / "settings.yaml").write_text("timezone: Europe/Berlin\n") + self._git("commit", "-am", "change tz", cwd=origin) + + result = sync_workspace(ws, tmp_path / "cfg", branch="main", validate=True) + assert result.ok and result.changed, result.message + assert "Europe/Berlin" in (ws / "config" / "settings.yaml").read_text() + + def test_end_to_end_invalid_not_applied(self, tmp_path): + origin = tmp_path / "origin" + origin.mkdir() + self._git("init", "-b", "main", cwd=origin) + (origin / "config" / "cron").mkdir(parents=True) + (origin / "config" / "settings.yaml").write_text("timezone: UTC\n") + self._git("add", "-A", cwd=origin) + self._git("commit", "-m", "init", cwd=origin) + + ws = tmp_path / "ws" + self._git("clone", str(origin), str(ws), cwd=tmp_path) + + # Push a broken cron file to the remote (new file → needs add). + (origin / "config" / "cron" / "jobs.yaml").write_text("jobs: [ broken: yaml\n") + self._git("add", "-A", cwd=origin) + self._git("commit", "-m", "break cron", cwd=origin) + + result = sync_workspace(ws, tmp_path / "cfg", branch="main", validate=True) + assert not result.ok and result.validation_errors + # The live working tree was NOT fast-forwarded. + assert not (ws / "config" / "cron" / "jobs.yaml").exists() + + def test_unset_env_ref_blocks_the_merge(self, tmp_path, monkeypatch): + """A bundle the daemon could not load must not reach the working tree. + + ``${VAR}`` (no ``:-default``) is the *required* form: load_config raises + on an unresolved one. Validation is lenient about those by default, + because CI has no secrets — but sync runs in the daemon, with the + daemon's environment, so leniency here just moves the failure to the + next restart, after the checkout has already moved. + """ + monkeypatch.delenv("NERVE_SYNC_TEST_SECRET", raising=False) + origin, ws = self._pair(tmp_path) + (origin / "config" / "settings.yaml").write_text( + "timezone: UTC\ntelegram:\n bot_token: ${NERVE_SYNC_TEST_SECRET}\n" + ) + self._git("commit", "-am", "needs a secret", cwd=origin) + + result = sync_workspace(ws, tmp_path / "cfg", branch="main", validate=True) + assert not result.ok + assert any("NERVE_SYNC_TEST_SECRET" in e for e in result.validation_errors) + assert "NERVE_SYNC_TEST_SECRET" not in ( + ws / "config" / "settings.yaml" + ).read_text() + + def test_env_ref_resolved_in_the_environment_merges(self, tmp_path, monkeypatch): + """The same bundle is fine once the variable is actually set.""" + monkeypatch.setenv("NERVE_SYNC_TEST_SECRET", "s3cret") + origin, ws = self._pair(tmp_path) + (origin / "config" / "settings.yaml").write_text( + "timezone: UTC\ntelegram:\n bot_token: ${NERVE_SYNC_TEST_SECRET}\n" + ) + self._git("commit", "-am", "needs a secret", cwd=origin) + + result = sync_workspace(ws, tmp_path / "cfg", branch="main", validate=True) + assert result.ok and result.changed, result.message + + def test_no_strict_env_lets_an_operator_pull_anyway(self, tmp_path, monkeypatch): + """The escape hatch for a shell that lacks the daemon's environment.""" + monkeypatch.delenv("NERVE_SYNC_TEST_SECRET", raising=False) + origin, ws = self._pair(tmp_path) + (origin / "config" / "settings.yaml").write_text( + "timezone: UTC\ntelegram:\n bot_token: ${NERVE_SYNC_TEST_SECRET}\n" + ) + self._git("commit", "-am", "needs a secret", cwd=origin) + + result = sync_workspace( + ws, tmp_path / "cfg", branch="main", validate=True, strict_env=False, + ) + assert result.ok and result.changed, result.message + + def test_unverifiable_gate_type_merges_but_is_reported(self, tmp_path): + """Validation does not load gate plugins, so an unrecognized gate type + is a warning, not an error — the bundle merges. Sync has to say so: + if nothing registers that type at run time the gate is dropped and the + job runs unconditionally, and this is the only place anyone would see + it before that happens.""" + origin, ws = self._pair(tmp_path) + (origin / "config" / "cron").mkdir() + (origin / "config" / "cron" / "jobs.yaml").write_text( + "jobs:\n" + " - id: nightly\n" + " schedule: '0 3 * * *'\n" + " prompt: go\n" + " run_if:\n" + " - type: not_a_builtin_gate\n" + ) + self._git("add", "-A", cwd=origin) + self._git("commit", "-m", "plugin gate", cwd=origin) + + result = sync_workspace(ws, tmp_path / "cfg", branch="main", validate=True) + assert result.ok and result.changed, result.message + assert any("not_a_builtin_gate" in w for w in result.validation_warnings) + + def test_cli_reports_the_missing_var_and_honors_no_strict_env( + self, tmp_path, monkeypatch, + ): + from click.testing import CliRunner + + import nerve.config as nerve_config + from nerve.cli import main + + monkeypatch.setattr(nerve_config, "_config", None, raising=False) + monkeypatch.delenv("NERVE_SYNC_TEST_SECRET", raising=False) + origin, ws = self._pair(tmp_path) + (origin / "config" / "settings.yaml").write_text( + "timezone: UTC\ntelegram:\n bot_token: ${NERVE_SYNC_TEST_SECRET}\n" + ) + self._git("commit", "-am", "needs a secret", cwd=origin) + config_dir = tmp_path / "cfg" + config_dir.mkdir() + (config_dir / "config.yaml").write_text(f"workspace: {ws}\n") + args = ["-c", str(config_dir), "config", "sync", "--branch", "main"] + + blocked = CliRunner().invoke(main, args) + assert blocked.exit_code == 1, blocked.output + assert "NERVE_SYNC_TEST_SECRET" in blocked.output + assert "Traceback" not in blocked.output + + pulled = CliRunner().invoke(main, [*args, "--no-strict-env"]) + assert pulled.exit_code == 0, pulled.output + + def test_non_utf8_git_output_does_not_raise(self, tmp_path): + """git hands back the bytes it was given; a strict decode would crash.""" + _origin, ws = self._pair(tmp_path) + git_config = ws / ".git" / "config" + git_config.write_bytes( + git_config.read_bytes() + b"\n[nervetest]\n\tval = \xff\xfe\n" + ) + r = sync._git(["config", "--get", "nervetest.val"], ws) + assert r.returncode == 0 and "�" in r.stdout + + def test_merge_without_validation_says_so(self, tmp_path): + """validate=False can be reached from a config key, a CLI flag, or an + env var exported empty. However it happened, the merge that skipped the + check has to leave a trace.""" + origin, ws = self._pair(tmp_path) + (origin / "config" / "settings.yaml").write_text("timezone: Europe/Berlin\n") + self._git("commit", "-am", "change tz", cwd=origin) + + result = sync_workspace(ws, tmp_path / "cfg", branch="main", validate=False) + assert result.ok and result.changed, result.message + assert any("validation is disabled" in w for w in result.validation_warnings) + + +@pytest.mark.skipif(not shutil.which("git"), reason="git not available") +class TestLocalChangesBlockTheMerge(_RealGit): + """Validation judges a clean checkout of the fetched commit; the merge lands + in the live tree. Everything that can differ between the two is covered here, + because each one makes the bundle on disk something other than the bundle + that passed — and ``--ff-only`` only objects when the incoming commit happens + to touch the same path. + """ + + def _upstream_commit(self, origin, text="timezone: Europe/Berlin\n"): + """A remote commit touching only settings.yaml, so a fast-forward of a + workspace dirty *elsewhere* would otherwise succeed.""" + (origin / "config" / "settings.yaml").write_text(text) + self._git("commit", "-am", "change tz", cwd=origin) + + def _sync(self, ws, tmp_path): + return sync_workspace(ws, tmp_path / "cfg", branch="main", validate=True) + + def test_locally_modified_tracked_file_elsewhere_in_subtree(self, tmp_path): + origin, ws = self._pair(tmp_path) + (origin / "config" / "cron").mkdir() + (origin / "config" / "cron" / "jobs.yaml").write_text("jobs: []\n") + self._git("add", "-A", cwd=origin) + self._git("commit", "-m", "add cron", cwd=origin) + self._git("pull", "-q", "--ff-only", "origin", "main", cwd=ws) + self._upstream_commit(origin) + # Dirty a tracked file the incoming commit does not touch, so git itself + # would fast-forward happily. + (ws / "config" / "cron" / "jobs.yaml").write_text("jobs: not-a-list\n") + + result = self._sync(ws, tmp_path) + assert not result.ok and not result.changed + assert "local changes" in result.message + assert "jobs.yaml" in result.message + assert "Europe/Berlin" not in (ws / "config" / "settings.yaml").read_text() + + def test_locally_deleted_tracked_file(self, tmp_path): + origin, ws = self._pair(tmp_path) + (origin / "config" / "extra.yaml").write_text("timezone: UTC\n") + self._git("add", "-A", cwd=origin) + self._git("commit", "-m", "add extra", cwd=origin) + self._git("pull", "-q", "--ff-only", "origin", "main", cwd=ws) + self._upstream_commit(origin) + (ws / "config" / "extra.yaml").unlink() + + result = self._sync(ws, tmp_path) + assert not result.ok and "extra.yaml" in result.message + + def test_staged_but_uncommitted_change(self, tmp_path): + origin, ws = self._pair(tmp_path) + self._upstream_commit(origin) + (ws / "config" / "staged.yaml").write_text("timezone: UTC\n") + self._git("add", "config/staged.yaml", cwd=ws) + + result = self._sync(ws, tmp_path) + assert not result.ok and "staged.yaml" in result.message + + def test_tracked_file_swapped_for_a_symlink_out_of_the_repo(self, tmp_path): + origin, ws = self._pair(tmp_path) + self._upstream_commit(origin) + outside = tmp_path / "outside.yaml" + outside.write_text("timezone: Pacific/Auckland\n") + target = ws / "config" / "settings.yaml" + target.unlink() + target.symlink_to(outside) + + result = self._sync(ws, tmp_path) + assert not result.ok and "settings.yaml" in result.message + + def test_untracked_gate_plugin_blocks_and_never_lands(self, tmp_path): + """The one with teeth. Gate plugins are imported and executed by the + daemon, validation never loads them by design, and an untracked one is + invisible to the validation checkout — so a box that is supposed to run + only reviewed remote config would run this, with sync reporting success. + """ + origin, ws = self._pair(tmp_path) + self._upstream_commit(origin) + gates = ws / "config" / "cron" / "gates" + gates.mkdir(parents=True) + plugin = gates / "unreviewed.py" + plugin.write_text("raise SystemExit('this would have run')\n") + + result = self._sync(ws, tmp_path) + assert not result.ok and not result.changed + assert "unreviewed.py" in result.message + # Nothing was applied, so the operator is not left believing the box is + # running the reviewed bundle. + assert "Europe/Berlin" not in (ws / "config" / "settings.yaml").read_text() + + def test_ignored_file_warns_but_does_not_block(self, tmp_path): + origin, ws = self._pair(tmp_path) + (origin / ".gitignore").write_text("config/*.local.yaml\n") + self._git("add", "-A", cwd=origin) + self._git("commit", "-m", "ignore", cwd=origin) + self._git("pull", "-q", "--ff-only", "origin", "main", cwd=ws) + self._upstream_commit(origin) + (ws / "config" / "secrets.local.yaml").write_text("token: xyz\n") + + result = self._sync(ws, tmp_path) + assert result.ok and result.changed, result.message + assert any("secrets.local.yaml" in w for w in result.validation_warnings) + + def test_submodule_in_the_subtree_is_flagged_as_unvalidated(self, tmp_path): + origin, ws = self._pair(tmp_path) + sha = self._git("rev-parse", "HEAD", cwd=origin).stdout.strip() + self._git( + "update-index", "--add", "--cacheinfo", f"160000,{sha},config/vendored", + cwd=origin, + ) + self._git("commit", "-m", "vendor config", cwd=origin) + + result = self._sync(ws, tmp_path) + assert result.ok, result.message + assert any("vendored" in w and "submodule" in w + for w in result.validation_warnings) + + def test_submodule_with_a_non_ascii_name_is_named_readably(self, tmp_path): + """Naming the path is the whole job of that warning. + + git C-escapes non-ASCII bytes in paths unless told not to, which would + turn the one actionable word in the message into ``\\303\\274`` noise. + """ + origin, ws = self._pair(tmp_path) + sha = self._git("rev-parse", "HEAD", cwd=origin).stdout.strip() + self._git( + "update-index", "--add", "--cacheinfo", f"160000,{sha},config/sübmodül", + cwd=origin, + ) + self._git("commit", "-m", "vendor config", cwd=origin) + + result = self._sync(ws, tmp_path) + assert result.ok, result.message + warning = next( + (w for w in result.validation_warnings if "submodule" in w), "" + ) + assert "config/sübmodül" in warning, warning + assert "\\303" not in warning, warning + + def test_changes_outside_the_config_subtree_do_not_block(self, tmp_path): + """A nerve workspace is also the agent's working directory. Refusing on + any dirt anywhere would refuse on nearly every real box.""" + origin, ws = self._pair(tmp_path) + self._upstream_commit(origin) + (ws / "notes.md").write_text("scratch\n") + (ws / "memory").mkdir() + (ws / "memory" / "today.md").write_text("things\n") + + result = self._sync(ws, tmp_path) + assert result.ok and result.changed, result.message + + def test_unreadable_status_fails_closed(self, tmp_path, monkeypatch): + """Not being able to establish that the tree is clean is not the same as + the tree being clean.""" + origin, ws = self._pair(tmp_path) + self._upstream_commit(origin) + real = sync._git + + def refuse_status(args, cwd): + if "status" in args: + return subprocess.CompletedProcess(args, 128, "", "fatal: nope") + return real(args, cwd) + + monkeypatch.setattr(sync, "_git", refuse_status) + result = self._sync(ws, tmp_path) + assert not result.ok and "local changes" in result.message + assert "Europe/Berlin" not in (ws / "config" / "settings.yaml").read_text() + + +@pytest.mark.skipif(not shutil.which("git"), reason="git not available") +class TestWorktreeHousekeeping(_RealGit): + def test_abandoned_validation_worktree_is_swept(self, tmp_path): + """``_validate_rev`` cleans up in a ``finally``, which covers exceptions + but not SIGKILL. What is left behind is a directory *and* a live entry in + .git/worktrees, so ``git worktree prune`` will never clear it — on a + five-minute cadence that grows without bound.""" + origin, ws = self._pair(tmp_path) + (origin / "config" / "settings.yaml").write_text("timezone: Europe/Berlin\n") + self._git("commit", "-am", "change tz", cwd=origin) + + # A validation worktree from a process that died mid-check. + stale = tmp_path / ".nerve-sync-crashed" + stale.mkdir() + self._git("worktree", "add", "--detach", str(stale / "wt"), "HEAD", cwd=ws) + os.utime(stale, (0, 0)) # older than the abandonment threshold + registrations = ws / ".git" / "worktrees" + assert list(registrations.iterdir()) + + result = sync_workspace(ws, tmp_path / "cfg", branch="main", validate=True) + assert result.ok and result.changed, result.message + assert not stale.exists() + # git drops the whole directory once the last worktree is unregistered. + assert not registrations.exists() or not list(registrations.iterdir()) + + def test_a_recent_worktree_is_left_alone(self, tmp_path): + """A sync running in another process (``nerve config sync`` alongside the + daemon loop) owns a directory this sweep must not delete under it.""" + origin, ws = self._pair(tmp_path) + (origin / "config" / "settings.yaml").write_text("timezone: Europe/Berlin\n") + self._git("commit", "-am", "change tz", cwd=origin) + live = tmp_path / ".nerve-sync-inflight" + live.mkdir() + self._git("worktree", "add", "--detach", str(live / "wt"), "HEAD", cwd=ws) + + result = sync_workspace(ws, tmp_path / "cfg", branch="main", validate=True) + assert result.ok, result.message + assert (live / "wt").exists() + + +class TestConcurrency: + def test_overlapping_syncs_are_serialized(self, tmp_path, monkeypatch): + """The periodic loop and POST /api/config/sync both drive syncs. Left to + race they trip git's own ref locks, and the failure surfaces as an + ff-only merge failure — blaming fast-forwardability for contention.""" + ws = tmp_path / "ws" + (ws / ".git").mkdir(parents=True) + fake = _FakeGit(head="a", upstream="a") + active: list[int] = [] + high_water: list[int] = [] + + def tracking_git(args, cwd): + active.append(1) + high_water.append(len(active)) + time.sleep(0.01) # long enough to overlap if nothing serializes + try: + return fake(args, cwd) + finally: + active.pop() + + monkeypatch.setattr(sync, "_git", tracking_git) + threads = [ + threading.Thread( + target=lambda: sync_workspace(ws, tmp_path / "cfg"), + ) + for _ in range(4) + ] + for t in threads: + t.start() + for t in threads: + t.join(timeout=10) + assert max(high_water) == 1 + + +class TestWorkspaceSyncConfig: + def test_defaults_disabled(self): + c = WorkspaceSyncConfig.from_dict({}) + assert c.enabled is False and c.interval_minutes == 1 and c.validate is True + + def test_from_dict(self): + c = WorkspaceSyncConfig.from_dict( + {"enabled": True, "branch": "main", "interval_minutes": 10, "validate": False} + ) + assert c.enabled and c.branch == "main" and c.interval_minutes == 10 and not c.validate + + def test_string_flags_are_parsed_not_tested_for_truthiness(self): + """``${VAR}`` interpolation yields a string, and ``bool("false")`` is + ``True`` — so a builder that casts eagerly turns every kill switch into + an on switch.""" + c = WorkspaceSyncConfig.from_dict( + {"enabled": "false", "validate": "0", "strict_env": "no", + "interval_minutes": "15"} + ) + assert c.enabled is False and c.validate is False and c.strict_env is False + assert c.interval_minutes == 15 + + def test_env_refs_can_turn_sync_off(self, tmp_path, monkeypatch): + """End to end through load_config: the operator sets the variable to + ``false`` and sync stays off.""" + from nerve.config import load_config + + monkeypatch.setenv("NERVE_SYNC_ENABLED", "false") + monkeypatch.setenv("NERVE_SYNC_VALIDATE", "no") + ws = tmp_path / "ws" + (ws / "config").mkdir(parents=True) + (ws / "config" / "settings.yaml").write_text( + "workspace_sync:\n" + " enabled: ${NERVE_SYNC_ENABLED}\n" + " validate: ${NERVE_SYNC_VALIDATE}\n" + ) + (tmp_path / "config.yaml").write_text(f"workspace: {ws}\n") + + loaded = load_config(tmp_path).workspace_sync + assert loaded.enabled is False and loaded.validate is False + + def test_validation_actually_skipped_when_env_says_off(self, tmp_path, monkeypatch): + """The flag has to reach the code, not just the dataclass: a truthy + ``"false"`` would make sync validate a bundle the operator told it not + to — and the mirror of that mistake is a sync that skips the check.""" + from nerve.config import load_config + + monkeypatch.setenv("NERVE_SYNC_VALIDATE", "false") + ws = tmp_path / "ws" + (ws / "config").mkdir(parents=True) + (ws / "config" / "settings.yaml").write_text( + "workspace_sync:\n validate: ${NERVE_SYNC_VALIDATE}\n" + ) + (tmp_path / "config.yaml").write_text(f"workspace: {ws}\n") + loaded = load_config(tmp_path).workspace_sync + + repo = tmp_path / "repo" + (repo / ".git").mkdir(parents=True) + calls = [] + monkeypatch.setattr(sync, "_git", _FakeGit(head="a", upstream="b")) + monkeypatch.setattr( + sync, "_validate_rev", + lambda *a, **k: calls.append(a) or ValidationResult(), + ) + result = sync_workspace(repo, tmp_path / "cfg", validate=loaded.validate) + assert result.ok and not calls + + +class TestApplySync: + @pytest.mark.asyncio + async def test_apply_triggers_both_reloads(self): + cron, engine = AsyncMock(), AsyncMock() + await sync._apply_sync(engine, cron) + cron.reload.assert_awaited_once() + engine.reload_mcp_config.assert_awaited_once() + + @pytest.mark.asyncio + async def test_apply_survives_reload_error(self): + cron = AsyncMock() + cron.reload.side_effect = RuntimeError("boom") + engine = AsyncMock() + await sync._apply_sync(engine, cron) # must not raise + engine.reload_mcp_config.assert_awaited_once() + + +class TestNeverRaises: + """``sync_workspace`` promises a SyncResult for every outcome. + + Two callers depend on it absolutely: the HTTP route, which has no handler + and would answer 500 with a stack trace, and the CLI, which would print + one. The daemon loop catches, but then logs a traceback where a config + problem belongs. So the guarantee is checked against the operations that + can actually fail, not against the ones someone remembered to wrap. + """ + + def _repo(self, tmp_path): + ws = tmp_path / "ws" + (ws / ".git").mkdir(parents=True) + return ws + + @pytest.mark.parametrize("exc", [ + FileNotFoundError(2, "No such file or directory: 'git'"), # git not installed + PermissionError(13, "Permission denied"), + NotADirectoryError(20, "Not a directory"), # cwd vanished + UnicodeDecodeError("utf-8", b"\xff", 0, 1, "invalid start byte"), + OSError("something the OS refused"), + ]) + def test_subprocess_failures_become_a_result(self, tmp_path, monkeypatch, exc): + ws = self._repo(tmp_path) + + def boom(*_a, **_k): + raise exc + + monkeypatch.setattr(sync.subprocess, "run", boom) + result = sync_workspace(ws, tmp_path / "cfg") + assert not result.ok and "git" in result.message + + def test_git_timeout_becomes_a_result(self, tmp_path, monkeypatch): + def slow(*_a, **_k): + raise subprocess.TimeoutExpired(["git"], 120) + + monkeypatch.setattr(sync.subprocess, "run", slow) + result = sync_workspace(self._repo(tmp_path), tmp_path / "cfg") + assert not result.ok and "timed out" in result.message + + def test_validator_exception_fails_closed(self, tmp_path, monkeypatch): + """The validator reads YAML and builds typed config; both can raise. + A sync that can't reach a verdict must refuse the merge, not crash.""" + import nerve.config_validate as cv + + ws = self._repo(tmp_path) + fake = _FakeGit(head="aaaaaaaa", upstream="bbbbbbbb") + monkeypatch.setattr(sync, "_git", fake) + + def boom(*_a, **_k): + raise RuntimeError("validator blew up") + + monkeypatch.setattr(cv, "validate_config_bundle", boom) + result = sync_workspace(ws, tmp_path / "cfg", validate=True) + assert not result.ok + assert any("validator blew up" in e for e in result.validation_errors) + assert not fake.did("merge") # fail closed + + @pytest.mark.skipif(os.geteuid() == 0, reason="root ignores directory permissions") + def test_unwritable_worktree_parent_fails_closed(self, tmp_path, monkeypatch): + """The throwaway worktree is created next to the repo; a read-only + parent (or a full disk) must not take the daemon down.""" + holder = tmp_path / "holder" + ws = holder / "ws" + (ws / ".git").mkdir(parents=True) + fake = _FakeGit(head="aaaaaaaa", upstream="bbbbbbbb") + monkeypatch.setattr(sync, "_git", fake) + os.chmod(holder, 0o500) + try: + result = sync_workspace(ws, tmp_path / "cfg", validate=True) + finally: + os.chmod(holder, 0o700) + assert not result.ok and result.validation_errors + assert not fake.did("merge") + + def test_unexpected_internal_failure_becomes_a_result(self, tmp_path, monkeypatch): + """The outer guard: the contract covers the whole call, including the + parts nobody anticipated.""" + def boom(*_a, **_k): + raise ValueError("something nobody thought of") + + monkeypatch.setattr(sync, "_rev", boom) + result = sync_workspace(self._repo(tmp_path), tmp_path / "cfg") + assert not result.ok and "something nobody thought of" in result.message + + @pytest.mark.asyncio + async def test_route_reports_400_not_500(self, tmp_path, monkeypatch): + """The end the contract exists for: a broken bundle is a client-visible + config error, not an internal server error.""" + import nerve.gateway.server as srv + from fastapi import HTTPException + + import nerve.gateway.routes.config as route_mod + + ws = self._repo(tmp_path) + fake_cfg = type("C", (), {})() + fake_cfg.workspace = str(ws) + fake_cfg.config_dir = str(tmp_path / "cfg") + fake_cfg.workspace_sync = WorkspaceSyncConfig(enabled=True, validate=True) + monkeypatch.setattr("nerve.config.get_config", lambda: fake_cfg) + monkeypatch.setattr(srv, "_cron_service", None, raising=False) + monkeypatch.setattr(sync, "_git", _FakeGit(head="aaaaaaaa", upstream="bbbbbbbb")) + monkeypatch.setattr( + sync, "_validate_rev", + lambda *_a, **_k: (_ for _ in ()).throw(RuntimeError("kaboom")), + ) + with pytest.raises(HTTPException) as ei: + await route_mod.sync_workspace_route(user={}) + assert ei.value.status_code == 400 + + +class _FakeClock: + """Replaces the loop's interval wait so cycles run without real time passing. + + The loop's only observable use of ``interval_minutes`` is the timeout it + hands to ``asyncio.wait_for``, so that is recorded rather than inferred. + ``cycles`` syncs run, then the stop event is set and the wait reports + "stopped" exactly as the real one would. + """ + + def __init__(self, cycles: int, stop: asyncio.Event): + self.cycles = cycles + self.stop = stop + self.timeouts: list[float] = [] + + async def wait_for(self, coro, timeout=None): + self.timeouts.append(timeout) + coro.close() # the loop passed stop_event.wait(); we never await it + if self.cycles <= 0: + self.stop.set() + return True + self.cycles -= 1 + raise asyncio.TimeoutError + + +def _run_cycles(monkeypatch, cycles): + """Wire a fake clock into the loop and return ``(stop_event, clock)``.""" + stop = asyncio.Event() + clock = _FakeClock(cycles, stop) + monkeypatch.setattr(sync.asyncio, "wait_for", clock.wait_for) + return stop, clock + + +def _sync_double(recorder, then=None): + """A ``sync_workspace`` stand-in that records the keywords it was called with, + then optionally runs ``then()`` to change the world between cycles. + + Takes ``**kwargs`` deliberately: the loop calls ``sync_workspace`` by keyword, + and a double that pins the positional shape turns a new parameter into an + empty recording — swallowed by the loop's own except-and-continue — instead of + a visible failure. + """ + def double(_workspace, _config_dir, **kwargs): + recorder.append(kwargs) + if then is not None: + then() + return sync.SyncResult(ok=True, changed=False) + + return double + + +def _loop_config(**overrides): + cfg = type("C", (), {})() + cfg.workspace = overrides.pop("workspace", "/tmp/ws") + cfg.config_dir = overrides.pop("config_dir", None) + overrides.setdefault("enabled", True) + overrides.setdefault("interval_minutes", 60) + cfg.workspace_sync = WorkspaceSyncConfig(**overrides) + return cfg + + +def _write_config(config_dir, workspace, **sync_keys): + config_dir.mkdir(parents=True, exist_ok=True) + lines = [f"workspace: {workspace}", "workspace_sync:"] + lines += [f" {k}: {v}" for k, v in sync_keys.items()] + (config_dir / "config.yaml").write_text("\n".join(lines) + "\n") + + +class TestPeriodicLoop: + @pytest.mark.asyncio + async def test_stop_event_exits_promptly(self): + stop = asyncio.Event() + stop.set() # already stopped → loop returns without syncing + await asyncio.wait_for( + sync.run_periodic_sync(_loop_config(), AsyncMock(), AsyncMock(), stop), + timeout=2, + ) + + @pytest.mark.asyncio + async def test_editing_config_on_disk_does_not_reach_the_loop( + self, tmp_path, monkeypatch, + ): + """What actually happens today, and what the docs must therefore say. + + The loop re-reads the process-wide config object every cycle, but nothing + refreshes that object after start-up, so an edited config.yaml is never + seen. If this test starts failing because something now reloads the + singleton, that is good news — update the loop's docstring and the + "workspace_sync changes need a restart" paragraph in docs/config.md, + which are written to this behavior. + """ + import nerve.config as nerve_config + + config_dir = tmp_path / "cfg" + _write_config(config_dir, tmp_path / "ws", enabled="true", branch="OLD") + monkeypatch.setattr( + nerve_config, "_config", nerve_config.load_config(config_dir), + ) + started_with = nerve_config.get_config() + + seen: list[dict] = [] + monkeypatch.setattr(sync, "sync_workspace", _sync_double( + seen, + # An operator edits the file between cycles. + then=lambda: _write_config( + config_dir, tmp_path / "ws", enabled="true", branch="NEW", + ), + )) + stop, _clock = _run_cycles(monkeypatch, 3) + await sync.run_periodic_sync(started_with, AsyncMock(), AsyncMock(), stop) + + assert [k["branch"] for k in seen] == ["OLD", "OLD", "OLD"] + assert nerve_config.get_config() is started_with + + @pytest.mark.asyncio + async def test_loop_follows_the_current_config_object(self, monkeypatch): + """The loop holds no snapshot of its own: whatever replaces the process + config is what the next cycle uses. Nothing replaces it today (see the + test above), so this is the guarantee that the loop is not the obstacle. + """ + import nerve.config as nerve_config + + first = _loop_config(branch="old", interval_minutes=60) + monkeypatch.setattr(nerve_config, "_config", first) + seen: list[dict] = [] + monkeypatch.setattr(sync, "sync_workspace", _sync_double( + seen, + then=lambda: nerve_config.set_config(_loop_config( + branch="new", validate=False, strict_env=False, interval_minutes=1, + )), + )) + stop, clock = _run_cycles(monkeypatch, 2) + await sync.run_periodic_sync(first, AsyncMock(), AsyncMock(), stop) + + assert [k["branch"] for k in seen] == ["old", "new"] + assert [k["validate"] for k in seen] == [True, False] + assert [k["strict_env"] for k in seen] == [True, False] + # interval_minutes 60 → 1 shows up as the next wait's timeout. + assert clock.timeouts[:3] == [3600, 3600, 60] + + @pytest.mark.asyncio + async def test_disabling_sync_stops_pulling(self, monkeypatch): + import nerve.config as nerve_config + + first = _loop_config() + monkeypatch.setattr(nerve_config, "_config", first) + calls: list[dict] = [] + monkeypatch.setattr(sync, "sync_workspace", _sync_double( + calls, + then=lambda: nerve_config.set_config(_loop_config(enabled=False)), + )) + stop, _clock = _run_cycles(monkeypatch, 3) + await sync.run_periodic_sync(first, AsyncMock(), AsyncMock(), stop) + assert len(calls) == 1 # first cycle only; then the flag went false + + @pytest.mark.asyncio + async def test_unreadable_config_does_not_kill_the_loop(self, monkeypatch): + """Re-reading is part of every cycle, so it must not be able to end the + task — a loop that dies silently never syncs again.""" + def boom(): + raise RuntimeError("config gone") + + monkeypatch.setattr("nerve.config.get_config", boom) + calls: list[dict] = [] + monkeypatch.setattr(sync, "sync_workspace", _sync_double(calls)) + stop, clock = _run_cycles(monkeypatch, 3) + await sync.run_periodic_sync(_loop_config(), AsyncMock(), AsyncMock(), stop) + assert clock.cycles == 0 and not calls # kept looping, synced nothing + + @pytest.mark.asyncio + async def test_failure_while_applying_does_not_kill_the_loop(self, monkeypatch): + """Everything a cycle does is inside the guard, applying included.""" + import nerve.config as nerve_config + + monkeypatch.setattr(nerve_config, "_config", _loop_config()) + calls: list[dict] = [] + + def double(_workspace, _config_dir, **kwargs): + calls.append(kwargs) + return sync.SyncResult(ok=True, changed=True, message="updated") + + monkeypatch.setattr(sync, "sync_workspace", double) + + async def explode(*_a, **_k): + raise RuntimeError("reload machinery is broken") + + monkeypatch.setattr(sync, "_apply_sync", explode) + stop, clock = _run_cycles(monkeypatch, 3) + await sync.run_periodic_sync( + nerve_config.get_config(), AsyncMock(), AsyncMock(), stop, + ) + assert clock.cycles == 0 and len(calls) == 3 + + +class TestSyncRoute: + @pytest.mark.asyncio + async def test_route_400_on_invalid(self, monkeypatch): + import nerve.gateway.server as srv + from fastapi import HTTPException + + import nerve.gateway.routes.config as route_mod + from nerve.sync_service import SyncResult + + fake_cfg = type("C", (), {})() + fake_cfg.workspace = "/tmp/ws" + fake_cfg.config_dir = "/tmp/cfg" + fake_cfg.workspace_sync = WorkspaceSyncConfig(enabled=True) + monkeypatch.setattr("nerve.config.get_config", lambda: fake_cfg) + monkeypatch.setattr(srv, "_cron_service", None, raising=False) + # The route imports sync_workspace from nerve.sync_service at call time. + monkeypatch.setattr( + sync, "sync_workspace", + lambda *a, **k: SyncResult(ok=False, message="bad", validation_errors=["e"]), + ) + with pytest.raises(HTTPException) as ei: + await route_mod.sync_workspace_route(user={}) + assert ei.value.status_code == 400