Merge branch 'refactor/groups-phase1'

Groups refactor, phases 1-3.

The root model replaces the old `upload` flag and group-wide `member_upload`
with per-root `writable`/`removable`/`ejected`, carried by a `RootSet` that
both front doors — the loopback API and signed MNP — reach through the same
`ops` functions. MNP goes to 1.1, additively: the roots table now rides on
`index_delta`, so a root added, removed, ejected or plugged reaches every
connected client instead of only whoever reloaded.

The group UI becomes a plugin architecture: an application is a registry
entry in `apps.js` plus its own files, with directories stored generically
by `ops.set_app_directories` under whatever the app is called. A reference
application, hidden behind `?dev=1`, is what makes that claim testable —
adding it is what found the two places still naming apps by hand.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011pvMdvLBG92jyhvD5pD6us

1 files changed, 423 insertions, 112 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
index 4c20c2a..a10504e 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -362,7 +362,11 @@ async def list_groups(state: dict) -> dict:
             "has_gek": bool(ctx.get("gek")),
             "file_count": idx.count if idx else 0,
             "index_version": idx.version if idx else 0,
-            "roots": roots.describe() if roots else [],
+            # With paths: this answers the loopback API, which is the
+            # operator's own channel. `meshbay-node root list` printed "?" for
+            # every directory without it — it was reading a field the member
+            # form of this deliberately omits.
+            "roots": roots.describe(with_paths=True) if roots else [],
             "peers": sum(1 for p in peers.values() if p.get("group_id") == gid),
         })
     roster = state.get("roster")
@@ -390,7 +394,7 @@ async def list_groups(state: dict) -> dict:
 
 
 async def attach_group(state: dict, name: str, shared_dir: str,
-                       upload_dir: str = "") -> dict:
+                       writable: bool = True) -> dict:
     """
     Write a new [[groups]] block into node.toml.
 
@@ -429,31 +433,24 @@ async def attach_group(state: dict, name: str, shared_dir: str,
         raise OpError(f"Cannot create {path}: {e}") from e
 
     conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH)
-    # Appended as text rather than re-serialised: node.toml is hand-written and
-    # full of comments explaining decisions, and a round trip through a TOML
-    # writer would throw all of that away.
     join_policy = group.get("join_policy", "invite")
     block = (f'\n[[groups]]\n'
              f'id          = "{group["id"]}"\n'
              f'name        = "{group["name"]}"\n'
              f'visibility  = "{group.get("visibility", "private")}"\n'
              f'join_policy = "{join_policy}"\n')
-    separate_upload = False
-    if upload_dir:
-        upload_path = Path(upload_dir).expanduser().resolve()
-        if upload_path != path.resolve():
-            separate_upload = True
-            try:
-                upload_path.mkdir(parents=True, exist_ok=True)
-            except OSError as e:
-                raise OpError(f"Cannot create {upload_path}: {e}") from e
-            block += f'upload_dir  = "{upload_path.as_posix()}"\n'
+    # No `upload_dir` here. `GroupConfig.__post_init__` still *reads* it, so an
+    # existing node.toml keeps working — but what it does on read is force every
+    # other root read-only and append that path as the one writable one, which
+    # is the model this refactor replaced. Writing it into a group created
+    # today would mean two mechanisms deciding the same thing, one of them
+    # invisible: `group add --dir X --writable --upload-dir Y` silently made X
+    # read-only. A second writable directory is `root add <path> --writable`.
     block += (f'\n  [[groups.roots]]\n'
               # Forward slashes: a Windows path in a TOML basic string is a
               # parse error (`\U`, `\a`, ... are escapes). pathlib reads `/`.
-              f'  path   = "{path.as_posix()}"\n')
-    if not separate_upload:
-        block += f'  upload = true\n'
+              f'  path     = "{path.as_posix()}"\n'
+              f'  writable = {"true" if writable else "false"}\n')
     try:
         with conf_path.open("a", encoding="utf-8", newline="\n") as f:
             f.write(block)
@@ -462,9 +459,8 @@ async def attach_group(state: dict, name: str, shared_dir: str,
 
     result = {"group_id": group["id"], "name": group["name"],
               "shared_dir": str(path), "config": str(conf_path),
+              "writable": writable,
               "note": "restart the node to pick it up"}
-    if separate_upload:
-        result["upload_dir"] = str(upload_path)
     return result
 
 
@@ -636,12 +632,13 @@ def _remove_roots_block(conf_path: Path, group_id: str,
                     conf_path.write_text("\n".join(new_lines), encoding="utf-8", newline="\n")
                     return
 
-    raise OpError(f"Root path not found in config", status=404)
+    raise OpError("Root path not found in config", status=404)
 
 
 async def add_root(state: dict, group_id: str, path: str, *,
                    name: str = "", kind: str = "generic",
-                   upload: bool = False) -> dict:
+                   writable: bool = False,
+                   removable: bool = False) -> dict:
     """
     Add a directory to a group, refusing anything ambiguous.
 
