From 87f09e9695cd4a72079a88da0b9061caa870a022 Mon Sep 17 00:00:00 2001
From: haikdc <haikdecie@gmail.com>
Date: Sat, 13 Jun 2026 22:01:17 -0700
Subject: [PATCH] [haik]: refactor naming convention and clean up dead code
 across backend outputs and service modules. Replace all leading-underscore
 private names with explicit p_/P_ prefix for module-internal symbols and drop
 the underscore for cross-module public APIs (load_all, save, load,
 walk_directory, find_free_port, etc). Rename the runtime module singleton
 from manager to RUNTIME_MANAGER. Remove unused legacy shims from client.py
 (submit_event, submit_session_close, record, set_test_sink, set_user_id,
 update_identity), drop clear() from ring_buffer.py and buffer.py, and remove
 the VIEW_BUILDER_SKILL backward-compat alias. Switch runtime.py from relative
 to absolute imports, rename the _d route param to data, and add public-usage
 comments to clarify cross-module API boundaries

---
 backend/apps/agents/agent_manager.py          |   6 +-
 backend/apps/outputs/html_inject.py           |  32 +-
 backend/apps/outputs/models.py                |   8 +-
 backend/apps/outputs/outputs.py               | 132 ++++----
 backend/apps/outputs/runtime.py               | 309 +++++++++---------
 backend/apps/outputs/runtime_proc.py          |  34 +-
 .../apps/outputs/view_builder_templates.py    | 114 +++----
 backend/apps/outputs/workspace_io.py          |  18 +-
 backend/apps/service/buffer.py                |  40 ++-
 backend/apps/service/client.py                | 210 +++++-------
 backend/apps/service/ring_buffer.py           |  19 +-
 backend/apps/service/service.py               | 116 ++++---
 backend/apps/service/version.py               |  15 +-
 backend/apps/settings/credentials.py          |   4 +-
 backend/main.py                               |   8 +-
 backend/tests/test_service.py                 |   2 +-
 16 files changed, 501 insertions(+), 566 deletions(-)

diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py
index d463a04a..c9c37fc8 100644
--- a/backend/apps/agents/agent_manager.py
+++ b/backend/apps/agents/agent_manager.py
@@ -893,10 +893,8 @@ class AgentManager:
                 if file_path:
                     try:
                         await asyncio.sleep(0.4)
-                        from backend.apps.outputs.runtime import (
-                            manager as _outputs_runtime_manager,
-                        )
-                        errs = _outputs_runtime_manager.drain_errors_for_path(file_path)
+                        from backend.apps.outputs.runtime import RUNTIME_MANAGER
+                        errs = RUNTIME_MANAGER.drain_errors_for_path(file_path)
                     except Exception:
                         errs = []
                     if errs:
diff --git a/backend/apps/outputs/html_inject.py b/backend/apps/outputs/html_inject.py
index 95fe04b5..8647416e 100644
--- a/backend/apps/outputs/html_inject.py
+++ b/backend/apps/outputs/html_inject.py
@@ -14,13 +14,13 @@ from jsonschema import validate as schema_validate, ValidationError as SchemaVal
 
 logger = logging.getLogger(__name__)
 
