From 9899d37c640f0100ddd133a906062d9bb4ce3922 Mon Sep 17 00:00:00 2001 From: haikdc Date: Sun, 5 Apr 2026 09:11:44 -0700 Subject: [PATCH] [Haik]: removed old agentsd.py websocket wrapper bc it seemed like bs (may have to re add in some bits later). Now gonna add in some singeltons where applicable --- backend/OLDmain.py | 93 +++++++++++++++++++ backend/apps/agents/utils/comms/ws.py | 26 +----- backend/main.py | 67 +++---------- .../app/pages/AgentChat/hooks/useAgentChat.ts | 6 -- .../src/app/pages/Views/hooks/useAutoRun.ts | 7 -- frontend/src/shared/ws/WebSocketManager.ts | 44 +-------- 6 files changed, 112 insertions(+), 131 deletions(-) create mode 100644 backend/OLDmain.py diff --git a/backend/OLDmain.py b/backend/OLDmain.py new file mode 100644 index 00000000..cc89d9cc --- /dev/null +++ b/backend/OLDmain.py @@ -0,0 +1,93 @@ +import logging +import os + +from backend.ports import get_backend_port + +logger = logging.getLogger(__name__) + +from backend.config.Apps import MainApp +from backend.apps.health.health import health +from backend.apps.agents.agents import agents +from backend.apps.agents.manager.ws_manager import ws_manager +from backend.apps.agents.ws_routes import handle_session_message, handle_dashboard_message +from backend.apps.skills.skills import skills +from backend.apps.tools_lib.tools_lib import tools_lib +from backend.apps.modes.modes import modes +from backend.apps.settings.settings import settings +from backend.apps.mcp_registry.mcp_registry import mcp_registry +from backend.apps.skill_registry.skill_registry import skill_registry +from backend.apps.outputs.outputs import outputs +from backend.apps.dashboards.dashboards import dashboards +from backend.apps.nine_router import nine_router +from backend.apps.analytics.analytics import analytics +from backend.apps.subscriptions.subscriptions import subscriptions +from fastapi.middleware.cors import CORSMiddleware +from fastapi import WebSocket, WebSocketDisconnect +import json + +main_app = MainApp([ + health, agents, skills, tools_lib, modes, settings, + mcp_registry, skill_registry, outputs, dashboards, nine_router, analytics, + subscriptions, +]) +app = main_app.app + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +@app.websocket("/ws/agents/{session_id}") +async def websocket_session(websocket: WebSocket, session_id: str): + await ws_manager.connect_session(session_id, websocket) + try: + while True: + data = await websocket.receive_text() + msg = json.loads(data) + await handle_session_message(session_id, msg.get("event"), msg.get("data", {})) + except WebSocketDisconnect: + ws_manager.disconnect_session(session_id, websocket) + + +@app.websocket("/ws/dashboard") +async def websocket_dashboard(websocket: WebSocket): + await ws_manager.connect_global(websocket) + try: + while True: + data = await websocket.receive_text() + msg = json.loads(data) + await handle_dashboard_message(msg.get("event"), msg.get("data", {})) + except WebSocketDisconnect: + ws_manager.disconnect_global(websocket) + + +if __name__ == "__main__": + import argparse + import uvicorn + + parser = argparse.ArgumentParser(description="OpenSwarm backend server") + parser.add_argument("--port", type=int, default=get_backend_port()) + parser.add_argument("--host", default=os.environ.get("OPENSWARM_HOST", "127.0.0.1")) + parser.add_argument("--reload", action="store_true", default=False) + args = parser.parse_args() + + os.environ["OPENSWARM_PORT"] = str(args.port) + + import uvicorn.config + + class _ReadyServer(uvicorn.Server): + """Subclass that prints a machine-readable READY line on startup.""" + async def startup(self, sockets=None): + await super().startup(sockets) + print(f"READY:PORT={args.port}", flush=True) + + if args.reload: + uvicorn.run("backend.main:app", host=args.host, port=args.port, reload=True) + else: + config = uvicorn.Config("backend.main:app", host=args.host, port=args.port) + server = _ReadyServer(config) + import asyncio + asyncio.run(server.serve()) diff --git a/backend/apps/agents/utils/comms/ws.py b/backend/apps/agents/utils/comms/ws.py index bc5249dd..4902b2a9 100644 --- a/backend/apps/agents/utils/comms/ws.py +++ b/backend/apps/agents/utils/comms/ws.py @@ -1,33 +1,18 @@ import json from fastapi import WebSocket -from typing import Dict, List +from typing import List from typeguard import typechecked -P_SESSION_CONNECTIONS: Dict[str, List[WebSocket]] = {} P_GLOBAL_CONNECTIONS: List[WebSocket] = [] -@typechecked -async def connect_session(session_id: str, ws: WebSocket) -> None: - await ws.accept() - P_SESSION_CONNECTIONS.setdefault(session_id, []).append(ws) - @typechecked async def connect_global(ws: WebSocket) -> None: await ws.accept() P_GLOBAL_CONNECTIONS.append(ws) -@typechecked -def disconnect_session(session_id: str, ws: WebSocket) -> None: - conns = P_SESSION_CONNECTIONS.get(session_id) - if not conns: - return - conns[:] = [c for c in conns if c is not ws] - if not conns: - del P_SESSION_CONNECTIONS[session_id] - - +# USED-IN: make_session_emitter, send_browser_command @typechecked def has_global_connections() -> bool: return len(P_GLOBAL_CONNECTIONS) > 0 @@ -38,14 +23,10 @@ def disconnect_global(ws: WebSocket) -> None: P_GLOBAL_CONNECTIONS[:] = [c for c in P_GLOBAL_CONNECTIONS if c is not ws] +# USED-IN: make_session_emitter @typechecked async def send_to_session(session_id: str, event: str, data: dict) -> None: payload = json.dumps({"event": event, "session_id": session_id, "data": data}) - for ws in P_SESSION_CONNECTIONS.get(session_id, []): - try: - await ws.send_text(payload) - except Exception: - pass for ws in P_GLOBAL_CONNECTIONS: try: await ws.send_text(payload) @@ -53,6 +34,7 @@ async def send_to_session(session_id: str, event: str, data: dict) -> None: pass +# USED-IN: send_browser_command @typechecked async def broadcast_global(event: str, data: dict) -> None: payload = json.dumps({"event": event, "data": data}) diff --git a/backend/main.py b/backend/main.py index cc89d9cc..81725e7e 100644 --- a/backend/main.py +++ b/backend/main.py @@ -1,34 +1,20 @@ -import logging +from typing import List import os - +import argparse +import socket +from typing import Optional +import uvicorn +from typeguard import typechecked from backend.ports import get_backend_port - -logger = logging.getLogger(__name__) - from backend.config.Apps import MainApp from backend.apps.health.health import health from backend.apps.agents.agents import agents -from backend.apps.agents.manager.ws_manager import ws_manager -from backend.apps.agents.ws_routes import handle_session_message, handle_dashboard_message -from backend.apps.skills.skills import skills -from backend.apps.tools_lib.tools_lib import tools_lib -from backend.apps.modes.modes import modes from backend.apps.settings.settings import settings -from backend.apps.mcp_registry.mcp_registry import mcp_registry -from backend.apps.skill_registry.skill_registry import skill_registry -from backend.apps.outputs.outputs import outputs from backend.apps.dashboards.dashboards import dashboards -from backend.apps.nine_router import nine_router -from backend.apps.analytics.analytics import analytics -from backend.apps.subscriptions.subscriptions import subscriptions from fastapi.middleware.cors import CORSMiddleware -from fastapi import WebSocket, WebSocketDisconnect -import json main_app = MainApp([ - health, agents, skills, tools_lib, modes, settings, - mcp_registry, skill_registry, outputs, dashboards, nine_router, analytics, - subscriptions, + health, agents, settings, dashboards ]) app = main_app.app @@ -40,47 +26,20 @@ app.add_middleware( allow_headers=["*"], ) -@app.websocket("/ws/agents/{session_id}") -async def websocket_session(websocket: WebSocket, session_id: str): - await ws_manager.connect_session(session_id, websocket) - try: - while True: - data = await websocket.receive_text() - msg = json.loads(data) - await handle_session_message(session_id, msg.get("event"), msg.get("data", {})) - except WebSocketDisconnect: - ws_manager.disconnect_session(session_id, websocket) - - -@app.websocket("/ws/dashboard") -async def websocket_dashboard(websocket: WebSocket): - await ws_manager.connect_global(websocket) - try: - while True: - data = await websocket.receive_text() - msg = json.loads(data) - await handle_dashboard_message(msg.get("event"), msg.get("data", {})) - except WebSocketDisconnect: - ws_manager.disconnect_global(websocket) - - if __name__ == "__main__": - import argparse - import uvicorn - parser = argparse.ArgumentParser(description="OpenSwarm backend server") + parser: argparse.ArgumentParser = argparse.ArgumentParser(description="OpenSwarm backend server") parser.add_argument("--port", type=int, default=get_backend_port()) parser.add_argument("--host", default=os.environ.get("OPENSWARM_HOST", "127.0.0.1")) parser.add_argument("--reload", action="store_true", default=False) - args = parser.parse_args() + args: argparse.Namespace = parser.parse_args() os.environ["OPENSWARM_PORT"] = str(args.port) - import uvicorn.config - - class _ReadyServer(uvicorn.Server): + class P_ReadyServer(uvicorn.Server): """Subclass that prints a machine-readable READY line on startup.""" - async def startup(self, sockets=None): + @typechecked + async def startup(self, sockets: Optional[List[socket.socket]] = None) -> None: await super().startup(sockets) print(f"READY:PORT={args.port}", flush=True) @@ -88,6 +47,6 @@ if __name__ == "__main__": uvicorn.run("backend.main:app", host=args.host, port=args.port, reload=True) else: config = uvicorn.Config("backend.main:app", host=args.host, port=args.port) - server = _ReadyServer(config) + server = P_ReadyServer(config) import asyncio asyncio.run(server.serve()) diff --git a/frontend/src/app/pages/AgentChat/hooks/useAgentChat.ts b/frontend/src/app/pages/AgentChat/hooks/useAgentChat.ts index 021109a3..870b418f 100644 --- a/frontend/src/app/pages/AgentChat/hooks/useAgentChat.ts +++ b/frontend/src/app/pages/AgentChat/hooks/useAgentChat.ts @@ -13,7 +13,6 @@ import { fetchSession, } from '@/shared/state/agentsSlice'; import { fetchModes } from '@/shared/state/modesSlice'; -import { createSessionWs } from '@/shared/ws/WebSocketManager'; import { setGlowingBrowserCards, fadeGlowingBrowserCards, clearGlowingBrowserCards } from '@/shared/state/dashboardLayoutSlice'; export interface QueuedMessage { @@ -39,7 +38,6 @@ export function useAgentChat({ sessionId: sessionIdProp }: UseAgentChatParams) { const [awaitingResponse, setAwaitingResponse] = useState(false); const [mode, setMode] = useState('agent'); const [model, setModel] = useState('sonnet'); - const wsRef = useRef | null>(null); const messageQueueRef = useRef([]); const [queueLength, setQueueLength] = useState(0); const [editingMessageId, setEditingMessageId] = useState(null); @@ -48,11 +46,7 @@ export function useAgentChat({ sessionId: sessionIdProp }: UseAgentChatParams) { useEffect(() => { if (!id || isDraft) return; - const ws = createSessionWs(id); - ws.connect(); - wsRef.current = ws; dispatch(fetchSession(id)); - return () => { ws.disconnect(); wsRef.current = null; }; }, [id, isDraft, dispatch]); useEffect(() => { if (session) setMode(session.mode); }, [session?.mode]); diff --git a/frontend/src/app/pages/Views/hooks/useAutoRun.ts b/frontend/src/app/pages/Views/hooks/useAutoRun.ts index d66fc298..2939a3a7 100644 --- a/frontend/src/app/pages/Views/hooks/useAutoRun.ts +++ b/frontend/src/app/pages/Views/hooks/useAutoRun.ts @@ -5,7 +5,6 @@ import { executeOutput, autoRunOutput, autoRunAgentOutput, cleanupAutoRunAgent, AutoRunConfig, } from '@/shared/state/outputsSlice'; -import { createSessionWs } from '@/shared/ws/WebSocketManager'; import { ChatInputHandle } from '../../AgentChat/ChatInput'; import { getDefault } from '../InputSchemaForm'; import type { ConsoleEntry } from '../ConsolePanel'; @@ -43,7 +42,6 @@ export function useAutoRun( const autoRunInitialized = useRef(false); const [autoRunSessionId, setAutoRunSessionId] = useState(null); - const autoRunWsRef = useRef | null>(null); const autoRunLogEndRef = useRef(null); const autoRunSession = useAppSelector((state) => @@ -108,9 +106,6 @@ export function useAutoRun( context_paths: config.contextPaths.map((cp) => ({ path: cp.path, type: cp.type })), })).unwrap(); setAutoRunSessionId(res.session_id); - const ws = createSessionWs(res.session_id); - ws.connect(); - autoRunWsRef.current = ws; } catch { setAutoRunning(false); } } else { try { @@ -173,7 +168,6 @@ export function useAutoRun( } } setAutoRunning(false); - if (autoRunWsRef.current) { autoRunWsRef.current.disconnect(); autoRunWsRef.current = null; } cleanupAutoRunAgent(autoRunSessionId).catch(() => {}); setTimeout(() => setAutoRunSessionId(null), 300); }, [autoRunSessionId, autoRunSessionStatus]); @@ -184,7 +178,6 @@ export function useAutoRun( useEffect(() => { return () => { - if (autoRunWsRef.current) { autoRunWsRef.current.disconnect(); autoRunWsRef.current = null; } if (autoRunSessionId) cleanupAutoRunAgent(autoRunSessionId).catch(() => {}); }; }, [autoRunSessionId]); diff --git a/frontend/src/shared/ws/WebSocketManager.ts b/frontend/src/shared/ws/WebSocketManager.ts index 781ca1ed..1422bafc 100644 --- a/frontend/src/shared/ws/WebSocketManager.ts +++ b/frontend/src/shared/ws/WebSocketManager.ts @@ -2,14 +2,9 @@ import { store } from '../state/store'; import { streamDelta } from '../state/agentsSlice'; import { type WSEvent, dispatchWsEvent } from './wsEventHandlers'; -interface WSManagerOptions { - skipStreamEvents?: boolean; -} - class WebSocketManager { private ws: WebSocket | null = null; private url: string; - private skipStreamEvents: boolean; private reconnectTimer: ReturnType | null = null; private reconnectDelay = 1000; private maxReconnectDelay = 30000; @@ -17,9 +12,8 @@ class WebSocketManager { private deltaBuffer: Map = new Map(); private flushScheduled = false; - constructor(url: string, options?: WSManagerOptions) { + constructor(url: string) { this.url = url; - this.skipStreamEvents = options?.skipStreamEvents ?? false; } private bufferDelta(sessionId: string, messageId: string, delta: string) { @@ -91,12 +85,6 @@ class WebSocketManager { private handleMessage(msg: WSEvent) { const { event, session_id, data } = msg; - if (this.skipStreamEvents) { - if (event === 'agent:stream_start' || event === 'agent:stream_delta' || event === 'agent:stream_end') { - return; - } - } - dispatchWsEvent(msg, { bufferDelta: (sid, mid, d) => this.bufferDelta(sid, mid, d), flushDeltas: () => this.flushDeltas(), @@ -114,30 +102,6 @@ class WebSocketManager { this.ws.send(JSON.stringify({ event, data })); } - sendMessage( - sessionId: string, - prompt: string, - opts?: { mode?: string; model?: string; provider?: string; images?: Array<{ data: string; media_type: string }> }, - ) { - this.send('agent:send_message', { - session_id: sessionId, - prompt, - ...opts, - }); - } - - sendApproval(requestId: string, behavior: 'allow' | 'deny', message?: string) { - this.send('agent:approval_response', { - request_id: requestId, - behavior, - message, - }); - } - - stopAgent(sessionId: string) { - this.send('agent:stop', { session_id: sessionId }); - } - on(event: string, handler: (data: any) => void) { if (!this.listeners.has(event)) { this.listeners.set(event, new Set()); @@ -153,8 +117,4 @@ class WebSocketManager { import { WS_BASE } from '@/shared/config'; -export const dashboardWs = new WebSocketManager(`${WS_BASE}/ws/dashboard`, { skipStreamEvents: true }); - -export function createSessionWs(sessionId: string): WebSocketManager { - return new WebSocketManager(`${WS_BASE}/ws/agents/${sessionId}`); -} \ No newline at end of file +export const dashboardWs = new WebSocketManager(`${WS_BASE}/ws/dashboard`); \ No newline at end of file