@@ -655,7 +652,8 @@ async def add_root(state: dict, group_id: str, path: str, *,
         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, "upload": upload})
+    specs.append({"path": path, "name": name, "kind": kind,
+                  "writable": writable, "removable": removable})
     try:
         built = RootSet.build(specs)
     except RootError as e:
@@ -674,15 +672,29 @@ async def add_root(state: dict, group_id: str, path: str, *,
         root_block += f'\n  name   = "{added.name}"'
     if kind != "generic":
         root_block += f'\n  kind   = "{added.kind}"'
-    if upload:
-        root_block += f'\n  upload = true'
+    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,
-        upload=added.upload, direct=added.direct))
+        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()}
@@ -714,25 +726,245 @@ async def remove_root(state: dict, group_id: str, root_name: str) -> dict:
         raise OpError("Cannot remove the only root", status=400)
 
     removed = cfg.roots[match_idx]
-    if removed.upload:
-        raise OpError(
-            "Cannot remove the upload root — file uploads and chat "
-            "attachments are stored there", status=400)
     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)
-    remaining = [asdict(r) for r in cfg.roots]
-    try:
-        built = RootSet.build(remaining)
-    except RootError:
-        built = None
+
+    # 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": built.describe() if built else []}
+            "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 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 = 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()}
+
+
+def _update_root_field(conf_path: Path, group_id: str,
+                       resolved_path: str, *,
+                       writable: bool, removable: bool) -> None:
+    """Update writable/removable fields on a root in node.toml."""
+    text = conf_path.read_text(encoding="utf-8")
+    lines = text.split("\n")
+
+    rng = _find_group_range(lines, group_id)
+    if rng is None:
+        raise OpError(f"Group {group_id[:8]} not found in {conf_path}")
+
+    start, end = rng
+    path_re = re.compile(r'^\s*path\s*=\s*"([^"]*)"')
+    writable_re = re.compile(r'^\s*(writable|upload)\s*=')
+    removable_re = re.compile(r'^\s*removable\s*=')
+    roots_starts: list[int] = []
+    for i in range(start + 1, end):
+        if lines[i].strip() == "[[groups.roots]]":
+            roots_starts.append(i)
+
+    for j, rs in enumerate(roots_starts):
+        rs_end = roots_starts[j + 1] if j + 1 < len(roots_starts) else end
+        found_path = False
+        for k in range(rs, rs_end):
+            m = path_re.match(lines[k])
+            if m:
+                try:
+                    p = str(Path(m.group(1)).expanduser().resolve())
+                except OSError:
+                    continue
+                if p == resolved_path:
+                    found_path = True
+                    break
+        if not found_path:
+            continue
+
+        writable_idx = None
+        removable_idx = None
+        for k in range(rs, rs_end):
+            if writable_re.match(lines[k]):
+                writable_idx = k
+            if removable_re.match(lines[k]):
+                removable_idx = k
+
+        if writable_idx is not None:
+            lines[writable_idx] = f"  writable = {'true' if writable else 'false'}"
+        else:
+            lines.insert(rs_end, f"  writable = {'true' if writable else 'false'}")
+            if removable_idx is not None and removable_idx >= rs_end:
+                removable_idx += 1
+            rs_end += 1
+
+        if removable_idx is not None:
+            lines[removable_idx] = f"  removable = {'true' if removable else 'false'}"
+        else:
+            lines.insert(rs_end, f"  removable = {'true' if removable else 'false'}")
+
+        conf_path.write_text("\n".join(lines), encoding="utf-8", newline="\n")
+        return
+
+    raise OpError("Root path not found in config", status=404)
 
 
 # ── Files ────────────────────────────────────────────────────────────────────
@@ -806,26 +1038,6 @@ async def clear_denylist(state: dict, *, subject: str = "") -> dict:
     return {"status": "cleared", "removed": removed, "subject": subject or "all"}
 
 
