[eric] settings: leading-_ -> p_/public in store+settings+credentials+redaction; promote cross-file API public (save_settings/atomic_write_settings/migrate_legacy_fields), drop redundant _save_settings alias, fix re-export + cache test refs

backend/apps/agents/manager/prompt/attachments.py

@@ -96,7 +96,7 @@ def resolve_attachments(context_paths: Optional[List], api_type: str, model: str
     """
     if not context_paths:
         return "", [], []
-    from backend.apps.settings.settings import _sniff_file_kind
+    from backend.apps.settings.settings import sniff_file_kind
     import base64 as _b64
     sections: List[str] = []
     native: List[dict] = []
@@ -182,7 +182,7 @@ def resolve_attachments(context_paths: Optional[List], api_type: str, model: str
             size = os.path.getsize(path)
             with open(path, "rb") as fh:
                 head = fh.read(4096)
-            kind, media_type = _sniff_file_kind(head, os.path.basename(path))
+            kind, media_type = sniff_file_kind(head, os.path.basename(path))
 
             if kind == "text":
                 with open(path, "r", errors="replace") as f:

backend/apps/service/client.py

@@ -72,13 +72,13 @@ def _get_install_id() -> str:
     if _install_id:
         return _install_id
     try:
-        from backend.apps.settings.store import load_settings, _save_settings
+        from backend.apps.settings.store import load_settings, save_settings
         s = load_settings()
         iid = getattr(s, "installation_id", None)
         if not iid:
             iid = uuid4().hex
             s.installation_id = iid
-            _save_settings(s)
+            save_settings(s)
         _install_id = iid
     except Exception:
         _install_id = uuid4().hex

backend/apps/service/service.py

@@ -125,13 +125,13 @@ async def service_lifespan():
     global _pulse_task, _drain_task, _9r_start_task
 
     try:
-        from backend.apps.settings.settings import load_settings, _save_settings
+        from backend.apps.settings.settings import load_settings, save_settings
         settings = load_settings()
 
         is_first_open = settings.first_opened_at is None
         if is_first_open:
             settings.first_opened_at = datetime.now().isoformat()
-            _save_settings(settings)
+            save_settings(settings)
 
         days_since_install = 0
         if settings.first_opened_at:

backend/apps/settings/credentials.py

@@ -31,7 +31,7 @@ def proxy_auth(settings: AppSettings) -> tuple[str | None, str | None]:
     return (None, None)
 
 
-def _check_9router() -> bool:
+def p_check_9router() -> bool:
     """Check if 9Router is running locally."""
     try:
         import httpx
@@ -50,7 +50,7 @@ def validate_credentials(settings: AppSettings, provider: str = "anthropic") ->
         return
 
     # 9Router proxies every provider, so if it's up we don't need keys here.
-    if _check_9router():
+    if p_check_9router():
         return
 
     if p == "anthropic":
@@ -135,7 +135,7 @@ def get_anthropic_client(settings: AppSettings) -> anthropic.AsyncAnthropic:
         return anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
 
     # Fall back to 9Router (free for users with Claude/ChatGPT/Gemini subscriptions).
-    if _check_9router():
+    if p_check_9router():
         return anthropic.AsyncAnthropic(
             api_key="9router",
             base_url="http://localhost:20128",

backend/apps/settings/redaction.py

@@ -19,17 +19,17 @@ from typing import Any
 
 from backend.common.secret_scan import looks_secret
 
-_SECRET_NAME_SUFFIXES = ("_key", "_token", "_secret")
+P_SECRET_NAME_SUFFIXES = ("_key", "_token", "_secret")
 # Not a credential and doesn't match the suffix rule, but a stable hardware-ish
 # fingerprint used for cohorting/abuse; keep it out of the agent's eyes too.
-_SECRET_EXTRA_FIELDS = frozenset({"installation_id"})
+P_SECRET_EXTRA_FIELDS = frozenset({"installation_id"})
 
 
 def is_secret_field(name: str) -> bool:
-    return name.endswith(_SECRET_NAME_SUFFIXES) or name in _SECRET_EXTRA_FIELDS
+    return name.endswith(P_SECRET_NAME_SUFFIXES) or name in P_SECRET_EXTRA_FIELDS
 
 
-def _value_is_secret_shaped(value: Any) -> bool:
+def p_value_is_secret_shaped(value: Any) -> bool:
     """Fail-safe behind the name rule: a field the name rule misses (a future
     secret with an off-convention name) is still caught if its VALUE looks like
     a credential (sk-..., ghp_..., Bearer ...). So a leak needs BOTH a bad name
@@ -37,7 +37,7 @@ def _value_is_secret_shaped(value: Any) -> bool:
     return isinstance(value, str) and looks_secret(value)
 
 
-def _redact_value(value: Any) -> dict[str, Any]:
+def p_redact_value(value: Any) -> dict[str, Any]:
     """A secret rendered as state, never content: configured + last 4 only."""
     if value is None or (isinstance(value, str) and value.strip() == ""):
         return {"configured": False}
@@ -50,19 +50,19 @@ def redact_settings(raw: dict[str, Any]) -> dict[str, Any]:
     {configured, last4}. Nested custom-provider api_keys are redacted too."""
     out: dict[str, Any] = {}
     for key, value in raw.items():
-        if is_secret_field(key) or _value_is_secret_shaped(value):
-            out[key] = _redact_value(value)
+        if is_secret_field(key) or p_value_is_secret_shaped(value):
+            out[key] = p_redact_value(value)
         elif key == "custom_providers" and isinstance(value, list):
-            out[key] = [_redact_custom_provider(cp) for cp in value]
+            out[key] = [p_redact_custom_provider(cp) for cp in value]
         else:
             out[key] = value
     return out
 
 
-def _redact_custom_provider(cp: Any) -> Any:
+def p_redact_custom_provider(cp: Any) -> Any:
     if not isinstance(cp, dict):
         return cp
     out = dict(cp)
     if "api_key" in out:
-        out["api_key"] = _redact_value(out.get("api_key"))
+        out["api_key"] = p_redact_value(out.get("api_key"))
     return out

backend/apps/settings/settings.py

@@ -16,9 +16,8 @@ from backend.apps.settings.store import (
     SETTINGS_FILE,
     load_settings,
     save_settings,
-    _save_settings,
-    _atomic_write_settings,
-    _migrate_legacy_fields,
+    atomic_write_settings,
+    migrate_legacy_fields,
 )
 
 logger = logging.getLogger(__name__)
@@ -77,13 +76,13 @@ async def settings_lifespan():
             await sync_custom_providers(getattr(s, "custom_providers", None) or [])
 
         _asyncio.create_task(_boot_router_then_sync())
-        _asyncio.create_task(_upload_dir_gc_loop())
+        _asyncio.create_task(p_upload_dir_gc_loop())
     except Exception as e:
         logger.warning(f"9Router sync startup failed: {e}")
     yield
 
 
-async def _upload_dir_gc_loop():
+async def p_upload_dir_gc_loop():
     """Daily GC of UPLOAD_DIR. Without this, every PDF/image the user
     drops sits in the OS temp dir forever, growing unbounded across
     sessions. We keep files for 7 days to make resume-after-restart
@@ -116,7 +115,7 @@ async def save_settings_async(settings_obj: AppSettings) -> None:
     """Async atomic save via thread pool; shares the lock with the sync variant."""
     payload = settings_obj.model_dump()
     loop = asyncio.get_running_loop()
-    await loop.run_in_executor(None, _atomic_write_settings, payload)
+    await loop.run_in_executor(None, atomic_write_settings, payload)
 
 
 @settings.router.get("")
@@ -157,15 +156,15 @@ import weakref as _weakref
 # the first loop that uses it and then errors on reuse from another loop (every
 # async test spins a fresh one). WeakKeyDictionary auto-drops a loop's lock once
 # the loop is gone.
-_settings_write_locks: "_weakref.WeakKeyDictionary" = _weakref.WeakKeyDictionary()
+p_settings_write_locks: "_weakref.WeakKeyDictionary" = _weakref.WeakKeyDictionary()
 
 
 def settings_write_lock() -> asyncio.Lock:
     loop = asyncio.get_running_loop()
-    lock = _settings_write_locks.get(loop)
+    lock = p_settings_write_locks.get(loop)
     if lock is None:
         lock = asyncio.Lock()
-        _settings_write_locks[loop] = lock
+        p_settings_write_locks[loop] = lock
     return lock
 
 
@@ -382,7 +381,7 @@ async def reset_system_prompt():
 # defaults EXCEPT the things a "reset my preferences" click must never silently
 # sever, your connections (server-owned subscription fields AND your pasted
 # provider credentials) and your identity. Hard-erase is the separate flow.
-_RESET_PRESERVE_FIELDS = SERVER_OWNED_FIELDS + (
+P_RESET_PRESERVE_FIELDS = SERVER_OWNED_FIELDS + (
     "anthropic_api_key",
     "openai_api_key",
     "google_api_key",
@@ -399,7 +398,7 @@ _RESET_PRESERVE_FIELDS = SERVER_OWNED_FIELDS + (
 async def reset_to_defaults():
     old = load_settings()
     fresh = AppSettings()
-    for k in _RESET_PRESERVE_FIELDS:
+    for k in P_RESET_PRESERVE_FIELDS:
         setattr(fresh, k, getattr(old, k, None))
     await save_settings_async(fresh)
     return {"ok": True, "settings": fresh.model_dump()}
@@ -416,7 +415,7 @@ UPLOAD_DIR = os.path.join(tempfile.gettempdir(), "self-swarm-uploads")
 os.makedirs(UPLOAD_DIR, exist_ok=True)
 
 
-def _sniff_file_kind(contents: bytes, name: str) -> tuple[str, str | None]:
+def sniff_file_kind(contents: bytes, name: str) -> tuple[str, str | None]:
     """Classify an uploaded file as text/pdf/image/binary so the agent
     layer can route it (inline as text, send as native document/image
     block, or refuse). Returns (kind, media_type)."""
@@ -457,7 +456,7 @@ def _sniff_file_kind(contents: bytes, name: str) -> tuple[str, str | None]:
         return ("binary", None)
 
 
-def _estimate_pdf_tokens(contents: bytes) -> int:
+def estimate_pdf_tokens(contents: bytes) -> int:
     """Conservative PDF token estimate without a parser dep.
 
     We use two signals and take the MAX so the chip + dry-run never
@@ -546,7 +545,7 @@ async def upload_files(files: list[UploadFile] = File(...)):
                 pass
             raise
 
-        kind, media_type = _sniff_file_kind(contents, safe_name)
+        kind, media_type = sniff_file_kind(contents, safe_name)
 
         if kind == "text":
             try:
@@ -556,7 +555,7 @@ async def upload_files(files: list[UploadFile] = File(...)):
             except Exception:
                 tokens_est = min(len(contents), 512_000) // 4
         elif kind == "pdf":
-            tokens_est = _estimate_pdf_tokens(contents)
+            tokens_est = estimate_pdf_tokens(contents)
         elif kind == "image":
             tokens_est = 1_500
         else:
@@ -574,14 +573,14 @@ async def upload_files(files: list[UploadFile] = File(...)):
     return JSONResponse({"files": results})
 
 
-class _SummarizeRequest(BaseModel):
+class p_SummarizeRequest(BaseModel):
     path: str
     target_tokens: int = 4_000
     primary_model: Optional[str] = None
 
 
 @settings.router.post("/summarize-file")
-async def summarize_file(req: _SummarizeRequest):
+async def summarize_file(req: p_SummarizeRequest):
     """Compress an attached file down to a fact-dense summary the agent can
     still reason over without paying the full token cost.
 

backend/apps/settings/store.py

@@ -22,7 +22,7 @@ logger = logging.getLogger(__name__)
 SETTINGS_FILE = os.path.join(DATA_DIR, "settings.json")
 
 
-def _migrate_legacy_fields(raw: dict) -> dict:
+def migrate_legacy_fields(raw: dict) -> dict:
     """Translate deprecated pre-launch field names ('managed', 'openswarm_auth_token') into production schema."""
     if raw.get("connection_mode") == "managed":
         raw["connection_mode"] = "openswarm-pro"
@@ -31,7 +31,7 @@ def _migrate_legacy_fields(raw: dict) -> dict:
     return raw
 
 
-def _coerce_settings(raw: dict) -> AppSettings:
+def p_coerce_settings(raw: dict) -> AppSettings:
     """Build AppSettings, surviving a settings.json written by a different app
     version. Unknown fields are already ignored by pydantic; the case this guards
     is a field whose TYPE drifted across versions (e.g. a list that is now a
@@ -55,7 +55,7 @@ def _coerce_settings(raw: dict) -> AppSettings:
             return AppSettings()
 
 
-def _preserve_corrupt_settings() -> None:
+def p_preserve_corrupt_settings() -> None:
     """Move an unparseable settings.json aside so boot proceeds on defaults while
     the original stays recoverable (the next save would otherwise overwrite it)."""
     try:
@@ -70,11 +70,11 @@ def _preserve_corrupt_settings() -> None:
 # so even a hand-edited file or an unexpected writer is picked up immediately. A stat
 # skips the open+parse+validate that Defender turns into 5-50ms on Windows. Copies on
 # both sides keep handler isolation: callers mutate their copy, never the cache.
-_cached_settings: AppSettings | None = None
-_cached_sig: tuple[int, int] | None = None
+p_cached_settings: AppSettings | None = None
+p_cached_sig: tuple[int, int] | None = None
 
 
-def _settings_sig() -> tuple[int, int] | None:
+def p_settings_sig() -> tuple[int, int] | None:
     try:
         st = os.stat(SETTINGS_FILE)
         return (st.st_mtime_ns, st.st_size)
@@ -86,38 +86,38 @@ def load_settings() -> AppSettings:
     """Load settings from JSON file, returning defaults if not found. Never raises
     on a corrupt or version-mismatched file: a single bad settings.json must not
     brick boot (it is read at startup, by the settings endpoint, and per dispatch)."""
-    global _cached_settings, _cached_sig
-    sig = _settings_sig()
-    if sig is not None and _cached_settings is not None and sig == _cached_sig:
-        return _cached_settings.model_copy(deep=True)
+    global p_cached_settings, p_cached_sig
+    sig = p_settings_sig()
+    if sig is not None and p_cached_settings is not None and sig == p_cached_sig:
+        return p_cached_settings.model_copy(deep=True)
     if os.path.exists(SETTINGS_FILE):
         try:
             with open(SETTINGS_FILE) as f:
                 raw = json.load(f)
         except (json.JSONDecodeError, OSError, ValueError):
-            _preserve_corrupt_settings()
+            p_preserve_corrupt_settings()
             return AppSettings()
         if not isinstance(raw, dict):
             # Valid JSON but not an object (e.g. a bare list/number); unusable.
-            _preserve_corrupt_settings()
+            p_preserve_corrupt_settings()
             return AppSettings()
-        settings = _coerce_settings(_migrate_legacy_fields(raw))
+        settings = p_coerce_settings(migrate_legacy_fields(raw))
         if settings.default_system_prompt is None:
             settings.default_system_prompt = DEFAULT_SYSTEM_PROMPT
-        _cached_settings = settings.model_copy(deep=True)
-        _cached_sig = sig
+        p_cached_settings = settings.model_copy(deep=True)
+        p_cached_sig = sig
         return settings
     return AppSettings()
 
 
 # threading.Lock guards every SETTINGS_FILE write; works for sync paths and async run_in_executor paths.
-_settings_write_lock = threading.Lock()
+p_settings_write_lock = threading.Lock()
 
 
-def _atomic_write_settings(payload: dict) -> None:
+def atomic_write_settings(payload: dict) -> None:
     """Atomic SETTINGS_FILE write; call via save_settings*, not directly."""
-    global _cached_settings, _cached_sig
-    with _settings_write_lock:
+    global p_cached_settings, p_cached_sig
+    with p_settings_write_lock:
         os.makedirs(DATA_DIR, exist_ok=True)
         fd, tmp = tempfile.mkstemp(prefix=".settings.", suffix=".tmp", dir=DATA_DIR)
         try:
@@ -128,10 +128,10 @@ def _atomic_write_settings(payload: dict) -> None:
                 try:
                     os.replace(tmp, SETTINGS_FILE)
                     # Refresh the cache inside the lock so cache order matches disk order.
-                    _cached_settings = _coerce_settings(_migrate_legacy_fields(dict(payload)))
-                    if _cached_settings.default_system_prompt is None:
-                        _cached_settings.default_system_prompt = DEFAULT_SYSTEM_PROMPT
-                    _cached_sig = _settings_sig()
+                    p_cached_settings = p_coerce_settings(migrate_legacy_fields(dict(payload)))
+                    if p_cached_settings.default_system_prompt is None:
+                        p_cached_settings.default_system_prompt = DEFAULT_SYSTEM_PROMPT
+                    p_cached_sig = p_settings_sig()
                     return
                 except PermissionError:
                     if attempt == 1:
@@ -147,8 +147,4 @@ def _atomic_write_settings(payload: dict) -> None:
 
 def save_settings(settings_obj: AppSettings) -> None:
     """Sync atomic persist; thread-safe. Async callers should prefer save_settings_async (Defender can stretch writes to 50-200ms)."""
-    _atomic_write_settings(settings_obj.model_dump())
-
-
-def _save_settings(settings_obj: AppSettings) -> None:
-    save_settings(settings_obj)
+    atomic_write_settings(settings_obj.model_dump())

backend/tests/test_auth_router.py

@@ -31,11 +31,11 @@ def client():
 @pytest.fixture
 def reset_settings():
     """Snapshot + restore settings around each test so writes don't leak."""
-    from backend.apps.settings.settings import load_settings, _save_settings
+    from backend.apps.settings.settings import load_settings, save_settings
 
     original = load_settings().model_copy(deep=True)
     yield
-    _save_settings(original)
+    save_settings(original)
 
 
 # ---------------------------------------------------------------------------
@@ -140,14 +140,14 @@ def test_signin_activate_short_token_rejected_locally(client, reset_settings):
 # ---------------------------------------------------------------------------
 
 def test_signout_clears_local_identity(client, reset_settings):
-    from backend.apps.settings.settings import load_settings, _save_settings
+    from backend.apps.settings.settings import load_settings, save_settings
     s = load_settings()
     s.user_id = "u-bye"
     s.user_email = "bye@example.com"
     s.signin_method = "google"
     s.openswarm_bearer_token = "bearer-to-revoke-xxxxxxxx"
     s.connection_mode = "openswarm-pro"
-    _save_settings(s)
+    save_settings(s)
 
     fake_response = AsyncMock()
     fake_response.status_code = 200
@@ -168,11 +168,11 @@ def test_signout_clears_local_identity(client, reset_settings):
 
 def test_signout_succeeds_even_when_cloud_unreachable(client, reset_settings):
     """A flaky network shouldn't strand the user signed-in locally."""
-    from backend.apps.settings.settings import load_settings, _save_settings
+    from backend.apps.settings.settings import load_settings, save_settings
     s = load_settings()
     s.user_id = "u-flaky"
     s.openswarm_bearer_token = "bearer-flaky-network-xxxx"
-    _save_settings(s)
+    save_settings(s)
 
     with patch("httpx.AsyncClient") as MockClient:
         instance = MockClient.return_value.__aenter__.return_value

backend/tests/test_disk_caches.py

@@ -15,8 +15,8 @@ def settings_tmp(tmp_path, monkeypatch):
     f = tmp_path / "settings.json"
     monkeypatch.setattr(store, "DATA_DIR", str(tmp_path))
     monkeypatch.setattr(store, "SETTINGS_FILE", str(f))
-    monkeypatch.setattr(store, "_cached_settings", None)
-    monkeypatch.setattr(store, "_cached_sig", None)
+    monkeypatch.setattr(store, "p_cached_settings", None)
+    monkeypatch.setattr(store, "p_cached_sig", None)
     return f
 
 

backend/tests/test_settings_meta_concurrency.py

@@ -29,20 +29,20 @@ def _auth_headers():
 
 @pytest.fixture
 def reset_settings():
-    from backend.apps.settings.settings import load_settings, _save_settings
+    from backend.apps.settings.settings import load_settings, save_settings
     original = load_settings().model_copy(deep=True)
     yield
-    _save_settings(original)
+    save_settings(original)
 
 
 @pytest.mark.asyncio
 async def test_concurrent_writes_to_different_fields_both_survive(reset_settings):
-    from backend.apps.settings.settings import load_settings, _save_settings
+    from backend.apps.settings.settings import load_settings, save_settings
 
     base = load_settings()
     base.theme = "dark"
     base.default_mode = "agent"
-    _save_settings(base)
+    save_settings(base)
 
     headers = _auth_headers()
     transport = httpx.ASGITransport(app=app)

backend/tests/test_settings_meta_endpoint.py

@@ -20,13 +20,13 @@ async def test_second_wall_restores_protected_credential_even_if_body_blanks_it(
     live credential blanked (a guard slip upstream), the second-wall restore puts
     it back. Proves the api-key guard isn't a single point of failure."""
     from backend.apps.settings.settings import (
-        apply_settings_update, settings_write_lock, load_settings, _save_settings,
+        apply_settings_update, settings_write_lock, load_settings, save_settings,
     )
     original = load_settings().model_copy(deep=True)
     try:
         s = load_settings()
         s.anthropic_api_key = "sk-live-KEEP-ME"
-        _save_settings(s)
+        save_settings(s)
         # A body that (as if a guard bug let it through) clears the live key.
         body = load_settings()
         body.anthropic_api_key = ""
@@ -41,7 +41,7 @@ async def test_second_wall_restores_protected_credential_even_if_body_blanks_it(
             await apply_settings_update(body2, protect_fields={"anthropic_api_key"})
         assert not load_settings().openai_api_key
     finally:
-        _save_settings(original)
+        save_settings(original)
 
 
 @pytest.fixture
@@ -55,10 +55,10 @@ def client():
 
 @pytest.fixture
 def reset_settings():
-    from backend.apps.settings.settings import load_settings, _save_settings
+    from backend.apps.settings.settings import load_settings, save_settings
     original = load_settings().model_copy(deep=True)
     yield
-    _save_settings(original)
+    save_settings(original)
 
 
 @pytest.fixture
@@ -67,13 +67,13 @@ def session_on_anthropic_key():
     Anthropic key powers it. Registered in agent_manager so the guard sees it."""
     from backend.apps.agents.agent_manager import agent_manager
     from backend.apps.agents.core.models import AgentSession
-    from backend.apps.settings.settings import load_settings, _save_settings
+    from backend.apps.settings.settings import load_settings, save_settings
 
     s = load_settings()
     s.connection_mode = "own_key"
     s.anthropic_api_key = "sk-ant-test-LIVE"
     s.openai_api_key = "sk-openai-test-OTHER"
-    _save_settings(s)
+    save_settings(s)
 
     sess = AgentSession(id="settings-meta-test", name="t", model="opus-4-8")
     agent_manager.sessions["settings-meta-test"] = sess

backend/tests/test_settings_meta_stdio_live.py

@@ -58,10 +58,10 @@ def live_backend():
 
 @pytest.fixture
 def reset_settings():
-    from backend.apps.settings.settings import load_settings, _save_settings
+    from backend.apps.settings.settings import load_settings, save_settings
     original = load_settings().model_copy(deep=True)
     yield
-    _save_settings(original)
+    save_settings(original)
 
 
 def _run_stdio(port: int, token: str, session_id: str, changes: dict) -> str:
@@ -84,7 +84,7 @@ def _run_stdio(port: int, token: str, session_id: str, changes: dict) -> str:
 
 def test_live_stdio_settingswrite_refuses_live_key_clears_other(live_backend, reset_settings):
     port, token = live_backend
-    from backend.apps.settings.settings import load_settings, _save_settings
+    from backend.apps.settings.settings import load_settings, save_settings
     from backend.apps.agents.agent_manager import agent_manager
     from backend.apps.agents.core.models import AgentSession
 
@@ -94,7 +94,7 @@ def test_live_stdio_settingswrite_refuses_live_key_clears_other(live_backend, re
     s.connection_mode = "own_key"
     s.anthropic_api_key = "sk-ant-LIVE-do-not-clear"
     s.openai_api_key = "sk-oai-OTHER-ok-to-clear"
-    _save_settings(s)
+    save_settings(s)
     agent_manager.sessions["live-stdio-test"] = AgentSession(id="live-stdio-test", name="t", model="opus-4-8")
 
     try:

backend/tests/test_settings_migration.py

@@ -27,25 +27,25 @@ def _write(path, obj):
         json.dump(obj, fh)
 
 
-# ---------------- _migrate_legacy_fields ----------------
+# ---------------- migrate_legacy_fields ----------------
 
 def test_migrate_managed_to_openswarm_pro():
-    assert store._migrate_legacy_fields({"connection_mode": "managed"})["connection_mode"] == "openswarm-pro"
+    assert store.migrate_legacy_fields({"connection_mode": "managed"})["connection_mode"] == "openswarm-pro"
 
 
 def test_migrate_auth_token_renamed_and_popped():
-    out = store._migrate_legacy_fields({"openswarm_auth_token": "tok"})
+    out = store.migrate_legacy_fields({"openswarm_auth_token": "tok"})
     assert out["openswarm_bearer_token"] == "tok"
     assert "openswarm_auth_token" not in out
 
 
 def test_migrate_does_not_clobber_existing_bearer():
-    out = store._migrate_legacy_fields({"openswarm_auth_token": "old", "openswarm_bearer_token": "new"})
+    out = store.migrate_legacy_fields({"openswarm_auth_token": "old", "openswarm_bearer_token": "new"})
     assert out["openswarm_bearer_token"] == "new"
 
 
 def test_migrate_leaves_modern_values_untouched():
-    out = store._migrate_legacy_fields({"connection_mode": "own_key"})
+    out = store.migrate_legacy_fields({"connection_mode": "own_key"})
     assert out["connection_mode"] == "own_key"
 
 

backend/tests/test_settings_patch.py

@@ -30,18 +30,18 @@ def client():
 
 @pytest.fixture
 def reset_settings():
-    from backend.apps.settings.settings import load_settings, _save_settings
+    from backend.apps.settings.settings import load_settings, save_settings
     original = load_settings().model_copy(deep=True)
     yield
-    _save_settings(original)
+    save_settings(original)
 
 
 def test_patch_changes_only_sent_fields(client, reset_settings):
-    from backend.apps.settings.settings import load_settings, _save_settings
+    from backend.apps.settings.settings import load_settings, save_settings
     s = load_settings()
     s.theme = "dark"
     s.default_mode = "chat"  # as if something else had set this
-    _save_settings(s)
+    save_settings(s)
 
     r = client.patch("/api/settings", json={"theme": "light"})
     assert r.status_code == 200, r.text
@@ -64,11 +64,11 @@ async def test_concurrent_renderer_patch_and_agent_write_both_survive(reset_sett
     """The renderer PATCHes one field while an autonomous agent writes another,
     at the same time. Both must land: the renderer never sends the agent's field,
     so it can't clobber it, and both reads happen fresh under the shared lock."""
-    from backend.apps.settings.settings import load_settings, _save_settings
+    from backend.apps.settings.settings import load_settings, save_settings
     base = load_settings()
     base.theme = "dark"
     base.default_mode = "agent"
-    _save_settings(base)
+    save_settings(base)
 
     transport = httpx.ASGITransport(app=app)
     async with httpx.AsyncClient(transport=transport, base_url="http://test", headers=_auth_headers()) as client:

backend/tests/test_settings_server_owned.py

@@ -28,11 +28,11 @@ def client():
 
 @pytest.fixture
 def reset_settings():
-    from backend.apps.settings.settings import load_settings, _save_settings
+    from backend.apps.settings.settings import load_settings, save_settings
 
     original = load_settings().model_copy(deep=True)
     yield
-    _save_settings(original)
+    save_settings(original)
 
 
 def _activate_pro(client, token="repro-bearer-0123456789abcdef"):

backend/tests/test_v2_invariants.py

@@ -1315,18 +1315,18 @@ def test_apply_context_window_silent_on_unknown_model():
 def test_estimate_pdf_tokens_floors_empty_pdf_at_byte_heuristic():
     """A truly empty / minimal PDF still returns a non-zero estimate so
     the dry-run guard doesn't allow many tiny PDFs through silently."""
-    from backend.apps.settings.settings import _estimate_pdf_tokens
-    assert _estimate_pdf_tokens(b"") >= 1_000
-    assert _estimate_pdf_tokens(b"%PDF-1.4\n") >= 1_000
+    from backend.apps.settings.settings import estimate_pdf_tokens
+    assert estimate_pdf_tokens(b"") >= 1_000
+    assert estimate_pdf_tokens(b"%PDF-1.4\n") >= 1_000
 
 
 def test_estimate_pdf_tokens_takes_max_of_pages_and_bytes():
     """An image-heavy PDF with low page count should still report high
     tokens via the byte-size signal; we never under-report."""
-    from backend.apps.settings.settings import _estimate_pdf_tokens
+    from backend.apps.settings.settings import estimate_pdf_tokens
     # 8MB PDF with 1 page (image-heavy), byte heuristic should dominate.
     fake = b"%PDF-1.4\n/Type /Pages /Count 1\n" + b"X" * (8 * 1024 * 1024)
-    tokens = _estimate_pdf_tokens(fake)
+    tokens = estimate_pdf_tokens(fake)
     # byte heuristic: 8MB / 80 = 100k tokens > pages * 750 = 750
     assert tokens >= 100_000
 
@@ -1334,9 +1334,9 @@ def test_estimate_pdf_tokens_takes_max_of_pages_and_bytes():
 def test_estimate_pdf_tokens_caps_malformed_count():
     """A PDF with /Count 999999 (malformed or hostile) does NOT bypass
     the 10k pages sanity cap; falls through to byte heuristic instead."""
-    from backend.apps.settings.settings import _estimate_pdf_tokens
+    from backend.apps.settings.settings import estimate_pdf_tokens
     fake = b"%PDF-1.4\n/Type /Pages /Count 999999\n"
-    t = _estimate_pdf_tokens(fake)
+    t = estimate_pdf_tokens(fake)
     # Should NOT be 999999 * 750 = 750 million.
     assert t < 50_000_000
 
@@ -1526,20 +1526,20 @@ def test_resolve_attachments_uses_os_path_basename_for_windows_paths():
 def test_sniff_file_kind_consistent_across_platforms():
     """The sniffer reads bytes, never paths. So platform doesn't matter
     for the classification logic, same bytes → same kind on Windows/Mac/Linux."""
-    from backend.apps.settings.settings import _sniff_file_kind
-    assert _sniff_file_kind(b"%PDF-1.4\n", "x.pdf") == ("pdf", "application/pdf")
-    assert _sniff_file_kind(b"\x89PNG\r\n\x1a\n", "x.png") == ("image", "image/png")
-    assert _sniff_file_kind(b"PK\x03\x04", "x.zip") == ("binary", None)
-    assert _sniff_file_kind(b"MZ\x90\x00", "x.exe") == ("binary", None)
-    assert _sniff_file_kind(b"hello world", "x.txt") == ("text", "text/plain")
+    from backend.apps.settings.settings import sniff_file_kind
+    assert sniff_file_kind(b"%PDF-1.4\n", "x.pdf") == ("pdf", "application/pdf")
+    assert sniff_file_kind(b"\x89PNG\r\n\x1a\n", "x.png") == ("image", "image/png")
+    assert sniff_file_kind(b"PK\x03\x04", "x.zip") == ("binary", None)
+    assert sniff_file_kind(b"MZ\x90\x00", "x.exe") == ("binary", None)
+    assert sniff_file_kind(b"hello world", "x.txt") == ("text", "text/plain")
 
 
 def test_estimate_pdf_tokens_consistent_across_platforms():
     """Same byte-level math regardless of OS."""
-    from backend.apps.settings.settings import _estimate_pdf_tokens
+    from backend.apps.settings.settings import estimate_pdf_tokens
     # 5MB PDF should always estimate ≥ 5MB/80 = 65536 tokens.
     fake = b"%PDF-1.4\n" + b"X" * (5 * 1024 * 1024)
-    assert _estimate_pdf_tokens(fake) >= 65000
+    assert estimate_pdf_tokens(fake) >= 65000
 
 
 def test_sniff_handles_windows_style_backslash_path_string():