refactor: consolidate writable-dir probe into decnet/paths.py

bus.factory and vectorstore.factory carried byte-identical copies of the
'env override -> writable runtime dir -> ~/.decnet fallback' probe. Move
it to decnet.paths.resolve_runtime_path and call it from both.

The mkdir-create variants (deployer topologies dir, _pid_dir candidate
iteration, personas_pool existence-precedence) are deliberately left
inline: they're different policies, not the same probe.

decnet/bus/factory.py

@@ -20,6 +20,7 @@ import os
 from typing import Any, cast
 
 from decnet.bus.base import BaseBus
+from decnet.paths import resolve_runtime_path
 
 
 def get_bus(**kwargs: Any) -> BaseBus:
@@ -58,14 +59,12 @@ def _default_socket_path() -> str:
     ``RuntimeDirectory=decnet`` sets it up with the right perms; the home
     fallback keeps dev boxes usable without systemd.
     """
-    explicit = os.environ.get("DECNET_BUS_SOCKET")
+    return resolve_runtime_path(
-    if explicit:
+        "bus.sock",
-        return explicit
+        env_var="DECNET_BUS_SOCKET",
-
+        runtime_dir="/run/decnet",
-    runtime_dir = "/run/decnet"
+        user_fallback="~/.decnet/bus.sock",
-    if os.path.isdir(runtime_dir) and os.access(runtime_dir, os.W_OK):
+    )
-        return f"{runtime_dir}/bus.sock"
-    return os.path.expanduser("~/.decnet/bus.sock")
 
 
 def _maybe_wrap_telemetry(bus: BaseBus) -> BaseBus:

decnet/paths.py

@@ -0,0 +1,41 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""Shared runtime filesystem path resolution.
+
+DECNET writes runtime state under a system dir provisioned by ``decnet
+init`` / systemd (``/var/lib/decnet`` for state, ``/run/decnet`` for
+sockets). On dev boxes without systemd, or in CI, that dir may be absent
+or read-only, so callers fall back to a per-user location.
+
+:func:`resolve_runtime_path` centralises the writable-dir probe that the
+vectorstore and bus backends previously copy-pasted verbatim.
+"""
+from __future__ import annotations
+
+import os
+
+
+def resolve_runtime_path(
+    filename: str,
+    *,
+    env_var: str,
+    runtime_dir: str,
+    user_fallback: str,
+) -> str:
+    """Resolve a runtime file path. Creates nothing.
+
+    Precedence:
+      1. ``$env_var`` if set (used verbatim).
+      2. ``runtime_dir/filename`` if ``runtime_dir`` exists and is writable.
+      3. ``user_fallback`` (``~`` expanded).
+
+    ``runtime_dir`` is *probed*, never created: it is meant to be
+    provisioned with the right ownership and perms by init/systemd, so
+    creating it here with whatever perms the current process happens to
+    have would be worse than falling back to the user path.
+    """
+    explicit = os.environ.get(env_var)
+    if explicit:
+        return explicit
+    if os.path.isdir(runtime_dir) and os.access(runtime_dir, os.W_OK):
+        return os.path.join(runtime_dir, filename)
+    return os.path.expanduser(user_fallback)

decnet/vectorstore/factory.py

@@ -26,6 +26,7 @@ import logging
 import os
 from typing import Any
 
+from decnet.paths import resolve_runtime_path
 from decnet.vectorstore.base import BaseVectorStore
 
 LOG = logging.getLogger(__name__)
@@ -65,10 +66,9 @@ def get_vectorstore(**kwargs: Any) -> BaseVectorStore:
 
 
 def _default_db_path() -> str:
-    explicit = os.environ.get("DECNET_VECTORSTORE_PATH")
+    return resolve_runtime_path(
-    if explicit:
+        "vectors.sqlite",
-        return explicit
+        env_var="DECNET_VECTORSTORE_PATH",
-    runtime_dir = "/var/lib/decnet"
+        runtime_dir="/var/lib/decnet",
-    if os.path.isdir(runtime_dir) and os.access(runtime_dir, os.W_OK):
+        user_fallback="~/.decnet/vectors.sqlite",
-        return f"{runtime_dir}/vectors.sqlite"
+    )
-    return os.path.expanduser("~/.decnet/vectors.sqlite")

tests/test_paths.py

@@ -0,0 +1,32 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""Tests for the shared runtime-path probe."""
+from __future__ import annotations
+
+from decnet.paths import resolve_runtime_path
+
+
+def _resolve(tmp_path, env=None, runtime_dir=None):
+    return resolve_runtime_path(
+        "x.sock",
+        env_var="DECNET_TEST_PATH",
+        runtime_dir=str(runtime_dir if runtime_dir is not None else tmp_path),
+        user_fallback="~/.decnet/x.sock",
+    )
+
+
+def test_env_override_wins(tmp_path, monkeypatch):
+    monkeypatch.setenv("DECNET_TEST_PATH", "/explicit/here.sock")
+    assert _resolve(tmp_path) == "/explicit/here.sock"
+
+
+def test_writable_runtime_dir(tmp_path, monkeypatch):
+    monkeypatch.delenv("DECNET_TEST_PATH", raising=False)
+    assert _resolve(tmp_path) == str(tmp_path / "x.sock")
+
+
+def test_falls_back_when_runtime_dir_absent(tmp_path, monkeypatch):
+    monkeypatch.delenv("DECNET_TEST_PATH", raising=False)
+    missing = tmp_path / "nope"  # does not exist → not a writable dir
+    result = _resolve(tmp_path, runtime_dir=missing)
+    assert result.endswith("/.decnet/x.sock")
+    assert "~" not in result  # tilde expanded