From fafbe4518750e7525d09ae775a81e6ed0e8dd509 Mon Sep 17 00:00:00 2001 From: haikdc Date: Sat, 13 Jun 2026 20:34:25 -0700 Subject: [PATCH] [haik]: refactor backend naming conventions across 14 files: rename module-internal functions from _fn to p_fn, module-level constants from _CONST to P_CONST, and cross-module public functions from _fn to fn (e.g. _load/_save to load/save in dashboards, _pending_oauth to PENDING_OAUTH in oauth_state, _nr to nr in sync, _process to P_PROCESS in process); hoist inline import json to module-level in modes.py; clean up redundant import subprocess alias and unused _aux_base variable; update all import sites in browser_agent, prompt_context, main, sync, and sync_custom --- backend/apps/agents/browser/browser_agent.py | 10 +- .../agents/manager/prompt/prompt_context.py | 2 +- backend/apps/dashboards/dashboards.py | 40 +++--- backend/apps/discord_mcp_shim/server.py | 104 +++++++------- backend/apps/google_workspace_mcp_shim/run.py | 6 +- backend/apps/mcp_registry/mcp_registry.py | 132 +++++++++--------- backend/apps/modes/modes.py | 30 ++-- backend/apps/nine_router/oauth.py | 54 +++---- backend/apps/nine_router/process.py | 125 ++++++++--------- backend/apps/nine_router/sync.py | 26 ++-- backend/apps/nine_router/sync_custom.py | 48 +++---- backend/apps/oauth_state.py | 16 +-- backend/apps/outputs/executor.py | 24 ++-- backend/main.py | 14 +- 14 files changed, 315 insertions(+), 316 deletions(-) diff --git a/backend/apps/agents/browser/browser_agent.py b/backend/apps/agents/browser/browser_agent.py index b9c2235d..ddc78055 100644 --- a/backend/apps/agents/browser/browser_agent.py +++ b/backend/apps/agents/browser/browser_agent.py @@ -2195,8 +2195,8 @@ def _find_reusable_card(dashboard_id: str, url: str, parent_session_id: str | No if not (dashboard_id and want): return "" try: - from backend.apps.dashboards.dashboards import _load - cards = _load(dashboard_id).layout.browser_cards + from backend.apps.dashboards.dashboards import load as load_dashboard + cards = load_dashboard(dashboard_id).layout.browser_cards except Exception: return "" from backend.apps.agents.agent_manager import agent_manager @@ -2218,10 +2218,10 @@ def _find_reusable_card(dashboard_id: str, url: str, parent_session_id: str | No async def _create_browser_card(dashboard_id: str, url: str, parent_session_id: str | None = None) -> str: """Create a new browser card on the dashboard and return its browser_id.""" - from backend.apps.dashboards.dashboards import _load, _save + from backend.apps.dashboards.dashboards import load as load_dashboard, save as save_dashboard from backend.apps.dashboards.models import BrowserCardPosition, BrowserTab - dashboard = _load(dashboard_id) + dashboard = load_dashboard(dashboard_id) browser_id = f"browser-{uuid4().hex[:8]}" tab_id = f"tab-{uuid4().hex[:8]}" tab = BrowserTab(id=tab_id, url=url or "https://www.google.com", title="") @@ -2238,7 +2238,7 @@ async def _create_browser_card(dashboard_id: str, url: str, parent_session_id: s ) dashboard.layout.browser_cards[browser_id] = card dashboard.updated_at = datetime.now() - _save(dashboard) + save_dashboard(dashboard) await ws_manager.broadcast_global("dashboard:browser_card_added", { "dashboard_id": dashboard_id, diff --git a/backend/apps/agents/manager/prompt/prompt_context.py b/backend/apps/agents/manager/prompt/prompt_context.py index ba243566..7b7c1a86 100644 --- a/backend/apps/agents/manager/prompt/prompt_context.py +++ b/backend/apps/agents/manager/prompt/prompt_context.py @@ -107,7 +107,7 @@ def build_browser_context(dashboard_id: str | None, selected_browser_ids: list[s if not dashboard_id: return None try: - from backend.apps.dashboards.dashboards import _load as load_dashboard + from backend.apps.dashboards.dashboards import load as load_dashboard dashboard = load_dashboard(dashboard_id) except Exception: return None diff --git a/backend/apps/dashboards/dashboards.py b/backend/apps/dashboards/dashboards.py index a6453304..7b0158a7 100644 --- a/backend/apps/dashboards/dashboards.py +++ b/backend/apps/dashboards/dashboards.py @@ -22,7 +22,7 @@ from backend.config.json_store import read_json_or_none, atomic_write_json OLD_LAYOUT_FILE = os.path.join(OLD_LAYOUT_DIR, "layout.json") -def _load_all() -> list[Dashboard]: +def p_load_all() -> list[Dashboard]: result = [] if not os.path.exists(DATA_DIR): return result @@ -40,11 +40,11 @@ def _load_all() -> list[Dashboard]: return result -def _save(dashboard: Dashboard): +def save(dashboard: Dashboard): atomic_write_json(os.path.join(DATA_DIR, f"{dashboard.id}.json"), dashboard.model_dump(mode="json")) -def _load(dashboard_id: str) -> Dashboard: +def load(dashboard_id: str) -> Dashboard: path = os.path.join(DATA_DIR, f"{dashboard_id}.json") data = read_json_or_none(path) if data is None: @@ -52,15 +52,15 @@ def _load(dashboard_id: str) -> Dashboard: return Dashboard(**data) -def _delete(dashboard_id: str): +def p_delete(dashboard_id: str): path = os.path.join(DATA_DIR, f"{dashboard_id}.json") if os.path.exists(path): os.remove(path) -def _migrate_if_needed(): +def p_migrate_if_needed(): """One-time migration: if no dashboards exist, create 'Dashboard 1' from old layout.""" - existing = _load_all() + existing = p_load_all() if existing: return @@ -78,7 +78,7 @@ def _migrate_if_needed(): logger.exception("Failed to read old layout.json, using empty layout") dashboard = Dashboard(name="Dashboard 1", layout=layout) - _save(dashboard) + save(dashboard) logger.info(f"Created default dashboard: {dashboard.id}") if os.path.exists(SESSIONS_DIR): @@ -102,7 +102,7 @@ def _migrate_if_needed(): @asynccontextmanager async def dashboards_lifespan(): os.makedirs(DATA_DIR, exist_ok=True) - _migrate_if_needed() + p_migrate_if_needed() yield @@ -111,7 +111,7 @@ dashboards = SubApp("dashboards", dashboards_lifespan) @dashboards.router.get("/list") async def list_dashboards(): - all_dashboards = _load_all() + all_dashboards = p_load_all() all_dashboards.sort(key=lambda d: d.updated_at or d.created_at, reverse=True) items = [] for d in all_dashboards: @@ -132,7 +132,7 @@ async def list_dashboards(): @dashboards.router.post("/create") async def create_dashboard(body: DashboardCreate): dashboard = Dashboard(name=body.name) - _save(dashboard) + save(dashboard) return dashboard.model_dump(mode="json") @@ -212,7 +212,7 @@ async def seed_orchestration_demo(dashboard_id: str): drags it into a new agent and asks for a PDF report; which delegates back to this seeded agent. """ - _load(dashboard_id) # validate dashboard exists + load(dashboard_id) # validate dashboard exists session_id = uuid4().hex now = datetime.now() @@ -294,7 +294,7 @@ async def seed_orchestration_demo(dashboard_id: str): @dashboards.router.post("/{dashboard_id}/generate-name") async def generate_name(dashboard_id: str): - dashboard = _load(dashboard_id) + dashboard = load(dashboard_id) if not dashboard.auto_named and dashboard.name != "Untitled Dashboard": return {"name": dashboard.name, "auto_named": dashboard.auto_named} @@ -319,7 +319,7 @@ async def generate_name(dashboard_id: str): from backend.apps.settings.credentials import get_anthropic_client_for_model from backend.apps.agents.providers.registry import resolve_aux_model global_settings = load_settings() - aux_model, _aux_base = await resolve_aux_model(global_settings, preferred_tier="haiku") + aux_model, _ = await resolve_aux_model(global_settings, preferred_tier="haiku") client = get_anthropic_client_for_model(global_settings, aux_model) # Mirrors generate_title's hardening: the tasks are inert text to LABEL, never answer, @@ -352,19 +352,19 @@ async def generate_name(dashboard_id: str): dashboard.name = fallback dashboard.auto_named = True dashboard.updated_at = datetime.now() - _save(dashboard) + save(dashboard) return {"name": dashboard.name, "auto_named": True} @dashboards.router.get("/{dashboard_id}") async def get_dashboard(dashboard_id: str): - dashboard = _load(dashboard_id) + dashboard = load(dashboard_id) return dashboard.model_dump(mode="json") @dashboards.router.put("/{dashboard_id}") async def update_dashboard(dashboard_id: str, body: DashboardUpdate): - dashboard = _load(dashboard_id) + dashboard = load(dashboard_id) if body.name is not None: dashboard.name = body.name dashboard.auto_named = False @@ -377,13 +377,13 @@ async def update_dashboard(dashboard_id: str, body: DashboardUpdate): # Only a real screenshot write moves the sort key; layout/rename saves don't reorder. dashboard.preview_updated_at = now dashboard.updated_at = now - _save(dashboard) + save(dashboard) return dashboard.model_dump(mode="json") @dashboards.router.delete("/{dashboard_id}") async def delete_dashboard(dashboard_id: str): - _load(dashboard_id) + load(dashboard_id) if os.path.exists(SESSIONS_DIR): for fname in os.listdir(SESSIONS_DIR): @@ -409,13 +409,13 @@ async def delete_dashboard(dashboard_id: str): except Exception: logger.warning(f"Failed to delete active session {sid} during dashboard deletion") - _delete(dashboard_id) + p_delete(dashboard_id) return {"ok": True} @dashboards.router.post("/{dashboard_id}/duplicate") async def duplicate_dashboard(dashboard_id: str): - source = _load(dashboard_id) + source = load(dashboard_id) source_data = source.model_dump(mode="json") new_id = uuid4().hex now = datetime.now().isoformat() diff --git a/backend/apps/discord_mcp_shim/server.py b/backend/apps/discord_mcp_shim/server.py index a806841f..caf61c98 100644 --- a/backend/apps/discord_mcp_shim/server.py +++ b/backend/apps/discord_mcp_shim/server.py @@ -24,10 +24,10 @@ ALLOWED_GUILDS = set( # -- MCP tool definitions (exposed to the agent) --------------------------- # Names match the original mcp-discord surface so prompts that referenced -# `discord_send` etc. keep working. inputSchema deliberately matches what +# `discordp_send` etc. keep working. inputSchema deliberately matches what # the original package documented. -TOOLS = [ +P_TOOLS = [ { "name": "discord_login", "description": "Verify the Discord bot helper is reachable. Returns the bot's joined guilds.", @@ -214,7 +214,7 @@ TOOLS = [ # -- HTTP plumbing --------------------------------------------------------- -def _call( +def p_call( method: str, path: str, *, @@ -267,17 +267,17 @@ def _call( return 0, f"Request failed: {e!r}" -def _err(text: str) -> dict: +def p_err(text: str) -> dict: return {"content": [{"type": "text", "text": f"Error: {text}"}], "isError": True} -def _ok(payload) -> dict: +def p_ok(payload) -> dict: if isinstance(payload, str): return {"content": [{"type": "text", "text": payload}]} return {"content": [{"type": "text", "text": json.dumps(payload, indent=2, default=str)}]} -def _check_guild(guild_id: str) -> str | None: +def p_check_guild(guild_id: str) -> str | None: """Return an error string if guild_id is outside the user-authorized set, else None. The set is sourced from OPENSWARM_DISCORD_GUILD_IDS env var (CSV) which @@ -297,71 +297,71 @@ def _check_guild(guild_id: str) -> str | None: # -- Tool implementations -------------------------------------------------- -def handle_tool_call(name: str, args: dict) -> dict: +def p_handle_tool_call(name: str, args: dict) -> dict: if name == "discord_login": - status, body = _call("GET", "/users/@me/guilds") + status, body = p_call("GET", "/users/@me/guilds") if status != 200: - return _err(f"Discord proxy unreachable (HTTP {status}): {body}") - return _ok({"connected": True, "guilds": body}) + return p_err(f"Discord proxy unreachable (HTTP {status}): {body}") + return p_ok({"connected": True, "guilds": body}) if name == "discord_get_server_info": gid = str(args.get("guild_id", "")) - if (e := _check_guild(gid)): return _err(e) - status, body = _call("GET", f"/guilds/{gid}") - return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}") + if (e := p_check_guild(gid)): return p_err(e) + status, body = p_call("GET", f"/guilds/{gid}") + return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}") if name == "discord_list_channels": gid = str(args.get("guild_id", "")) - if (e := _check_guild(gid)): return _err(e) - status, body = _call("GET", f"/guilds/{gid}/channels") - return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}") + if (e := p_check_guild(gid)): return p_err(e) + status, body = p_call("GET", f"/guilds/{gid}/channels") + return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}") if name == "discord_create_text_channel": gid = str(args.get("guild_id", "")) - if (e := _check_guild(gid)): return _err(e) + if (e := p_check_guild(gid)): return p_err(e) payload: dict = {"name": args.get("name", ""), "type": 0} if args.get("parent_id"): payload["parent_id"] = args["parent_id"] if args.get("topic"): payload["topic"] = args["topic"] - status, body = _call("POST", f"/guilds/{gid}/channels", body=payload) - return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}") + status, body = p_call("POST", f"/guilds/{gid}/channels", body=payload) + return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}") if name == "discord_create_category": gid = str(args.get("guild_id", "")) - if (e := _check_guild(gid)): return _err(e) - status, body = _call("POST", f"/guilds/{gid}/channels", body={"name": args.get("name", ""), "type": 4}) - return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}") + if (e := p_check_guild(gid)): return p_err(e) + status, body = p_call("POST", f"/guilds/{gid}/channels", body={"name": args.get("name", ""), "type": 4}) + return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}") if name == "discord_edit_category": cid = str(args.get("channel_id", "")) payload: dict = {} if args.get("name"): payload["name"] = args["name"] - status, body = _call("PATCH", f"/channels/{cid}", body=payload) - return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}") + status, body = p_call("PATCH", f"/channels/{cid}", body=payload) + return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}") if name == "discord_delete_category" or name == "discord_delete_channel": cid = str(args.get("channel_id", "")) - status, body = _call("DELETE", f"/channels/{cid}") - return _ok({"deleted": True}) if status in (200, 204) else _err(f"HTTP {status}: {body}") + status, body = p_call("DELETE", f"/channels/{cid}") + return p_ok({"deleted": True}) if status in (200, 204) else p_err(f"HTTP {status}: {body}") if name == "discord_send": cid = str(args.get("channel_id", "")) content = str(args.get("content", "")) - status, body = _call("POST", f"/channels/{cid}/messages", body={"content": content}) - return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}") + status, body = p_call("POST", f"/channels/{cid}/messages", body={"content": content}) + return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}") if name == "discord_read_messages": cid = str(args.get("channel_id", "")) limit = max(1, min(int(args.get("limit", 50) or 50), 100)) - status, body = _call("GET", f"/channels/{cid}/messages", query={"limit": limit}) - return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}") + status, body = p_call("GET", f"/channels/{cid}/messages", query={"limit": limit}) + return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}") if name == "discord_add_reaction": cid = str(args.get("channel_id", "")) mid = str(args.get("message_id", "")) emoji = str(args.get("emoji", "")) # Discord's URL needs the emoji urlencoded; passes through. - status, body = _call("PUT", f"/channels/{cid}/messages/{mid}/reactions/{urllib.parse.quote(emoji, safe='')}/@me") - return _ok({"added": emoji}) if status in (200, 204) else _err(f"HTTP {status}: {body}") + status, body = p_call("PUT", f"/channels/{cid}/messages/{mid}/reactions/{urllib.parse.quote(emoji, safe='')}/@me") + return p_ok({"added": emoji}) if status in (200, 204) else p_err(f"HTTP {status}: {body}") if name == "discord_add_multiple_reactions": cid = str(args.get("channel_id", "")) @@ -369,46 +369,46 @@ def handle_tool_call(name: str, args: dict) -> dict: emojis = args.get("emojis", []) or [] results = [] for e in emojis: - status, body = _call("PUT", f"/channels/{cid}/messages/{mid}/reactions/{urllib.parse.quote(str(e), safe='')}/@me") + status, body = p_call("PUT", f"/channels/{cid}/messages/{mid}/reactions/{urllib.parse.quote(str(e), safe='')}/@me") results.append({"emoji": e, "ok": status in (200, 204), "status": status}) - return _ok({"reactions": results}) + return p_ok({"reactions": results}) if name == "discord_get_forum_channels": gid = str(args.get("guild_id", "")) - if (e := _check_guild(gid)): return _err(e) - status, body = _call("GET", f"/guilds/{gid}/channels") - if status != 200: return _err(f"HTTP {status}: {body}") + if (e := p_check_guild(gid)): return p_err(e) + status, body = p_call("GET", f"/guilds/{gid}/channels") + if status != 200: return p_err(f"HTTP {status}: {body}") # Filter to type 15 (forum). Discord channel types reference: # GUILD_FORUM = 15 forums = [ch for ch in (body or []) if isinstance(ch, dict) and ch.get("type") == 15] - return _ok(forums) + return p_ok(forums) if name == "discord_create_forum_post": fid = str(args.get("forum_id", "")) - status, body = _call("POST", f"/channels/{fid}/threads", body={ + status, body = p_call("POST", f"/channels/{fid}/threads", body={ "name": args.get("name", ""), "message": {"content": args.get("content", "")}, }) - return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}") + return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}") if name == "discord_get_forum_post": cid = str(args.get("channel_id", "")) mid = str(args.get("message_id", "")) - status, body = _call("GET", f"/channels/{cid}/messages/{mid}") - return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}") + status, body = p_call("GET", f"/channels/{cid}/messages/{mid}") + return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}") if name == "discord_reply_to_forum": cid = str(args.get("channel_id", "")) content = str(args.get("content", "")) - status, body = _call("POST", f"/channels/{cid}/messages", body={"content": content}) - return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}") + status, body = p_call("POST", f"/channels/{cid}/messages", body={"content": content}) + return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}") - return _err(f"Unknown tool: {name}") + return p_err(f"Unknown tool: {name}") # -- JSON-RPC stdio loop --------------------------------------------------- -def _send(id_, result=None, error=None): +def p_send(id_, result=None, error=None): msg = {"jsonrpc": "2.0", "id": id_} if error is not None: msg["error"] = error @@ -433,7 +433,7 @@ def main(): params = msg.get("params", {}) or {} if method == "initialize": - _send(id_, { + p_send(id_, { "protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": {"name": "openswarm-discord", "version": "1.0.0"}, @@ -441,18 +441,18 @@ def main(): elif method == "notifications/initialized": pass elif method == "tools/list": - _send(id_, {"tools": TOOLS}) + p_send(id_, {"tools": P_TOOLS}) elif method == "tools/call": name = params.get("name", "") args = params.get("arguments", {}) or {} try: - _send(id_, handle_tool_call(name, args)) + p_send(id_, p_handle_tool_call(name, args)) except Exception as e: - _send(id_, _err(f"shim crashed: {e!r}")) + p_send(id_, p_err(f"shim crashed: {e!r}")) elif method == "ping": - _send(id_, {}) + p_send(id_, {}) elif id_ is not None: - _send(id_, error={"code": -32601, "message": f"Method not found: {method}"}) + p_send(id_, error={"code": -32601, "message": f"Method not found: {method}"}) if __name__ == "__main__": diff --git a/backend/apps/google_workspace_mcp_shim/run.py b/backend/apps/google_workspace_mcp_shim/run.py index 103c5932..a4b48189 100644 --- a/backend/apps/google_workspace_mcp_shim/run.py +++ b/backend/apps/google_workspace_mcp_shim/run.py @@ -24,7 +24,7 @@ from google.oauth2.credentials import Credentials @functools.lru_cache(maxsize=1) -def _patched_get_credentials(): +def p_patched_get_credentials(): refresh_token = os.environ.get("GOOGLE_WORKSPACE_REFRESH_TOKEN") if not refresh_token: raise ValueError("GOOGLE_WORKSPACE_REFRESH_TOKEN env var is required") @@ -40,10 +40,10 @@ def _patched_get_credentials(): ) -gauth.get_credentials = _patched_get_credentials # vulture-ignore: get_credentials +gauth.get_credentials = p_patched_get_credentials # vulture-ignore: get_credentials -from google_workspace_mcp import __main__ as _gw_main # noqa: E402,F401 +from google_workspace_mcp import __main__ as gw_main # noqa: E402,F401 from google_workspace_mcp.app import mcp # noqa: E402 diff --git a/backend/apps/mcp_registry/mcp_registry.py b/backend/apps/mcp_registry/mcp_registry.py index 4a8bdeac..1bfe4370 100644 --- a/backend/apps/mcp_registry/mcp_registry.py +++ b/backend/apps/mcp_registry/mcp_registry.py @@ -12,21 +12,21 @@ from backend.config.Apps import SubApp logger = logging.getLogger(__name__) -REGISTRY_BASE = "https://registry.modelcontextprotocol.io/v0.1" -PAGE_LIMIT = 100 -REFRESH_INTERVAL_S = 3600 +P_REGISTRY_BASE = "https://registry.modelcontextprotocol.io/v0.1" +P_PAGE_LIMIT = 100 +P_REFRESH_INTERVAL_S = 3600 -GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "") -GITHUB_BATCH = 4000 if GITHUB_TOKEN else 50 -GITHUB_CONCURRENT = 10 +P_GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "") +P_GITHUB_BATCH = 4000 if P_GITHUB_TOKEN else 50 +P_GITHUB_CONCURRENT = 10 -_cache: dict[str, dict] = {} -_cache_updated_at: float = 0 -_refresh_task: Optional[asyncio.Task] = None -_stars_cache: dict[str, int] = {} +P_CACHE: dict[str, dict] = {} +P_CACHE_UPDATED_AT: float = 0 +P_REFRESH_TASK: Optional[asyncio.Task] = None +P_STARS_CACHE: dict[str, int] = {} -def _extract_gh_repo(repo_url: str) -> Optional[str]: +def p_extract_gh_repo(repo_url: str) -> Optional[str]: """Parse 'owner/repo' from a GitHub URL.""" if not repo_url or "github.com" not in repo_url: return None @@ -42,7 +42,7 @@ def _extract_gh_repo(repo_url: str) -> Optional[str]: return None -def _extract_server(entry: dict) -> Optional[dict]: +def p_extract_server(entry: dict) -> Optional[dict]: """Extract a flat server record from a registry entry, keeping only latest versions.""" meta = entry.get("_meta", {}).get("io.modelcontextprotocol.registry/official", {}) if not meta.get("isLatest"): @@ -96,7 +96,7 @@ def _extract_server(entry: dict) -> Optional[dict]: } -async def _fetch_all_servers() -> dict[str, dict]: +async def p_fetch_all_servers() -> dict[str, dict]: """Paginate through the full registry and return a dict keyed by server name.""" servers: dict[str, dict] = {} cursor: Optional[str] = None @@ -104,12 +104,12 @@ async def _fetch_all_servers() -> dict[str, dict]: async with httpx.AsyncClient(timeout=30.0) as client: while True: - params: dict = {"limit": PAGE_LIMIT} + params: dict = {"limit": P_PAGE_LIMIT} if cursor: params["cursor"] = cursor try: - resp = await client.get(f"{REGISTRY_BASE}/servers", params=params) + resp = await client.get(f"{P_REGISTRY_BASE}/servers", params=params) resp.raise_for_status() data = resp.json() except Exception as e: @@ -121,7 +121,7 @@ async def _fetch_all_servers() -> dict[str, dict]: break for entry in entries: - record = _extract_server(entry) + record = p_extract_server(entry) if record: servers[record["name"]] = record @@ -135,16 +135,16 @@ async def _fetch_all_servers() -> dict[str, dict]: return servers -GOOGLE_README_URL = "https://raw.githubusercontent.com/google/mcp/main/README.md" -GOOGLE_ICON_URL = "https://github.com/google.png?size=64" -_ENTRY_RE = re.compile(r"\[\*\*(.+?)\*\*\]\((.+?)\)(?:[,\s]*(.+))?") +P_GOOGLE_README_URL = "https://raw.githubusercontent.com/google/mcp/main/README.md" +P_GOOGLE_ICON_URL = "https://github.com/google.png?size=64" +P_ENTRY_RE = re.compile(r"\[\*\*(.+?)\*\*\]\((.+?)\)(?:[,\s]*(.+))?") -def _slugify(name: str) -> str: +def p_slugify(name: str) -> str: return re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-") -def _parse_google_readme(text: str) -> dict[str, dict]: +def p_parse_google_readme(text: str) -> dict[str, dict]: servers: dict[str, dict] = {} section: Optional[str] = None @@ -164,7 +164,7 @@ def _parse_google_readme(text: str) -> dict[str, dict]: if section is None: continue - m = _ENTRY_RE.search(stripped) + m = P_ENTRY_RE.search(stripped) if not m: continue @@ -172,7 +172,7 @@ def _parse_google_readme(text: str) -> dict[str, dict]: url = m.group(2).strip() desc_raw = (m.group(3) or "").strip().rstrip(".") - slug = _slugify(title) + slug = p_slugify(title) key = f"google/{slug}" is_github = "github.com" in url or "go.dev" in url @@ -195,7 +195,7 @@ def _parse_google_readme(text: str) -> dict[str, dict]: "repositoryUrl": repo_url, "remoteUrl": "", "remoteType": remote_type, - "iconUrl": GOOGLE_ICON_URL, + "iconUrl": P_GOOGLE_ICON_URL, "environmentVariables": [], "keywords": ["google", section], "license": "Apache-2.0", @@ -206,13 +206,13 @@ def _parse_google_readme(text: str) -> dict[str, dict]: return servers -async def _fetch_google_servers() -> dict[str, dict]: +async def p_fetch_google_servers() -> dict[str, dict]: """Fetch and parse Google's MCP server catalog from their GitHub README.""" try: async with httpx.AsyncClient(timeout=15.0) as client: - resp = await client.get(GOOGLE_README_URL) + resp = await client.get(P_GOOGLE_README_URL) resp.raise_for_status() - servers = _parse_google_readme(resp.text) + servers = p_parse_google_readme(resp.text) logger.info(f"Google MCP catalog: parsed {len(servers)} servers") return servers except Exception as e: @@ -220,40 +220,40 @@ async def _fetch_google_servers() -> dict[str, dict]: return {} -async def _fetch_github_stars(servers: dict[str, dict]): +async def p_fetch_github_stars(servers: dict[str, dict]): """Batch-fetch GitHub star counts for servers with GitHub repos. Uses an in-memory cache so stars accumulate across refresh cycles even when rate-limited (60 req/hr unauthenticated, 5 000 with GITHUB_TOKEN). """ - global _stars_cache + global P_STARS_CACHE needed: list[str] = [] for srv in servers.values(): - gh = _extract_gh_repo(srv.get("repositoryUrl", "")) - if gh and gh not in _stars_cache and gh not in needed: + gh = p_extract_gh_repo(srv.get("repositoryUrl", "")) + if gh and gh not in P_STARS_CACHE and gh not in needed: needed.append(gh) if not needed: - logger.info(f"GitHub stars: all {len(_stars_cache)} repos cached, 0 to fetch") - _apply_stars(servers) + logger.info(f"GitHub stars: all {len(P_STARS_CACHE)} repos cached, 0 to fetch") + p_apply_stars(servers) return - to_fetch = needed[: GITHUB_BATCH] + to_fetch = needed[: P_GITHUB_BATCH] logger.info( f"GitHub stars: fetching {len(to_fetch)} repos " - f"({len(_stars_cache)} cached, {len(needed)} pending)" + f"({len(P_STARS_CACHE)} cached, {len(needed)} pending)" ) headers: dict[str, str] = {"Accept": "application/vnd.github.v3+json"} - if GITHUB_TOKEN: - headers["Authorization"] = f"token {GITHUB_TOKEN}" + if P_GITHUB_TOKEN: + headers["Authorization"] = f"token {P_GITHUB_TOKEN}" - sem = asyncio.Semaphore(GITHUB_CONCURRENT) + sem = asyncio.Semaphore(P_GITHUB_CONCURRENT) rate_limited = False fetched = 0 - async def _fetch_one(client: httpx.AsyncClient, repo: str): + async def p_fetch_one(client: httpx.AsyncClient, repo: str): nonlocal rate_limited, fetched if rate_limited: return @@ -265,56 +265,56 @@ async def _fetch_github_stars(servers: dict[str, dict]): f"https://api.github.com/repos/{repo}", headers=headers ) if resp.status_code == 200: - _stars_cache[repo] = resp.json().get("stargazers_count", 0) + P_STARS_CACHE[repo] = resp.json().get("stargazers_count", 0) fetched += 1 elif resp.status_code in (403, 429): rate_limited = True logger.warning("GitHub API rate-limited, stopping star fetch") elif resp.status_code == 404: - _stars_cache[repo] = 0 + P_STARS_CACHE[repo] = 0 fetched += 1 except Exception as exc: logger.debug(f"GitHub stars fetch failed for {repo}: {exc}") async with httpx.AsyncClient(timeout=15.0) as client: - await asyncio.gather(*[_fetch_one(client, r) for r in to_fetch]) + await asyncio.gather(*[p_fetch_one(client, r) for r in to_fetch]) - logger.info(f"GitHub stars: fetched {fetched} new, {len(_stars_cache)} total cached") - _apply_stars(servers) + logger.info(f"GitHub stars: fetched {fetched} new, {len(P_STARS_CACHE)} total cached") + p_apply_stars(servers) -def _apply_stars(servers: dict[str, dict]): +def p_apply_stars(servers: dict[str, dict]): for srv in servers.values(): - gh = _extract_gh_repo(srv.get("repositoryUrl", "")) - srv["stars"] = _stars_cache.get(gh) if gh else None + gh = p_extract_gh_repo(srv.get("repositoryUrl", "")) + srv["stars"] = P_STARS_CACHE.get(gh) if gh else None -async def _refresh_loop(): +async def p_refresh_loop(): """Background loop that refreshes the cache on startup and then hourly.""" - global _cache, _cache_updated_at + global P_CACHE, P_CACHE_UPDATED_AT while True: try: community, google = await asyncio.gather( - _fetch_all_servers(), - _fetch_google_servers(), + p_fetch_all_servers(), + p_fetch_google_servers(), ) - _cache = {**community, **google} - await _fetch_github_stars(_cache) - _cache_updated_at = time.time() + P_CACHE = {**community, **google} + await p_fetch_github_stars(P_CACHE) + P_CACHE_UPDATED_AT = time.time() except Exception as e: logger.exception(f"MCP registry refresh error: {e}") - await asyncio.sleep(REFRESH_INTERVAL_S) + await asyncio.sleep(P_REFRESH_INTERVAL_S) @asynccontextmanager async def mcp_registry_lifespan(): - global _refresh_task - _refresh_task = asyncio.create_task(_refresh_loop()) + global P_REFRESH_TASK + P_REFRESH_TASK = asyncio.create_task(p_refresh_loop()) yield - if _refresh_task: - _refresh_task.cancel() + if P_REFRESH_TASK: + P_REFRESH_TASK.cancel() try: - await _refresh_task + await P_REFRESH_TASK except asyncio.CancelledError: pass @@ -324,13 +324,13 @@ mcp_registry = SubApp("mcp-registry", mcp_registry_lifespan) @mcp_registry.router.get("/stats") async def registry_stats(): - google = sum(1 for s in _cache.values() if s.get("source") == "google") - community = sum(1 for s in _cache.values() if s.get("source") == "community") + google = sum(1 for s in P_CACHE.values() if s.get("source") == "google") + community = sum(1 for s in P_CACHE.values() if s.get("source") == "community") return { - "total": len(_cache), + "total": len(P_CACHE), "google": google, "community": community, - "lastUpdated": _cache_updated_at, + "lastUpdated": P_CACHE_UPDATED_AT, } @@ -342,7 +342,7 @@ async def registry_search( sort: str = Query("name", description="Sort by: name, stars"), source: str = Query("", description="Filter by source: google, community, or empty for all"), ): - pool = _cache.values() + pool = P_CACHE.values() if source: pool = [s for s in pool if s.get("source") == source] @@ -387,7 +387,7 @@ async def registry_search( @mcp_registry.router.get("/detail/{server_name:path}") async def registry_detail(server_name: str): - srv = _cache.get(server_name) + srv = P_CACHE.get(server_name) if not srv: return {"error": "Server not found"}, 404 return {"server": srv} diff --git a/backend/apps/modes/modes.py b/backend/apps/modes/modes.py index 9cc897fc..499fee84 100644 --- a/backend/apps/modes/modes.py +++ b/backend/apps/modes/modes.py @@ -1,5 +1,6 @@ import os import logging +import json from contextlib import asynccontextmanager from fastapi import HTTPException from backend.config.Apps import SubApp @@ -18,10 +19,9 @@ async def modes_lifespan(): chat_path = os.path.join(DATA_DIR, "chat.json") if os.path.exists(chat_path): try: - import json as _json - with open(chat_path) as _f: - _data = _json.load(_f) - if _data.get("is_builtin") is True and _data.get("id") == "chat": + with open(chat_path) as f: + json_data = json.load(f) + if json_data.get("is_builtin") is True and json_data.get("id") == "chat": os.remove(chat_path) logger.info("Removed deprecated built-in chat.json (merged into ask)") except Exception: @@ -29,14 +29,14 @@ async def modes_lifespan(): for builtin in BUILTIN_MODES: path = os.path.join(DATA_DIR, f"{builtin.id}.json") if not os.path.exists(path): - _save(builtin) + p_save(builtin) yield modes = SubApp("modes", modes_lifespan) -def _load_all() -> list[Mode]: +def p_load_all() -> list[Mode]: result = [] if not os.path.exists(DATA_DIR): return result @@ -52,11 +52,11 @@ def _load_all() -> list[Mode]: return result -def _save(mode: Mode): +def p_save(mode: Mode): atomic_write_json(os.path.join(DATA_DIR, f"{mode.id}.json"), mode.model_dump()) -def _load(mode_id: str) -> Mode: +def p_load(mode_id: str) -> Mode: data = read_json_or_none(os.path.join(DATA_DIR, f"{mode_id}.json")) if data is None: raise HTTPException(status_code=404, detail="Mode not found") @@ -72,12 +72,12 @@ def load_mode(mode_id: str) -> Mode | None: @modes.router.get("/list") async def list_modes(): builtin_defaults = {m.id: m.model_dump() for m in BUILTIN_MODES} - return {"modes": [m.model_dump() for m in _load_all()], "builtin_defaults": builtin_defaults} + return {"modes": [m.model_dump() for m in p_load_all()], "builtin_defaults": builtin_defaults} @modes.router.get("/{mode_id}") async def get_mode(mode_id: str): - return _load(mode_id).model_dump() + return p_load(mode_id).model_dump() @modes.router.post("/create") @@ -93,16 +93,16 @@ async def create_mode(body: ModeCreate): default_folder=body.default_folder, is_builtin=False, ) - _save(mode) + p_save(mode) return {"ok": True, "mode": mode.model_dump()} @modes.router.put("/{mode_id}") async def update_mode(mode_id: str, body: ModeUpdate): - mode = _load(mode_id) + mode = p_load(mode_id) for k, v in body.model_dump(exclude_unset=True).items(): setattr(mode, k, v) - _save(mode) + p_save(mode) return {"ok": True, "mode": mode.model_dump()} @@ -112,13 +112,13 @@ async def reset_mode(mode_id: str): builtin = next((m for m in BUILTIN_MODES if m.id == mode_id), None) if not builtin: raise HTTPException(status_code=400, detail="Only built-in modes can be reset") - _save(builtin) + p_save(builtin) return {"ok": True, "mode": builtin.model_dump()} @modes.router.delete("/{mode_id}") async def delete_mode(mode_id: str): - mode = _load(mode_id) + mode = p_load(mode_id) if mode.is_builtin: raise HTTPException(status_code=403, detail="Cannot delete built-in modes") path = os.path.join(DATA_DIR, f"{mode_id}.json") diff --git a/backend/apps/nine_router/oauth.py b/backend/apps/nine_router/oauth.py index 55b7ba85..b21b0e93 100644 --- a/backend/apps/nine_router/oauth.py +++ b/backend/apps/nine_router/oauth.py @@ -11,7 +11,7 @@ import os import httpx from .process import NINE_ROUTER_API, NINE_ROUTER_PORT, NINE_ROUTER_V1 -from backend.apps.oauth_state import _pending_oauth, _mark_oauth_completed +from backend.apps.oauth_state import PENDING_OAUTH, mark_oauth_completed logger = logging.getLogger(__name__) @@ -22,9 +22,9 @@ logger = logging.getLogger(__name__) # 1455 that serves the same postMessage/BroadcastChannel/localStorage relay so # the frontend's existing popup + msgHandler flow works unchanged. -_CODEX_CALLBACK_PORT = 1455 -_CODEX_CALLBACK_PATH = "/auth/callback" -_CODEX_CALLBACK_HTML = b""" +P_CODEX_CALLBACK_PORT = 1455 +P_CODEX_CALLBACK_PATH = "/auth/callback" +P_CODEX_CALLBACK_HTML = b""" Authorization Complete """ -async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base_events.Server | None: +async def p_start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base_events.Server | None: """Spawn a one-shot HTTP listener on 127.0.0.1:1455 for the Codex OAuth callback. Serves GET /auth/callback with _CODEX_CALLBACK_HTML. After serving the @@ -80,7 +80,7 @@ async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base callback_served = asyncio.Event() - async def _handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter): + async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter): try: # Read the request line ("GET /auth/callback?... HTTP/1.1\r\n") raw_request_line = await asyncio.wait_for(reader.readline(), timeout=5.0) @@ -96,7 +96,7 @@ async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base path = parts[1] if len(parts) >= 2 else "" method = parts[0] if parts else "" - if method == "GET" and path.startswith(_CODEX_CALLBACK_PATH): + if method == "GET" and path.startswith(P_CODEX_CALLBACK_PATH): # Parse code/state out of the query string and exchange # server-side before serving the HTML. Duplicate exchanges # are harmless (single-use auth codes fail the second call, @@ -109,7 +109,7 @@ async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base code = (q.get("code") or [""])[0] state = (q.get("state") or [""])[0] if code and state: - pending = _pending_oauth.pop(state, None) + pending = PENDING_OAUTH.pop(state, None) if pending: try: await exchange_oauth( @@ -119,7 +119,7 @@ async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base pending["code_verifier"], state, ) - _mark_oauth_completed(state) + mark_oauth_completed(state) logger.info( f"Codex callback: server-side exchange succeeded for state {state[:8]}..." ) @@ -129,14 +129,14 @@ async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base # /agents/subscriptions/exchange still # has a shot. Safe because we only popped # it a moment ago. - _pending_oauth[state] = pending + PENDING_OAUTH[state] = pending logger.debug( f"Codex callback: server-side exchange failed ({e}); leaving for frontend retry" ) except Exception as e: logger.debug(f"Codex callback listener pre-exchange error: {e}") - body = _CODEX_CALLBACK_HTML + body = P_CODEX_CALLBACK_HTML response = ( b"HTTP/1.1 200 OK\r\n" b"Content-Type: text/html; charset=utf-8\r\n" @@ -165,17 +165,17 @@ async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base pass try: - server = await asyncio.start_server(_handle, "127.0.0.1", _CODEX_CALLBACK_PORT) + server = await asyncio.start_server(handle, "127.0.0.1", P_CODEX_CALLBACK_PORT) except OSError as e: # Port already in use; probably another Codex connect attempt still # running, or an actual Codex CLI process holding 1455. Log and bail. logger.warning( - f"Could not start Codex callback listener on port {_CODEX_CALLBACK_PORT}: {e}. " + f"Could not start Codex callback listener on port {P_CODEX_CALLBACK_PORT}: {e}. " "If another connection attempt is in progress, wait for it to finish or time out." ) return None - async def _lifecycle(): + async def lifecycle(): try: await asyncio.wait_for(callback_served.wait(), timeout=timeout) # Give the served HTML a moment to run its JS (postMessage + @@ -193,8 +193,8 @@ async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base except Exception: pass - asyncio.create_task(_lifecycle()) - logger.info(f"Started Codex callback listener on http://localhost:{_CODEX_CALLBACK_PORT}{_CODEX_CALLBACK_PATH}") + asyncio.create_task(lifecycle()) + logger.info(f"Started Codex callback listener on http://localhost:{P_CODEX_CALLBACK_PORT}{P_CODEX_CALLBACK_PATH}") return server @@ -207,14 +207,14 @@ async def _start_codex_callback_listener(timeout: float = 300.0) -> asyncio.base # embed detection + regional checks); system browser surfaces the real error. # The callback for gemini-cli/antigravity lands on /api/subscriptions/callback # and runs the exchange server-side; codex uses its fixed 1455 listener. -_EXTERNAL_BROWSER_PROVIDERS: set[str] = {"gemini-cli", "antigravity", "codex"} +P_EXTERNAL_BROWSER_PROVIDERS: set[str] = {"gemini-cli", "antigravity", "codex"} -def _should_use_external_browser(provider: str) -> bool: - return provider in _EXTERNAL_BROWSER_PROVIDERS +def p_should_use_external_browser(provider: str) -> bool: + return provider in P_EXTERNAL_BROWSER_PROVIDERS -def _backend_port() -> int: +def p_backend_port() -> int: """Best-effort lookup of the OpenSwarm backend HTTP port. Falls back to 8324 (the default in backend/main.py) if OPENSWARM_PORT @@ -228,7 +228,7 @@ def _backend_port() -> int: return 8324 -def _callback_uri_for_provider(provider: str) -> str: +def p_callback_uri_for_provider(provider: str) -> str: """Return the redirect URI to pass to 9Router's authorize endpoint. Most providers accept 9Router's built-in callback page at port 20128. @@ -243,9 +243,9 @@ def _callback_uri_for_provider(provider: str) -> str: on OpenSwarm's port rather than 9Router's. """ if provider == "codex": - return f"http://localhost:{_CODEX_CALLBACK_PORT}{_CODEX_CALLBACK_PATH}" - if provider in _EXTERNAL_BROWSER_PROVIDERS: - return f"http://localhost:{_backend_port()}/api/subscriptions/callback" + return f"http://localhost:{P_CODEX_CALLBACK_PORT}{P_CODEX_CALLBACK_PATH}" + if provider in P_EXTERNAL_BROWSER_PROVIDERS: + return f"http://localhost:{p_backend_port()}/api/subscriptions/callback" return f"http://localhost:{NINE_ROUTER_PORT}/callback" @@ -271,9 +271,9 @@ async def start_oauth(provider: str) -> dict: except Exception: pass - callback_url = _callback_uri_for_provider(provider) + callback_url = p_callback_uri_for_provider(provider) if provider == "codex": - await _start_codex_callback_listener() + await p_start_codex_callback_listener() r = await client.get( f"{NINE_ROUTER_API}/oauth/{provider}/authorize", @@ -287,7 +287,7 @@ async def start_oauth(provider: str) -> dict: "code_verifier": data.get("codeVerifier", ""), "state": data.get("state", ""), "redirect_uri": callback_url, - "use_external_browser": _should_use_external_browser(provider), + "use_external_browser": p_should_use_external_browser(provider), } diff --git a/backend/apps/nine_router/process.py b/backend/apps/nine_router/process.py index a34d5c44..77ff59ed 100644 --- a/backend/apps/nine_router/process.py +++ b/backend/apps/nine_router/process.py @@ -23,9 +23,9 @@ import httpx logger = logging.getLogger(__name__) NINE_ROUTER_PORT = 20128 -NINE_ROUTER_URL = f"http://localhost:{NINE_ROUTER_PORT}" -NINE_ROUTER_API = f"{NINE_ROUTER_URL}/api" -NINE_ROUTER_V1 = f"{NINE_ROUTER_URL}/v1" +P_NINE_ROUTER_URL = f"http://localhost:{NINE_ROUTER_PORT}" +NINE_ROUTER_API = f"{P_NINE_ROUTER_URL}/api" +NINE_ROUTER_V1 = f"{P_NINE_ROUTER_URL}/v1" # Pinned 9router npm package version. Stays at 0.3.60. # @@ -56,7 +56,7 @@ NINE_ROUTER_V1 = f"{NINE_ROUTER_URL}/v1" # `openai-compatible` provider-node that honors `baseUrl`. NINE_ROUTER_NPM_VERSION = "0.3.60" -_process: subprocess.Popen | None = None +P_PROCESS: subprocess.Popen | None = None # Short TTL cache for positive is_running() results. The probe is a sync # httpx.get that blocks the event loop, and under load (9Router busy @@ -65,46 +65,46 @@ _process: subprocess.Popen | None = None # negatives without masking a real crash for more than _IS_RUNNING_TTL seconds. # Negative results are NOT cached so startup detection in ensure_running() # remains correct. -_IS_RUNNING_TTL = 10.0 -_is_running_last_ok: float = 0.0 +P_IS_RUNNING_TTL = 10.0 +P_IS_RUNNING_LAST_OK: float = 0.0 def is_running() -> bool: """Check if 9Router is running.""" - global _is_running_last_ok + global P_IS_RUNNING_LAST_OK now = time.monotonic() - if now - _is_running_last_ok < _IS_RUNNING_TTL: + if now - P_IS_RUNNING_LAST_OK < P_IS_RUNNING_TTL: return True try: r = httpx.get(f"{NINE_ROUTER_V1}/models", timeout=2.0) if r.status_code == 200: - _is_running_last_ok = now + P_IS_RUNNING_LAST_OK = now return True return False except Exception: return False -def _find_9router_dir() -> str | None: +def p_find_9router_dir() -> str | None: """Locate the bundled 9Router directory (works in both dev and packaged mode).""" - _is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1" + is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1" - if _is_packaged: - _resources = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) - _candidate = os.path.join(_resources, "router") - if os.path.isdir(_candidate): - return _candidate + if is_packaged: + resources = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) + candidate = os.path.join(resources, "router") + if os.path.isdir(candidate): + return candidate else: - _backend_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - _project_root = os.path.dirname(_backend_dir) - _candidate = os.path.join(_project_root, "router") - if os.path.isdir(_candidate): - return _candidate + backend_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + project_root = os.path.dirname(backend_dir) + candidate = os.path.join(project_root, "router") + if os.path.isdir(candidate): + return candidate return None -def _gpt5_patch_path() -> str | None: +def p_gpt5_patch_path() -> str | None: """Absolute path to backend/apps/agents/9router_gpt5_patch.js, used as `node --require ` when spawning 9router. @@ -127,7 +127,7 @@ def _gpt5_patch_path() -> str | None: return candidate if os.path.exists(candidate) else None -def _find_node() -> str | None: +def p_find_node() -> str | None: """Find a Node.js binary (works in both dev and packaged mode). Priority order: @@ -157,7 +157,7 @@ def _find_node() -> str | None: return None -def _dev_router_cache_dir() -> str: +def p_dev_router_cache_dir() -> str: """Cache dir for the npm 9router package used in dev mode. Pinned per version so bumping NINE_ROUTER_NPM_VERSION triggers a fresh @@ -169,7 +169,7 @@ def _dev_router_cache_dir() -> str: return os.path.join(base, "openswarm-router", NINE_ROUTER_NPM_VERSION) -def _ensure_router_cached() -> str | None: +def p_ensure_router_cached() -> str | None: """Ensure the npm 9router package is installed in the dev cache. Returns the absolute path to `app/server.js` on success, or None if @@ -181,7 +181,7 @@ def _ensure_router_cached() -> str | None: no update-check spinner, and no accidental-quit foot-gun when a non-developer right-clicks the "9" tray icon and picks Quit. """ - cache_dir = _dev_router_cache_dir() + cache_dir = p_dev_router_cache_dir() server_js = os.path.join(cache_dir, "node_modules", "9router", "app", "server.js") if os.path.exists(server_js): return server_js @@ -223,22 +223,21 @@ def _ensure_router_cached() -> str | None: async def ensure_running(): """Start 9Router if not already running.""" - global _process - _is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1" + global P_PROCESS + is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1" if is_running(): # In dev mode, kill stale standalone servers (from previous builds) # so we can start `next dev` which always uses latest source code - if not _is_packaged: - import subprocess as _sp + if not is_packaged: try: - result = _sp.run( + result = subprocess.run( ["pgrep", "-f", "next-server"], capture_output=True, text=True, timeout=3, ) if result.stdout.strip(): logger.info("Dev mode: killing stale standalone 9Router to use next dev instead") - _sp.run(["pkill", "-f", "next-server"], timeout=5) + subprocess.run(["pkill", "-f", "next-server"], timeout=5) await asyncio.sleep(2) else: logger.info("9Router already running on port %d", NINE_ROUTER_PORT) @@ -249,28 +248,28 @@ async def ensure_running(): else: logger.info("9Router already running on port %d", NINE_ROUTER_PORT) return - _9router_dir = _find_9router_dir() + nine_router_dir = p_find_9router_dir() - if _is_packaged and _9router_dir: + if is_packaged and nine_router_dir: # Packaged mode; run the pre-built standalone server staged at # /router/server.js by scripts/fetch-router.sh at build time. - standalone_server = os.path.join(_9router_dir, "server.js") + standalone_server = os.path.join(nine_router_dir, "server.js") if not os.path.exists(standalone_server): - standalone_server = os.path.join(_9router_dir, ".next", "standalone", "server.js") + standalone_server = os.path.join(nine_router_dir, ".next", "standalone", "server.js") if not os.path.exists(standalone_server): - logger.warning("9Router standalone build not found in %s", _9router_dir) + logger.warning("9Router standalone build not found in %s", nine_router_dir) return - node = _find_node() + node = p_find_node() if not node: logger.warning("Node.js not found; cannot start 9Router in packaged mode.") return logger.info("Starting 9Router (production) on port %d...", NINE_ROUTER_PORT) cmd = [node] - _patch = _gpt5_patch_path() - if _patch: - cmd += ["--require", _patch] + patch = p_gpt5_patch_path() + if patch: + cmd += ["--require", patch] cmd.append(standalone_server) cwd = os.path.dirname(standalone_server) env = {**os.environ, "PORT": str(NINE_ROUTER_PORT), "NODE_ENV": "production"} @@ -284,11 +283,11 @@ async def ensure_running(): # avoids its menu-bar tray icon (which users confusingly quit, # silently killing their subscription routing), its update-check # spinner, and the interactive TUI. - cached_server = _ensure_router_cached() + cached_server = p_ensure_router_cached() if not cached_server: return - node = _find_node() + node = p_find_node() if not node: logger.warning("Node.js not found; cannot start 9Router in dev mode.") return @@ -298,9 +297,9 @@ async def ensure_running(): NINE_ROUTER_NPM_VERSION, NINE_ROUTER_PORT, ) cmd = [node] - _patch = _gpt5_patch_path() - if _patch: - cmd += ["--require", _patch] + patch = p_gpt5_patch_path() + if patch: + cmd += ["--require", patch] cmd.append(cached_server) cwd = os.path.dirname(cached_server) env = {**os.environ, "PORT": str(NINE_ROUTER_PORT), "NODE_ENV": "production"} @@ -311,29 +310,29 @@ async def ensure_running(): # before launching the backend; output will then be appended to # backend/data/9router.log line-buffered, which can be `tail -f`'d. if os.environ.get("OPENSWARM_DEBUG_9ROUTER"): - _log_path = os.path.join( + log_path = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "data", "9router.log", ) - os.makedirs(os.path.dirname(_log_path), exist_ok=True) - _stdout = open(_log_path, "a", buffering=1) # line-buffered - _stderr = subprocess.STDOUT - logger.info(f"9Router debug logging enabled → {_log_path}") + os.makedirs(os.path.dirname(log_path), exist_ok=True) + stdout = open(log_path, "a", buffering=1) # line-buffered + stderr = subprocess.STDOUT + logger.info(f"9Router debug logging enabled → {log_path}") else: - _stdout = subprocess.DEVNULL - _stderr = subprocess.DEVNULL + stdout = subprocess.DEVNULL + stderr = subprocess.DEVNULL try: - _process = subprocess.Popen( + P_PROCESS = subprocess.Popen( cmd, cwd=cwd, - stdout=_stdout, - stderr=_stderr, + stdout=stdout, + stderr=stderr, env=env, ) - timeout = 20 if _is_packaged else 30 + timeout = 20 if is_packaged else 30 for _ in range(timeout * 2): await asyncio.sleep(0.5) if is_running(): @@ -347,17 +346,17 @@ async def ensure_running(): def stop(): """Stop the 9Router subprocess.""" - global _process - if _process: + global P_PROCESS + if P_PROCESS: try: - _process.terminate() - _process.wait(timeout=5) + P_PROCESS.terminate() + P_PROCESS.wait(timeout=5) except Exception: try: - _process.kill() + P_PROCESS.kill() except Exception: pass - _process = None + P_PROCESS = None logger.info("9Router stopped") diff --git a/backend/apps/nine_router/sync.py b/backend/apps/nine_router/sync.py index fe32ddab..312b2e02 100644 --- a/backend/apps/nine_router/sync.py +++ b/backend/apps/nine_router/sync.py @@ -12,12 +12,12 @@ from .process import NINE_ROUTER_API logger = logging.getLogger(__name__) -def _nr(): +def nr(): """The package module. is_running / get_providers / httpx are read off it at call time so tests that patch `backend.apps.nine_router.` still take effect after the split (they used to all live on one module).""" - from backend.apps import nine_router - return nine_router + from backend.apps.nine_router import process + return process # API-key auth (provider="gemini", authType="apikey") and OAuth hit different # Google quotas: OAuth uses the Code Assist free tier (aggressively rate-limited; @@ -35,9 +35,9 @@ NINE_ROUTER_CLAUDE_PRO_NAME = "OpenSwarm Pro (OpenSwarm-managed)" NINE_ROUTER_OPENAI_KEYED_PREFIX = "cp-openai" -async def _find_keyed_connection(provider: str, name: str) -> dict | None: +async def find_keyed_connection(provider: str, name: str) -> dict | None: """Return the 9Router connection we manage for this provider, if any.""" - conns = await _nr().get_providers() + conns = await nr().get_providers() if not isinstance(conns, list): return None for c in conns: @@ -51,7 +51,7 @@ async def _find_keyed_connection(provider: str, name: str) -> dict | None: return None -async def _sync_apikey_provider( +async def p_sync_apikey_provider( provider: str, api_key: str | None, name: str, @@ -59,12 +59,12 @@ async def _sync_apikey_provider( label: str, ) -> None: """Create/update/delete an OpenSwarm-managed apikey connection. Silent if 9Router is down.""" - if not _nr().is_running(): + if not nr().is_running(): return - existing = await _find_keyed_connection(provider, name) + existing = await find_keyed_connection(provider, name) try: - async with _nr().httpx.AsyncClient(timeout=5.0) as client: + async with nr().httpx.AsyncClient(timeout=5.0) as client: if api_key: payload = { "provider": provider, @@ -100,7 +100,7 @@ async def _sync_apikey_provider( async def sync_gemini_api_key(api_key: str | None) -> None: """Mirror google_api_key into 9Router; bypasses Code Assist's tight quota.""" - await _sync_apikey_provider( + await p_sync_apikey_provider( "gemini", api_key, NINE_ROUTER_KEYED_NAME, label="Gemini" ) @@ -128,12 +128,12 @@ async def sync_openai_api_key(api_key: str | None) -> None: above) so 9Router's translator dispatches to this provider-node instead of the built-in `openai` provider. """ - from .sync_custom import _sync_openai_compat_node - await _sync_openai_compat_node(api_key) + from backend.apps.nine_router.sync_custom import sync_openai_compat_node + await sync_openai_compat_node(api_key) async def sync_openrouter_api_key(api_key: str | None) -> None: """Mirror openrouter_api_key into 9Router; supplies bearer for openrouter/ routes.""" - await _sync_apikey_provider( + await p_sync_apikey_provider( "openrouter", api_key, NINE_ROUTER_OPENROUTER_KEYED_NAME, label="OpenRouter" ) diff --git a/backend/apps/nine_router/sync_custom.py b/backend/apps/nine_router/sync_custom.py index 7696270c..fddb2a1f 100644 --- a/backend/apps/nine_router/sync_custom.py +++ b/backend/apps/nine_router/sync_custom.py @@ -10,12 +10,12 @@ spawns the subprocess (that's process.py's job). import logging -from .process import NINE_ROUTER_API -from .sync import ( +from backend.apps.nine_router.process import NINE_ROUTER_API +from backend.apps.nine_router.sync import ( NINE_ROUTER_CLAUDE_PRO_NAME, NINE_ROUTER_OPENAI_KEYED_PREFIX, - _find_keyed_connection, - _nr, + find_keyed_connection, + nr, ) logger = logging.getLogger(__name__) @@ -25,18 +25,18 @@ logger = logging.getLogger(__name__) NINE_ROUTER_CUSTOM_NAME_SUFFIX = " (OpenSwarm-managed)" -async def _sync_openai_compat_node(api_key: str | None) -> None: +async def sync_openai_compat_node(api_key: str | None) -> None: """Create / update / delete the openai-compatible node + connection pair we use to ferry OpenAI requests through openai-passthrough.""" - if not _nr().is_running(): + if not nr().is_running(): return - import os as _os - port = _os.environ.get("OPENSWARM_PORT", "8324") + import os + port = os.environ.get("OPENSWARM_PORT", "8324") base_url = f"http://127.0.0.1:{port}/api/openai-passthrough/v1" managed_name = f"OpenAI{NINE_ROUTER_CUSTOM_NAME_SUFFIX}" try: - async with _nr().httpx.AsyncClient(timeout=5.0) as client: + async with nr().httpx.AsyncClient(timeout=5.0) as client: r = await client.get(f"{NINE_ROUTER_API}/provider-nodes") existing_nodes = (r.json().get("nodes") if r.status_code == 200 else []) or [] except Exception as e: @@ -50,7 +50,7 @@ async def _sync_openai_compat_node(api_key: str | None) -> None: if not api_key: if existing_node: try: - async with _nr().httpx.AsyncClient(timeout=5.0) as client: + async with nr().httpx.AsyncClient(timeout=5.0) as client: await client.delete(f"{NINE_ROUTER_API}/provider-nodes/{existing_node['id']}") logger.info("9Router: removed OpenAI compat node (key cleared)") except Exception as e: @@ -66,7 +66,7 @@ async def _sync_openai_compat_node(api_key: str | None) -> None: } node_id: str | None = existing_node.get("id") if existing_node else None try: - async with _nr().httpx.AsyncClient(timeout=5.0) as client: + async with nr().httpx.AsyncClient(timeout=5.0) as client: if existing_node: await client.put( f"{NINE_ROUTER_API}/provider-nodes/{existing_node['id']}", @@ -92,7 +92,7 @@ async def _sync_openai_compat_node(api_key: str | None) -> None: return try: - existing_conn = await _find_keyed_connection(node_id, managed_name) + existing_conn = await find_keyed_connection(node_id, managed_name) conn_payload = { "provider": node_id, "authType": "apikey", @@ -100,7 +100,7 @@ async def _sync_openai_compat_node(api_key: str | None) -> None: "apiKey": api_key, "priority": 0, } - async with _nr().httpx.AsyncClient(timeout=5.0) as client: + async with nr().httpx.AsyncClient(timeout=5.0) as client: if existing_conn: await client.patch( f"{NINE_ROUTER_API}/providers/{existing_conn['id']}", @@ -117,7 +117,7 @@ async def _sync_openai_compat_node(api_key: str | None) -> None: logger.warning(f"9Router OpenAI compat connection sync failed: {e}") -def _custom_provider_slug(name: str) -> str: +def p_custom_provider_slug(name: str) -> str: """Slugify a user-supplied custom-provider name for use as a 9Router prefix. Always returns a non-empty alnum-and-dash string.""" import re @@ -157,11 +157,11 @@ async def sync_custom_providers(providers: list) -> None: no longer in `providers` is deleted (which cascades to its connection). Silent no-op when 9Router isn't running. """ - if not _nr().is_running(): + if not nr().is_running(): return try: - async with _nr().httpx.AsyncClient(timeout=5.0) as client: + async with nr().httpx.AsyncClient(timeout=5.0) as client: r = await client.get(f"{NINE_ROUTER_API}/provider-nodes") existing_nodes = (r.json().get("nodes") if r.status_code == 200 else []) or [] except Exception as e: @@ -187,7 +187,7 @@ async def sync_custom_providers(providers: list) -> None: # Bearer header even with auth disabled. Substitute a placeholder; real # auth deployments always have api_key set. api_key = api_key.strip() or "no-auth-required" - slug = _custom_provider_slug(name) + slug = p_custom_provider_slug(name) prefix = f"cp-{slug}" seen_prefixes.add(prefix) managed_name = f"{name.strip()}{NINE_ROUTER_CUSTOM_NAME_SUFFIX}" @@ -201,7 +201,7 @@ async def sync_custom_providers(providers: list) -> None: "type": "openai-compatible", } try: - async with _nr().httpx.AsyncClient(timeout=5.0) as client: + async with nr().httpx.AsyncClient(timeout=5.0) as client: if node: await client.put( f"{NINE_ROUTER_API}/provider-nodes/{node['id']}", @@ -228,7 +228,7 @@ async def sync_custom_providers(providers: list) -> None: continue try: - existing_conn = await _find_keyed_connection(node_id, managed_name) + existing_conn = await find_keyed_connection(node_id, managed_name) conn_payload = { "provider": node_id, "authType": "apikey", @@ -236,7 +236,7 @@ async def sync_custom_providers(providers: list) -> None: "apiKey": api_key, "priority": 0, } - async with _nr().httpx.AsyncClient(timeout=5.0) as client: + async with nr().httpx.AsyncClient(timeout=5.0) as client: if existing_conn: await client.patch( f"{NINE_ROUTER_API}/providers/{existing_conn['id']}", @@ -259,7 +259,7 @@ async def sync_custom_providers(providers: list) -> None: if prefix in seen_prefixes: continue try: - async with _nr().httpx.AsyncClient(timeout=5.0) as client: + async with nr().httpx.AsyncClient(timeout=5.0) as client: await client.delete(f"{NINE_ROUTER_API}/provider-nodes/{node['id']}") logger.info(f"9Router: removed orphaned custom node {prefix}") except Exception as e: @@ -280,15 +280,15 @@ async def sync_openswarm_pro_as_claude(bearer_token: str | None, proxy_url: str the OpenSwarm-Pro-backed Claude connection and routes the search call through our cloud; same quota the user's Pro subscription already covers, no extra cost.""" - if not _nr().is_running(): + if not nr().is_running(): return # 9Router's POST /api/providers only accepts direct-API provider ids # for apikey auth; `claude` is the subscription/IDE id, `anthropic` # is the direct-API id. Use `anthropic`. - existing = await _find_keyed_connection("anthropic", NINE_ROUTER_CLAUDE_PRO_NAME) + existing = await find_keyed_connection("anthropic", NINE_ROUTER_CLAUDE_PRO_NAME) try: - async with _nr().httpx.AsyncClient(timeout=5.0) as client: + async with nr().httpx.AsyncClient(timeout=5.0) as client: if bearer_token and proxy_url: payload = { "provider": "anthropic", diff --git a/backend/apps/oauth_state.py b/backend/apps/oauth_state.py index 26e3dca2..dd488dbb 100644 --- a/backend/apps/oauth_state.py +++ b/backend/apps/oauth_state.py @@ -1,18 +1,18 @@ # In-memory store for pending OAuth flows (state -> {provider, code_verifier, redirect_uri}) -_pending_oauth: dict[str, dict] = {} +PENDING_OAUTH: dict[str, dict] = {} # Recently-completed OAuth states so the /api/subscriptions/callback handler # can distinguish a legitimate duplicate callback (browser prefetch, refresh, # or Google redirect retry after a slow first response) from a truly stale # request. Bounded FIFO, drops the oldest entries once it grows past # _MAX_COMPLETED_OAUTH so it can't leak memory. -_completed_oauth: list[str] = [] -_MAX_COMPLETED_OAUTH = 64 +COMPLETED_OAUTH: list[str] = [] +P_MAX_COMPLETED_OAUTH = 64 -def _mark_oauth_completed(state: str) -> None: - if state in _completed_oauth: +def mark_oauth_completed(state: str) -> None: + if state in COMPLETED_OAUTH: return - _completed_oauth.append(state) + COMPLETED_OAUTH.append(state) # Trim head if we've outgrown the bound - while len(_completed_oauth) > _MAX_COMPLETED_OAUTH: - _completed_oauth.pop(0) + while len(COMPLETED_OAUTH) > P_MAX_COMPLETED_OAUTH: + COMPLETED_OAUTH.pop(0) diff --git a/backend/apps/outputs/executor.py b/backend/apps/outputs/executor.py index ba0121ff..8393fa4c 100644 --- a/backend/apps/outputs/executor.py +++ b/backend/apps/outputs/executor.py @@ -17,7 +17,7 @@ TIMEOUT_SECONDS = 30 # pairs with cwd=tempdir + minimal env so the blast radius is small even if # a payload slips past. Keep this list to "data shaping" libraries; no I/O, # no networking, no subprocess. -_ALLOWED_MODULES = frozenset({ +P_ALLOWED_MODULES = frozenset({ "json", "math", "re", "datetime", "collections", "itertools", "functools", "statistics", "decimal", "fractions", "random", "string", "textwrap", "unicodedata", "csv", "copy", "enum", @@ -29,7 +29,7 @@ _ALLOWED_MODULES = frozenset({ # calls (e.g. `eval(...)`) are caught here. Attribute-style calls # (`__builtins__.eval(...)`) are blocked by the preamble's `delattr` loop in # the subprocess. -_BLOCKED_BUILTINS = frozenset({ +P_BLOCKED_BUILTINS = frozenset({ "exec", "eval", "compile", "__import__", "open", "input", "breakpoint", "exit", "quit", }) @@ -62,7 +62,7 @@ def get_code_warnings(code: str) -> list[str]: if isinstance(node, ast.Import): for alias in node.names: root = alias.name.split(".")[0] - if root not in _ALLOWED_MODULES: + if root not in P_ALLOWED_MODULES: msg = f"Imports '{alias.name}' (outside the safe-data-shaping allowlist)" if msg not in seen: seen.add(msg) @@ -70,13 +70,13 @@ def get_code_warnings(code: str) -> list[str]: elif isinstance(node, ast.ImportFrom): if node.module: root = node.module.split(".")[0] - if root not in _ALLOWED_MODULES: + if root not in P_ALLOWED_MODULES: msg = f"Imports from '{node.module}' (outside the safe-data-shaping allowlist)" if msg not in seen: seen.add(msg) warnings.append(msg) elif isinstance(node, ast.Call): - if isinstance(node.func, ast.Name) and node.func.id in _BLOCKED_BUILTINS: + if isinstance(node.func, ast.Name) and node.func.id in P_BLOCKED_BUILTINS: msg = f"Calls builtin '{node.func.id}()' which can escape the sandbox" if msg not in seen: seen.add(msg) @@ -84,7 +84,7 @@ def get_code_warnings(code: str) -> list[str]: return warnings -def _validate_code_safety(code: str) -> None: +def p_validate_code_safety(code: str) -> None: """Raise UnsafeCodeError on the first AST-visible risk. Thin wrapper around get_code_warnings for callers that want the strict-reject behavior (the default `execute_backend_code` path). Callers that want @@ -99,7 +99,7 @@ def _validate_code_safety(code: str) -> None: # Env vars we always scrub from the subprocess, regardless of strict-vs-force. # These are the keys an attacker would actually want; install token, provider # API keys, cloud credentials. Everything else is local-machine convenience. -_SCRUBBED_ENV_KEYS = frozenset({ +P_SCRUBBED_ENV_KEYS = frozenset({ "OPENSWARM_AUTH_TOKEN", "ANTHROPIC_API_KEY", "OPENAI_API_KEY", @@ -116,7 +116,7 @@ _SCRUBBED_ENV_KEYS = frozenset({ }) -def _minimal_env(force: bool = False) -> dict: +def p_minimal_env(force: bool = False) -> dict: """Build the env for the executor subprocess. Strict mode (force=False): only language essentials. AST-validated code @@ -130,11 +130,11 @@ def _minimal_env(force: bool = False) -> dict: minus credentials, so an `open(os.path.expanduser("~/data.csv"))` actually works instead of silently misbehaving. - Both modes scrub _SCRUBBED_ENV_KEYS so even force-mode code never + Both modes scrub P_SCRUBBED_ENV_KEYS so even force-mode code never sees the install token or provider API keys. """ if force: - env = {k: v for k, v in os.environ.items() if k not in _SCRUBBED_ENV_KEYS} + env = {k: v for k, v in os.environ.items() if k not in P_SCRUBBED_ENV_KEYS} env["PYTHONDONTWRITEBYTECODE"] = "1" # Force UTF-8 even if the parent somehow lacked it (dev mode where # Electron didn't inject PYTHONUTF8). Without this, a child reading @@ -192,7 +192,7 @@ async def execute_backend_code( """ if not skip_validation: - _validate_code_safety(code) + p_validate_code_safety(code) preamble = ( "import json, sys, io, builtins\n" @@ -229,7 +229,7 @@ async def execute_backend_code( stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=workdir, - env=_minimal_env(force=skip_validation), + env=p_minimal_env(force=skip_validation), ) try: diff --git a/backend/main.py b/backend/main.py index dc496659..1a2323df 100644 --- a/backend/main.py +++ b/backend/main.py @@ -22,9 +22,9 @@ from fastapi.responses import JSONResponse, HTMLResponse from fastapi import Request from backend.apps.oauth_state import ( - _pending_oauth, - _completed_oauth, - _mark_oauth_completed, + PENDING_OAUTH, + COMPLETED_OAUTH, + mark_oauth_completed, ) from backend.config.Apps import MainApp from backend.apps.health.health import health @@ -435,7 +435,7 @@ async def browser_command(request: Request): @app.get("/api/subscriptions/pending/{state}") async def subscriptions_pending(state: str): """Return pending OAuth data for a state param. Called by 9Router's callback page.""" - pending = _pending_oauth.get(state) + pending = PENDING_OAUTH.get(state) if not pending: return JSONResponse({"error": "not found"}, status_code=404, headers={"Access-Control-Allow-Origin": "*"}) @@ -483,12 +483,12 @@ async def subscriptions_callback(request: Request): desc = html.escape(request.query_params.get("error_description", error)) return HTMLResponse(f'

Authorization failed

{desc}

') - pending = _pending_oauth.pop(state, None) + pending = PENDING_OAUTH.pop(state, None) if not pending: # Either a duplicate callback for a state we've already exchanged, # or a truly stale state. Duplicates are the expected case: # Chrome's prefetcher and some extensions speculatively GET URLs. - if state and state in _completed_oauth: + if state and state in COMPLETED_OAUTH: logger.info(f"Duplicate OAuth callback for state {state[:8]}... (already completed)") return HTMLResponse(_SUCCESS_HTML) logger.warning(f"OAuth callback with unknown state {state[:8] if state else '(empty)'}...") @@ -506,7 +506,7 @@ async def subscriptions_callback(request: Request): safe_e = html.escape(str(e)) return HTMLResponse(f'

Connection failed

{safe_e}

') - _mark_oauth_completed(state) + mark_oauth_completed(state) logger.info(f"OAuth exchange succeeded for provider={pending.get('provider')}") return HTMLResponse(_SUCCESS_HTML)