WEBDIENSTE/openswarm
1
0
Fork 0
mirror of https://github.com/openswarm-ai/openswarm.git synced 2026-09-10 11:47:43 +02:00
Code Issues Packages Projects Releases Wiki Activity

[eric] apps: leading-_ -> p_ for function-local vars/args/nested-fns across 53 apps modules (scope-aware AST rename, module symbols untouched, p-private 0)

Browse Source
This commit is contained in:
ciregenz
2026-06-23 21:11:44 -07:00
parent 2e4bcd9e89
commit 798448acca
34 changed files with 192 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/agents/agents.py
+15 -15
View File
@@ -78,7 +78,7 @@ async def send_message(session_id: str, body: dict):
from backend.apps.agents.core.mcp_preflight import run_preflight
from backend.apps.agents.core.ws_manager import ws_manager as _ws
async def _emit_preflight():
async def p_emit_preflight():
try:
result = await run_preflight(prompt, task_id=session_id)
if result.get("suggestions") or result.get("is_vague"):
@@ -91,7 +91,7 @@ async def send_message(session_id: str, body: dict):
pass
import asyncio as _asyncio
_asyncio.create_task(_emit_preflight())
_asyncio.create_task(p_emit_preflight())
except Exception:
pass
@@ -568,17 +568,17 @@ async def list_models():
conns = await _9r_providers()
raw_providers = {c.get("provider", "") for c in conns if c.get("isActive") or c.get("testStatus") == "active"}
# 9Router uses "claude"; our models use api="anthropic". Map across.
_9R_TO_API = {
p_9R_TO_API = {
"claude": "anthropic",
"codex": "codex",
"gemini-cli": "gemini-cli",
"antigravity": "gemini-cli", # AG = same Gemini models, separate OAuth.
}
connected = raw_providers | {_9R_TO_API.get(p, p) for p in raw_providers}
connected = raw_providers | {p_9R_TO_API.get(p, p) for p in raw_providers}
except Exception as e:
logger.debug(f"Failed to fetch 9Router providers: {e}")
def _serialize(models: list[dict]) -> list[dict]:
def p_serialize(models: list[dict]) -> list[dict]:
# Tiers describe the model; billing_kind describes the wallet. Pricing shown only for paid.
from backend.apps.agents.providers.registry import (
COST_PER_1M_TOKENS,
@@ -588,8 +588,8 @@ async def list_models():
out = []
for m in models:
input_cost = output_cost = 0.0
for (_p, _v), rates in COST_PER_1M_TOKENS.items():
if _v == m["value"]:
for (p_p, p_v), rates in COST_PER_1M_TOKENS.items():
if p_v == m["value"]:
input_cost, output_cost = rates
break
api = m.get("api", "")
@@ -634,16 +634,16 @@ async def list_models():
# Pro mode splits into Pro proxy + Anthropic alternates; own-key collapses to one adaptive group.
notes: list[dict] = []
if is_openswarm_pro:
result["OpenSwarm Pro"] = _serialize(adaptive)
result["OpenSwarm Pro"] = p_serialize(adaptive)
anth_alternates: list[dict] = []
if has_claude_sub:
anth_alternates += cc_variants
if has_api_key:
anth_alternates += api_variants
if anth_alternates:
result["Anthropic"] = _serialize(anth_alternates)
result["Anthropic"] = p_serialize(anth_alternates)
elif has_api_key or has_claude_sub:
rows = _serialize(adaptive)
rows = p_serialize(adaptive)
# When an Anthropic key is set, these adaptive rows run on it: own-key routing prefers the
# user's key over any sub (agent_manager + anthropic_proxy._pick_upstream), so it holds even
# with a Claude sub connected. Label + bucket as API key (not 9router-state dependent).
@@ -657,7 +657,7 @@ async def list_models():
# on our pinned 9Router) have no adaptive twin to relabel, so add them or they vanish.
adaptive_ids = {m.get("model_id") for m in adaptive}
api_only = [m for m in api_variants if m.get("model_id") not in adaptive_ids]
rows = _serialize(api_only) + rows
rows = p_serialize(api_only) + rows
elif has_claude_sub:
# Only a sub: the adaptive rows route through 9router's cc/ lane, so they're covered
# by the subscription, not pay-per-use.
@@ -666,12 +666,12 @@ async def list_models():
# Sub-only models with no adaptive twin (Fable 5) won't ride the relabeled rows, so add their cc/ entry.
adaptive_ids = {m.get("model_id") for m in adaptive}
cc_only = [m for m in cc_variants if m.get("model_id") not in adaptive_ids]
rows = _serialize(cc_only) + rows
rows = p_serialize(cc_only) + rows
# With BOTH a key and a sub the adaptive rows above run on the key, so also surface the
# subscription (cc) variants; they route via 9router's cc/ lane and stay selectable, the
# way OpenAI/Gemini show both a subscription row and an API-key row.
if has_api_key and has_claude_sub:
rows += _serialize(cc_variants)
rows += p_serialize(cc_variants)
result["Anthropic"] = rows
has_openai_key = bool(getattr(settings, "openai_api_key", None))
@@ -698,8 +698,8 @@ async def list_models():
if not nine_router_up or api not in connected:
continue
in_cost = out_cost = 0.0
for (_p, _v), rates in _CPM.items():
if _v == m["value"]:
for (p_p, p_v), rates in _CPM.items():
if p_v == m["value"]:
in_cost, out_cost = rates
break
billing_kind = _cbk_native(
backend/apps/agents/browser/browser_metrics.py
+2 -2
View File
@@ -209,13 +209,13 @@ def p_maybe_self_audit() -> None:
if p_task_count % P_AUDIT_EVERY_N != 0:
return
def _run():
def p_run():
try:
from backend.apps.agents.browser import browser_self_audit
browser_self_audit.run_and_write()
except Exception:
pass
try:
threading.Thread(target=_run, name="browser-self-audit", daemon=True).start()
threading.Thread(target=p_run, name="browser-self-audit", daemon=True).start()
except Exception:
pass
backend/apps/agents/browser/browser_skills.py
+5 -5
View File
@@ -171,11 +171,11 @@ def template_task(task: str) -> tuple[str, list[str]]:
"""Replace each quoted span with a fixed token; return (templated, [values])."""
values: list[str] = []
def _repl(m):
def p_repl(m):
values.append(m.group(1))
return P_SLOT_TOKEN
return P_QUOTE_RE.sub(_repl, task or ""), values
return P_QUOTE_RE.sub(p_repl, task or ""), values
def compute_sig(task: str) -> str:
@@ -238,7 +238,7 @@ def distill_steps(action_log: list[dict]) -> list[dict]:
steps: list[dict] = []
productive_count = 0
def _emit_simple(tool, inp):
def p_emit_simple(tool, inp):
nonlocal productive_count
if tool in ("BrowserType", "type") and inp.get("selector") is not None:
steps.append({"tool": "BrowserType", "params": {"selector": inp.get("selector"), "text": inp.get("text", "")}})
@@ -284,7 +284,7 @@ def distill_steps(action_log: list[dict]) -> list[dict]:
steps.append({"tool": "BrowserClickByName", "params": {"role": (r or {}).get("clicked_role", ""), "name": name}})
productive_count += 1
continue
if not _emit_simple(st, sp):
if not p_emit_simple(st, sp):
return []
continue
if tool == "BrowserNavigate" and inp.get("url"):
@@ -757,7 +757,7 @@ def render_route_hint(skill: dict, task: str, score: float) -> tuple[str, list[t
_, values = template_task(task)
# first_unsafe_step is the batching boundary (it stops at composer typing
# too); the IRREVERSIBLE flag goes only on genuinely outward-facing clicks
unsafe_i, _why = first_unsafe_step(steps)
unsafe_i, p_why = first_unsafe_step(steps)
lines = []
for i, s in enumerate(steps):
mark = ""
backend/apps/agents/browser/browser_wait.py
+6 -6
View File
@@ -106,12 +106,12 @@ async def smart_wait(execute_fn, browser_id, tab_id, max_ms, *, until="",
last_elems = None
elems_changed_at = start # DOM-settle clock: when the element count last changed
def _elapsed():
def p_elapsed():
return (time.monotonic() - start) * 1000
while _elapsed() < max_ms:
await asyncio.sleep(min(poll_ms, max(0, max_ms - _elapsed())) / 1000)
if _elapsed() >= max_ms:
while p_elapsed() < max_ms:
await asyncio.sleep(min(poll_ms, max(0, max_ms - p_elapsed())) / 1000)
if p_elapsed() >= max_ms:
break
# Bound each probe so a wedged tab can't make us inherit the 30s command
# timeout. A timeout is a not-responding signal (not a verdict): count
@@ -156,13 +156,13 @@ async def smart_wait(execute_fn, browser_id, tab_id, max_ms, *, until="",
break
continue
if decide_stop(probe.get("ready"), probe.get("quiet", 0), dom_stable_ms,
probe.get("found"), _elapsed(),
probe.get("found"), p_elapsed(),
floor_ms=floor_ms, settle_window_ms=quiet_window_ms):
settled = True
found = bool(probe.get("found"))
break
waited = round(_elapsed())
waited = round(p_elapsed())
if found:
state = "found target"
elif settled:
backend/apps/agents/core/mcp_preflight.py
+1 -1
View File
@@ -181,7 +181,7 @@ def p_decorate(llm_suggestion: dict, available: list[CuratedEntry]) -> dict | No
async def p_call_classifier(settings, prompt: str, available: list[CuratedEntry], task_id: str | None = None) -> dict:
"""One aux-model call, returns validated JSON {is_vague, suggestions}."""
aux_model, _base = await resolve_aux_model(settings, preferred_tier="haiku")
aux_model, p_base = await resolve_aux_model(settings, preferred_tier="haiku")
client = get_anthropic_client_for_model(settings, aux_model)
catalog_lines = "\n".join(
backend/apps/agents/manager/prompt/attachments.py
+1 -1
View File
@@ -307,6 +307,6 @@ def resolve_attachments(context_paths: Optional[List], api_type: str, model: str
# blocks emitted, so behavior is the safe text-only old path).
@typechecked
def resolve_context_paths(context_paths: Optional[List]) -> str:
text, _native, refusals = resolve_attachments(context_paths, api_type="anthropic", model="")
text, p_native, refusals = resolve_attachments(context_paths, api_type="anthropic", model="")
refusal_text = "\n\n".join(refusals)
return "\n\n".join(p for p in (text, refusal_text) if p)
backend/apps/agents/providers/registry.py
+5 -5
View File
@@ -169,7 +169,7 @@ def find_custom_provider_for_value(settings, value: str):
if not isinstance(value, str) or not value.startswith(_CUSTOM_VALUE_PREFIX):
return None
rest = value[len(_CUSTOM_VALUE_PREFIX):]
slug, _sep, _bare = rest.partition("/")
slug, p_sep, p_bare = rest.partition("/")
if not slug:
return None
for cp in getattr(settings, "custom_providers", None) or []:
@@ -204,7 +204,7 @@ def find_builtin_model(short_name: str) -> dict | None:
}
if isinstance(short_name, str) and short_name.startswith(_CUSTOM_VALUE_PREFIX):
rest = short_name[len(_CUSTOM_VALUE_PREFIX):]
slug, _sep, bare_model = rest.partition("/")
slug, p_sep, bare_model = rest.partition("/")
if slug and bare_model:
# Routing string `cp-<slug>/<model>` matches the prefix we use
# when sync_custom_providers registers the provider node.
@@ -280,7 +280,7 @@ def resolve_model_id_for_sdk(short_name: str, settings: AppSettings) -> str:
# connected AG sub is preferred over the AI Studio key, which otherwise
# silently shadowed it. The map is AG's allowlist; pro variants 404/400 on
# AG and are deliberately absent, so they fall through to the key.
_ANTIGRAVITY_MAP = {
P_ANTIGRAVITY_MAP = {
# gemini-3-pro-preview disabled: AG returns 404 even with active conn.
# gemini-3.1-pro-preview disabled: AG's `gemini-3.1-pro-high` variant
# 400s every request with "invalid argument" (the `-high` thinking-
@@ -293,7 +293,7 @@ def resolve_model_id_for_sdk(short_name: str, settings: AppSettings) -> str:
rid = entry.get("router_model_id", "")
if isinstance(rid, str) and rid.startswith("gc/"):
suffix = rid[len("gc/"):]
ag_suffix = _ANTIGRAVITY_MAP.get(suffix)
ag_suffix = P_ANTIGRAVITY_MAP.get(suffix)
if ag_suffix and p_antigravity_connected():
return "ag/" + ag_suffix
if getattr(settings, "google_api_key", None):
@@ -392,7 +392,7 @@ def get_context_window(provider: str, model: str, settings: AppSettings | None =
bare_model = model
if isinstance(model, str) and model.startswith(_CUSTOM_VALUE_PREFIX):
rest = model[len(_CUSTOM_VALUE_PREFIX):]
_slug, _sep, bare_model = rest.partition("/")
p_slug, p_sep, bare_model = rest.partition("/")
for cp in getattr(settings, "custom_providers", []):
for m in (getattr(cp, "models", None) or []):
if m.get("value") == bare_model or m.get("id") == bare_model:
backend/apps/agents/proxy/anthropic_proxy.py
+12 -12
View File
@@ -184,9 +184,9 @@ def scrub_request_for_openai_gpt5(body: bytes) -> bytes:
if "temperature" in parsed and parsed["temperature"] != 1:
parsed.pop("temperature", None)
mutated = True
for _k in ("top_p", "top_k", "frequency_penalty", "presence_penalty",
for p_k in ("top_p", "top_k", "frequency_penalty", "presence_penalty",
"logprobs", "top_logprobs", "logit_bias"):
if parsed.pop(_k, None) is not None:
if parsed.pop(p_k, None) is not None:
mutated = True
try:
before = json.dumps(parsed.get("messages"), sort_keys=True) if "messages" in parsed else ""
@@ -426,22 +426,22 @@ async def proxy(rest: str, request: Request):
forward_to_openrouter as _forward_or,
)
from backend.apps.settings.settings import load_settings as _load
_s = _load()
p_s = _load()
if p_is_openai_max_completion_tokens_model(model):
_oak = (getattr(_s, "openai_api_key", "") or "").strip()
if _should_bypass_oai(parsed_for_bypass, _oak):
p_oak = (getattr(p_s, "openai_api_key", "") or "").strip()
if _should_bypass_oai(parsed_for_bypass, p_oak):
status, body_stream, hdrs = await _forward_oai(
parsed_for_bypass, _oak,
parsed_for_bypass, p_oak,
)
return StreamingResponse(
body_stream, status_code=status, headers=hdrs,
media_type=hdrs.get("content-type", "text/event-stream"),
)
if p_is_openrouter_model(model):
_ork = (getattr(_s, "openrouter_api_key", "") or "").strip()
if _should_bypass_or(parsed_for_bypass, _ork):
p_ork = (getattr(p_s, "openrouter_api_key", "") or "").strip()
if _should_bypass_or(parsed_for_bypass, p_ork):
status, body_stream, hdrs = await _forward_or(
parsed_for_bypass, _ork,
parsed_for_bypass, p_ork,
)
return StreamingResponse(
body_stream, status_code=status, headers=hdrs,
@@ -479,11 +479,11 @@ async def proxy(rest: str, request: Request):
# the retry, which hangs the whole turn for the full read window. Bound Gemini
# so a stalled first response fails fast (~2 min) instead of stalling ~10 min;
# other providers keep the generous window for long reasoning turns.
_read_timeout = 120.0 if p_is_gemini_model(model) else 600.0
p_read_timeout = 120.0 if p_is_gemini_model(model) else 600.0
try:
if wants_stream:
client = httpx.AsyncClient(timeout=httpx.Timeout(_read_timeout, connect=30.0))
client = httpx.AsyncClient(timeout=httpx.Timeout(p_read_timeout, connect=30.0))
req = client.build_request(
request.method, url, content=body, headers=forward_headers,
params=dict(request.query_params),
@@ -507,7 +507,7 @@ async def proxy(rest: str, request: Request):
media_type=upstream.headers.get("content-type", "text/event-stream"),
)
else:
async with httpx.AsyncClient(timeout=httpx.Timeout(_read_timeout, connect=30.0)) as client:
async with httpx.AsyncClient(timeout=httpx.Timeout(p_read_timeout, connect=30.0)) as client:
r = await client.request(
request.method, url, content=body, headers=forward_headers,
params=dict(request.query_params),
backend/apps/dashboards/dashboards.py
+1 -1
View File
@@ -323,7 +323,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, p_aux_base = 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,
backend/apps/mcp_registry/mcp_registry.py
+2 -2
View File
@@ -253,7 +253,7 @@ async def p_fetch_github_stars(servers: dict[str, dict]):
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
@@ -277,7 +277,7 @@ async def p_fetch_github_stars(servers: dict[str, dict]):
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(p_stars_cache)} total cached")
p_apply_stars(servers)
backend/apps/modes/modes.py
+3 -3
View File
@@ -20,9 +20,9 @@ async def modes_lifespan():
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 p_f:
p_data = _json.load(p_f)
if p_data.get("is_builtin") is True and p_data.get("id") == "chat":
os.remove(chat_path)
logger.info("Removed deprecated built-in chat.json (merged into ask)")
except Exception:
backend/apps/nine_router/oauth.py
+4 -4
View File
@@ -80,7 +80,7 @@ async def p_start_codex_callback_listener(timeout: float = 300.0) -> asyncio.bas
callback_served = asyncio.Event()
async def _handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
async def p_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)
@@ -165,7 +165,7 @@ async def p_start_codex_callback_listener(timeout: float = 300.0) -> asyncio.bas
pass
try:
server = await asyncio.start_server(_handle, "127.0.0.1", P_CODEX_CALLBACK_PORT)
server = await asyncio.start_server(p_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.
@@ -175,7 +175,7 @@ async def p_start_codex_callback_listener(timeout: float = 300.0) -> asyncio.bas
)
return None
async def _lifecycle():
async def p_lifecycle():
try:
await asyncio.wait_for(callback_served.wait(), timeout=timeout)
# Give the served HTML a moment to run its JS (postMessage +
@@ -193,7 +193,7 @@ async def p_start_codex_callback_listener(timeout: float = 300.0) -> asyncio.bas
except Exception:
pass
asyncio.create_task(_lifecycle())
asyncio.create_task(p_lifecycle())
logger.info(f"Started Codex callback listener on http://localhost:{P_CODEX_CALLBACK_PORT}{P_CODEX_CALLBACK_PATH}")
return server
backend/apps/nine_router/process.py
+40 -40
View File
@@ -213,20 +213,20 @@ def cli_auth_headers() -> dict[str, str]:
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"
p_is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1"
if _is_packaged:
if p_is_packaged:
import sys
_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
p_resources = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
p_candidate = os.path.join(p_resources, "router")
if os.path.isdir(p_candidate):
return p_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
p_backend_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
p_project_root = os.path.dirname(p_backend_dir)
p_candidate = os.path.join(p_project_root, "router")
if os.path.isdir(p_candidate):
return p_candidate
return None
@@ -396,12 +396,12 @@ async def ensure_running():
async def p_ensure_running_impl():
"""Start 9Router if not already running."""
global p_process
_is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1"
p_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:
if not p_is_packaged:
import subprocess as _sp
try:
result = _sp.run(
@@ -422,20 +422,20 @@ async def p_ensure_running_impl():
logger.info("9Router already running on port %d", NINE_ROUTER_PORT)
return
p_rotate_request_log()
_9router_dir = p_find_9router_dir()
_patch = p_gpt5_patch_path()
p_9router_dir = p_find_9router_dir()
p_patch = p_gpt5_patch_path()
if _is_packaged:
if p_is_packaged:
# Packaged: run the pre-built standalone server staged at
# <resources>/router/server.js by fetch-router at build time. We do NOT
# fall back to the dev npm path here, a user machine has no npm, so that
# only ever fails silently; every miss is reported instead.
if not _9router_dir:
if not p_9router_dir:
p_report_start_failure("router_not_bundled")
return
standalone_server = os.path.join(_9router_dir, "server.js")
standalone_server = os.path.join(p_9router_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(p_9router_dir, ".next", "standalone", "server.js")
if not os.path.exists(standalone_server):
p_report_start_failure("server_missing", router_dir_found=True)
return
@@ -444,7 +444,7 @@ async def p_ensure_running_impl():
p_report_start_failure("node_not_found", router_dir_found=True, server_found=True)
return
logger.info("Starting 9Router (production) on port %d...", NINE_ROUTER_PORT)
cmd = [node, f"--max-old-space-size={P_NODE_HEAP_MB}"] + (["--require", _patch] if _patch else []) + [standalone_server]
cmd = [node, f"--max-old-space-size={P_NODE_HEAP_MB}"] + (["--require", p_patch] if p_patch else []) + [standalone_server]
cwd = os.path.dirname(standalone_server)
env = {**os.environ, "PORT": str(NINE_ROUTER_PORT), "NODE_ENV": "production"}
if node == os.environ.get("OPENSWARM_ELECTRON_PATH"):
@@ -464,7 +464,7 @@ async def p_ensure_running_impl():
"Starting 9Router (dev cache, 9router@%s) on port %d...",
NINE_ROUTER_NPM_VERSION, NINE_ROUTER_PORT,
)
cmd = [node, f"--max-old-space-size={P_NODE_HEAP_MB}"] + (["--require", _patch] if _patch else []) + [cached_server]
cmd = [node, f"--max-old-space-size={P_NODE_HEAP_MB}"] + (["--require", p_patch] if p_patch else []) + [cached_server]
cwd = os.path.dirname(cached_server)
env = {**os.environ, "PORT": str(NINE_ROUTER_PORT), "NODE_ENV": "production"}
@@ -473,30 +473,30 @@ async def p_ensure_running_impl():
# whole reason #90 was un-diagnosable). Packaged prod (NODE_ENV=production
# standalone) is quiet, so one fixed temp file, truncated each start attempt,
# won't grow; dev keeps its chatty-Next.js DEVNULL unless debug is set.
_cap_path = os.path.join(tempfile.gettempdir(), "openswarm-9router-start.log")
_cap_file = None
if _is_packaged:
p_cap_path = os.path.join(tempfile.gettempdir(), "openswarm-9router-start.log")
p_cap_file = None
if p_is_packaged:
try:
_cap_file = open(_cap_path, "wb")
_stdout, _stderr = _cap_file, subprocess.STDOUT
p_cap_file = open(p_cap_path, "wb")
p_stdout, p_stderr = p_cap_file, subprocess.STDOUT
except OSError:
_stdout, _stderr = subprocess.DEVNULL, subprocess.DEVNULL
p_stdout, p_stderr = subprocess.DEVNULL, subprocess.DEVNULL
elif os.environ.get("OPENSWARM_DEBUG_9ROUTER"):
_log_path = os.path.join(
p_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, _stderr = open(_log_path, "a", buffering=1), subprocess.STDOUT
logger.info(f"9Router debug logging enabled → {_log_path}")
os.makedirs(os.path.dirname(p_log_path), exist_ok=True)
p_stdout, p_stderr = open(p_log_path, "a", buffering=1), subprocess.STDOUT
logger.info(f"9Router debug logging enabled → {p_log_path}")
else:
_stdout, _stderr = subprocess.DEVNULL, subprocess.DEVNULL
p_stdout, p_stderr = subprocess.DEVNULL, subprocess.DEVNULL
try:
p_process = subprocess.Popen(cmd, cwd=cwd, stdout=_stdout, stderr=_stderr, env=env)
if _cap_file is not None:
_cap_file.close() # the child holds its own fd; the parent copy isn't needed
timeout = 20 if _is_packaged else 30
p_process = subprocess.Popen(cmd, cwd=cwd, stdout=p_stdout, stderr=p_stderr, env=env)
if p_cap_file is not None:
p_cap_file.close() # the child holds its own fd; the parent copy isn't needed
timeout = 20 if p_is_packaged else 30
for _ in range(timeout * 2):
await asyncio.sleep(0.5)
if is_running():
@@ -506,19 +506,19 @@ async def p_ensure_running_impl():
# exit code (non-None = it crashed; None = wedged or just slow).
p_report_start_failure(
"not_ready_in_time",
detail=p_read_capture_tail(_cap_path) if _is_packaged else "",
detail=p_read_capture_tail(p_cap_path) if p_is_packaged else "",
returncode=p_process.poll(),
timeout_s=timeout,
)
except Exception as e:
if _cap_file is not None and not _cap_file.closed:
if p_cap_file is not None and not p_cap_file.closed:
try:
_cap_file.close()
p_cap_file.close()
except OSError:
pass
p_report_start_failure(
"spawn_exception",
detail=f"{e}\n{p_read_capture_tail(_cap_path) if _is_packaged else ''}",
detail=f"{e}\n{p_read_capture_tail(p_cap_path) if p_is_packaged else ''}",
)
backend/apps/outputs/html_inject.py
+2 -2
View File
@@ -147,7 +147,7 @@ def inject_token_into_relative_urls(html: str, token: str) -> str:
if not token:
return html
def _patch(match: re.Match) -> str:
def p_patch(match: re.Match) -> str:
attr, quote, url = match.group(1), match.group(2), match.group(3)
lowered = url.lower().lstrip()
if lowered.startswith(P_ABSOLUTE_URL_PREFIXES):
@@ -164,7 +164,7 @@ 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 P_HREF_SRC_ATTR_RE.sub(_patch, html)
return P_HREF_SRC_ATTR_RE.sub(p_patch, html)
def decode_data_param(d: str) -> tuple[str, str]:
backend/apps/outputs/outputs.py
+5 -5
View File
@@ -78,7 +78,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, p_d: 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))
@@ -91,7 +91,7 @@ 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")
input_json, result_json = decode_data_param(p_d) if p_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, with_runtime=True)
# Iframe sub-resource fetches (<link>, <script src>, <img>) drop the
@@ -104,7 +104,7 @@ async def serve_workspace_file(workspace_id: str, filepath: str, _d: str = ""):
@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, p_d: str = ""):
"""Serve a file from a saved output's files dict. For index.html, inject OUTPUT data."""
output = load(output_id)
content = output.files.get(filepath)
@@ -112,7 +112,7 @@ async def serve_output_file(output_id: str, filepath: str, _d: str = ""):
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")
input_json, result_json = decode_data_param(p_d) if p_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, with_runtime=True)
content = inject_token_into_relative_urls(content, get_auth_token())
@@ -623,7 +623,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, p_aux_base = await resolve_aux_model(load_settings(), preferred_tier="sonnet")
except ValueError as e:
return {
"message": f"Error: {str(e)}",
backend/apps/outputs/publish_build.py
+2 -2
View File
@@ -79,7 +79,7 @@ async def build_static(output: Output) -> Optional[str]:
env={**os.environ, "NODE_ENV": "production"},
)
try:
_out, err = await asyncio.wait_for(proc.communicate(), timeout=P_BUILD_TIMEOUT)
p_out, err = await asyncio.wait_for(proc.communicate(), timeout=P_BUILD_TIMEOUT)
except asyncio.TimeoutError:
proc.kill()
await proc.wait()
@@ -120,7 +120,7 @@ def collect_bundle(output: Output, dist_dir: Optional[str]) -> bytes:
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tar:
if dist_dir:
for root, _dirs, files in os.walk(dist_dir):
for root, p_dirs, files in os.walk(dist_dir):
for fn in files:
full = os.path.join(root, fn)
if os.path.islink(full):
backend/apps/outputs/publish_scan.py
+3 -3
View File
@@ -51,8 +51,8 @@ def collect_source(output: Output) -> dict[str, str]:
src[name] = content
if is_webapp(output):
root = workspace_dir(output)
for base, _dirs, fnames in os.walk(root):
_dirs[:] = [d for d in _dirs if d not in WALK_SKIP_DIRS]
for base, p_dirs, fnames in os.walk(root):
p_dirs[:] = [d for d in p_dirs if d not in WALK_SKIP_DIRS]
for fn in fnames:
if not fn.lower().endswith(P_SCAN_EXTS):
continue
@@ -115,7 +115,7 @@ async def llm_findings(src: dict[str, str], settings) -> tuple[list[str], str]:
from backend.apps.settings.credentials import get_anthropic_client_for_model
from backend.apps.agents.core.aux_llm import safe_resp_text
try:
model, _base = await resolve_aux_model(settings, preferred_tier="haiku")
model, p_base = await resolve_aux_model(settings, preferred_tier="haiku")
except Exception:
return [], "clean"
client = get_anthropic_client_for_model(settings, model)
backend/apps/outputs/runtime.py
+5 -5
View File
@@ -335,7 +335,7 @@ class AppRuntime:
# end doesn't double-release if a success path beat it.
lock_released = False
def _release_boot_lock() -> None:
def p_release_boot_lock() -> None:
nonlocal lock_released
if lock_released:
return
@@ -377,7 +377,7 @@ 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()
p_release_boot_lock()
return
except (OSError, asyncio.TimeoutError):
pass
@@ -395,7 +395,7 @@ class AppRuntime:
# 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()
p_release_boot_lock()
async def _start_old_mode(self) -> bool:
if not self.has_backend_file:
@@ -487,10 +487,10 @@ class AppRuntime:
except Exception:
pass
def _unsub() -> None:
def p_unsub() -> None:
self._subscribers.discard(cb)
return _unsub
return p_unsub
def _broadcast(self, line: LogLine) -> None:
self.log_buffer.append(line)
backend/apps/outputs/view_builder_templates.py
+2 -2
View File
@@ -566,7 +566,7 @@ def warm_cache_in_background() -> None:
if node_done and venv_done:
return
def _runner() -> None:
def p_runner() -> None:
try:
ensure_warm_cache()
except Exception:
@@ -577,7 +577,7 @@ def warm_cache_in_background() -> None:
logger.exception("background warm python venv crashed")
p_warm_cache_thread = threading.Thread(
target=_runner, daemon=True, name="webapp-template-warm-cache"
target=p_runner, daemon=True, name="webapp-template-warm-cache"
)
p_warm_cache_thread.start()
backend/apps/service/client.py
+2 -2
View File
@@ -328,13 +328,13 @@ def p_schedule(coro) -> None:
return
import threading
def _run():
def p_run():
try:
asyncio.run(coro)
except Exception:
pass
threading.Thread(target=_run, daemon=True).start()
threading.Thread(target=p_run, daemon=True).start()
# --------------------------------------------------------------------------
backend/apps/service/service.py
+2 -2
View File
@@ -271,7 +271,7 @@ async def usage_summary():
for s in agent_manager.get_all_sessions():
sessions.append(s.model_dump(mode="json"))
def _is_real(sess: dict) -> bool:
def p_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:
@@ -281,7 +281,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 p_is_real(s)]
total_sessions = len(sessions)
total_cost = sum(s.get("cost_usd", 0) for s in sessions)
backend/apps/service/version.py
+5 -5
View File
@@ -23,11 +23,11 @@ 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")
p_here = os.path.dirname(os.path.abspath(__file__))
p_repo = os.path.dirname(os.path.dirname(os.path.dirname(p_here)))
p_pkg = os.path.join(p_repo, "electron", "package.json")
with open(p_pkg, encoding="utf-8") as p_f:
return json.load(p_f).get("version", "unknown")
except (OSError, ValueError, KeyError):
return "unknown"
backend/apps/settings/settings.py
+10 -10
View File
@@ -39,7 +39,7 @@ async def settings_lifespan():
s = load_settings()
import asyncio as _asyncio
async def _boot_router_then_sync():
async def p_boot_router_then_sync():
"""Boot 9Router then push key-based connections (sequential: sync helpers no-op pre-boot)."""
needs_router = any([
getattr(s, "google_api_key", None),
@@ -75,7 +75,7 @@ async def settings_lifespan():
await sync_openswarm_pro_as_claude(bearer, base)
await sync_custom_providers(getattr(s, "custom_providers", None) or [])
_asyncio.create_task(_boot_router_then_sync())
_asyncio.create_task(p_boot_router_then_sync())
_asyncio.create_task(p_upload_dir_gc_loop())
except Exception as e:
logger.warning(f"9Router sync startup failed: {e}")
@@ -297,7 +297,7 @@ async def apply_settings_update(body: AppSettings, protect_fields: set[str] | No
# Off the request path: ensure_running() can take 5min on first install (npm pull) and would freeze the loop.
if google_changed or openai_changed or openrouter_changed or custom_providers_changed:
async def _boot_and_sync_keys(
async def p_boot_and_sync_keys(
google_key: str | None,
openai_key: str | None,
openrouter_key: str | None,
@@ -330,7 +330,7 @@ async def apply_settings_update(body: AppSettings, protect_fields: set[str] | No
except Exception as e:
logger.warning(f"Background apikey sync failed: {e}")
asyncio.create_task(_boot_and_sync_keys(
asyncio.create_task(p_boot_and_sync_keys(
getattr(body, "google_api_key", None),
getattr(body, "openai_api_key", None),
getattr(body, "openrouter_api_key", None),
@@ -611,7 +611,7 @@ async def summarize_file(req: p_SummarizeRequest):
from backend.apps.agents.providers.registry import resolve_aux_model, get_api_type
from backend.apps.settings.credentials import get_anthropic_client_for_model
s = load_settings()
aux_model, _base = await resolve_aux_model(
aux_model, p_base = await resolve_aux_model(
s,
preferred_tier="haiku",
primary_api=get_api_type(req.primary_model) if req.primary_model else None,
@@ -635,7 +635,7 @@ async def summarize_file(req: p_SummarizeRequest):
CHUNK_CHARS = 200_000
is_chunked = len(raw) > CHUNK_CHARS
async def _summarize_block(text: str, target_tokens: int, label: str) -> str:
async def p_summarize_block(text: str, target_tokens: int, label: str) -> str:
user = (
f"Target length: ~{target_tokens} tokens.\n\n"
f"<document path=\"{label}\">\n{text}\n</document>\n\n"
@@ -657,7 +657,7 @@ async def summarize_file(req: p_SummarizeRequest):
return out
if not is_chunked:
summary = await _summarize_block(raw, req.target_tokens, os.path.basename(src))
summary = await p_summarize_block(raw, req.target_tokens, os.path.basename(src))
else:
chunks = [raw[i:i + CHUNK_CHARS] for i in range(0, len(raw), CHUNK_CHARS)]
per_chunk_budget = max(800, req.target_tokens // len(chunks) + 600)
@@ -667,15 +667,15 @@ async def summarize_file(req: p_SummarizeRequest):
# provider's per-key rate limit, and a single user summarizing
# one file will never hit that.
partials = await asyncio.gather(*[
_summarize_block(ch, per_chunk_budget, f"{os.path.basename(src)} (part {i + 1} of {len(chunks)})")
p_summarize_block(ch, per_chunk_budget, f"{os.path.basename(src)} (part {i + 1} of {len(chunks)})")
for i, ch in enumerate(chunks)
])
merge_input = "\n\n".join(f"## Part {i + 1}\n{p}" for i, p in enumerate(partials))
summary = await _summarize_block(merge_input, req.target_tokens, f"merged summary of {os.path.basename(src)}")
summary = await p_summarize_block(merge_input, req.target_tokens, f"merged summary of {os.path.basename(src)}")
except Exception as e:
raise HTTPException(status_code=502, detail=f"summarize failed: {e}")
base, _ext = os.path.splitext(src)
base, p_ext = os.path.splitext(src)
dest = f"{base}.summary.txt"
counter = 1
while os.path.exists(dest):
backend/apps/skill_registry/skill_registry.py
+3 -3
View File
@@ -326,14 +326,14 @@ def select_skill_paths(tree: list[dict], skill_id: str) -> tuple[str, list[str]]
if not candidates:
raise ValueError(f"no SKILL.md for '{skill_id}' in this repo")
def _rank(p: str) -> tuple:
def p_rank(p: str) -> tuple:
if p == f"{skill_id}/SKILL.md":
return (0, 0, p)
if p == f"skills/{skill_id}/SKILL.md":
return (1, p.count("/"), p)
return (2, p.count("/"), p)
skill_md = min(candidates, key=_rank)
skill_md = min(candidates, key=p_rank)
skill_dir = skill_md[: -len("/SKILL.md")] if "/" in skill_md else ""
prefix = (skill_dir + "/") if skill_dir else ""
members = [p for p in blobs if (p.startswith(prefix) if prefix else "/" not in p)]
@@ -400,7 +400,7 @@ async def resolve_community_skill(source: str, skill_id: str) -> dict:
if "SKILL.md" not in files:
raise ValueError("SKILL.md could not be fetched")
meta, _body = p_parse_frontmatter(files["SKILL.md"])
meta, p_body = p_parse_frontmatter(files["SKILL.md"])
# Reuse the .swarm importer's content scan: flag files holding secret-shaped
# literals (the author's leaked key, or a sketchy skill) so the user sees it
# before installing from an unvetted repo.
backend/apps/subscription/free_trial.py
+2 -2
View File
@@ -182,11 +182,11 @@ async def arm_free_trial(settings_obj) -> dict:
# over a sub that's merely still loading. CAPPED on purpose: a genuinely
# sub-less user exhausts these in ~1.2s and falls through to arm, so this
# never waits on a subscription that doesn't exist.
for _i in range(5):
for p_i in range(5):
if await p_has_connected_subscription():
has_sub = True
break
if _i < 4:
if p_i < 4:
await asyncio.sleep(0.3)
if own or has_sub:
# A real model exists now (key, custom provider, or a 9Router sub). If we
backend/apps/swarm/closure.py
+2 -2
View File
@@ -88,7 +88,7 @@ def p_assemble(root_type: EntityType, root_id: str):
counts: dict[str, int] = {}
for key in order:
etype, _lid = key
etype, p_lid = key
inst = nodes[key]
bid = local_to_bundle[key]
payloads[bid] = scrub_payload(inst.serialize(ctx))
@@ -304,7 +304,7 @@ def p_read_files(sandbox: str, ref: EntityRef) -> dict[str, bytes]:
out: dict[str, bytes] = {}
if not os.path.isdir(base):
return out
for root, _dirs, fnames in os.walk(base):
for root, p_dirs, fnames in os.walk(base):
for fn in fnames:
full = os.path.join(root, fn)
with open(full, "rb") as f:
backend/apps/swarm/entities/dashboards.py
+1 -1
View File
@@ -90,7 +90,7 @@ class DashboardExportable:
"parent_session_id": remap.local(parent) if parent else None,
}
browser_cards = {}
for _bkey, card in (layout.get("browser_cards") or {}).items():
for p_bkey, card in (layout.get("browser_cards") or {}).items():
nbid = "browser-" + uuid4().hex[:10]
c = dict(card)
c["browser_id"] = nbid
backend/apps/swarm/entities/skills.py
+1 -1
View File
@@ -99,7 +99,7 @@ class SkillExportable:
def p_read_supporting_files(skill_dir: str) -> dict[str, bytes]:
"""Every file in a skill folder except SKILL.md, as {relpath: bytes}."""
out: dict[str, bytes] = {}
for root, _dirs, names in os.walk(skill_dir):
for root, p_dirs, names in os.walk(skill_dir):
for n in names:
full = os.path.join(root, n)
rel = os.path.relpath(full, skill_dir)
backend/apps/swarm/redact.py
+3 -3
View File
@@ -57,19 +57,19 @@ def scrub_payload(value: Any) -> Any:
return value
def find_denied_keys(value: Any, _path: str = "") -> list[str]:
def find_denied_keys(value: Any, p_path: str = "") -> list[str]:
"""Audit used by ziputil.pack as the last line of defense: the paths of any
denied key still present. Empty list means clean."""
found: list[str] = []
if isinstance(value, dict):
for k, v in value.items():
here = f"{_path}.{k}" if _path else str(k)
here = f"{p_path}.{k}" if p_path else str(k)
if isinstance(k, str) and is_denied_key(k):
found.append(here)
found.extend(find_denied_keys(v, here))
elif isinstance(value, list):
for i, v in enumerate(value):
found.extend(find_denied_keys(v, f"{_path}[{i}]"))
found.extend(find_denied_keys(v, f"{p_path}[{i}]"))
return found
backend/apps/swarm/ziputil.py
+1 -1
View File
@@ -71,7 +71,7 @@ def p_sandbox_entries(sandbox: str) -> dict[str, bytes]:
relpath so it matches the keys pack() hashed (cross-platform)."""
out: dict[str, bytes] = {}
root = os.path.realpath(sandbox)
for base, _dirs, fnames in os.walk(root):
for base, p_dirs, fnames in os.walk(root):
for fn in fnames:
full = os.path.join(base, fn)
rel = os.path.relpath(full, root).replace(os.sep, "/")
backend/apps/tools_lib/mcp_config.py
+29 -29
View File
@@ -20,9 +20,9 @@ def p_extra_bin_dirs() -> list[str]:
"""Well-known user-local bin directories that may not be on PATH in packaged apps."""
home = os.path.expanduser("~")
# Bundled uv-bin (ships uvx for non-dev users)
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
p_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
dirs = [
os.path.join(_backend, "uv-bin"),
os.path.join(p_backend, "uv-bin"),
os.path.join(home, ".bun", "bin"),
os.path.join(home, ".cargo", "bin"),
os.path.join(home, ".local", "bin"),
@@ -59,19 +59,19 @@ def resolve_command(command: str) -> str | None:
suffixes = [""] + os.environ.get("PATHEXT", ".COM;.EXE;.BAT;.CMD").lower().split(os.pathsep)
else:
suffixes = [""]
def _probe(directory: str) -> str | None:
def p_probe(directory: str) -> str | None:
for suffix in suffixes:
candidate = os.path.join(directory, command + suffix)
if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
return candidate
return None
for d in p_extra_bin_dirs():
hit = _probe(d)
hit = p_probe(d)
if hit:
return hit
# Check bundled uv-bin directory (ships uv/uvx for non-dev users)
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
return _probe(os.path.join(_backend, "uv-bin"))
p_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
return p_probe(os.path.join(p_backend, "uv-bin"))
def augmented_path() -> str:
@@ -130,9 +130,9 @@ def derive_mcp_config(tool: ToolDefinition) -> Optional[dict]:
# proxy that forwards the refresh to our cloud's pool-aware
# /api/oauth/google/refresh endpoint; CLIENT_ID/SECRET become
# unused placeholders (gauth.py only validates non-empty).
_port = os.environ.get("OPENSWARM_PORT", "8324")
p_port = os.environ.get("OPENSWARM_PORT", "8324")
env["GOOGLE_WORKSPACE_TOKEN_URI"] = (
f"http://127.0.0.1:{_port}/api/tools/google-oauth-token"
f"http://127.0.0.1:{p_port}/api/tools/google-oauth-token"
)
env.setdefault("GOOGLE_WORKSPACE_CLIENT_ID", "openswarm-proxy")
env.setdefault("GOOGLE_WORKSPACE_CLIENT_SECRET", "openswarm-proxy")
@@ -167,9 +167,9 @@ def derive_mcp_config(tool: ToolDefinition) -> Optional[dict]:
# The shim runs as a subprocess and needs to import
# `backend.apps.discord_mcp_shim`; set PYTHONPATH to the project
# root (parent of the backend/ dir) so that import resolves.
_project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
p_project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
existing_pp = env.get("PYTHONPATH") or os.environ.get("PYTHONPATH", "")
env["PYTHONPATH"] = (_project_root + os.pathsep + existing_pp) if existing_pp else _project_root
env["PYTHONPATH"] = (p_project_root + os.pathsep + existing_pp) if existing_pp else p_project_root
# Microsoft 365 MCP: use a stable token cache path shared across process spawns
if tool.name.lower() == "microsoft 365" and config.get("type") == "stdio":
@@ -196,7 +196,7 @@ def derive_mcp_config(tool: ToolDefinition) -> Optional[dict]:
if config["command"] in ("npx", "bunx"):
pkg_name = next((a for a in (config.get("args") or []) if not a.startswith("-")), None)
if pkg_name:
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
p_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
electron_path = os.environ.get("OPENSWARM_ELECTRON_PATH")
# Two bundle layouts in mcp-bundles/, checked in priority order:
#
@@ -217,8 +217,8 @@ def derive_mcp_config(tool: ToolDefinition) -> Optional[dict]:
# Scoped names get flattened ("@softeria/ms-365-mcp-server"
# -> "softeria-ms-365-mcp-server") for filesystem safety.
safe_bundle = pkg_name.replace("/", "-").replace("@", "")
bundle_dir_path = os.path.join(_backend, "mcp-bundles", safe_bundle, "dist", "index.js")
bundle_file_path = os.path.join(_backend, "mcp-bundles", f"{safe_bundle}.js")
bundle_dir_path = os.path.join(p_backend, "mcp-bundles", safe_bundle, "dist", "index.js")
bundle_file_path = os.path.join(p_backend, "mcp-bundles", f"{safe_bundle}.js")
bundle_path = None
if os.path.isfile(bundle_dir_path):
bundle_path = bundle_dir_path
@@ -242,7 +242,7 @@ def derive_mcp_config(tool: ToolDefinition) -> Optional[dict]:
else:
# Check for pre-installed npm package (works in both dev and packaged modes)
safe_dir = pkg_name.replace("/", "-").replace("@", "")
npm_dir = os.path.join(_backend, "npm-servers", safe_dir)
npm_dir = os.path.join(p_backend, "npm-servers", safe_dir)
pkg_json_path = os.path.join(npm_dir, "node_modules", pkg_name, "package.json")
if os.path.isfile(pkg_json_path):
import json as _json
@@ -272,23 +272,23 @@ def derive_mcp_config(tool: ToolDefinition) -> Optional[dict]:
env.setdefault("PYTHONPATH", "")
# Point uv/uvx at our bundled Python; avoids macOS CLT popup on fresh Macs
# and avoids downloading Python at runtime
_is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1"
_is_windows = sys.platform == "win32"
if _is_packaged:
_resources = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
if _is_windows:
_bundled_python = os.path.join(_resources, "python-env", "python.exe")
p_is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1"
p_is_windows = sys.platform == "win32"
if p_is_packaged:
p_resources = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
if p_is_windows:
p_bundled_python = os.path.join(p_resources, "python-env", "python.exe")
else:
_bundled_python = os.path.join(_resources, "python-env", "bin", "python3")
if os.path.exists(_bundled_python):
env.setdefault("UV_PYTHON", _bundled_python)
p_bundled_python = os.path.join(p_resources, "python-env", "bin", "python3")
if os.path.exists(p_bundled_python):
env.setdefault("UV_PYTHON", p_bundled_python)
else:
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
if _is_windows:
_venv_python = os.path.join(_backend, ".venv", "Scripts", "python.exe")
p_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
if p_is_windows:
p_venv_python = os.path.join(p_backend, ".venv", "Scripts", "python.exe")
else:
_venv_python = os.path.join(_backend, ".venv", "bin", "python3")
if os.path.exists(_venv_python):
env.setdefault("UV_PYTHON", _venv_python)
p_venv_python = os.path.join(p_backend, ".venv", "bin", "python3")
if os.path.exists(p_venv_python):
env.setdefault("UV_PYTHON", p_venv_python)
return config
backend/apps/tools_lib/mcp_discovery.py
+8 -8
View File
@@ -124,7 +124,7 @@ def p_try_heal_npx_cache(stderr: str) -> str | None:
return hash_
async def discover_mcp_tools_stdio(command: str, args: list[str] | None = None, env: dict | None = None, _attempt: int = 0) -> list[dict]:
async def discover_mcp_tools_stdio(command: str, args: list[str] | None = None, env: dict | None = None, p_attempt: int = 0) -> list[dict]:
"""Spawn a stdio MCP server process and call tools/list via JSON-RPC over stdin/stdout.
On the first attempt, a failure that looks like corrupted npx cache
@@ -155,7 +155,7 @@ async def discover_mcp_tools_stdio(command: str, args: list[str] | None = None,
# the opaque "discovery failed" we used to show.
stderr_tail: list[str] = []
async def _drain_stderr() -> None:
async def p_drain_stderr() -> None:
try:
while True:
chunk = await proc.stderr.readline()
@@ -169,14 +169,14 @@ async def discover_mcp_tools_stdio(command: str, args: list[str] | None = None,
except Exception:
return
stderr_task = asyncio.create_task(_drain_stderr())
stderr_task = asyncio.create_task(p_drain_stderr())
async def p_send(msg: dict) -> None:
line = json.dumps(msg) + "\n"
proc.stdin.write(line.encode())
await proc.stdin.drain()
async def _recv(timeout_s: float = 30.0) -> dict:
async def p_recv(timeout_s: float = 30.0) -> dict:
"""Read JSON-RPC responses, skipping notification lines (no 'id' field)."""
while True:
line = await asyncio.wait_for(proc.stdout.readline(), timeout=timeout_s)
@@ -217,12 +217,12 @@ async def discover_mcp_tools_stdio(command: str, args: list[str] | None = None,
# AV-scan every file npm writes; total install time often exceeds
# 60 s and occasionally pushes past 90 s. Subsequent reads run
# against an already-running server and stay at the default 30 s.
await _recv(timeout_s=120.0)
await p_recv(timeout_s=120.0)
await p_send({"jsonrpc": "2.0", "method": "notifications/initialized"})
await p_send({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}})
data = await _recv()
data = await p_recv()
tools_list = data.get("result", {}).get("tools", [])
return [{"name": t.get("name", ""), "description": t.get("description", ""), "inputSchema": t.get("inputSchema")} for t in tools_list]
@@ -231,8 +231,8 @@ async def discover_mcp_tools_stdio(command: str, args: list[str] | None = None,
# Heal-on-corrupt-npx-cache still triggers from the EOF branch,
# which now includes the full stderr tail in `e.detail`; so the
# ERR_MODULE_NOT_FOUND signature is still discoverable here.
if _attempt == 0 and p_try_heal_npx_cache(str(e.detail) if e.detail is not None else ""):
return await discover_mcp_tools_stdio(command, args, env, _attempt=1)
if p_attempt == 0 and p_try_heal_npx_cache(str(e.detail) if e.detail is not None else ""):
return await discover_mcp_tools_stdio(command, args, env, p_attempt=1)
raise
except asyncio.TimeoutError:
# Most common cause: cold npx cache on Windows. The npm install
backend/apps/tools_lib/oauth_tokens.py
+3 -3
View File
@@ -165,9 +165,9 @@ def m365_server_script() -> str:
package.json) because cli.js reads __dirname/../package.json for the
--version flag; see scripts/build-app.sh `build_mcp_bundle_dir`.
"""
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
p_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
bundle = os.path.join(
_backend, "mcp-bundles", "softeria-ms-365-mcp-server", "dist", "index.js",
p_backend, "mcp-bundles", "softeria-ms-365-mcp-server", "dist", "index.js",
)
if os.path.isfile(bundle):
return bundle
@@ -175,7 +175,7 @@ def m365_server_script() -> str:
# was left over from before the bundle migration. Will return the legacy
# path; if that doesn't exist either, the caller raises a clear error.
return os.path.join(
_backend, "npm-servers", "softeria-ms-365-mcp-server",
p_backend, "npm-servers", "softeria-ms-365-mcp-server",
"node_modules", "@softeria", "ms-365-mcp-server", "dist", "index.js",
)
backend/apps/tools_lib/tools_lib.py
+4 -4
View File
@@ -505,7 +505,7 @@ async def m365_device_login(tool_id: str):
import threading
login_state: dict = {"proc": proc, "status": "waiting_for_code", "device_code": "", "device_code_url": "", "email": None, "output": ""}
def _read_output():
def p_read_output():
import re
for line in proc.stdout:
login_state["output"] += line
@@ -544,7 +544,7 @@ async def m365_device_login(tool_id: str):
else:
login_state["status"] = "error"
thread = threading.Thread(target=_read_output, daemon=True)
thread = threading.Thread(target=p_read_output, daemon=True)
thread.start()
p_m365_login_processes[tool_id] = login_state
@@ -643,11 +643,11 @@ async def oauth_start(tool_id: str):
)
from backend.config.install_id import get_install_id
install_id = get_install_id()
_port = os.environ.get("OPENSWARM_PORT", "8324")
p_port = os.environ.get("OPENSWARM_PORT", "8324")
params = {
"install_id": install_id,
"tool_id": tool_id,
"local_port": _port,
"local_port": p_port,
}
auth_url = (
f"{OPENSWARM_OAUTH_BASE_URL}/api/oauth/{proxied}/start?"