-# ── Upload policy ───────────────────────────────────────────────────────────
-
-async def set_member_upload(state: dict, group_id: str, allowed: bool) -> dict:
-    """
-    Turn uploading by ordinary members on or off.
-
-    The setting lives on the node (roster.db), not on the hub and not in
-    node.toml — changing it must not rewrite the operator's config file,
-    and must not need a restart.
-    """
-    roster = _roster(state)
-    ctx = _group_ctx(state, group_id)
-    await roster.set_member_upload(group_id, allowed,
-                                   set_by=state.get("node_user_id", ""))
-    ctx["member_upload"] = allowed
-    log.info("Upload policy: %s for group %s", "on" if allowed else "off",
-             group_id[:8])
-    return {"allowed": allowed, "group_id": group_id}
-
-
 # ── Node settings ────────────────────────────────────────────────────────────
 
 async def get_node_settings(state: dict) -> dict:
@@ -924,12 +1136,16 @@ async def set_enabled_apps(state: dict, group_id: str, apps: list[str]) -> dict:
     """
     Which group "applications" (Chat, Files, ...) are shown to members.
 
-    Same shape as `set_member_upload`: lives on the node (roster.db), takes
+    Same shape as other signed ops: lives on the node (roster.db), takes
     effect without a restart, and is signed by the operator (webrtc_server.py
     checks the caller's own admin-authority allow-list before this runs).
     """
     roster = _roster(state)
     ctx = _group_ctx(state, group_id)
+    # See the same guard in webrtc_server._do_apps_enabled: Files cannot be
+    # turned off, and both writers put it at the front so the two agree.
+    if "files" not in apps:
+        apps = ["files"] + list(apps)
     await roster.set_enabled_apps(group_id, apps,
                                   set_by=state.get("node_user_id", ""))
     ctx["enabled_apps"] = apps
