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

[haik]: refactor: standardize backend naming and import structure - strip leading underscores from all internal functions/constants across agents, settings, session, and provider modules in favor of public names or P_ prefix for module-private symbols; replace nine_router/init.py barrel re-exports with direct submodule imports (process, sync, sync_custom, oauth); hoist inline stdlib imports to module level; delete dead openai_passthrough.py GPT-5 proxy and the redundant _save_settings alias

Browse Source
This commit is contained in:
haikdc
2026-06-13 19:30:34 -07:00
parent 211f939560
commit f0d3cabc47
28 changed files with 384 additions and 606 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/agents/agent_manager.py
+103 -103
View File
@@ -33,41 +33,41 @@ from backend.apps.agents.core.error_classify import (
is_unknown_model_error,
)
from backend.apps.agents.manager.session.session_store import (
_delete_session_file,
_load_all_session_data,
_load_session_data,
_save_session,
delete_session_file,
load_all_session_data,
load_session_data,
save_session,
build_search_text,
)
from backend.apps.agents.manager.session.cloud_sync import _sync_session_close
from backend.apps.agents.manager.session.workspace_git import _detect_git_identity, _ensure_cwd_git_repo
from backend.apps.agents.manager.session.cloud_sync import sync_session_close
from backend.apps.agents.manager.session.workspace_git import detect_git_identity, ensure_cwd_git_repo
from backend.apps.agents.manager.prompt.tool_catalog import (
FULL_TOOLS,
_get_all_known_tool_names,
_get_denied_tool_names,
_is_fully_denied,
get_all_known_tool_names,
get_denied_tool_names,
is_fully_denied,
)
from backend.apps.agents.core.aux_llm import safe_resp_text, clean_short_label
from backend.apps.agents.manager.session.history_compaction import (
_build_history_prefix,
_get_branch_messages,
_truncate_large_tool_result,
build_history_prefix,
get_branch_messages,
truncate_large_tool_result,
)
from backend.apps.agents.manager.prompt.prompt_context import (
_build_browser_context,
_build_selected_app_context,
_build_connected_tools_context,
_build_mcp_registry_summary,
_compose_system_prompt,
_resolve_attached_skills,
_resolve_forced_tools,
_resolve_mode,
build_browser_context,
build_selected_app_context,
build_connected_tools_context,
build_mcp_registry_summary,
compose_system_prompt,
resolve_attached_skills,
resolve_forced_tools,
resolve_mode,
)
from backend.apps.agents.manager.prompt.attachments import (
_build_dir_tree,
_build_prompt_content,
_resolve_attachments,
_resolve_context_paths,
build_dir_tree,
build_prompt_content,
resolve_attachments,
resolve_context_paths,
)
logger = logging.getLogger(__name__)
@@ -119,7 +119,7 @@ def get_all_tool_names() -> list[str]:
if t.mcp_config
and t.enabled
and t.auth_status in ("configured", "connected")
and not _is_fully_denied(t)
and not is_fully_denied(t)
]
return builtin_tools + mcp_names
@@ -130,7 +130,7 @@ class AgentManager:
self.tasks: dict[str, asyncio.Task] = {}
def _resolve_mode(self, mode_id: str) -> tuple[list[str], str | None, str | None]:
return _resolve_mode(mode_id, get_all_tool_names)
return resolve_mode(mode_id, get_all_tool_names)
async def _build_mcp_servers(
self,
@@ -178,7 +178,7 @@ class AgentManager:
logger.info(f"[MCP-DEBUG] GATED {server_name}: not in session.active_mcps, model must call MCPActivate first")
continue
if _is_fully_denied(tool):
if is_fully_denied(tool):
logger.info(f"[MCP-DEBUG] SKIPPED {tool.name}: fully denied")
continue
@@ -206,19 +206,19 @@ class AgentManager:
return mcp_servers
def _build_connected_tools_context(self, allowed_tools: list[str]) -> str | None:
return _build_connected_tools_context(allowed_tools, get_all_tool_names)
return build_connected_tools_context(allowed_tools, get_all_tool_names)
def _build_browser_context(self, dashboard_id: str | None, selected_browser_ids: list[str] | None = None) -> str | None:
return _build_browser_context(dashboard_id, selected_browser_ids)
return build_browser_context(dashboard_id, selected_browser_ids)
def _build_selected_app_context(self, selected_app_output_ids: list[str] | None) -> str | None:
return _build_selected_app_context(selected_app_output_ids)
return build_selected_app_context(selected_app_output_ids)
def _build_mcp_registry_summary(self, allowed_tools: list[str], active_mcps: list[str]) -> str | None:
return _build_mcp_registry_summary(allowed_tools, active_mcps, get_all_tool_names)
return build_mcp_registry_summary(allowed_tools, active_mcps, get_all_tool_names)
def _compose_system_prompt(self, default_prompt: str | None, mode_prompt: str | None, session_prompt: str | None, connected_tools_ctx: str | None = None, browser_ctx: str | None = None, mcp_registry_ctx: str | None = None) -> str | None:
return _compose_system_prompt(default_prompt, mode_prompt, session_prompt, connected_tools_ctx, browser_ctx, mcp_registry_ctx)
return compose_system_prompt(default_prompt, mode_prompt, session_prompt, connected_tools_ctx, browser_ctx, mcp_registry_ctx)
async def launch_agent(self, config: AgentConfig) -> AgentSession:
session_id = uuid4().hex
@@ -291,9 +291,9 @@ class AgentManager:
effective_cwd = os.path.join(_home, ".openswarm", "workspaces", session_id)
os.makedirs(effective_cwd, exist_ok=True)
_ensure_cwd_git_repo(effective_cwd, _home)
ensure_cwd_git_repo(effective_cwd, _home)
repo_url, branch_name = _detect_git_identity(effective_cwd)
repo_url, branch_name = detect_git_identity(effective_cwd)
session = AgentSession(
id=session_id,
@@ -322,13 +322,13 @@ class AgentManager:
return session
def _build_dir_tree(self, root: str, max_depth: int = 4, prefix: str = "") -> list[str]:
return _build_dir_tree(root, max_depth, prefix)
return build_dir_tree(root, max_depth, prefix)
def _resolve_forced_tools(self, forced_tools: list[str] | None) -> str:
return _resolve_forced_tools(forced_tools)
return resolve_forced_tools(forced_tools)
def _resolve_attached_skills(self, attached_skills: list | None) -> str:
return _resolve_attached_skills(attached_skills)
return resolve_attached_skills(attached_skills)
# ------------------------------------------------------------------
# Compaction & token guard (Phase 2)
@@ -351,12 +351,12 @@ class AgentManager:
sets compacted_through_msg_id and emits a context_status event.
Never modifies session.messages, originals stay around for the
UI drawer; only the history *sent to the SDK* is trimmed (handled
in _build_history_prefix lookups).
in build_history_prefix lookups).
"""
ctx_used = session.tokens.get("input", 0) / max(1, session.context_window)
if not force and ctx_used < session.compact_threshold_pct:
return False
msgs = _get_branch_messages(session)
msgs = get_branch_messages(session)
if len(msgs) < 4:
return False
# Summarize everything up to (but not including) the last 6
@@ -373,13 +373,13 @@ class AgentManager:
return True
def _build_prompt_content(self, prompt: str, images: list | None = None, context_paths: list | None = None, forced_tools: list[str] | None = None, attached_skills: list | None = None, api_type: str = "anthropic", model: str = ""):
return _build_prompt_content(prompt, images, context_paths, forced_tools, attached_skills, api_type, model)
return build_prompt_content(prompt, images, context_paths, forced_tools, attached_skills, api_type, model)
def _resolve_attachments(self, context_paths: list | None, api_type: str, model: str) -> tuple[str, list[dict], list[str]]:
return _resolve_attachments(context_paths, api_type, model)
return resolve_attachments(context_paths, api_type, model)
def _resolve_context_paths(self, context_paths: list | None) -> str:
return _resolve_context_paths(context_paths)
return resolve_context_paths(context_paths)
async def _run_agent_loop(self, session_id: str, prompt: str, images: list | None = None, context_paths: list | None = None, forced_tools: list[str] | None = None, attached_skills: list | None = None, fork_session: bool = False, selected_browser_ids: list[str] | None = None, selected_app_output_ids: list[str] | None = None):
"""Run the Claude Agent SDK query loop for a session."""
@@ -387,8 +387,8 @@ class AgentManager:
if not session:
return
from backend.apps.agents.providers.registry import get_api_type as _get_api_type
_api = _get_api_type(session.model)
from backend.apps.agents.providers.registry import get_api_type as idk_get_api_type
_api = idk_get_api_type(session.model)
prompt_content = self._build_prompt_content(
prompt, images, context_paths, forced_tools, attached_skills,
api_type=_api, model=session.model,
@@ -997,7 +997,7 @@ class AgentManager:
# at *write* time (before the next turn ships history to the
# SDK) so the bloat never re-enters context.
try:
truncated_content, blob_path = _truncate_large_tool_result(
truncated_content, blob_path = truncate_large_tool_result(
result_msg.content, session.id, result_msg.id
)
if blob_path:
@@ -1248,8 +1248,8 @@ class AgentManager:
# settings is NOT enough; it must be a *-api route model. Everyone
# else registers openswarm-web and cascades through /api/web/search.
from backend.apps.agents.tools.web import anthropic_web_search_is_reliable
from backend.apps.agents.providers.registry import _find_builtin_model as _fbm_web
_web_model_entry = _fbm_web(session.model)
from backend.apps.agents.providers.registry import find_builtin_model as fbm_web
_web_model_entry = fbm_web(session.model)
_uses_direct_anthropic_api = (
_web_model_entry is not None
and _web_model_entry.get("route") == "api"
@@ -1344,8 +1344,8 @@ class AgentManager:
None,
)
if tool_def:
denied = _get_denied_tool_names(tool_def)
known = _get_all_known_tool_names(tool_def)
denied = get_denied_tool_names(tool_def)
known = get_all_known_tool_names(tool_def)
for tn in known - denied:
policy = tool_def.tool_permissions.get(tn, "ask")
if policy == "always_allow":
@@ -1465,12 +1465,12 @@ class AgentManager:
}
# cc/cx/gc/ag/gemini/openrouter prefixes force 9Router; route="api"
# bypasses to the provider's host directly; otherwise Pro proxy or key.
from backend.apps.nine_router import is_running as _9r_running
from backend.apps.agents.providers.registry import _NINEROUTER_MODEL_PREFIXES
resolved_is_9router = isinstance(resolved_model, str) and resolved_model.startswith(_NINEROUTER_MODEL_PREFIXES)
from backend.apps.nine_router.process import is_running
from backend.apps.agents.providers.registry import NINEROUTER_MODEL_PREFIXES
resolved_is_9router = isinstance(resolved_model, str) and resolved_model.startswith(NINEROUTER_MODEL_PREFIXES)
from backend.apps.agents.providers.registry import _find_builtin_model
_model_entry = _find_builtin_model(session.model)
from backend.apps.agents.providers.registry import find_builtin_model
_model_entry = find_builtin_model(session.model)
_is_pinned_api_route = (
_model_entry is not None
and _model_entry.get("route") == "api"
@@ -1513,18 +1513,18 @@ class AgentManager:
# User-configured OpenAI-compatible endpoint (Ollama Cloud,
# Together, local Ollama, etc.). Routes through 9Router's
# openai-compatible provider node we synced from settings.
from backend.apps.nine_router import ensure_running as _9r_ensure_c
if not _9r_running():
from backend.apps.nine_router.process import ensure_running
if not is_running():
logger.info(f"[MCP-DEBUG] custom provider selected but 9Router not running; waiting for startup")
await _9r_ensure_c()
if not _9r_running():
await ensure_running()
if not is_running():
raise ValueError(
"9Router could not start. Custom OpenAI-compatible "
"providers need 9Router to translate the Anthropic "
"protocol, install Node.js and restart the app."
)
from backend.apps.agents.providers.registry import _find_custom_provider_for_value
cp = _find_custom_provider_for_value(global_settings, session.model)
from backend.apps.agents.providers.registry import find_custom_provider_for_value
cp = find_custom_provider_for_value(global_settings, session.model)
env = {
"ANTHROPIC_API_KEY": "9router",
"ANTHROPIC_BASE_URL": "http://localhost:20128",
@@ -1538,8 +1538,8 @@ class AgentManager:
# CLI can issue requests. Servers that DO check auth always
# have a real key configured.
env["OPENAI_API_KEY"] = (cp.api_key or "").strip() or "no-auth-required"
from backend.apps.nine_router import normalize_openai_compat_base_url as _norm_cp_url
env["OPENAI_BASE_URL"] = _norm_cp_url(cp.base_url or "")
from backend.apps.nine_router.sync_custom import normalize_openai_compat_base_url
env["OPENAI_BASE_URL"] = normalize_openai_compat_base_url(cp.base_url or "")
# Pin subagent ids, without these, CLI's default Haiku 4.5
# gets sent to the custom provider and 404s.
if global_settings.anthropic_api_key:
@@ -1575,11 +1575,11 @@ class AgentManager:
# CLI's WebSearch delegation needs an Anthropic-shaped lane;
# if the user has no Anthropic key/sub/Pro, fall back to OR's
# resold Claude so subagents stay on the same OR billing.
if not _9r_running():
from backend.apps.nine_router import ensure_running as _9r_ensure
if not is_running():
from backend.apps.nine_router.process import ensure_running
logger.info(f"[MCP-DEBUG] OpenRouter selected but 9Router not running; waiting for startup")
await _9r_ensure()
if not _9r_running():
await ensure_running()
if not is_running():
raise ValueError(
"9Router could not start. OpenRouter routing requires "
"Node.js, install it and restart the app, or pick a "
@@ -1619,7 +1619,7 @@ class AgentManager:
elif api_type == "anthropic" and not resolved_is_9router and global_settings.anthropic_api_key:
options_kwargs["env"] = {"ANTHROPIC_API_KEY": global_settings.anthropic_api_key}
logger.info("[MCP-DEBUG] Using direct Anthropic API key")
elif _9r_running():
elif is_running():
# Gemini-bound ids go through the local proxy for schema scrubbing;
# everything else hits 9Router directly.
_is_gemini_bound = (
@@ -1663,10 +1663,10 @@ class AgentManager:
logger.info(f"[MCP-DEBUG] Using 9Router (api_type={api_type})")
else:
if api_type != "anthropic":
from backend.apps.nine_router import ensure_running as _9r_ensure
from backend.apps.nine_router.process import ensure_running
logger.info(f"[MCP-DEBUG] 9Router not running for non-Anthropic model {session.model}; waiting for startup")
await _9r_ensure()
if _9r_running():
await ensure_running()
if is_running():
options_kwargs["env"] = {
"ANTHROPIC_API_KEY": "9router",
"ANTHROPIC_BASE_URL": "http://localhost:20128",
@@ -1726,7 +1726,7 @@ class AgentManager:
# the git-init block in launch_agent, leaving them
# without a valid HEAD. Ensure it here so subagent
# worktree-add always works.
_ensure_cwd_git_repo(session.cwd)
ensure_cwd_git_repo(session.cwd)
options_kwargs["cwd"] = session.cwd
try:
@@ -1764,7 +1764,7 @@ class AgentManager:
# Fresh-restart path: some session changes must not reuse the
# CLI's resume transcript. MCPActivate needs a new transport so
# tool schemas are reread; branch edits/switches need the model
# to see only _get_branch_messages(session), not facts from the
# to see only get_branch_messages(session), not facts from the
# old branch's SDK transcript. Soft restart: drop resume +
# sdk_session_id, replay local history via the prompt, let the
# SDK build a clean session from the current app state.
@@ -1785,8 +1785,8 @@ class AgentManager:
if session.needs_fork:
session.needs_fork = False
elif len(session.messages) > 1:
history = _build_history_prefix(
_get_branch_messages(session),
history = build_history_prefix(
get_branch_messages(session),
cutoff_msg_id=session.compacted_through_msg_id,
)
if history:
@@ -1990,11 +1990,11 @@ class AgentManager:
# answer text (e.g. 13).
if not _turn_thinking_text_parts or force_provider_unavailable:
try:
from backend.apps.nine_router import (
from backend.apps.nine_router.process import (
get_latest_reasoning_tokens,
is_running as _9r_running,
is_running
)
if _9r_running():
if is_running():
rt = await get_latest_reasoning_tokens(model_hint=session.model)
if rt and rt > 0:
upstream_reasoning_tokens = rt
@@ -2039,11 +2039,11 @@ class AgentManager:
turn_tokens = max(_turn_output_tokens, heuristic_tokens)
else:
try:
from backend.apps.nine_router import (
from backend.apps.nine_router.process import (
get_latest_reasoning_tokens,
is_running as _9r_running,
is_running,
)
if _9r_running():
if is_running():
rt = await get_latest_reasoning_tokens(model_hint=session.model)
if rt and rt > 0:
turn_tokens = rt
@@ -2704,7 +2704,7 @@ class AgentManager:
cost = 0.0
elif isinstance(resolved_model, str) and resolved_model.startswith("openrouter/"):
# SDK assumes Anthropic rates → 50-100× off for OR.
from backend.apps.agents.providers.registry import get_openrouter_pricing
from backend.apps.agents.providers.openrouter import get_openrouter_pricing
pricing = get_openrouter_pricing(resolved_model)
if pricing:
in_rate, out_rate = pricing
@@ -2725,7 +2725,7 @@ class AgentManager:
# ($30 instead of $0.04 per Mehmet-style
# 4-PDF turn). Use the published per-model
# rates instead.
from backend.apps.agents.providers.registry import get_direct_pricing
from backend.apps.agents.providers.openrouter import get_direct_pricing
pricing = get_direct_pricing(resolved_model) or get_direct_pricing(session.model)
if pricing:
in_rate, out_rate = pricing
@@ -3112,7 +3112,7 @@ class AgentManager:
"session": session.model_dump(mode="json"),
})
try:
_save_session(session_id, session.model_dump(mode="json"))
save_session(session_id, session.model_dump(mode="json"))
except Exception as e:
logger.warning(f"Failed to snapshot session {session_id}: {e}")
@@ -3270,7 +3270,7 @@ class AgentManager:
"""Send a follow-up message to an existing session."""
session = self.sessions.get(session_id)
if not session:
data = _load_session_data(session_id)
data = load_session_data(session_id)
if data:
session = AgentSession(**data)
_apply_context_window(session)
@@ -3293,8 +3293,8 @@ class AgentManager:
# responses with placeholder text). Forking starts a new CLI
# session so history is re-sent fresh in whichever format the
# new provider expects.
from backend.apps.agents.providers.registry import get_api_type as _get_api_type_for_model
if _get_api_type_for_model(session.model) != _get_api_type_for_model(model):
from backend.apps.agents.providers.registry import get_api_type as get_api_type_for_model
if get_api_type_for_model(session.model) != get_api_type_for_model(model):
session.needs_fork = True
logger.info(f"[MCP-DEBUG] Forking session: api_type changed {session.model}→{model}")
@@ -3523,7 +3523,7 @@ class AgentManager:
"session": session.model_dump(mode="json"),
})
try:
_save_session(session_id, session.model_dump(mode="json"))
save_session(session_id, session.model_dump(mode="json"))
except Exception as e:
logger.warning(f"Failed to snapshot session {session_id}: {e}")
@@ -3822,8 +3822,8 @@ class AgentManager:
return
try:
from backend.apps.agents.providers.registry import _find_builtin_model
entry = _find_builtin_model(session.model)
from backend.apps.agents.providers.registry import find_builtin_model
entry = find_builtin_model(session.model)
if not entry or entry.get("api") != "anthropic":
return # other providers handle caching automatically
@@ -3962,7 +3962,7 @@ class AgentManager:
return build_search_text(session, max_len)
def _sync_session_close(self, session: AgentSession, close_reason: str = "user"):
_sync_session_close(session, close_reason)
sync_session_close(session, close_reason)
async def close_session(self, session_id: str) -> None:
"""Close a session: pause the agent if running, persist to JSON file,
@@ -4002,7 +4002,7 @@ class AgentManager:
doc_data = session.model_dump(mode="json")
doc_data["search_text"] = self._build_search_text(session)
_save_session(session_id, doc_data)
save_session(session_id, doc_data)
await ws_manager.send_to_session(session_id, "agent:closed", {
"session_id": session_id,
@@ -4041,7 +4041,7 @@ class AgentManager:
self.sessions.pop(session_id, None)
self.tasks.pop(session_id, None)
_delete_session_file(session_id)
delete_session_file(session_id)
logger.info(f"Session {session_id} permanently deleted")
async def resume_session(self, session_id: str) -> AgentSession:
@@ -4049,7 +4049,7 @@ class AgentManager:
if session_id in self.sessions:
return self.sessions[session_id]
data = _load_session_data(session_id)
data = load_session_data(session_id)
if data is None:
raise ValueError(f"Session {session_id} not found in history")
@@ -4064,7 +4064,7 @@ class AgentManager:
# chat permanently removed it from history on the next restart.
# The disk copy stays as the durable record; subsequent turn
# completions and close_session calls overwrite it via
# _save_session, so memory and disk stay in sync.
# save_session, so memory and disk stay in sync.
await ws_manager.send_to_session(session_id, "agent:status", {
"session_id": session_id,
@@ -4083,7 +4083,7 @@ class AgentManager:
dashboard_id: str | None = None,
) -> dict:
"""Return paginated, optionally filtered summaries of closed sessions."""
all_data = _load_all_session_data()
all_data = load_all_session_data()
all_data.sort(key=lambda pair: pair[1].get("closed_at") or "", reverse=True)
q_lower = q.strip().lower()
@@ -4118,7 +4118,7 @@ class AgentManager:
async def reconcile_on_startup(self) -> None:
"""Mark any stale running sessions as stopped."""
for sid, data in _load_all_session_data():
for sid, data in load_all_session_data():
dirty = False
if data.get("status") in ("running", "waiting_approval"):
data["status"] = "stopped"
@@ -4130,7 +4130,7 @@ class AgentManager:
data["mode"] = "ask"
dirty = True
if dirty:
_save_session(sid, data)
save_session(sid, data)
async def persist_all_sessions(self) -> None:
"""Flush every in-memory session to JSON files (for graceful shutdown)."""
@@ -4147,7 +4147,7 @@ class AgentManager:
self._sync_session_close(session, close_reason="shutdown")
doc_data = session.model_dump(mode="json")
doc_data["search_text"] = self._build_search_text(session)
_save_session(session_id, doc_data)
save_session(session_id, doc_data)
logger.info(f"Persisted session {session_id} on shutdown")
self.sessions.clear()
self.tasks.clear()
@@ -4159,7 +4159,7 @@ class AgentManager:
shutdown). Sessions with closed_at were explicitly closed by the user
and stay on disk so the history endpoint can still serve them.
"""
for sid, data in _load_all_session_data():
for sid, data in load_all_session_data():
try:
session = AgentSession(**data)
except Exception as e:
@@ -4172,14 +4172,14 @@ class AgentManager:
session.pending_approvals = []
_apply_context_window(session)
self.sessions[session.id] = session
_delete_session_file(sid)
delete_session_file(sid)
logger.info(f"Restored session {session.id}")
async def duplicate_session(self, session_id: str, dashboard_id: str | None = None, up_to_message_id: str | None = None) -> AgentSession:
"""Create an independent copy of a session with the same chat history."""
source = self.sessions.get(session_id)
if not source:
data = _load_session_data(session_id)
data = load_session_data(session_id)
if data is None:
raise ValueError(f"Session {session_id} not found")
source = AgentSession(**data)
@@ -4262,7 +4262,7 @@ class AgentManager:
"""Fork an existing session and send it a new message, returning the result."""
source = self.sessions.get(source_session_id)
if not source:
data = _load_session_data(source_session_id)
data = load_session_data(source_session_id)
if data is None:
raise ValueError(f"Session {source_session_id} not found")
source = AgentSession(**data)
@@ -4387,7 +4387,7 @@ class AgentManager:
results.append(s.model_dump(mode="json"))
seen.add(s.id)
for sid, data in _load_all_session_data():
for sid, data in load_all_session_data():
if sid in seen:
continue
if data.get("mode") == "browser-agent" and data.get("parent_session_id") == parent_session_id:
backend/apps/agents/agents.py
+18 -15
View File
@@ -337,7 +337,8 @@ async def clear_session(session_id: str):
@agents.router.get("/subscriptions/status")
async def subscriptions_status():
"""Check if 9Router is running and list connected providers."""
from backend.apps.nine_router import is_running, get_providers, get_models
from backend.apps.nine_router.process import is_running, get_providers
from backend.apps.nine_router.oauth import get_models
if not is_running():
return {"running": False, "providers": [], "models": []}
connections = await get_providers()
@@ -349,7 +350,8 @@ async def subscriptions_status():
@agents.router.post("/subscriptions/connect")
async def subscriptions_connect(body: dict):
"""Start OAuth flow for a subscription provider."""
from backend.apps.nine_router import is_running, ensure_running, start_oauth
from backend.apps.nine_router.process import is_running, ensure_running
from backend.apps.nine_router.oauth import start_oauth
provider = body.get("provider", "")
if not provider:
raise HTTPException(status_code=400, detail="provider required")
@@ -386,7 +388,7 @@ async def subscriptions_connect(body: dict):
@agents.router.post("/subscriptions/poll")
async def subscriptions_poll(body: dict):
"""Poll for OAuth completion."""
from backend.apps.nine_router import poll_oauth
from backend.apps.nine_router.oauth import poll_oauth
provider = body.get("provider", "")
device_code = body.get("device_code", "")
if not provider or not device_code:
@@ -410,7 +412,7 @@ async def subscriptions_poll(body: dict):
@agents.router.post("/subscriptions/exchange")
async def subscriptions_exchange(body: dict):
"""Exchange OAuth code for tokens via 9Router."""
from backend.apps.nine_router import exchange_oauth
from backend.apps.nine_router.oauth import exchange_oauth
provider = body.get("provider", "")
code = body.get("code", "")
redirect_uri = body.get("redirect_uri", "")
@@ -434,7 +436,8 @@ async def subscriptions_exchange(body: dict):
@agents.router.get("/subscriptions/models")
async def subscriptions_models():
"""List all models available through connected subscriptions."""
from backend.apps.nine_router import is_running, get_models
from backend.apps.nine_router.process import is_running
from backend.apps.nine_router.oauth import get_models
if not is_running():
return {"models": []}
models = await get_models()
@@ -456,7 +459,7 @@ async def probe_model(body: dict):
_NINEROUTER_MODEL_PREFIXES,
)
from backend.apps.settings.settings import load_settings
from backend.apps.nine_router import is_running as _9r_running
from backend.apps.nine_router.process import is_running
settings = load_settings()
api_type = get_api_type(short_name)
resolved = resolve_model_id_for_sdk(short_name, settings)
@@ -474,7 +477,7 @@ async def probe_model(body: dict):
)
if resolved_is_9router:
if not _9r_running():
if not is_running():
return {"ok": True, "skipped": True}
client = anthropic.AsyncAnthropic(api_key="9router", base_url="http://localhost:20128")
elif route == "api" and api_type == "anthropic" and getattr(settings, "anthropic_api_key", None):
@@ -488,7 +491,7 @@ async def probe_model(body: dict):
elif api_type == "anthropic" and getattr(settings, "anthropic_api_key", None):
client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
else:
if not _9r_running():
if not is_running():
return {"ok": True, "skipped": True}
client = anthropic.AsyncAnthropic(api_key="9router", base_url="http://localhost:20128")
@@ -521,16 +524,16 @@ async def probe_model(body: dict):
async def list_models():
"""Picker model list, grouped by provider, intersected with available creds."""
from backend.apps.agents.providers.registry import BUILTIN_MODELS
from backend.apps.nine_router import is_running as _9r_running, get_providers as _9r_providers
from backend.apps.nine_router.process import is_running, get_providers
from backend.apps.settings.settings import load_settings
settings = load_settings()
nine_router_up = _9r_running()
nine_router_up = is_running()
connected: set[str] = set()
if nine_router_up:
try:
conns = await _9r_providers()
conns = await get_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 = {
@@ -697,7 +700,7 @@ async def list_models():
# Fetch OpenRouter catalog directly (independent of 9Router) so picker fills the moment a key lands.
if has_openrouter_key:
try:
from backend.apps.agents.providers.registry import fetch_openrouter_models
from backend.apps.agents.providers.openrouter import fetch_openrouter_models
or_models = await fetch_openrouter_models(settings.openrouter_api_key)
except Exception as e:
logger.debug(f"OpenRouter catalog fetch failed: {e}")
@@ -743,14 +746,14 @@ async def list_models():
result[f"OpenRouter · {pretty}"] = entries
# Custom OpenAI-compatible providers (Ollama Cloud, Together, etc); addressed via custom/<slug>/<model_id>.
from backend.apps.agents.providers.registry import _custom_provider_slug_for_lookup
from backend.apps.agents.providers.registry import custom_provider_slug_for_lookup
for cp in (getattr(settings, "custom_providers", None) or []):
cp_name = (getattr(cp, "name", "") or "").strip()
cp_base_url = (getattr(cp, "base_url", "") or "").strip()
cp_models = getattr(cp, "models", None) or []
if not cp_name or not cp_base_url or not cp_models:
continue
slug = _custom_provider_slug_for_lookup(cp_name)
slug = custom_provider_slug_for_lookup(cp_name)
entries: list[dict] = []
for m in cp_models:
bare = (m.get("value") or m.get("id") or "").strip()
@@ -786,7 +789,7 @@ _PROVIDER_CASCADE_REMOVES: dict[str, list[str]] = {
async def _delete_provider_connections(providers: list[str]) -> int:
"""Delete 9Router connections in `providers`; returns count removed, silent on 9Router unreachable."""
import httpx
from backend.apps.nine_router import NINE_ROUTER_API, get_providers
from backend.apps.nine_router.process import NINE_ROUTER_API, get_providers
try:
connections = await get_providers()
except Exception:
backend/apps/agents/core/mcp_preflight.py
+17 -17
View File
@@ -17,9 +17,9 @@ logger = logging.getLogger(__name__)
# Curated shortlist; `id` MUST match ToolDefinition.name exactly or the enabled/dismissed filter no-ops and the modal renders nothing.
CuratedEntry = dict[str, Any]
P_CuratedEntry = dict[str, Any]
CURATED_SHORTLIST: list[CuratedEntry] = [
P_CURATED_SHORTLIST: list[P_CuratedEntry] = [
{
"id": "Google Workspace",
"title": "Google Workspace",
@@ -69,18 +69,18 @@ CURATED_SHORTLIST: list[CuratedEntry] = [
# Short-circuit for obviously-local prompts where no MCP helps. Saves ~200ms + ~$0.0001 per launch.
_PATH_LIKE = re.compile(r"^[./~]|/[\w\-]+/|\.[a-zA-Z]{1,5}\b")
_SHELL_PREFIX = re.compile(r"^\s*[\$!/]")
P_PATH_LIKE = re.compile(r"^[./~]|/[\w\-]+/|\.[a-zA-Z]{1,5}\b")
P_SHELL_PREFIX = re.compile(r"^\s*[\$!/]")
def _is_obviously_local(prompt: str) -> bool:
def p_is_obviously_local(prompt: str) -> bool:
"""True for prompts that obviously can't benefit from MCP (very short, shell-ish, single path)."""
s = prompt.strip()
if len(s) < 8:
return True
if _SHELL_PREFIX.match(s):
if P_SHELL_PREFIX.match(s):
return True
if " " not in s and _PATH_LIKE.search(s):
if " " not in s and P_PATH_LIKE.search(s):
return True
return False
@@ -92,21 +92,21 @@ async def run_preflight(prompt: str, timeout_s: float = 2.0) -> dict:
if not prompt or not prompt.strip():
return default
if _is_obviously_local(prompt):
if p_is_obviously_local(prompt):
return default
try:
settings = load_settings()
available = _build_available_shortlist(settings)
available = p_build_available_shortlist(settings)
result = await asyncio.wait_for(
_call_classifier(settings, prompt, available),
p_call_classifier(settings, prompt, available),
timeout=timeout_s,
)
# Re-validate ids against the curated shortlist so hallucinations can't reach the frontend.
valid_ids = {e["id"] for e in CURATED_SHORTLIST}
valid_ids = {e["id"] for e in P_CURATED_SHORTLIST}
result["suggestions"] = [
_decorate(s, available) for s in result.get("suggestions", [])
p_decorate(s, available) for s in result.get("suggestions", [])
if isinstance(s, dict) and s.get("id") in valid_ids
]
result["suggestions"] = [s for s in result["suggestions"] if s is not None]
@@ -123,7 +123,7 @@ async def run_preflight(prompt: str, timeout_s: float = 2.0) -> dict:
return default
def _build_available_shortlist(settings) -> list[CuratedEntry]:
def p_build_available_shortlist(settings) -> list[P_CuratedEntry]:
"""Curated entries that are NOT currently enabled and NOT dismissed."""
try:
enabled_names = {t.name for t in load_all_tools() if getattr(t, "enabled", False)}
@@ -133,12 +133,12 @@ def _build_available_shortlist(settings) -> list[CuratedEntry]:
dismissed = set((getattr(settings, "dismissed_mcp_suggestions", {}) or {}).keys())
return [
entry for entry in CURATED_SHORTLIST
entry for entry in P_CURATED_SHORTLIST
if entry["id"] not in enabled_names and entry["id"] not in dismissed
]
def _decorate(llm_suggestion: dict, available: list[CuratedEntry]) -> dict | None:
def p_decorate(llm_suggestion: dict, available: list[P_CuratedEntry]) -> dict | None:
"""Expand an LLM-returned {id, reason} into the full frontend shape."""
entry = next((e for e in available if e["id"] == llm_suggestion["id"]), None)
if entry is None:
@@ -151,9 +151,9 @@ def _decorate(llm_suggestion: dict, available: list[CuratedEntry]) -> dict | Non
}
async def _call_classifier(settings, prompt: str, available: list[CuratedEntry]) -> dict:
async def p_call_classifier(settings, prompt: str, available: list[P_CuratedEntry]) -> dict:
"""One aux-model call, returns validated JSON {is_vague, suggestions}."""
aux_model, _base = await resolve_aux_model(settings, preferred_tier="haiku")
aux_model, _ = await resolve_aux_model(settings, preferred_tier="haiku")
client = get_anthropic_client_for_model(settings, aux_model)
catalog_lines = "\n".join(
backend/apps/agents/core/openai_passthrough.py
-121
View File
@@ -1,121 +0,0 @@
"""Tiny OpenAI passthrough renaming max_tokens to max_completion_tokens for GPT-5; 9Router 0.3.60 is pinned and doesn't know the change."""
import json
import logging
from contextlib import asynccontextmanager
import httpx
from fastapi import Request
from fastapi.responses import JSONResponse, StreamingResponse
from backend.config.Apps import SubApp
logger = logging.getLogger(__name__)
@asynccontextmanager
async def openai_passthrough_lifespan():
yield
openai_passthrough = SubApp("openai-passthrough", openai_passthrough_lifespan)
# Mirrors anthropic_proxy.py's GPT-5 matcher; duplicated to avoid the cross-module dep.
_GPT5_PREFIXES = ("gpt-5",)
_OPENAI_UPSTREAM = "https://api.openai.com/v1"
_HOP_HEADERS = {
"host", "content-length", "connection", "keep-alive",
"proxy-authenticate", "proxy-authorization", "te", "trailers",
"transfer-encoding", "upgrade",
}
def _is_gpt5(model: str) -> bool:
m = (model or "").strip().lower()
if not m:
return False
for prefix in ("openai/", "cx/", "openrouter/", "or:openai/", "cp/", "cp-"):
if m.startswith(prefix):
m = m[len(prefix):]
break
return any(m.startswith(p) for p in _GPT5_PREFIXES)
def _scrub_max_tokens(body: bytes) -> bytes:
"""Rename max_tokens to max_completion_tokens for GPT-5; bytes in/out, never raises."""
if not body:
return body
try:
parsed = json.loads(body)
except Exception:
return body
if not isinstance(parsed, dict):
return body
model = str(parsed.get("model") or "")
if not _is_gpt5(model):
return body
if "max_tokens" in parsed and "max_completion_tokens" not in parsed:
parsed["max_completion_tokens"] = parsed.pop("max_tokens")
return json.dumps(parsed).encode("utf-8")
if "max_tokens" in parsed and "max_completion_tokens" in parsed:
parsed.pop("max_tokens", None)
return json.dumps(parsed).encode("utf-8")
return body
@openai_passthrough.router.api_route(
"/v1/{rest:path}",
methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"],
)
async def passthrough(rest: str, request: Request):
body = await request.body()
body = _scrub_max_tokens(body)
forward_headers: dict[str, str] = {}
for k, v in request.headers.items():
if k.lower() in _HOP_HEADERS:
continue
forward_headers[k] = v
upstream_url = f"{_OPENAI_UPSTREAM}/{rest}"
if request.url.query:
upstream_url = f"{upstream_url}?{request.url.query}"
# Stream upstream body back; httpx handles SSE without buffering the full response.
client = httpx.AsyncClient(timeout=httpx.Timeout(connect=10.0, read=300.0, write=60.0, pool=30.0))
try:
upstream_req = client.build_request(
request.method,
upstream_url,
headers=forward_headers,
content=body,
)
upstream_resp = await client.send(upstream_req, stream=True)
except httpx.HTTPError as e:
await client.aclose()
logger.warning("openai-passthrough upstream error: %s", e)
return JSONResponse(
{"error": {"message": str(e), "type": "upstream_error"}},
status_code=502,
)
response_headers: dict[str, str] = {}
for k, v in upstream_resp.headers.items():
if k.lower() in _HOP_HEADERS:
continue
response_headers[k] = v
async def streamer():
try:
async for chunk in upstream_resp.aiter_raw():
yield chunk
finally:
await upstream_resp.aclose()
await client.aclose()
return StreamingResponse(
streamer(),
status_code=upstream_resp.status_code,
headers=response_headers,
)
backend/apps/agents/core/seq_log.py
+26 -26
View File
@@ -13,12 +13,12 @@ from typing import AsyncIterator, Optional
logger = logging.getLogger(__name__)
# 500 events covers a 30s drop even at ~20Hz thinking deltas (~50KB/session).
BUFFER_LIMIT = 500
P_BUFFER_LIMIT = 500
TERMINAL_STATUSES = {"completed", "stopped", "error"}
P_TERMINAL_STATUSES = {"completed", "stopped", "error"}
class _SessionSeqLog:
class P_SessionSeqLog:
"""Per-session lock + monotonic seq + recent-event ring buffer."""
__slots__ = ("lock", "seq", "buffer")
@@ -27,43 +27,43 @@ class _SessionSeqLog:
self.lock: asyncio.Lock = asyncio.Lock()
self.seq: int = 0
# (seq, json_payload_str): pre-serialized so replays don't redo json.dumps per reconnect.
self.buffer: deque[tuple[int, str]] = deque(maxlen=BUFFER_LIMIT)
self.buffer: deque[tuple[int, str]] = deque(maxlen=P_BUFFER_LIMIT)
class SeqLogStore:
"""Process-wide store. Per-session locks live inside `_SessionSeqLog`."""
def __init__(self, persist_dir: Optional[str] = None) -> None:
self._per_session: dict[str, _SessionSeqLog] = {}
self.p_per_session: dict[str, P_SessionSeqLog] = {}
# Coarse lock guards only the setdefault path; never crosses an await.
self._dict_lock = asyncio.Lock()
self._persist_dir = persist_dir
self.p_dict_lock = asyncio.Lock()
self.p_persist_dir = persist_dir
if persist_dir:
try:
os.makedirs(persist_dir, exist_ok=True)
except Exception:
logger.warning("seq_log: failed to create persist dir %s", persist_dir)
async def _get_or_create(self, session_id: str) -> _SessionSeqLog:
log = self._per_session.get(session_id)
async def p_get_or_create(self, session_id: str) -> P_SessionSeqLog:
log = self.p_per_session.get(session_id)
if log is not None:
return log
async with self._dict_lock:
async with self.p_dict_lock:
log = self._per_session.get(session_id)
if log is None:
log = _SessionSeqLog()
self._per_session[session_id] = log
log = P_SessionSeqLog()
self.p_per_session[session_id] = log
return log
def _peek(self, session_id: str) -> Optional[_SessionSeqLog]:
return self._per_session.get(session_id)
def p_peek(self, session_id: str) -> Optional[P_SessionSeqLog]:
return self.p_per_session.get(session_id)
@asynccontextmanager
async def stamp(
self, session_id: str, event: str, data: dict
) -> AsyncIterator[tuple[int, str]]:
"""Atomically assign seq, buffer, and yield (seq, payload); caller's send must happen inside the with-block."""
log = await self._get_or_create(session_id)
log = await self.p_get_or_create(session_id)
async with log.lock:
log.seq += 1
seq = log.seq
@@ -81,7 +81,7 @@ class SeqLogStore:
self, session_id: str, last_seq: int
) -> tuple[Optional[int], Optional[int], list[str]]:
"""Return (oldest_buffered_seq, newest_buffered_seq, events)."""
log = self._peek(session_id)
log = self.p_peek(session_id)
if log is None:
return (None, None, [])
# asyncio is single-threaded; deque list() is safe vs concurrent append/eviction. No lock needed for read.
@@ -95,21 +95,21 @@ class SeqLogStore:
def current_seq(self, session_id: str) -> int:
"""Last assigned seq, or 0 if no log exists for the session."""
log = self._peek(session_id)
log = self.p_peek(session_id)
return log.seq if log else 0
def _terminal_path(self, session_id: str) -> Optional[str]:
if not self._persist_dir:
def p_terminal_path(self, session_id: str) -> Optional[str]:
if not self.p_persist_dir:
return None
# Session ids are uuid4 hex; sanitize anyway against path traversal.
safe = "".join(c for c in session_id if c.isalnum() or c in ("-", "_"))
if not safe:
return None
return os.path.join(self._persist_dir, f"{safe}.json")
return os.path.join(self.p_persist_dir, f"{safe}.json")
def persist_terminal(self, session_id: str, payload_str: str) -> None:
"""Atomic write of a terminal event for post-restart clients; best-effort, never blocks broadcast."""
path = self._terminal_path(session_id)
path = self.p_terminal_path(session_id)
if not path:
return
try:
@@ -123,7 +123,7 @@ class SeqLogStore:
)
def load_terminal(self, session_id: str) -> Optional[str]:
path = self._terminal_path(session_id)
path = self.p_terminal_path(session_id)
if not path or not os.path.exists(path):
return None
try:
@@ -134,8 +134,8 @@ class SeqLogStore:
def clear(self, session_id: str) -> None:
"""Drop in-memory log and persisted terminal; for full deletion only, closed-but-retained sessions keep it."""
self._per_session.pop(session_id, None)
path = self._terminal_path(session_id)
self.p_per_session.pop(session_id, None)
path = self.p_terminal_path(session_id)
if path and os.path.exists(path):
try:
os.remove(path)
@@ -143,7 +143,7 @@ class SeqLogStore:
pass
def _default_persist_dir() -> Optional[str]:
def p_default_persist_dir() -> Optional[str]:
try:
from backend.config.paths import DATA_ROOT
return os.path.join(DATA_ROOT, "agents", "terminal_events")
@@ -151,4 +151,4 @@ def _default_persist_dir() -> Optional[str]:
return None
seq_log = SeqLogStore(persist_dir=_default_persist_dir())
SEQ_LOG = SeqLogStore(persist_dir=p_default_persist_dir())
backend/apps/agents/core/ws_manager.py
+8 -8
View File
@@ -3,7 +3,7 @@ import json
import logging
from fastapi import WebSocket
from backend.apps.agents.core.seq_log import TERMINAL_STATUSES, seq_log
from backend.apps.agents.core.seq_log import TERMINAL_STATUSES, SEQ_LOG
logger = logging.getLogger(__name__)
@@ -42,7 +42,7 @@ async def _await_reconnect(has_conn) -> bool:
class ConnectionManager:
"""Manages WebSocket connections and HITL approval bridging; events flow through seq_log so reconnects can replay."""
"""Manages WebSocket connections and HITL approval bridging; events flow through SEQ_LOG so reconnects can replay."""
def __init__(self):
self.connections: dict[str, list[WebSocket]] = {}
@@ -75,7 +75,7 @@ class ConnectionManager:
async def send_to_session(self, session_id: str, event: str, data: dict):
"""Broadcast a session event with monotonic sequencing; terminal statuses also persist to disk."""
async with seq_log.stamp(session_id, event, data) as (seq, payload_str):
async with SEQ_LOG.stamp(session_id, event, data) as (seq, payload_str):
for ws in list(self.connections.get(session_id, [])):
try:
await ws.send_text(payload_str)
@@ -88,13 +88,13 @@ class ConnectionManager:
logger.debug("send_to_session: global send failed", exc_info=True)
# Persist under the lock so a concurrent running status can't race past and overwrite with stale state.
if event == "agent:status" and data.get("status") in TERMINAL_STATUSES:
seq_log.persist_terminal(session_id, payload_str)
SEQ_LOG.persist_terminal(session_id, payload_str)
async def replay_to(
self, session_id: str, websocket: WebSocket, last_seq: int
) -> dict:
"""Replay buffered events with seq > last_seq; returns ack envelope for the resume handshake."""
oldest, newest, events = seq_log.replay(session_id, last_seq)
oldest, newest, events = SEQ_LOG.replay(session_id, last_seq)
# Gap-check first: if last_seq predates the buffer, signal REST-refresh; last_seq=0 means fresh client (full replay).
if last_seq > 0 and oldest is not None and last_seq < oldest - 1:
@@ -150,7 +150,7 @@ class ConnectionManager:
"to_seq": newest,
}
terminal = seq_log.load_terminal(session_id)
terminal = SEQ_LOG.load_terminal(session_id)
if terminal is not None:
try:
await websocket.send_text(terminal)
@@ -211,7 +211,7 @@ class ConnectionManager:
return out
async def broadcast_global(self, event: str, data: dict):
"""Send to all dashboard connections; bypasses seq_log (dashboard resumes via full state refetch)."""
"""Send to all dashboard connections; bypasses SEQ_LOG (dashboard resumes via full state refetch)."""
payload = json.dumps({"event": event, "data": data})
for ws in list(self.global_connections):
try:
@@ -286,7 +286,7 @@ class ConnectionManager:
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
# sent into that gap is lost forever (broadcast skips seq_log).
# sent into that gap is lost forever (broadcast skips SEQ_LOG).
# The renderer dedupes by request_id so re-sends can't double-act.
while True:
await self.broadcast_global("browser:command", payload)
backend/apps/agents/manager/prompt/attachments.py
+16 -16
View File
@@ -1,9 +1,9 @@
import os
from backend.apps.agents.manager.prompt.prompt_context import _resolve_attached_skills, _resolve_forced_tools
from backend.apps.agents.manager.prompt.prompt_context import resolve_attached_skills, resolve_forced_tools
def _build_dir_tree(root: str, max_depth: int = 4, prefix: str = "") -> list[str]:
def build_dir_tree(root: str, max_depth: int = 4, prefix: str = "") -> list[str]:
"""Build a recursive directory tree listing."""
lines = []
try:
@@ -17,12 +17,12 @@ def _build_dir_tree(root: str, max_depth: int = 4, prefix: str = "") -> list[str
for d in dirs:
lines.append(f"{prefix}{d}/")
if max_depth > 1:
sub = _build_dir_tree(os.path.join(root, d), max_depth - 1, prefix + " ")
sub = build_dir_tree(os.path.join(root, d), max_depth - 1, prefix + " ")
lines.extend(sub)
return lines
def _build_prompt_content(prompt: str, images: list | None = None, context_paths: list | None = None, forced_tools: list[str] | None = None, attached_skills: list | None = None, api_type: str = "anthropic", model: str = ""):
def build_prompt_content(prompt: str, images: list | None = None, context_paths: list | None = None, forced_tools: list[str] | None = None, attached_skills: list | None = None, api_type: str = "anthropic", model: str = ""):
"""Build message content for the Anthropic SDK's prompt stream.
Routes attachments per provider:
@@ -42,11 +42,11 @@ def _build_prompt_content(prompt: str, images: list | None = None, context_paths
anything binary, since native shape varies wildly. Caller can
opt-in to the OR file-parser via a separate plugins config.
"""
context_text, native_blocks, refusals = _resolve_attachments(
context_text, native_blocks, refusals = resolve_attachments(
context_paths, api_type=api_type, model=model,
)
forced_tools_text = _resolve_forced_tools(forced_tools)
skills_text = _resolve_attached_skills(attached_skills)
forced_tools_text = resolve_forced_tools(forced_tools)
skills_text = resolve_attached_skills(attached_skills)
refusal_text = "\n\n".join(refusals)
parts = [p for p in (forced_tools_text, context_text, refusal_text, skills_text, prompt) if p]
@@ -69,7 +69,7 @@ def _build_prompt_content(prompt: str, images: list | None = None, context_paths
return content
def _resolve_attachments(context_paths: list | None, api_type: str, model: str) -> tuple[str, list[dict], list[str]]:
def resolve_attachments(context_paths: list | None, api_type: str, model: str) -> tuple[str, list[dict], list[str]]:
"""Split context_paths into:
- inline text (returned as the existing <context_file> block string)
- native content blocks for this provider (PDFs/images)
@@ -90,8 +90,8 @@ def _resolve_attachments(context_paths: list | None, api_type: str, model: str)
"""
if not context_paths:
return "", [], []
from backend.apps.settings.settings import _sniff_file_kind
import base64 as _b64
from backend.apps.settings.settings import sniff_file_kind
import base64 as b64
sections: list[str] = []
native: list[dict] = []
refusals: list[str] = []
@@ -164,7 +164,7 @@ def _resolve_attachments(context_paths: list | None, api_type: str, model: str)
sections.append(f"[Context: {path}, not found]")
continue
if cp_type == "directory" and os.path.isdir(path):
tree_lines = _build_dir_tree(path, max_depth=4)
tree_lines = build_dir_tree(path, max_depth=4)
sections.append(
f"<context_directory path=\"{path}\">\n{chr(10).join(tree_lines)}\n</context_directory>"
)
@@ -176,7 +176,7 @@ def _resolve_attachments(context_paths: list | None, api_type: str, model: str)
size = os.path.getsize(path)
with open(path, "rb") as fh:
head = fh.read(4096)
kind, media_type = _sniff_file_kind(head, os.path.basename(path))
kind, media_type = sniff_file_kind(head, os.path.basename(path))
if kind == "text":
with open(path, "r", errors="replace") as f:
@@ -230,7 +230,7 @@ def _resolve_attachments(context_paths: list | None, api_type: str, model: str)
)
continue
with open(path, "rb") as fh:
data_b64 = _b64.b64encode(fh.read()).decode("ascii")
data_b64 = b64.b64encode(fh.read()).decode("ascii")
block = {
"type": "document",
"source": {
@@ -263,7 +263,7 @@ def _resolve_attachments(context_paths: list | None, api_type: str, model: str)
)
continue
with open(path, "rb") as fh:
data_b64 = _b64.b64encode(fh.read()).decode("ascii")
data_b64 = b64.b64encode(fh.read()).decode("ascii")
native.append({
"type": "image",
"source": {
@@ -299,7 +299,7 @@ def _resolve_attachments(context_paths: list | None, api_type: str, model: str)
# Legacy entry point retained for any external caller; routes to the
# new attachment resolver with anthropic-default routing (no native
# blocks emitted, so behavior is the safe text-only old path).
def _resolve_context_paths(context_paths: list | None) -> str:
text, _native, refusals = _resolve_attachments(context_paths, api_type="anthropic", model="")
def resolve_context_paths(context_paths: list | None) -> str:
text, _, 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/manager/prompt/prompt_context.py
+12 -12
View File
@@ -5,10 +5,10 @@ from backend.apps.tools_lib.tools_lib import (
_load_all as load_all_tools,
)
from backend.apps.tools_lib.mcp_config import sanitize_mcp_server_name
from backend.apps.agents.manager.prompt.tool_catalog import _get_denied_tool_names, _is_fully_denied
from backend.apps.agents.manager.prompt.tool_catalog import get_denied_tool_names, is_fully_denied
def _resolve_mode(mode_id: str, get_all_tool_names: Callable[[], list[str]]) -> tuple[list[str], str | None, str | None]:
def resolve_mode(mode_id: str, get_all_tool_names: Callable[[], list[str]]) -> tuple[list[str], str | None, str | None]:
"""Return (tools, system_prompt, default_folder) resolved from the mode store."""
mode_def = load_mode(mode_id)
if mode_def:
@@ -17,7 +17,7 @@ def _resolve_mode(mode_id: str, get_all_tool_names: Callable[[], list[str]]) ->
return get_all_tool_names(), None, None
def _build_connected_tools_context(allowed_tools: list[str], get_all_tool_names: Callable[[], list[str]]) -> str | None:
def build_connected_tools_context(allowed_tools: list[str], get_all_tool_names: Callable[[], list[str]]) -> str | None:
"""Build a context block describing connected MCP tools and their accounts.
Tools set to 'deny' and fully-denied servers are excluded.
@@ -31,11 +31,11 @@ def _build_connected_tools_context(allowed_tools: list[str], get_all_tool_names:
if tool_ref not in allowed_tools and allowed_tools != get_all_tool_names():
continue
if _is_fully_denied(tool):
if is_fully_denied(tool):
continue
server_name = sanitize_mcp_server_name(tool.name)
denied = _get_denied_tool_names(tool)
denied = get_denied_tool_names(tool)
tool_descs = {
k: v for k, v in tool.tool_permissions.get("_tool_descriptions", {}).items()
if k not in denied
@@ -98,7 +98,7 @@ def _build_connected_tools_context(allowed_tools: list[str], get_all_tool_names:
)
def _build_browser_context(dashboard_id: str | None, selected_browser_ids: list[str] | None = None) -> str | None:
def build_browser_context(dashboard_id: str | None, selected_browser_ids: list[str] | None = None) -> str | None:
"""Build a context block listing browser cards and delegation instructions.
Only browser cards explicitly selected by the user are included.
@@ -174,7 +174,7 @@ def _build_browser_context(dashboard_id: str | None, selected_browser_ids: list[
return "\n".join(lines)
def _build_selected_app_context(selected_app_output_ids: list[str] | None) -> str | None:
def build_selected_app_context(selected_app_output_ids: list[str] | None) -> str | None:
"""Build a context block for dashboard App cards the user selected to edit.
Resolves each Output id to its on-disk workspace so the agent edits the
@@ -233,7 +233,7 @@ def _build_selected_app_context(selected_app_output_ids: list[str] | None) -> st
)
def _build_mcp_registry_summary(allowed_tools: list[str], active_mcps: list[str], get_all_tool_names: Callable[[], list[str]]) -> str | None:
def build_mcp_registry_summary(allowed_tools: list[str], active_mcps: list[str], get_all_tool_names: Callable[[], list[str]]) -> str | None:
"""Compact registry of installed MCP servers, one line per server.
This is the visible surface that drives the activation gate: the model
@@ -261,7 +261,7 @@ def _build_mcp_registry_summary(allowed_tools: list[str], active_mcps: list[str]
tool_ref = f"mcp:{tool.name}"
if tool_ref not in allowed_tools and allowed_tools != get_all_tool_names():
continue
if _is_fully_denied(tool):
if is_fully_denied(tool):
continue
server_name = sanitize_mcp_server_name(tool.name)
desc = (getattr(tool, "description", None) or "").strip()
@@ -354,14 +354,14 @@ AGENT_IDENTITY = (
)
def _compose_system_prompt(default_prompt: str | None, mode_prompt: str | None, session_prompt: str | None, connected_tools_ctx: str | None = None, browser_ctx: str | None = None, mcp_registry_ctx: str | None = None) -> str | None:
def compose_system_prompt(default_prompt: str | None, mode_prompt: str | None, session_prompt: str | None, connected_tools_ctx: str | None = None, browser_ctx: str | None = None, mcp_registry_ctx: str | None = None) -> str | None:
# Identity always leads so it overrides the preset's Claude Code persona, even
# when the user has no custom default/mode/session prompt of their own.
parts = [AGENT_IDENTITY] + [p for p in (default_prompt, mode_prompt, session_prompt, connected_tools_ctx, mcp_registry_ctx, browser_ctx) if p]
return "\n\n".join(parts)
def _resolve_forced_tools(forced_tools: list[str] | None) -> str:
def resolve_forced_tools(forced_tools: list[str] | None) -> str:
"""Build a context block describing explicitly requested tools."""
if not forced_tools:
return ""
@@ -401,7 +401,7 @@ def _resolve_forced_tools(forced_tools: list[str] | None) -> str:
)
def _resolve_attached_skills(attached_skills: list | None) -> str:
def resolve_attached_skills(attached_skills: list | None) -> str:
"""Build a context block injecting attached skill content into the prompt."""
if not attached_skills:
return ""
backend/apps/agents/manager/prompt/tool_catalog.py
+5 -5
View File
@@ -14,7 +14,7 @@ FULL_TOOLS = [
]
def _get_denied_tool_names(tool) -> set[str]:
def get_denied_tool_names(tool) -> set[str]:
"""Return the set of MCP sub-tool names whose permission is 'deny'."""
return {
key for key, value in tool.tool_permissions.items()
@@ -22,14 +22,14 @@ def _get_denied_tool_names(tool) -> set[str]:
}
def _get_all_known_tool_names(tool) -> set[str]:
def get_all_known_tool_names(tool) -> set[str]:
"""Return all known sub-tool names for an MCP tool (from _tool_descriptions)."""
return set(tool.tool_permissions.get("_tool_descriptions", {}).keys())
def _is_fully_denied(tool) -> bool:
def is_fully_denied(tool) -> bool:
"""True when every known sub-tool on this MCP server is set to 'deny'."""
known = _get_all_known_tool_names(tool)
known = get_all_known_tool_names(tool)
if not known:
return False
return known <= _get_denied_tool_names(tool)
return known <= get_denied_tool_names(tool)
backend/apps/agents/manager/session/cloud_sync.py
+3 -3
View File
@@ -1,10 +1,10 @@
from datetime import datetime
from backend.apps.agents.core.models import AgentSession
from backend.apps.service.client import sync as _sync
from backend.apps.service.client import sync
def _sync_session_close(session: AgentSession, close_reason: str = "user"):
def sync_session_close(session: AgentSession, close_reason: str = "user"):
"""Submit the session state to the cloud on close. The cloud
consumes the dump however it sees fit; the desktop just hands off
a snapshot. Skipped for mock sessions so dev runs don't post to
@@ -31,6 +31,6 @@ def _sync_session_close(session: AgentSession, close_reason: str = "user"):
dump = session.model_dump(mode="json")
if not dump.get("closed_at"):
dump["closed_at"] = datetime.now().isoformat()
_sync(dump)
sync(dump)
except Exception:
pass
backend/apps/agents/manager/session/history_compaction.py
+3 -3
View File
@@ -8,7 +8,7 @@ from backend.config.paths import SESSIONS_DIR
logger = logging.getLogger(__name__)
def _get_branch_messages(session) -> list:
def get_branch_messages(session) -> list:
"""Return the linear message list for the active branch, walking the branch tree."""
branch_id = session.active_branch_id or "main"
branch = session.branches.get(branch_id)
@@ -48,7 +48,7 @@ def _get_branch_messages(session) -> list:
return result
def _build_history_prefix(messages, cutoff_msg_id: str | None = None) -> str:
def build_history_prefix(messages, cutoff_msg_id: str | None = None) -> str:
"""Format branch messages into a conversation summary for context injection.
When `cutoff_msg_id` is provided (session.compacted_through_msg_id), drop every
@@ -71,7 +71,7 @@ def _build_history_prefix(messages, cutoff_msg_id: str | None = None) -> str:
return "<prior_conversation>\n" + "\n".join(lines) + "\n</prior_conversation>"
def _truncate_large_tool_result(content: object, session_id: str, msg_id: str, max_bytes: int = 50_000) -> tuple[object, str | None]:
def truncate_large_tool_result(content: object, session_id: str, msg_id: str, max_bytes: int = 50_000) -> tuple[object, str | None]:
"""Spill a large tool_result body to disk, return a truncated
inline replacement plus the on-disk path (or None if untouched).
backend/apps/agents/manager/session/session_store.py
+9 -9
View File
@@ -4,32 +4,32 @@ from backend.apps.agents.core.models import AgentSession
from backend.config.json_store import read_json_or_none, atomic_write_json
def _sessions_dir() -> str:
def p_sessions_dir() -> str:
# Resolve live so test patches on either the paths module or the
# agent_manager facade re-export land on the same directory.
from backend.config.paths import SESSIONS_DIR
return SESSIONS_DIR
def _save_session(session_id: str, doc_data: dict):
sessions_dir = _sessions_dir()
def save_session(session_id: str, doc_data: dict):
sessions_dir = p_sessions_dir()
os.makedirs(sessions_dir, exist_ok=True)
atomic_write_json(os.path.join(sessions_dir, f"{session_id}.json"), doc_data)
def _load_session_data(session_id: str) -> dict | None:
return read_json_or_none(os.path.join(_sessions_dir(), f"{session_id}.json"))
def load_session_data(session_id: str) -> dict | None:
return read_json_or_none(os.path.join(p_sessions_dir(), f"{session_id}.json"))
def _delete_session_file(session_id: str):
path = os.path.join(_sessions_dir(), f"{session_id}.json")
def delete_session_file(session_id: str):
path = os.path.join(p_sessions_dir(), f"{session_id}.json")
if os.path.exists(path):
os.remove(path)
def _load_all_session_data() -> list[tuple[str, dict]]:
def load_all_session_data() -> list[tuple[str, dict]]:
results = []
sessions_dir = _sessions_dir()
sessions_dir = p_sessions_dir()
if not os.path.exists(sessions_dir):
return results
for fname in os.listdir(sessions_dir):
backend/apps/agents/manager/session/workspace_git.py
+22 -23
View File
@@ -1,10 +1,11 @@
import logging
import subprocess
import os
logger = logging.getLogger(__name__)
def _ensure_cwd_git_repo(cwd: str, home: str | None = None) -> None:
def ensure_cwd_git_repo(cwd: str, home: str | None = None) -> None:
"""Idempotently make `cwd` into a git repo with a valid HEAD.
The CLI's built-in Agent tool uses `isolation: "worktree"` to spawn
@@ -30,36 +31,35 @@ def _ensure_cwd_git_repo(cwd: str, home: str | None = None) -> None:
return
if not os.path.isdir(cwd):
return
import subprocess as _sp_git
# Case A: cwd is inside some git repo (possibly parent). Verify
# HEAD resolves. If the enclosing repo is broken (e.g. a stray
# `.git` in $HOME with no commits, which makes workspaces
# under ~/.openswarm/workspaces/ inherit a broken HEAD), we
# need to init a fresh repo AT cwd so it shadows the parent.
_inside = _sp_git.run(
inside = subprocess.run(
["git", "rev-parse", "--is-inside-work-tree"],
cwd=cwd,
stdout=_sp_git.PIPE, stderr=_sp_git.DEVNULL, timeout=5,
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=5,
)
if _inside.returncode == 0 and b"true" in _inside.stdout:
if inside.returncode == 0 and b"true" in inside.stdout:
# Check HEAD resolves (has at least one commit).
_head = _sp_git.run(
head = subprocess.run(
["git", "rev-parse", "--verify", "HEAD"],
cwd=cwd,
stdout=_sp_git.DEVNULL, stderr=_sp_git.DEVNULL, timeout=5,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=5,
)
if _head.returncode == 0:
if head.returncode == 0:
return # parent repo is healthy, leave it alone
# Parent repo exists but HEAD is broken.
if os.path.isdir(os.path.join(cwd, ".git")):
# .git is directly here, commit to fix it.
_sp_git.run(
subprocess.run(
["git", "-c", "user.email=openswarm@local",
"-c", "user.name=OpenSwarm",
"commit", "--allow-empty", "-q", "-m", "openswarm init"],
cwd=cwd,
stdout=_sp_git.DEVNULL, stderr=_sp_git.DEVNULL, timeout=10,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10,
)
return
# .git is in a parent dir (broken home-dir repo, etc.).
@@ -68,23 +68,23 @@ def _ensure_cwd_git_repo(cwd: str, home: str | None = None) -> None:
# Case B: cwd is not a git repo at all (or parent is broken):
# init + empty commit here.
_sp_git.run(
subprocess.run(
["git", "init", "-q", "-b", "main"],
cwd=cwd,
stdout=_sp_git.DEVNULL, stderr=_sp_git.DEVNULL, timeout=10,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10,
)
_sp_git.run(
subprocess.run(
["git", "-c", "user.email=openswarm@local",
"-c", "user.name=OpenSwarm",
"commit", "--allow-empty", "-q", "-m", "openswarm init"],
cwd=cwd,
stdout=_sp_git.DEVNULL, stderr=_sp_git.DEVNULL, timeout=10,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10,
)
except Exception as _e:
logger.info(f"[agent-cwd] git init skipped: {_e}")
except Exception as e:
logger.info(f"[agent-cwd] git init skipped: {e}")
def _detect_git_identity(cwd: str) -> tuple[str | None, str | None]:
def detect_git_identity(cwd: str) -> tuple[str | None, str | None]:
"""Resolve the origin remote and current branch for `cwd`.
Used to label sessions in the session list ("Agent on owner/repo
@@ -97,10 +97,9 @@ def _detect_git_identity(cwd: str) -> tuple[str | None, str | None]:
if not cwd or not os.path.isdir(cwd):
return (None, None)
try:
import subprocess as _sp
url_proc = _sp.run(
url_proc = subprocess.run(
["git", "remote", "get-url", "origin"],
cwd=cwd, stdout=_sp.PIPE, stderr=_sp.DEVNULL, timeout=3,
cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=3,
)
repo_url: str | None = None
if url_proc.returncode == 0:
@@ -113,9 +112,9 @@ def _detect_git_identity(cwd: str) -> tuple[str | None, str | None]:
repo_url = f"{scheme}://{rest}"
else:
repo_url = raw
branch_proc = _sp.run(
branch_proc = subprocess.run(
["git", "branch", "--show-current"],
cwd=cwd, stdout=_sp.PIPE, stderr=_sp.DEVNULL, timeout=3,
cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=3,
)
branch_name: str | None = None
if branch_proc.returncode == 0:
backend/apps/agents/providers/openrouter.py
+23 -23
View File
@@ -3,6 +3,8 @@
from __future__ import annotations
import logging
import time
import httpx
logger = logging.getLogger(__name__)
@@ -10,13 +12,13 @@ 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}
P_9ROUTER_CACHE: dict = {"available": None, "checked_at": 0}
# Per-model published pricing in $/1M tokens (input, output) for direct
@@ -25,7 +27,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 +55,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 +63,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,26 +73,24 @@ 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]:
"""Return OR's tool-capable chat catalog. Cached. Never raises."""
import time as _time
if not api_key:
invalidate_openrouter_cache()
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
now = time.monotonic()
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"]
import httpx
return P_OR_MODELS_CACHE["models"]
try:
async with httpx.AsyncClient(timeout=8.0) as client:
r = await client.get(
@@ -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/pricing.py
+6 -6
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import re
# ---------------------------------------------------------------------------
# Curated model tiers; Intelligence, Speed, Cost on a 1-5 scale
@@ -185,7 +186,7 @@ MODEL_TIERS: dict[str, tuple[int, int, int]] = {
}
def _heuristic_tiers(label: str, output_cost_per_1m: float, reasoning: bool) -> tuple[int, int, int]:
def p_heuristic_tiers(label: str, output_cost_per_1m: float, reasoning: bool) -> tuple[int, int, int]:
"""Fallback tier scoring for models not in MODEL_TIERS. Tries to
extract a parameter count from the label (8B/70B/235B/etc.) and
use that as a stronger size signal than cost alone, since open-
@@ -204,7 +205,6 @@ def _heuristic_tiers(label: str, output_cost_per_1m: float, reasoning: bool) ->
- inverse of size, with name keywords as ±1 nudges.
Cost: pure cost bucket.
"""
import re as _re
out = output_cost_per_1m or 0.0
# Cost bucket; same 5-tier cost ladder as before.
@@ -225,7 +225,7 @@ def _heuristic_tiers(label: str, output_cost_per_1m: float, reasoning: bool) ->
# clearly above 1B (so we don't pick up version numbers).
lower = (label or "").lower()
param_b = 0.0
for m in _re.finditer(r"\b(\d{1,4}(?:\.\d+)?)\s*b\b", lower):
for m in re.finditer(r"\b(\d{1,4}(?:\.\d+)?)\s*b\b", lower):
try:
v = float(m.group(1))
if v >= 1 and v > param_b:
@@ -259,9 +259,9 @@ def _heuristic_tiers(label: str, output_cost_per_1m: float, reasoning: bool) ->
# Speed inverse of intel.
speed = 6 - intel
if _re.search(r"\b(mini|lite|flash|haiku|nano|small|fast|turbo|micro|tiny)\b", lower):
if re.search(r"\b(mini|lite|flash|haiku|nano|small|fast|turbo|micro|tiny)\b", lower):
speed += 1
if _re.search(r"\b(opus|ultra|max|xlarge|titan|huge)\b", lower):
if re.search(r"\b(opus|ultra|max|xlarge|titan|huge)\b", lower):
speed -= 1
if reasoning and intel >= 4:
# Frontier reasoning models burn lots of tokens on hidden
@@ -310,7 +310,7 @@ def compute_tiers(
if c in MODEL_TIERS:
return MODEL_TIERS[c]
return _heuristic_tiers(label, output_cost_per_1m, reasoning)
return p_heuristic_tiers(label, output_cost_per_1m, reasoning)
def compute_billing_kind(
backend/apps/agents/providers/registry.py
+40 -39
View File
@@ -10,9 +10,10 @@ from __future__ import annotations
import logging
from typing import Any, TYPE_CHECKING
import httpx
from .openrouter import (
_OPENROUTER_VALUE_PREFIX,
OPENROUTER_VALUE_PREFIX,
)
if TYPE_CHECKING:
@@ -21,7 +22,7 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
# Full set of model-id prefixes that force routing through 9Router.
_NINEROUTER_MODEL_PREFIXES = ("cc/", "cx/", "gc/", "ag/", "gemini/", "openrouter/")
NINEROUTER_MODEL_PREFIXES = ("cc/", "cx/", "gc/", "ag/", "gemini/", "openrouter/")
# Entry fields: value, label, context_window, model_id, router_model_id, api,
# subscription_only, reasoning, route ("cc"|"api"|"openrouter"|None).
@@ -151,10 +152,10 @@ BUILTIN_MODELS: dict[str, list[dict[str, Any]]] = {
# Model resolution (used by the live claude_agent_sdk path)
# ---------------------------------------------------------------------------
_CUSTOM_VALUE_PREFIX = "custom/"
P_CUSTOM_VALUE_PREFIX = "custom/"
def _custom_provider_slug_for_lookup(name: str) -> str:
def custom_provider_slug_for_lookup(name: str) -> str:
"""Mirror nine_router._custom_provider_slug; duplicated here to avoid
importing from nine_router (circular: nine_router imports from settings)."""
import re
@@ -162,22 +163,22 @@ def _custom_provider_slug_for_lookup(name: str) -> str:
return s or "custom"
def _find_custom_provider_for_value(settings, value: str):
def find_custom_provider_for_value(settings, value: str):
"""Look up the CustomProvider whose slug matches the slug encoded in a
`custom/<slug>/<model_id>` picker value. Returns None if no match."""
if not isinstance(value, str) or not value.startswith(_CUSTOM_VALUE_PREFIX):
if not isinstance(value, str) or not value.startswith(P_CUSTOM_VALUE_PREFIX):
return None
rest = value[len(_CUSTOM_VALUE_PREFIX):]
slug, _sep, _bare = rest.partition("/")
rest = value[len(P_CUSTOM_VALUE_PREFIX):]
slug, _, _ = rest.partition("/")
if not slug:
return None
for cp in getattr(settings, "custom_providers", None) or []:
if _custom_provider_slug_for_lookup(getattr(cp, "name", "")) == slug:
if custom_provider_slug_for_lookup(getattr(cp, "name", "")) == slug:
return cp
return None
def _find_builtin_model(short_name: str) -> dict | None:
def find_builtin_model(short_name: str) -> dict | None:
"""Look up a model entry by its short `value`.
OpenRouter entries (prefixed `or:<vendor>/<model>`) and custom-provider
@@ -188,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,
@@ -201,9 +202,9 @@ def _find_builtin_model(short_name: str) -> dict | None:
"route": "openrouter",
"reasoning": False,
}
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("/")
if isinstance(short_name, str) and short_name.startswith(P_CUSTOM_VALUE_PREFIX):
rest = short_name[len(P_CUSTOM_VALUE_PREFIX):]
slug, _, 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.
@@ -222,13 +223,25 @@ def _find_builtin_model(short_name: str) -> dict | None:
def get_api_type(short_name: str) -> str:
entry = _find_builtin_model(short_name)
entry = find_builtin_model(short_name)
return (entry or {}).get("api", "anthropic")
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-
# budget alias on AG requires a thinking_config the CLI doesn't
# emit). Falls through to gc/gemini-3.1-pro-preview, which works
# for non-tool turns; multi-step tool turns still hit the
# thoughtSignature validator but that's a separate fight.
"gemini-3-flash-preview": "gemini-3-flash",
"gemini-3.1-flash-lite-preview": "gemini-3-flash",
}
def resolve_model_id_for_sdk(short_name: str, settings: AppSettings) -> str:
"""Short model name → id string for ClaudeAgentOptions."""
entry = _find_builtin_model(short_name)
entry = find_builtin_model(short_name)
if entry is None:
return short_name
if entry.get("route") == "cc":
@@ -250,28 +263,16 @@ def resolve_model_id_for_sdk(short_name: str, settings: AppSettings) -> str:
# AG bypasses the thoughtSignature validator that breaks multi-step tool
# turns on gc/. Without it, every Gemini turn 400s after the first tool
# call with "Thought signature is not valid".
_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-
# budget alias on AG requires a thinking_config the CLI doesn't
# emit). Falls through to gc/gemini-3.1-pro-preview, which works
# for non-tool turns; multi-step tool turns still hit the
# thoughtSignature validator but that's a separate fight.
"gemini-3-flash-preview": "gemini-3-flash",
"gemini-3.1-flash-lite-preview": "gemini-3-flash",
}
if entry.get("api") == "gemini-cli":
rid = entry.get("router_model_id", "")
if isinstance(rid, str) and rid.startswith("gc/"):
suffix = rid[len("gc/"):]
if getattr(settings, "google_api_key", None):
return "gemini/" + suffix
ag_suffix = _ANTIGRAVITY_MAP.get(suffix)
ag_suffix = P_ANTIGRAVITY_MAP.get(suffix)
if ag_suffix:
try:
import httpx as _httpx
r = _httpx.get("http://localhost:20128/api/providers", timeout=2.0)
r = httpx.get("http://localhost:20128/api/providers", timeout=2.0)
if r.status_code == 200:
data = r.json()
conns = data.get("connections", []) if isinstance(data, dict) else (data if isinstance(data, list) else [])
@@ -305,14 +306,14 @@ async def resolve_aux_model(
or_sonnet = "openrouter/anthropic/claude-sonnet-4.5"
bare = haiku_bare if preferred_tier == "haiku" else sonnet_bare
or_aux = or_haiku if preferred_tier == "haiku" else or_sonnet
from backend.apps.nine_router import is_running as _9r_running, get_providers as _9r_providers
from backend.apps.nine_router.process import is_running, get_providers
base_url = "http://localhost:20128"
connected: set[str] = set()
if _9r_running():
if is_running():
try:
connections = await _9r_providers()
connections = await get_providers()
connected = {c.get("provider") for c in connections if c.get("isActive")}
except Exception:
connected = set()
@@ -340,7 +341,7 @@ async def resolve_aux_model(
if getattr(settings, "anthropic_api_key", None):
return (bare, None)
if not _9r_running():
if not is_running():
raise ValueError(
"No AI provider configured for auxiliary LLM call. "
"Set an Anthropic API key or connect a subscription."
@@ -375,9 +376,9 @@ def get_context_window(model: str, settings: AppSettings | None = None) -> int:
# bare-model tail against any custom provider's models list.
if settings:
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("/")
if isinstance(model, str) and model.startswith(P_CUSTOM_VALUE_PREFIX):
rest = model[len(P_CUSTOM_VALUE_PREFIX):]
_, _, 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/auth/router.py
+3 -3
View File
@@ -52,7 +52,7 @@ async def _sync_pro_routing(settings_obj) -> None:
paying user into pro mode and sign-out must tear the lane down so a
revoked bearer doesn't linger in the router."""
try:
from backend.apps.nine_router import sync_pro_routing
from backend.apps.nine_router.sync_custom import sync_pro_routing
await sync_pro_routing(settings_obj)
except Exception as e:
logger.debug("pro routing sync skipped: %s", e)
@@ -211,7 +211,7 @@ async def signout():
# Best-effort: failures here shouldn't block the sign-out itself.
try:
from backend.apps.agents.agent_manager import agent_manager
from backend.apps.agents.agent_manager import _save_session
from backend.apps.agents.manager.session.session_store import save_session
running = list(agent_manager.tasks.keys())
for session_id in running:
@@ -227,7 +227,7 @@ async def signout():
if sess.sdk_session_id:
sess.sdk_session_id = None
try:
_save_session(sess.id, sess.model_dump(mode="json"))
save_session(sess.id, sess.model_dump(mode="json"))
except Exception as e:
logger.warning("signout: save_session(%s) failed: %s", sess.id, e)
backend/apps/dashboards/dashboards.py
+2 -2
View File
@@ -421,7 +421,7 @@ async def duplicate_dashboard(dashboard_id: str):
now = datetime.now().isoformat()
from backend.apps.agents.agent_manager import agent_manager
from backend.apps.agents.manager.session.session_store import _save_session
from backend.apps.agents.manager.session.session_store import save_session
source_layout = source_data.get("layout", {}) or {}
source_browser_cards = source_layout.get("browser_cards", {}) or {}
@@ -471,7 +471,7 @@ async def duplicate_dashboard(dashboard_id: str):
new_sess.browser_id = browser_id_remap[old_browser_id]
if old_parent_sid and old_parent_sid in session_id_remap:
new_sess.parent_session_id = session_id_remap[old_parent_sid]
_save_session(new_sess.id, new_sess.model_dump(mode="json"))
save_session(new_sess.id, new_sess.model_dump(mode="json"))
new_cards: dict[str, dict] = {}
for old_sid, card in source_cards.items():
backend/apps/nine_router/__init__.py
-87
View File
@@ -1,87 +0,0 @@
"""Auto-start and manage the 9Router subprocess.
9Router is a free AI subscription proxy that lets users connect their
Claude/ChatGPT/Gemini subscriptions to OpenSwarm without API keys. It runs
silently on port 20128 and exposes an OpenAI-compatible API at
localhost:20128/v1.
This package was split out of a single ~1190-line module. The public API is
unchanged: every `from backend.apps.nine_router import X` keeps resolving via
the re-exports below.
- process.py: subprocess lifecycle (the single owner of the process handle),
constants, ports/URLs, the pinned NPM version, path resolution, stats.
- sync.py: Gemini/OpenAI/OpenRouter API-key sync.
- sync_custom.py: custom OpenAI-compatible provider + OpenSwarm Pro sync.
- oauth.py: OAuth start/poll/exchange + the Codex 1455 callback listener.
"""
import httpx # noqa: F401 patch point: tests stub backend.apps.nine_router.httpx.AsyncClient
from .process import (
NINE_ROUTER_API,
NINE_ROUTER_NPM_VERSION,
NINE_ROUTER_PORT,
NINE_ROUTER_URL,
NINE_ROUTER_V1,
ensure_running,
get_latest_reasoning_tokens,
get_providers,
get_usage_stats,
is_running,
stop,
)
from .sync import (
NINE_ROUTER_CLAUDE_PRO_NAME,
NINE_ROUTER_KEYED_NAME,
NINE_ROUTER_OPENAI_KEYED_NAME,
NINE_ROUTER_OPENAI_KEYED_PREFIX,
NINE_ROUTER_OPENROUTER_KEYED_NAME,
sync_gemini_api_key,
sync_openai_api_key,
sync_openrouter_api_key,
)
from .sync_custom import (
NINE_ROUTER_CUSTOM_NAME_SUFFIX,
normalize_openai_compat_base_url,
sync_custom_providers,
sync_openswarm_pro_as_claude,
sync_pro_routing,
)
from .oauth import (
exchange_oauth,
get_models,
poll_oauth,
start_oauth,
)
__all__ = [
"NINE_ROUTER_API",
"NINE_ROUTER_NPM_VERSION",
"NINE_ROUTER_PORT",
"NINE_ROUTER_URL",
"NINE_ROUTER_V1",
"NINE_ROUTER_CLAUDE_PRO_NAME",
"NINE_ROUTER_KEYED_NAME",
"NINE_ROUTER_OPENAI_KEYED_NAME",
"NINE_ROUTER_OPENAI_KEYED_PREFIX",
"NINE_ROUTER_OPENROUTER_KEYED_NAME",
"NINE_ROUTER_CUSTOM_NAME_SUFFIX",
"ensure_running",
"stop",
"is_running",
"get_usage_stats",
"get_latest_reasoning_tokens",
"get_providers",
"get_models",
"start_oauth",
"poll_oauth",
"exchange_oauth",
"sync_gemini_api_key",
"sync_openai_api_key",
"sync_openrouter_api_key",
"sync_custom_providers",
"sync_openswarm_pro_as_claude",
"sync_pro_routing",
"normalize_openai_compat_base_url",
]
backend/apps/service/service.py
+13 -12
View File
@@ -75,8 +75,8 @@ async def _pulse_loop():
cost_delta = 0.0
try:
from backend.apps.nine_router import get_usage_stats, is_running as _9r_running
if _9r_running():
from backend.apps.nine_router.process import get_usage_stats, is_running
if is_running():
stats = await get_usage_stats()
if stats:
cur_cost = stats.get("totalCost", 0) or 0
@@ -124,13 +124,14 @@ async def service_lifespan():
global _pulse_task, _drain_task
try:
from backend.apps.settings.settings import load_settings, _save_settings
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import save_settings
settings = load_settings()
is_first_open = settings.first_opened_at is None
if is_first_open:
settings.first_opened_at = datetime.now().isoformat()
_save_settings(settings)
save_settings(settings)
days_since_install = 0
if settings.first_opened_at:
@@ -192,8 +193,8 @@ async def service_lifespan():
logger.debug(f"Service startup event failed (non-critical): {e}")
try:
from backend.apps.nine_router import ensure_running as ensure_9router
await ensure_9router()
from backend.apps.nine_router.process import ensure_running
await ensure_running()
except Exception as e:
logger.debug(f"9Router auto-start skipped: {e}")
@@ -219,8 +220,8 @@ async def service_lifespan():
_drain_task = None
try:
from backend.apps.nine_router import stop as stop_9router
stop_9router()
from backend.apps.nine_router.process import stop
stop()
except Exception:
pass
@@ -321,8 +322,8 @@ async def usage_summary():
completed = status_counts.get("completed", 0)
completion_rate = completed / total_sessions if total_sessions > 0 else 0
from backend.apps.nine_router import get_usage_stats, is_running as _9r_running
nine_router_stats = await get_usage_stats() if _9r_running() else None
from backend.apps.nine_router.process import get_usage_stats, is_running
nine_router_stats = await get_usage_stats() if is_running() else None
if nine_router_stats and nine_router_stats.get("totalCost", 0) > 0:
cost_source = "9router"
@@ -382,8 +383,8 @@ async def usage_summary():
@service.router.get("/cost-breakdown")
async def cost_breakdown(period: str = "7d"):
from backend.apps.nine_router import get_usage_stats, is_running as _9r_running
if not _9r_running():
from backend.apps.nine_router.process import get_usage_stats, is_running
if not is_running():
return {"available": False, "by_model": {}, "by_provider": {}}
stats = await get_usage_stats(period)
if not stats:
backend/apps/settings/settings.py
+26 -40
View File
@@ -24,19 +24,12 @@ logger = logging.getLogger(__name__)
async def settings_lifespan():
os.makedirs(DATA_DIR, exist_ok=True)
try:
from backend.apps.nine_router import (
ensure_running as _9r_ensure,
is_running as _9r_running,
sync_gemini_api_key,
sync_openai_api_key,
sync_openrouter_api_key,
sync_openswarm_pro_as_claude,
sync_custom_providers,
)
from backend.apps.nine_router.process import ensure_running, is_running
from backend.apps.nine_router.sync import sync_gemini_api_key, sync_openai_api_key, sync_openrouter_api_key
from backend.apps.nine_router.sync_custom import sync_openswarm_pro_as_claude, sync_custom_providers
s = load_settings()
import asyncio as _asyncio
async def _boot_router_then_sync():
async def 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),
@@ -47,7 +40,7 @@ async def settings_lifespan():
])
if needs_router:
try:
await _9r_ensure()
await ensure_running()
except Exception as e:
logger.warning(f"9Router lifespan boot failed: {e}")
# Reconcile, don't just add: pass the key OR None so a cleared/never-set key
@@ -55,7 +48,7 @@ async def settings_lifespan():
# old add-only guards left a zombie managed key alive after disconnect, which
# kept routing to it (the "still defaults to gemini") and blocked the free
# trial from arming. Only acts when 9Router is already up (_sync no-ops if not).
if _9r_running():
if is_running():
await sync_gemini_api_key(getattr(s, "google_api_key", None) or None)
await sync_openai_api_key(getattr(s, "openai_api_key", None) or None)
await sync_openrouter_api_key(getattr(s, "openrouter_api_key", None) or None)
@@ -66,14 +59,14 @@ 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(_upload_dir_gc_loop())
asyncio.create_task(boot_router_then_sync())
asyncio.create_task(p_upload_dir_gc_loop())
except Exception as e:
logger.warning(f"9Router sync startup failed: {e}")
yield
async def _upload_dir_gc_loop():
async def p_upload_dir_gc_loop():
"""Daily GC of UPLOAD_DIR. Without this, every PDF/image the user
drops sits in the OS temp dir forever, growing unbounded across
sessions. We keep files for 7 days to make resume-after-restart
@@ -81,7 +74,6 @@ async def _upload_dir_gc_loop():
by the OS but not aggressively; Windows temp is not. Belt and braces.
Errors are swallowed: a chmod hiccup or in-use lock should never
crash the backend."""
import asyncio as _a
while True:
try:
now = time.time()
@@ -96,7 +88,7 @@ async def _upload_dir_gc_loop():
continue
except Exception:
pass
await _a.sleep(24 * 3600)
await asyncio.sleep(24 * 3600)
settings = SubApp("settings", settings_lifespan)
@@ -137,7 +129,7 @@ SERVER_OWNED_FIELDS = (
@settings.router.put("")
async def update_settings(body: AppSettings):
from backend.apps.service.client import sync as _sync
from backend.apps.service.client import sync
old = load_settings()
for k in SERVER_OWNED_FIELDS:
@@ -154,9 +146,8 @@ async def update_settings(body: AppSettings):
body.free_trial_token = None
body.free_trial_remaining = None
try:
import asyncio as _aio
from backend.apps.nine_router import sync_pro_routing as _spr
_aio.create_task(_spr(body)) # drop the now-stale free-trial 9router node
from backend.apps.nine_router.sync_custom import sync_pro_routing
asyncio.create_task(sync_pro_routing(body)) # drop the now-stale free-trial 9router node
except Exception:
pass
@@ -164,11 +155,11 @@ async def update_settings(body: AppSettings):
"claude_subscription_token", "openai_subscription_token", "gemini_subscription_token",
"openswarm_bearer_token", "free_trial_token", "installation_id"}
safe = {k: v for k, v in body.model_dump().items() if k not in secret_keys}
_sync(safe)
sync(safe)
if (body.user_email and body.user_email != getattr(old, "user_email", None)) or \
(body.user_name and body.user_name != getattr(old, "user_name", None)):
from backend.apps.service.client import identify as _identify
from backend.apps.service.client import identify
id_props = {}
if body.user_email:
id_props["email"] = body.user_email
@@ -179,7 +170,7 @@ async def update_settings(body: AppSettings):
if body.user_referral_source:
id_props["referral_source"] = body.user_referral_source
if id_props:
_identify(id_props)
identify(id_props)
await save_settings_async(body)
@@ -208,14 +199,14 @@ async def update_settings(body: AppSettings):
if openrouter_changed:
try:
from backend.apps.agents.providers.registry import invalidate_openrouter_cache
from backend.apps.agents.providers.openrouter import invalidate_openrouter_cache
invalidate_openrouter_cache()
except Exception:
pass
# 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 boot_and_sync_keys(
google_key: str | None,
openai_key: str | None,
openrouter_key: str | None,
@@ -227,16 +218,11 @@ async def update_settings(body: AppSettings):
need_boot: bool,
):
try:
from backend.apps.nine_router import (
ensure_running as _9r_ensure,
is_running as _9r_running,
sync_gemini_api_key,
sync_openai_api_key,
sync_openrouter_api_key,
sync_custom_providers,
)
if need_boot and not _9r_running():
await _9r_ensure()
from backend.apps.nine_router.process import ensure_running, is_running
from backend.apps.nine_router.sync import sync_gemini_api_key, sync_openai_api_key, sync_openrouter_api_key
from backend.apps.nine_router.sync_custom import sync_custom_providers
if need_boot and not is_running():
await ensure_running()
if do_google:
await sync_gemini_api_key(google_key or None)
if do_openai:
@@ -248,7 +234,7 @@ async def update_settings(body: AppSettings):
except Exception as e:
logger.warning(f"Background apikey sync failed: {e}")
asyncio.create_task(_boot_and_sync_keys(
asyncio.create_task(boot_and_sync_keys(
getattr(body, "google_api_key", None),
getattr(body, "openai_api_key", None),
getattr(body, "openrouter_api_key", None),
@@ -306,7 +292,7 @@ UPLOAD_DIR = os.path.join(tempfile.gettempdir(), "self-swarm-uploads")
os.makedirs(UPLOAD_DIR, exist_ok=True)
def _sniff_file_kind(contents: bytes) -> tuple[str, str | None]:
def sniff_file_kind(contents: bytes) -> tuple[str, str | None]:
"""Classify an uploaded file as text/pdf/image/binary so the agent
layer can route it (inline as text, send as native document/image
block, or refuse). Returns (kind, media_type)."""
@@ -436,7 +422,7 @@ async def upload_files(files: list[UploadFile] = File(...)):
pass
raise
kind, media_type = _sniff_file_kind(contents, safe_name)
kind, media_type = sniff_file_kind(contents, safe_name)
if kind == "text":
try:
backend/apps/settings/store.py
+1 -5
View File
@@ -147,8 +147,4 @@ def _atomic_write_settings(payload: dict) -> None:
def save_settings(settings_obj: AppSettings) -> None:
"""Sync atomic persist; thread-safe. Async callers should prefer save_settings_async (Defender can stretch writes to 50-200ms)."""
_atomic_write_settings(settings_obj.model_dump())
def _save_settings(settings_obj: AppSettings) -> None:
save_settings(settings_obj)
_atomic_write_settings(settings_obj.model_dump())
backend/apps/subscription/free_trial.py
+4 -4
View File
@@ -100,10 +100,10 @@ async def _has_connected_subscription() -> bool:
connections live in 9Router, not settings, so the sync check above misses
them; this catches a sub connected while the trial was armed."""
try:
from backend.apps.nine_router import is_running as _9r_running, get_providers as _9r_providers
if not _9r_running():
from backend.apps.nine_router.process import is_running, get_providers
if not is_running():
return False
conns = await _9r_providers()
conns = await get_providers()
return any(
c.get("isActive") and c.get("provider") in ("claude", "codex", "gemini-cli")
for c in conns
@@ -118,7 +118,7 @@ def _proxy_base(settings_obj) -> str:
async def _sync_routing(settings_obj) -> None:
try:
from backend.apps.nine_router import sync_pro_routing
from backend.apps.nine_router.sync_custom import sync_pro_routing
await sync_pro_routing(settings_obj)
except Exception as e:
logger.debug("free-trial routing sync skipped: %s", e)
backend/apps/subscription/router.py
+1 -1
View File
@@ -39,7 +39,7 @@ async def _sync_pro_routing(settings_obj) -> None:
non-Claude primaries). PUT /api/settings no longer carries these fields,
so the state-change endpoints here are the only trigger left."""
try:
from backend.apps.nine_router import sync_pro_routing
from backend.apps.nine_router.sync_custom import sync_pro_routing
await sync_pro_routing(settings_obj)
except Exception as e:
logger.debug("pro routing sync skipped: %s", e)
backend/apps/web/web.py
+5 -5
View File
@@ -193,16 +193,16 @@ async def _refresh_9r_connected() -> set[str]:
(e.g. {"claude", "codex", "antigravity", "gemini-cli"}). Cached for
20s to keep search/fetch endpoints snappy."""
global _NINE_ROUTER_CONNECTED, _NINE_ROUTER_CACHE_AT
import time as _t
now = _t.time()
import time
now = time.time()
if now - _NINE_ROUTER_CACHE_AT < 20.0:
return _NINE_ROUTER_CONNECTED
try:
from backend.apps.nine_router import is_running as _9r_running, get_providers as _9r_providers
if not _9r_running():
from backend.apps.nine_router.process import is_running, get_providers
if not is_running():
_NINE_ROUTER_CONNECTED = set()
else:
conns = await _9r_providers()
conns = await get_providers()
_NINE_ROUTER_CONNECTED = {
c.get("provider")
for c in conns
backend/config/json_store.py
+1 -1
View File
@@ -9,7 +9,7 @@ Two jobs:
on a garbled/unreadable file, so a single corrupt file can't crash a whole
load-all path (and take down boot or a page with it).
Settings/seq_log/auth keep their own inlined atomic writers; this is for the
Settings/SEQ_LOG/auth keep their own inlined atomic writers; this is for the
stores that were still doing plain open()+dump().
"""
import json
backend/main.py
+3 -3
View File
@@ -204,13 +204,13 @@ async def websocket_session(websocket: WebSocket, session_id: str):
last_seq = int(payload.get("last_seq") or 0)
connection_uuid = payload.get("connection_uuid") or ""
ack = await ws_manager.replay_to(session_id, websocket, last_seq)
from backend.apps.agents.core.seq_log import seq_log as _sl
from backend.apps.agents.core.seq_log import SEQ_LOG
await websocket.send_text(json.dumps({
"event": "server:hello",
"session_id": session_id,
"data": {
"connection_uuid": connection_uuid,
"current_seq": _sl.current_seq(session_id),
"current_seq": SEQ_LOG.current_seq(session_id),
"ack": ack,
},
}))
@@ -494,7 +494,7 @@ async def subscriptions_callback(request: Request):
logger.warning(f"OAuth callback with unknown state {state[:8] if state else '(empty)'}...")
return HTMLResponse('<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>Session expired</h2><p style="color:#888">Please try connecting again.</p></div></body></html>')
from backend.apps.nine_router import exchange_oauth
from backend.apps.nine_router.oauth import exchange_oauth
try:
await exchange_oauth(pending["provider"], code, pending["redirect_uri"], pending["code_verifier"], state)
except Exception as e:
backend/tests/test_disconnect_resilience.py
+14 -14
View File
@@ -46,27 +46,27 @@ from fastapi.testclient import TestClient
_TMPROOT = tempfile.mkdtemp(prefix="openswarm-disconnect-test-")
os.environ.setdefault("OPENSWARM_DATA_DIR", _TMPROOT)
# Push the seq_log persist dir to a deterministic location too.
# Push the SEQ_LOG persist dir to a deterministic location too.
_SEQ_DIR = os.path.join(_TMPROOT, "seq_terminals")
os.makedirs(_SEQ_DIR, exist_ok=True)
@pytest.fixture(autouse=True)
def _patch_persist_dir():
"""Force the seq_log to use our tmp dir so we can assert on disk state."""
from backend.apps.agents.core import seq_log as sl_mod
"""Point the shared SEQ_LOG singleton at our tmp dir so we can assert on disk state.
# Rebuild the singleton with our test dir.
new_store = sl_mod.SeqLogStore(persist_dir=_SEQ_DIR)
monkey = patch.object(sl_mod, "seq_log", new_store)
monkey.start()
# Also patch the symbol re-exported into ws_manager's import scope.
from backend.apps.agents.core import ws_manager as wm_mod
wm_monkey = patch.object(wm_mod, "seq_log", new_store)
wm_monkey.start()
yield new_store
monkey.stop()
wm_monkey.stop()
ws_manager imports SEQ_LOG by value, so both modules hold the same object;
patching the object's attribute is visible everywhere with one patch, no
need to rebind a name in each import scope.
"""
from backend.apps.agents.core.seq_log import SEQ_LOG
os.makedirs(_SEQ_DIR, exist_ok=True)
with patch.object(SEQ_LOG, "_persist_dir", _SEQ_DIR):
# Fresh per-session state so each test is isolated.
SEQ_LOG._per_session.clear()
yield SEQ_LOG
SEQ_LOG._per_session.clear()
# ---------------------------------------------------------------------------