feat(node): Phase 2 server side — one directory setting for every app

`video_root` (a string), `audio_root` (a string) and `photo_roots` (a list)
said the same thing three ways: three roster accessors, three ops, three MNP
messages, three admin-op subjects. They become `set_app_directories(app_key,
paths)` and its single-directory wrapper, stored under `<app>_directories` and
keyed by the app's registry name — so an application can be added without
touching this layer, which is the whole claim of the plugin architecture.

The three old names still work. Their MNP messages are handled, and the roster
falls back to the old key when the new one is unset, so a node upgraded into
this keeps working with no migration step — the plan called for a script, and
a script nobody runs on the machine where it matters is worse than a fallback.

Two things are new rather than moved:

The paths are validated. The setters this replaces accepted anything, so a
typo — or a path left behind when a root was removed — was stored happily and
then matched no entry, leaving an app showing an empty tab with nothing to
distinguish "misconfigured" from "no files yet". Deliberately not
`RootSet.resolve()`: that also refuses a currently-unavailable root, and an
operator must be able to point an app at a library on a drive they ejected.

The legacy scalar is derived, never stored. `video_root` still rides on the
handshake ack for MNP 1.0 clients; kept as a second stored value it would
drift from the list within one run, which reads as "it works after a restart".

Also here: chat's own two settings (a directory, which must be on a read-write
root because it is a destination rather than a view, and a link-preview switch
gating the unfurl path — checked before the cache, or turning it off would
still serve every preview already fetched), the `app_directories`,
`chat_directory` and `chat_link_preview` MNP messages, the plural
`<app>_directories` on the handshake ack, and `music` as the app's one
identifier where storage said `audio` and the registry said `music`.

The Music enricher now resolves a boundary per configured directory rather
than one for the group: with several, a single boundary is wrong for all but
one of them, and for Music that is the difference between reading a folder as
an artist and reading it as a release.

Suite: 11 failures, all pre-existing.

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

8 files changed, 329 insertions, 25 deletions

