WEBDIENSTE/openswarm
1
0
Fork 0
mirror of https://github.com/openswarm-ai/openswarm.git synced 2026-09-26 19:44:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

[eric] agents: leading-_ -> p_/P_/public in proxy+providers/openrouter+ws_manager+mcp-servers+oauth_state; promote cross-file API public, fix monkeypatch+attr test refs

Browse Source
This commit is contained in:
ciregenz
2026-06-23 20:00:06 -07:00
parent 991af1b580
commit f703b8aa02
17 changed files with 205 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/agents/agents.py
+3 -3
View File
@@ -435,9 +435,9 @@ async def subscriptions_exchange(body: dict):
"""Exchange OAuth code for tokens via 9Router."""
from backend.apps.nine_router import exchange_oauth
from backend.apps.oauth_state import (
_pending_oauth as pending_oauth,
_completed_oauth as completed_oauth,
_mark_oauth_completed as mark_completed,
pending_oauth as pending_oauth,
completed_oauth as completed_oauth,
mark_oauth_completed as mark_completed,
)
provider = body.get("provider", "")
code = body.get("code", "")
backend/apps/agents/browser/browser_agent.py
+2 -2
View File
@@ -77,7 +77,7 @@ from backend.apps.agents.browser.browser_schema import (
SYSTEM_PROMPT,
)
from backend.apps.agents.core.models import AgentSession, ApprovalRequest, Message
from backend.apps.agents.core.ws_manager import ws_manager, _await_reconnect
from backend.apps.agents.core.ws_manager import ws_manager, await_reconnect
from backend.apps.tools_lib.tools_lib import load_builtin_permissions
logger = logging.getLogger(__name__)
@@ -2290,7 +2290,7 @@ async def run_browser_agents(
# corpse before card-gone detection trips. But a CPU-starved renderer can
# briefly drop its WS then auto-reconnect, so wait (capped) for it to come
# back before refusing, turning a load blip into a pause, not a failed run.
if not ws_manager.global_connections and not await _await_reconnect(lambda: bool(ws_manager.global_connections)):
if not ws_manager.global_connections and not await await_reconnect(lambda: bool(ws_manager.global_connections)):
logger.warning("[browser-agent] dispatch refused: no dashboard after reconnect wait")
return [{
"summary": (
backend/apps/agents/browser_agent_mcp_server.py
+2 -2
View File
@@ -156,7 +156,7 @@ def call_backend(tasks: list[dict]) -> dict:
MAX_IMAGE_B64_BYTES = 400_000
def _sniff_image_mime(b64: str) -> str:
def p_sniff_image_mime(b64: str) -> str:
"""PNG vs JPEG from the base64 magic bytes. Capture now sends JPEG, but older
callers / cached shots may be PNG, so we label by content, not assumption."""
if b64.startswith("/9j/"):
@@ -214,7 +214,7 @@ def format_result(result: dict) -> dict:
screenshot = result.get("final_screenshot")
if screenshot:
image_data = screenshot
mime_type = _sniff_image_mime(screenshot)
mime_type = p_sniff_image_mime(screenshot)
if len(image_data) > MAX_IMAGE_B64_BYTES:
compressed = compress_screenshot(image_data)
backend/apps/agents/core/ws_manager.py
+10 -10
View File
@@ -12,28 +12,28 @@ logger = logging.getLogger(__name__)
# operate on an already-loaded page and should be quick; navigation legitimately
# loads the network so it gets a longer leash. Was a flat 30s, which let one
# wedged page spin for ~20 minutes across retries.
_BROWSER_CMD_TIMEOUT_DEFAULT = 15.0 # modest load headroom; still "short" so a wedged tab fails fast
_BROWSER_CMD_TIMEOUTS = {
BROWSER_CMD_TIMEOUT_DEFAULT = 15.0 # modest load headroom; still "short" so a wedged tab fails fast
BROWSER_CMD_TIMEOUTS = {
"navigate": 25.0, # a real page load can be slow (more leash under load)
"replay_route": 20.0, # an API fetch can be slow
"wait": 12.0, # smart-wait already caps itself well under this
}
_BROWSER_CMD_REBROADCAST_S = 3.0
BROWSER_CMD_REBROADCAST_S = 3.0
# A CPU-starved renderer can briefly drop its WS (a missed heartbeat) and the
# frontend auto-reconnects a beat later; bridge that gap instead of hard-failing
# a live run into it. Short enough that a genuinely-closed window still fails
# quickly (and no LLM turns are ever burned waiting); long enough to ride out a
# reconnect even on a loaded machine.
_WS_RECONNECT_WAIT_S = 8.0
P_WS_RECONNECT_WAIT_S = 8.0
async def _await_reconnect(has_conn) -> bool:
"""Poll up to _WS_RECONNECT_WAIT_S for a dashboard socket to (re)appear.
async def await_reconnect(has_conn) -> bool:
"""Poll up to P_WS_RECONNECT_WAIT_S for a dashboard socket to (re)appear.
`has_conn` is a 0-arg callable returning truthy when connected."""
if has_conn():
return True
waited = 0.0
while waited < _WS_RECONNECT_WAIT_S:
while waited < P_WS_RECONNECT_WAIT_S:
await asyncio.sleep(0.5)
waited += 0.5
if has_conn():
@@ -260,7 +260,7 @@ class ConnectionManager:
self, request_id: str, action: str, browser_id: str, params: dict, tab_id: str = ""
) -> dict:
"""Send a browser command to the frontend and wait for the result."""
if not self.global_connections and not await _await_reconnect(lambda: bool(self.global_connections)):
if not self.global_connections and not await await_reconnect(lambda: bool(self.global_connections)):
return {"error": "No dashboard is connected. Open the dashboard to use browser tools."}
loop = asyncio.get_event_loop()
@@ -282,7 +282,7 @@ class ConnectionManager:
# so it gets a longer leash; everything else fails fast. A one-off slow
# command just times out and the next success resets the agent's streak,
# so only a SUSTAINED hang trips the fast-fail abort.
timeout = _BROWSER_CMD_TIMEOUTS.get(action, _BROWSER_CMD_TIMEOUT_DEFAULT)
timeout = BROWSER_CMD_TIMEOUTS.get(action, BROWSER_CMD_TIMEOUT_DEFAULT)
deadline = loop.time() + timeout
# Re-broadcast until a client answers: a silently-dead dashboard
# socket takes up to ~35s of heartbeat to notice, and a command
@@ -294,7 +294,7 @@ class ConnectionManager:
if remaining <= 0:
return {"error": "Browser command timed out"}
done, _ = await asyncio.wait(
{future}, timeout=min(_BROWSER_CMD_REBROADCAST_S, remaining)
{future}, timeout=min(BROWSER_CMD_REBROADCAST_S, remaining)
)
if done:
return future.result()
backend/apps/agents/providers/openrouter.py
+18 -18
View File
@@ -10,11 +10,11 @@ OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
# `or:` prefix on picker values so resolve_model_id_for_sdk recognises them
# without a side-table.
_OPENROUTER_VALUE_PREFIX = "or:"
OPENROUTER_VALUE_PREFIX = "or:"
_OR_MODELS_TTL_OK = 3600.0
_OR_MODELS_TTL_FAIL = 30.0
_or_models_cache: dict = {"models": None, "fetched_at": 0.0, "ok": False}
P_OR_MODELS_TTL_OK = 3600.0
P_OR_MODELS_TTL_FAIL = 30.0
p_or_models_cache: dict = {"models": None, "fetched_at": 0.0, "ok": False}
_9router_cache: dict = {"available": None, "checked_at": 0}
@@ -25,7 +25,7 @@ _9router_cache: dict = {"available": None, "checked_at": 0}
# Anthropic rates; for any non-Anthropic upstream the SDK number is
# 50-1000x wrong and we MUST recompute. Used by agent_manager's cost
# recompute logic.
_DIRECT_API_PRICING: dict[str, tuple[float, float]] = {
P_DIRECT_API_PRICING: dict[str, tuple[float, float]] = {
# OpenAI GPT-5.x family (source: platform.openai.com/docs/pricing).
"gpt-5.5": (1.25, 10.00),
"gpt-5.4": (1.25, 10.00),
@@ -53,7 +53,7 @@ def get_direct_pricing(model_id: str) -> tuple[float, float] | None:
if bare.startswith(prefix):
bare = bare[len(prefix):]
break
return _DIRECT_API_PRICING.get(bare)
return P_DIRECT_API_PRICING.get(bare)
def get_openrouter_pricing(resolved_model: str) -> tuple[float, float] | None:
@@ -61,7 +61,7 @@ def get_openrouter_pricing(resolved_model: str) -> tuple[float, float] | None:
if not isinstance(resolved_model, str) or not resolved_model.startswith("openrouter/"):
return None
bare = resolved_model[len("openrouter/"):]
for m in _or_models_cache.get("models") or []:
for m in p_or_models_cache.get("models") or []:
if m.get("model_id") == bare:
return (
float(m.get("input_cost_per_1m", 0.0)),
@@ -71,9 +71,9 @@ def get_openrouter_pricing(resolved_model: str) -> tuple[float, float] | None:
def invalidate_openrouter_cache() -> None:
_or_models_cache["models"] = None
_or_models_cache["fetched_at"] = 0.0
_or_models_cache["ok"] = False
p_or_models_cache["models"] = None
p_or_models_cache["fetched_at"] = 0.0
p_or_models_cache["ok"] = False
async def fetch_openrouter_models(api_key: str | None) -> list[dict]:
@@ -84,11 +84,11 @@ async def fetch_openrouter_models(api_key: str | None) -> list[dict]:
return []
now = _time.monotonic()
fetched_at = _or_models_cache["fetched_at"]
if _or_models_cache["models"] is not None:
ttl = _OR_MODELS_TTL_OK if _or_models_cache["ok"] else _OR_MODELS_TTL_FAIL
fetched_at = p_or_models_cache["fetched_at"]
if p_or_models_cache["models"] is not None:
ttl = P_OR_MODELS_TTL_OK if p_or_models_cache["ok"] else P_OR_MODELS_TTL_FAIL
if now - fetched_at < ttl:
return _or_models_cache["models"]
return p_or_models_cache["models"]
import httpx
try:
@@ -98,12 +98,12 @@ async def fetch_openrouter_models(api_key: str | None) -> list[dict]:
headers={"Authorization": f"Bearer {api_key}"},
)
if r.status_code != 200:
_or_models_cache.update(models=[], fetched_at=now, ok=False)
p_or_models_cache.update(models=[], fetched_at=now, ok=False)
logger.debug(f"OpenRouter /models returned {r.status_code}")
return []
raw = r.json().get("data") or []
except Exception as e:
_or_models_cache.update(models=[], fetched_at=now, ok=False)
p_or_models_cache.update(models=[], fetched_at=now, ok=False)
logger.debug(f"OpenRouter /models fetch failed: {e}")
return []
@@ -155,7 +155,7 @@ async def fetch_openrouter_models(api_key: str | None) -> list[dict]:
except (TypeError, ValueError):
max_completion = None
out.append({
"value": f"{_OPENROUTER_VALUE_PREFIX}{model_id}",
"value": f"{OPENROUTER_VALUE_PREFIX}{model_id}",
"label": label,
"context_window": ctx,
"model_id": model_id,
@@ -170,5 +170,5 @@ async def fetch_openrouter_models(api_key: str | None) -> list[dict]:
"max_completion_tokens": max_completion,
})
_or_models_cache.update(models=out, fetched_at=now, ok=True)
p_or_models_cache.update(models=out, fetched_at=now, ok=True)
return out
backend/apps/agents/providers/registry.py
+3 -3
View File
@@ -12,7 +12,7 @@ import logging
from typing import Any, TYPE_CHECKING
from .openrouter import (
_OPENROUTER_VALUE_PREFIX,
OPENROUTER_VALUE_PREFIX,
fetch_openrouter_models,
get_direct_pricing,
get_openrouter_pricing,
@@ -189,8 +189,8 @@ def find_builtin_model(short_name: str) -> dict | None:
for m in models:
if m.get("value") == short_name:
return m
if isinstance(short_name, str) and short_name.startswith(_OPENROUTER_VALUE_PREFIX):
bare = short_name[len(_OPENROUTER_VALUE_PREFIX):]
if isinstance(short_name, str) and short_name.startswith(OPENROUTER_VALUE_PREFIX):
bare = short_name[len(OPENROUTER_VALUE_PREFIX):]
if bare:
return {
"value": short_name,
backend/apps/agents/proxy/anthropic_proxy.py
+50 -50
View File
@@ -21,7 +21,7 @@ async def anthropic_proxy_lifespan():
anthropic_proxy = SubApp("anthropic-proxy", anthropic_proxy_lifespan)
_CLAUDE_MODEL_PREFIXES = (
P_CLAUDE_MODEL_PREFIXES = (
"claude-",
"claude/",
"sonnet",
@@ -30,10 +30,10 @@ _CLAUDE_MODEL_PREFIXES = (
"cc/",
)
_GEMINI_MODEL_PREFIXES = ("gemini/", "gc/", "ag/")
P_GEMINI_MODEL_PREFIXES = ("gemini/", "gc/", "ag/")
# Own-key Gemini ("gemini-3-flash-api" etc.) skips the gemini/ prefix; match bare names so $schema scrub still fires.
_GEMINI_BARE_MODEL_PATTERNS = ("gemini-",)
P_GEMINI_BARE_MODEL_PATTERNS = ("gemini-",)
# Gemini's function_declarations validator accepts only a small OpenAPI subset.
# A denylist was whack-a-mole: every new JSON Schema construct that slipped
@@ -43,20 +43,20 @@ _GEMINI_BARE_MODEL_PATTERNS = ("gemini-",)
# anyOf whose other branch is `{"type":"null"}`) into the `nullable` flag Gemini
# actually understands. Everything dropped is advisory; the model still reads it
# from `description`. The win is structural: an unknown future key can't 400 us.
_GEMINI_ALLOWED_SCHEMA_KEYS = {
P_GEMINI_ALLOWED_SCHEMA_KEYS = {
"type", "description", "nullable", "enum", "items", "properties",
"required", "minimum", "maximum", "minItems", "maxItems",
}
_GEMINI_NULL_TYPES = {"null", None}
P_GEMINI_NULL_TYPES = {"null", None}
def _normalize_schema_for_gemini(node):
def normalize_schema_for_gemini(node):
"""Allowlist-rewrite a JSON Schema node into the subset Gemini accepts.
Returns a NEW node (callers must assign the result); folds union/anyOf
nullability into `nullable`. Never raises on odd input."""
if isinstance(node, list):
return [_normalize_schema_for_gemini(v) for v in node]
return [normalize_schema_for_gemini(v) for v in node]
if not isinstance(node, dict):
return node
@@ -69,11 +69,11 @@ def _normalize_schema_for_gemini(node):
if isinstance(branches, list) and branches:
picked = None
for b in branches:
if isinstance(b, dict) and b.get("type") in _GEMINI_NULL_TYPES and len(b) == 1:
if isinstance(b, dict) and b.get("type") in P_GEMINI_NULL_TYPES and len(b) == 1:
nullable = True
elif picked is None:
picked = b
base = _normalize_schema_for_gemini(picked) if isinstance(picked, dict) else {}
base = normalize_schema_for_gemini(picked) if isinstance(picked, dict) else {}
if nullable and isinstance(base, dict):
base["nullable"] = True
return base
@@ -81,7 +81,7 @@ def _normalize_schema_for_gemini(node):
out = {}
t = node.get("type")
if isinstance(t, list): # ["string", "null"] -> "string" + nullable
non_null = [x for x in t if x not in _GEMINI_NULL_TYPES]
non_null = [x for x in t if x not in P_GEMINI_NULL_TYPES]
if len(non_null) != len(t):
nullable = True
t = non_null[0] if non_null else None
@@ -89,12 +89,12 @@ def _normalize_schema_for_gemini(node):
out["type"] = t
for k, v in node.items():
if k in ("type", "nullable") or k not in _GEMINI_ALLOWED_SCHEMA_KEYS:
if k in ("type", "nullable") or k not in P_GEMINI_ALLOWED_SCHEMA_KEYS:
continue
if k == "properties" and isinstance(v, dict):
out[k] = {pk: _normalize_schema_for_gemini(pv) for pk, pv in v.items()}
out[k] = {pk: normalize_schema_for_gemini(pv) for pk, pv in v.items()}
elif k == "items":
out[k] = _normalize_schema_for_gemini(v)
out[k] = normalize_schema_for_gemini(v)
else:
out[k] = v
@@ -104,10 +104,10 @@ def _normalize_schema_for_gemini(node):
# GPT-5.x rejects max_tokens; needs max_completion_tokens. Anthropic-format wire still emits max_tokens; we rename on the way out.
_OPENAI_MAX_COMPLETION_TOKENS_MODELS = ("gpt-5",)
P_OPENAI_MAX_COMPLETION_TOKENS_MODELS = ("gpt-5",)
def _is_openai_max_completion_tokens_model(model: str) -> bool:
def p_is_openai_max_completion_tokens_model(model: str) -> bool:
"""Match every shape a GPT-5 name might arrive in (bare, api-suffixed, openai/-prefixed, cx/-routed)."""
m = (model or "").strip().lower()
if not m:
@@ -116,10 +116,10 @@ def _is_openai_max_completion_tokens_model(model: str) -> bool:
if m.startswith(prefix):
m = m[len(prefix):]
break
return any(m.startswith(p) for p in _OPENAI_MAX_COMPLETION_TOKENS_MODELS)
return any(m.startswith(p) for p in P_OPENAI_MAX_COMPLETION_TOKENS_MODELS)
def _rewrite_document_to_openai_file(parsed: dict) -> None:
def p_rewrite_document_to_openai_file(parsed: dict) -> None:
"""In-place: rewrite Anthropic base64 `image` blocks to OpenAI `image_url` (PDFs go via OpenRouter)."""
msgs = parsed.get("messages") if isinstance(parsed, dict) else None
if not isinstance(msgs, list):
@@ -160,7 +160,7 @@ def _rewrite_document_to_openai_file(parsed: dict) -> None:
}
def _scrub_request_for_openai_gpt5(body: bytes) -> bytes:
def scrub_request_for_openai_gpt5(body: bytes) -> bytes:
"""Rename max_tokens→max_completion_tokens for GPT-5 AND rewrite any
Anthropic document blocks to OpenAI type:file shape so PDFs flow
natively on GPT-5.x vision models. Bytes in/out; never raises."""
@@ -190,7 +190,7 @@ def _scrub_request_for_openai_gpt5(body: bytes) -> bytes:
mutated = True
try:
before = json.dumps(parsed.get("messages"), sort_keys=True) if "messages" in parsed else ""
_rewrite_document_to_openai_file(parsed)
p_rewrite_document_to_openai_file(parsed)
after = json.dumps(parsed.get("messages"), sort_keys=True) if "messages" in parsed else ""
if before != after:
mutated = True
@@ -199,7 +199,7 @@ def _scrub_request_for_openai_gpt5(body: bytes) -> bytes:
return json.dumps(parsed).encode("utf-8") if mutated else body
def _rewrite_document_to_image(parsed: dict) -> None:
def p_rewrite_document_to_image(parsed: dict) -> None:
"""In-place: rewrite Anthropic `document` (PDF) AND `image` content
blocks → OpenAI `image_url` shape with a `data:` URL. Critical fix
for 9router 0.3.60 which **only translates `image_url` blocks** to
@@ -245,15 +245,15 @@ def _rewrite_document_to_image(parsed: dict) -> None:
}
_OPENROUTER_MODEL_PREFIXES = ("openrouter/", "or:")
P_OPENROUTER_MODEL_PREFIXES = ("openrouter/", "or:")
def _is_openrouter_model(model: str) -> bool:
def p_is_openrouter_model(model: str) -> bool:
m = (model or "").strip().lower()
return any(m.startswith(p) for p in _OPENROUTER_MODEL_PREFIXES)
return any(m.startswith(p) for p in P_OPENROUTER_MODEL_PREFIXES)
def _inject_openrouter_file_parser(body: bytes) -> bytes:
def inject_openrouter_file_parser(body: bytes) -> bytes:
"""When the request has document blocks AND is bound for OpenRouter,
inject the file-parser plugin so OR's universal PDF support kicks in
on any model (free models get pdf-text engine; native PDF models can
@@ -293,7 +293,7 @@ def _inject_openrouter_file_parser(body: bytes) -> bytes:
return json.dumps(parsed).encode("utf-8")
def _scrub_request_for_gemini(body: bytes) -> bytes:
def scrub_request_for_gemini(body: bytes) -> bytes:
"""Strip Gemini-incompatible schema keys from request tools AND
rewrite Anthropic document blocks to image-shape so 9router's
inline_data translator picks them up. Bytes-in/out, never raises."""
@@ -309,19 +309,19 @@ def _scrub_request_for_gemini(body: bytes) -> bytes:
if not isinstance(t, dict):
continue
if isinstance(t.get("input_schema"), (dict, list)):
t["input_schema"] = _normalize_schema_for_gemini(t["input_schema"])
t["input_schema"] = normalize_schema_for_gemini(t["input_schema"])
if isinstance(t.get("parameters"), (dict, list)):
t["parameters"] = _normalize_schema_for_gemini(t["parameters"])
t["parameters"] = normalize_schema_for_gemini(t["parameters"])
try:
if isinstance(parsed, dict):
_rewrite_document_to_image(parsed)
p_rewrite_document_to_image(parsed)
except Exception:
pass
return json.dumps(parsed).encode("utf-8")
# Hop-by-hop headers or auth we replace with the upstream-specific value.
_HOP_HEADERS = {
P_HOP_HEADERS = {
"host",
"content-length",
"authorization",
@@ -337,22 +337,22 @@ _HOP_HEADERS = {
}
def _is_claude_model(model: str) -> bool:
def p_is_claude_model(model: str) -> bool:
m = (model or "").strip().lower()
return m.startswith(_CLAUDE_MODEL_PREFIXES)
return m.startswith(P_CLAUDE_MODEL_PREFIXES)
def _is_gemini_model(model: str) -> bool:
def p_is_gemini_model(model: str) -> bool:
m = (model or "").strip().lower()
if m.startswith(_GEMINI_MODEL_PREFIXES):
if m.startswith(P_GEMINI_MODEL_PREFIXES):
return True
# Bare-name match for own-key Gemini; excludes anthropic-routed gemini (those carry "/").
if "/" in m:
return False
return any(m.startswith(p) for p in _GEMINI_BARE_MODEL_PATTERNS)
return any(m.startswith(p) for p in P_GEMINI_BARE_MODEL_PATTERNS)
def _pick_upstream(model: str) -> tuple[str, dict[str, str]]:
def p_pick_upstream(model: str) -> tuple[str, dict[str, str]]:
"""Return (base_url_without_v1, auth_headers) for this model.
Routing for Claude-family models:
@@ -364,7 +364,7 @@ def _pick_upstream(model: str) -> tuple[str, dict[str, str]]:
from backend.apps.settings.settings import load_settings
s = load_settings()
if _is_claude_model(model):
if p_is_claude_model(model):
if getattr(s, "connection_mode", "own_key") == "openswarm-pro":
bearer = getattr(s, "openswarm_bearer_token", "") or ""
proxy = (getattr(s, "openswarm_proxy_url", "") or "https://api.openswarm.com").rstrip("/")
@@ -390,7 +390,7 @@ def _pick_upstream(model: str) -> tuple[str, dict[str, str]]:
methods=["GET", "HEAD", "OPTIONS"],
include_in_schema=False,
)
async def _healthcheck():
async def p_healthcheck():
"""CLI healthchecks the proxy root; return 200 so it doesn't 404."""
return {"ok": True}
@@ -427,7 +427,7 @@ async def proxy(rest: str, request: Request):
)
from backend.apps.settings.settings import load_settings as _load
_s = _load()
if _is_openai_max_completion_tokens_model(model):
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):
status, body_stream, hdrs = await _forward_oai(
@@ -437,7 +437,7 @@ async def proxy(rest: str, request: Request):
body_stream, status_code=status, headers=hdrs,
media_type=hdrs.get("content-type", "text/event-stream"),
)
if _is_openrouter_model(model):
if p_is_openrouter_model(model):
_ork = (getattr(_s, "openrouter_api_key", "") or "").strip()
if _should_bypass_or(parsed_for_bypass, _ork):
status, body_stream, hdrs = await _forward_or(
@@ -448,18 +448,18 @@ async def proxy(rest: str, request: Request):
media_type=hdrs.get("content-type", "text/event-stream"),
)
if _is_gemini_model(model):
body = _scrub_request_for_gemini(body)
if _is_openai_max_completion_tokens_model(model):
body = _scrub_request_for_openai_gpt5(body)
if _is_openrouter_model(model):
body = _inject_openrouter_file_parser(body)
if p_is_gemini_model(model):
body = scrub_request_for_gemini(body)
if p_is_openai_max_completion_tokens_model(model):
body = scrub_request_for_openai_gpt5(body)
if p_is_openrouter_model(model):
body = inject_openrouter_file_parser(body)
base_url, auth_headers = _pick_upstream(model)
base_url, auth_headers = p_pick_upstream(model)
forward_headers: dict[str, str] = {}
for k, v in request.headers.items():
if k.lower() in _HOP_HEADERS:
if k.lower() in P_HOP_HEADERS:
continue
# CLI carries our install token as x-api-key; never forward (leak + shadows real upstream auth).
if k.lower() == "x-api-key":
@@ -479,7 +479,7 @@ 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 _is_gemini_model(model) else 600.0
_read_timeout = 120.0 if p_is_gemini_model(model) else 600.0
try:
if wants_stream:
@@ -503,7 +503,7 @@ async def proxy(rest: str, request: Request):
streamer(),
status_code=upstream.status_code,
headers={k: v for k, v in upstream.headers.items()
if k.lower() not in _HOP_HEADERS},
if k.lower() not in P_HOP_HEADERS},
media_type=upstream.headers.get("content-type", "text/event-stream"),
)
else:
@@ -515,7 +515,7 @@ async def proxy(rest: str, request: Request):
return JSONResponse(
content=r.json() if r.headers.get("content-type", "").startswith("application/json") else {"raw": r.text},
status_code=r.status_code,
headers={k: v for k, v in r.headers.items() if k.lower() not in _HOP_HEADERS},
headers={k: v for k, v in r.headers.items() if k.lower() not in P_HOP_HEADERS},
)
except httpx.TimeoutException:
return JSONResponse({"error": "upstream timeout"}, status_code=504)
backend/apps/agents/proxy/anthropic_to_openai.py
+29 -29
View File
@@ -27,8 +27,8 @@ import httpx
logger = logging.getLogger(__name__)
_OPENAI_UPSTREAM = "https://api.openai.com/v1"
_OPENROUTER_UPSTREAM = "https://openrouter.ai/api/v1"
P_OPENAI_UPSTREAM = "https://api.openai.com/v1"
P_OPENROUTER_UPSTREAM = "https://openrouter.ai/api/v1"
# Concurrency cap for bypass-route requests. Each in-flight request holds
# the base64'd PDF (raw_bytes * 1.33) in memory across httpx's request
@@ -38,18 +38,18 @@ _OPENROUTER_UPSTREAM = "https://openrouter.ai/api/v1"
# concurrent probes. Cap at 2 so a Mehmet-style multi-PDF attach in one
# session can't take the whole backend down. Requests above the cap
# queue rather than fail.
_BYPASS_CONCURRENCY = 2
_bypass_sema = asyncio.Semaphore(_BYPASS_CONCURRENCY)
BYPASS_CONCURRENCY = 2
bypass_sema = asyncio.Semaphore(BYPASS_CONCURRENCY)
# Hard per-request body size ceiling. Anthropic API caps at 32MB,
# OpenAI Chat Completions at 50MB, OpenRouter at whatever underlying
# model accepts. We refuse anything over 40MB raw (≈53MB base64) before
# we even build the request body, so a malicious or accidental huge
# attach never reaches the in-memory pipeline.
_BYPASS_MAX_RAW_BYTES = 40 * 1024 * 1024
P_BYPASS_MAX_RAW_BYTES = 40 * 1024 * 1024
def _has_document_block(parsed: dict) -> bool:
def p_has_document_block(parsed: dict) -> bool:
msgs = parsed.get("messages")
if not isinstance(msgs, list):
return False
@@ -76,7 +76,7 @@ def should_bypass_9router(parsed: dict, api_key: str | None) -> bool:
return False
if parsed.get("tools"):
return False
return _has_document_block(parsed)
return p_has_document_block(parsed)
def should_bypass_9router_for_openrouter(parsed: dict, api_key: str | None) -> bool:
@@ -91,10 +91,10 @@ def should_bypass_9router_for_openrouter(parsed: dict, api_key: str | None) -> b
return False
if parsed.get("tools"):
return False
return _has_document_block(parsed)
return p_has_document_block(parsed)
def _content_blocks_to_openai(content) -> list[dict]:
def p_content_blocks_to_openai(content) -> list[dict]:
"""Convert Anthropic content blocks → OpenAI Chat Completions parts."""
if isinstance(content, str):
return [{"type": "text", "text": content}]
@@ -161,7 +161,7 @@ def translate_request(parsed: dict) -> dict:
role = m.get("role")
if role not in ("user", "assistant"):
continue
msgs_out.append({"role": role, "content": _content_blocks_to_openai(m.get("content"))})
msgs_out.append({"role": role, "content": p_content_blocks_to_openai(m.get("content"))})
openai_body["messages"] = msgs_out
mt = parsed.get("max_tokens")
@@ -177,12 +177,12 @@ def translate_request(parsed: dict) -> dict:
return openai_body
def _sse_event(event: str, data: dict) -> bytes:
def p_sse_event(event: str, data: dict) -> bytes:
"""Encode an Anthropic-format SSE event."""
return f"event: {event}\ndata: {json.dumps(data)}\n\n".encode("utf-8")
async def _translate_response_stream(
async def p_translate_response_stream(
upstream: httpx.Response, model: str,
) -> AsyncIterator[bytes]:
"""Convert OpenAI Chat Completions SSE → Anthropic Messages SSE.
@@ -227,7 +227,7 @@ async def _translate_response_stream(
if not started:
usage = (ev.get("usage") or {})
input_tokens = int(usage.get("prompt_tokens") or 0)
yield _sse_event("message_start", {
yield p_sse_event("message_start", {
"type": "message_start",
"message": {
"id": msg_id,
@@ -256,13 +256,13 @@ async def _translate_response_stream(
delta_text = delta.get("content")
if isinstance(delta_text, str) and delta_text:
if not block_opened:
yield _sse_event("content_block_start", {
yield p_sse_event("content_block_start", {
"type": "content_block_start",
"index": 0,
"content_block": {"type": "text", "text": ""},
})
block_opened = True
yield _sse_event("content_block_delta", {
yield p_sse_event("content_block_delta", {
"type": "content_block_delta",
"index": 0,
"delta": {"type": "text_delta", "text": delta_text},
@@ -278,15 +278,15 @@ async def _translate_response_stream(
finally:
if started:
if block_opened:
yield _sse_event("content_block_stop", {
yield p_sse_event("content_block_stop", {
"type": "content_block_stop", "index": 0,
})
yield _sse_event("message_delta", {
yield p_sse_event("message_delta", {
"type": "message_delta",
"delta": {"stop_reason": stop_reason, "stop_sequence": None},
"usage": {"input_tokens": input_tokens, "output_tokens": output_tokens},
})
yield _sse_event("message_stop", {"type": "message_stop"})
yield p_sse_event("message_stop", {"type": "message_stop"})
async def forward_to_openai(
@@ -295,7 +295,7 @@ async def forward_to_openai(
"""Translate + forward an Anthropic request to OpenAI Chat Completions.
Returns (status, body_stream, response_headers)."""
openai_body = translate_request(parsed)
return await _forward(openai_body, api_key, f"{_OPENAI_UPSTREAM}/chat/completions")
return await p_forward(openai_body, api_key, f"{P_OPENAI_UPSTREAM}/chat/completions")
async def forward_to_openrouter(
@@ -312,10 +312,10 @@ async def forward_to_openrouter(
break
openai_body["model"] = bare
openai_body["plugins"] = [{"id": "file-parser", "pdf": {"engine": "pdf-text"}}]
return await _forward(openai_body, api_key, f"{_OPENROUTER_UPSTREAM}/chat/completions")
return await p_forward(openai_body, api_key, f"{P_OPENROUTER_UPSTREAM}/chat/completions")
def _estimate_body_bytes(body_json: dict) -> int:
def estimate_body_bytes(body_json: dict) -> int:
"""Sum the base64 payload bytes across content blocks. Used as a
cheap pre-flight check before httpx serializes the body."""
total = 0
@@ -337,12 +337,12 @@ def _estimate_body_bytes(body_json: dict) -> int:
return total
async def _forward(
async def p_forward(
body_json: dict, api_key: str, url: str,
) -> tuple[int, AsyncIterator[bytes], dict[str, str]]:
# Pre-flight size check. base64 expands ~4/3 so 40MB raw → 53MB b64.
raw_estimate = int(_estimate_body_bytes(body_json) * 0.75)
if raw_estimate > _BYPASS_MAX_RAW_BYTES:
raw_estimate = int(estimate_body_bytes(body_json) * 0.75)
if raw_estimate > P_BYPASS_MAX_RAW_BYTES:
async def reject():
payload = json.dumps({
@@ -351,7 +351,7 @@ async def _forward(
"type": "invalid_request_error",
"message": (
f"Attached files total ~{raw_estimate // (1024*1024)} MB, "
f"over the {_BYPASS_MAX_RAW_BYTES // (1024*1024)} MB per-request "
f"over the {P_BYPASS_MAX_RAW_BYTES // (1024*1024)} MB per-request "
"cap on this provider lane. Detach a file or split across "
"separate turns."
),
@@ -372,14 +372,14 @@ async def _forward(
# ~40MB request body + a streaming response buffer, and the OS
# OOM-kills the backend (observed on macOS during a 3-PDF probe
# burst). Semaphore serializes excess requests instead of failing.
await _bypass_sema.acquire()
await bypass_sema.acquire()
client = httpx.AsyncClient(timeout=httpx.Timeout(600.0, connect=30.0))
try:
req = client.build_request("POST", url, json=body_json, headers=headers)
upstream = await client.send(req, stream=True)
except Exception:
await client.aclose()
_bypass_sema.release()
bypass_sema.release()
raise
async def streamer():
@@ -388,7 +388,7 @@ async def _forward(
raw = await upstream.aread()
yield raw
return
async for chunk in _translate_response_stream(upstream, body_json["model"]):
async for chunk in p_translate_response_stream(upstream, body_json["model"]):
yield chunk
finally:
try:
@@ -397,7 +397,7 @@ async def _forward(
try:
await client.aclose()
finally:
_bypass_sema.release()
bypass_sema.release()
return upstream.status_code, streamer(), {
"content-type": "text/event-stream" if upstream.status_code < 400 else "application/json",
backend/apps/agents/session_credential.py
+10 -10
View File
@@ -37,7 +37,7 @@ if TYPE_CHECKING:
# AppSettings fields holding a user-writable API key, keyed by provider api-type.
# Blanking whichever of these powers the current run is the one suicide the guard
# stops. Anything not here (subscription tokens, bearers) is not settings-writable.
_API_KEY_FIELD_BY_API: dict[str, str] = {
P_API_KEY_FIELD_BY_API: dict[str, str] = {
"anthropic": "anthropic_api_key",
"openai": "openai_api_key",
"codex": "openai_api_key",
@@ -47,7 +47,7 @@ _API_KEY_FIELD_BY_API: dict[str, str] = {
# Every settings field that can hold an API key (the full guarded set). Custom
# providers keep their keys inside the custom_providers list, guarded separately.
ALL_API_KEY_FIELDS: frozenset[str] = frozenset(_API_KEY_FIELD_BY_API.values())
ALL_API_KEY_FIELDS: frozenset[str] = frozenset(P_API_KEY_FIELD_BY_API.values())
CredentialKind = Literal["api_key", "subscription", "unknown"]
@@ -72,7 +72,7 @@ class PoweringCredential:
label: str = ""
def _custom_slug_for_model(model_value: str, settings: AppSettings) -> str | None:
def p_custom_slug_for_model(model_value: str, settings: AppSettings) -> str | None:
cp = find_custom_provider_for_value(settings, model_value)
if cp is not None:
return custom_provider_slug_for_lookup(getattr(cp, "name", ""))
@@ -98,7 +98,7 @@ def resolve_powering_credential(model_value: str, settings: AppSettings) -> Powe
# placeholder key, so suicide is removing the provider ENTRY, not blanking
# its key; the guard keys off the slug.
if api == "custom":
slug = _custom_slug_for_model(model_value, settings)
slug = p_custom_slug_for_model(model_value, settings)
return PoweringCredential(
kind="api_key", provider="custom",
protected_custom_slug=slug,
@@ -107,7 +107,7 @@ def resolve_powering_credential(model_value: str, settings: AppSettings) -> Powe
# Explicit API-key route: the matching *_api_key field is the live one.
if route == "api":
field = _API_KEY_FIELD_BY_API.get(api)
field = P_API_KEY_FIELD_BY_API.get(api)
if field:
return PoweringCredential(kind="api_key", provider=api, protected_field=field,
label=f"{field} (powers this run)")
@@ -151,7 +151,7 @@ def resolve_powering_credential(model_value: str, settings: AppSettings) -> Powe
label=f"{api or 'unknown'} provider (unclassified)")
def _is_blank(value: Any) -> bool:
def p_is_blank(value: Any) -> bool:
"""A credential write that removes the credential: None, "", or whitespace."""
if value is None:
return True
@@ -160,7 +160,7 @@ def _is_blank(value: Any) -> bool:
return False
def _powering_custom_slug_present(new_providers: Any, slug: str) -> bool:
def p_powering_custom_slug_present(new_providers: Any, slug: str) -> bool:
"""True if the powering custom provider's entry still exists after the write."""
if not isinstance(new_providers, list):
return False
@@ -183,13 +183,13 @@ def write_would_suicide(field: str, new_value: Any, powering: PoweringCredential
# local provider's placeholder key being blanked is not. When the run is
# unknown, any custom run could be the live one, so refuse a vanish.
if powering.kind == "api_key" and powering.provider == "custom" and powering.protected_custom_slug:
return not _powering_custom_slug_present(new_value, powering.protected_custom_slug)
return not p_powering_custom_slug_present(new_value, powering.protected_custom_slug)
if powering.kind == "unknown":
return not _powering_custom_slug_present(new_value, powering.protected_custom_slug or "")
return not p_powering_custom_slug_present(new_value, powering.protected_custom_slug or "")
return False
if field in ALL_API_KEY_FIELDS:
if not _is_blank(new_value):
if not p_is_blank(new_value):
return False
if powering.kind == "unknown":
return True
backend/apps/agents/settings_meta_server.py
+4 -4
View File
@@ -92,7 +92,7 @@ def call_backend(action: str, payload: dict) -> dict:
return {"error": str(e)}
def _format_read(settings: dict) -> str:
def p_format_read(settings: dict) -> str:
"""Render redacted settings compactly so the model spends tokens on the
values it can act on, not on JSON punctuation."""
lines = ["Current OpenSwarm Settings (secrets shown as configured/not):"]
@@ -106,7 +106,7 @@ def _format_read(settings: dict) -> str:
return "\n".join(lines)
def _format_write(outcomes: dict) -> str:
def p_format_write(outcomes: dict) -> str:
applied = [f for f, o in outcomes.items() if o.get("status") == "applied"]
parts = []
if applied:
@@ -128,7 +128,7 @@ def handle_tool_call(tool_name: str, arguments: dict) -> dict:
result = call_backend("read", {})
if "error" in result:
return {"content": [{"type": "text", "text": f"Error: {result['error']}"}], "isError": True}
return {"content": [{"type": "text", "text": _format_read(result.get("settings", {}))}]}
return {"content": [{"type": "text", "text": p_format_read(result.get("settings", {}))}]}
if tool_name == "SettingsWrite":
changes = arguments.get("changes")
@@ -137,7 +137,7 @@ def handle_tool_call(tool_name: str, arguments: dict) -> dict:
result = call_backend("write", {"changes": changes})
if "error" in result:
return {"content": [{"type": "text", "text": f"Error: {result['error']}"}], "isError": True}
return {"content": [{"type": "text", "text": _format_write(result.get("outcomes", {}))}]}
return {"content": [{"type": "text", "text": p_format_write(result.get("outcomes", {}))}]}
return {"content": [{"type": "text", "text": f"Unknown tool: {tool_name}"}], "isError": True}
backend/apps/agents/web_mcp_server.py
+3 -3
View File
@@ -76,7 +76,7 @@ def send_response(id_, result=None, error=None):
sys.stdout.flush()
def _post(url: str, body: dict, timeout: float = 60.0) -> dict:
def p_post(url: str, body: dict, timeout: float = 60.0) -> dict:
payload = json.dumps(body).encode()
headers = {"Content-Type": "application/json"}
if BACKEND_AUTH:
@@ -107,7 +107,7 @@ def handle_tool_call(tool_name: str, arguments: dict) -> dict:
body = {"query": query, "num_results": num}
if PRIMARY_HINT:
body["primary"] = PRIMARY_HINT
r = _post(SEARCH_URL, body, timeout=45.0)
r = p_post(SEARCH_URL, body, timeout=45.0)
if "error" in r:
return {"content": [{"type": "text", "text": f"Search failed: {r['error']}"}], "isError": True}
results = r.get("results", "")
@@ -127,7 +127,7 @@ def handle_tool_call(tool_name: str, arguments: dict) -> dict:
body["prompt"] = str(prompt)
if PRIMARY_HINT:
body["primary"] = PRIMARY_HINT
r = _post(FETCH_URL, body, timeout=45.0)
r = p_post(FETCH_URL, body, timeout=45.0)
if "error" in r:
return {"content": [{"type": "text", "text": f"Fetch failed: {r['error']}"}], "isError": True}
content = r.get("content", "")
backend/apps/nine_router/oauth.py
+4 -4
View File
@@ -11,7 +11,7 @@ import os
import httpx
from .process import NINE_ROUTER_API, NINE_ROUTER_PORT, NINE_ROUTER_V1, cli_auth_headers
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__)
@@ -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,7 +129,7 @@ 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"
)
backend/apps/oauth_state.py
+9 -9
View File
@@ -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
# MAX_COMPLETED_OAUTH so it can't leak memory.
completed_oauth: list[str] = []
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) > MAX_COMPLETED_OAUTH:
completed_oauth.pop(0)
backend/main.py
+10 -10
View File
@@ -22,10 +22,10 @@ from fastapi.responses import JSONResponse, HTMLResponse
from fastapi import Request
from backend.apps.oauth_state import (
_pending_oauth,
_completed_oauth,
_MAX_COMPLETED_OAUTH,
_mark_oauth_completed,
pending_oauth,
completed_oauth,
MAX_COMPLETED_OAUTH,
mark_oauth_completed,
)
from backend.config.Apps import MainApp
from backend.apps.health.health import health
@@ -438,7 +438,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": "*"})
@@ -467,10 +467,10 @@ async def subscriptions_callback(request: Request):
Must be idempotent: the browser can legitimately hit this URL more than
once (Chrome prefetch, user refresh, Google retrying a slow first
redirect). The first call consumes `_pending_oauth[state]`, so a second
redirect). The first call consumes `pending_oauth[state]`, so a second
call would otherwise render a misleading "Session expired" even though
the connection is already saved. To handle that, we track recently-
completed state values in `_completed_oauth` and return the success
completed state values in `completed_oauth` and return the success
page whenever we see a duplicate.
"""
code = request.query_params.get("code", "")
@@ -486,12 +486,12 @@ async def subscriptions_callback(request: Request):
desc = html.escape(request.query_params.get("error_description", error))
return HTMLResponse(f'<html><body style="background:#1a1a1a;color:#fff;display:flex;align-items:center;justify-content:center;height:100vh;font-family:sans-serif"><div style="text-align:center"><h2>Authorization failed</h2><p style="color:#888">{desc}</p></div></body></html>')
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)'}...")
@@ -509,7 +509,7 @@ async def subscriptions_callback(request: Request):
safe_e = html.escape(str(e))
return HTMLResponse(f'<html><body style="background:#1a1a1a;color:#fff;display:flex;align-items:center;justify-content:center;height:100vh;font-family:sans-serif"><div style="text-align:center"><h2>Connection failed</h2><p style="color:#888">{safe_e}</p></div></body></html>')
_mark_oauth_completed(state)
mark_oauth_completed(state)
logger.info(f"OAuth exchange succeeded for provider={pending.get('provider')}")
return HTMLResponse(_SUCCESS_HTML)
backend/tests/test_browser_command_timeout.py
+11 -11
View File
@@ -26,19 +26,19 @@ def _mgr():
def test_timeout_map_reads_are_short_navigation_longer():
# reads/clicks act on a loaded page -> short; navigation loads network -> longer
assert wsm._BROWSER_CMD_TIMEOUT_DEFAULT <= 15
assert wsm._BROWSER_CMD_TIMEOUTS["navigate"] <= 25
assert wsm._BROWSER_CMD_TIMEOUTS["navigate"] > wsm._BROWSER_CMD_TIMEOUT_DEFAULT
assert wsm.BROWSER_CMD_TIMEOUT_DEFAULT <= 15
assert wsm.BROWSER_CMD_TIMEOUTS["navigate"] <= 25
assert wsm.BROWSER_CMD_TIMEOUTS["navigate"] > wsm.BROWSER_CMD_TIMEOUT_DEFAULT
# the old flat 30s is gone for the common path
assert wsm._BROWSER_CMD_TIMEOUT_DEFAULT < 30
assert wsm.BROWSER_CMD_TIMEOUT_DEFAULT < 30
@pytest.mark.asyncio
async def test_hung_command_returns_fast_at_the_bound(monkeypatch):
# shrink the bounds so the test is quick, then never resolve the future:
# the command must return a timeout error at ~the (default) bound, not hang.
monkeypatch.setattr(wsm, "_BROWSER_CMD_TIMEOUT_DEFAULT", 0.3)
monkeypatch.setattr(wsm, "_BROWSER_CMD_TIMEOUTS", {"navigate": 0.6})
monkeypatch.setattr(wsm, "BROWSER_CMD_TIMEOUT_DEFAULT", 0.3)
monkeypatch.setattr(wsm, "BROWSER_CMD_TIMEOUTS", {"navigate": 0.6})
m = _mgr()
t0 = time.monotonic()
res = await m.send_browser_command("rid1", "get_text", "b1", {}) # never resolved
@@ -49,8 +49,8 @@ async def test_hung_command_returns_fast_at_the_bound(monkeypatch):
@pytest.mark.asyncio
async def test_navigate_gets_the_longer_leash(monkeypatch):
monkeypatch.setattr(wsm, "_BROWSER_CMD_TIMEOUT_DEFAULT", 0.3)
monkeypatch.setattr(wsm, "_BROWSER_CMD_TIMEOUTS", {"navigate": 0.7})
monkeypatch.setattr(wsm, "BROWSER_CMD_TIMEOUT_DEFAULT", 0.3)
monkeypatch.setattr(wsm, "BROWSER_CMD_TIMEOUTS", {"navigate": 0.7})
m = _mgr()
t0 = time.monotonic()
await m.send_browser_command("rid2", "navigate", "b1", {"url": "x"})
@@ -62,8 +62,8 @@ async def test_navigate_gets_the_longer_leash(monkeypatch):
async def test_lost_first_delivery_heals_via_rebroadcast(monkeypatch):
# a silently-dead socket eats the first broadcast; the re-send after the
# rebroadcast interval must reach the (reconnected) client and succeed
monkeypatch.setattr(wsm, "_BROWSER_CMD_TIMEOUT_DEFAULT", 5.0)
monkeypatch.setattr(wsm, "_BROWSER_CMD_REBROADCAST_S", 0.1)
monkeypatch.setattr(wsm, "BROWSER_CMD_TIMEOUT_DEFAULT", 5.0)
monkeypatch.setattr(wsm, "BROWSER_CMD_REBROADCAST_S", 0.1)
m = _mgr()
sends = []
@@ -83,7 +83,7 @@ async def test_lost_first_delivery_heals_via_rebroadcast(monkeypatch):
@pytest.mark.asyncio
async def test_a_resolved_command_returns_immediately(monkeypatch):
# a healthy command returns the moment the renderer resolves it, not at the bound
monkeypatch.setattr(wsm, "_BROWSER_CMD_TIMEOUT_DEFAULT", 5.0)
monkeypatch.setattr(wsm, "BROWSER_CMD_TIMEOUT_DEFAULT", 5.0)
m = _mgr()
async def _resolve_soon():
backend/tests/test_browser_fast_path.py
+1 -1
View File
@@ -102,7 +102,7 @@ def test_dispatch_refused_when_no_dashboard_connected(monkeypatch):
# Dispatch now waits briefly for a momentary WS drop to reconnect; with a
# genuinely-closed window that wait just elapses and it still refuses without
# dispatching an agent or burning a turn. Zero the wait so the test is instant.
monkeypatch.setattr(wsm, "_WS_RECONNECT_WAIT_S", 0.0)
monkeypatch.setattr(wsm, "P_WS_RECONNECT_WAIT_S", 0.0)
assert not wsm.ws_manager.global_connections
results = asyncio.run(run_browser_agents(tasks=[{"task": "go to example.com"}], model="sonnet"))
assert len(results) == 1
backend/tests/test_v2_invariants.py
+36 -36
View File
@@ -1605,7 +1605,7 @@ def test_gemini_proxy_rewrites_document_to_openai_image_url_for_9router():
Anthropic-shape image/document blocks get stringified. We rewrite to
OpenAI image_url with data: URL so 9router emits Gemini inlineData."""
import json
from backend.apps.agents.proxy.anthropic_proxy import _scrub_request_for_gemini
from backend.apps.agents.proxy.anthropic_proxy import scrub_request_for_gemini
body = json.dumps({
"model": "gemini-3.1-pro-preview",
"messages": [{
@@ -1620,7 +1620,7 @@ def test_gemini_proxy_rewrites_document_to_openai_image_url_for_9router():
],
}],
}).encode("utf-8")
out = json.loads(_scrub_request_for_gemini(body))
out = json.loads(scrub_request_for_gemini(body))
blocks = out["messages"][0]["content"]
assert blocks[0]["type"] == "text"
assert blocks[1]["type"] == "image_url"
@@ -1631,7 +1631,7 @@ def test_gemini_proxy_also_rewrites_anthropic_image_blocks_to_image_url():
"""Same fix applies to plain images: Anthropic image → OpenAI image_url
with data: URL, so 9router's filter preserves it instead of stringifying."""
import json
from backend.apps.agents.proxy.anthropic_proxy import _scrub_request_for_gemini
from backend.apps.agents.proxy.anthropic_proxy import scrub_request_for_gemini
body = json.dumps({
"model": "gemini-3-pro-preview",
"messages": [{
@@ -1645,7 +1645,7 @@ def test_gemini_proxy_also_rewrites_anthropic_image_blocks_to_image_url():
],
}],
}).encode("utf-8")
out = json.loads(_scrub_request_for_gemini(body))
out = json.loads(scrub_request_for_gemini(body))
block = out["messages"][0]["content"][0]
assert block["type"] == "image_url"
assert block["image_url"]["url"] == "data:image/png;base64,iVBORw0KGgo="
@@ -1692,7 +1692,7 @@ def test_gemini_translated_block_matches_9router_image_url_filter():
(it stringifies any other shape). Our translator must emit exactly that
shape for PDFs and images both."""
import json
from backend.apps.agents.proxy.anthropic_proxy import _scrub_request_for_gemini
from backend.apps.agents.proxy.anthropic_proxy import scrub_request_for_gemini
body = json.dumps({
"model": "gemini-3.1-pro-preview",
"messages": [{"role": "user", "content": [
@@ -1703,7 +1703,7 @@ def test_gemini_translated_block_matches_9router_image_url_filter():
}},
]}],
}).encode("utf-8")
out = json.loads(_scrub_request_for_gemini(body))
out = json.loads(scrub_request_for_gemini(body))
block = out["messages"][0]["content"][0]
assert block["type"] == "image_url"
assert "image_url" in block
@@ -1719,16 +1719,16 @@ def test_gemini_schema_normalizer_allowlists_and_folds_nullable():
Live-confirmed against the Gemini API 2026-06-14."""
import json
from backend.apps.agents.proxy.anthropic_proxy import (
_normalize_schema_for_gemini, _scrub_request_for_gemini,
normalize_schema_for_gemini, scrub_request_for_gemini,
)
# union type -> single type + nullable
assert _normalize_schema_for_gemini({"type": ["string", "null"], "description": "d"}) == \
assert normalize_schema_for_gemini({"type": ["string", "null"], "description": "d"}) == \
{"type": "string", "description": "d", "nullable": True}
# anyOf-with-null -> chosen branch + nullable, allowed constraint preserved
assert _normalize_schema_for_gemini({"anyOf": [{"type": "integer", "minimum": 0}, {"type": "null"}]}) == \
assert normalize_schema_for_gemini({"anyOf": [{"type": "integer", "minimum": 0}, {"type": "null"}]}) == \
{"type": "integer", "minimum": 0, "nullable": True}
# forbidden keys dropped, enum kept
assert _normalize_schema_for_gemini({
assert normalize_schema_for_gemini({
"type": "object", "additionalProperties": False, "title": "T",
"properties": {"u": {"type": "string", "format": "uri", "$comment": "x", "minLength": 2},
"d": {"type": "string", "enum": ["a", "b"]}},
@@ -1749,7 +1749,7 @@ def test_gemini_schema_normalizer_allowlists_and_folds_nullable():
"size": {"type": ["integer", "null"], "minimum": 1, "default": 10},
"url": {"type": "string", "format": "uri", "$comment": "c"}},
"required": ["filter"]}}]}).encode()
schema = json.loads(_scrub_request_for_gemini(body))["tools"][0]["input_schema"]
schema = json.loads(scrub_request_for_gemini(body))["tools"][0]["input_schema"]
seen, stack = set(), [schema]
while stack:
n = stack.pop()
@@ -1765,12 +1765,12 @@ def test_gpt5_param_scrub_drops_unsupported_sampling_knobs():
penalty/logprobs family. Both the proxy and the passthrough must strip them.
Live-confirmed the 400s against the OpenAI API 2026-06-14."""
import json
from backend.apps.agents.proxy.anthropic_proxy import _scrub_request_for_openai_gpt5
from backend.apps.agents.proxy.anthropic_proxy import scrub_request_for_openai_gpt5
from backend.apps.agents.core.openai_passthrough import scrub_gpt5_params
dirty = json.dumps({"model": "gpt-5", "messages": [{"role": "user", "content": "hi"}],
"max_tokens": 200, "temperature": 0, "top_p": 0.9,
"frequency_penalty": 0.5, "presence_penalty": 0.1, "logprobs": True}).encode()
for fn in (_scrub_request_for_openai_gpt5, scrub_gpt5_params):
for fn in (scrub_request_for_openai_gpt5, scrub_gpt5_params):
out = json.loads(fn(dirty))
assert out.get("max_completion_tokens") == 200 and "max_tokens" not in out, fn.__name__
for k in ("temperature", "top_p", "frequency_penalty", "presence_penalty", "logprobs"):
@@ -1790,7 +1790,7 @@ def test_openrouter_plugin_array_matches_docs():
at the top level. Engines: pdf-text (free, deprecated → cloudflare),
mistral-ocr ($2/1k pages), native (model-supported)."""
import json
from backend.apps.agents.proxy.anthropic_proxy import _inject_openrouter_file_parser
from backend.apps.agents.proxy.anthropic_proxy import inject_openrouter_file_parser
body = json.dumps({
"model": "openrouter/qwen/qwen-2.5-72b-instruct",
"messages": [{"role": "user", "content": [
@@ -1799,7 +1799,7 @@ def test_openrouter_plugin_array_matches_docs():
}},
]}],
}).encode("utf-8")
out = json.loads(_inject_openrouter_file_parser(body))
out = json.loads(inject_openrouter_file_parser(body))
plugins = out["plugins"]
assert isinstance(plugins, list)
fp = [p for p in plugins if p.get("id") == "file-parser"][0]
@@ -1815,7 +1815,7 @@ def test_openai_translated_image_block_matches_image_url_data_uri():
translator rewrites Anthropic image blocks; document blocks are
refused upstream in agent_manager."""
import json
from backend.apps.agents.proxy.anthropic_proxy import _scrub_request_for_openai_gpt5
from backend.apps.agents.proxy.anthropic_proxy import scrub_request_for_openai_gpt5
body = json.dumps({
"model": "gpt-5.5",
"max_tokens": 100,
@@ -1827,7 +1827,7 @@ def test_openai_translated_image_block_matches_image_url_data_uri():
}},
]}],
}).encode("utf-8")
out = json.loads(_scrub_request_for_openai_gpt5(body))
out = json.loads(scrub_request_for_openai_gpt5(body))
block = out["messages"][0]["content"][0]
assert block["type"] == "image_url"
assert block["image_url"]["url"] == "data:image/png;base64,iVBORw0KGgo="
@@ -1837,7 +1837,7 @@ def test_openai_proxy_rewrites_image_block_only_documents_pass_through():
"""OpenAI image_url only accepts image/* mime; documents are refused
upstream. Translator handles images, leaves documents untouched."""
import json
from backend.apps.agents.proxy.anthropic_proxy import _scrub_request_for_openai_gpt5
from backend.apps.agents.proxy.anthropic_proxy import scrub_request_for_openai_gpt5
body = json.dumps({
"model": "gpt-5.5",
"max_tokens": 500,
@@ -1853,7 +1853,7 @@ def test_openai_proxy_rewrites_image_block_only_documents_pass_through():
],
}],
}).encode("utf-8")
out = json.loads(_scrub_request_for_openai_gpt5(body))
out = json.loads(scrub_request_for_openai_gpt5(body))
blocks = out["messages"][0]["content"]
assert blocks[0]["type"] == "text"
assert blocks[1]["type"] == "image_url"
@@ -1865,13 +1865,13 @@ def test_openai_proxy_rewrites_image_block_only_documents_pass_through():
def test_openai_proxy_skips_rewrite_when_no_document():
"""Pure text turn on GPT-5 should only get the max_tokens rename."""
import json
from backend.apps.agents.proxy.anthropic_proxy import _scrub_request_for_openai_gpt5
from backend.apps.agents.proxy.anthropic_proxy import scrub_request_for_openai_gpt5
body = json.dumps({
"model": "gpt-5.5",
"max_tokens": 100,
"messages": [{"role": "user", "content": "hi"}],
}).encode("utf-8")
out = json.loads(_scrub_request_for_openai_gpt5(body))
out = json.loads(scrub_request_for_openai_gpt5(body))
assert out["messages"][0]["content"] == "hi"
assert out.get("max_completion_tokens") == 100
@@ -1881,7 +1881,7 @@ def test_openai_proxy_defensive_on_malformed_document_blocks():
through untouched so the upstream returns a proper error rather
than us silently dropping the file."""
import json
from backend.apps.agents.proxy.anthropic_proxy import _scrub_request_for_openai_gpt5
from backend.apps.agents.proxy.anthropic_proxy import scrub_request_for_openai_gpt5
body = json.dumps({
"model": "gpt-5.5",
"messages": [{
@@ -1893,7 +1893,7 @@ def test_openai_proxy_defensive_on_malformed_document_blocks():
],
}],
}).encode("utf-8")
out = json.loads(_scrub_request_for_openai_gpt5(body))
out = json.loads(scrub_request_for_openai_gpt5(body))
for b in out["messages"][0]["content"]:
assert b["type"] == "document"
@@ -1941,7 +1941,7 @@ def test_openrouter_proxy_injects_file_parser_plugin_when_document_present():
"""OR's universal-PDF feature requires top-level plugins:[{id:file-parser,...}].
When a document block is in the request bound for OR, inject it."""
import json
from backend.apps.agents.proxy.anthropic_proxy import _inject_openrouter_file_parser
from backend.apps.agents.proxy.anthropic_proxy import inject_openrouter_file_parser
body = json.dumps({
"model": "openrouter/qwen/qwen-2.5-72b-instruct",
"messages": [{
@@ -1956,7 +1956,7 @@ def test_openrouter_proxy_injects_file_parser_plugin_when_document_present():
],
}],
}).encode("utf-8")
out = json.loads(_inject_openrouter_file_parser(body))
out = json.loads(inject_openrouter_file_parser(body))
plugins = out.get("plugins")
assert isinstance(plugins, list) and len(plugins) >= 1
fp = next((p for p in plugins if p.get("id") == "file-parser"), None)
@@ -1967,19 +1967,19 @@ def test_openrouter_proxy_skips_plugin_when_no_document():
"""No document block → don't inject the plugin (costs nothing, but
keeps the request body clean)."""
import json
from backend.apps.agents.proxy.anthropic_proxy import _inject_openrouter_file_parser
from backend.apps.agents.proxy.anthropic_proxy import inject_openrouter_file_parser
body = json.dumps({
"model": "openrouter/qwen/qwen-2.5-72b-instruct",
"messages": [{"role": "user", "content": "just a question"}],
}).encode("utf-8")
out = json.loads(_inject_openrouter_file_parser(body))
out = json.loads(inject_openrouter_file_parser(body))
assert "plugins" not in out
def test_openrouter_proxy_dedupes_existing_file_parser_plugin():
"""If a caller already provided file-parser, don't duplicate it."""
import json
from backend.apps.agents.proxy.anthropic_proxy import _inject_openrouter_file_parser
from backend.apps.agents.proxy.anthropic_proxy import inject_openrouter_file_parser
body = json.dumps({
"model": "openrouter/qwen/qwen-2.5-72b-instruct",
"plugins": [{"id": "file-parser", "pdf": {"engine": "mistral-ocr"}}],
@@ -1990,7 +1990,7 @@ def test_openrouter_proxy_dedupes_existing_file_parser_plugin():
],
}],
}).encode("utf-8")
out = json.loads(_inject_openrouter_file_parser(body))
out = json.loads(inject_openrouter_file_parser(body))
fps = [p for p in out["plugins"] if p.get("id") == "file-parser"]
assert len(fps) == 1
assert fps[0]["pdf"]["engine"] == "mistral-ocr" # caller's engine wins
@@ -2001,7 +2001,7 @@ def test_gemini_proxy_defensive_on_malformed_blocks():
must NOT be rewritten; they pass through so the upstream sees the
error rather than a silently-corrupted block."""
import json
from backend.apps.agents.proxy.anthropic_proxy import _scrub_request_for_gemini
from backend.apps.agents.proxy.anthropic_proxy import scrub_request_for_gemini
body = json.dumps({
"model": "gemini-3.1-pro-preview",
"messages": [{
@@ -2014,7 +2014,7 @@ def test_gemini_proxy_defensive_on_malformed_blocks():
],
}],
}).encode("utf-8")
out = json.loads(_scrub_request_for_gemini(body))
out = json.loads(scrub_request_for_gemini(body))
for b in out["messages"][0]["content"]:
assert b["type"] == "document"
@@ -2064,23 +2064,23 @@ def test_bypass_estimate_body_bytes_sums_image_url_and_file_blocks():
"""The size estimator must sum payload bytes across BOTH content
types the translator emits (image_url with data: URL, file with
file_data) so the pre-flight reject can fire before httpx serializes."""
from backend.apps.agents.proxy.anthropic_to_openai import _estimate_body_bytes
from backend.apps.agents.proxy.anthropic_to_openai import estimate_body_bytes
body = {"messages": [{"role": "user", "content": [
{"type": "text", "text": "hi"},
{"type": "image_url", "image_url": {"url": "data:image/png;base64,QUJDRA=="}}, # 8 bytes b64
{"type": "file", "file": {"file_data": "data:application/pdf;base64,RUZHSA=="}}, # 8 bytes b64
]}]}
assert _estimate_body_bytes(body) == 16
assert estimate_body_bytes(body) == 16
def test_bypass_concurrency_semaphore_serializes_excess_requests():
"""The semaphore caps in-flight bypass requests to prevent OOM. Cap=2
means a third concurrent request waits rather than allocating another
~40MB buffer."""
from backend.apps.agents.proxy.anthropic_to_openai import _bypass_sema, _BYPASS_CONCURRENCY
assert _BYPASS_CONCURRENCY == 2
from backend.apps.agents.proxy.anthropic_to_openai import bypass_sema, BYPASS_CONCURRENCY
assert BYPASS_CONCURRENCY == 2
# Initial value matches the cap (no in-flight at import time).
assert _bypass_sema._value == _BYPASS_CONCURRENCY
assert bypass_sema._value == BYPASS_CONCURRENCY
def test_anthropic_to_openai_should_bypass_fires_for_gpt5_pdf():