refactor(node): split ops.py into the ops package

Each section of ops.py becomes a module of meshbay_node/ops/ (core,
node_toml, members, chat, groups, roots, files, settings, apps), cut as
text; ops/__init__.py keeps the docstring and re-exports every name, so
`ops.<name>` is unchanged for every caller. Logger name unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 285 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/ops/roots.py b/packages/meshbay-node/src/meshbay_node/ops/roots.py
new file mode 100644
index 0000000..e3e2781
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/ops/roots.py
@@ -0,0 +1,285 @@
+"""A group's directories: adding, removing, changing and ejecting them."""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import asdict
+from pathlib import Path
+
+from meshbay_node.config import DEFAULT_CONFIG_PATH
+from meshbay_node.ops.core import OpError, _config, _group_ctx, _roster
+from meshbay_node.ops.node_toml import _insert_roots_block, _remove_roots_block, _update_root_field
+from meshbay_node.roots import RootError, RootSet, off_disk
+
+log = logging.getLogger("meshbay_node.ops")
+
+
+async def add_root(state: dict, group_id: str, path: str, *,
+                   name: str = "", kind: str = "generic",
+                   writable: bool = False,
+                   removable: bool = False) -> dict:
+    """
+    Add a directory to a group, refusing anything ambiguous.
+
+    Validated against the group's existing roots *before* being written, so a
+    config that would be refused at startup is refused here instead — where the
+    operator is watching and can fix it.
+    """
+    config = _config(state)
+    cfg = next((g for g in config.groups if g.id == group_id), None)
+    if cfg is None:
+        raise OpError("Group not configured on this node", status=404)
+
+    specs = [asdict(r) for r in cfg.roots]
+    specs.append({"path": path, "name": name, "kind": kind,
+                  "writable": writable, "removable": removable})
+    try:
+        built = RootSet.build(specs)
+    except RootError as e:
+        raise OpError(str(e)) from e
+
+    added = built.roots[-1]
+
+    try:
+        added.path.mkdir(parents=True, exist_ok=True)
+    except OSError as e:
+        raise OpError(f"Cannot create {added.path}: {e}") from e
+
+    conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH)
+    root_block = f'  [[groups.roots]]\n  path   = "{added.path.as_posix()}"'
+    if name:
+        root_block += f'\n  name   = "{added.name}"'
+    if kind != "generic":
+        root_block += f'\n  kind   = "{added.kind}"'
+    if writable:
+        root_block += '\n  writable = true'
+    if removable:
+        root_block += '\n  removable = true'
+    _insert_roots_block(conf_path, group_id, root_block)
+
+    from meshbay_node.config import RootSpec
+    cfg.roots.append(RootSpec(
+        path=str(added.path), name=added.name, kind=added.kind,
+        writable=added.writable, removable=added.removable))
+
+    # Deliberately *not* mutating the live RootSet in place.
+    #
+    # `DirectoryIndexer.retarget` decides what to scan by diffing the names it
+    # already has against the ones it is given — so handing it the same object,
+    # edited, means the new root is in both sides of the comparison and is
+    # never scanned. It would appear in the table and stay permanently empty.
+    # `_reload_config_inner` diffs the same way and would likewise conclude
+    # nothing changed. The caller reloads instead, which builds a fresh set
+    # from the file this just wrote.
+    #
+    # `built` is that set, computed here only to validate and to answer with;
+    # what the node serves comes from the reload.
+    log.info("Root added: %s → group %s", added.name, group_id[:8])
+    return {"status": "added", "name": added.name, "path": str(added.path),
+            "group_id": group_id, "roots": built.describe()}
+
+
+async def remove_root(state: dict, group_id: str, root_name: str) -> dict:
+    """Remove a named root from a group. At least one root must remain."""
+    config = _config(state)
+    cfg = next((g for g in config.groups if g.id == group_id), None)
+    if cfg is None:
+        raise OpError("Group not configured on this node", status=404)
+
+    from meshbay_common.paths import fold
+
+    from meshbay_node.roots import derive_name
+    target = fold(root_name)
+    match_idx = None
+    for i, r in enumerate(cfg.roots):
+        try:
+            rname = r.name or derive_name(Path(r.path).expanduser().resolve())
+        except Exception:
+            continue
+        if fold(rname) == target:
+            match_idx = i
+            break
+
+    if match_idx is None:
+        raise OpError(f"No root named {root_name!r} in this group", status=404)
+    if len(cfg.roots) < 2:
+        raise OpError("Cannot remove the only root", status=400)
+
+    removed = cfg.roots[match_idx]
+    resolved = str(Path(removed.path).expanduser().resolve())
+
+    conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH)
+    _remove_roots_block(conf_path, group_id, resolved)
+
+    cfg.roots.pop(match_idx)
+
+    # Not mutating the live set here either — see `add_root`. Dropping the
+    # root from it would leave `retarget` unable to tell that its entries
+    # should go, so the removed directory's files would stay in the index.
+    #
+    # Built from the config this just edited, and never returned empty: an
+    # empty list is a *valid answer* meaning "this group has no directories",
+    # which the client cannot tell from "the node could not say" — it would
+    # blank the operator's table on an op that succeeded.
+    result_roots = RootSet.build([asdict(r) for r in cfg.roots]).describe()
+
+    log.info("Root removed: %s from group %s", root_name, group_id[:8])
+    return {"status": "removed", "name": root_name, "group_id": group_id,
+            "roots": result_roots}
+
+
+async def update_root(state: dict, group_id: str, root_name: str, *,
+                      writable: bool | None = None,
+                      removable: bool | None = None) -> dict:
+    """Toggle writable/removable on an existing root without removing it."""
+    config = _config(state)
+    cfg = next((g for g in config.groups if g.id == group_id), None)
+    if cfg is None:
+        raise OpError("Group not configured on this node", status=404)
+
+    from meshbay_common.paths import fold
+
+    from meshbay_node.roots import RootSet
+    target = fold(root_name)
+    match = None
+    for r in cfg.roots:
+        rname = r.name or str(Path(r.path).name)
+        if fold(rname) == target:
+            match = r
+            break
+    if match is None:
+        raise OpError(f"No root named {root_name!r} in this group", status=404)
+
+    changed = False
+    if writable is not None and match.writable != writable:
+        match.writable = writable
+        changed = True
+    if removable is not None and match.removable != removable:
+        match.removable = removable
+        changed = True
+
+    if not changed:
+        specs = [asdict(r) for r in cfg.roots]
+        built = RootSet.build(specs)
+        return {"status": "unchanged", "name": root_name, "group_id": group_id,
+                "roots": built.describe()}
+
+    conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH)
+    _update_root_field(conf_path, group_id, str(Path(match.path).expanduser().resolve()),
+                       writable=match.writable, removable=match.removable)
+
+    # Update the live RootSet so GET /api/groups returns correct data
+    # immediately, without waiting for the async reload to finish.
+    live_roots: RootSet | None = state.get("groups_ctx", {}).get(
+        group_id, {}).get("roots")
+    if live_roots:
+        for lr in live_roots.roots:
+            lr_name = lr.name or str(Path(lr.path).name)
+            if fold(lr_name) == target:
+                if writable is not None:
+                    lr.writable = writable
+                if removable is not None:
+                    lr.removable = removable
+                break
+
+    # Built from config when there is no live set, never returned empty: an
+    # empty list is a *valid answer* meaning "this group has no directories",
+    # and the client cannot tell it from "the node could not say". It would
+    # blank the operator's table on an op that succeeded.
+    result_roots = (live_roots.describe() if live_roots
+                    else RootSet.build([asdict(r) for r in cfg.roots]).describe())
+
+    log.info("Root updated: %s (writable=%s, removable=%s) in group %s",
+             root_name, match.writable, match.removable, group_id[:8])
+    return {"status": "updated", "name": root_name, "group_id": group_id,
+            "roots": result_roots}
+
+
+async def eject_root(state: dict, group_id: str, root_name: str) -> dict:
+    """Mark a removable root as ejected so the operator can safely unplug."""
+    config = _config(state)
+    cfg = next((g for g in config.groups if g.id == group_id), None)
+    if cfg is None:
+        raise OpError("Group not configured on this node", status=404)
+
+    from meshbay_common.paths import fold
+    target = fold(root_name)
+    ctx = _group_ctx(state, group_id)
+    roots: RootSet | None = ctx.get("roots")
+    if not roots:
+        raise OpError("Group has no roots", status=503)
+
+    root = None
+    for r in roots:
+        if fold(r.name) == target:
+            root = r
+            break
+    if root is None:
+        raise OpError(f"No root named {root_name!r} in this group", status=404)
+    if not root.removable:
+        raise OpError(f"Root {root_name!r} is not marked as removable", status=400)
+    if root.ejected:
+        return {"status": "already_ejected", "name": root_name,
+                "group_id": group_id, "roots": roots.describe()}
+
+    # The indexer stops its watchdog and freezes the entries; it holds the same
+    # RootSet object, but the flags are set here too so a context whose indexer
+    # was replaced by a retarget cannot be left disagreeing with the roster.
+    indexer = state.get("indexers", {}).get(group_id)
+    if indexer:
+        indexer.eject_root(root_name)
+    root.ejected = True
+    root.available = False
+
+    await _roster(state).set_root_ejected(
+        group_id, root_name, True, set_by=state.get("node_user_id", ""))
+
+    log.info("Root ejected: %s from group %s", root_name, group_id[:8])
+    return {"status": "ejected", "name": root_name, "group_id": group_id,
+            "roots": roots.describe()}
+
+
+async def plug_root(state: dict, group_id: str, root_name: str) -> dict:
+    """Re-enable an ejected root after the device is plugged back in."""
+    config = _config(state)
+    cfg = next((g for g in config.groups if g.id == group_id), None)
+    if cfg is None:
+        raise OpError("Group not configured on this node", status=404)
+
+    from meshbay_common.paths import fold
+    target = fold(root_name)
+    ctx = _group_ctx(state, group_id)
+    roots: RootSet | None = ctx.get("roots")
+    if not roots:
+        raise OpError("Group has no roots", status=503)
+
+    root = None
+    for r in roots:
+        if fold(r.name) == target:
+            root = r
+            break
+    if root is None:
+        raise OpError(f"No root named {root_name!r} in this group", status=404)
+    if not root.ejected:
+        return {"status": "already_plugged", "name": root_name,
+                "group_id": group_id, "roots": roots.describe()}
+    if not await off_disk(roots, root.is_live):
+        raise OpError(
+            f"Directory not found: {root.path}. Is the device connected?",
+            status=409)
+
+    # Persisted before the rescan, which can take minutes on a large library:
+    # a crash halfway through must leave the root plugged, not ejected with
+    # entries half rebuilt.
+    await _roster(state).set_root_ejected(
+        group_id, root_name, False, set_by=state.get("node_user_id", ""))
+
+    indexer = state.get("indexers", {}).get(group_id)
+    if indexer:
+        await indexer.plug_root(root_name)
+    root.ejected = False
+    root.available = await off_disk(roots, root.is_live)
+
+    log.info("Root plugged: %s in group %s", root_name, group_id[:8])
+    return {"status": "plugged", "name": root_name, "group_id": group_id,
+            "roots": roots.describe()}