fix: tighten managed local package install safety and restore session compat

Agent-Logs-Url: https://github.com/MervinPraison/PraisonAI/sessions/be07f0d7-3e03-4987-808d-5f87bfcae00b

Co-authored-by: MervinPraison <454862+MervinPraison@users.noreply.github.com>

Author: Copilot

src/praisonai/praisonai/integrations/managed_local.py

@@ -476,42 +476,42 @@ def _install_packages(self) -> None:
         if not pip_pkgs:
             return
 
+        compute_install_succeeded = False
+
         # If compute provider is available, use it
         if self._compute is not None:
             # Use sandbox_type from config
             sandbox_type = self._cfg.get("sandbox_type", "subprocess")
             logger.info("[local_managed] installing packages via compute provider (%s): %s", sandbox_type, pip_pkgs)
             try:
-                import asyncio
-                # Run async operation in sync context
-                loop = asyncio.get_event_loop()
-                if loop.is_running():
-                    # We're in async context, need to run in executor
+                try:
+                    asyncio.get_running_loop()
+                    # We're in async context, run in a separate thread loop.
                     import threading
-                    result = [None]
                     exception = [None]
-                    
+
                     def run_install():
                         try:
-                            new_loop = asyncio.new_event_loop()
-                            asyncio.set_event_loop(new_loop)
-                            result[0] = new_loop.run_until_complete(self._install_via_compute(pip_pkgs))
+                            asyncio.run(self._install_via_compute(pip_pkgs))
                         except Exception as e:
                             exception[0] = e
-                        finally:
-                            new_loop.close()
-                    
+
                     thread = threading.Thread(target=run_install)
                     thread.start()
                     thread.join()
-                    
+
                     if exception[0]:
                         raise exception[0]
-                else:
-                    loop.run_until_complete(self._install_via_compute(pip_pkgs))
+                except RuntimeError:
+                    # No running loop in this thread.
+                    asyncio.run(self._install_via_compute(pip_pkgs))
+                compute_install_succeeded = True
             except Exception as e:
                 logger.warning("[local_managed] compute package install failed: %s", e)
                 # Fall through to host installation if allowed
+
+        if compute_install_succeeded:
+            return
 
         # Host installation - only if explicitly allowed
         if not self._cfg.get("host_packages_ok", False):
@@ -544,7 +544,8 @@ async def _install_via_compute(self, pip_pkgs: List[str]) -> None:
             await self.provision_compute(config=config)
         else:
             # Install on existing instance
-            cmd = f"pip install -q {' '.join(pip_pkgs)}"
+            import shlex
+            cmd = "pip install -q " + " ".join(shlex.quote(pkg) for pkg in pip_pkgs)
             await self.execute_in_compute(cmd, timeout=120)
 
     def _ensure_agent(self) -> Any:
@@ -773,29 +774,32 @@ def interrupt(self) -> None:
     # ------------------------------------------------------------------
     def retrieve_session(self) -> Dict[str, Any]:
         """Retrieve current session metadata using unified SessionInfo schema."""
+        if not self._session_id:
+            return {}
+
         self._sync_usage()
 
         # Use unified SessionInfo schema for consistency with Anthropic backend
         try:
             from praisonaiagents.managed import SessionInfo
             session_info = SessionInfo(
-                id=self._session_id or "",
-                status="idle" if self._session_id else "none",
+                id=self._session_id,
+                status="idle",
                 usage={
                     "input_tokens": self.total_input_tokens,
                     "output_tokens": self.total_output_tokens,
-                } if self._session_id else None
+                }
             )
             return session_info.to_dict()
         except ImportError:
             # Fallback to old format if SessionInfo not available
             return {
-                "id": self._session_id or "",
-                "status": "idle" if self._session_id else "none",
+                "id": self._session_id,
+                "status": "idle",
                 "usage": {
                     "input_tokens": self.total_input_tokens,
                     "output_tokens": self.total_output_tokens,
-                } if self._session_id else None,
+                },
             }
 
     def list_sessions(self, **kwargs) -> List[Dict[str, Any]]:

src/praisonai/tests/unit/integrations/test_managed_agents.py

@@ -7,7 +7,7 @@
 """
 
 import pytest
-from unittest.mock import Mock, patch, MagicMock
+from unittest.mock import Mock, patch, MagicMock, AsyncMock
 
 
 def test_managed_config_dataclass():
@@ -298,6 +298,48 @@ def test_backward_compatible_aliases():
     assert ManagedBackendConfig == ManagedConfig
 
 
+def test_local_retrieve_session_no_session_returns_empty_dict():
+    """Local backend should preserve empty-session behavior."""
+    from praisonai.integrations.managed_local import LocalManagedAgent
+
+    managed = LocalManagedAgent()
+    assert managed.retrieve_session() == {}
+
+
+@patch("praisonai.integrations.managed_local.subprocess.run")
+def test_local_install_packages_prefers_compute_and_skips_host(mock_subprocess_run):
+    """Successful compute install must not fall through to host install."""
+    from praisonai.integrations.managed_local import LocalManagedAgent, LocalManagedConfig
+
+    managed = LocalManagedAgent(
+        config=LocalManagedConfig(packages={"pip": ["requests"]})
+    )
+    managed._compute = object()
+    managed._install_via_compute = AsyncMock(return_value=None)
+
+    managed._install_packages()
+
+    managed._install_via_compute.assert_awaited_once_with(["requests"])
+    mock_subprocess_run.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_local_install_via_compute_quotes_package_names():
+    """Package names passed through shell command should be shell-escaped."""
+    from praisonai.integrations.managed_local import LocalManagedAgent
+
+    managed = LocalManagedAgent()
+    managed._compute_instance_id = "inst_123"
+    managed.execute_in_compute = AsyncMock(return_value={"stdout": "", "stderr": "", "exit_code": 0})
+
+    await managed._install_via_compute(["requests", "bad;echo pwned"])
+
+    managed.execute_in_compute.assert_awaited_once()
+    command = managed.execute_in_compute.await_args.args[0]
+    assert "pip install -q " in command
+    assert "'bad;echo pwned'" in command
+
+
 @patch('praisonai.integrations.managed_agents.logger')
 def test_logging_integration(mock_logger):
     """Test that managed agents include proper logging."""
@@ -308,4 +350,4 @@ def test_logging_integration(mock_logger):
     managed.reset_all()
 
     # Verify logging is available (don't assert specific calls since they may not happen in unit tests)
-    assert mock_logger is not None
+    assert mock_logger is not None