diff --git a/backend/apps/agents/agents.py b/backend/apps/agents/agents.py
index 59ad351d..02d144d9 100644
--- a/backend/apps/agents/agents.py
+++ b/backend/apps/agents/agents.py
@@ -13,7 +13,7 @@ import logging
 logger = logging.getLogger(__name__)
 
 # Dedup concurrent generate-group-meta calls; collapses the 429 thundering herd by sharing one upstream Future per (session, group).
-_group_meta_inflight: dict[tuple[str, str], asyncio.Future] = {}
+p_group_meta_inflight: dict[tuple[str, str], asyncio.Future] = {}
 
 @asynccontextmanager
 async def agents_lifespan():
@@ -164,7 +164,7 @@ async def generate_group_meta(session_id: str, body: dict):
     is_refinement = body.get("is_refinement", False)
     key = (session_id, group_id)
     if not is_refinement:
-        existing = _group_meta_inflight.get(key)
+        existing = p_group_meta_inflight.get(key)
         if existing is not None and not existing.done():
             try:
                 return await existing
@@ -174,7 +174,7 @@ async def generate_group_meta(session_id: str, body: dict):
 
     future: asyncio.Future = asyncio.get_event_loop().create_future()
     if not is_refinement:
-        _group_meta_inflight[key] = future
+        p_group_meta_inflight[key] = future
     try:
         result = await agent_manager.generate_group_meta(
             session_id,
@@ -191,8 +191,8 @@ async def generate_group_meta(session_id: str, body: dict):
             future.set_exception(e)
         raise
     finally:
-        if not is_refinement and _group_meta_inflight.get(key) is future:
-            _group_meta_inflight.pop(key, None)
+        if not is_refinement and p_group_meta_inflight.get(key) is future:
+            p_group_meta_inflight.pop(key, None)
 
 @agents.router.patch("/sessions/{session_id}")
 async def update_session(session_id: str, body: dict):
@@ -383,10 +383,10 @@ async def subscriptions_connect(body: dict):
             raise HTTPException(status_code=503, detail="9Router not available. Please install Node.js.")
 
     # Reconnecting gemini-cli must wipe antigravity; registry prefers AG and a stale AG token would 400 after gemini-cli refreshes.
-    cascade = _PROVIDER_CASCADE_REMOVES.get(provider, [])
+    cascade = P_PROVIDER_CASCADE_REMOVES.get(provider, [])
     if cascade:
         try:
-            await _delete_provider_connections(cascade)
+            await p_delete_provider_connections(cascade)
         except Exception:
             pass
 
@@ -809,12 +809,12 @@ async def list_models():
 
 
 # gemini-cli and antigravity are two Google OAuth lanes; registry prefers AG, so we cascade-wipe AG when reconnecting gemini-cli to avoid stale-AG 400s. One-directional: AG operations MUST NOT cascade back.
-_PROVIDER_CASCADE_REMOVES: dict[str, list[str]] = {
+P_PROVIDER_CASCADE_REMOVES: dict[str, list[str]] = {
     "gemini-cli": ["antigravity"],
 }
 
 
-async def _delete_provider_connections(providers: list[str]) -> int:
+async def p_delete_provider_connections(providers: list[str]) -> int:
     """Delete 9Router connections in `providers`; returns count removed, silent on 9Router unreachable."""
     import httpx
     from backend.apps.nine_router import NINE_ROUTER_API, get_providers
@@ -842,8 +842,8 @@ async def subscriptions_disconnect(body: dict):
         raise HTTPException(status_code=400, detail="provider required")
 
     try:
-        to_remove = [provider, *_PROVIDER_CASCADE_REMOVES.get(provider, [])]
-        removed = await _delete_provider_connections(to_remove)
+        to_remove = [provider, *P_PROVIDER_CASCADE_REMOVES.get(provider, [])]
+        removed = await p_delete_provider_connections(to_remove)
         if removed:
             from backend.apps.service.client import sync as _sync
             from backend.apps.settings.settings import load_settings
diff --git a/backend/apps/agents/core/error_classify.py b/backend/apps/agents/core/error_classify.py
index 05830c9a..78716188 100644
--- a/backend/apps/agents/core/error_classify.py
+++ b/backend/apps/agents/core/error_classify.py
@@ -7,7 +7,7 @@ from typeguard import typechecked
 # error string to telemetry. own_key mode means the subprocess stderr can echo
 # the user's OWN provider key, so this scrub is the wall between a diagnostic
 # and a key leak; over-redacting is fine, leaking is not.
-_TELEMETRY_SECRET_PATTERNS = (
+P_TELEMETRY_SECRET_PATTERNS = (
     re.compile(r"sk-ant-[A-Za-z0-9_\-]{12,}"),
     re.compile(r"sk-[A-Za-z0-9_\-]{16,}"),
     re.compile(r"AIza[A-Za-z0-9_\-]{20,}"),
@@ -23,7 +23,7 @@ def redact_for_telemetry(text: str, *, limit: int = 2000) -> str:
     error/stderr string goes through here before it leaves the machine."""
     if not text:
         return ""
-    for pat in _TELEMETRY_SECRET_PATTERNS:
+    for pat in P_TELEMETRY_SECRET_PATTERNS:
         text = pat.sub("[redacted]", text)
     return text[-limit:]
 
@@ -51,7 +51,7 @@ TRANSIENT_CAPACITY_PATTERNS = re.compile(
 # zero tokens. That is NOT auth, reconnecting won't help, the request shape is
 # wrong, so we classify it apart and stop the catch-all from showing a
 # "reconnect your subscription" card for a tool-schema 400.
-_TRANSLATION_ERROR_PATTERNS = re.compile(
+P_TRANSLATION_ERROR_PATTERNS = re.compile(
     r"(?:function_declarations"
     r"|invalid_argument"
     r"|invalid\s+json\s+payload"
@@ -128,7 +128,7 @@ def is_translation_error(exc: BaseException, extra_text: str = "") -> bool:
     combined = f"{exc!s}\n{extra_text}".strip()
     if not combined:
         return False
-    return bool(_TRANSLATION_ERROR_PATTERNS.search(combined))
+    return bool(P_TRANSLATION_ERROR_PATTERNS.search(combined))
 
 
 @typechecked
diff --git a/backend/apps/auth/router.py b/backend/apps/auth/router.py
index 49fd4f13..9c912a34 100644
--- a/backend/apps/auth/router.py
+++ b/backend/apps/auth/router.py
@@ -40,14 +40,14 @@ async def auth_lifespan():
 auth = SubApp("auth", auth_lifespan)
 
 
-def _proxy_url() -> str:
+def p_proxy_url() -> str:
     settings_obj = load_settings()
     url = (getattr(settings_obj, "openswarm_proxy_url", None)
            or OPENSWARM_DEFAULT_PROXY_URL)
     return url.rstrip("/")
 
 
-async def _sync_pro_routing(settings_obj) -> None:
+async def p_sync_pro_routing(settings_obj) -> None:
     """Mirror connection state into 9Router's Claude lane; sign-in can flip a
     paying user into pro mode and sign-out must tear the lane down so a
     revoked bearer doesn't linger in the router."""
@@ -58,7 +58,7 @@ async def _sync_pro_routing(settings_obj) -> None:
         logger.debug("pro routing sync skipped: %s", e)
 
 
-def _sync_identity_to_service(settings_obj) -> None:
+def p_sync_identity_to_service(settings_obj) -> None:
     """Push user_id + email + signin_method into the service-sync identify
     pipeline so every event from this user has the right Person properties."""
     try:
@@ -101,7 +101,7 @@ async def signin_activate(body: SigninActivateRequest):
     if not body.token or len(body.token) < 16:
         raise HTTPException(status_code=400, detail="Invalid token")
 
-    proxy = _proxy_url()
+    proxy = p_proxy_url()
     try:
         async with httpx.AsyncClient(timeout=10.0) as client:
             r = await client.post(
@@ -157,8 +157,8 @@ async def signin_activate(body: SigninActivateRequest):
         settings_obj.openswarm_proxy_url = proxy
 
     await save_settings_async(settings_obj)
-    _sync_identity_to_service(settings_obj)
-    await _sync_pro_routing(settings_obj)
+    p_sync_identity_to_service(settings_obj)
+    await p_sync_pro_routing(settings_obj)
 
     return {
         "ok": True,
@@ -185,7 +185,7 @@ async def signout():
     """
     settings_obj = load_settings()
     bearer = getattr(settings_obj, "openswarm_bearer_token", None)
-    proxy = _proxy_url()
+    proxy = p_proxy_url()
     if bearer:
         try:
             async with httpx.AsyncClient(timeout=5.0) as client:
@@ -245,6 +245,6 @@ async def signout():
     settings_obj.openswarm_subscription_expires = None
     settings_obj.openswarm_usage_cached = None
     await save_settings_async(settings_obj)
-    _sync_identity_to_service(settings_obj)
-    await _sync_pro_routing(settings_obj)
+    p_sync_identity_to_service(settings_obj)
+    await p_sync_pro_routing(settings_obj)
     return {"ok": True}
diff --git a/backend/apps/discord_mcp_shim/server.py b/backend/apps/discord_mcp_shim/server.py
index a806841f..0c49758e 100644
--- a/backend/apps/discord_mcp_shim/server.py
+++ b/backend/apps/discord_mcp_shim/server.py
@@ -214,7 +214,7 @@ TOOLS = [
 
 # -- HTTP plumbing ---------------------------------------------------------
 
-def _call(
+def p_call(
     method: str,
     path: str,
     *,
@@ -267,17 +267,17 @@ def _call(
         return 0, f"Request failed: {e!r}"
 
 
-def _err(text: str) -> dict:
+def p_err(text: str) -> dict:
     return {"content": [{"type": "text", "text": f"Error: {text}"}], "isError": True}
 
 
-def _ok(payload) -> dict:
+def p_ok(payload) -> dict:
     if isinstance(payload, str):
         return {"content": [{"type": "text", "text": payload}]}
     return {"content": [{"type": "text", "text": json.dumps(payload, indent=2, default=str)}]}
 
 
-def _check_guild(guild_id: str) -> str | None:
+def p_check_guild(guild_id: str) -> str | None:
     """Return an error string if guild_id is outside the user-authorized set, else None.
 
     The set is sourced from OPENSWARM_DISCORD_GUILD_IDS env var (CSV) which
@@ -299,69 +299,69 @@ def _check_guild(guild_id: str) -> str | None:
 
 def handle_tool_call(name: str, args: dict) -> dict:
     if name == "discord_login":
-        status, body = _call("GET", "/users/@me/guilds")
+        status, body = p_call("GET", "/users/@me/guilds")
         if status != 200:
-            return _err(f"Discord proxy unreachable (HTTP {status}): {body}")
-        return _ok({"connected": True, "guilds": body})
+            return p_err(f"Discord proxy unreachable (HTTP {status}): {body}")
+        return p_ok({"connected": True, "guilds": body})
 
     if name == "discord_get_server_info":
         gid = str(args.get("guild_id", ""))
-        if (e := _check_guild(gid)): return _err(e)
-        status, body = _call("GET", f"/guilds/{gid}")
-        return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}")
+        if (e := p_check_guild(gid)): return p_err(e)
+        status, body = p_call("GET", f"/guilds/{gid}")
+        return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}")
 
     if name == "discord_list_channels":
         gid = str(args.get("guild_id", ""))
-        if (e := _check_guild(gid)): return _err(e)
-        status, body = _call("GET", f"/guilds/{gid}/channels")
-        return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}")
+        if (e := p_check_guild(gid)): return p_err(e)
+        status, body = p_call("GET", f"/guilds/{gid}/channels")
+        return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}")
 
     if name == "discord_create_text_channel":
         gid = str(args.get("guild_id", ""))
-        if (e := _check_guild(gid)): return _err(e)
+        if (e := p_check_guild(gid)): return p_err(e)
         payload: dict = {"name": args.get("name", ""), "type": 0}
         if args.get("parent_id"): payload["parent_id"] = args["parent_id"]
         if args.get("topic"): payload["topic"] = args["topic"]
-        status, body = _call("POST", f"/guilds/{gid}/channels", body=payload)
-        return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}")
+        status, body = p_call("POST", f"/guilds/{gid}/channels", body=payload)
+        return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}")
 
     if name == "discord_create_category":
         gid = str(args.get("guild_id", ""))
