diff --git a/ai-hub/app/api/routes/nodes.py b/ai-hub/app/api/routes/nodes.py index f2bc95f..6aba850 100644 --- a/ai-hub/app/api/routes/nodes.py +++ b/ai-hub/app/api/routes/nodes.py @@ -646,7 +646,7 @@ # ================================================================== @router.get("/{node_id}/fs/ls", response_model=schemas.DirectoryListing, summary="List Directory Content") - def fs_ls( + async def fs_ls( node_id: str, path: str = ".", session_id: str = "__fs_explorer__", @@ -659,14 +659,14 @@ """ _require_node_access(user_id, node_id, db) try: - # Defensive check for orchestrator service injection try: orchestrator = services.orchestrator except AttributeError: logger.error("[FS] Orchestrator service not found in ServiceContainer.") raise HTTPException(status_code=500, detail="Agent Orchestrator service is starting or unavailable.") - res = orchestrator.assistant.ls(node_id, path, session_id=session_id) + loop = asyncio.get_event_loop() + res = await loop.run_in_executor(None, lambda: orchestrator.assistant.ls(node_id, path, session_id=session_id)) if not res: logger.error(f"[FS] Received empty response from node {node_id} for path {path}") @@ -679,7 +679,6 @@ files = res.get("files", []) - # M6: Check sync status ONLY for real user sessions, not for the node-wide navigator if session_id != "__fs_explorer__": workspace_mirror = orchestrator.mirror.get_workspace_path(session_id) for f in files: @@ -694,26 +693,25 @@ raise HTTPException(status_code=500, detail=f"Internal Server Error: {str(e)}") @router.get("/{node_id}/fs/cat", summary="Read File Content") - def fs_cat( + async def fs_cat( node_id: str, path: str, session_id: str = "__fs_explorer__", user_id: str = Header(..., alias="X-User-ID"), db: Session = Depends(get_db) ): - """ - Read the content of a file on a remote node. - """ + """Read the content of a file on a remote node.""" _require_node_access(user_id, node_id, db) try: orchestrator = services.orchestrator - res = orchestrator.assistant.cat(node_id, path, session_id=session_id) + loop = asyncio.get_event_loop() + res = await loop.run_in_executor(None, lambda: orchestrator.assistant.cat(node_id, path, session_id=session_id)) if not res: raise HTTPException(status_code=500, detail="Node returned an empty response.") if isinstance(res, dict) and "error" in res: status_code = 404 if "not found" in res["error"].lower() else 500 raise HTTPException(status_code=status_code, detail=res["error"]) - return res # Expecting {"content": "..."} + return res except HTTPException: raise except AttributeError: @@ -723,30 +721,24 @@ raise HTTPException(status_code=500, detail=str(e)) @router.post("/{node_id}/fs/touch", summary="Create File or Directory") - def fs_touch( + async def fs_touch( node_id: str, req: schemas.FileWriteRequest, user_id: str = Header(..., alias="X-User-ID"), db: Session = Depends(get_db) ): - """ - Create a new file or directory on the node. - """ + """Create a new file or directory on the node.""" _require_node_access(user_id, node_id, db) try: orchestrator = services.orchestrator - res = orchestrator.assistant.write( - node_id, - req.path, - req.content.encode('utf-8') if isinstance(req.content, str) else req.content, - req.is_dir, - session_id=req.session_id - ) + loop = asyncio.get_event_loop() + content = req.content.encode('utf-8') if isinstance(req.content, str) else req.content + res = await loop.run_in_executor(None, lambda: orchestrator.assistant.write(node_id, req.path, content, req.is_dir, session_id=req.session_id)) if not res: raise HTTPException(status_code=500, detail="Node returned an empty response.") if isinstance(res, dict) and "error" in res: raise HTTPException(status_code=500, detail=res["error"]) - return res # Expecting {"success": bool, "message": str} + return res except HTTPException: raise except AttributeError: @@ -831,19 +823,18 @@ raise HTTPException(status_code=500, detail=str(e)) @router.post("/{node_id}/fs/rm", summary="Delete File/Directory") - def fs_rm( + async def fs_rm( node_id: str, req: schemas.FileDeleteRequest, user_id: str = Header(..., alias="X-User-ID"), db: Session = Depends(get_db) ): - """ - Delete a file or directory from a remote node. - """ + """Delete a file or directory from a remote node.""" _require_node_access(user_id, node_id, db) try: orchestrator = services.orchestrator - res = orchestrator.assistant.rm(node_id, req.path, session_id=req.session_id) + loop = asyncio.get_event_loop() + res = await loop.run_in_executor(None, lambda: orchestrator.assistant.rm(node_id, req.path, session_id=req.session_id)) if not res: raise HTTPException(status_code=500, detail="Node returned an empty response.") return res @@ -854,7 +845,7 @@ raise HTTPException(status_code=500, detail=str(e)) @router.post("/{node_id}/fs/move", summary="Move/Rename File or Directory") - def fs_move( + async def fs_move( node_id: str, req: schemas.FileMoveRequest, user_id: str = Header(..., alias="X-User-ID"), @@ -864,13 +855,14 @@ _require_node_access(user_id, node_id, db) try: orchestrator = services.orchestrator - return orchestrator.assistant.move(req.session_id, req.old_path, req.new_path) + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, lambda: orchestrator.assistant.move(req.session_id, req.old_path, req.new_path)) except Exception as e: logger.error(f"[FS] Move error: {e}") raise HTTPException(status_code=500, detail=str(e)) @router.post("/{node_id}/fs/copy", summary="Copy File or Directory") - def fs_copy( + async def fs_copy( node_id: str, req: schemas.FileCopyRequest, user_id: str = Header(..., alias="X-User-ID"), @@ -880,7 +872,8 @@ _require_node_access(user_id, node_id, db) try: orchestrator = services.orchestrator - return orchestrator.assistant.copy(req.session_id, req.old_path, req.new_path) + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, lambda: orchestrator.assistant.copy(req.session_id, req.old_path, req.new_path)) except Exception as e: logger.error(f"[FS] Copy error: {e}") raise HTTPException(status_code=500, detail=str(e))