From f0d3cabc47a6c15e8e89d2e05aba32df1eb06bc4 Mon Sep 17 00:00:00 2001 From: haikdc Date: Sat, 13 Jun 2026 19:30:34 -0700 Subject: [PATCH] [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 --- backend/apps/agents/agent_manager.py | 206 +++++++++--------- backend/apps/agents/agents.py | 33 +-- backend/apps/agents/core/mcp_preflight.py | 34 +-- .../apps/agents/core/openai_passthrough.py | 121 ---------- backend/apps/agents/core/seq_log.py | 52 ++--- backend/apps/agents/core/ws_manager.py | 16 +- .../apps/agents/manager/prompt/attachments.py | 32 +-- .../agents/manager/prompt/prompt_context.py | 24 +- .../agents/manager/prompt/tool_catalog.py | 10 +- .../apps/agents/manager/session/cloud_sync.py | 6 +- .../manager/session/history_compaction.py | 6 +- .../agents/manager/session/session_store.py | 18 +- .../agents/manager/session/workspace_git.py | 45 ++-- backend/apps/agents/providers/openrouter.py | 46 ++-- backend/apps/agents/providers/pricing.py | 12 +- backend/apps/agents/providers/registry.py | 79 +++---- backend/apps/auth/router.py | 6 +- backend/apps/dashboards/dashboards.py | 4 +- backend/apps/nine_router/__init__.py | 87 -------- backend/apps/service/service.py | 25 ++- backend/apps/settings/settings.py | 66 +++--- backend/apps/settings/store.py | 6 +- backend/apps/subscription/free_trial.py | 8 +- backend/apps/subscription/router.py | 2 +- backend/apps/web/web.py | 10 +- backend/config/json_store.py | 2 +- backend/main.py | 6 +- backend/tests/test_disconnect_resilience.py | 28 +-- 28 files changed, 384 insertions(+), 606 deletions(-) delete mode 100644 backend/apps/agents/core/openai_passthrough.py delete mode 100644 backend/apps/nine_router/__init__.py diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py index c19a47cc..d463a04a 100644 --- a/backend/apps/agents/agent_manager.py +++ b/backend/apps/agents/agent_manager.py @@ -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: diff --git a/backend/apps/agents/agents.py b/backend/apps/agents/agents.py index 9b0891f6..7f6ba3bb 100644 --- a/backend/apps/agents/agents.py +++ b/backend/apps/agents/agents.py @@ -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//. - 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: diff --git a/backend/apps/agents/core/mcp_preflight.py b/backend/apps/agents/core/mcp_preflight.py index 9de5ca70..daf977a6 100644 --- a/backend/apps/agents/core/mcp_preflight.py +++ b/backend/apps/agents/core/mcp_preflight.py @@ -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( diff --git a/backend/apps/agents/core/openai_passthrough.py b/backend/apps/agents/core/openai_passthrough.py deleted file mode 100644 index 72a635f0..00000000 --- a/backend/apps/agents/core/openai_passthrough.py +++ /dev/null @@ -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, - ) diff --git a/backend/apps/agents/core/seq_log.py b/backend/apps/agents/core/seq_log.py index fcd1ea39..092f45e2 100644 --- a/backend/apps/agents/core/seq_log.py +++ b/backend/apps/agents/core/seq_log.py @@ -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()) diff --git a/backend/apps/agents/core/ws_manager.py b/backend/apps/agents/core/ws_manager.py index 0a92e2dd..4abd8f85 100644 --- a/backend/apps/agents/core/ws_manager.py +++ b/backend/apps/agents/core/ws_manager.py @@ -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) diff --git a/backend/apps/agents/manager/prompt/attachments.py b/backend/apps/agents/manager/prompt/attachments.py index 01233a02..e70483e2 100644 --- a/backend/apps/agents/manager/prompt/attachments.py +++ b/backend/apps/agents/manager/prompt/attachments.py @@ -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 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"\n{chr(10).join(tree_lines)}\n" ) @@ -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) diff --git a/backend/apps/agents/manager/prompt/prompt_context.py b/backend/apps/agents/manager/prompt/prompt_context.py index f1f28474..ba243566 100644 --- a/backend/apps/agents/manager/prompt/prompt_context.py +++ b/backend/apps/agents/manager/prompt/prompt_context.py @@ -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 "" diff --git a/backend/apps/agents/manager/prompt/tool_catalog.py b/backend/apps/agents/manager/prompt/tool_catalog.py index 88a7286b..bd6336a4 100644 --- a/backend/apps/agents/manager/prompt/tool_catalog.py +++ b/backend/apps/agents/manager/prompt/tool_catalog.py @@ -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) diff --git a/backend/apps/agents/manager/session/cloud_sync.py b/backend/apps/agents/manager/session/cloud_sync.py index 09c4e5af..4f1b4981 100644 --- a/backend/apps/agents/manager/session/cloud_sync.py +++ b/backend/apps/agents/manager/session/cloud_sync.py @@ -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 diff --git a/backend/apps/agents/manager/session/history_compaction.py b/backend/apps/agents/manager/session/history_compaction.py index f644e2e6..dbda9958 100644 --- a/backend/apps/agents/manager/session/history_compaction.py +++ b/backend/apps/agents/manager/session/history_compaction.py @@ -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 "\n" + "\n".join(lines) + "\n" -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). diff --git a/backend/apps/agents/manager/session/session_store.py b/backend/apps/agents/manager/session/session_store.py index a1dac45e..cd4e12a9 100644 --- a/backend/apps/agents/manager/session/session_store.py +++ b/backend/apps/agents/manager/session/session_store.py @@ -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): diff --git a/backend/apps/agents/manager/session/workspace_git.py b/backend/apps/agents/manager/session/workspace_git.py index c9dd68cb..8c2e7023 100644 --- a/backend/apps/agents/manager/session/workspace_git.py +++ b/backend/apps/agents/manager/session/workspace_git.py @@ -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: diff --git a/backend/apps/agents/providers/openrouter.py b/backend/apps/agents/providers/openrouter.py index 9c231243..3888baf6 100644 --- a/backend/apps/agents/providers/openrouter.py +++ b/backend/apps/agents/providers/openrouter.py @@ -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 diff --git a/backend/apps/agents/providers/pricing.py b/backend/apps/agents/providers/pricing.py index 63035aa3..52957c74 100644 --- a/backend/apps/agents/providers/pricing.py +++ b/backend/apps/agents/providers/pricing.py @@ -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( diff --git a/backend/apps/agents/providers/registry.py b/backend/apps/agents/providers/registry.py index 5b2c77bf..6200a4fb 100644 --- a/backend/apps/agents/providers/registry.py +++ b/backend/apps/agents/providers/registry.py @@ -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//` 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:/`) 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-/` 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: diff --git a/backend/apps/auth/router.py b/backend/apps/auth/router.py index dbc892e7..6765607c 100644 --- a/backend/apps/auth/router.py +++ b/backend/apps/auth/router.py @@ -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) diff --git a/backend/apps/dashboards/dashboards.py b/backend/apps/dashboards/dashboards.py index 4bf9365e..a6453304 100644 --- a/backend/apps/dashboards/dashboards.py +++ b/backend/apps/dashboards/dashboards.py @@ -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(): diff --git a/backend/apps/nine_router/__init__.py b/backend/apps/nine_router/__init__.py deleted file mode 100644 index ce818d3d..00000000 --- a/backend/apps/nine_router/__init__.py +++ /dev/null @@ -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", -] diff --git a/backend/apps/service/service.py b/backend/apps/service/service.py index bec5bef1..10779541 100644 --- a/backend/apps/service/service.py +++ b/backend/apps/service/service.py @@ -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: diff --git a/backend/apps/settings/settings.py b/backend/apps/settings/settings.py index 51f52cb7..8b721d0c 100644 --- a/backend/apps/settings/settings.py +++ b/backend/apps/settings/settings.py @@ -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: diff --git a/backend/apps/settings/store.py b/backend/apps/settings/store.py index f7decb62..b1ddf5ee 100644 --- a/backend/apps/settings/store.py +++ b/backend/apps/settings/store.py @@ -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()) \ No newline at end of file diff --git a/backend/apps/subscription/free_trial.py b/backend/apps/subscription/free_trial.py index cd897293..b6d3e1e3 100644 --- a/backend/apps/subscription/free_trial.py +++ b/backend/apps/subscription/free_trial.py @@ -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) diff --git a/backend/apps/subscription/router.py b/backend/apps/subscription/router.py index 00b23a05..5034a76b 100644 --- a/backend/apps/subscription/router.py +++ b/backend/apps/subscription/router.py @@ -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) diff --git a/backend/apps/web/web.py b/backend/apps/web/web.py index 60336e65..4f4b99bb 100644 --- a/backend/apps/web/web.py +++ b/backend/apps/web/web.py @@ -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 diff --git a/backend/config/json_store.py b/backend/config/json_store.py index 46984696..2231c391 100644 --- a/backend/config/json_store.py +++ b/backend/config/json_store.py @@ -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 diff --git a/backend/main.py b/backend/main.py index fe186778..dc496659 100644 --- a/backend/main.py +++ b/backend/main.py @@ -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('

Session expired

Please try connecting again.

') - 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: diff --git a/backend/tests/test_disconnect_resilience.py b/backend/tests/test_disconnect_resilience.py index d39fb54c..e00db6e8 100644 --- a/backend/tests/test_disconnect_resilience.py +++ b/backend/tests/test_disconnect_resilience.py @@ -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() # ---------------------------------------------------------------------------