-        if (e := _check_guild(gid)): return _err(e)
-        status, body = _call("POST", f"/guilds/{gid}/channels", body={"name": args.get("name", ""), "type": 4})
-        return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}")
+        if (e := p_check_guild(gid)): return p_err(e)
+        status, body = p_call("POST", f"/guilds/{gid}/channels", body={"name": args.get("name", ""), "type": 4})
+        return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}")
 
     if name == "discord_edit_category":
         cid = str(args.get("channel_id", ""))
         payload: dict = {}
         if args.get("name"): payload["name"] = args["name"]
-        status, body = _call("PATCH", f"/channels/{cid}", body=payload)
-        return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}")
+        status, body = p_call("PATCH", f"/channels/{cid}", body=payload)
+        return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}")
 
     if name == "discord_delete_category" or name == "discord_delete_channel":
         cid = str(args.get("channel_id", ""))
-        status, body = _call("DELETE", f"/channels/{cid}")
-        return _ok({"deleted": True}) if status in (200, 204) else _err(f"HTTP {status}: {body}")
+        status, body = p_call("DELETE", f"/channels/{cid}")
+        return p_ok({"deleted": True}) if status in (200, 204) else p_err(f"HTTP {status}: {body}")
 
     if name == "discord_send":
         cid = str(args.get("channel_id", ""))
         content = str(args.get("content", ""))
-        status, body = _call("POST", f"/channels/{cid}/messages", body={"content": content})
-        return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}")
+        status, body = p_call("POST", f"/channels/{cid}/messages", body={"content": content})
+        return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}")
 
     if name == "discord_read_messages":
         cid = str(args.get("channel_id", ""))
         limit = max(1, min(int(args.get("limit", 50) or 50), 100))
-        status, body = _call("GET", f"/channels/{cid}/messages", query={"limit": limit})
-        return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}")
+        status, body = p_call("GET", f"/channels/{cid}/messages", query={"limit": limit})
+        return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}")
 
     if name == "discord_add_reaction":
         cid = str(args.get("channel_id", ""))
         mid = str(args.get("message_id", ""))
         emoji = str(args.get("emoji", ""))
         # Discord's URL needs the emoji urlencoded; passes through.
-        status, body = _call("PUT", f"/channels/{cid}/messages/{mid}/reactions/{urllib.parse.quote(emoji, safe='')}/@me")
-        return _ok({"added": emoji}) if status in (200, 204) else _err(f"HTTP {status}: {body}")
+        status, body = p_call("PUT", f"/channels/{cid}/messages/{mid}/reactions/{urllib.parse.quote(emoji, safe='')}/@me")
+        return p_ok({"added": emoji}) if status in (200, 204) else p_err(f"HTTP {status}: {body}")
 
     if name == "discord_add_multiple_reactions":
         cid = str(args.get("channel_id", ""))
@@ -369,46 +369,46 @@ def handle_tool_call(name: str, args: dict) -> dict:
         emojis = args.get("emojis", []) or []
         results = []
         for e in emojis:
-            status, body = _call("PUT", f"/channels/{cid}/messages/{mid}/reactions/{urllib.parse.quote(str(e), safe='')}/@me")
+            status, body = p_call("PUT", f"/channels/{cid}/messages/{mid}/reactions/{urllib.parse.quote(str(e), safe='')}/@me")
             results.append({"emoji": e, "ok": status in (200, 204), "status": status})
-        return _ok({"reactions": results})
+        return p_ok({"reactions": results})
 
     if name == "discord_get_forum_channels":
         gid = str(args.get("guild_id", ""))
-        if (e := _check_guild(gid)): return _err(e)
-        status, body = _call("GET", f"/guilds/{gid}/channels")
-        if status != 200: return _err(f"HTTP {status}: {body}")
+        if (e := p_check_guild(gid)): return p_err(e)
+        status, body = p_call("GET", f"/guilds/{gid}/channels")
+        if status != 200: return p_err(f"HTTP {status}: {body}")
         # Filter to type 15 (forum). Discord channel types reference:
         #   GUILD_FORUM = 15
         forums = [ch for ch in (body or []) if isinstance(ch, dict) and ch.get("type") == 15]