@@ -946,7 +1162,7 @@ async def set_tmdb_config(state: dict, token: str | None = None,
     and in what language it queries TMDB (docs/mediacenter.md §5.5).
 
     Node-wide (roster.py group_settings, group_id="") rather than per-group
-    like set_member_upload/set_enabled_apps: the token and the shared-cache
+    like set_enabled_apps: the token and the shared-cache
     language are one operator's budget and one credential, not a per-group
     or per-viewer concern. Whether TMDB is used *at all* is the per-group
     decision set_tmdb_enabled below makes instead. `token=""` explicitly
@@ -1007,76 +1223,171 @@ async def set_musicbrainz_enabled(state: dict, group_id: str, enabled: bool) ->
     return {"enabled": enabled, "group_id": group_id}
 
 
-async def set_video_root(state: dict, group_id: str, path: str) -> dict:
+# ── App directories ──────────────────────────────────────────────────────────
+
+def _validate_app_dirs(state: dict, group_id: str, paths: list[str], *,
+                       require_writable: bool) -> list[str]:
     """
-    Which folder (possibly a subfolder of a shared root) is the Videos app's
-    entry point for this group. Same shape as set_enabled_apps: lives on the
-    node (roster.db), takes effect without a restart, signed by the operator.
-    `path=""` clears it — the Videos tab then asks for one to be chosen
-    before anything (including TMDB enrichment, docs/mediacenter.md §5.2)
-    runs, rather than defaulting to the whole shared index.
+    Every path an app is pointed at must live inside one of the group's roots.
+
+    The per-app setters this replaces validated nothing: a typo, or a path left
+    behind by a root that was removed, was stored and then quietly matched no
+    entry — an app showing an empty tab with no way to tell "misconfigured"
+    from "no files yet". Refusing at the point of setting is the only moment
+    the operator is present to be told.
 
-    A non-empty path fires (never awaits) a sweep of whatever that folder
-    already contains: the ordinary per-change enrichment path only ever
-    looks at files new since the last broadcast, so anything already sitting
-    in a folder before it became the video_root would otherwise never be
-    picked up.
+    Not `RootSet.resolve()`, deliberately: that also refuses a directory whose
+    root is currently *unavailable*, and an operator must be able to configure
+    a library on a drive they have unplugged. What is checked here is the
+    shape — inside a named root, no traversal — which does not change with
+    what happens to be mounted.
     """
-    roster = _roster(state)
-    ctx = _group_ctx(state, group_id)
-    await roster.set_video_root(group_id, path, set_by=state.get("node_user_id", ""))
-    ctx["video_root"] = path
-    log.info("Videos root for group %s: %r", group_id[:8], path)
-    if path:
-        enrich_fn = state.get("enrich_video_root_fn")
-        if enrich_fn:
-            asyncio.ensure_future(enrich_fn(group_id))
-    return {"path": path, "group_id": group_id}
+    roots: RootSet | None = _group_ctx(state, group_id).get("roots")
+    if roots is None:
+        raise OpError("Group has no roots", status=503)
 
+    clean: list[str] = []
+    for raw in paths:
+        path = str(raw or "").strip().strip("/")
+        if not path:
+            continue
+        if ".." in path.split("/"):
+            raise OpError(f"{path!r} is not a directory inside this group",
+                          status=400)
+        found = roots.split(path)
+        if found is None:
+            raise OpError(
+                f"{path!r} is not inside any of this group's shared "
+                f"directories", status=400,
+                extra={"available": roots.names})
+        root, _tail = found
+        if require_writable and not root.writable:
+            raise OpError(
+                f"{root.name!r} is read-only, and this setting needs a "
+                f"directory that accepts uploads", status=400)
+        clean.append(path)
+    return sorted(set(clean))
 
-async def set_audio_root(state: dict, group_id: str, path: str) -> dict:
+
+async def set_app_directories(state: dict, group_id: str, app_key: str,
+                              paths: list[str], *,
+                              require_writable: bool = False) -> dict:
     """
-    Same shape as set_video_root above — the Music app's own entry point,
-    added later (docs/musicbay.md's original "no root, works over the
-    whole shared tree" simplification didn't hold up against a real messy
-    library). `path=""` clears it — the Music tab then asks for one to be
-    chosen before anything (including tag/cover enrichment) runs, rather
-    than defaulting to the whole shared index.
+    Which folder(s) inside the group's shared roots an application works over.
+
+    One function for every app, keyed by the app's own name: adding an
+    application is a registry entry and a settings component, not another
+    near-identical op here. It replaces `set_video_root`, `set_audio_root` and
+    `set_photo_roots`, which differed only in the key they wrote and whether
+    they took a string or a list.
+
+    Empty means nothing configured, which every app reads as "show nothing
+    until an operator has chosen" — never "the whole group index". Pointing an
+    app at the whole library is a decision, not a default nobody made.
+
+    A change always fires (never awaits) a sweep of what the new directories
+    already contain: the ordinary per-change enrichment path only looks at
+    entries new since the last broadcast, so files already sitting in a folder
+    when it was chosen would otherwise never be picked up.
     """
     roster = _roster(state)
     ctx = _group_ctx(state, group_id)
-    await roster.set_audio_root(group_id, path, set_by=state.get("node_user_id", ""))
-    ctx["audio_root"] = path
-    log.info("Music root for group %s: %r", group_id[:8], path)
-    if path:
-        enrich_fn = state.get("enrich_audio_root_fn")
-        if enrich_fn:
-            asyncio.ensure_future(enrich_fn(group_id))
-    return {"path": path, "group_id": group_id}
+    clean = _validate_app_dirs(state, group_id, paths,
+                               require_writable=require_writable)
+    await roster.set_app_directories(group_id, app_key, clean,
+                                     set_by=state.get("node_user_id", ""))
+    ctx[f"{app_key}_directories"] = clean
+    # The scalar the handshake ack still publishes for MNP 1.0 clients is
+    # derived, and has to be re-derived here: leaving it behind would make the
+    # ack disagree with the list within a single run, and only until a restart
+    # — the shape of bug that reads as "it works after a restart".
+    from meshbay_node.roster import Roster
+    alias = Roster.ctx_alias(app_key, clean)
+    if alias:
+        ctx[alias[0]] = alias[1]
+    log.info("%s directories for group %s: %s", app_key, group_id[:8],
+             ", ".join(clean) or "(none)")
+
+    enrich = (state.get("enrich_app_dirs_fns") or {}).get(app_key)
+    if enrich:
+        asyncio.ensure_future(enrich(group_id))
+    return {"app": app_key, "directories": clean, "group_id": group_id}
+
+
+async def set_app_directory(state: dict, group_id: str, app_key: str,
+                            path: str, *,
+                            require_writable: bool = False) -> dict:
+    """
+    The single-directory form, for an app that only ever wants one.
+
+    Stored as a one-element list like every other app, because two storage
+    shapes for one idea is what made `video_root` (scalar) and `photo_roots`
+    (list) need separate ops, separate MNP messages and separate widgets to
+    say the same thing. `path=""` clears it.
+    """
+    result = await set_app_directories(
+        state, group_id, app_key, [path] if path else [],
+        require_writable=require_writable)
+    dirs = result["directories"]
+    return {**result, "path": dirs[0] if dirs else ""}
+
+
+# The per-app wrappers MNP still names. They exist so an MNP 1.0 client's
+# `video_root` / `audio_root` / `photo_roots` messages keep working; nothing
+# new should be added here — a new app calls the generic pair above.
+
+async def set_video_root(state: dict, group_id: str, path: str) -> dict:
+    result = await set_app_directory(state, group_id, "video", path)
+    return {"path": result["path"], "group_id": group_id}
+
+
+async def set_audio_root(state: dict, group_id: str, path: str) -> dict:
+    # "music", not "audio": the app's registry key is what identifies it
+    # everywhere, and `audio_root` is only the name the setting used to have.
+    result = await set_app_directory(state, group_id, "music", path)
+    return {"path": result["path"], "group_id": group_id}
 
 
 async def set_photo_roots(state: dict, group_id: str, roots: list[str]) -> dict:
+    result = await set_app_directories(state, group_id, "photo", roots)
+    return {"roots": result["directories"], "group_id": group_id}
+
+
+# ── Chat ─────────────────────────────────────────────────────────────────────
+
+async def set_chat_directory(state: dict, group_id: str, path: str) -> dict:
+    """
+    Where chat attachments are written.
+
+    `require_writable`, unlike every other app directory: this one is a
+    *destination*, not a view. Pointing it at a read-only root would produce an
+    attachment button that fails at the moment somebody uses it, which is the
+    failure mode the RO/RW model exists to move earlier.
     """
-    Which folder(s) are the Photos app's entry points for this group. Unlike
-    `set_video_root`/`set_audio_root`, the whole *set* is replaced in one
-    call (docs/photos.md §2.1) — signed once, same shape as
-    `set_enabled_apps`, rather than one op per root added/removed.
+    return await set_app_directory(state, group_id, "chat", path,
+                                   require_writable=True)
 
-    Always fires a sweep, even to an empty list: a root just added needs its
-    existing contents enriched (nothing else re-visits already-indexed
-    entries), and a root just removed leaves its cache entries harmlessly
-    unused rather than needing any cleanup — re-sweeping the new set costs
-    nothing when it's empty.
+
+async def set_chat_link_preview(state: dict, group_id: str,
+                                enabled: bool) -> dict:
+    """
+    Whether the node fetches a page's title and image when a member posts a
+    link.
+
+    Outbound third-party traffic on the operator's connection, caused by a
+    message they did not write and pointing at a URL they did not choose — so
+    it is theirs to switch off, on the same reasoning as the per-group TMDB
+    switch. Absent means on, because that is what the node did before this
+    existed.
     """
     roster = _roster(state)
     ctx = _group_ctx(state, group_id)
-    await roster.set_photo_roots(group_id, roots, set_by=state.get("node_user_id", ""))
-    ctx["photo_roots"] = roots
-    log.info("Photo roots for group %s: %s", group_id[:8], ", ".join(sorted(roots)) or "(none)")
-    enrich_fn = state.get("enrich_photo_roots_fn")
-    if enrich_fn:
-        asyncio.ensure_future(enrich_fn(group_id))
-    return {"roots": roots, "group_id": group_id}
+    await roster.set_chat_link_preview(group_id, enabled,
+                                       set_by=state.get("node_user_id", ""))
+    ctx["chat_link_preview"] = enabled
+    log.info("Chat link previews for group %s: %s", group_id[:8],
+             "on" if enabled else "off")
+    return {"enabled": enabled, "group_id": group_id}
 
 
 # ── Scan settings ────────────────────────────────────────────────────────────
@@ -1086,7 +1397,7 @@ async def set_scan_settings(state: dict, group_id: str, reconcile_interval_secs:
     """
     How often the indexer's reconciliation backstop runs, and how long a
     changed file is left alone before being hashed (indexer.py
-    DirectoryIndexer). Persisted like set_member_upload/set_enabled_apps —
+    DirectoryIndexer). Persisted like set_enabled_apps —
     but there is also a *live* DirectoryIndexer object to update, since it
     reads these once at construction and runs its own background loop with
     them rather than consulting groups_ctx on every use.