-MODEL_MAP = {
+P_MODEL_MAP = {
     "sonnet": "claude-sonnet-4-20250514",
     "opus": "claude-opus-4-20250514",
     "haiku": "claude-haiku-4-5-20251001",
 }
 
-def _get_anthropic_client(api_model: str | None = None):
+def get_anthropic_client(api_model: str | None = None):
     """Create an AsyncAnthropic client using the API key from app settings.
 
     When `api_model` is provided and carries a 9Router prefix (cc/, cx/, gc/),
@@ -39,7 +39,7 @@ def _get_anthropic_client(api_model: str | None = None):
     return get_anthropic_client(settings)
 
 
-def _validate_against_schema(data: dict, schema: dict) -> str | None:
+def validate_against_schema(data: dict, schema: dict) -> str | None:
     """Validate *data* against *schema*. Return an error string or None."""
     try:
         schema_validate(instance=data, schema=schema)
@@ -49,7 +49,7 @@ def _validate_against_schema(data: dict, schema: dict) -> str | None:
         return f"Schema validation failed at {path}: {exc.message}"
 
 
-def _build_data_injection(input_json: str, result_json: str, backend_url_json: str = "null") -> str:
+def p_build_data_injection(input_json: str, result_json: str, backend_url_json: str = "null") -> str:
     """Build a <script> tag that sets OUTPUT_INPUT / OUTPUT_BACKEND_RESULT /
     OUTPUT_BACKEND_URL and listens for postMessage updates.
 
@@ -76,8 +76,8 @@ def _build_data_injection(input_json: str, result_json: str, backend_url_json: s
     )
 
 
-def _inject_data_into_html(html: str, input_json: str = "{}", result_json: str = "null", backend_url_json: str = "null") -> str:
-    injection = _build_data_injection(input_json, result_json, backend_url_json)
+def inject_data_into_html(html: str, input_json: str = "{}", result_json: str = "null", backend_url_json: str = "null") -> str:
+    injection = p_build_data_injection(input_json, result_json, backend_url_json)
     if "</head>" in html:
         return html.replace("</head>", f"{injection}\n</head>", 1)
     if "<body" in html:
@@ -85,13 +85,13 @@ def _inject_data_into_html(html: str, input_json: str = "{}", result_json: str =
     return f"{injection}\n{html}"
 
 
-def _backend_url_for_workspace(workspace_id: str) -> str:
+def backend_url_for_workspace(workspace_id: str) -> str:
     """Return the JSON-encoded backend URL for the given workspace, or
     "null" if no runtime is active. Cheap inline lookup so serve_workspace_file
     doesn't have to think about it."""
     try:
-        from backend.apps.outputs.runtime import manager as runtime_manager
-        rt = runtime_manager.get(workspace_id)
+        from backend.apps.outputs.runtime import RUNTIME_MANAGER
+        rt = RUNTIME_MANAGER.get(workspace_id)
         if rt and rt.running and rt.port:
             return json.dumps(f"http://127.0.0.1:{rt.port}")
     except Exception:
@@ -103,18 +103,18 @@ def _backend_url_for_workspace(workspace_id: str) -> str:
 # external (CDNs, mailto) or non-network references that the auth middleware
 # never sees. Anything else is treated as a same-origin relative URL pointing
 # at our /api/outputs/.../serve/ subtree, which DOES need the token.
-_ABSOLUTE_URL_PREFIXES = (
+P_ABSOLUTE_URL_PREFIXES = (
     "http://", "https://", "//", "data:", "blob:",
     "mailto:", "tel:", "javascript:", "about:", "#",
 )
 
-_HREF_SRC_ATTR_RE = re.compile(
+P_HREF_SRC_ATTR_RE = re.compile(
     r"""(\s(?:href|src))\s*=\s*(["'])([^"']+)\2""",
     re.IGNORECASE,
 )
 
 
-def _inject_token_into_relative_urls(html: str, token: str) -> str:
+def inject_token_into_relative_urls(html: str, token: str) -> str:
     """Append `?token=<t>` to every relative href/src in the served HTML.
 
     Browsers strip the parent iframe URL's query string before resolving
@@ -126,10 +126,10 @@ def _inject_token_into_relative_urls(html: str, token: str) -> str:
     if not token:
         return html
 
-    def _patch(match: re.Match) -> str:
+    def patch(match: re.Match) -> str:
         attr, quote, url = match.group(1), match.group(2), match.group(3)
         lowered = url.lower().lstrip()
-        if lowered.startswith(_ABSOLUTE_URL_PREFIXES):
+        if lowered.startswith(P_ABSOLUTE_URL_PREFIXES):
             return match.group(0)
         if "token=" in url:
             return match.group(0)
@@ -143,10 +143,10 @@ def _inject_token_into_relative_urls(html: str, token: str) -> str:
         sep = "&" if "?" in base else "?"
         return f'{attr}={quote}{base}{sep}token={token}{frag}{quote}'
 
-    return _HREF_SRC_ATTR_RE.sub(_patch, html)
+    return P_HREF_SRC_ATTR_RE.sub(patch, html)
 
 
-def _decode_data_param(d: str) -> tuple[str, str]:
+def decode_data_param(d: str) -> tuple[str, str]:
     """Decode the base64-encoded _d query param into (input_json, result_json)."""
     try:
         decoded = json.loads(base64.b64decode(d))
diff --git a/backend/apps/outputs/models.py b/backend/apps/outputs/models.py
index 4b844a1f..f915df0c 100644
--- a/backend/apps/outputs/models.py
+++ b/backend/apps/outputs/models.py
@@ -28,7 +28,7 @@ class Output(BaseModel):
 
     @model_validator(mode="before")
     @classmethod
-    def _migrate_flat_fields(cls, data: Any) -> Any:
+    def p_migrate_flat_fields(cls, data: Any) -> Any:
         """Migrate legacy frontend_code/backend_code fields into the files dict."""
         if not isinstance(data, dict):
             return data
@@ -71,7 +71,7 @@ class OutputCreate(BaseModel):
 
     @model_validator(mode="before")
     @classmethod
-    def _migrate_flat_fields(cls, data: Any) -> Any:
+    def p_migrate_flat_fields(cls, data: Any) -> Any:
         if not isinstance(data, dict):
             return data
         if "files" not in data or not data["files"]:
@@ -101,7 +101,7 @@ class OutputUpdate(BaseModel):
 
     @model_validator(mode="before")
     @classmethod
-    def _migrate_flat_fields(cls, data: Any) -> Any:
+    def p_migrate_flat_fields(cls, data: Any) -> Any:
         if not isinstance(data, dict):
             return data
         if "files" not in data:
@@ -166,7 +166,7 @@ class WorkspaceSeedRequest(BaseModel):
 
     @model_validator(mode="before")
     @classmethod
-    def _migrate_flat_fields(cls, data: Any) -> Any:
+    def p_migrate_flat_fields(cls, data: Any) -> Any:
         """Accept legacy frontend_code/backend_code/schema_json fields."""
         if not isinstance(data, dict):
             return data
diff --git a/backend/apps/outputs/outputs.py b/backend/apps/outputs/outputs.py
index ef02e9fd..4da9935e 100644
--- a/backend/apps/outputs/outputs.py
+++ b/backend/apps/outputs/outputs.py
@@ -22,18 +22,18 @@ from backend.apps.outputs.view_builder_templates import (
 from backend.apps.settings.settings import load_settings
 from backend.config.paths import OUTPUTS_DIR as DATA_DIR, OUTPUTS_WORKSPACE_DIR as WORKSPACE_DIR
 from backend.apps.outputs.html_inject import (
-    _get_anthropic_client,
-    _validate_against_schema,
-    _inject_data_into_html,
-    _backend_url_for_workspace,
-    _inject_token_into_relative_urls,
-    _decode_data_param,
+    get_anthropic_client,
+    validate_against_schema,
+    inject_data_into_html,
+    backend_url_for_workspace,
+    inject_token_into_relative_urls,
+    decode_data_param,
 )
 from backend.apps.outputs.workspace_io import (
-    _load_all,
-    _save,
-    _load,
-    _walk_directory,
+    load_all,
+    save,
+    load,
+    walk_directory,
 )
 from backend.apps.outputs.prompts import VIBE_CODE_SYSTEM_PROMPT
 
@@ -52,8 +52,8 @@ async def outputs_lifespan():
         # main backend dies, leaving ghost listeners on the .env-pinned
         # ports that block the next OpenSwarm launch's reload preview.
         try:
-            from backend.apps.outputs.runtime import manager as runtime_manager
-            killed = await runtime_manager.stop_all()
+            from backend.apps.outputs.runtime import RUNTIME_MANAGER
+            killed = await RUNTIME_MANAGER.stop_all()
             if killed:
                 logger.info("outputs lifespan: reaped %d workspace runtimes on shutdown", killed)
         except Exception:
@@ -68,7 +68,7 @@ outputs = SubApp("outputs", outputs_lifespan)
 # ---------------------------------------------------------------------------
 
 @outputs.router.get("/workspace/{workspace_id}/serve/{filepath:path}")
-async def serve_workspace_file(workspace_id: str, filepath: str, _d: str = ""):
+async def serve_workspace_file(workspace_id: str, filepath: str, data: str = ""):
     """Serve a file from a workspace folder. For index.html, inject OUTPUT data."""
     folder = os.path.join(WORKSPACE_DIR, workspace_id)
     full_path = os.path.normpath(os.path.join(folder, filepath))
@@ -81,31 +81,31 @@ async def serve_workspace_file(workspace_id: str, filepath: str, _d: str = ""):
         content = f.read()
 
     if filepath == "index.html":
-        input_json, result_json = _decode_data_param(_d) if _d else ("{}", "null")
-        backend_url_json = _backend_url_for_workspace(workspace_id)
-        content = _inject_data_into_html(content, input_json, result_json, backend_url_json)
+        input_json, result_json = decode_data_param(data) if data else ("{}", "null")
+        backend_url_json = backend_url_for_workspace(workspace_id)
+        content = inject_data_into_html(content, input_json, result_json, backend_url_json)
         # Iframe sub-resource fetches (<link>, <script src>, <img>) drop the
         # parent's ?token= query string, so rewrite the HTML to put the token
         # back on every relative URL; otherwise sub-resources 401.
-        content = _inject_token_into_relative_urls(content, get_auth_token())
+        content = inject_token_into_relative_urls(content, get_auth_token())
 
     mime, _ = mimetypes.guess_type(filepath)
     return Response(content=content, media_type=mime or "text/plain")
 
 
 @outputs.router.get("/{output_id}/serve/{filepath:path}")
-async def serve_output_file(output_id: str, filepath: str, _d: str = ""):
+async def serve_output_file(output_id: str, filepath: str, data: str = ""):
     """Serve a file from a saved output's files dict. For index.html, inject OUTPUT data."""
-    output = _load(output_id)
+    output = load(output_id)
     content = output.files.get(filepath)
     if content is None:
         raise HTTPException(status_code=404, detail="File not found in output")
 
     if filepath == "index.html":
-        input_json, result_json = _decode_data_param(_d) if _d else ("{}", "null")
-        backend_url_json = _backend_url_for_workspace(output.workspace_id) if output.workspace_id else "null"
-        content = _inject_data_into_html(content, input_json, result_json, backend_url_json)
-        content = _inject_token_into_relative_urls(content, get_auth_token())
+        input_json, result_json = decode_data_param(data) if data else ("{}", "null")
+        backend_url_json = backend_url_for_workspace(output.workspace_id) if output.workspace_id else "null"
+        content = inject_data_into_html(content, input_json, result_json, backend_url_json)
+        content = inject_token_into_relative_urls(content, get_auth_token())
 
     mime, _ = mimetypes.guess_type(filepath)
     return Response(content=content, media_type=mime or "text/plain")
@@ -117,7 +117,7 @@ async def serve_output_file(output_id: str, filepath: str, _d: str = ""):
 
 @outputs.router.get("/list")
 async def list_outputs():
-    return {"outputs": [o.model_dump() for o in _load_all()]}
+    return {"outputs": [o.model_dump() for o in load_all()]}
 
 
 @outputs.router.get("/workspace/{workspace_id}")
@@ -127,7 +127,7 @@ async def read_workspace(workspace_id: str):
     if not os.path.isdir(folder):
         raise HTTPException(status_code=404, detail="Workspace not found")
 
-    files = _walk_directory(folder)
+    files = walk_directory(folder)
 
     meta = None
     if "meta.json" in files:
@@ -171,7 +171,7 @@ def sync_output_from_meta_json(workspace_id: str) -> bool:
         description = str(meta.get("description") or "").strip()
         if not name and not description:
             return False
-        matching = [o for o in _load_all() if o.workspace_id == workspace_id]
+        matching = [o for o in load_all() if o.workspace_id == workspace_id]
         if not matching:
             return False
         output = matching[0]
@@ -187,7 +187,7 @@ def sync_output_from_meta_json(workspace_id: str) -> bool:
             changed = True
         if changed:
             output.updated_at = datetime.now().isoformat()
-            _save(output)
+            save(output)
         return changed
     except (OSError, json.JSONDecodeError, ValueError):
         return False
@@ -223,18 +223,18 @@ def ensure_webapp_workspace_seeded_and_registered(
         os.makedirs(folder, exist_ok=True)
         already_seeded = os.path.exists(os.path.join(folder, "run.sh"))
         if not already_seeded:
-            from backend.apps.outputs.runtime import _find_free_port
-            frontend_port = _find_free_port()
+            from backend.apps.outputs.runtime_proc import find_free_port
+            frontend_port = find_free_port()
             seed_webapp_template_workspace(folder, frontend_port)
             with open(os.path.join(folder, "SKILL.md"), "w", encoding="utf-8") as f:
                 f.write(load_app_builder_skill())
-        existing = [o for o in _load_all() if o.workspace_id == workspace_id]
+        existing = [o for o in load_all() if o.workspace_id == workspace_id]
         if existing:
             output = existing[0]
             if session_id and output.session_id != session_id:
                 output.session_id = session_id
                 output.updated_at = datetime.now().isoformat()
-                _save(output)
+                save(output)
             return output.id
         now = datetime.now().isoformat()
         output = Output(
@@ -247,7 +247,7 @@ def ensure_webapp_workspace_seeded_and_registered(
             created_at=now,
             updated_at=now,
         )
-        _save(output)
+        save(output)
         return output.id
     except Exception:
         logger.exception("ensure_webapp_workspace_seeded_and_registered failed for %s", workspace_id)
@@ -293,16 +293,16 @@ async def seed_workspace(body: WorkspaceSeedRequest):
         # dirs_exist_ok=True + copytree). If `run.sh` already exists,
         # the workspace was seeded on a previous visit; skip the file
         # copy and only re-derive the frontend port from .env.
-        from backend.apps.outputs.runtime import _find_free_port, _read_env_value
+        from backend.apps.outputs.runtime_proc import find_free_port, read_env_value
         already_seeded = os.path.exists(os.path.join(folder, "run.sh"))
         if already_seeded:
-            fp_raw = _read_env_value(os.path.join(folder, ".env"), "FRONTEND_PORT")
+            fp_raw = read_env_value(os.path.join(folder, ".env"), "FRONTEND_PORT")
             try:
-                frontend_port = int(fp_raw) if fp_raw else _find_free_port()
+                frontend_port = int(fp_raw) if fp_raw else find_free_port()
             except (TypeError, ValueError):
-                frontend_port = _find_free_port()
+                frontend_port = find_free_port()
         else:
-            frontend_port = _find_free_port()
+            frontend_port = find_free_port()
             seed_webapp_template_workspace(folder, frontend_port)
             # SKILL.md still goes in workspace root; agent reads it for
             # context. Live content (user-editable via Skills page) is
@@ -323,7 +323,7 @@ async def seed_workspace(body: WorkspaceSeedRequest):
         # remains the source of truth for the code.
         output_id: Optional[str] = None
         try:
-            existing = [o for o in _load_all() if o.workspace_id == body.workspace_id]
+            existing = [o for o in load_all() if o.workspace_id == body.workspace_id]
             if existing:
                 output_id = existing[0].id
             else:
@@ -337,7 +337,7 @@ async def seed_workspace(body: WorkspaceSeedRequest):
                     created_at=now,
                     updated_at=now,
                 )
-                _save(output)
+                save(output)
                 output_id = output.id
         except Exception:
             logger.exception("seed-time Output create failed for %s", body.workspace_id)
@@ -387,10 +387,10 @@ async def seed_workspace(body: WorkspaceSeedRequest):
 # ---------------------------------------------------------------------------
 
 
-def _runtime_status_payload(workspace_id: str) -> dict:
-    from backend.apps.outputs.runtime import manager as runtime_manager
-    from backend.apps.outputs.runtime import _is_new_mode
-    rt = runtime_manager.get(workspace_id)
+def runtime_status_payload(workspace_id: str) -> dict:
+    from backend.apps.outputs.runtime import RUNTIME_MANAGER
+    from backend.apps.outputs.runtime_proc import is_new_mode
+    rt = RUNTIME_MANAGER.get(workspace_id)
     if not rt:
         # Even without a live runtime, the editor needs is_new_mode to
         # decide whether the preview pane should fall back to the legacy
@@ -399,7 +399,7 @@ def _runtime_status_payload(workspace_id: str) -> dict:
         # Compute from disk so a failed runtime/start still gives the
         # client the right hint instead of dumping it onto a 404.
         folder = os.path.join(WORKSPACE_DIR, workspace_id)
-        is_new = _is_new_mode(folder) if os.path.isdir(folder) else False
+        is_new = is_new_mode(folder) if os.path.isdir(folder) else False
         return {
             "running": False,
             "port": None,
@@ -431,16 +431,16 @@ async def runtime_start(workspace_id: str):
     folder = os.path.join(WORKSPACE_DIR, workspace_id)
     if not os.path.isdir(folder):
         raise HTTPException(status_code=404, detail="Workspace not found")
-    from backend.apps.outputs.runtime import manager as runtime_manager
-    await runtime_manager.attach(workspace_id, os.path.abspath(folder))
-    return _runtime_status_payload(workspace_id)
+    from backend.apps.outputs.runtime import RUNTIME_MANAGER
+    await RUNTIME_MANAGER.attach(workspace_id, os.path.abspath(folder))
+    return runtime_status_payload(workspace_id)
 
 
 @outputs.router.post("/workspace/{workspace_id}/runtime/stop")
 async def runtime_stop(workspace_id: str):
-    from backend.apps.outputs.runtime import manager as runtime_manager
-    await runtime_manager.detach(workspace_id)
-    return _runtime_status_payload(workspace_id)
+    from backend.apps.outputs.runtime import RUNTIME_MANAGER
+    await RUNTIME_MANAGER.detach(workspace_id)
+    return runtime_status_payload(workspace_id)
 
 
 @outputs.router.post("/workspace/{workspace_id}/runtime/restart")
@@ -448,19 +448,19 @@ async def runtime_restart(workspace_id: str):
     folder = os.path.join(WORKSPACE_DIR, workspace_id)
     if not os.path.isdir(folder):
         raise HTTPException(status_code=404, detail="Workspace not found")
-    from backend.apps.outputs.runtime import manager as runtime_manager
+    from backend.apps.outputs.runtime import RUNTIME_MANAGER
     # Restart only if something's attached; otherwise this is a no-op
     # silently (a hard-reload click while the runtime was already torn
     # down; we'd rather not silently respawn an orphan).
-    rt = runtime_manager.get(workspace_id)
+    rt = RUNTIME_MANAGER.get(workspace_id)
     if rt:
-        await runtime_manager.restart(workspace_id, os.path.abspath(folder))
-    return _runtime_status_payload(workspace_id)
+        await RUNTIME_MANAGER.restart(workspace_id, os.path.abspath(folder))
+    return runtime_status_payload(workspace_id)
 
 
 @outputs.router.get("/workspace/{workspace_id}/runtime/status")
 async def runtime_get_status(workspace_id: str):
-    return _runtime_status_payload(workspace_id)
+    return runtime_status_payload(workspace_id)
 
 
 @outputs.router.post("/shutdown-all")
@@ -469,8 +469,8 @@ async def runtime_shutdown_all():
     will-quit so app subprocesses die BEFORE the main backend gets
     SIGTERM'd; without it `bash run.sh` + its vite/uvicorn descendants
     reparent to PID 1 and squat on .env-pinned ports forever."""
-    from backend.apps.outputs.runtime import manager as runtime_manager
-    killed = await runtime_manager.stop_all()
+    from backend.apps.outputs.runtime import RUNTIME_MANAGER
+    killed = await RUNTIME_MANAGER.stop_all()
     return {"ok": True, "killed": killed}
 
 
@@ -519,7 +519,7 @@ async def delete_workspace_file(workspace_id: str, filepath: str):
 
 @outputs.router.get("/{output_id}")
 async def get_output(output_id: str):
-    return _load(output_id).model_dump()
+    return load(output_id).model_dump()
 
 
 @outputs.router.post("/create")
@@ -535,14 +535,14 @@ async def create_output(body: OutputCreate):
         created_at=now,
         updated_at=now,
     )
-    _save(output)
+    save(output)
     pass
     return {"ok": True, "output": output.model_dump()}
 
 
 @outputs.router.put("/{output_id}")
 async def update_output(output_id: str, body: OutputUpdate):
-    output = _load(output_id)
+    output = load(output_id)
     # exclude_unset, NOT exclude_none: a PUT that explicitly sends session_id=null
     # (the Apps stale-link self-heal) must clear the field. exclude_none silently
     # dropped that null, so the dead pointer never cleared and the app 404'd on
@@ -554,13 +554,13 @@ async def update_output(output_id: str, body: OutputUpdate):
     # Only a real screenshot write moves the sort key; files/linkage saves don't reorder.
     if body.thumbnail is not None:
         output.preview_updated_at = now
-    _save(output)
+    save(output)
     return {"ok": True, "output": output.model_dump()}
 
 
 @outputs.router.delete("/{output_id}")
 async def delete_output(output_id: str):
-    _load(output_id)
+    load(output_id)
     path = os.path.join(DATA_DIR, f"{output_id}.json")
     if os.path.exists(path):
         os.remove(path)
@@ -589,7 +589,7 @@ async def vibe_code(body: VibeCodeRequest):
 
     from backend.apps.agents.providers.registry import resolve_aux_model
     try:
-        aux_model, _aux_base = await resolve_aux_model(load_settings(), preferred_tier="sonnet")
+        aux_model, _ = await resolve_aux_model(load_settings(), preferred_tier="sonnet")
     except ValueError as e:
         return {
             "message": f"Error: {str(e)}",
@@ -597,7 +597,7 @@ async def vibe_code(body: VibeCodeRequest):
             "backend_code": body.current_backend_code,
             "input_schema": body.current_schema,
         }
-    client = _get_anthropic_client(aux_model)
+    client = get_anthropic_client(aux_model)
     try:
         resp = await client.messages.create(
             model=aux_model,
@@ -648,9 +648,9 @@ async def vibe_code(body: VibeCodeRequest):
 
 @outputs.router.post("/execute")
 async def execute_output(body: OutputExecute):
-    output = _load(body.output_id)
+    output = load(body.output_id)
 
-    validation_err = _validate_against_schema(body.input_data, output.input_schema)
+    validation_err = validate_against_schema(body.input_data, output.input_schema)
     if validation_err:
         return OutputExecuteResult(
             output_id=output.id,
diff --git a/backend/apps/outputs/runtime.py b/backend/apps/outputs/runtime.py
index 77f5fe64..9fe64584 100644
--- a/backend/apps/outputs/runtime.py
+++ b/backend/apps/outputs/runtime.py
@@ -10,7 +10,7 @@ from dataclasses import dataclass
 from typing import Callable, Optional
 
 
-def _resolve_bash() -> str:
+def p_resolve_bash() -> str:
     # Windows: Python's subprocess uses Windows-style PATH resolution and doesn't follow Git Bash's Unix-style entries like /mingw64/bin/..., so a bare "bash" call hits [WinError 2]. shutil.which goes through Windows PATHEXT lookup; fall back to the conventional Git for Windows install path so users without bash in their Windows PATH still work. POSIX: just return "bash" since the kernel finds it via PATH like any other exec.
     found = shutil.which("bash")
     if found:
@@ -25,29 +25,29 @@ def _resolve_bash() -> str:
                 return candidate
     return "bash"
 
-from .runtime_proc import (
-    _ERROR_PATTERNS,
-    _FRONTEND_BIND_POLL_INTERVAL,
-    _FRONTEND_BIND_TIMEOUT_SECONDS,
-    _LOG_BUFFER_LINES,
-    _MAX_IDLE_RUNTIMES,
-    _RECENT_ERRORS_MAX,
-    _TERMINATE_GRACE_SECONDS,
-    _background_priority_kwargs,
-    _find_free_port,
-    _is_new_mode,
-    _is_port_free,
-    _kill_descendant_tree,
-    _read_env_value,
-    _resume_process_tree,
-    _suspend_process_tree,
-    _write_env_value,
+from backend.apps.outputs.runtime_proc import (
+    ERROR_PATTERNS,
+    FRONTEND_BIND_POLL_INTERVAL,
+    FRONTEND_BIND_TIMEOUT_SECONDS,
+    LOG_BUFFER_LINES,
+    MAX_IDLE_RUNTIMES,
+    RECENT_ERRORS_MAX,
+    TERMINATE_GRACE_SECONDS,
+    background_priority_kwargs,
+    find_free_port,
+    is_new_mode,
+    is_port_free,
+    kill_descendant_tree,
+    read_env_value,
+    resume_process_tree,
+    suspend_process_tree,
+    write_env_value,
 )
 
 logger = logging.getLogger(__name__)
 
-# Module-level lock so only ONE vite optimizeDeps runs at a time; must be acquired before manager._lock to avoid deadlock with manager.attach.
-_vite_boot_lock = asyncio.Lock()
+# Module-level lock so only ONE vite optimizeDeps runs at a time; must be acquired before manager.p_lock to avoid deadlock with manager.attach.
+P_VITE_BOOT_LOCK = asyncio.Lock()
 
 
 @dataclass
@@ -82,28 +82,28 @@ class AppRuntime:
         self.frontend_port: Optional[int] = None
         # New-mode only: flips True once something is actually listening
         # on frontend_port (we kick off a background poll task in
-        # _start_new_mode). frontend_url returns null until this flips,
+        # p_start_new_mode). frontend_url returns null until this flips,
         # so the preview pane doesn't try to navigate to an unbound port
         # and show a "Site can't be reached" error mid-npm-install.
-        self._frontend_ready: bool = False
+        self.p_frontend_ready: bool = False
         # True while the process tree is SIGSTOP'd in the idle pool. A frozen
         # vite still holds its port but can't answer it, so frontend_url must
         # stay null while suspended (else the webview loads a dead port = the
         # ERR_FAILED on fast app-switching).
-        self._suspended: bool = False
+        self.p_suspended: bool = False
         self.process: Optional[asyncio.subprocess.Process] = None
-        self.log_buffer: deque[LogLine] = deque(maxlen=_LOG_BUFFER_LINES)
-        self._subscribers: set[LogSubscriber] = set()
+        self.log_buffer: deque[LogLine] = deque(maxlen=LOG_BUFFER_LINES)
+        self.p_subscribers: set[LogSubscriber] = set()
         # Recent build/runtime errors scraped from stderr; drained by
         # the agent's post-tool hook after Write/Edit so the agent sees
         # vite/babel/uvicorn errors in its next turn and can self-fix
         # instead of leaving the user with a red iframe overlay.
-        self.recent_errors: deque[str] = deque(maxlen=_RECENT_ERRORS_MAX)
-        self._stdout_task: Optional[asyncio.Task] = None
-        self._stderr_task: Optional[asyncio.Task] = None
-        self._wait_task: Optional[asyncio.Task] = None
-        self._frontend_ready_task: Optional[asyncio.Task] = None
-        self._lock = asyncio.Lock()
+        self.recent_errors: deque[str] = deque(maxlen=RECENT_ERRORS_MAX)
+        self.p_stdout_task: Optional[asyncio.Task] = None
+        self.p_stderr_task: Optional[asyncio.Task] = None
+        self.p_wait_task: Optional[asyncio.Task] = None
+        self.p_frontend_ready_task: Optional[asyncio.Task] = None
+        self.p_lock = asyncio.Lock()
 
     def drain_errors(self) -> list[str]:
         """Pop and return all accumulated error lines. Used by the
@@ -123,21 +123,21 @@ class AppRuntime:
 
     @property
     def is_new_mode(self) -> bool:
-        return _is_new_mode(self.workspace_path)
+        return is_new_mode(self.workspace_path)
 
     @property
     def frontend_url(self) -> Optional[str]:
-        # Gated on `_frontend_ready` (set by the background bind-poll
-        # task in _start_new_mode) so the preview pane only switches
+        # Gated on `p_frontend_ready` (set by the background bind-poll
+        # task in p_start_new_mode) so the preview pane only switches
         # over once Vite is actually accepting connections. Without
         # this, the editor flashes a "Site can't be reached" error
         # while `npm install` is running.
         # Also gated on `running`: a vite that crashed or got orphaned still
-        # has _frontend_ready=True, and handing the webview that dead port is
+        # has p_frontend_ready=True, and handing the webview that dead port is
         # the ERR_FAILED you see on reopen. No live process, no URL.
-        # And gated on `not _suspended`: a SIGSTOP'd idle runtime is "running"
+        # And gated on `not p_suspended`: a SIGSTOP'd idle runtime is "running"
         # (returncode is None) but frozen, so its port won't answer.
-        if self.frontend_port and self._frontend_ready and self.running and not self._suspended:
+        if self.frontend_port and self.p_frontend_ready and self.running and not self.p_suspended:
             return f"http://127.0.0.1:{self.frontend_port}/"
         return None
 
@@ -160,11 +160,11 @@ class AppRuntime:
         exists so the Terminal pane can host `[FRONTEND]` lines.
 
         New-mode spawns are serialized through the module-level
-        `_vite_boot_lock` (see comment at the lock declaration) so a
+        `P_VITE_BOOT_LOCK` (see comment at the lock declaration) so a
         burst of "create 3 apps in 5 seconds" doesn't trigger 3 parallel
         MUI pre-bundle runs each pegging a core.
         """
-        async with self._lock:
+        async with self.p_lock:
             if self.running:
                 return True
 
@@ -175,45 +175,45 @@ class AppRuntime:
                 # vite emits "frontend ready" (or its 180s timeout
                 # fires), which is the moment the next workspace can
                 # start its own vite without competing for the same
-                # CPU. See `_await_frontend_bind` for the release.
-                await _vite_boot_lock.acquire()
+                # CPU. See `p_await_frontend_bind` for the release.
+                await P_VITE_BOOT_LOCK.acquire()
                 try:
-                    ok = await self._start_new_mode()
+                    ok = await self.p_start_new_mode()
                     if not ok:
                         # Spawn failed before the bind-poll task was
                         # created; release synchronously so we don't
                         # wedge the next workspace.
-                        _vite_boot_lock.release()
+                        P_VITE_BOOT_LOCK.release()
                     return ok
                 except Exception:
-                    _vite_boot_lock.release()
+                    P_VITE_BOOT_LOCK.release()
                     raise
-            return await self._start_old_mode()
+            return await self.p_start_old_mode()
 
-    async def _start_new_mode(self) -> bool:
+    async def p_start_new_mode(self) -> bool:
         env_path = os.path.join(self.workspace_path, ".env")
-        fp_raw = _read_env_value(env_path, "FRONTEND_PORT")
-        bp_raw = _read_env_value(env_path, "BACKEND_PORT")
+        fp_raw = read_env_value(env_path, "FRONTEND_PORT")
+        bp_raw = read_env_value(env_path, "BACKEND_PORT")
         # FRONTEND_PORT is allocated by seed_workspace; should always be
         # a number. If missing, fall back to a fresh allocation (rare
         # edge case: workspace seeded by an older OpenSwarm).
         try:
-            self.frontend_port = int(fp_raw) if fp_raw else _find_free_port()
+            self.frontend_port = int(fp_raw) if fp_raw else find_free_port()
         except ValueError:
-            self.frontend_port = _find_free_port()
+            self.frontend_port = find_free_port()
         # Port-collision safety net: if a ghost subprocess from a prior
         # OpenSwarm run is still bound to the persisted port (force-quit,
         # crash, OS killed the parent before stop_all could reap), Vite
         # would EADDRINUSE silently. Re-probe and reallocate, then rewrite
         # .env so the bash run.sh subprocess reads the new port.
-        if self.frontend_port and not _is_port_free(self.frontend_port):
-            new_port = _find_free_port()
-            self._broadcast(LogLine(
+        if self.frontend_port and not is_port_free(self.frontend_port):
+            new_port = find_free_port()
+            self.p_broadcast(LogLine(
                 "runtime",
                 f"[runtime] persisted FRONTEND_PORT {self.frontend_port} is in use; reallocating to {new_port}",
             ))
             self.frontend_port = new_port
-            _write_env_value(env_path, "FRONTEND_PORT", str(new_port))
+            write_env_value(env_path, "FRONTEND_PORT", str(new_port))
         # BACKEND_PORT may be the literal string "NONE" (frontend-only
         # app; the common case) or a number once `backend_init.sh` has
         # run. Only populate self.port when there's a real backend.
@@ -224,18 +224,18 @@ class AppRuntime:
                 self.port = None
             # Same collision check for the backend port; a leaked uvicorn
             # from a prior session would otherwise block the new spawn.
-            if self.port and not _is_port_free(self.port):
-                new_port = _find_free_port()
-                self._broadcast(LogLine(
+            if self.port and not is_port_free(self.port):
+                new_port = find_free_port()
+                self.p_broadcast(LogLine(
                     "runtime",
                     f"[runtime] persisted BACKEND_PORT {self.port} is in use; reallocating to {new_port}",
                 ))
                 self.port = new_port
-                _write_env_value(env_path, "BACKEND_PORT", str(new_port))
+                write_env_value(env_path, "BACKEND_PORT", str(new_port))
         else:
             self.port = None
 
-        env = self._spawn_env_base()
+        env = self.p_spawn_env_base()
         # bash run.sh reads .env itself; we don't need to set
         # FRONTEND_PORT / BACKEND_PORT here. We DO export the install
         # paths so the template's `backend/run.sh` can find our
@@ -244,47 +244,47 @@ class AppRuntime:
         # the more reliable read site for subshells.)
         # NOTE: keep these in sync with seed_webapp_template_workspace.
         from backend.apps.outputs.view_builder_templates import (
-            _DEBUGGER_PATH,
-            _TEMPLATE_BACKEND_PATH,
+            DEBUGGER_PATH,
+            TEMPLATE_BACKEND_PATH,
         )
-        env["OPENSWARM_DEBUGGER_PATH"] = _DEBUGGER_PATH
-        env["OPENSWARM_TEMPLATE_BACKEND_PATH"] = _TEMPLATE_BACKEND_PATH
+        env["OPENSWARM_DEBUGGER_PATH"] = DEBUGGER_PATH
+        env["OPENSWARM_TEMPLATE_BACKEND_PATH"] = TEMPLATE_BACKEND_PATH
 
         try:
             self.process = await asyncio.create_subprocess_exec(
-                _resolve_bash(), "run.sh",
+                p_resolve_bash(), "run.sh",
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
                 cwd=self.workspace_path,
                 env=env,
-                **_background_priority_kwargs(),
+                **background_priority_kwargs(),
             )
         except Exception as e:
             logger.exception("failed to start new-mode runtime for %s", self.workspace_id)
-            self._broadcast(LogLine("runtime", f"[runtime] failed to start: {e}"))
+            self.p_broadcast(LogLine("runtime", f"[runtime] failed to start: {e}"))
             self.frontend_port = None
             self.port = None
             self.process = None
             return False
         backend_note = f" + backend on {self.port}" if self.port else ""
-        self._broadcast(LogLine("runtime", f"[runtime] bash run.sh started; frontend on {self.frontend_port}{backend_note} (pid {self.process.pid})"))
-        self._stdout_task = asyncio.create_task(self._pipe_stream(self.process.stdout, "stdout"))
-        self._stderr_task = asyncio.create_task(self._pipe_stream(self.process.stderr, "stderr"))
-        self._wait_task = asyncio.create_task(self._await_exit())
+        self.p_broadcast(LogLine("runtime", f"[runtime] bash run.sh started; frontend on {self.frontend_port}{backend_note} (pid {self.process.pid})"))
+        self.p_stdout_task = asyncio.create_task(self.p_pipe_stream(self.process.stdout, "stdout"))
+        self.p = asyncio.create_task(self.p_pipe_stream(self.process.stderr, "stderr"))
+        self.p_wait_task = asyncio.create_task(self.p_await_exit())
         # Kick off the port-bind poller so frontend_url flips on once
         # Vite is actually accepting connections.
-        self._frontend_ready = False
-        self._frontend_ready_task = asyncio.create_task(self._await_frontend_bind())
+        self.p_frontend_ready = False
+        self.p_frontend_ready_task = asyncio.create_task(self.p_await_frontend_bind())
         return True
 
-    async def _await_frontend_bind(self) -> None:
-        """Poll `frontend_port` every _FRONTEND_BIND_POLL_INTERVAL until
+    async def p_await_frontend_bind(self) -> None:
+        """Poll `frontend_port` every FRONTEND_BIND_POLL_INTERVAL until
         something binds (Vite dev server) or we hit the timeout. Emits a
         `[runtime]` log line on success/failure so the Terminal pane
-        shows the transition; flips `_frontend_ready` which the
+        shows the transition; flips `p_frontend_ready` which the
         `frontend_url` property reads.
 
-        Also responsible for releasing the module-level `_vite_boot_lock`
+        Also responsible for releasing the module-level `P_VITE_BOOT_LOCK`
        ; every exit path (success, process death, hard timeout) MUST
         release exactly once so the next queued workspace can start its
         own vite spawn. A try/finally on the lock guarantees that even
@@ -293,13 +293,13 @@ class AppRuntime:
         # end doesn't double-release if a success path beat it.
         lock_released = False
 
-        def _release_boot_lock() -> None:
+        def release_boot_lock() -> None:
             nonlocal lock_released
             if lock_released:
                 return
             lock_released = True
             try:
-                _vite_boot_lock.release()
+                P_VITE_BOOT_LOCK.release()
             except RuntimeError:
                 # Lock already released (e.g. start() failure path
                 # released synchronously before spawning the poll task).
@@ -309,7 +309,7 @@ class AppRuntime:
             if not self.frontend_port:
                 return
             port = self.frontend_port
-            deadline = asyncio.get_event_loop().time() + _FRONTEND_BIND_TIMEOUT_SECONDS
+            deadline = asyncio.get_event_loop().time() + FRONTEND_BIND_TIMEOUT_SECONDS
             while asyncio.get_event_loop().time() < deadline:
                 # Stop polling if the process died; pointless to keep
                 # checking a port nothing will bind.
@@ -326,8 +326,8 @@ class AppRuntime:
                         await writer.wait_closed()
                     except Exception:
                         pass
-                    self._frontend_ready = True
-                    self._broadcast(LogLine(
+                    self.p_frontend_ready = True
+                    self.p_broadcast(LogLine(
                         "runtime",
                         f"[runtime] frontend ready at http://127.0.0.1:{port}/",
                     ))
@@ -335,32 +335,32 @@ class AppRuntime:
                     # ready; the next queued workspace can start its
                     # own bundle now even though we'll keep streaming
                     # logs for this one.
-                    _release_boot_lock()
+                    release_boot_lock()
                     return
                 except (OSError, asyncio.TimeoutError):
                     pass
-                await asyncio.sleep(_FRONTEND_BIND_POLL_INTERVAL)
+                await asyncio.sleep(FRONTEND_BIND_POLL_INTERVAL)
             # Timed out; keep the runtime up (Terminal might show useful
             # errors) but surface why the preview never appeared.
-            self._broadcast(LogLine(
+            self.p_broadcast(LogLine(
                 "runtime",
                 f"[runtime] frontend did NOT bind on port {port} after "
-                f"{_FRONTEND_BIND_TIMEOUT_SECONDS}s; check the Terminal "
+                f"{FRONTEND_BIND_TIMEOUT_SECONDS}s; check the Terminal "
                 f"for npm/vite errors.",
             ))
         finally:
             # Catches process-death return, timeout fall-through, and
-            # any exception in the poll body. _release_boot_lock is
+            # any exception in the poll body. release_boot_lock is
             # idempotent so this is safe even after the success path
             # already released.
-            _release_boot_lock()
+            release_boot_lock()
 
-    async def _start_old_mode(self) -> bool:
+    async def p_start_old_mode(self) -> bool:
         if not self.has_backend_file:
             self.port = None
             return False
-        self.port = _find_free_port()
-        env = self._spawn_env_base()
+        self.port = find_free_port()
+        env = self.p_spawn_env_base()
         env["PORT"] = str(self.port)
         env["BACKEND_PORT"] = str(self.port)  # alias; both common names work
         try:
@@ -373,21 +373,21 @@ class AppRuntime:
                 stderr=asyncio.subprocess.PIPE,
                 cwd=self.workspace_path,
                 env=env,
-                **_background_priority_kwargs(),
+                **background_priority_kwargs(),
             )
         except Exception as e:
             logger.exception("failed to start backend for %s", self.workspace_id)
-            self._broadcast(LogLine("runtime", f"[runtime] failed to start: {e}"))
+            self.p_broadcast(LogLine("runtime", f"[runtime] failed to start: {e}"))
             self.port = None
             self.process = None
             return False
-        self._broadcast(LogLine("runtime", f"[runtime] backend started on port {self.port} (pid {self.process.pid})"))
-        self._stdout_task = asyncio.create_task(self._pipe_stream(self.process.stdout, "stdout"))
-        self._stderr_task = asyncio.create_task(self._pipe_stream(self.process.stderr, "stderr"))
-        self._wait_task = asyncio.create_task(self._await_exit())
+        self.p_broadcast(LogLine("runtime", f"[runtime] backend started on port {self.port} (pid {self.process.pid})"))
+        self.p_stdout_task = asyncio.create_task(self.p_pipe_stream(self.process.stdout, "stdout"))
+        self.p = asyncio.create_task(self.p_pipe_stream(self.process.stderr, "stderr"))
+        self.p_wait_task = asyncio.create_task(self.p_await_exit())
         return True
 
-    def _spawn_env_base(self) -> dict[str, str]:
+    def p_spawn_env_base(self) -> dict[str, str]:
         """Inherited env minus the install token. Backend.py can hit our
         REST API back via its own creds if it really needs to, but it
         shouldn't inherit the host process's token by default."""
@@ -402,33 +402,33 @@ class AppRuntime:
         return env
 
     async def stop(self) -> None:
-        async with self._lock:
+        async with self.p_lock:
             if not self.process or self.process.returncode is not None:
                 # Still cancel the bind poller in case stop() races a
                 # never-launched runtime; defensive no-op otherwise.
-                if self._frontend_ready_task and not self._frontend_ready_task.done():
-                    self._frontend_ready_task.cancel()
+                if self.p_frontend_ready_task and not self.p_frontend_ready_task.done():
+                    self.p_frontend_ready_task.cancel()
                 return
             try:
                 # Walk the descendant tree first so vite/uvicorn grandchildren
                 # die before bash exits and orphans them to PID 1. The webapp
                 # template's run.sh only traps EXIT, not TERM, so a flat
                 # SIGTERM to bash kills bash silently and leaves vite alive.
-                _kill_descendant_tree(self.process.pid, "TERM")
+                kill_descendant_tree(self.process.pid, "TERM")
                 self.process.terminate()
                 try:
-                    await asyncio.wait_for(self.process.wait(), timeout=_TERMINATE_GRACE_SECONDS)
+                    await asyncio.wait_for(self.process.wait(), timeout=TERMINATE_GRACE_SECONDS)
                 except asyncio.TimeoutError:
-                    _kill_descendant_tree(self.process.pid, "KILL")
+                    kill_descendant_tree(self.process.pid, "KILL")
                     self.process.kill()
                     await self.process.wait()
             except ProcessLookupError:
                 pass
             # Cancel the bind poller so it stops scanning a port that's
             # gone away, and reset the readiness flag.
-            if self._frontend_ready_task and not self._frontend_ready_task.done():
-                self._frontend_ready_task.cancel()
-            self._frontend_ready = False
+            if self.p_frontend_ready_task and not self.p_frontend_ready_task.done():
+                self.p_frontend_ready_task.cancel()
+            self.p_frontend_ready = False
 
     async def restart(self) -> bool:
         await self.stop()
@@ -438,36 +438,36 @@ class AppRuntime:
         """Register a log subscriber. Immediately replays the ring buffer
         so a Terminal pane that opens mid-session shows context. Returns
         an unsubscribe function."""
-        self._subscribers.add(cb)
+        self.p_subscribers.add(cb)
         for line in list(self.log_buffer):
             try:
                 cb(line)
             except Exception:
                 pass
 
-        def _unsub() -> None:
-            self._subscribers.discard(cb)
+        def unsub() -> None:
+            self.p_subscribers.discard(cb)
 
-        return _unsub
+        return unsub
 
-    def _broadcast(self, line: LogLine) -> None:
+    def p_broadcast(self, line: LogLine) -> None:
         self.log_buffer.append(line)
         # Snapshot subscribers; they can self-remove during dispatch.
-        for cb in list(self._subscribers):
+        for cb in list(self.p_subscribers):
             try:
                 cb(line)
             except Exception:
                 pass
 
-    def _maybe_capture_error(self, text: str) -> None:
+    def p_maybe_capture_error(self, text: str) -> None:
         """If a stderr/stdout line matches a known build-error pattern,
         record it for the next agent-tool drain. Tests every line , 
         cheap (single regex search) and only the matching ones land in
         the buffer."""
-        if _ERROR_PATTERNS.search(text):
+        if ERROR_PATTERNS.search(text):
             self.recent_errors.append(text.rstrip())
 
-    async def _pipe_stream(self, stream: Optional[asyncio.StreamReader], name: str) -> None:
+    async def p_pipe_stream(self, stream: Optional[asyncio.StreamReader], name: str) -> None:
         if stream is None:
             return
         try:
@@ -477,21 +477,21 @@ class AppRuntime:
                     break
                 text = raw.decode(errors="replace").rstrip("\r\n")
                 if text:
-                    self._broadcast(LogLine(name, text))
+                    self.p_broadcast(LogLine(name, text))
                     if name == "stderr" or name == "stdout":
-                        self._maybe_capture_error(text)
+                        self.p_maybe_capture_error(text)
         except Exception:
             logger.exception("log pipe error (%s) for %s", name, self.workspace_id)
 
-    async def _await_exit(self) -> None:
+    async def p_await_exit(self) -> None:
         if not self.process:
             return
         rc = await self.process.wait()
         # Unclean death (vite crash, OOM, orphaned parent) must drop readiness;
         # otherwise frontend_url keeps advertising a dead port and the preview
         # navigates into ERR_FAILED. stop() already does this for clean stops.
-        self._frontend_ready = False
-        self._broadcast(LogLine("runtime", f"[runtime] backend exited with code {rc}"))
+        self.p_frontend_ready = False
+        self.p_broadcast(LogLine("runtime", f"[runtime] backend exited with code {rc}"))
 
 
 class AppRuntimeManager:
@@ -502,30 +502,32 @@ class AppRuntimeManager:
     spawns; final detach moves the runtime into an LRU idle pool
     instead of stopping it immediately; so re-clicking a recent App
     is instant. The oldest runtime gets reaped once the pool exceeds
-    _MAX_IDLE_RUNTIMES."""
+    MAX_IDLE_RUNTIMES."""
 
     def __init__(self) -> None:
+        # NOTE: None of the attributes ever got modified externally, so they're all private.
         # workspace_id → AppRuntime, currently has >=1 subscriber.
         self.runtimes: dict[str, AppRuntime] = {}
-        self._attached: dict[str, int] = {}
+        self.p_attached: dict[str, int] = {}
         # workspace_id → AppRuntime with no subscribers but still
         # alive. OrderedDict gives O(1) move_to_end + popitem(last=False)
         # for LRU semantics.
-        self._idle_lru: "OrderedDict[str, AppRuntime]" = OrderedDict()
-        self._lock = asyncio.Lock()
+        self.p_idle_lru: "OrderedDict[str, AppRuntime]" = OrderedDict()
+        self.p_lock = asyncio.Lock()
 
+    # Public - used by outputs.py
     async def attach(self, workspace_id: str, workspace_path: str) -> AppRuntime:
         revived = False
         # Defined here so every code path below leaves it bound; the
         # revive-idle branch used to skip the assignment, leaving the
         # post-lock `if dead is not None:` check throwing UnboundLocalError.
         dead: Optional[AppRuntime] = None
-        async with self._lock:
+        async with self.p_lock:
             rt = self.runtimes.get(workspace_id)
             if rt is None:
                 # Maybe the runtime is sitting idle in the LRU; revive
                 # it without paying the spawn cost again.
-                idle_rt = self._idle_lru.pop(workspace_id, None)
+                idle_rt = self.p_idle_lru.pop(workspace_id, None)
                 if idle_rt is not None and idle_rt.running:
                     rt = idle_rt
                     rt.workspace_path = workspace_path
@@ -533,8 +535,8 @@ class AppRuntimeManager:
                     revived = True
                     # SIGCONT the process tree if A2 had it paused while
                     # idle. Pair with the SIGSTOP in detach() below.
-                    _resume_process_tree(rt.process)
-                    rt._suspended = False
+                    resume_process_tree(rt.process)
+                    rt.p_suspended = False
                 else:
                     if idle_rt is not None:
                         # Stale idle entry; process died while idling.
@@ -549,7 +551,7 @@ class AppRuntimeManager:
                 # folder), trust the latest caller; they have the
                 # current truth.
                 rt.workspace_path = workspace_path
-            self._attached[workspace_id] = self._attached.get(workspace_id, 0) + 1
+            self.p_attached[workspace_id] = self.p_attached.get(workspace_id, 0) + 1
         if not revived and not rt.running:
             await rt.start()
         # Stop any dead idle runtime outside the lock to avoid blocking.
@@ -560,15 +562,16 @@ class AppRuntimeManager:
                 logger.exception("failed to reap dead idle runtime %s", workspace_id)
         return rt
 
+    # Public - used by outputs.py
     async def detach(self, workspace_id: str) -> None:
         to_idle: Optional[AppRuntime] = None
         to_reap: list[AppRuntime] = []
-        async with self._lock:
-            count = self._attached.get(workspace_id, 0) - 1
+        async with self.p_lock:
+            count = self.p_attached.get(workspace_id, 0) - 1
             if count > 0:
-                self._attached[workspace_id] = count
+                self.p_attached[workspace_id] = count
                 return
-            self._attached.pop(workspace_id, None)
+            self.p_attached.pop(workspace_id, None)
             rt = self.runtimes.pop(workspace_id, None)
             if rt is None:
                 return
@@ -579,21 +582,21 @@ class AppRuntimeManager:
             if not rt.running:
                 to_reap.append(rt)
             else:
-                self._idle_lru[workspace_id] = rt
-                self._idle_lru.move_to_end(workspace_id)
-                _suspend_process_tree(rt.process)
-                rt._suspended = True
-                while len(self._idle_lru) > _MAX_IDLE_RUNTIMES:
-                    _, old_rt = self._idle_lru.popitem(last=False)
+                self.p_idle_lru[workspace_id] = rt
+                self.p_idle_lru.move_to_end(workspace_id)
+                suspend_process_tree(rt.process)
+                rt.p_suspended = True
+                while len(self.p_idle_lru) > MAX_IDLE_RUNTIMES:
+                    _, old_rt = self.p_idle_lru.popitem(last=False)
                     # Reaping a stopped process: SIGCONT first so the
                     # SIGTERM in stop() can be delivered cleanly (a
                     # SIGSTOP'd process can't run its own shutdown).
-                    _resume_process_tree(old_rt.process)
+                    resume_process_tree(old_rt.process)
                     to_reap.append(old_rt)
             to_idle = rt if rt.running else None
 
         # Stop any reaped runtimes OUTSIDE the lock. stop() is async and
-        # can take up to _TERMINATE_GRACE_SECONDS; holding the lock for
+        # can take up to TERMINATE_GRACE_SECONDS; holding the lock for
         # it would block every other attach/detach.
         for old in to_reap:
             try:
@@ -601,8 +604,9 @@ class AppRuntimeManager:
             except Exception:
                 logger.exception("failed to reap idle runtime %s", workspace_id)
         if to_idle is not None:
-            logger.debug("workspace %s idled (LRU size now %d)", workspace_id, len(self._idle_lru))
+            logger.debug("workspace %s idled (LRU size now %d)", workspace_id, len(self.p_idle_lru))
 
+    # Public - used by html_inject.py, main.py, outputs.py
     def get(self, workspace_id: str) -> Optional[AppRuntime]:
         # Active subscribers see the live runtime; idle-pool members
         # are also accessible so a status probe between detach and
@@ -610,8 +614,9 @@ class AppRuntimeManager:
         rt = self.runtimes.get(workspace_id)
         if rt is not None:
             return rt
-        return self._idle_lru.get(workspace_id)
+        return self.p_idle_lru.get(workspace_id)
 
+    # Public - used by agent_manager.py
     def drain_errors_for_path(self, file_path: str) -> list[str]:
         """If `file_path` falls under one of the live workspace
         runtimes' workspace_path, drain that workspace's recent
@@ -629,7 +634,7 @@ class AppRuntimeManager:
         # navigated away from the workspace mid-build, but the agent
         # could still be editing files; the LRU keeps the runtime alive
         # for ~3 idle slots.
-        for rt in (*self.runtimes.values(), *self._idle_lru.values()):
+        for rt in (*self.runtimes.values(), *self.p_idle_lru.values()):
             try:
                 ws_root = os.path.abspath(rt.workspace_path)
             except Exception:
@@ -638,8 +643,9 @@ class AppRuntimeManager:
                 return rt.drain_errors()
         return []
 
+    # Public - used by outputs.py
     async def restart(self, workspace_id: str, workspace_path: Optional[str] = None) -> Optional[AppRuntime]:
-        rt = self.runtimes.get(workspace_id) or self._idle_lru.get(workspace_id)
+        rt = self.runtimes.get(workspace_id) or self.p_idle_lru.get(workspace_id)
         if rt is None:
             return None
         if workspace_path:
@@ -647,6 +653,7 @@ class AppRuntimeManager:
         await rt.restart()
         return rt
 
+    # Public - used by outputs.py
     async def stop_all(self) -> int:
         """Terminate every active + idle workspace subprocess. Called on
         FastAPI lifespan shutdown AND from Electron's pre-quit POST. Without
@@ -657,16 +664,16 @@ class AppRuntimeManager:
         so they can run their own shutdown. Parallel via gather; with the
         per-runtime 3s SIGTERM grace, worst case is one ~3s wait rather than
         N*3s. Idempotent; safe to invoke from multiple shutdown paths."""
-        async with self._lock:
+        async with self.p_lock:
             victims: list[AppRuntime] = []
             for rt in list(self.runtimes.values()):
                 victims.append(rt)
-            for rt in list(self._idle_lru.values()):
-                _resume_process_tree(rt.process)
+            for rt in list(self.p_idle_lru.values()):
+                resume_process_tree(rt.process)
                 victims.append(rt)
             self.runtimes.clear()
-            self._idle_lru.clear()
-            self._attached.clear()
+            self.p_idle_lru.clear()
+            self.p_attached.clear()
         if not victims:
             return 0
         await asyncio.gather(
@@ -676,4 +683,4 @@ class AppRuntimeManager:
         return len(victims)
 
 
-manager = AppRuntimeManager()
+RUNTIME_MANAGER = AppRuntimeManager()
diff --git a/backend/apps/outputs/runtime_proc.py b/backend/apps/outputs/runtime_proc.py
index 9d699444..8eef8104 100644
--- a/backend/apps/outputs/runtime_proc.py
+++ b/backend/apps/outputs/runtime_proc.py
@@ -14,24 +14,24 @@ from typing import Optional
 logger = logging.getLogger(__name__)
 
 # SIGTERM grace; well-behaved servers shut down under a second so 3s is enough.
-_TERMINATE_GRACE_SECONDS = 3
+TERMINATE_GRACE_SECONDS = 3
 
 # 180s covers npm install (60-90s on typical hardware) plus the Vite bind.
-_FRONTEND_BIND_TIMEOUT_SECONDS = 180
+FRONTEND_BIND_TIMEOUT_SECONDS = 180
 # 80ms probe: dropping from 500ms was pure user-visible preview latency win; cheap on localhost.
-_FRONTEND_BIND_POLL_INTERVAL = 0.08
+FRONTEND_BIND_POLL_INTERVAL = 0.08
 
 # 2000 lines per runtime; lets a Terminal tab opened mid-session replay context. ~few hundred KB at worst.
-_LOG_BUFFER_LINES = 2000
+LOG_BUFFER_LINES = 2000
 
 # Idle runtimes kept in LRU; trades memory for instant switch-back, beyond 1 because typical users ping-pong 2-3 apps.
-_MAX_IDLE_RUNTIMES = 3
+MAX_IDLE_RUNTIMES = 3
 
 # Cap on recent error lines the agent gets; 50 is enough for babel error + stack + a few warnings.
-_RECENT_ERRORS_MAX = 50
+RECENT_ERRORS_MAX = 50
 
 # Narrow regex for build errors (vite, babel, tsc, uvicorn); keeps routine logs out of agent context.
-_ERROR_PATTERNS = re.compile(
+ERROR_PATTERNS = re.compile(
     r"(?:"
     r"\[plugin:[^\]]+\]|"      # vite plugin errors
     r"SyntaxError|"            # node / babel
@@ -49,7 +49,7 @@ _ERROR_PATTERNS = re.compile(
 )
 
 
-def _suspend_process_tree(proc) -> None:
+def suspend_process_tree(proc) -> None:
     """Send SIGSTOP to a workspace's subprocess so it consumes 0% CPU
     while sitting in the LRU idle pool. The signal is delivered to the
     PROCESS GROUP (negative PID) when the child is a session leader,
@@ -71,7 +71,7 @@ def _suspend_process_tree(proc) -> None:
         pass
 
 
-def _resume_process_tree(proc) -> None:
+def resume_process_tree(proc) -> None:
     """SIGCONT a previously-suspended workspace process. Pair with
     _suspend_process_tree. Microsecond cost; idempotent if the process
     was never paused."""
@@ -85,7 +85,7 @@ def _resume_process_tree(proc) -> None:
         pass
 
 
-def _background_priority_kwargs() -> dict:
+def background_priority_kwargs() -> dict:
     """Return the kwargs that lower the spawned subprocess's OS priority
     to a "background" level. On POSIX this is `preexec_fn=os.nice(10)`,
     which sets the child's nice to +10 BEFORE exec (so the renice covers
@@ -112,7 +112,7 @@ def _background_priority_kwargs() -> dict:
     return {"preexec_fn": lambda: os.nice(10)}
 
 
-def _find_free_port() -> int:
+def find_free_port() -> int:
     """Ask the kernel for an unused localhost port. There's a tiny race
     between this socket closing and the backend re-binding, but we hand
     each port to exactly one runtime so no caller competes for it, and
@@ -122,7 +122,7 @@ def _find_free_port() -> int:
         return s.getsockname()[1]
 
 
-def _kill_descendant_tree(pid: int, sig_name: str = "TERM") -> None:
+def kill_descendant_tree(pid: int, sig_name: str = "TERM") -> None:
     """Recursively signal every descendant of `pid`, leaves-first. The
     webapp template's run.sh installs `trap cleanup EXIT` (no TERM), so a
     plain SIGTERM to the bash wrapper exits bash silently and leaves
@@ -154,7 +154,7 @@ def _kill_descendant_tree(pid: int, sig_name: str = "TERM") -> None:
     except Exception:
         children = []
     for child in children:
-        _kill_descendant_tree(child, sig_name)
+        kill_descendant_tree(child, sig_name)
     sig = getattr(signal, f"SIG{sig_name}", signal.SIGTERM)
     for child in children:
         try:
@@ -163,7 +163,7 @@ def _kill_descendant_tree(pid: int, sig_name: str = "TERM") -> None:
             pass
 
 
-def _is_port_free(port: int) -> bool:
+def is_port_free(port: int) -> bool:
     """True if nothing currently holds a TCP listener on 127.0.0.1:port.
     Cheap kernel-probe; resolves on bind success. Used as the cross-session
     safety net: if a prior OpenSwarm run left a ghost subprocess holding
@@ -177,7 +177,7 @@ def _is_port_free(port: int) -> bool:
         return False
 
 
-def _write_env_value(env_path: str, key: str, value: str) -> None:
+def write_env_value(env_path: str, key: str, value: str) -> None:
     """Update KEY=VALUE in an existing `.env`, preserving every other
     line. Creates the file if missing. Used when a persisted port collides
     with a ghost from a prior session and we have to reallocate before
@@ -210,7 +210,7 @@ def _write_env_value(env_path: str, key: str, value: str) -> None:
         logger.exception("failed writing %s=%s to %s", key, value, env_path)
 
 
-def _is_new_mode(workspace_path: str) -> bool:
+def is_new_mode(workspace_path: str) -> bool:
     """A workspace is "new-mode" (webapp-template scaffold) if it has a
     `run.sh` at its root. Old-mode workspaces are flat `index.html`-only
     apps that pre-date the template swap; they're served by OpenSwarm's
@@ -222,7 +222,7 @@ def _is_new_mode(workspace_path: str) -> bool:
     return os.path.isfile(os.path.join(workspace_path, "run.sh"))
 
 
-def _read_env_value(env_path: str, key: str) -> Optional[str]:
+def read_env_value(env_path: str, key: str) -> Optional[str]:
     """Parse one value out of a workspace's `.env` without the cost of a
     full subprocess-source. Strips quotes + trailing comments. Returns
     None if the file or key is missing."""
diff --git a/backend/apps/outputs/view_builder_templates.py b/backend/apps/outputs/view_builder_templates.py
index 3cc089f7..f1a16a33 100644
--- a/backend/apps/outputs/view_builder_templates.py
+++ b/backend/apps/outputs/view_builder_templates.py
@@ -13,7 +13,7 @@ import threading
 logger = logging.getLogger(__name__)
 
 
-def _resolve_npm() -> list[str] | None:
+def p_resolve_npm() -> list[str] | None:
     """Resolve an invokable npm command. Windows ships npm as npm.cmd (a
     batch shim), which Python's subprocess won't find via a bare "npm";
     and the packaged Electron build bundles only node.exe (no npm) but
@@ -42,7 +42,7 @@ def _resolve_npm() -> list[str] | None:
     return None
 
 
-def _resolve_python() -> str:
+def p_resolve_python() -> str:
     """The interpreter to build warm/workspace venvs with. sys.executable
     is the running backend's python (bundled standalone in the packaged
     build, system python in dev) and is always valid, sidestepping the
@@ -69,8 +69,8 @@ WEBAPP_TEMPLATE_DIR = os.path.join(os.path.dirname(__file__), "webapp_template")
 # Bundled default; used as the read-once fallback if the user-editable
 # copy at ~/.claude/skills/app_builder_skill.md has been removed despite
 # the built-in flag (defensive; shouldn't happen in normal use).
-with open(APP_BUILDER_SKILL_SOURCE_PATH, encoding="utf-8") as _f:
-    APP_BUILDER_SKILL_DEFAULT = _f.read()
+with open(APP_BUILDER_SKILL_SOURCE_PATH, encoding="utf-8") as f:
+    APP_BUILDER_SKILL_DEFAULT = f.read()
 
 
 def load_app_builder_skill() -> str:
@@ -88,13 +88,7 @@ def load_app_builder_skill() -> str:
             pass
     return APP_BUILDER_SKILL_DEFAULT
 
-
-# Backward-compat alias. Older callers import VIEW_BUILDER_SKILL directly , 
-# point them at the same content as the user-editable version so a "frozen
-# at import" stale copy can't drift from what the skills page shows.
-VIEW_BUILDER_SKILL = APP_BUILDER_SKILL_DEFAULT
-
-VIEW_TEMPLATE_INDEX = """\
+P_VIEW_TEMPLATE_INDEX = """\
 <!DOCTYPE html>
 <html lang="en">
 <head>
@@ -139,7 +133,7 @@ VIEW_TEMPLATE_INDEX = """\
 </html>
 """
 
-VIEW_TEMPLATE_SCHEMA = """\
+P_VIEW_TEMPLATE_SCHEMA = """\
 {
   "type": "object",
   "properties": {},
@@ -147,7 +141,7 @@ VIEW_TEMPLATE_SCHEMA = """\
 }
 """
 
-VIEW_TEMPLATE_META = """\
+P_VIEW_TEMPLATE_META = """\
 {
   "name": "",
   "description": ""
@@ -155,9 +149,9 @@ VIEW_TEMPLATE_META = """\
 """
 
 VIEW_TEMPLATE_FILES = {
-    "index.html": VIEW_TEMPLATE_INDEX,
-    "schema.json": VIEW_TEMPLATE_SCHEMA,
-    "meta.json": VIEW_TEMPLATE_META,
+    "index.html": P_VIEW_TEMPLATE_INDEX,
+    "schema.json": P_VIEW_TEMPLATE_SCHEMA,
+    "meta.json": P_VIEW_TEMPLATE_META,
 }
 
 
@@ -165,7 +159,7 @@ VIEW_TEMPLATE_FILES = {
 # webapp_template (new-mode) seed helpers
 # ---------------------------------------------------------------------------
 
-def _ignore_backend(src: str, names: list[str]) -> list[str]:
+def p_ignore_backend(src: str, names: list[str]) -> list[str]:
     """copytree filter; when copying the template root, drop only the
     top-level `backend/` directory. Subdirectories named `backend` deeper
     in the tree (none today, but defensively scoped) are unaffected."""
@@ -174,10 +168,10 @@ def _ignore_backend(src: str, names: list[str]) -> list[str]:
     return []
 
 
-_DEBUGGER_PATH = os.path.abspath(
+DEBUGGER_PATH = os.path.abspath(
     os.path.join(os.path.dirname(__file__), "..", "..", "..", "debugger")
 )
-_TEMPLATE_BACKEND_PATH = os.path.abspath(os.path.join(WEBAPP_TEMPLATE_DIR, "backend"))
+TEMPLATE_BACKEND_PATH = os.path.abspath(os.path.join(WEBAPP_TEMPLATE_DIR, "backend"))
 
 
 # ---------------------------------------------------------------------------
@@ -191,8 +185,8 @@ _TEMPLATE_BACKEND_PATH = os.path.abspath(os.path.join(WEBAPP_TEMPLATE_DIR, "back
 # until the user clears ~/.openswarm/cache.
 # ---------------------------------------------------------------------------
 
-_warm_cache_lock = threading.Lock()
-_warm_cache_thread: threading.Thread | None = None
+P_WARM_CACHE_LOCK = threading.Lock()
+P_WARM_CACHE_THREAD: threading.Thread | None = None
 
 
 # Pre-built node_modules archive bundled with packaged releases. Generated
@@ -202,25 +196,25 @@ _warm_cache_thread: threading.Thread | None = None
 # ~3 s vs ~22 s for the live install. Stale archives (package.json bumped
 # but archive not rebuilt) are silently ignored, so the live-install
 # fallback always wins on correctness.
-_BUNDLED_ARCHIVE_DIR = os.path.join(
+P_BUNDLED_ARCHIVE_DIR = os.path.join(
     os.path.dirname(__file__), "webapp_template_cache"
 )
 
 
-def _bundled_archive_path_for(digest: str) -> str:
+def p_bundled_archive_path_for(digest: str) -> str:
     """Sha-tagged archive path so a stale archive from a prior template
     version is automatically skipped instead of overwriting the cache with
     out-of-date modules."""
-    return os.path.join(_BUNDLED_ARCHIVE_DIR, f"node_modules.{digest}.tar.gz")
+    return os.path.join(P_BUNDLED_ARCHIVE_DIR, f"node_modules.{digest}.tar.gz")
 
 
-def _try_extract_bundled_archive(cache_dir: str, digest: str) -> bool:
+def p_try_extract_bundled_archive(cache_dir: str, digest: str) -> bool:
     """Unpack the sha-tagged bundled archive into `cache_dir` if one
     exists for the current template digest. Returns True on success,
     False to signal the caller should fall back to a live `npm install`.
     The archive is built from the same package.json + package-lock.json
     sha so the extracted tree is byte-equivalent to `npm ci`."""
-    archive_path = _bundled_archive_path_for(digest)
+    archive_path = p_bundled_archive_path_for(digest)
     if not os.path.exists(archive_path):
         return False
     try:
@@ -252,7 +246,7 @@ def _try_extract_bundled_archive(cache_dir: str, digest: str) -> bool:
         return False
 
 
-def _warm_cache_digest() -> str:
+def p_warm_cache_digest() -> str:
     """Sha of the template's frontend/package.json; used as the cache
     key + the bundled-archive filename suffix so a package.json bump
     invalidates both at once."""
@@ -264,34 +258,34 @@ def _warm_cache_digest() -> str:
         return "fallback"
 
 
-def _warm_cache_dir() -> str:
+def p_warm_cache_dir() -> str:
     """Path the warm node_modules lives under. Hashed by package.json so
     upgrades automatically force a re-populate."""
     base = os.environ.get("OPENSWARM_WEBAPP_CACHE_DIR") or os.path.expanduser(
         "~/.openswarm/cache/webapp_template_node_modules"
     )
-    return os.path.join(base, _warm_cache_digest())
+    return os.path.join(base, p_warm_cache_digest())
 
 
-def _ensure_warm_cache() -> str | None:
+def p_ensure_warm_cache() -> str | None:
     """Populate the warm-cache node_modules if missing. Returns the
     absolute path to the populated `node_modules` directory, or None on
     failure. Thread-safe; concurrent callers block on a single install
     instead of racing. Idempotent and fast after the first call."""
-    cache_dir = _warm_cache_dir()
+    cache_dir = p_warm_cache_dir()
     cache_modules = os.path.join(cache_dir, "node_modules")
 
     if os.path.isdir(cache_modules):
         return cache_modules
 
-    with _warm_cache_lock:
+    with P_WARM_CACHE_LOCK:
         if os.path.isdir(cache_modules):
             return cache_modules
         # Fast path: pre-built archive shipped inside the release. The
         # build script generates this so users hitting OpenSwarm for the
         # first time skip the ~22 s live `npm install`. Falls through on
         # any failure so dev installs (no archive) keep working.
-        if _try_extract_bundled_archive(cache_dir, _warm_cache_digest()):
+        if p_try_extract_bundled_archive(cache_dir, p_warm_cache_digest()):
             logger.info("webapp-template: warm cache ready from bundled archive")
             return cache_modules
         try:
@@ -304,7 +298,7 @@ def _ensure_warm_cache() -> str | None:
             tmpl_lock = os.path.join(WEBAPP_TEMPLATE_DIR, "frontend", "package-lock.json")
             shutil.copyfile(tmpl_pkg, os.path.join(cache_dir, "package.json"))
             base_flags = ["--prefer-offline", "--no-audit", "--no-fund", "--loglevel=error"]
-            npm = _resolve_npm()
+            npm = p_resolve_npm()
             if npm is None:
                 logger.info("webapp-template: no npm available; skipping warm cache (workspace will install on first run)")
                 return None
@@ -344,11 +338,11 @@ def _ensure_warm_cache() -> str | None:
             return None
 
 
-def _link_node_modules(workspace_dir: str) -> None:
+def p_link_node_modules(workspace_dir: str) -> None:
     """After copytree, point the workspace's frontend/node_modules at
     the warm-cache directory. Safe fallback; if the cache isn't ready,
     the workspace's run.sh will fall through to its own install path."""
-    cache_modules = _ensure_warm_cache()
+    cache_modules = p_ensure_warm_cache()
     if not cache_modules:
         return
     target = os.path.join(workspace_dir, "frontend", "node_modules")
@@ -392,10 +386,10 @@ def _link_node_modules(workspace_dir: str) -> None:
 # otherwise pays per workspace.
 # ---------------------------------------------------------------------------
 
-_warm_venv_lock = threading.Lock()
+P_WARM_VENV_LOCK = threading.Lock()
 
 
-def _warm_venv_dir() -> str:
+def p_warm_venv_dir() -> str:
     """Cache root for the shared backend venv, keyed by a sha of the
     template backend's pyproject.toml so a dep bump auto-invalidates."""
     pyproject = os.path.join(WEBAPP_TEMPLATE_DIR, "backend", "pyproject.toml")
@@ -410,18 +404,18 @@ def _warm_venv_dir() -> str:
     return os.path.join(base, digest)
 
 
-def _ensure_warm_python_venv() -> str | None:
+def p_ensure_warm_python_venv() -> str | None:
     """Populate the warm-cache backend venv if missing. Returns the
     absolute path to the populated `.venv` directory, or None on
     failure. Thread-safe and idempotent; fast return after first call."""
-    cache_dir = _warm_venv_dir()
+    cache_dir = p_warm_venv_dir()
     venv_dir = os.path.join(cache_dir, ".venv")
     sentinel = os.path.join(cache_dir, ".populated")
 
     if os.path.isfile(sentinel) and os.path.isdir(venv_dir):
         return venv_dir
 
-    with _warm_venv_lock:
+    with P_WARM_VENV_LOCK:
         if os.path.isfile(sentinel) and os.path.isdir(venv_dir):
             return venv_dir
         try:
@@ -433,7 +427,7 @@ def _ensure_warm_python_venv() -> str | None:
             # `python.exe`. On macOS/Linux the versioned candidates
             # match first so we don't accidentally pick a system
             # Python 2.x via the bare name.
-            py = _resolve_python()
+            py = p_resolve_python()
 
             # Wipe any half-populated venv from a previous crashed run.
             if os.path.isdir(venv_dir):
@@ -481,28 +475,28 @@ def warm_cache_in_background() -> None:
     node_modules cache and the backend-venv cache so the user's FIRST
     webapp-template seed doesn't pay the install costs. No-op (fast
     return) if both caches are already there or a thread is in flight."""
-    global _warm_cache_thread
-    if _warm_cache_thread is not None and _warm_cache_thread.is_alive():
+    global P_WARM_CACHE_THREAD
+    if P_WARM_CACHE_THREAD is not None and P_WARM_CACHE_THREAD.is_alive():
         return
-    node_done = os.path.isdir(os.path.join(_warm_cache_dir(), "node_modules"))
-    venv_done = os.path.isfile(os.path.join(_warm_venv_dir(), ".populated"))
+    node_done = os.path.isdir(os.path.join(p_warm_cache_dir(), "node_modules"))
+    venv_done = os.path.isfile(os.path.join(p_warm_venv_dir(), ".populated"))
     if node_done and venv_done:
         return
 
-    def _runner() -> None:
+    def runner() -> None:
         try:
-            _ensure_warm_cache()
+            p_ensure_warm_cache()
         except Exception:
             logger.exception("background warm node_modules crashed")
         try:
-            _ensure_warm_python_venv()
+            p_ensure_warm_python_venv()
         except Exception:
             logger.exception("background warm python venv crashed")
 
-    _warm_cache_thread = threading.Thread(
-        target=_runner, daemon=True, name="webapp-template-warm-cache"
+    P_WARM_CACHE_THREAD = threading.Thread(
+        target=runner, daemon=True, name="webapp-template-warm-cache"
     )
-    _warm_cache_thread.start()
+    P_WARM_CACHE_THREAD.start()
 
 
 # Trigger pre-warm on module import; backend startup hits this and the
@@ -512,7 +506,7 @@ def warm_cache_in_background() -> None:
 warm_cache_in_background()
 
 
-def _patch_env_port(env_path: str, key: str, value: str) -> None:
+def p_patch_env_port(env_path: str, key: str, value: str) -> None:
     """Idempotent in-place rewrite: `KEY=...` → `KEY=value`. Appends if
     the key isn't present. Preserves surrounding lines untouched."""
     if not os.path.exists(env_path):
@@ -557,12 +551,12 @@ def seed_webapp_template_workspace(workspace_dir: str, frontend_port: int) -> No
     shutil.copytree(
         WEBAPP_TEMPLATE_DIR,
         workspace_dir,
-        ignore=_ignore_backend,
+        ignore=p_ignore_backend,
         dirs_exist_ok=True,
     )
     # Symlink the workspace's frontend/node_modules at the warm cache so
     # `npm install` can be skipped entirely by the workspace run.sh.
-    _link_node_modules(workspace_dir)
+    p_link_node_modules(workspace_dir)
     env_path = os.path.join(workspace_dir, ".env")
     env_example_path = os.path.join(workspace_dir, ".env.example")
     src_example = os.path.join(WEBAPP_TEMPLATE_DIR, ".env.example")
@@ -579,17 +573,17 @@ def seed_webapp_template_workspace(workspace_dir: str, frontend_port: int) -> No
         with open(env_path, "w", encoding="utf-8") as f:
             f.write("BACKEND_PORT=NONE\nFRONTEND_PORT=4949\n")
 
-    _patch_env_port(env_path, "FRONTEND_PORT", str(frontend_port))
-    _patch_env_port(env_example_path, "FRONTEND_PORT", str(frontend_port))
+    p_patch_env_port(env_path, "FRONTEND_PORT", str(frontend_port))
+    p_patch_env_port(env_example_path, "FRONTEND_PORT", str(frontend_port))
 
     # Install-specific paths; .env only.
-    _patch_env_port(env_path, "OPENSWARM_TEMPLATE_BACKEND_PATH", _TEMPLATE_BACKEND_PATH)
-    _patch_env_port(env_path, "OPENSWARM_DEBUGGER_PATH", _DEBUGGER_PATH)
+    p_patch_env_port(env_path, "OPENSWARM_TEMPLATE_BACKEND_PATH", TEMPLATE_BACKEND_PATH)
+    p_patch_env_port(env_path, "OPENSWARM_DEBUGGER_PATH", DEBUGGER_PATH)
     # Backend-venv warm-cache path; backend_init.sh checks this for a
     # pre-populated `.venv/` to cp -aR into the workspace instead of
     # paying the ~25s venv-create + pip-install cost. Written even if
     # the cache isn't ready yet; backend_init.sh re-checks at run time.
-    _patch_env_port(env_path, "OPENSWARM_BACKEND_VENV_CACHE", _warm_venv_dir())
+    p_patch_env_port(env_path, "OPENSWARM_BACKEND_VENV_CACHE", p_warm_venv_dir())
 
     # Make the shipped scripts executable. tarball/git extracts may strip
     # the +x bit depending on how the snapshot was vendored.
diff --git a/backend/apps/outputs/workspace_io.py b/backend/apps/outputs/workspace_io.py
index ccb75b28..cb09df0a 100644
--- a/backend/apps/outputs/workspace_io.py
+++ b/backend/apps/outputs/workspace_io.py
@@ -13,7 +13,7 @@ from backend.config.json_store import read_json_or_none, atomic_write_json
 logger = logging.getLogger(__name__)
 
 
-def _load_all() -> list[Output]:
+def load_all() -> list[Output]:
     result = []
     if not os.path.exists(DATA_DIR):
         return result
@@ -29,11 +29,11 @@ def _load_all() -> list[Output]:
     return result
 
 
-def _save(output: Output):
+def save(output: Output):
     atomic_write_json(os.path.join(DATA_DIR, f"{output.id}.json"), output.model_dump())
 
 
-def _load(output_id: str) -> Output:
+def load(output_id: str) -> Output:
     data = read_json_or_none(os.path.join(DATA_DIR, f"{output_id}.json"))
     if data is None:
         raise HTTPException(status_code=404, detail="Output not found")
@@ -54,7 +54,7 @@ def load_output(output_id: str) -> Output | None:
 # agent is active. Result: backend CPU pegged on JSON-serializing
 # auto-generated chunks the frontend will then throw away. The frontend
 # already filters these for display; this skip is the real fix.
-_WALK_SKIP_DIRS = frozenset({
+P_WALK_SKIP_DIRS = frozenset({
     "node_modules",
     ".vite",
     ".vite-cache",
@@ -75,10 +75,10 @@ _WALK_SKIP_DIRS = frozenset({
 # they're not what the user/agent is editing. Anything over the cap
 # returns a truncated stub the frontend treats as "open the file
 # directly to see full contents."
-_WALK_MAX_FILE_BYTES = 256 * 1024
+P_WALK_MAX_FILE_BYTES = 256 * 1024
 
 
-def _walk_directory(folder: str) -> dict[str, str]:
+def walk_directory(folder: str) -> dict[str, str]:
     """Walk a directory tree and return {relative_path: content} for all
     text files the user is actually authoring. Skips build/install
     directories AND truncates oversize files; both critical for the
@@ -92,7 +92,7 @@ def _walk_directory(folder: str) -> dict[str, str]:
         # Doing it here means we never even stat the children, so a
         # 10k-file `.venv/` costs ~one stat (on the dir itself) instead
         # of 10k.
-        dirs[:] = [d for d in dirs if d not in _WALK_SKIP_DIRS]
+        dirs[:] = [d for d in dirs if d not in P_WALK_SKIP_DIRS]
         for fname in filenames:
             full_path = os.path.join(root, fname)
             # Normalize to forward-slash keys so the frontend's
@@ -106,10 +106,10 @@ def _walk_directory(folder: str) -> dict[str, str]:
                 # Stat first; cheap, lets us skip giant files without
                 # opening + reading them.
                 size = os.path.getsize(full_path)
-                if size > _WALK_MAX_FILE_BYTES:
+                if size > P_WALK_MAX_FILE_BYTES:
                     files[rel_path] = (
                         f"// [openswarm] file truncated ({size} bytes > "
-                        f"{_WALK_MAX_FILE_BYTES} byte cap). Open directly "
+                        f"{P_WALK_MAX_FILE_BYTES} byte cap). Open directly "
                         f"to view full contents."
                     )
                     continue
diff --git a/backend/apps/service/buffer.py b/backend/apps/service/buffer.py
index c8d3dad2..5a0caccf 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 p_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:
@@ -54,11 +54,12 @@ def _conn(spool_path: str) -> Iterator[sqlite3.Connection]:
         c.close()
 
 
+# Public - Used by client.py
 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, p_conn(spool_path) as c:
         c.execute(
             "INSERT INTO spool (kind, payload, created_at) VALUES (?, ?, ?)",
             (kind, body, now),
@@ -68,8 +69,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,23 +87,24 @@ 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
 
 
+# Public - Used by client.py
 def drain(spool_path: str, batch_size: int = 50) -> list[tuple[int, str, dict]]:
     """Read up to `batch_size` oldest entries. Returns (id, kind, payload)
     triples; caller is responsible for calling `acknowledge(ids)` once the
     cloud accepts them."""
     if not os.path.exists(spool_path):
         return []
-    with _lock, _conn(spool_path) as c:
+    with P_LOCK, p_conn(spool_path) as c:
         rows = c.execute(
             "SELECT id, kind, payload FROM spool ORDER BY id ASC LIMIT ?",
             (batch_size,),
@@ -113,30 +115,26 @@ 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, p_conn(spool_path) as c:
                 c.execute("DELETE FROM spool WHERE id = ?", (rid,))
             logger.warning("Dropped corrupt spool row id=%s", rid)
     return out
 
 
+# Public - Used by client.py
 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, p_conn(spool_path) as c:
         c.executemany("DELETE FROM spool WHERE id = ?", [(i,) for i in ids])
 
 
+# Public - Used by service.py
 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, p_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:
-        c.execute("DELETE FROM spool")
+    return int(row[0]) if row else 0
\ No newline at end of file
diff --git a/backend/apps/service/client.py b/backend/apps/service/client.py
index 6c445f33..15cb6613 100644
--- a/backend/apps/service/client.py
+++ b/backend/apps/service/client.py
@@ -1,6 +1,6 @@
 """Operational state forwarder.
 
-Single public surface: `submit(kind, payload)`. The desktop hands off
+Single public surface: `p_submit(kind, payload)`. The desktop hands off
 opaque payload dicts; the cloud at api.openswarm.com is responsible for
 parsing and routing them. The desktop has no schema knowledge.
 
@@ -34,26 +34,27 @@ 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()
+P_TEST_SINK: Optional[Any] = None
+P_INSTALL_ID: Optional[str] = None
+P_USER_ID: Optional[str] = None
+P_INFLIGHT: int = 0
+P_INFLIGHT_LOCK = asyncio.Lock()
+P_DRAIN_LOCK = asyncio.Lock()
 
 
-def _spool_path() -> str:
+# Public - Used by service.py
+def spool_path() -> str:
     try:
         from backend.config.paths import SETTINGS_DIR
         return os.path.join(SETTINGS_DIR, "service_spool.db")
@@ -61,34 +62,28 @@ def _spool_path() -> str:
         return os.path.expanduser("~/.openswarm/data/service_spool.db")
 
 
-def set_test_sink(fn: Optional[Any]) -> None:
-    """Test seam; receives every submission instead of the network."""
-    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 P_INSTALL_ID
+    if P_INSTALL_ID:
+        return P_INSTALL_ID
     try:
-        from backend.apps.settings.store import load_settings, _save_settings
+        from backend.apps.settings.store import load_settings, save_settings
         s = load_settings()
         iid = getattr(s, "installation_id", None)
         if not iid:
             iid = uuid4().hex
             s.installation_id = iid
-            _save_settings(s)
-        _install_id = iid
+            save_settings(s)
+        P_INSTALL_ID = iid
     except Exception:
-        _install_id = uuid4().hex
-    return _install_id
+        P_INSTALL_ID = uuid4().hex
+    return P_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()
@@ -107,12 +102,8 @@ def _get_user_id() -> Optional[str]:
         return None
 
 
-def set_user_id(uid: Optional[str]) -> None:
-    global _user_id
-    _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 +121,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:
@@ -156,8 +147,8 @@ def _envelope() -> dict:
             except Exception:
                 pass
         if not ianatz:
-            import datetime as _dt
-            local_tz = _dt.datetime.now().astimezone().tzinfo
+            import datetime as dt
+            local_tz = dt.datetime.now().astimezone().tzinfo
             if local_tz:
                 ianatz = str(local_tz)
         if ianatz:
@@ -182,20 +173,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 +194,43 @@ 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 P_TEST_SINK is not None:
         try:
-            _test_sink(kind, body)
+            P_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)
 
 
+# Public - Used by service.py
 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 +239,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,15 +256,16 @@ async def drain_spool(batch_size: int = 50) -> int:
 # Public API
 # --------------------------------------------------------------------------
 
-def _log(kind: str) -> None:
+def p_log(kind: str) -> None:
     """Append to the rolling operational log for diagnostics."""
     try:
-        from backend.apps.service.ring_buffer import record
-        record(kind)
+        from backend.apps.service.ring_buffer import record as ring_record
+        ring_record(kind)
     except Exception:
         pass
 
 
+# Public - Used by agents.py, settings.py, cloud_sync.py, subscription.router.py, service.py
 def sync(data: dict | None = None) -> None:
     """Sync operational state to the cloud. Single entry point.
 
@@ -288,29 +281,29 @@ 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")
-    if _test_sink is not None:
+    p_log("s")
+    if P_TEST_SINK is not None:
         try:
-            _test_sink("s", body)
+            P_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(payload: dict) -> None:
+def p_submit(payload: dict) -> None:
     """Routes through sync(). The cloud demuxes by payload shape (state /
     sync / diagnostic / event), so kind here is informational; the routing
     happens server-side in openswarm-cloud/src/routes/service/ingest.ts.
@@ -318,7 +311,7 @@ def submit(payload: dict) -> None:
     sync(payload)
 
 
-def _schedule(coro) -> None:
+def p_schedule(coro) -> None:
     try:
         loop = asyncio.get_running_loop()
     except RuntimeError:
@@ -328,75 +321,32 @@ def _schedule(coro) -> None:
         return
     import threading
 
-    def _run():
+    def run_coro():
         try:
             asyncio.run(coro)
         except Exception:
             pass
 
-    threading.Thread(target=_run, daemon=True).start()
+    threading.Thread(target=run_coro, daemon=True).start()
 
 
 # --------------------------------------------------------------------------
-# Backwards-compat shims for legacy call sites. New code calls submit()
+# Backwards-compat shims for legacy call sites. New code calls p_submit(()
 # directly. These keep the ~50 existing import sites in the codebase
 # working unchanged. Removed in a future cleanup once nothing imports
 # from older import paths.
 # --------------------------------------------------------------------------
 
-def submit_event(
-    surface: str,
-    action: str,
-    props: Optional[dict] = None,
-    *,
-    session_id: Optional[str] = None,
-    dashboard_id: Optional[str] = None,
-) -> None:
-    """Legacy event-shape submit. Bundles surface/action into the opaque
-    payload and hands off via submit()."""
-    p = {
-        "surface": surface,
-        "action": action,
-        "props": props or {},
-        "session_id": session_id,
-        "dashboard_id": dashboard_id,
-    }
-    submit("event", p)
-
-
-def submit_session_close(session_dump: dict, activity: Optional[dict] = None) -> None:
-    submit("session", {"usage_window": session_dump, "activity": activity or {}})
-
-
+# Public - Used by agent_manager.py
 def submit_diagnostic(diagnostic: dict) -> None:
     try:
         from backend.apps.service.ring_buffer import snapshot
         diagnostic["recent_log"] = snapshot()
     except Exception:
         pass
-    submit("diagnostic", {"diagnostic": diagnostic})
-
-
-def update_identity(extra: Optional[dict] = None) -> None:
-    submit("state", {"identity": extra or {}})
-
-
-def record(
-    event_type: str,
-    properties: Optional[dict] = None,
-    session_id: Optional[str] = None,
-    dashboard_id: Optional[str] = None,
-) -> None:
-    """Legacy collector.record() shim; splits dotted name into surface/action."""
-    if "." in event_type:
-        surface, action = event_type.split(".", 1)
-    else:
-        surface, action = event_type, "fired"
-    submit_event(
-        surface=surface, action=action, props=properties or {},
-        session_id=session_id, dashboard_id=dashboard_id,
-    )
+    p_submit("diagnostic", {"diagnostic": diagnostic})
 
 
+# Public - Used by settings.py, auth.router.py, subscription.router.py
 def identify(extra_properties: Optional[dict] = None) -> None:
-    update_identity(extra_properties or {})
+    p_submit("state", {"identity": extra_properties or {}})
diff --git a/backend/apps/service/ring_buffer.py b/backend/apps/service/ring_buffer.py
index d9006b6b..17a9a344 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,5 @@ 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)
-
-
-def clear() -> None:
-    with _lock:
-        _buffer.clear()
+    with P_LOCK:
+        return list(P_BUFFER)
\ No newline at end of file
diff --git a/backend/apps/service/service.py b/backend/apps/service/service.py
index 10779541..233e9fb4 100644
--- a/backend/apps/service/service.py
+++ b/backend/apps/service/service.py
@@ -26,22 +26,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.client import sync, drain_spool, spool_path
 from backend.apps.service.version import 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
 
-_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:
@@ -51,25 +51,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)
+            import datetime as dt
+            P_PULSE_HOURS.add(dt.datetime.now().hour)
         except Exception:
             pass
 
@@ -80,40 +80,34 @@ async def _pulse_loop():
                 stats = await get_usage_stats()
                 if stats:
                     cur_cost = stats.get("totalCost", 0) or 0
-                    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)
+                    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({
+                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()
+            await drain_spool()
         except Exception:
             pass
         await asyncio.sleep(60)
@@ -121,7 +115,7 @@ async def _drain_loop():
 
 @asynccontextmanager
 async def service_lifespan():
-    global _pulse_task, _drain_task
+    global P_PULSE_TASK, P_DRAIN_TASK
 
     try:
         from backend.apps.settings.settings import load_settings
@@ -153,7 +147,7 @@ async def service_lifespan():
         for cp in getattr(settings, "custom_providers", []):
             providers.append(cp.name)
 
-        svc.sync({
+        sync({
             "os": platform.system(),
             "platform": platform.platform(),
             "provider_count": len(providers),
@@ -188,7 +182,7 @@ async def service_lifespan():
         if is_paying and getattr(settings, "openswarm_subscription_expires", None):
             id_props["subscription_expires"] = settings.openswarm_subscription_expires
 
-        svc.sync({"identity": id_props})
+        sync({"identity": id_props})
     except Exception as e:
         logger.debug(f"Service startup event failed (non-critical): {e}")
 
@@ -198,26 +192,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
 
     try:
         from backend.apps.nine_router.process import stop
@@ -235,7 +229,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
@@ -253,11 +247,11 @@ 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"))
 
-    def _is_real(sess: dict) -> bool:
+    def is_real(sess: dict) -> bool:
         # "Real" = actually ran. Empty draft/abandoned sessions (no assistant turn, no tokens,
         # no active time) otherwise inflate the count and drag every average toward zero.
         if (sess.get("agent_active_ms") or 0) > 0 or (sess.get("cost_usd") or 0) > 0:
@@ -267,7 +261,7 @@ async def usage_summary():
             return True
         return any(m.get("role") == "assistant" for m in sess.get("messages", []))
 
-    sessions = [s for s in sessions if _is_real(s)]
+    sessions = [s for s in sessions if is_real(s)]
 
     total_sessions = len(sessions)
     total_cost = sum(s.get("cost_usd", 0) for s in sessions)
@@ -438,25 +432,25 @@ async def post_submit(body=Body(...)):
         for item in body:
             if isinstance(item, dict):
                 if any(k in item for k in ("s", "a", "p")):
-                    svc.sync(item)
+                    sync(item)
                     continue
                 kind = item.get("kind") or ""
                 payload = item.get("payload") or {}
                 if isinstance(payload, dict):
                     payload.setdefault("kind", kind)
-                    svc.sync(payload)
+                    sync(payload)
         return {"ok": True}
     if not isinstance(body, dict):
         return {"ok": False, "error": "JSON object or array required"}
     # Shape 1: frontend `report()`; flat {s, a, p, ...}
     if any(k in body for k in ("s", "a", "p")):
-        svc.sync(body)
+        sync(body)
         return {"ok": True}
     # Shape 2: legacy {kind, payload}
     kind = body.get("kind") or ""
     payload = body.get("payload")
     if kind and isinstance(payload, dict):
-        svc.sync(payload)
+        sync(payload)
         return {"ok": True}
     return {"ok": False, "error": "expected {s,a,p,...} or {kind,payload}"}
 
@@ -474,7 +468,7 @@ async def post_event(body: dict):
     if not action:
         action = "fired"
 
-    svc.sync({
+    sync({
         "s": str(surface)[:64],
         "a": str(action)[:64],
         "p": body.get("props") or body.get("properties") or {},
@@ -485,4 +479,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(spool_path())}
diff --git a/backend/apps/service/version.py b/backend/apps/service/version.py
index c7fbb402..495231f2 100644
--- a/backend/apps/service/version.py
+++ b/backend/apps/service/version.py
@@ -7,8 +7,7 @@ path to electron/package.json hops the same three dirnames up to the repo root.
 import json
 import os
 
-
-def _read_app_version() -> str:
+def p_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
@@ -23,13 +22,13 @@ def _read_app_version() -> str:
     # app_version="unknown" pre-fix. Kept for backward compatibility with
     # dev runs and as a safety net if the env var is ever unset.
     try:
-        _here = os.path.dirname(os.path.abspath(__file__))
-        _repo = os.path.dirname(os.path.dirname(os.path.dirname(_here)))
-        _pkg = os.path.join(_repo, "electron", "package.json")
-        with open(_pkg, encoding="utf-8") as _f:
-            return json.load(_f).get("version", "unknown")
+        here = os.path.dirname(os.path.abspath(__file__))
+        repo = os.path.dirname(os.path.dirname(os.path.dirname(here)))
+        pkg = os.path.join(repo, "electron", "package.json")
+        with open(pkg, encoding="utf-8") as f:
+            return json.load(f).get("version", "unknown")
     except (OSError, ValueError, KeyError):
         return "unknown"
 
 
-APP_VERSION = _read_app_version()
+APP_VERSION = p_read_app_version()
diff --git a/backend/apps/settings/credentials.py b/backend/apps/settings/credentials.py
index f21231f5..0e3979b7 100644
--- a/backend/apps/settings/credentials.py
+++ b/backend/apps/settings/credentials.py
@@ -31,7 +31,7 @@ def proxy_auth(settings: AppSettings) -> tuple[str | None, str | None]:
     return (None, None)
 
 
-def _check_9router() -> bool:
+def p_check_9router() -> bool:
     """Check if 9Router is running locally."""
     try:
         import httpx
@@ -56,7 +56,7 @@ def get_anthropic_client(settings: AppSettings) -> anthropic.AsyncAnthropic:
         return anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
 
     # Fall back to 9Router (free for users with Claude/ChatGPT/Gemini subscriptions).
-    if _check_9router():
+    if p_check_9router():
         return anthropic.AsyncAnthropic(
             api_key="9router",
             base_url="http://localhost:20128",
diff --git a/backend/main.py b/backend/main.py
index 1a2323df..5f3a758e 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -288,8 +288,8 @@ async def websocket_runtime_logs(websocket: WebSocket, workspace_id: str):
     if not _ws_auth_ok(websocket):
         return
     await websocket.accept()
-    from backend.apps.outputs.runtime import manager as runtime_manager
-    rt = runtime_manager.get(workspace_id)
+    from backend.apps.outputs.runtime import RUNTIME_MANAGER
+    rt = RUNTIME_MANAGER.get(workspace_id)
     if rt is None:
         # No active runtime, surface that to the client and close. The
         # frontend will call /runtime/start and reconnect. Also emit a
@@ -298,8 +298,8 @@ async def websocket_runtime_logs(websocket: WebSocket, workspace_id: str):
         # webapp_template workspaces instead of falling back to the
         # legacy /serve/index.html URL (which 404s in new-mode).
         try:
-            from backend.apps.outputs.outputs import _runtime_status_payload
-            status = _runtime_status_payload(workspace_id)
+            from backend.apps.outputs.outputs import runtime_status_payload
+            status = runtime_status_payload(workspace_id)
             await websocket.send_text(json.dumps({
                 "event": "runtime:status",
                 "workspace_id": workspace_id,
diff --git a/backend/tests/test_service.py b/backend/tests/test_service.py
index ac01d67d..62629b8f 100644
--- a/backend/tests/test_service.py
+++ b/backend/tests/test_service.py
@@ -244,7 +244,7 @@ def test_buffer_clear(tmp_path):
     from backend.apps.service import buffer
     spool = str(tmp_path / "s.db")
     buffer.enqueue(spool, "s:/x", {}, now=time.time())
-    buffer.clear(spool)
+    buffer.p_clear(spool) # p-private-ignore
     assert buffer.count(spool) == 0