diff --git a/packages/meshbay-node/tests/test_app_directories.py b/packages/meshbay-node/tests/test_app_directories.py
new file mode 100644
index 0000000..3ede1b6
--- /dev/null
+++ b/packages/meshbay-node/tests/test_app_directories.py
@@ -0,0 +1,292 @@
+"""
+One shape for every application's directories.
+
+`video_root` (a string), `audio_root` (a string) and `photo_roots` (a list)
+said the same thing three ways, and each needed its own op, its own MNP message
+and its own settings widget. They are one function keyed by the app's own name
+now, which is what lets an application be added without touching this layer at
+all — the whole claim of the plugin architecture.
+
+Two properties are new rather than moved, and both matter more than the tidying:
+
+* **the paths are validated.** The setters this replaces accepted anything. A
+  typo, or a path left behind when a root was removed, was stored happily and
+  then matched no entry — an app showing an empty tab, with nothing to
+  distinguish "misconfigured" from "no files yet". The moment of setting is the
+  only one where the operator is present to be told;
+* **the legacy scalar is derived, never stored.** `video_root` still rides on
+  the handshake ack for MNP 1.0 clients. Kept as a second stored value it would
+  drift from the list within one run — the shape of bug that reads as "it works
+  after a restart".
+"""
+
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_node import ops
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import RootSet
+from meshbay_node.roster import Roster
+
+pytestmark = pytest.mark.asyncio
+
+GROUP = "g" * 32
+
+
+async def _state(tmp_path: Path, *, writable: bool = True) -> tuple[dict, Roster]:
+    media = tmp_path / "Media"
+    (media / "Films").mkdir(parents=True)
+    (media / "Albums").mkdir()
+    published = tmp_path / "Published"
+    published.mkdir()
+
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    roots = RootSet.build([
+        {"path": str(media), "writable": writable},
+        {"path": str(published)},
+    ])
+    state = {
+        "roster": roster,
+        "node_user_id": "operator",
+        "groups_ctx": {GROUP: {"index": index, "roots": roots}},
+        "config": SimpleNamespace(groups=[SimpleNamespace(id=GROUP, roots=[])]),
+    }
+    return state, roster
+
+
+# ── One function, any app ────────────────────────────────────────────────────
+
+async def test_an_app_nobody_wrote_code_for_stores_its_directories(tmp_path):
+    """
+    The point of the generic pair. Nothing in ops.py, roster.py or the daemon
+    names this app, and it round-trips anyway — which is the difference between
+    a plugin architecture and a list of special cases.
+    """
+    state, roster = await _state(tmp_path)
+    try:
+        await ops.set_app_directories(state, GROUP, "helloworld", ["Media/Films"])
+        assert await roster.app_directories(GROUP, "helloworld") == ["Media/Films"]
+    finally:
+        await roster.close()
+
+
+async def test_directories_are_deduplicated_and_ordered(tmp_path):
+    """
+    The stored form is what the operator signs a subject built from, on both
+    sides. Two clients sending the same set in different orders must produce
+    the same bytes, or one of them refuses to sign its own request.
+    """
+    state, roster = await _state(tmp_path)
+    try:
+        out = await ops.set_app_directories(
+            state, GROUP, "video", ["Media/Films", "Media", "Media/Films"])
+        assert out["directories"] == ["Media", "Media/Films"]
+    finally:
+        await roster.close()
+
+
+async def test_a_single_directory_app_stores_a_one_element_list(tmp_path):
+    state, roster = await _state(tmp_path)
+    try:
+        out = await ops.set_app_directory(state, GROUP, "chat", "Media/Films",
+                                          require_writable=True)
+        assert out["path"] == "Media/Films"
+        assert await roster.app_directories(GROUP, "chat") == ["Media/Films"]
+
+        cleared = await ops.set_app_directory(state, GROUP, "chat", "")
+        assert cleared["path"] == ""
+        assert await roster.app_directories(GROUP, "chat") == []
+    finally:
+        await roster.close()
+
+
+# ── Validation ───────────────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("bad", [
+    "Nowhere", "Nowhere/Deeper", "/etc", "Media/../../etc", "..",
+])
+async def test_a_directory_outside_every_root_is_refused(tmp_path, bad):
+    state, roster = await _state(tmp_path)
+    try:
+        with pytest.raises(ops.OpError):
+            await ops.set_app_directories(state, GROUP, "video", [bad])
+        assert await roster.app_directories(GROUP, "video") == []
+    finally:
+        await roster.close()
+
+
+async def test_a_read_only_root_is_refused_where_writability_is_required(tmp_path):
+    """
+    Chat's directory is a destination, not a view. Storing one on a read-only
+    root would produce a paperclip that fails at the moment somebody uses it,
+    which is the failure the RO/RW model exists to move earlier.
+    """
+    state, roster = await _state(tmp_path)
+    try:
+        with pytest.raises(ops.OpError, match="read-only"):
+            await ops.set_app_directory(state, GROUP, "chat", "Published",
+                                        require_writable=True)
+        # The same path is fine for an app that only reads it.
+        await ops.set_app_directories(state, GROUP, "video", ["Published"])
+        assert await roster.app_directories(GROUP, "video") == ["Published"]
+    finally:
+        await roster.close()
+
+
+async def test_a_directory_on_an_unplugged_drive_can_still_be_configured(tmp_path):
+    """
+    Deliberately *not* `RootSet.resolve()`, which also refuses a root that is
+    currently unavailable. An operator must be able to point an app at a
+    library on a drive they have ejected — what is checked is the shape, which
+    does not change with what happens to be mounted.
+    """
+    state, roster = await _state(tmp_path)
+    roots = state["groups_ctx"][GROUP]["roots"]
+    roots.roots[0].available = False
+    try:
+        out = await ops.set_app_directories(state, GROUP, "video", ["Media/Films"])
+        assert out["directories"] == ["Media/Films"]
+    finally:
+        await roster.close()
+
+
+# ── The derived scalar ───────────────────────────────────────────────────────
+
+async def test_the_legacy_scalar_follows_the_list_in_the_live_context(tmp_path):
+    """
+    `video_root` rides on the handshake ack for MNP 1.0 clients and is read
+    from the group context. Left behind by a save, it would disagree with the
+    list until the next restart.
+    """
+    state, roster = await _state(tmp_path)
+    ctx = state["groups_ctx"][GROUP]
+    try:
+        await ops.set_app_directories(state, GROUP, "video",
+                                      ["Media/Films", "Media/Albums"])
+        assert ctx["video_directories"] == ["Media/Albums", "Media/Films"]
+        assert ctx["video_root"] == "Media/Albums", (
+            "the scalar must be the first of the list, not a stale value")
+
+        await ops.set_app_directories(state, GROUP, "video", [])
+        assert ctx["video_root"] == ""
+    finally:
+        await roster.close()
+
+
+async def test_the_photo_alias_stays_a_list_and_chat_stays_a_string(tmp_path):
+    """The alias table has to carry the shape, not just the name."""
+    state, roster = await _state(tmp_path)
+    ctx = state["groups_ctx"][GROUP]
+    try:
+        await ops.set_app_directories(state, GROUP, "photo",
+                                      ["Media/Films", "Media/Albums"])
+        assert ctx["photo_roots"] == ["Media/Albums", "Media/Films"]
+        await ops.set_app_directory(state, GROUP, "chat", "Media",
+                                    require_writable=True)
+        assert ctx["chat_directory"] == "Media"
+    finally:
+        await roster.close()
+
+
+# ── Reading what an older node stored ────────────────────────────────────────
+
+async def test_an_existing_video_root_is_read_without_a_migration(tmp_path):
+    """
+    A node upgraded into this reads its old key until the first save through
+    the new path. Requiring a migration script to run before the Videos tab
+    works again would be a step nobody performs on the machine where it
+    matters.
+    """
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        await roster.set_setting(GROUP, "video_root", "Media/Films")
+        assert await roster.app_directories(GROUP, "video") == ["Media/Films"]
+
+        await roster.set_setting(GROUP, "audio_root", "Media/Albums")
+        assert await roster.app_directories(GROUP, "music") == ["Media/Albums"]
+
+        await roster.set_setting(GROUP, "photo_roots", '["A", "B"]')
+        assert await roster.app_directories(GROUP, "photo") == ["A", "B"]
+    finally:
+        await roster.close()
+
+
+async def test_an_empty_legacy_value_means_nothing_configured(tmp_path):
+    """`video_root = ""` was how "unset" was spelled; it must not become `[""]`."""
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        await roster.set_setting(GROUP, "video_root", "")
+        assert await roster.app_directories(GROUP, "video") == []
+    finally:
+        await roster.close()
+
+
+async def test_the_new_key_wins_over_the_legacy_one(tmp_path):
+    """
+    Both present is a group saved once through the new path. Reading the old
+    key there would undo that save on every load.
+    """
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        await roster.set_setting(GROUP, "video_root", "Old/Place")
+        await roster.set_app_directories(GROUP, "video", ["New/Place"])
+        assert await roster.app_directories(GROUP, "video") == ["New/Place"]
+    finally:
+        await roster.close()
+
+
+async def test_an_app_with_no_legacy_name_simply_has_none(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        assert await roster.app_directories(GROUP, "helloworld") == []
+    finally:
+        await roster.close()
+
+
+# ── Chat's own settings ──────────────────────────────────────────────────────
+
+async def test_link_previews_default_on_and_survive_a_restart(tmp_path):
+    """
+    Absent means on, because that is what the node did before the switch
+    existed — an upgrade must not silently change what a group's chat does.
+    """
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        assert await roster.chat_link_preview(GROUP) is True
+        await roster.set_chat_link_preview(GROUP, False, set_by="op")
+        assert await roster.chat_link_preview(GROUP) is False
+    finally:
+        await roster.close()
+
+    reopened = Roster(db_path=tmp_path / "roster.db")
+    await reopened.open()
+    try:
+        assert await reopened.chat_link_preview(GROUP) is False
+        assert await reopened.chat_link_preview("other") is True, (
+            "one group's setting must not answer for another")
+    finally:
+        await reopened.close()
+
+
+async def test_setting_link_previews_updates_the_live_context(tmp_path):
+    """
+    The unfurl handler reads the context, not the database — it runs per
+    message and a round trip there would be absurd. So the two are kept in
+    step by the op that changes it.
+    """
+    state, roster = await _state(tmp_path)
+    try:
+        await ops.set_chat_link_preview(state, GROUP, False)
+        assert state["groups_ctx"][GROUP]["chat_link_preview"] is False
+    finally:
+        await roster.close()
diff --git a/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py b/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py
index deab1bd..5a8f9b1 100644
--- a/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py
+++ b/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py
@@ -102,7 +102,7 @@ async def test_only_entries_under_the_configured_root_are_enriched(tmp_path):
 
     daemon = await _make_daemon(tmp_path, shared, group_id)
     try:
-        await daemon._roster.set_audio_root(group_id, "shared/Music", set_by="op")
+        await daemon._roster.set_app_directories(group_id, "music", ["shared/Music"], set_by="op")
         indexer = DirectoryIndexer(
             roots=one_root(shared), group_id=group_id,
             sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
@@ -142,8 +142,10 @@ async def test_setting_the_audio_root_sweeps_what_it_already_contains(tmp_path):
         # this file all along.
         state = {
             "roster": daemon._roster,
-            "groups_ctx": {group_id: {}},
-            "enrich_audio_root_fn": daemon._enrich_audio_root_now,
+            # Real roots, because ops now refuses a directory that is not
+            # inside one — the per-app setters this replaced validated nothing.
+            "groups_ctx": {group_id: {"roots": one_root(shared)}},
+            "enrich_app_dirs_fns": {"music": daemon._enrich_audio_root_now},
         }
         await ops.set_audio_root(state, group_id, "shared/Music")
         await asyncio.sleep(0.05)  # let the fire-and-forget sweep actually run
@@ -197,8 +199,10 @@ async def test_the_same_file_shared_into_two_groups_enriches_in_both(tmp_path):
             "the fixture itself must produce identical content hashes — "
             "otherwise this test isn't exercising the collision at all")
 
-        await daemon._roster.set_audio_root(group_a, "shared_a/Music", set_by="op")
-        await daemon._roster.set_audio_root(group_b, "shared_b/Music", set_by="op")
+        await daemon._roster.set_app_directories(
+            group_a, "music", ["shared_a/Music"], set_by="op")
+        await daemon._roster.set_app_directories(
+            group_b, "music", ["shared_b/Music"], set_by="op")
 
         await daemon._enrich_new_audio_entries(indexer_a, list(indexer_a.index.entries))
         await daemon._enrich_new_audio_entries(indexer_b, list(indexer_b.index.entries))
diff --git a/packages/meshbay-node/tests/test_audio_root_policy.py b/packages/meshbay-node/tests/test_audio_root_policy.py
index 576c08a..e2e9254 100644
--- a/packages/meshbay-node/tests/test_audio_root_policy.py
+++ b/packages/meshbay-node/tests/test_audio_root_policy.py
@@ -125,17 +125,20 @@ async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
     roster = Roster(db_path=tmp_path / "roster.db")
     await roster.open()
     try:
-        assert await roster.audio_root("g1") == "", "absent must mean unset"
-        await roster.set_audio_root("g1", "shared/Music", set_by="op")
-        assert await roster.audio_root("g1") == "shared/Music"
+        assert await roster.app_directories("g1", "music") == [], (
+            "absent must mean unset")
+        await roster.set_app_directories("g1", "music", ["shared/Music"],
+                                         set_by="op")
+        assert await roster.app_directories("g1", "music") == ["shared/Music"]
     finally:
         await roster.close()
 
     reopened = Roster(db_path=tmp_path / "roster.db")
     await reopened.open()
     try:
-        assert await reopened.audio_root("g1") == "shared/Music"
-        assert await reopened.audio_root("g2") == "", "one group's setting must not answer for another"
+        assert await reopened.app_directories("g1", "music") == ["shared/Music"]
+        assert await reopened.app_directories("g2", "music") == [], (
+            "one group's setting must not answer for another")
     finally:
         await reopened.close()
 
@@ -239,7 +242,7 @@ async def test_a_real_signed_save_persists_and_survives_a_fresh_roster_read(tmp_
 
         assert session._ctx["groups"][GROUP]["audio_root"] == "shared/Music", (
             "the live in-memory context must reflect the new root immediately")
-        assert await roster.audio_root(GROUP) == "shared/Music", (
+        assert await roster.app_directories(GROUP, "music") == ["shared/Music"], (
             "the same Roster instance must read back what it just wrote")
     finally:
         await roster.close()
@@ -249,7 +252,7 @@ async def test_a_real_signed_save_persists_and_survives_a_fresh_roster_read(tmp_
     # actually answers "does it survive a reload".
     reopened = await open_roster(tmp_path)
     try:
-        assert await reopened.audio_root(GROUP) == "shared/Music", (
+        assert await reopened.app_directories(GROUP, "music") == ["shared/Music"], (
             "a freshly-opened Roster against the same db file must see the "
             "committed value — anything else means the write was never "
             "durable in the first place")
diff --git a/packages/meshbay-node/tests/test_rename_reenrichment.py b/packages/meshbay-node/tests/test_rename_reenrichment.py
index 7a77368..f6761d9 100644
--- a/packages/meshbay-node/tests/test_rename_reenrichment.py
+++ b/packages/meshbay-node/tests/test_rename_reenrichment.py
@@ -37,8 +37,8 @@ def _free_port() -> int:
 
 
 class _StubRoster:
-    async def video_root(self, group_id):
-        return "shared"
+    async def app_directories(self, group_id, app_key):
+        return ["shared"] if app_key == "video" else []
 
 
 class _SpyEnricher:
diff --git a/packages/meshbay-node/tests/test_root_eject.py b/packages/meshbay-node/tests/test_root_eject.py
index 0ec36a4..f57fb92 100644
--- a/packages/meshbay-node/tests/test_root_eject.py
+++ b/packages/meshbay-node/tests/test_root_eject.py
@@ -17,7 +17,6 @@ exactly what an operator does after noticing a drive fell off — silently undid
 the eject, and the next scan read an empty mount point as an erased library.
 """
 
-import asyncio
 from pathlib import Path
 
 import pytest
diff --git a/packages/meshbay-node/tests/test_startup_scan_enrichment.py b/packages/meshbay-node/tests/test_startup_scan_enrichment.py
index 65b9728..d2cbc3e 100644
--- a/packages/meshbay-node/tests/test_startup_scan_enrichment.py
+++ b/packages/meshbay-node/tests/test_startup_scan_enrichment.py
@@ -60,8 +60,9 @@ async def test_a_file_already_on_disk_at_startup_gets_enrichment_scheduled(tmp_p
         data_dir=tmp_path / "data",
     )
     class _StubRoster:
-        async def video_root(self, group_id):
-            return "shared"  # the root itself, i.e. "enrich the whole thing"
+        async def app_directories(self, group_id, app_key):
+            # The root itself, i.e. "enrich the whole thing".
+            return ["shared"] if app_key == "video" else []
 
     daemon = NodeDaemon(config)
     daemon._broadcast_coalesce_secs = 0.01  # real value would make this test wait 0.5s
diff --git a/packages/meshbay-node/tests/test_video_root_gates_enrichment.py b/packages/meshbay-node/tests/test_video_root_gates_enrichment.py
index 9cfb819..b88ecaf 100644
--- a/packages/meshbay-node/tests/test_video_root_gates_enrichment.py
+++ b/packages/meshbay-node/tests/test_video_root_gates_enrichment.py
@@ -99,7 +99,7 @@ async def test_only_entries_under_the_configured_root_are_enriched(tmp_path):
 
     daemon = await _make_daemon(tmp_path, shared, group_id)
     try:
-        await daemon._roster.set_video_root(group_id, "shared/Movies", set_by="op")
+        await daemon._roster.set_app_directories(group_id, "video", ["shared/Movies"], set_by="op")
         indexer = DirectoryIndexer(
             roots=one_root(shared), group_id=group_id,
             sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
@@ -139,8 +139,10 @@ async def test_setting_the_video_root_sweeps_what_it_already_contains(tmp_path):
         # this file all along.
         state = {
             "roster": daemon._roster,
-            "groups_ctx": {group_id: {}},
-            "enrich_video_root_fn": daemon._enrich_video_root_now,
+            # Real roots, because ops now refuses a directory that is not
+            # inside one — the per-app setters this replaced validated nothing.
+            "groups_ctx": {group_id: {"roots": one_root(shared)}},
+            "enrich_app_dirs_fns": {"video": daemon._enrich_video_root_now},
         }
         await ops.set_video_root(state, group_id, "shared/Movies")
         await asyncio.sleep(0.05)  # let the fire-and-forget sweep actually run
diff --git a/packages/meshbay-node/tests/test_video_root_policy.py b/packages/meshbay-node/tests/test_video_root_policy.py
index 8cc1540..8d8c45a 100644
--- a/packages/meshbay-node/tests/test_video_root_policy.py
+++ b/packages/meshbay-node/tests/test_video_root_policy.py
@@ -126,16 +126,19 @@ async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
     roster = Roster(db_path=tmp_path / "roster.db")
     await roster.open()
     try:
-        assert await roster.video_root("g1") == "", "absent must mean the whole group index"
-        await roster.set_video_root("g1", "shared/Movies", set_by="op")
-        assert await roster.video_root("g1") == "shared/Movies"
+        assert await roster.app_directories("g1", "video") == [], (
+            "absent must mean nothing configured")
+        await roster.set_app_directories("g1", "video", ["shared/Movies"],
+                                         set_by="op")
+        assert await roster.app_directories("g1", "video") == ["shared/Movies"]
     finally:
         await roster.close()
 
     reopened = Roster(db_path=tmp_path / "roster.db")
     await reopened.open()
     try:
-        assert await reopened.video_root("g1") == "shared/Movies"
-        assert await reopened.video_root("g2") == "", "one group's setting must not answer for another"
+        assert await reopened.app_directories("g1", "video") == ["shared/Movies"]
+        assert await reopened.app_directories("g2", "video") == [], (
+            "one group's setting must not answer for another")
     finally:
         await reopened.close()