-        return _ok(forums)
+        return p_ok(forums)
 
     if name == "discord_create_forum_post":
         fid = str(args.get("forum_id", ""))
-        status, body = _call("POST", f"/channels/{fid}/threads", body={
+        status, body = p_call("POST", f"/channels/{fid}/threads", body={
             "name": args.get("name", ""),
             "message": {"content": args.get("content", "")},
         })
-        return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}")
+        return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}")
 
     if name == "discord_get_forum_post":
         cid = str(args.get("channel_id", ""))
         mid = str(args.get("message_id", ""))
-        status, body = _call("GET", f"/channels/{cid}/messages/{mid}")
-        return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}")
+        status, body = p_call("GET", f"/channels/{cid}/messages/{mid}")
+        return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}")
 
     if name == "discord_reply_to_forum":
         cid = str(args.get("channel_id", ""))
         content = str(args.get("content", ""))
-        status, body = _call("POST", f"/channels/{cid}/messages", body={"content": content})
-        return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}")
+        status, body = p_call("POST", f"/channels/{cid}/messages", body={"content": content})
+        return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}")
 
-    return _err(f"Unknown tool: {name}")
+    return p_err(f"Unknown tool: {name}")
 
 
 # -- JSON-RPC stdio loop ---------------------------------------------------
 
-def _send(id_, result=None, error=None):
+def p_send(id_, result=None, error=None):
     msg = {"jsonrpc": "2.0", "id": id_}
     if error is not None:
         msg["error"] = error
@@ -433,7 +433,7 @@ def main():
         params = msg.get("params", {}) or {}
 
         if method == "initialize":
