[eric] swarm+nine_router: leading-_ -> p_/public across swarm entities/closure/redact/ziputil + nine_router process/sync/oauth; promote cross-file public, fix monkeypatch refs

2026-09-13 05:07:40 +02:00 · 2026-06-23 20:35:06 -07:00
parent 71b2b33fae
commit 10aef1ff18
16 changed files with 230 additions and 230 deletions
@@ -22,9 +22,9 @@ logger = logging.getLogger(__name__)
 # 1455 that serves the same postMessage/BroadcastChannel/localStorage relay so
 # the frontend's existing popup + msgHandler flow works unchanged.

-_CODEX_CALLBACK_PORT = 1455
-_CODEX_CALLBACK_PATH = "/auth/callback"
-_CODEX_CALLBACK_HTML = b"""<!DOCTYPE html>
+P_CODEX_CALLBACK_PORT = 1455
+P_CODEX_CALLBACK_PATH = "/auth/callback"
+P_CODEX_CALLBACK_HTML = b"""<!DOCTYPE html>
 <html><head><meta charset="utf-8"><title>Authorization Complete</title>
 <style>body{font-family:-apple-system,system-ui,sans-serif;background:#111;color:#eee;
 text-align:center;padding:60px 20px;margin:0}h1{font-weight:600;margin:0 0 12px}