-            _send(id_, {
+            p_send(id_, {
                 "protocolVersion": "2024-11-05",
                 "capabilities": {"tools": {}},
                 "serverInfo": {"name": "openswarm-discord", "version": "1.0.0"},
@@ -441,18 +441,18 @@ def main():
         elif method == "notifications/initialized":
             pass
         elif method == "tools/list":
-            _send(id_, {"tools": TOOLS})
+            p_send(id_, {"tools": TOOLS})
         elif method == "tools/call":
             name = params.get("name", "")
             args = params.get("arguments", {}) or {}
             try:
-                _send(id_, handle_tool_call(name, args))
+                p_send(id_, handle_tool_call(name, args))
             except Exception as e:
-                _send(id_, _err(f"shim crashed: {e!r}"))
+                p_send(id_, p_err(f"shim crashed: {e!r}"))
         elif method == "ping":
-            _send(id_, {})
+            p_send(id_, {})
         elif id_ is not None:
-            _send(id_, error={"code": -32601, "message": f"Method not found: {method}"})
+            p_send(id_, error={"code": -32601, "message": f"Method not found: {method}"})
 
 
 if __name__ == "__main__":
diff --git a/backend/apps/google_workspace_mcp_shim/run.py b/backend/apps/google_workspace_mcp_shim/run.py
index 189b107f..524979e2 100644
--- a/backend/apps/google_workspace_mcp_shim/run.py
+++ b/backend/apps/google_workspace_mcp_shim/run.py
@@ -24,7 +24,7 @@ from google.oauth2.credentials import Credentials
 
 
 @functools.lru_cache(maxsize=1)
-def _patched_get_credentials():
+def p_patched_get_credentials():
     refresh_token = os.environ.get("GOOGLE_WORKSPACE_REFRESH_TOKEN")
     if not refresh_token:
         raise ValueError("GOOGLE_WORKSPACE_REFRESH_TOKEN env var is required")
@@ -40,7 +40,7 @@ def _patched_get_credentials():
     )
 
 
-gauth.get_credentials = _patched_get_credentials
+gauth.get_credentials = p_patched_get_credentials
 
 
 from google_workspace_mcp import __main__ as _gw_main  # noqa: E402,F401
diff --git a/backend/apps/service/buffer.py b/backend/apps/service/buffer.py
index 21a9dc3f..e9d185a1 100644
--- a/backend/apps/service/buffer.py
+++ b/backend/apps/service/buffer.py
@@ -26,18 +26,18 @@ logger = logging.getLogger(__name__)
 # Cap the spool at 50 MB on disk. SQLite's overhead means the actual ceiling
 # on retained payloads is somewhat smaller, which is fine; this is a
 # best-effort cushion, not a guaranteed retention window.
-_MAX_BYTES = 50 * 1024 * 1024
+P_MAX_BYTES = 50 * 1024 * 1024
 
 # Trim 25% when we cross the cap so we don't trim on every insert.
-_TRIM_TARGET_FRACTION = 0.75
+P_TRIM_TARGET_FRACTION = 0.75
 
-_lock = threading.Lock()
+p_lock = threading.Lock()
 
 
 @contextmanager
-def _conn(spool_path: str) -> Iterator[sqlite3.Connection]:
+def conn(spool_path: str) -> Iterator[sqlite3.Connection]:
     """Open a connection that auto-commits and ensures the table exists.
-    Caller holds `_lock` for the duration of the context."""
+    Caller holds `p_lock` for the duration of the context."""
     os.makedirs(os.path.dirname(spool_path), exist_ok=True)
     c = sqlite3.connect(spool_path, isolation_level=None, timeout=5.0)
     try:
@@ -58,7 +58,7 @@ def enqueue(spool_path: str, kind: str, payload: dict, *, now: float) -> None:
     """Append a submission to the spool. Drops the oldest if the spool is
     over the byte cap."""
     body = json.dumps(payload, separators=(",", ":"), default=str)
-    with _lock, _conn(spool_path) as c:
+    with p_lock, conn(spool_path) as c:
         c.execute(
             "INSERT INTO spool (kind, payload, created_at) VALUES (?, ?, ?)",
             (kind, body, now),
@@ -68,8 +68,8 @@ def enqueue(spool_path: str, kind: str, payload: dict, *, now: float) -> None:
             size = os.path.getsize(spool_path)
         except OSError:
             size = 0
-        if size > _MAX_BYTES:
-            target = int(_MAX_BYTES * _TRIM_TARGET_FRACTION)
+        if size > P_MAX_BYTES:
+            target = int(P_MAX_BYTES * P_TRIM_TARGET_FRACTION)
             # Delete oldest rows until we're back under target. Use a
             # reasonable batch size so we don't block forever.
             dropped = 0
@@ -86,11 +86,11 @@ def enqueue(spool_path: str, kind: str, payload: dict, *, now: float) -> None:
                 if new_size <= target:
                     break
             if dropped:
-                logger.warning("Spool over %d MB cap; dropped %d oldest entries", _MAX_BYTES // (1024 * 1024), dropped)
+                logger.warning("Spool over %d MB cap; dropped %d oldest entries", P_MAX_BYTES // (1024 * 1024), dropped)
             # VACUUM is expensive; only run if we still appear oversized after
             # trimming, otherwise free pages get reused on next insert.
             try:
-                if os.path.getsize(spool_path) > _MAX_BYTES:
+                if os.path.getsize(spool_path) > P_MAX_BYTES:
                     c.execute("VACUUM")
             except (OSError, sqlite3.DatabaseError):
                 pass
@@ -102,7 +102,7 @@ def drain(spool_path: str, batch_size: int = 50) -> list[tuple[int, str, dict]]:
     cloud accepts them."""
     if not os.path.exists(spool_path):
         return []
-    with _lock, _conn(spool_path) as c:
+    with p_lock, conn(spool_path) as c:
         rows = c.execute(
             "SELECT id, kind, payload FROM spool ORDER BY id ASC LIMIT ?",
             (batch_size,),
@@ -113,7 +113,7 @@ def drain(spool_path: str, batch_size: int = 50) -> list[tuple[int, str, dict]]:
             out.append((rid, kind, json.loads(body)))
         except json.JSONDecodeError:
             # Corrupt row; discard so it doesn't block draining behind it.
-            with _lock, _conn(spool_path) as c:
+            with p_lock, conn(spool_path) as c:
                 c.execute("DELETE FROM spool WHERE id = ?", (rid,))
             logger.warning("Dropped corrupt spool row id=%s", rid)
     return out
@@ -123,7 +123,7 @@ def acknowledge(spool_path: str, ids: list[int]) -> None:
     """Remove rows the cloud has accepted."""
     if not ids:
         return
-    with _lock, _conn(spool_path) as c:
+    with p_lock, conn(spool_path) as c:
         c.executemany("DELETE FROM spool WHERE id = ?", [(i,) for i in ids])
 
 
@@ -131,12 +131,12 @@ def count(spool_path: str) -> int:
     """Return the number of pending entries. Used for tests + debug UI."""
     if not os.path.exists(spool_path):
         return 0
-    with _lock, _conn(spool_path) as c:
+    with p_lock, conn(spool_path) as c:
         row = c.execute("SELECT COUNT(*) FROM spool").fetchone()
     return int(row[0]) if row else 0
 
 
 def clear(spool_path: str) -> None:
     """Delete all pending entries. Tests + manual reset only."""
-    with _lock, _conn(spool_path) as c:
+    with p_lock, conn(spool_path) as c:
         c.execute("DELETE FROM spool")
diff --git a/backend/apps/service/client.py b/backend/apps/service/client.py
index 63298184..d823fa4a 100644
--- a/backend/apps/service/client.py
+++ b/backend/apps/service/client.py
@@ -34,26 +34,26 @@ from backend.apps.service.version import APP_VERSION
 
 logger = logging.getLogger(__name__)
 
-_DEFAULT_BASE = "https://api.openswarm.com"
-_PATH_BY_KIND = {
+P_DEFAULT_BASE = "https://api.openswarm.com"
+P_PATH_BY_KIND = {
     "state": "/api/service/state",
     "session": "/api/service/sync",
     "diagnostic": "/api/service/diagnostics",
     "event": "/api/service/event",
 }
 
-_TIMEOUT_SECONDS = 5.0
-_MAX_INFLIGHT = 16
+P_TIMEOUT_SECONDS = 5.0
+P_MAX_INFLIGHT = 16
 
-_test_sink: Optional[Any] = None
-_install_id: Optional[str] = None
-_user_id: Optional[str] = None
-_inflight = 0
-_inflight_lock = asyncio.Lock()
-_drain_lock = asyncio.Lock()
+test_sink: Optional[Any] = None
+install_id: Optional[str] = None
+p_user_id: Optional[str] = None
+p_inflight = 0
+p_inflight_lock = asyncio.Lock()
+p_drain_lock = asyncio.Lock()
 
 
-def _spool_path() -> str:
+def spool_path() -> str:
     try:
         from backend.config.paths import SETTINGS_DIR
         return os.path.join(SETTINGS_DIR, "service_spool.db")
@@ -63,14 +63,14 @@ def _spool_path() -> str:
 
 def set_test_sink(fn: Optional[Any]) -> None:
     """Test seam; receives every submission instead of the network."""
-    global _test_sink
-    _test_sink = fn
+    global test_sink
+    test_sink = fn
 
 
-def _get_install_id() -> str:
-    global _install_id
-    if _install_id:
-        return _install_id
+def p_get_install_id() -> str:
+    global install_id
+    if install_id:
+        return install_id
     try:
         from backend.apps.settings.store import load_settings, save_settings
         s = load_settings()
@@ -79,16 +79,16 @@ def _get_install_id() -> str:
             iid = uuid4().hex
             s.installation_id = iid
             save_settings(s)
-        _install_id = iid
+        install_id = iid
     except Exception:
-        _install_id = uuid4().hex
-    return _install_id
+        install_id = uuid4().hex
+    return install_id
 
 
-def _get_user_id() -> Optional[str]:
-    global _user_id
-    if _user_id:
-        return _user_id
+def p_get_user_id() -> Optional[str]:
+    global p_user_id
+    if p_user_id:
+        return p_user_id
     try:
         from backend.apps.settings.store import load_settings
         s = load_settings()
@@ -108,11 +108,11 @@ def _get_user_id() -> Optional[str]:
 
 
 def set_user_id(uid: Optional[str]) -> None:
-    global _user_id
-    _user_id = uid or None
+    global p_user_id
+    p_user_id = uid or None
 
 
-def _is_enabled(kind: str) -> bool:
+def p_is_enabled(kind: str) -> bool:
     """Honour user opt-out. Diagnostic always flows (errors block usability);
     state + session honour the toggle."""
     if kind == "diagnostic":
@@ -130,10 +130,10 @@ def _is_enabled(kind: str) -> bool:
         return True
 
 
-def _envelope() -> dict:
+def p_envelope() -> dict:
     """Identity + environment metadata stamped on every submission."""
-    env: dict[str, Any] = {"install_id": _get_install_id()}
-    uid = _get_user_id()
+    env: dict[str, Any] = {"install_id": p_get_install_id()}
+    uid = p_get_user_id()
     if uid:
         env["user_id"] = uid
     try:
@@ -182,20 +182,20 @@ def _envelope() -> dict:
     return env
 
 
-def _base_url() -> str:
+def p_base_url() -> str:
     try:
         from backend.apps.settings.store import load_settings
         from backend.apps.settings.credentials import OPENSWARM_DEFAULT_PROXY_URL
         s = load_settings()
         return (getattr(s, "openswarm_proxy_url", None) or OPENSWARM_DEFAULT_PROXY_URL).rstrip("/")
     except Exception:
-        return _DEFAULT_BASE
+        return P_DEFAULT_BASE
 
 
-async def _post(path: str, body: dict) -> int | None:
-    url = f"{_base_url()}{path}"
+async def p_post(path: str, body: dict) -> int | None:
+    url = f"{p_base_url()}{path}"
     try:
-        async with httpx.AsyncClient(timeout=_TIMEOUT_SECONDS) as c:
+        async with httpx.AsyncClient(timeout=P_TIMEOUT_SECONDS) as c:
             r = await c.post(url, json=body)
         return r.status_code
     except Exception as e:
@@ -203,42 +203,42 @@ async def _post(path: str, body: dict) -> int | None:
         return None
 
 
-def _delivered(status: int | None) -> bool:
+def p_delivered(status: int | None) -> bool:
     return status is not None and 200 <= status < 300
 
 
 # 429/timeouts/5xx/network are worth retrying; other 4xx means the payload itself is rejected and retrying forever would just poison the spool.
-def _retryable(status: int | None) -> bool:
+def p_retryable(status: int | None) -> bool:
     return status is None or status >= 500 or status in (408, 429)
 
 
-async def _post_or_spool(path: str, body: dict, kind: str) -> None:
-    global _inflight
-    if _test_sink is not None:
+async def p_post_or_spool(path: str, body: dict, kind: str) -> None:
+    global p_inflight
+    if test_sink is not None:
         try:
-            _test_sink(kind, body)
+            test_sink(kind, body)
         except Exception as e:
             logger.debug("test sink raised: %s", e)
         return
-    async with _inflight_lock:
-        if _inflight >= _MAX_INFLIGHT:
-            buffer.enqueue(_spool_path(), f"{kind}:{path}", body, now=time.time())
+    async with p_inflight_lock:
+        if p_inflight >= P_MAX_INFLIGHT:
+            buffer.enqueue(spool_path(), f"{kind}:{path}", body, now=time.time())
             return
-        _inflight += 1
+        p_inflight += 1
     try:
-        status = await _post(path, body)
-        if _retryable(status):
-            buffer.enqueue(_spool_path(), f"{kind}:{path}", body, now=time.time())
-        elif not _delivered(status):
+        status = await p_post(path, body)
+        if p_retryable(status):
+            buffer.enqueue(spool_path(), f"{kind}:{path}", body, now=time.time())
+        elif not p_delivered(status):
             logger.warning("service POST %s rejected with HTTP %s; payload dropped", path, status)
     finally:
-        async with _inflight_lock:
-            _inflight = max(0, _inflight - 1)
+        async with p_inflight_lock:
+            p_inflight = max(0, p_inflight - 1)
 
 
 async def drain_spool(batch_size: int = 50) -> int:
-    async with _drain_lock:
-        entries = buffer.drain(_spool_path(), batch_size=batch_size)
+    async with p_drain_lock:
+        entries = buffer.drain(spool_path(), batch_size=batch_size)
         if not entries:
             return 0
         succeeded: list[int] = []
@@ -247,16 +247,16 @@ async def drain_spool(batch_size: int = 50) -> int:
             if not path:
                 succeeded.append(rid)
                 continue
-            status = await _post(path, body)
-            if _delivered(status):
+            status = await p_post(path, body)
+            if p_delivered(status):
                 succeeded.append(rid)
-            elif _retryable(status):
+            elif p_retryable(status):
                 break
             else:
                 logger.warning("service replay %s rejected with HTTP %s; dropping spooled row", path, status)
                 succeeded.append(rid)
         if succeeded:
-            buffer.acknowledge(_spool_path(), succeeded)
+            buffer.acknowledge(spool_path(), succeeded)
         return len(succeeded)
 
 
@@ -264,7 +264,7 @@ async def drain_spool(batch_size: int = 50) -> int:
 # Public API
 # --------------------------------------------------------------------------
 
-def _log(kind: str, payload: dict) -> None:
+def p_log(kind: str, payload: dict) -> None:
     """Append to the rolling operational log for diagnostics."""
     try:
         from backend.apps.service.ring_buffer import record
@@ -288,26 +288,26 @@ def sync(data: dict | None = None) -> None:
     Fire-and-forget; never raises.
     """
     payload = data or {}
-    if not _is_enabled("state"):
+    if not p_is_enabled("state"):
         return
     body = {
-        "client_state": _envelope(),
+        "client_state": p_envelope(),
         "d": payload,
         "t": time.time(),
         "submission_id": uuid4().hex,
     }
-    _log("s", payload)
-    if _test_sink is not None:
+    p_log("s", payload)
+    if test_sink is not None:
         try:
-            _test_sink("s", body)
+            test_sink("s", body)
         except Exception as e:
             logger.debug("test sink raised: %s", e)
         return
-    _schedule(_post_or_spool(_DEFAULT_SYNC_PATH, body, "s"))
+    p_schedule(p_post_or_spool(P_DEFAULT_SYNC_PATH, body, "s"))
 
 
 # Internal routing; the cloud has one endpoint for everything.
-_DEFAULT_SYNC_PATH = "/api/service/sync"
+P_DEFAULT_SYNC_PATH = "/api/service/sync"
 
 
 def submit(kind: str, payload: dict) -> None:
@@ -318,7 +318,7 @@ def submit(kind: str, payload: dict) -> None:
     sync(payload)
 
 
-def _schedule(coro) -> None:
+def p_schedule(coro) -> None:
     try:
         loop = asyncio.get_running_loop()
     except RuntimeError:
diff --git a/backend/apps/service/ring_buffer.py b/backend/apps/service/ring_buffer.py
index d9006b6b..50212579 100644
--- a/backend/apps/service/ring_buffer.py
+++ b/backend/apps/service/ring_buffer.py
@@ -6,15 +6,15 @@ import threading
 import time
 from collections import deque
 
-_MAX_SIZE = 50
-_lock = threading.Lock()
-_buffer: deque[dict] = deque(maxlen=_MAX_SIZE)
+P_MAX_SIZE = 50
+p_lock = threading.Lock()
+p_buffer: deque[dict] = deque(maxlen=P_MAX_SIZE)
 
 
 def record(label: str, **meta: str | int | float | None) -> None:
     """Append an entry. Oldest drops when full."""
-    with _lock:
-        _buffer.append({
+    with p_lock:
+        p_buffer.append({
             "l": label,
             "t": time.time(),
             **{k: v for k, v in meta.items() if v is not None},
@@ -23,10 +23,10 @@ def record(label: str, **meta: str | int | float | None) -> None:
 
 def snapshot() -> list[dict]:
     """Return a copy of the current buffer, oldest first."""
-    with _lock:
-        return list(_buffer)
+    with p_lock:
+        return list(p_buffer)
 
 
 def clear() -> None:
-    with _lock:
-        _buffer.clear()
+    with p_lock:
+        p_buffer.clear()
diff --git a/backend/apps/service/service.py b/backend/apps/service/service.py
index eb665946..67ed7cd7 100644
--- a/backend/apps/service/service.py
+++ b/backend/apps/service/service.py
@@ -27,22 +27,22 @@ from fastapi import Body
 from backend.config.Apps import SubApp
 from backend.config.paths import SESSIONS_DIR
 from backend.apps.service import client as svc
-from backend.apps.service.version import APP_VERSION, _read_app_version
+from backend.apps.service.version import APP_VERSION, read_app_version
 
 logger = logging.getLogger(__name__)
 
-_pulse_task: asyncio.Task | None = None
-_drain_task: asyncio.Task | None = None
+p_pulse_task: asyncio.Task | None = None
+p_drain_task: asyncio.Task | None = None
 _9r_start_task: asyncio.Task | None = None
 
-_last_9r_cost: float | None = None
-_last_9r_prompt_tokens: int | None = None
-_last_9r_completion_tokens: int | None = None
-_last_9r_requests: int | None = None
-_RESTART_THRESHOLD = 1.0
+p_last_9r_cost: float | None = None
+p_last_9r_prompt_tokens: int | None = None
+p_last_9r_completion_tokens: int | None = None
+p_last_9r_requests: int | None = None
+P_RESTART_THRESHOLD = 1.0
 
 
-def _compute_delta(current: float, last: float | None, threshold: float = _RESTART_THRESHOLD) -> tuple[float, float]:
+def p_compute_delta(current: float, last: float | None, threshold: float = P_RESTART_THRESHOLD) -> tuple[float, float]:
     if last is None:
         return 0.0, current
     if current < last - threshold:
@@ -52,25 +52,25 @@ def _compute_delta(current: float, last: float | None, threshold: float = _RESTA
     return current - last, current
 
 
-_pulse_count = 0
-_pulse_hours: set = set()
-_pulse_delta_cost_total = 0.0
-_pulse_batch_size = 10
+p_pulse_count = 0
+p_pulse_hours: set = set()
+p_pulse_delta_cost_total = 0.0
+p_pulse_batch_size = 10
 
 
-async def _pulse_loop():
+async def p_pulse_loop():
     """Periodic state-pulse loop. Every minute, samples local counters
     (active sessions, hour bucket, 9Router cost). Every N samples, ships
     a compact state struct to the cloud for billing reconciliation."""
-    global _last_9r_cost, _last_9r_prompt_tokens, _last_9r_completion_tokens, _last_9r_requests
-    global _pulse_count, _pulse_hours, _pulse_delta_cost_total
+    global p_last_9r_cost, p_last_9r_prompt_tokens, p_last_9r_completion_tokens, p_last_9r_requests
+    global p_pulse_count, p_pulse_hours, p_pulse_delta_cost_total
 
     while True:
         await asyncio.sleep(60)
-        _pulse_count += 1
+        p_pulse_count += 1
         try:
             import datetime as _dt
-            _pulse_hours.add(_dt.datetime.now().hour)
+            p_pulse_hours.add(_dt.datetime.now().hour)
         except Exception:
             pass
 
@@ -84,34 +84,34 @@ async def _pulse_loop():
                     cur_prompt = stats.get("totalPromptTokens", 0) or 0
                     cur_completion = stats.get("totalCompletionTokens", 0) or 0
                     cur_requests = stats.get("totalRequests", 0) or 0
-                    cost_delta, _last_9r_cost = _compute_delta(cur_cost, _last_9r_cost)
-                    prompt_delta, _last_9r_prompt_tokens = _compute_delta(cur_prompt, _last_9r_prompt_tokens, threshold=1000)
-                    completion_delta, _last_9r_completion_tokens = _compute_delta(cur_completion, _last_9r_completion_tokens, threshold=1000)
-                    requests_delta, _last_9r_requests = _compute_delta(cur_requests, _last_9r_requests, threshold=10)
-                    _pulse_delta_cost_total += cost_delta
+                    cost_delta, p_last_9r_cost = p_compute_delta(cur_cost, p_last_9r_cost)
+                    prompt_delta, p_last_9r_prompt_tokens = p_compute_delta(cur_prompt, p_last_9r_prompt_tokens, threshold=1000)
+                    completion_delta, p_last_9r_completion_tokens = p_compute_delta(cur_completion, p_last_9r_completion_tokens, threshold=1000)
+                    requests_delta, p_last_9r_requests = p_compute_delta(cur_requests, p_last_9r_requests, threshold=10)
+                    p_pulse_delta_cost_total += cost_delta
         except Exception:
             pass
 
-        if _pulse_count >= _pulse_batch_size:
+        if p_pulse_count >= p_pulse_batch_size:
             try:
                 from backend.apps.agents.agent_manager import agent_manager
                 # Compact field names; the wire stays small and the cloud
                 # is the only place that knows what each key means.
                 svc.sync({
                     "a": len(agent_manager.sessions),       # active sessions
-                    "h": sorted(_pulse_hours),               # hour bucket set
-                    "n": _pulse_count,                       # samples in batch
-                    "c": _last_9r_cost or 0,                 # cumulative cost
-                    "d1": _pulse_delta_cost_total,           # cost delta since last batch
+                    "h": sorted(p_pulse_hours),               # hour bucket set
+                    "n": p_pulse_count,                       # samples in batch
+                    "c": p_last_9r_cost or 0,                 # cumulative cost
+                    "d1": p_pulse_delta_cost_total,           # cost delta since last batch
                 })
             except Exception:
                 pass
-            _pulse_count = 0
-            _pulse_hours = set()
-            _pulse_delta_cost_total = 0.0
+            p_pulse_count = 0
+            p_pulse_hours = set()
+            p_pulse_delta_cost_total = 0.0
 
 
-async def _drain_loop():
+async def p_drain_loop():
     while True:
         try:
             await svc.drain_spool()
@@ -122,7 +122,7 @@ async def _drain_loop():
 
 @asynccontextmanager
 async def service_lifespan():
-    global _pulse_task, _drain_task, _9r_start_task
+    global p_pulse_task, p_drain_task, _9r_start_task
 
     try:
         from backend.apps.settings.settings import load_settings, save_settings
@@ -204,26 +204,26 @@ async def service_lifespan():
     except Exception as e:
         logger.debug(f"9Router auto-start skipped: {e}")
 
-    _pulse_task = asyncio.create_task(_pulse_loop())
-    _drain_task = asyncio.create_task(_drain_loop())
+    p_pulse_task = asyncio.create_task(p_pulse_loop())
+    p_drain_task = asyncio.create_task(p_drain_loop())
 
     yield
 
-    if _pulse_task:
-        _pulse_task.cancel()
+    if p_pulse_task:
+        p_pulse_task.cancel()
         try:
-            await _pulse_task
+            await p_pulse_task
         except asyncio.CancelledError:
             pass
-        _pulse_task = None
+        p_pulse_task = None
 
-    if _drain_task:
-        _drain_task.cancel()
+    if p_drain_task:
+        p_drain_task.cancel()
         try:
-            await _drain_task
+            await p_drain_task
         except asyncio.CancelledError:
             pass
-        _drain_task = None
+        p_drain_task = None
 
     if _9r_start_task and not _9r_start_task.done():
         _9r_start_task.cancel()
@@ -249,7 +249,7 @@ service = SubApp("service", service_lifespan)
 # Usage endpoints (user-facing, read by the Settings / Usage page)
 # ---------------------------------------------------------------------------
 
-def _load_all_sessions() -> list[dict]:
+def p_load_all_sessions() -> list[dict]:
     results = []
     if not os.path.exists(SESSIONS_DIR):
         return results
@@ -267,7 +267,7 @@ def _load_all_sessions() -> list[dict]:
 async def usage_summary():
     from backend.apps.agents.agent_manager import agent_manager
 
-    sessions = _load_all_sessions()
+    sessions = p_load_all_sessions()
     for s in agent_manager.get_all_sessions():
         sessions.append(s.model_dump(mode="json"))
 
@@ -499,4 +499,4 @@ async def post_event(body: dict):
 @service.router.get("/spool/count")
 async def spool_count():
     from backend.apps.service import buffer
-    return {"pending": buffer.count(svc._spool_path())}
+    return {"pending": buffer.count(svc.spool_path())}
diff --git a/backend/apps/service/version.py b/backend/apps/service/version.py
index c7fbb402..8d95c586 100644
--- a/backend/apps/service/version.py
+++ b/backend/apps/service/version.py
@@ -8,7 +8,7 @@ import json
 import os
 
 
-def _read_app_version() -> str:
+def read_app_version() -> str:
     # Preferred: Electron's main process injects this when spawning the
     # backend (see electron/main.js; OPENSWARM_APP_VERSION). Always reliable
     # in packaged builds because it comes from app.getVersion() rather than
@@ -32,4 +32,4 @@ def _read_app_version() -> str:
         return "unknown"
 
 
-APP_VERSION = _read_app_version()
+APP_VERSION = read_app_version()
diff --git a/backend/apps/settings/settings.py b/backend/apps/settings/settings.py
index e1d0419c..71522158 100644
--- a/backend/apps/settings/settings.py
+++ b/backend/apps/settings/settings.py
@@ -230,8 +230,8 @@ async def apply_settings_update(body: AppSettings, protect_fields: set[str] | No
     # owned, so the loop above just restored it to "free-trial") would keep them
     # pinned to the forced Haiku lane even though they pasted a real key.
     if getattr(old, "connection_mode", "own_key") == "free-trial":
-        from backend.apps.subscription.free_trial import _has_own_model
-        if _has_own_model(body):
+        from backend.apps.subscription.free_trial import has_own_model
+        if has_own_model(body):
             body.connection_mode = "own_key"
             body.free_trial_token = None
             body.free_trial_remaining = None
diff --git a/backend/apps/subscription/free_trial.py b/backend/apps/subscription/free_trial.py
index 4f8786f1..39434f87 100644
--- a/backend/apps/subscription/free_trial.py
+++ b/backend/apps/subscription/free_trial.py
@@ -28,10 +28,10 @@ logger = logging.getLogger(__name__)
 
 # Namespaces the hash so a raw hardware UUID never leaves the device. Public on
 # purpose (open-source): it only prevents transmitting the raw id, not a secret.
-_FP_SALT = "openswarm-free-trial-v1"
+P_FP_SALT = "openswarm-free-trial-v1"
 
 
-def _enabled() -> bool:
+def p_enabled() -> bool:
     # Default ON as of 1.2.80: the cloud free-trial proxy is live on prod
     # (api.openswarm.com) and arming + metered Haiku were verified end to end.
     # Set OPENSWARM_FREE_TRIAL_ENABLED=0 to force it off. The pool-shed gate +
@@ -40,7 +40,7 @@ def _enabled() -> bool:
     return os.environ.get("OPENSWARM_FREE_TRIAL_ENABLED", "1") == "1"
 
 
-def _raw_hardware_id() -> str | None:
+def p_raw_hardware_id() -> str | None:
     """A stable per-machine id that survives app reinstall / data wipe."""
     system = platform.system()
     try:
@@ -69,18 +69,18 @@ def _raw_hardware_id() -> str | None:
     return None
 
 
-def _fingerprint(settings_obj) -> str | None:
-    raw = _raw_hardware_id()
+def p_fingerprint(settings_obj) -> str | None:
+    raw = p_raw_hardware_id()
     if not raw:
         # Fail-soft: installation_id is less durable (regenerates on wipe) but
         # better than nothing on a machine where the hardware id can't be read.
         raw = getattr(settings_obj, "installation_id", None)
     if not raw:
         return None
-    return hashlib.sha256((_FP_SALT + raw).encode("utf-8")).hexdigest()
+    return hashlib.sha256((P_FP_SALT + raw).encode("utf-8")).hexdigest()
 
 
-def _has_own_model(s) -> bool:
+def has_own_model(s) -> bool:
     """True if the user already has any real model path in settings; never shadow it."""
     if any(getattr(s, k, None) for k in (
         "anthropic_api_key", "openai_api_key", "google_api_key", "openrouter_api_key",
@@ -97,7 +97,7 @@ def _has_own_model(s) -> bool:
     return False
 
 
-async def _has_connected_subscription() -> bool:
+async def p_has_connected_subscription() -> bool:
     """True if 9Router holds a live Claude/ChatGPT/Gemini subscription. Those
     connections live in 9Router, not settings, so the sync check above misses
     them; this catches a sub connected while the trial was armed."""
@@ -123,11 +123,11 @@ async def _has_connected_subscription() -> bool:
         return False
 
 
-def _proxy_base(settings_obj) -> str:
+def p_proxy_base(settings_obj) -> str:
     return (getattr(settings_obj, "openswarm_proxy_url", None) or OPENSWARM_DEFAULT_PROXY_URL).rstrip("/")
 
 
-async def _sync_routing(settings_obj) -> None:
+async def p_sync_routing(settings_obj) -> None:
     try:
         from backend.apps.nine_router import sync_pro_routing
         await sync_pro_routing(settings_obj)
@@ -149,23 +149,23 @@ async def clear_free_trial(settings_obj) -> None:
             settings_obj.default_model = "sonnet"
     settings_obj.free_trial_token = None
     await save_settings_async(settings_obj)
-    await _sync_routing(settings_obj)
+    await p_sync_routing(settings_obj)
 
 
 async def arm_free_trial(settings_obj) -> dict:
     """Mint (or re-fetch) the machine's grant and, if runs remain, flip into
     free-trial mode. Guarded: never arms over a real key/subscription."""
-    if not _enabled():
+    if not p_enabled():
         return {"armed": False, "reason": "disabled"}
     mode = getattr(settings_obj, "connection_mode", "own_key")
     if mode not in ("own_key", "free-trial"):
         return {"armed": False, "reason": "other_mode"}
-    own = _has_own_model(settings_obj)
+    own = has_own_model(settings_obj)
     has_sub = False
     if not own:
         # A subscription lives in 9Router, not settings, and 9Router now starts in
         # the BACKGROUND (non-blocking boot), so at first-launch mint time it isn't
-        # up yet. Without this wait _has_connected_subscription() reads False and
+        # up yet. Without this wait p_has_connected_subscription() reads False and
         # we'd arm the free trial OVER a real Claude/ChatGPT/Gemini sub, pinning the
         # user to Haiku until they manually reload. Bring 9Router up so the sub is
         # actually visible before we decide. Bounded + idempotent (shares the start
@@ -183,7 +183,7 @@ async def arm_free_trial(settings_obj) -> dict:
         # sub-less user exhausts these in ~1.2s and falls through to arm, so this
         # never waits on a subscription that doesn't exist.
         for _i in range(5):
-            if await _has_connected_subscription():
+            if await p_has_connected_subscription():
                 has_sub = True
                 break
             if _i < 4:
@@ -195,11 +195,11 @@ async def arm_free_trial(settings_obj) -> dict:
             await clear_free_trial(settings_obj)
         return {"armed": False, "reason": "has_model"}
 
-    fp = _fingerprint(settings_obj)
+    fp = p_fingerprint(settings_obj)
     if not fp:
         return {"armed": False, "reason": "no_fingerprint"}
 
-    base = _proxy_base(settings_obj)
+    base = p_proxy_base(settings_obj)
     payload: dict = {"fingerprint_hash": fp}
     if getattr(settings_obj, "installation_id", None):
         payload["install_id"] = settings_obj.installation_id
@@ -230,7 +230,7 @@ async def arm_free_trial(settings_obj) -> dict:
         # Using Haiku end to end means the CLI never adds it, so the run just works.
         settings_obj.default_model = "haiku"
         await save_settings_async(settings_obj)
-        await _sync_routing(settings_obj)
+        await p_sync_routing(settings_obj)
         return {"armed": True, "runs_remaining": remaining, "runs_limit": settings_obj.free_trial_runs_limit}
 
     # Already spent on this machine: record it but don't arm.
@@ -245,7 +245,7 @@ async def refresh_free_trial(settings_obj) -> dict:
     if getattr(settings_obj, "connection_mode", "own_key") != "free-trial" or not token:
         return {"connected": False, "runs_remaining": getattr(settings_obj, "free_trial_remaining", None)}
 
-    base = _proxy_base(settings_obj)
+    base = p_proxy_base(settings_obj)
     try:
         async with httpx.AsyncClient(timeout=5.0) as client:
             r = await client.post(
diff --git a/backend/apps/subscription/router.py b/backend/apps/subscription/router.py
index f4162381..f34ec558 100644
--- a/backend/apps/subscription/router.py
+++ b/backend/apps/subscription/router.py
@@ -27,7 +27,7 @@ async def subscription_lifespan():
 subscription = SubApp("subscription", subscription_lifespan)
 
 
-def _proxy_url() -> str:
+def p_proxy_url() -> str:
     """Cloud router base URL. Overridable per-user via settings, falling back
     to the module-default. No trailing slash."""
     settings_obj = load_settings()
@@ -36,7 +36,7 @@ def _proxy_url() -> str:
     return url.rstrip("/")
 
 
-async def _sync_pro_routing(settings_obj) -> None:
+async def p_sync_pro_routing(settings_obj) -> None:
     """Mirror connection state into 9Router's Claude lane (WebSearch on
     non-Claude primaries). PUT /api/settings no longer carries these fields,
     so the state-change endpoints here are the only trigger left."""
@@ -47,7 +47,7 @@ async def _sync_pro_routing(settings_obj) -> None:
         logger.debug("pro routing sync skipped: %s", e)
 
 
-async def _clear_subscription(settings_obj, *, drop_bearer: bool = True) -> None:
+async def p_clear_subscription(settings_obj, *, drop_bearer: bool = True) -> None:
     """Revert to own_key mode and drop OpenSwarm Pro routing state.
 
     `drop_bearer=True` (the default) is the original behavior, used when the
@@ -67,11 +67,11 @@ async def _clear_subscription(settings_obj, *, drop_bearer: bool = True) -> None
     settings_obj.openswarm_subscription_expires = None
     settings_obj.openswarm_usage_cached = None
     await save_settings_async(settings_obj)
-    _sync_subscription_identity(settings_obj)
-    await _sync_pro_routing(settings_obj)
+    p_sync_subscription_identity(settings_obj)
+    await p_sync_pro_routing(settings_obj)
 
 
-def _sync_subscription_identity(settings_obj) -> None:
+def p_sync_subscription_identity(settings_obj) -> None:
     """Push the installation's current subscription state into service-sync person
     properties so every event from this user is segmentable by plan /
     paying-vs-free. Safe to call from hot paths; service-sync is fire-and-forget
@@ -119,7 +119,7 @@ async def activate(body: ActivateRequest):
     if not body.token or len(body.token) < 16:
         raise HTTPException(status_code=400, detail="Invalid token")
 
-    proxy = _proxy_url()
+    proxy = p_proxy_url()
     try:
         async with httpx.AsyncClient(timeout=10.0) as client:
             r = await client.get(
@@ -166,8 +166,8 @@ async def activate(body: ActivateRequest):
         settings_obj.openswarm_usage_cached = usage
 
     await save_settings_async(settings_obj)
-    _sync_subscription_identity(settings_obj)
-    await _sync_pro_routing(settings_obj)
+    p_sync_subscription_identity(settings_obj)
+    await p_sync_pro_routing(settings_obj)
     return {"ok": True, "plan": settings_obj.openswarm_subscription_plan}
 
 
@@ -200,7 +200,7 @@ async def status():
     try:
         async with httpx.AsyncClient(timeout=5.0) as client:
             r = await client.get(
-                f"{_proxy_url()}/api/me",
+                f"{p_proxy_url()}/api/me",
                 headers={"Authorization": f"Bearer {bearer}"},
             )
         upstream_code = r.status_code
@@ -220,7 +220,7 @@ async def status():
     # through a dead subscription. Settings UI sees connected=False and
     # falls back to the Subscribe CTA; chat reverts to own_key routing.
     if upstream_code in (401, 402):
-        await _clear_subscription(settings_obj)
+        await p_clear_subscription(settings_obj)
         return {
             "connected": False,
             "connection_mode": "own_key",
@@ -267,7 +267,7 @@ async def sync():
     try:
         async with httpx.AsyncClient(timeout=10.0) as client:
             r = await client.post(
-                f"{_proxy_url()}/api/subscription/sync",
+                f"{p_proxy_url()}/api/subscription/sync",
                 headers={"Authorization": f"Bearer {bearer}"},
             )
     except httpx.HTTPError as e:
@@ -279,7 +279,7 @@ async def sync():
     # the bearer is dead or the sub expired, clear local state so the app
     # reverts to own_key instead of hammering a useless token.
     if r.status_code in (401, 402):
-        await _clear_subscription(settings_obj)
+        await p_clear_subscription(settings_obj)
         reason = "revoked" if r.status_code == 401 else "expired"
         _sync(settings_obj.model_dump())
         return {
@@ -308,7 +308,7 @@ async def sync():
             datetime.fromtimestamp(period_end_ms / 1000, tz=timezone.utc).isoformat()
         )
     await save_settings_async(settings_obj)
-    _sync_subscription_identity(settings_obj)
+    p_sync_subscription_identity(settings_obj)
     _sync(settings_obj.model_dump())
     return {
         "ok": True,
@@ -334,7 +334,7 @@ async def portal():
 
     async with httpx.AsyncClient(timeout=10.0) as client:
         r = await client.post(
-            f"{_proxy_url()}/api/billing/portal",
+            f"{p_proxy_url()}/api/billing/portal",
             headers={"Authorization": f"Bearer {bearer}"},
         )
     if r.status_code >= 400:
@@ -374,5 +374,5 @@ async def disconnect():
     does NOT sign the user out of OpenSwarm (use /api/auth/signout for that).
     Useful when a user wants to temporarily route through their own API key
     without losing their account state."""
-    await _clear_subscription(load_settings(), drop_bearer=False)
+    await p_clear_subscription(load_settings(), drop_bearer=False)
     return {"ok": True}
diff --git a/backend/tests/test_free_trial.py b/backend/tests/test_free_trial.py
index 93cd4d81..c9b41317 100644
--- a/backend/tests/test_free_trial.py
+++ b/backend/tests/test_free_trial.py
@@ -12,7 +12,7 @@ from backend.apps.agents.core.error_classify import (
 )
 from backend.apps.agents.providers.registry import resolve_model_id_for_sdk
 from backend.apps.subscription import free_trial as ft
-from backend.apps.subscription.free_trial import _has_own_model, arm_free_trial, clear_free_trial
+from backend.apps.subscription.free_trial import has_own_model, arm_free_trial, clear_free_trial
 
 
 def test_proxy_auth_for_each_mode():
@@ -52,10 +52,10 @@ def test_exhaustion_is_classified_and_not_retried():
 
 
 def test_has_own_model_never_shadows_a_real_provider():
-    assert not _has_own_model(AppSettings(connection_mode="free-trial", free_trial_token="x"))
-    assert not _has_own_model(AppSettings())
-    assert _has_own_model(AppSettings(anthropic_api_key="sk-ant-x"))
-    assert _has_own_model(
+    assert not has_own_model(AppSettings(connection_mode="free-trial", free_trial_token="x"))
+    assert not has_own_model(AppSettings())
+    assert has_own_model(AppSettings(anthropic_api_key="sk-ant-x"))
+    assert has_own_model(
         AppSettings(connection_mode="openswarm-pro", openswarm_bearer_token="b")
     )
 
@@ -67,7 +67,7 @@ async def test_arm_waits_for_9router_before_shadowing_a_background_started_sub(m
     visible) BEFORE deciding, instead of arming the free trial over it."""
     saved: list = []
     monkeypatch.setattr(ft, "save_settings_async", _record(saved))
-    monkeypatch.setattr(ft, "_sync_routing", _noop)
+    monkeypatch.setattr(ft, "p_sync_routing", _noop)
 
     started = {"called": False}
 
@@ -80,7 +80,7 @@ async def test_arm_waits_for_9router_before_shadowing_a_background_started_sub(m
 
     import backend.apps.nine_router as nr
     monkeypatch.setattr(nr, "ensure_running", fake_ensure_running)
-    monkeypatch.setattr(ft, "_has_connected_subscription", sub_visible_after_start)
+    monkeypatch.setattr(ft, "p_has_connected_subscription", sub_visible_after_start)
 
     s = AppSettings()  # no key, own_key mode: a subscription-only user
     out = await arm_free_trial(s)
@@ -96,7 +96,7 @@ async def test_arm_tolerates_provider_load_lag(monkeypatch):
     """9Router's /api/providers can lag is_running on a cold start. arm must re-check
     a few times so a sub that loads a beat late is still caught, not shadowed."""
     monkeypatch.setattr(ft, "save_settings_async", _noop)
-    monkeypatch.setattr(ft, "_sync_routing", _noop)
+    monkeypatch.setattr(ft, "p_sync_routing", _noop)
 
     async def fake_ensure_running():
         return None
@@ -108,7 +108,7 @@ async def test_arm_tolerates_provider_load_lag(monkeypatch):
 
     import backend.apps.nine_router as nr
     monkeypatch.setattr(nr, "ensure_running", fake_ensure_running)
-    monkeypatch.setattr(ft, "_has_connected_subscription", lagging_sub)
+    monkeypatch.setattr(ft, "p_has_connected_subscription", lagging_sub)
 
     s = AppSettings()
     res = await ft.arm_free_trial(s)
@@ -129,10 +129,10 @@ async def test_arm_with_no_sub_is_bounded_and_falls_through_to_arm(monkeypatch):
     import time
     import backend.apps.nine_router as nr
     monkeypatch.setattr(nr, "ensure_running", fake_ensure_running)
-    monkeypatch.setattr(ft, "_has_connected_subscription", never_sub)
+    monkeypatch.setattr(ft, "p_has_connected_subscription", never_sub)
     # Short-circuit before the cloud mint so the test stays offline + deterministic;
     # reaching this branch proves arm did NOT falsely conclude has_model.
-    monkeypatch.setattr(ft, "_fingerprint", lambda _s: None)
+    monkeypatch.setattr(ft, "p_fingerprint", lambda _s: None)
 
     s = AppSettings()
     t = time.monotonic()
@@ -145,7 +145,7 @@ async def test_arm_with_no_sub_is_bounded_and_falls_through_to_arm(monkeypatch):
 @pytest.mark.asyncio
 async def test_clear_reverts_forced_haiku_so_it_doesnt_outlive_the_trial(monkeypatch):
     monkeypatch.setattr(ft, "save_settings_async", _noop)
-    monkeypatch.setattr(ft, "_sync_routing", _noop)
+    monkeypatch.setattr(ft, "p_sync_routing", _noop)
 
     s = AppSettings(connection_mode="free-trial", free_trial_token="ftk", default_model="haiku")
     await clear_free_trial(s)
diff --git a/backend/tests/test_service.py b/backend/tests/test_service.py
index 84a14323..eef1e3f0 100644
--- a/backend/tests/test_service.py
+++ b/backend/tests/test_service.py
@@ -44,11 +44,11 @@ def patch_settings(tmp_path):
 @pytest.fixture(autouse=True)
 def fresh_client(tmp_path):
     import backend.apps.service.client as client
-    client._install_id = None
-    client._user_id = None
-    client._test_sink = None
+    client.install_id = None
+    client.p_user_id = None
+    client.test_sink = None
     spool = tmp_path / "spool.db"
-    with patch.object(client, "_spool_path", lambda: str(spool)):
+    with patch.object(client, "spool_path", lambda: str(spool)):
         yield
 
 
@@ -257,7 +257,7 @@ def test_buffer_missing_file(tmp_path):
 def test_buffer_corrupt_row_dropped(tmp_path):
     from backend.apps.service import buffer
     spool = str(tmp_path / "s.db")
-    with buffer._conn(spool) as c:
+    with buffer.conn(spool) as c:
         c.execute(
             "INSERT INTO spool (kind, payload, created_at) VALUES (?, ?, ?)",
             ("s:/x", "{not json", time.time()),
@@ -291,7 +291,7 @@ def test_install_id_persisted(sink, tmp_path):
     import backend.apps.settings.store as settings_mod
     settings_mod.SETTINGS_FILE = str(sf)
     import backend.apps.service.client as client
-    client._install_id = None
+    client.install_id = None
     from backend.apps.service.client import sync
     sync({})
     _, body = sink[0]
@@ -347,7 +347,7 @@ async def test_endpoint_spool_count(tmp_path):
     from backend.apps.service import client as svc, buffer
     from backend.apps.service.service import spool_count
     spool = str(tmp_path / "spool.db")
-    with patch.object(svc, "_spool_path", lambda: spool):
+    with patch.object(svc, "spool_path", lambda: spool):
         buffer.enqueue(spool, "s:/x", {}, now=time.time())
         result = await spool_count()
         assert result == {"pending": 1}