@@ -59,10 +59,10 @@ p{color:#888;margin:0}</style></head><body>
 </body></html>"""


-async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base_events.Server | None:
+async def p_start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base_events.Server | None:
    """Spawn a one-shot HTTP listener on 127.0.0.1:1455 for the Codex OAuth callback.

-    Serves GET /auth/callback with _CODEX_CALLBACK_HTML. After serving the
+    Serves GET /auth/callback with P_CODEX_CALLBACK_HTML. After serving the
    callback (or after `timeout` seconds with no callback) the listener
    closes itself in a background task. Safe to call even if 1455 is busy ,
    logs the collision and returns None so start_oauth can still proceed and
@@ -96,7 +96,7 @@ async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base
            path = parts[1] if len(parts) >= 2 else ""
            method = parts[0] if parts else ""

-            if method == "GET" and path.startswith(_CODEX_CALLBACK_PATH):
+            if method == "GET" and path.startswith(P_CODEX_CALLBACK_PATH):
                # Parse code/state out of the query string and exchange
                # server-side before serving the HTML. Duplicate exchanges
                # are harmless (single-use auth codes fail the second call,
@@ -136,7 +136,7 @@ async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base
                except Exception as e:
                    logger.debug(f"Codex callback listener pre-exchange error: {e}")

-                body = _CODEX_CALLBACK_HTML
+                body = P_CODEX_CALLBACK_HTML
                response = (
                    b"HTTP/1.1 200 OK\r\n"
                    b"Content-Type: text/html; charset=utf-8\r\n"
@@ -165,12 +165,12 @@ async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base
                pass

    try:
-        server = await asyncio.start_server(_handle, "127.0.0.1", _CODEX_CALLBACK_PORT)
+        server = await asyncio.start_server(_handle, "127.0.0.1", P_CODEX_CALLBACK_PORT)
    except OSError as e:
        # Port already in use; probably another Codex connect attempt still
        # running, or an actual Codex CLI process holding 1455. Log and bail.
        logger.warning(
-            f"Could not start Codex callback listener on port {_CODEX_CALLBACK_PORT}: {e}. "
+            f"Could not start Codex callback listener on port {P_CODEX_CALLBACK_PORT}: {e}. "
            "If another connection attempt is in progress, wait for it to finish or time out."
        )
        return None
@@ -194,7 +194,7 @@ async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base
                pass

    asyncio.create_task(_lifecycle())
-    logger.info(f"Started Codex callback listener on http://localhost:{_CODEX_CALLBACK_PORT}{_CODEX_CALLBACK_PATH}")
+    logger.info(f"Started Codex callback listener on http://localhost:{P_CODEX_CALLBACK_PORT}{P_CODEX_CALLBACK_PATH}")
    return server


@@ -211,15 +211,15 @@ async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base
 #   in one cookie jar.
 # The callback for gemini-cli/antigravity lands on /api/subscriptions/callback
 # and runs the exchange server-side; codex uses its fixed 1455 listener; claude
-# is special-cased in _callback_uri_for_provider below.
-_EXTERNAL_BROWSER_PROVIDERS: set[str] = {"gemini-cli", "antigravity", "codex", "claude"}
+# is special-cased in p_callback_uri_for_provider below.
+P_EXTERNAL_BROWSER_PROVIDERS: set[str] = {"gemini-cli", "antigravity", "codex", "claude"}


-def _should_use_external_browser(provider: str) -> bool:
-    return provider in _EXTERNAL_BROWSER_PROVIDERS
+def p_should_use_external_browser(provider: str) -> bool:
+    return provider in P_EXTERNAL_BROWSER_PROVIDERS


-def _backend_port() -> int:
+def p_backend_port() -> int:
    """Best-effort lookup of the OpenSwarm backend HTTP port.

    Falls back to 8324 (the default in backend/main.py) if OPENSWARM_PORT
@@ -233,27 +233,27 @@ def _backend_port() -> int:
        return 8324


-def _callback_uri_for_provider(provider: str) -> str:
+def p_callback_uri_for_provider(provider: str) -> str:
    """Return the redirect URI to pass to 9Router's authorize endpoint.

    Most providers accept 9Router's built-in callback page at port 20128.
    Special cases:
    - Codex/OpenAI's OAuth client is bound to a fixed
      http://localhost:1455/auth/callback URI; handled by
-      _start_codex_callback_listener above.
+      p_start_codex_callback_listener above.
    - Gemini/Google's OAuth consent page rejects embedded browsers, so we
      route the callback through OpenSwarm's backend endpoint at
      /api/subscriptions/callback (backend/main.py) which runs the
      exchange itself.
    """
    if provider == "codex":
-        return f"http://localhost:{_CODEX_CALLBACK_PORT}{_CODEX_CALLBACK_PATH}"
+        return f"http://localhost:{P_CODEX_CALLBACK_PORT}{P_CODEX_CALLBACK_PATH}"
    # Anthropic's OAuth client only whitelists localhost:20128/callback;
    # 9router_gpt5_patch.js 302-rewrites the hit to the backend handler.
    if provider == "claude":
        return f"http://localhost:{NINE_ROUTER_PORT}/callback"
-    if provider in _EXTERNAL_BROWSER_PROVIDERS:
-        return f"http://localhost:{_backend_port()}/api/subscriptions/callback"
+    if provider in P_EXTERNAL_BROWSER_PROVIDERS:
+        return f"http://localhost:{p_backend_port()}/api/subscriptions/callback"
    return f"http://localhost:{NINE_ROUTER_PORT}/callback"


@@ -279,9 +279,9 @@ async def start_oauth(provider: str) -> dict:
        except Exception:
            pass

-        callback_url = _callback_uri_for_provider(provider)
+        callback_url = p_callback_uri_for_provider(provider)
        if provider == "codex":
-            await _start_codex_callback_listener()
+            await p_start_codex_callback_listener()

        r = await client.get(
            f"{NINE_ROUTER_API}/oauth/{provider}/authorize",
@@ -295,7 +295,7 @@ async def start_oauth(provider: str) -> dict:
            "code_verifier": data.get("codeVerifier", ""),
            "state": data.get("state", ""),
            "redirect_uri": callback_url,
-            "use_external_browser": _should_use_external_browser(provider),
+            "use_external_browser": p_should_use_external_browser(provider),
        }


@@ -63,43 +63,43 @@ NINE_ROUTER_NPM_VERSION = os.environ.get("OPENSWARM_ROUTER_VERSION", "0.3.60")
 #   1. rotate that log before we spawn 9Router when it gets large, so growth can't run away;
 #   2. give node an explicit, generous heap ceiling for legitimate large multimodal bodies.
 # Neither touches routing, so WebSearch/WebFetch translation and the 0.3.60 pin are unaffected.
-_REQUEST_LOG_PATH = os.path.expanduser("~/.9router/request-details.json")
-_REQUEST_LOG_MAX_BYTES = 5 * 1024 * 1024
-_NODE_HEAP_MB = 4096
+P_REQUEST_LOG_PATH = os.path.expanduser("~/.9router/request-details.json")
+P_REQUEST_LOG_MAX_BYTES = 5 * 1024 * 1024
+P_NODE_HEAP_MB = 4096


-def _rotate_request_log() -> None:
+def p_rotate_request_log() -> None:
    """Rotate ~/.9router/request-details.json to a single .0 backup when it grows past the cap,
    BEFORE 9Router is spawned (never racing a live writer). 9Router recreates a fresh file, exactly
    like a clean install. The only consumer is the 'most recent 5' reasoning-token lookup, which
    already tolerates an empty/missing file, so no feature loses data it depends on."""
    try:
-        if os.path.exists(_REQUEST_LOG_PATH) and os.path.getsize(_REQUEST_LOG_PATH) > _REQUEST_LOG_MAX_BYTES:
-            os.replace(_REQUEST_LOG_PATH, _REQUEST_LOG_PATH + ".0")
+        if os.path.exists(P_REQUEST_LOG_PATH) and os.path.getsize(P_REQUEST_LOG_PATH) > P_REQUEST_LOG_MAX_BYTES:
+            os.replace(P_REQUEST_LOG_PATH, P_REQUEST_LOG_PATH + ".0")
            logger.info(
                "9Router request log rotated (exceeded %d MB) to avoid the router OOM",
-                _REQUEST_LOG_MAX_BYTES // (1024 * 1024),
+                P_REQUEST_LOG_MAX_BYTES // (1024 * 1024),
            )
    except Exception as e:
        logger.debug("9Router request-log rotation skipped: %s", e)


-_process: subprocess.Popen | None = None
+p_process: subprocess.Popen | None = None

 # Serializes ensure_running() so a background auto-start and a concurrent
 # dispatch-time ensure can't both spawn 9Router (double-bind on :20128). Lazily
 # created so module import doesn't require a running event loop.
-_start_lock: "asyncio.Lock | None" = None
+p_start_lock: "asyncio.Lock | None" = None

 # Short TTL cache for positive is_running() results. The probe is a sync
 # httpx.get that blocks the event loop, and under load (9Router busy
 # streaming inference) it can exceed its 2s timeout and return False even
 # though 9Router is fine. Caching a recent True result avoids those false
-# negatives without masking a real crash for more than _IS_RUNNING_TTL seconds.
+# negatives without masking a real crash for more than P_IS_RUNNING_TTL seconds.
 # Negative results are NOT cached so startup detection in ensure_running()
 # remains correct.
-_IS_RUNNING_TTL = 10.0
-_is_running_last_ok: float = 0.0
+P_IS_RUNNING_TTL = 10.0
+p_is_running_last_ok: float = 0.0


 def is_running() -> bool:
@@ -115,9 +115,9 @@ def is_running() -> bool:
    first; a down 9Router is detected in <~0.3s instead of ~7s. Only when the
    port is open do we do the HTTP confirm. 9Router binds 0.0.0.0 (the warm app
    reaches it via 127.0.0.1 today), so this changes timing, not reachability."""
-    global _is_running_last_ok
+    global p_is_running_last_ok
    now = time.monotonic()
-    if now - _is_running_last_ok < _IS_RUNNING_TTL:
+    if now - p_is_running_last_ok < P_IS_RUNNING_TTL:
        return True
    try:
        with socket.create_connection(("127.0.0.1", NINE_ROUTER_PORT), timeout=0.3):
@@ -127,14 +127,14 @@ def is_running() -> bool:
    try:
        r = httpx.get(f"http://127.0.0.1:{NINE_ROUTER_PORT}/v1/models", timeout=2.0)
        if r.status_code == 200:
-            _is_running_last_ok = now
+            p_is_running_last_ok = now
            return True
        return False
    except Exception:
        return False


-def _nine_router_data_dir() -> str:
+def p_nine_router_data_dir() -> str:
    """Where 9Router persists machine-id + auth/cli-secret, the two files we
    hash into the /api/* auth token on 0.4.x. Mirrors 9Router's own default
    (DATA_DIR env, else ~/.9router on unix, %APPDATA%/9router on win) so we read
@@ -151,7 +151,7 @@ def _nine_router_data_dir() -> str:
    return os.path.join(os.path.expanduser("~"), ".9router")


-_cli_token_cache: str | None = None
+p_cli_token_cache: str | None = None


 def cli_auth_token() -> str | None:
@@ -162,13 +162,13 @@ def cli_auth_token() -> str | None:
    when missing so connect/sync can auth before that self-call. Returns None on
    0.3.60 (no machine-id) or when 9Router isn't up, so the caller sends no
    header and the old auth-free path is untouched. Never raises."""
-    global _cli_token_cache
-    if _cli_token_cache:
-        return _cli_token_cache
+    global p_cli_token_cache
+    if p_cli_token_cache:
+        return p_cli_token_cache
    if not is_running():
        return None
    try:
-        data_dir = _nine_router_data_dir()
+        data_dir = p_nine_router_data_dir()
        try:
            with open(os.path.join(data_dir, "machine-id"), encoding="utf-8") as f:
                machine_id = f.read().strip()
@@ -198,7 +198,7 @@ def cli_auth_token() -> str | None:
        tok = hashlib.sha256(
            (machine_id + "9r-cli-auth" + cli_secret).encode("utf-8")
        ).hexdigest()[:16]
-        _cli_token_cache = tok
+        p_cli_token_cache = tok
        return tok
    except Exception:
        return None
@@ -211,7 +211,7 @@ def cli_auth_headers() -> dict[str, str]:
    return {"x-9r-cli-token": tok} if tok else {}


-def _find_9router_dir() -> str | None:
+def p_find_9router_dir() -> str | None:
    """Locate the bundled 9Router directory (works in both dev and packaged mode)."""
    _is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1"

@@ -231,7 +231,7 @@ def _find_9router_dir() -> str | None:
    return None


-def _gpt5_patch_path() -> str | None:
+def p_gpt5_patch_path() -> str | None:
    """Absolute path to backend/apps/agents/9router_gpt5_patch.js, used as
    `node --require <path>` when spawning 9router.

@@ -254,7 +254,7 @@ def _gpt5_patch_path() -> str | None:
    return candidate if os.path.exists(candidate) else None


-def _find_node() -> str | None:
+def p_find_node() -> str | None:
    """Find a Node.js binary (works in both dev and packaged mode).

    Priority order:
@@ -284,7 +284,7 @@ def _find_node() -> str | None:
    return None


-def _dev_router_cache_dir() -> str:
+def p_dev_router_cache_dir() -> str:
    """Cache dir for the npm 9router package used in dev mode.

    Pinned per version so bumping NINE_ROUTER_NPM_VERSION triggers a fresh
@@ -296,7 +296,7 @@ def _dev_router_cache_dir() -> str:
    return os.path.join(base, "openswarm-router", NINE_ROUTER_NPM_VERSION)


-def _ensure_router_cached() -> str | None:
+def p_ensure_router_cached() -> str | None:
    """Ensure the npm 9router package is installed in the dev cache.

    Returns the absolute path to `app/server.js` on success, or None if
@@ -308,7 +308,7 @@ def _ensure_router_cached() -> str | None:
    no update-check spinner, and no accidental-quit foot-gun when a
    non-developer right-clicks the "9" tray icon and picks Quit.
    """
-    cache_dir = _dev_router_cache_dir()
+    cache_dir = p_dev_router_cache_dir()
    server_js = os.path.join(cache_dir, "node_modules", "9router", "app", "server.js")
    if os.path.exists(server_js):
        return server_js
@@ -348,7 +348,7 @@ def _ensure_router_cached() -> str | None:
    return server_js if os.path.exists(server_js) else None


-def _read_capture_tail(path: str, limit: int = 6000) -> str:
+def p_read_capture_tail(path: str, limit: int = 6000) -> str:
    """Tail of the 9Router start-capture file, where the real spawn error lands.
    Best-effort; empty string on any hiccup so telemetry never breaks boot."""
    try:
@@ -361,7 +361,7 @@ def _read_capture_tail(path: str, limit: int = 6000) -> str:
        return ""


-def _report_start_failure(reason: str, *, detail: str = "", **fields: Any) -> None:
+def p_report_start_failure(reason: str, *, detail: str = "", **fields: Any) -> None:
    """9Router didn't come up. Log it and ship a scrubbed diagnostic so a user's
    'every model exits 1' is finally explained from our side instead of a silent
    warning. The stderr tail can echo an own_key, so it rides the same scrub as
@@ -386,16 +386,16 @@ def _report_start_failure(reason: str, *, detail: str = "", **fields: Any) -> No
 async def ensure_running():
    """Start 9Router if not already running. Serialized so concurrent callers
    (the background auto-start + a dispatch-time ensure) can't double-spawn."""
-    global _start_lock
-    if _start_lock is None:
-        _start_lock = asyncio.Lock()
-    async with _start_lock:
-        await _ensure_running_impl()
+    global p_start_lock
+    if p_start_lock is None:
+        p_start_lock = asyncio.Lock()
+    async with p_start_lock:
+        await p_ensure_running_impl()


-async def _ensure_running_impl():
+async def p_ensure_running_impl():
    """Start 9Router if not already running."""
-    global _process
+    global p_process
    _is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1"

    if is_running():
@@ -421,9 +421,9 @@ async def _ensure_running_impl():
        else:
            logger.info("9Router already running on port %d", NINE_ROUTER_PORT)
            return
-    _rotate_request_log()
-    _9router_dir = _find_9router_dir()
-    _patch = _gpt5_patch_path()
+    p_rotate_request_log()
+    _9router_dir = p_find_9router_dir()
+    _patch = p_gpt5_patch_path()

    if _is_packaged:
        # Packaged: run the pre-built standalone server staged at
@@ -431,20 +431,20 @@ async def _ensure_running_impl():
        # fall back to the dev npm path here, a user machine has no npm, so that
        # only ever fails silently; every miss is reported instead.
        if not _9router_dir:
-            _report_start_failure("router_not_bundled")
+            p_report_start_failure("router_not_bundled")
            return
        standalone_server = os.path.join(_9router_dir, "server.js")
        if not os.path.exists(standalone_server):
            standalone_server = os.path.join(_9router_dir, ".next", "standalone", "server.js")
        if not os.path.exists(standalone_server):
-            _report_start_failure("server_missing", router_dir_found=True)
+            p_report_start_failure("server_missing", router_dir_found=True)
            return
-        node = _find_node()
+        node = p_find_node()
        if not node:
-            _report_start_failure("node_not_found", router_dir_found=True, server_found=True)
+            p_report_start_failure("node_not_found", router_dir_found=True, server_found=True)
            return
        logger.info("Starting 9Router (production) on port %d...", NINE_ROUTER_PORT)
-        cmd = [node, f"--max-old-space-size={_NODE_HEAP_MB}"] + (["--require", _patch] if _patch else []) + [standalone_server]
+        cmd = [node, f"--max-old-space-size={P_NODE_HEAP_MB}"] + (["--require", _patch] if _patch else []) + [standalone_server]
        cwd = os.path.dirname(standalone_server)
        env = {**os.environ, "PORT": str(NINE_ROUTER_PORT), "NODE_ENV": "production"}
        if node == os.environ.get("OPENSWARM_ELECTRON_PATH"):
@@ -453,10 +453,10 @@ async def _ensure_running_impl():
        # Dev: install the pinned npm package into a local cache once, then spawn
        # `node app/server.js` directly (bypasses the package cli.js tray icon
        # users confusingly quit, its update-check spinner, and the TUI).
-        cached_server = _ensure_router_cached()
+        cached_server = p_ensure_router_cached()
        if not cached_server:
            return
-        node = _find_node()
+        node = p_find_node()
        if not node:
            logger.warning("Node.js not found; cannot start 9Router in dev mode.")
            return
@@ -464,7 +464,7 @@ async def _ensure_running_impl():
            "Starting 9Router (dev cache, 9router@%s) on port %d...",
            NINE_ROUTER_NPM_VERSION, NINE_ROUTER_PORT,
        )
-        cmd = [node, f"--max-old-space-size={_NODE_HEAP_MB}"] + (["--require", _patch] if _patch else []) + [cached_server]
+        cmd = [node, f"--max-old-space-size={P_NODE_HEAP_MB}"] + (["--require", _patch] if _patch else []) + [cached_server]
        cwd = os.path.dirname(cached_server)
        env = {**os.environ, "PORT": str(NINE_ROUTER_PORT), "NODE_ENV": "production"}

@@ -493,7 +493,7 @@ async def _ensure_running_impl():
        _stdout, _stderr = subprocess.DEVNULL, subprocess.DEVNULL

    try:
-        _process = subprocess.Popen(cmd, cwd=cwd, stdout=_stdout, stderr=_stderr, env=env)
+        p_process = subprocess.Popen(cmd, cwd=cwd, stdout=_stdout, stderr=_stderr, env=env)
        if _cap_file is not None:
            _cap_file.close()  # the child holds its own fd; the parent copy isn't needed
        timeout = 20 if _is_packaged else 30
@@ -504,10 +504,10 @@ async def _ensure_running_impl():
                return
        # Verify-at-boot: it never answered. Report with the captured tail + the
        # exit code (non-None = it crashed; None = wedged or just slow).
-        _report_start_failure(
+        p_report_start_failure(
            "not_ready_in_time",
-            detail=_read_capture_tail(_cap_path) if _is_packaged else "",
-            returncode=_process.poll(),
+            detail=p_read_capture_tail(_cap_path) if _is_packaged else "",
+            returncode=p_process.poll(),
            timeout_s=timeout,
        )
    except Exception as e:
@@ -516,25 +516,25 @@ async def _ensure_running_impl():
                _cap_file.close()
            except OSError:
                pass
-        _report_start_failure(
+        p_report_start_failure(
            "spawn_exception",
-            detail=f"{e}\n{_read_capture_tail(_cap_path) if _is_packaged else ''}",
+            detail=f"{e}\n{p_read_capture_tail(_cap_path) if _is_packaged else ''}",
        )


 def stop():
    """Stop the 9Router subprocess."""
-    global _process
-    if _process:
+    global p_process
+    if p_process:
        try:
-            _process.terminate()
-            _process.wait(timeout=5)
+            p_process.terminate()
+            p_process.wait(timeout=5)
        except Exception:
            try:
-                _process.kill()
+                p_process.kill()
            except Exception:
                pass
-        _process = None
+        p_process = None
        logger.info("9Router stopped")


@@ -12,7 +12,7 @@ from .process import NINE_ROUTER_API, cli_auth_headers
 logger = logging.getLogger(__name__)


-def _nr():
+def nr():
    """The package module. is_running / get_providers / httpx are read off it
    at call time so tests that patch `backend.apps.nine_router.<name>` still
    take effect after the split (they used to all live on one module)."""
@@ -35,9 +35,9 @@ NINE_ROUTER_CLAUDE_PRO_NAME = "OpenSwarm Pro (OpenSwarm-managed)"
 NINE_ROUTER_OPENAI_KEYED_PREFIX = "cp-openai"


-async def _find_keyed_connection(provider: str, name: str) -> dict | None:
+async def find_keyed_connection(provider: str, name: str) -> dict | None:
    """Return the 9Router connection we manage for this provider, if any."""
-    conns = await _nr().get_providers()
+    conns = await nr().get_providers()
    if not isinstance(conns, list):
        return None
    for c in conns:
@@ -51,7 +51,7 @@ async def _find_keyed_connection(provider: str, name: str) -> dict | None:
    return None


-async def _sync_apikey_provider(
+async def p_sync_apikey_provider(
    provider: str,
    api_key: str | None,
    name: str,
@@ -59,12 +59,12 @@ async def _sync_apikey_provider(
    label: str,
 ) -> None:
    """Create/update/delete an OpenSwarm-managed apikey connection. Silent if 9Router is down."""
-    if not _nr().is_running():
+    if not nr().is_running():
        return

-    existing = await _find_keyed_connection(provider, name)
+    existing = await find_keyed_connection(provider, name)
    try:
-        async with _nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
+        async with nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
            if api_key:
                payload = {
                    "provider": provider,
@@ -100,7 +100,7 @@ async def _sync_apikey_provider(

 async def sync_gemini_api_key(api_key: str | None) -> None:
    """Mirror google_api_key into 9Router; bypasses Code Assist's tight quota."""
-    await _sync_apikey_provider(
+    await p_sync_apikey_provider(
        "gemini", api_key, NINE_ROUTER_KEYED_NAME, label="Gemini"
    )

@@ -128,12 +128,12 @@ async def sync_openai_api_key(api_key: str | None) -> None:
    above) so 9Router's translator dispatches to this provider-node
    instead of the built-in `openai` provider.
    """
-    from .sync_custom import _sync_openai_compat_node
-    await _sync_openai_compat_node(api_key)
+    from .sync_custom import sync_openai_compat_node
+    await sync_openai_compat_node(api_key)


 async def sync_openrouter_api_key(api_key: str | None) -> None:
    """Mirror openrouter_api_key into 9Router; supplies bearer for openrouter/ routes."""
-    await _sync_apikey_provider(
+    await p_sync_apikey_provider(
        "openrouter", api_key, NINE_ROUTER_OPENROUTER_KEYED_NAME, label="OpenRouter"
    )
@@ -14,8 +14,8 @@ from .process import NINE_ROUTER_API, cli_auth_headers
 from .sync import (
    NINE_ROUTER_CLAUDE_PRO_NAME,
    NINE_ROUTER_OPENAI_KEYED_PREFIX,
-    _find_keyed_connection,
-    _nr,
+    find_keyed_connection,
+    nr,
 )

 logger = logging.getLogger(__name__)
@@ -25,10 +25,10 @@ logger = logging.getLogger(__name__)
 NINE_ROUTER_CUSTOM_NAME_SUFFIX = " (OpenSwarm-managed)"


-async def _sync_openai_compat_node(api_key: str | None) -> None:
+async def sync_openai_compat_node(api_key: str | None) -> None:
    """Create / update / delete the openai-compatible node + connection
    pair we use to ferry OpenAI requests through openai-passthrough."""
-    if not _nr().is_running():
+    if not nr().is_running():
        return
    import os as _os
    port = _os.environ.get("OPENSWARM_PORT", "8324")
@@ -36,7 +36,7 @@ async def _sync_openai_compat_node(api_key: str | None) -> None:
    managed_name = f"OpenAI{NINE_ROUTER_CUSTOM_NAME_SUFFIX}"

    try:
-        async with _nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
+        async with nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
            r = await client.get(f"{NINE_ROUTER_API}/provider-nodes")
            existing_nodes = (r.json().get("nodes") if r.status_code == 200 else []) or []
    except Exception as e:
@@ -50,7 +50,7 @@ async def _sync_openai_compat_node(api_key: str | None) -> None:
    if not api_key:
        if existing_node:
            try:
-                async with _nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
+                async with nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
                    await client.delete(f"{NINE_ROUTER_API}/provider-nodes/{existing_node['id']}")
                logger.info("9Router: removed OpenAI compat node (key cleared)")
            except Exception as e:
@@ -66,7 +66,7 @@ async def _sync_openai_compat_node(api_key: str | None) -> None:
    }
    node_id: str | None = existing_node.get("id") if existing_node else None
    try:
-        async with _nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
+        async with nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
            if existing_node:
                await client.put(
                    f"{NINE_ROUTER_API}/provider-nodes/{existing_node['id']}",
@@ -92,7 +92,7 @@ async def _sync_openai_compat_node(api_key: str | None) -> None:
        return

    try:
-        existing_conn = await _find_keyed_connection(node_id, managed_name)
+        existing_conn = await find_keyed_connection(node_id, managed_name)
        conn_payload = {
            "provider": node_id,
            "authType": "apikey",
@@ -100,7 +100,7 @@ async def _sync_openai_compat_node(api_key: str | None) -> None:
            "apiKey": api_key,
            "priority": 0,
        }
-        async with _nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
+        async with nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
            if existing_conn:
                await client.patch(
                    f"{NINE_ROUTER_API}/providers/{existing_conn['id']}",
@@ -117,7 +117,7 @@ async def _sync_openai_compat_node(api_key: str | None) -> None:
        logger.warning(f"9Router OpenAI compat connection sync failed: {e}")


-def _custom_provider_slug(name: str) -> str:
+def p_custom_provider_slug(name: str) -> str:
    """Slugify a user-supplied custom-provider name for use as a 9Router prefix.
    Always returns a non-empty alnum-and-dash string."""
    import re
@@ -157,11 +157,11 @@ async def sync_custom_providers(providers: list) -> None:
    no longer in `providers` is deleted (which cascades to its connection).
    Silent no-op when 9Router isn't running.
    """
-    if not _nr().is_running():
+    if not nr().is_running():
        return

    try:
-        async with _nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
+        async with nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
            r = await client.get(f"{NINE_ROUTER_API}/provider-nodes")
            existing_nodes = (r.json().get("nodes") if r.status_code == 200 else []) or []
    except Exception as e:
@@ -187,7 +187,7 @@ async def sync_custom_providers(providers: list) -> None:
        # Bearer header even with auth disabled. Substitute a placeholder; real
        # auth deployments always have api_key set.
        api_key = api_key.strip() or "no-auth-required"
-        slug = _custom_provider_slug(name)
+        slug = p_custom_provider_slug(name)
        prefix = f"cp-{slug}"
        seen_prefixes.add(prefix)
        managed_name = f"{name.strip()}{NINE_ROUTER_CUSTOM_NAME_SUFFIX}"
@@ -201,7 +201,7 @@ async def sync_custom_providers(providers: list) -> None:
            "type": "openai-compatible",
        }
        try:
-            async with _nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
+            async with nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
                if node:
                    await client.put(
                        f"{NINE_ROUTER_API}/provider-nodes/{node['id']}",
@@ -228,7 +228,7 @@ async def sync_custom_providers(providers: list) -> None:
            continue

        try:
-            existing_conn = await _find_keyed_connection(node_id, managed_name)
+            existing_conn = await find_keyed_connection(node_id, managed_name)
            conn_payload = {
                "provider": node_id,
                "authType": "apikey",
@@ -236,7 +236,7 @@ async def sync_custom_providers(providers: list) -> None:
                "apiKey": api_key,
                "priority": 0,
            }
-            async with _nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
+            async with nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
                if existing_conn:
                    await client.patch(
                        f"{NINE_ROUTER_API}/providers/{existing_conn['id']}",
@@ -259,7 +259,7 @@ async def sync_custom_providers(providers: list) -> None:
        if prefix in seen_prefixes:
            continue
        try:
-            async with _nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
+            async with nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
                await client.delete(f"{NINE_ROUTER_API}/provider-nodes/{node['id']}")
                logger.info(f"9Router: removed orphaned custom node {prefix}")
        except Exception as e:
@@ -280,15 +280,15 @@ async def sync_openswarm_pro_as_claude(bearer_token: str | None, proxy_url: str
    the OpenSwarm-Pro-backed Claude connection and routes the search
    call through our cloud; same quota the user's Pro subscription
    already covers, no extra cost."""
-    if not _nr().is_running():
+    if not nr().is_running():
        return

    # 9Router's POST /api/providers only accepts direct-API provider ids
    # for apikey auth; `claude` is the subscription/IDE id, `anthropic`
    # is the direct-API id. Use `anthropic`.
-    existing = await _find_keyed_connection("anthropic", NINE_ROUTER_CLAUDE_PRO_NAME)
+    existing = await find_keyed_connection("anthropic", NINE_ROUTER_CLAUDE_PRO_NAME)
    try:
-        async with _nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
+        async with nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client:
            if bearer_token and proxy_url:
                payload = {
                    "provider": "anthropic",
@@ -31,15 +31,15 @@ from .registry import IMPORT_ORDER, get_exportable
 from .ziputil import MANIFEST_NAME, BundleError, has_member, is_zip, pack, read_manifest, unpack, verify_checksum


-def _now() -> str:
+def p_now() -> str:
    return datetime.now(timezone.utc).isoformat()


-def _created_with() -> str:
+def p_created_with() -> str:
    return os.environ.get("OPENSWARM_VERSION") or "OpenSwarm"


-class _Ctx:
+class p_Ctx:
    def __init__(self, local_to_bundle: dict[tuple, str]):
        self._m = local_to_bundle

@@ -49,7 +49,7 @@ class _Ctx:

 # ---------- export ----------

-def _assemble(root_type: EntityType, root_id: str):
+def p_assemble(root_type: EntityType, root_id: str):
    root_cls = get_exportable(root_type)
    if root_cls is None:
        raise BundleError(f"can't share a {root_type.value} yet")
@@ -79,7 +79,7 @@ def _assemble(root_type: EntityType, root_id: str):
                queue.append((dep.type, dep.local_id, dinst))

    local_to_bundle = {key: uuid4().hex for key in order}
-    ctx = _Ctx(local_to_bundle)
+    ctx = p_Ctx(local_to_bundle)
    payloads: dict[str, dict] = {}
    files: dict[str, bytes] = {}
    entities: list[EntityRef] = []
@@ -102,11 +102,11 @@ def _assemble(root_type: EntityType, root_id: str):
                edges.append(DependencyEdge(from_=bid, to=local_to_bundle[dkey], relation=dep.relation))
        requirements.extend(inst.requirements())

-    requirements = _dedupe_requirements(requirements)
+    requirements = p_dedupe_requirements(requirements)
    root_bid = local_to_bundle[(root_type, root_id)]
    manifest = Manifest(
-        created_with=_created_with(),
-        created_at=_now(),
+        created_with=p_created_with(),
+        created_at=p_now(),
        bundle_id=uuid4().hex,
        root=EntityRef(type=root_type, bundle_id=root_bid, name=root.name, path=f"entities/{root_bid}"),
        entities=entities,
@@ -123,16 +123,16 @@ def _assemble(root_type: EntityType, root_id: str):


 def build_manifest(root_type: EntityType, root_id: str) -> Manifest:
-    return _assemble(root_type, root_id)[0]
+    return p_assemble(root_type, root_id)[0]


 def build_bundle(root_type: EntityType, root_id: str) -> tuple[bytes, str]:
-    manifest, payloads, files = _assemble(root_type, root_id)
+    manifest, payloads, files = p_assemble(root_type, root_id)
    raw = pack(manifest.model_dump(by_alias=True, mode="json"), payloads, files)
    return raw, manifest.root.name


-def _dedupe_requirements(reqs: list[Requirement]) -> list[Requirement]:
+def p_dedupe_requirements(reqs: list[Requirement]) -> list[Requirement]:
    out: dict[tuple, Requirement] = {}
    for r in reqs:
        k = (r.kind, r.key)
@@ -210,24 +210,24 @@ def stage_upload(raw: bytes, filename: str) -> tuple[str, Manifest, list[str]]:
                shutil.rmtree(sandbox, ignore_errors=True)
                raise BundleError("this .swarm was made by a newer OpenSwarm; please update")
            return sandbox, manifest, warnings
-        return _stage_skill_from_zip(raw, filename, warnings)
-    return _stage_skill_from_markdown(raw, filename, warnings)
+        return stage_skill_from_zip(raw, filename, warnings)
+    return p_stage_skill_from_markdown(raw, filename, warnings)


-def _name_from_filename(filename: str) -> str:
+def p_name_from_filename(filename: str) -> str:
    base = os.path.splitext(os.path.basename(filename or "skill"))[0]
    return base.replace("-", " ").replace("_", " ").strip().title() or "Imported Skill"


-def _stage_skill_from_markdown(raw: bytes, filename: str, warnings: list[str]):
+def p_stage_skill_from_markdown(raw: bytes, filename: str, warnings: list[str]):
    try:
        content = raw.decode("utf-8")
    except UnicodeDecodeError:
        raise BundleError("unrecognized file; expected a .swarm or a .md skill")
-    return _synth_single_skill(content, _name_from_filename(filename), warnings)
+    return p_synth_single_skill(content, p_name_from_filename(filename), warnings)


-def _stage_skill_from_zip(raw: bytes, filename: str, warnings: list[str]):
+def stage_skill_from_zip(raw: bytes, filename: str, warnings: list[str]):
    with zipfile.ZipFile(io.BytesIO(raw)) as zf:
        mds = [n for n in zf.namelist() if n.lower().endswith(".md") and not n.endswith("/")]
        target = next((n for n in mds if os.path.basename(n).lower() == "skill.md"), None)
@@ -253,10 +253,10 @@ def _stage_skill_from_zip(raw: bytes, filename: str, warnings: list[str]):
                warnings.append("some oversized/extra supporting files were skipped")
                continue
            extra_files[rel] = zf.read(n)
-    return _synth_single_skill(content, _name_from_filename(filename), warnings, extra_files)
+    return p_synth_single_skill(content, p_name_from_filename(filename), warnings, extra_files)


-def _synth_single_skill(content: str, name: str, warnings: list[str], extra_files: dict[str, bytes] | None = None):
+def p_synth_single_skill(content: str, name: str, warnings: list[str], extra_files: dict[str, bytes] | None = None):
    bid = uuid4().hex
    sandbox = tempfile.mkdtemp(prefix="swarm-import-")
    edir = os.path.join(sandbox, "entities", bid)
@@ -266,10 +266,10 @@ def _synth_single_skill(content: str, name: str, warnings: list[str], extra_file
    with open(os.path.join(edir, "payload.json"), "w", encoding="utf-8") as f:
        json.dump(payload, f)
    # Supporting files ride the same entities/<bid>/files/<rel> channel the
-    # commit reader (_read_files) feeds into import_, so a zip-of-SKILL.md
+    # commit reader (p_read_files) feeds into import_, so a zip-of-SKILL.md
    # round-trips as a folder skill instead of getting flattened.
    for rel, data in (extra_files or {}).items():
-        dest = _safe_join(edir, os.path.join("files", rel))
+        dest = p_safe_join(edir, os.path.join("files", rel))
        os.makedirs(os.path.dirname(dest), exist_ok=True)
        with open(dest, "wb") as f:
            f.write(data)
@@ -285,7 +285,7 @@ def _synth_single_skill(content: str, name: str, warnings: list[str], extra_file

 # ---------- import: commit ----------

-def _safe_join(sandbox: str, rel: str) -> str:
+def p_safe_join(sandbox: str, rel: str) -> str:
    dest = os.path.realpath(os.path.join(sandbox, rel))
    root = os.path.realpath(sandbox)
    if dest != root and not dest.startswith(root + os.sep):
@@ -293,14 +293,14 @@ def _safe_join(sandbox: str, rel: str) -> str:
    return dest


-def _read_payload(sandbox: str, ref: EntityRef) -> dict:
-    path = _safe_join(sandbox, os.path.join(ref.path, "payload.json"))
+def p_read_payload(sandbox: str, ref: EntityRef) -> dict:
+    path = p_safe_join(sandbox, os.path.join(ref.path, "payload.json"))
    with open(path, encoding="utf-8") as f:
        return json.load(f)


-def _read_files(sandbox: str, ref: EntityRef) -> dict[str, bytes]:
-    base = _safe_join(sandbox, os.path.join(ref.path, "files"))
+def p_read_files(sandbox: str, ref: EntityRef) -> dict[str, bytes]:
+    base = p_safe_join(sandbox, os.path.join(ref.path, "files"))
    out: dict[str, bytes] = {}
    if not os.path.isdir(base):
        return out
@@ -326,7 +326,7 @@ def review_bundle(sandbox: str, manifest: Manifest):
        if e.type != EntityType.app:
            continue
        any_app = True
-        r = scan_app_files(_read_files(sandbox, e))
+        r = scan_app_files(p_read_files(sandbox, e))
        findings.extend(r.findings)
        scanned.extend(r.scanned_files)
        if r.verdict != "clean":
@@ -341,13 +341,13 @@ def detect_conflicts(sandbox: str, manifest: Manifest) -> list[IncludeItem]:
        check = getattr(cls, "conflict", None) if cls else None
        if not check:
            continue
-        msg = check(_read_payload(sandbox, e))
+        msg = check(p_read_payload(sandbox, e))
        if msg:
            out.append(IncludeItem(type=e.type, name=e.name, detail=msg))
    return out


-def _topo_order(manifest: Manifest) -> list[EntityRef]:
+def p_topo_order(manifest: Manifest) -> list[EntityRef]:
    entities = {e.bundle_id: e for e in manifest.entities}
    deps: dict[str, set[str]] = {bid: set() for bid in entities}
    for edge in manifest.edges:
@@ -372,11 +372,11 @@ def commit(sandbox: str, manifest: Manifest, accept_requirements: list[str]):
    created: dict[str, list[str]] = {}
    trail: list[tuple] = []  # (impl_cls, new_local_id) for rollback, newest last
    try:
-        for e in _topo_order(manifest):
+        for e in p_topo_order(manifest):
            cls = get_exportable(e.type)
            if cls is None:
                raise BundleError(f"can't import a {e.type.value} yet")
-            new_id = cls.import_(_read_payload(sandbox, e), _read_files(sandbox, e), remap)
+            new_id = cls.import_(p_read_payload(sandbox, e), p_read_files(sandbox, e), remap)
            remap.assign(e.bundle_id, new_id)
            created.setdefault(e.type.value, []).append(new_id)
            trail.append((cls, new_id))
@@ -18,7 +18,7 @@ from backend.config.paths import OUTPUTS_DIR, OUTPUTS_WORKSPACE_DIR
 from ..exportable import DepRef, ExportContext, RemapTable
 from ..models import EntityType, Requirement

-_MAX_APP_FILE = 25 * 1024 * 1024  # matches ziputil per-entry cap
+P_MAX_APP_FILE = 25 * 1024 * 1024  # matches ziputil per-entry cap


 class AppExportable:
@@ -61,7 +61,7 @@ class AppExportable:
                if os.path.islink(full):
                    continue
                try:
-                    if os.path.getsize(full) > _MAX_APP_FILE:
+                    if os.path.getsize(full) > P_MAX_APP_FILE:
                        continue
                    with open(full, "rb") as f:
                        data = f.read()
@@ -85,13 +85,13 @@ class AppExportable:
        for rel, data in files.items():
            if not rel.startswith("workspace/"):
                continue
-            dest = _safe_join(folder, rel[len("workspace/"):])
+            dest = p_safe_join(folder, rel[len("workspace/"):])
            os.makedirs(os.path.dirname(dest), exist_ok=True)
            with open(dest, "wb") as f:
                f.write(data)
            wrote_workspace = True
        if wrote_workspace:
-            _localize_env(folder)
+            p_localize_env(folder)

        o = Output(
            name=payload.get("name") or "Imported App",
@@ -115,7 +115,7 @@ class AppExportable:
            os.remove(p)


-def _safe_join(folder: str, rel: str) -> str:
+def p_safe_join(folder: str, rel: str) -> str:
    dest = os.path.realpath(os.path.join(folder, rel))
    root = os.path.realpath(folder)
    if dest != root and not dest.startswith(root + os.sep):
@@ -123,7 +123,7 @@ def _safe_join(folder: str, rel: str) -> str:
    return dest


-def _free_port() -> int:
+def p_free_port() -> int:
    s = socket.socket()
    try:
        s.bind(("127.0.0.1", 0))
@@ -132,7 +132,7 @@ def _free_port() -> int:
        s.close()


-def _localize_env(folder: str) -> None:
+def p_localize_env(folder: str) -> None:
    """Regenerate the workspace .env on the importer's machine: a fresh port plus
    this install's absolute template/debugger paths (the source's were dropped)."""
    env_path = os.path.join(folder, ".env")
@@ -151,7 +151,7 @@ def _localize_env(folder: str) -> None:
        )
    except Exception:
        return
-    patch_env_port(env_path, "FRONTEND_PORT", str(_free_port()))
+    patch_env_port(env_path, "FRONTEND_PORT", str(p_free_port()))
    patch_env_port(env_path, "OPENSWARM_TEMPLATE_BACKEND_PATH", TEMPLATE_BACKEND_PATH)
    patch_env_port(env_path, "OPENSWARM_DEBUGGER_PATH", DEBUGGER_PATH)
    try:
@@ -23,7 +23,7 @@ class DashboardExportable:

    @classmethod
    def load(cls, local_id: str) -> "DashboardExportable | None":
-        data = _read(local_id)
+        data = p_read(local_id)
        if data is None:
            return None
        return cls(local_id, data.get("name") or "Dashboard", data)
@@ -111,21 +111,21 @@ class DashboardExportable:
                "expanded_session_ids": expanded,
            },
        }
-        _write(new_did, doc)
-        _retag_sessions(cards.keys(), new_did)
+        p_write(new_did, doc)
+        p_retag_sessions(cards.keys(), new_did)
        return new_did

    @classmethod
    def rollback(cls, local_id: str) -> None:
        import os
-        d = _dash_dir()
+        d = p_dash_dir()
        if d:
            p = os.path.join(d, f"{local_id}.json")
            if os.path.exists(p):
                os.remove(p)


-def _dash_dir() -> str | None:
+def p_dash_dir() -> str | None:
    try:
        from backend.config.paths import DASHBOARDS_DIR
        return DASHBOARDS_DIR
@@ -133,22 +133,22 @@ def _dash_dir() -> str | None:
        return None


-def _read(did: str) -> dict | None:
+def p_read(did: str) -> dict | None:
    import os
    from backend.config.json_store import read_json_or_none
-    d = _dash_dir()
+    d = p_dash_dir()
    return read_json_or_none(os.path.join(d, f"{did}.json")) if d else None


-def _write(did: str, doc: dict) -> None:
+def p_write(did: str, doc: dict) -> None:
    import os
    from backend.config.json_store import atomic_write_json
-    d = _dash_dir()
+    d = p_dash_dir()
    if d:
        atomic_write_json(os.path.join(d, f"{did}.json"), doc)


-def _retag_sessions(session_ids, dashboard_id: str) -> None:
+def p_retag_sessions(session_ids, dashboard_id: str) -> None:
    # Best-effort: a hiccup here must not orphan the just-written dashboard.
    from backend.apps.agents.manager.session.session_store import load_session_data, save_session
    for sid in session_ids:
@@ -10,7 +10,7 @@ from ..exportable import DepRef, ExportContext, RemapTable
 from ..models import EntityType, Requirement

 # Machine-relative or install-owned fields that must not ride along.
-_DROP = {"is_builtin", "default_folder"}
+P_DROP = {"is_builtin", "default_folder"}


 class ModeExportable:
@@ -23,7 +23,7 @@ class ModeExportable:

    @classmethod
    def load(cls, local_id: str) -> "ModeExportable | None":
-        store = _store()
+        store = p_store()
        if store is None:
            return None
        m = store.load_mode(local_id)
@@ -33,7 +33,7 @@ class ModeExportable:
        return cls(local_id, d.get("name") or local_id, d)

    def serialize(self, ctx: ExportContext) -> dict:
-        return {k: v for k, v in self._data.items() if k not in _DROP}
+        return {k: v for k, v in self._data.items() if k not in P_DROP}

    def files(self) -> dict[str, bytes]:
        return {}
@@ -46,8 +46,8 @@ class ModeExportable:

    @classmethod
    def import_(cls, payload: dict, files: dict[str, bytes], remap: RemapTable) -> str:
-        store = _store()
-        model = _model()
+        store = p_store()
+        model = p_model()
        if store is None or model is None:
            from ..ziputil import BundleError
            raise BundleError("can't import this mode on this build")
@@ -63,7 +63,7 @@ class ModeExportable:
        return mid


-def _store():
+def p_store():
    try:
        from backend.apps.modes import modes
        return modes
@@ -71,7 +71,7 @@ def _store():
        return None


-def _model():
+def p_model():
    try:
        from backend.apps.modes.models import Mode
        return Mode
@@ -15,11 +15,11 @@ from uuid import uuid4
 from ..exportable import DepRef, ExportContext, RemapTable
 from ..models import EntityType, Requirement, RequirementKind

-_BUILTIN_MODES = {"agent", "ask", "plan", "view-builder", "skill-builder"}
+P_BUILTIN_MODES = {"agent", "ask", "plan", "view-builder", "skill-builder"}
 # Transcript fields ride along so the shared agent keeps its history; ids inside
 # (message ids, branch ids, their parent/fork refs) are self-consistent within
 # the one session file, so they carry verbatim with no remap.
-_KEEP = (
+P_KEEP = (
    "name", "provider", "model", "mode", "system_prompt", "allowed_tools",
    "max_turns", "thinking_level",
    "messages", "branches", "active_branch_id", "tool_group_meta",
@@ -52,14 +52,14 @@ class SessionExportable:
        return cls(local_id, d.get("name") or "Agent", d)

    def serialize(self, ctx: ExportContext) -> dict:
-        return {k: self._data.get(k) for k in _KEEP if k in self._data}
+        return {k: self._data.get(k) for k in P_KEEP if k in self._data}

    def files(self) -> dict[str, bytes]:
        return {}

    def dependencies(self) -> list[DepRef]:
        mode = self._data.get("mode")
-        if mode and mode not in _BUILTIN_MODES:
+        if mode and mode not in P_BUILTIN_MODES:
            return [DepRef(EntityType.mode, mode, "uses_mode")]
        return []

@@ -71,7 +71,7 @@ class SessionExportable:
                detail="An agent here uses this action.",
            ))
        mode = self._data.get("mode") or "agent"
-        if mode in _BUILTIN_MODES and mode != "agent":
+        if mode in P_BUILTIN_MODES and mode != "agent":
            reqs.append(Requirement(
                kind=RequirementKind.builtin_mode, key=mode, label=f"{mode} mode",
                detail="A built-in mode an agent runs in.",
@@ -42,7 +42,7 @@ class SkillExportable:
        }
        files: dict[str, bytes] = {}
        if kind == "folder":
-            files = _read_supporting_files(os.path.join(store.SKILLS_DIR, local_id))
+            files = p_read_supporting_files(os.path.join(store.SKILLS_DIR, local_id))
        return cls(local_id, name, payload, files)

    def serialize(self, ctx: ExportContext) -> dict:
@@ -60,14 +60,14 @@ class SkillExportable:
    @classmethod
    def conflict(cls, payload: dict) -> str | None:
        slug = payload.get("slug") or ""
-        if slug and _slug_taken(slug):
+        if slug and p_slug_taken(slug):
            return "already exists; will be added as a copy"
        return None

    @classmethod
    def import_(cls, payload: dict, files: dict[str, bytes], remap: RemapTable) -> str:
        base = (payload.get("slug") or payload.get("name") or "skill").lower().replace(" ", "-")
-        slug = _free_slug(base)
+        slug = p_free_slug(base)
        meta = {
            "name": payload.get("name", slug),
            "description": payload.get("description", ""),
@@ -96,7 +96,7 @@ class SkillExportable:
            store._save_index(index)


-def _read_supporting_files(skill_dir: str) -> dict[str, bytes]:
+def p_read_supporting_files(skill_dir: str) -> dict[str, bytes]:
    """Every file in a skill folder except SKILL.md, as {relpath: bytes}."""
    out: dict[str, bytes] = {}
    for root, _dirs, names in os.walk(skill_dir):
@@ -113,7 +113,7 @@ def _read_supporting_files(skill_dir: str) -> dict[str, bytes]:
    return out


-def _slug_taken(slug: str) -> bool:
+def p_slug_taken(slug: str) -> bool:
    return (
        slug in store._load_index()
        or os.path.isfile(os.path.join(store.SKILLS_DIR, f"{slug}.md"))
@@ -121,14 +121,14 @@ def _slug_taken(slug: str) -> bool:
    )


-def _free_slug(base: str) -> str:
+def p_free_slug(base: str) -> str:
    base = base or "skill"
-    if not _slug_taken(base):
+    if not p_slug_taken(base):
        return base
    cand = f"{base}-imported"
-    if not _slug_taken(cand):
+    if not p_slug_taken(cand):
        return cand
    i = 2
-    while _slug_taken(f"{base}-imported-{i}"):
+    while p_slug_taken(f"{base}-imported-{i}"):
        i += 1
    return f"{base}-imported-{i}"
@@ -13,18 +13,18 @@ from __future__ import annotations
 from ..exportable import DepRef, ExportContext, RemapTable
 from ..models import EntityType, Requirement, RequirementKind

-_BUILTIN_MODES = {"agent", "ask", "plan", "view-builder", "skill-builder"}
+P_BUILTIN_MODES = {"agent", "ask", "plan", "view-builder", "skill-builder"}

 # Run-state, machine-linkage, and identifiers that must not ride along.
-_DROP_FIELDS = {
+P_DROP_FIELDS = {
    "id", "source_session_id", "dashboard_id", "edit_agent_session_id",
    "last_run_at", "last_run_status", "last_run_id", "next_run_at",
    "created_at", "updated_at", "cost_cap_usd_monthly",
 }


-def _sanitize_workflow(data: dict) -> dict:
-    out = {k: v for k, v in data.items() if k not in _DROP_FIELDS}
+def sanitize_workflow(data: dict) -> dict:
+    out = {k: v for k, v in data.items() if k not in P_DROP_FIELDS}
    sched = dict(out.get("schedule") or {})
    if sched:
        sched["enabled"] = False
@@ -52,7 +52,7 @@ class WorkflowExportable:

    @classmethod
    def load(cls, local_id: str) -> "WorkflowExportable | None":
-        store = _store()
+        store = p_store()
        if store is None:
            return None
        wf = store.get_workflow(local_id)
@@ -62,7 +62,7 @@ class WorkflowExportable:
        return cls(local_id, data.get("title") or "Untitled workflow", data)

    def serialize(self, ctx: ExportContext) -> dict:
-        return _sanitize_workflow(self._data)
+        return sanitize_workflow(self._data)

    def files(self) -> dict[str, bytes]:
        return {}
@@ -78,7 +78,7 @@ class WorkflowExportable:
                detail="This workflow uses this action.",
            ))
        mode = self._data.get("mode") or "agent"
-        if mode in _BUILTIN_MODES and mode != "agent":
+        if mode in P_BUILTIN_MODES and mode != "agent":
            reqs.append(Requirement(
                kind=RequirementKind.builtin_mode, key=mode, label=f"{mode} mode",
                detail="A built-in mode this workflow runs in.",
@@ -92,12 +92,12 @@ class WorkflowExportable:

    @classmethod
    def import_(cls, payload: dict, files: dict[str, bytes], remap: RemapTable) -> str:
-        store = _store()
-        model = _model()
+        store = p_store()
+        model = p_model()
        if store is None or model is None:
            from ..ziputil import BundleError
            raise BundleError("this build doesn't support workflows yet; please update OpenSwarm")
-        clean = _sanitize_workflow(payload)
+        clean = sanitize_workflow(payload)
        clean.pop("id", None)  # fresh id via the model's default_factory
        wf = model(**clean)
        store.save_workflow(wf)
@@ -105,7 +105,7 @@ class WorkflowExportable:

    @classmethod
    def rollback(cls, local_id: str) -> None:
-        store = _store()
+        store = p_store()
        if store is not None:
            try:
                store.delete_workflow(local_id)
@@ -113,7 +113,7 @@ class WorkflowExportable:
                pass


-def _store():
+def p_store():
    try:
        from backend.apps.workflows import storage
        return storage
@@ -121,7 +121,7 @@ def _store():
        return None


-def _model():
+def p_model():
    try:
        from backend.apps.workflows.models import Workflow
        return Workflow
@@ -8,7 +8,7 @@ import re
 from typing import Any

 # Substrings that mark a field name as secret (matched case-insensitively).
-_DENY_SUBSTRINGS = (
+P_DENY_SUBSTRINGS = (
    "api_key", "apikey", "secret", "password", "passwd", "credential", "oauth",
    "bearer", "subscription_token", "access_token", "refresh_token",
    "session_token", "auth_token", "private_key",
@@ -16,7 +16,7 @@ _DENY_SUBSTRINGS = (

 # Exact field names that are sensitive or per-install identity (the substring
 # pass alone would miss these).
-_DENY_EXACT = {
+P_DENY_EXACT = {
    "token", "installation_id", "user_id", "free_trial_token",
    "free_trial_remaining", "free_trial_runs_limit", "openswarm_bearer_token",
    "openswarm_usage_cached", "connected_account_email", "oauth_tokens",
@@ -35,9 +35,9 @@ from backend.common.secret_scan import (  # noqa: E402

 def is_denied_key(key: str) -> bool:
    k = key.lower()
-    if k in _DENY_EXACT:
+    if k in P_DENY_EXACT:
        return True
-    return any(sub in k for sub in _DENY_SUBSTRINGS)
+    return any(sub in k for sub in P_DENY_SUBSTRINGS)


 def scrub_payload(value: Any) -> Any:
@@ -25,31 +25,31 @@ from .ziputil import MAX_TOTAL_BYTES, BundleError

 logger = logging.getLogger(__name__)

-_STAGING: dict[str, dict] = {}
-_STAGING_TTL = 30 * 60  # 30 minutes
+P_STAGING: dict[str, dict] = {}
+P_STAGING_TTL = 30 * 60  # 30 minutes


-def _gc_staging() -> None:
+def p_gc_staging() -> None:
    now = time.time()
-    for token in list(_STAGING):
-        if now - _STAGING[token]["created_at"] > _STAGING_TTL:
-            _discard(token)
+    for token in list(P_STAGING):
+        if now - P_STAGING[token]["created_at"] > P_STAGING_TTL:
+            p_discard(token)


-def _discard(token: str) -> None:
-    entry = _STAGING.pop(token, None)
+def p_discard(token: str) -> None:
+    entry = P_STAGING.pop(token, None)
    if entry:
        shutil.rmtree(entry["sandbox"], ignore_errors=True)


@asynccontextmanager
 async def swarm_lifespan():
-    _gc_staging()
+    p_gc_staging()
    try:
        yield
    finally:
-        for token in list(_STAGING):
-            _discard(token)
+        for token in list(P_STAGING):
+            p_discard(token)


 swarm = SubApp("swarm", swarm_lifespan)
@@ -93,9 +93,9 @@ async def import_preflight(file: UploadFile = File(...)) -> ImportPreflightRespo
        review = closure.review_bundle(sandbox, manifest)
    except BundleError as e:
        raise HTTPException(status_code=400, detail=str(e))
-    _gc_staging()
+    p_gc_staging()
    token = uuid.uuid4().hex
-    _STAGING[token] = {"sandbox": sandbox, "manifest": manifest, "created_at": time.time()}
+    P_STAGING[token] = {"sandbox": sandbox, "manifest": manifest, "created_at": time.time()}
    return ImportPreflightResponse(
        summary=closure.summarize(manifest),
        staging_token=token,
@@ -107,7 +107,7 @@ async def import_preflight(file: UploadFile = File(...)) -> ImportPreflightRespo

@swarm.router.post("/import/commit")
 async def import_commit(body: ImportCommitRequest) -> ImportCommitResponse:
-    entry = _STAGING.get(body.staging_token)
+    entry = P_STAGING.get(body.staging_token)
    if not entry:
        raise HTTPException(status_code=404, detail="import session expired; please re-open the file")
    try:
@@ -117,7 +117,7 @@ async def import_commit(body: ImportCommitRequest) -> ImportCommitResponse:
    except BundleError as e:
        raise HTTPException(status_code=400, detail=str(e))
    finally:
-        _discard(body.staging_token)
+        p_discard(body.staging_token)
    if root_id is None:
        raise HTTPException(status_code=400, detail="bundle has no root entity")
    return ImportCommitResponse(
@@ -26,7 +26,7 @@ class BundleError(Exception):
    """Bundle is malformed or unsafe. Message is safe to show the user."""


-def _content_digest(entries: dict[str, bytes]) -> str:
+def p_content_digest(entries: dict[str, bytes]) -> str:
    """Order-independent sha256 over every non-manifest entry (path + bytes)."""
    h = hashlib.sha256()
    for path in sorted(entries):
@@ -57,7 +57,7 @@ def pack(manifest: dict, payloads: dict[str, dict], files: dict[str, bytes]) ->
        entries[f"entities/{bid}/payload.json"] = json.dumps(payload, indent=2).encode("utf-8")
    for path, data in files.items():
        entries[path] = data
-    manifest = {**manifest, "checksum": _content_digest(entries)}
+    manifest = {**manifest, "checksum": p_content_digest(entries)}
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
        zf.writestr(MANIFEST_NAME, json.dumps(manifest, indent=2))
@@ -66,7 +66,7 @@ def pack(manifest: dict, payloads: dict[str, dict], files: dict[str, bytes]) ->
    return buf.getvalue()


-def _sandbox_entries(sandbox: str) -> dict[str, bytes]:
+def p_sandbox_entries(sandbox: str) -> dict[str, bytes]:
    """Every file under the sandbox except the manifest, keyed by forward-slash
    relpath so it matches the keys pack() hashed (cross-platform)."""
    out: dict[str, bytes] = {}
@@ -88,11 +88,11 @@ def verify_checksum(sandbox: str, manifest: dict) -> None:
    expected = manifest.get("checksum")
    if not expected:
        return
-    if _content_digest(_sandbox_entries(sandbox)) != expected:
+    if p_content_digest(p_sandbox_entries(sandbox)) != expected:
        raise BundleError("this .swarm looks corrupted or was modified")


-def _safe_member_path(name: str, sandbox: str) -> str:
+def p_safe_member_path(name: str, sandbox: str) -> str:
    if name.startswith(("/", "\\")) or (len(name) > 1 and name[1] == ":"):
        raise BundleError("bundle contains an absolute path")
    dest = os.path.realpath(os.path.join(sandbox, name))
@@ -141,7 +141,7 @@ def unpack(raw: bytes) -> str:
        for zi in infos:
            if zi.is_dir():
                continue
-            dest = _safe_member_path(zi.filename, sandbox)
+            dest = p_safe_member_path(zi.filename, sandbox)
            os.makedirs(os.path.dirname(dest), exist_ok=True)
            with zf.open(zi) as src, open(dest, "wb") as out:
                while True:
@@ -209,12 +209,12 @@ async def test_create_writes_folder_and_supersedes_legacy_flat(skills_dir):

 def test_stage_zip_carries_supporting_files_into_sandbox():
    import io as _io, zipfile, os as _os, shutil
-    from backend.apps.swarm.closure import _stage_skill_from_zip
+    from backend.apps.swarm.closure import stage_skill_from_zip
    buf = _io.BytesIO()
    with zipfile.ZipFile(buf, "w") as zf:
        zf.writestr("my-skill/SKILL.md", "do it")
        zf.writestr("my-skill/scripts/run.sh", "echo hi")
-    sandbox, manifest, warnings = _stage_skill_from_zip(buf.getvalue(), "my-skill.zip", [])
+    sandbox, manifest, warnings = stage_skill_from_zip(buf.getvalue(), "my-skill.zip", [])
    try:
        bid = manifest.entities[0].bundle_id
        files_dir = _os.path.join(sandbox, "entities", bid, "files")
@@ -134,7 +134,7 @@ def test_app_export_drops_machine_env(tmp_path, monkeypatch):


 def test_workflow_sanitize_disables_schedule_and_strips_pii():
-    from backend.apps.swarm.entities.workflows import _sanitize_workflow
+    from backend.apps.swarm.entities.workflows import sanitize_workflow
    raw = {
        "id": "wf123",
        "title": "Daily digest",
@@ -147,7 +147,7 @@ def test_workflow_sanitize_disables_schedule_and_strips_pii():
        "mode": "agent",
        "provider": "anthropic",
    }
-    out = _sanitize_workflow(raw)
+    out = sanitize_workflow(raw)
    # An imported workflow must not auto-run or carry the sharer's identity.
    assert out["schedule"]["enabled"] is False
    assert out["schedule"]["runs_count"] == 0
@@ -384,8 +384,8 @@ def test_dashboard_import_remaps_to_fresh_local_ids(monkeypatch):
    from backend.apps.swarm.exportable import RemapTable

    written: dict = {}
-    monkeypatch.setattr(dmod, "_write", lambda did, doc: written.update({did: doc}))
-    monkeypatch.setattr(dmod, "_retag_sessions", lambda ids, did: None)
+    monkeypatch.setattr(dmod, "p_write", lambda did, doc: written.update({did: doc}))
+    monkeypatch.setattr(dmod, "p_retag_sessions", lambda ids, did: None)
    remap = RemapTable()
    remap.assign("SBID", "newsess")
    remap.assign("ABID", "newapp")
@@ -422,8 +422,8 @@ def test_dashboard_remap_invariant_generative(monkeypatch):
    from backend.apps.swarm.models import EntityType

    written: dict = {}
-    monkeypatch.setattr(dmod, "_write", lambda did, doc: written.update({did: doc}))
-    monkeypatch.setattr(dmod, "_retag_sessions", lambda ids, did: None)
+    monkeypatch.setattr(dmod, "p_write", lambda did, doc: written.update({did: doc}))
+    monkeypatch.setattr(dmod, "p_retag_sessions", lambda ids, did: None)

    rng = random.Random(1234)
    for _ in range(60):