diff --git a/src/praisonai-agents/praisonaiagents/managed/__init__.py b/src/praisonai-agents/praisonaiagents/managed/__init__.py
index 3d5142e8c..6acafe065 100644
--- a/src/praisonai-agents/praisonaiagents/managed/__init__.py
+++ b/src/praisonai-agents/praisonaiagents/managed/__init__.py
@@ -25,8 +25,17 @@
     ComputeConfig,
     InstanceInfo,
     InstanceStatus,
+    SessionInfo,
 )
 
+# Lazy re-export of ManagedBackendProtocol from agent.protocols
+# Following AGENTS.md protocol-driven design principles
+def __getattr__(name: str):
+    if name == "ManagedBackendProtocol":
+        from ..agent.protocols import ManagedBackendProtocol
+        return ManagedBackendProtocol
+    raise AttributeError(f"module '{__name__}' has no attribute '{name}'")
+
 __all__ = [
     "ManagedEvent",
     "AgentMessageEvent",
@@ -41,4 +50,6 @@
     "ComputeConfig",
     "InstanceInfo",
     "InstanceStatus",
+    "SessionInfo",
+    "ManagedBackendProtocol",  # Lazy re-export from agent.protocols
 ]
diff --git a/src/praisonai-agents/praisonaiagents/managed/protocols.py b/src/praisonai-agents/praisonaiagents/managed/protocols.py
index 7bf2e5bdd..e1997f4d6 100644
--- a/src/praisonai-agents/praisonaiagents/managed/protocols.py
+++ b/src/praisonai-agents/praisonaiagents/managed/protocols.py
@@ -86,6 +86,36 @@ class InstanceInfo:
     metadata: Dict[str, Any] = field(default_factory=dict)
 
 
+@dataclass
+class SessionInfo:
+    """Unified session information schema for all managed backends.
+    
+    Provides a consistent interface for session metadata across
+    Anthropic Managed Agents and Local Managed Agents.
+    """
+    id: str
+    status: Optional[str] = None
+    usage: Optional[Dict[str, int]] = None
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "SessionInfo":
+        """Create SessionInfo from backend-specific dict."""
+        return cls(
+            id=data["id"],
+            status=data.get("status"),
+            usage=data.get("usage")
+        )
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dict format expected by ManagedBackendProtocol."""
+        result = {"id": self.id}
+        if self.status is not None:
+            result["status"] = self.status
+        if self.usage is not None:
+            result["usage"] = self.usage
+        return result
+
+
 @runtime_checkable
 class ComputeProviderProtocol(Protocol):
     """Protocol for compute backends that host managed agent sandboxes.
diff --git a/src/praisonai/praisonai/cli/commands/managed.py b/src/praisonai/praisonai/cli/commands/managed.py
index 3e564a3a2..41a4b2db8 100644
--- a/src/praisonai/praisonai/cli/commands/managed.py
+++ b/src/praisonai/praisonai/cli/commands/managed.py
@@ -211,6 +211,32 @@ def managed_multi(
         typer.echo(f"Output tokens: {managed.total_output_tokens}")
 
 
+@sessions_app.command("delete")
+def sessions_delete(
+    session_id: str = typer.Argument(..., help="Session ID to delete (sesn_01...)"),
+    force: bool = typer.Option(False, "--force", "-f", help="Skip confirmation prompt"),
+):
+    """Delete a managed session.
+
+    Example:
+        praisonai managed sessions delete sesn_01AbCdEf
+        praisonai managed sessions delete sesn_01AbCdEf --force
+    """
+    if not force:
+        confirm = typer.confirm(f"Are you sure you want to delete session {session_id}?")
+        if not confirm:
+            typer.echo("Cancelled.")
+            return
+
+    try:
+        client = _get_client()
+        client.beta.sessions.delete(session_id)
+        typer.echo(f"Deleted session: {session_id}")
+    except Exception as e:
+        typer.echo(f"Error deleting session: {e}", err=True)
+        raise typer.Exit(1)
+
+
 # ─────────────────────────────────────────────────────────────────────────────
 # sessions sub-commands
 # ─────────────────────────────────────────────────────────────────────────────
@@ -393,8 +419,38 @@ def agents_update(
     if not kwargs:
         typer.echo("Nothing to update. Pass --name, --system, or --model.")
         raise typer.Exit(0)
-    updated = client.beta.agents.update(agent_id, **kwargs)
-    typer.echo(f"Updated agent: {updated.id} (v{getattr(updated,'version','')})")
+    try:
+        updated = client.beta.agents.update(agent_id, **kwargs)
+        typer.echo(f"Updated agent: {updated.id} (v{getattr(updated,'version','')})")
+    except Exception as e:
+        typer.echo(f"Error updating agent {agent_id}: {e}", err=True)
+        raise typer.Exit(1)
+
+
+@agents_app.command("delete")
+def agents_delete(
+    agent_id: str = typer.Argument(..., help="Agent ID to delete (agent_01...)"),
+    force: bool = typer.Option(False, "--force", "-f", help="Skip confirmation prompt"),
+):
+    """Delete a managed agent.
+
+    Example:
+        praisonai managed agents delete agent_01AbCdEf
+        praisonai managed agents delete agent_01AbCdEf --force
+    """
+    if not force:
+        confirm = typer.confirm(f"Are you sure you want to delete agent {agent_id}?")
+        if not confirm:
+            typer.echo("Cancelled.")
+            return
+
+    try:
+        client = _get_client()
+        client.beta.agents.delete(agent_id)
+        typer.echo(f"Deleted agent: {agent_id}")
+    except Exception as e:
+        typer.echo(f"Error deleting agent: {e}", err=True)
+        raise typer.Exit(1)
 
 
 # ─────────────────────────────────────────────────────────────────────────────
@@ -441,6 +497,68 @@ def envs_get(
         typer.echo(f"Config: {cfg}")
 
 
+@envs_app.command("update")
+def envs_update(
+    env_id: str = typer.Argument(..., help="Environment ID to update (env_01...)"),
+    name: Optional[str] = typer.Option(None, "--name", "-n", help="New environment name"),
+    config: Optional[str] = typer.Option(None, "--config", "-c", help="New environment config (JSON)"),
+):
+    """Update a managed environment.
+
+    Example:
+        praisonai managed envs update env_01AbCdEf --name "Updated Env"
+        praisonai managed envs update env_01AbCdEf --config '{"packages": ["numpy"]}'
+    """
+    client = _get_client()
+    kwargs = {}
+    if name:
+        kwargs["name"] = name
+    if config:
+        import json
+        try:
+            kwargs["config"] = json.loads(config)
+        except json.JSONDecodeError as e:
+            typer.echo(f"Error parsing config JSON: {e}", err=True)
+            raise typer.Exit(1)
+    
+    if not kwargs:
+        typer.echo("No update fields provided. Use --name or --config.", err=True)
+        raise typer.Exit(1)
+    
+    try:
+        updated = client.beta.environments.update(env_id, **kwargs)
+        typer.echo(f"Updated environment: {updated.id}")
+    except Exception as e:
+        typer.echo(f"Error updating environment: {e}", err=True)
+        raise typer.Exit(1)
+
+
+@envs_app.command("delete")
+def envs_delete(
+    env_id: str = typer.Argument(..., help="Environment ID to delete (env_01...)"),
+    force: bool = typer.Option(False, "--force", "-f", help="Skip confirmation prompt"),
+):
+    """Delete a managed environment.
+
+    Example:
+        praisonai managed envs delete env_01AbCdEf
+        praisonai managed envs delete env_01AbCdEf --force
+    """
+    if not force:
+        confirm = typer.confirm(f"Are you sure you want to delete environment {env_id}?")
+        if not confirm:
+            typer.echo("Cancelled.")
+            return
+
+    try:
+        client = _get_client()
+        client.beta.environments.delete(env_id)
+        typer.echo(f"Deleted environment: {env_id}")
+    except Exception as e:
+        typer.echo(f"Error deleting environment: {e}", err=True)
+        raise typer.Exit(1)
+
+
 # ─────────────────────────────────────────────────────────────────────────────
 # ids sub-commands  (save / restore / show — no Anthropic IDs are user-defined)
 # ─────────────────────────────────────────────────────────────────────────────
diff --git a/src/praisonai/praisonai/integrations/managed_agents.py b/src/praisonai/praisonai/integrations/managed_agents.py
index 761384abc..7355e4ce2 100644
--- a/src/praisonai/praisonai/integrations/managed_agents.py
+++ b/src/praisonai/praisonai/integrations/managed_agents.py
@@ -565,22 +565,29 @@ def interrupt(self) -> None:
     # retrieve_session — ManagedBackendProtocol
     # ------------------------------------------------------------------
     def retrieve_session(self) -> Dict[str, Any]:
-        """Retrieve current session metadata and usage from the API."""
+        """Retrieve current session metadata and usage from the API using unified schema."""
         if not self._session_id:
             return {}
         client = self._get_client()
         sess = client.beta.sessions.retrieve(self._session_id)
-        result: Dict[str, Any] = {
-            "id": getattr(sess, "id", self._session_id),
-            "status": getattr(sess, "status", None),
-        }
+        
+        # Build usage dict if available
+        usage_dict = None
         usage = getattr(sess, "usage", None)
         if usage:
-            result["usage"] = {
+            usage_dict = {
                 "input_tokens": getattr(usage, "input_tokens", 0),
                 "output_tokens": getattr(usage, "output_tokens", 0),
             }
-        return result
+        
+        # Use unified SessionInfo schema for consistency with Local backend
+        from praisonaiagents.managed import SessionInfo
+        session_info = SessionInfo(
+            id=getattr(sess, "id", self._session_id),
+            status=getattr(sess, "status", None),
+            usage=usage_dict
+        )
+        return session_info.to_dict()
 
     # ------------------------------------------------------------------
     # list_sessions — ManagedBackendProtocol
diff --git a/src/praisonai/praisonai/integrations/managed_local.py b/src/praisonai/praisonai/integrations/managed_local.py
index dc27489fc..336c7cff5 100644
--- a/src/praisonai/praisonai/integrations/managed_local.py
+++ b/src/praisonai/praisonai/integrations/managed_local.py
@@ -38,6 +38,15 @@
     Optional,
 )
 
+
+class ManagedSandboxRequired(Exception):
+    """Raised when sandbox execution is required but not available.
+    
+    This prevents dangerous host-side operations like package installation
+    when compute providers are configured but not properly wired.
+    """
+    pass
+
 logger = logging.getLogger(__name__)
 
 _DEFAULT_SYSTEM = "You are a helpful coding assistant."
@@ -83,6 +92,9 @@ class LocalManagedConfig:
     env: Dict[str, str] = field(default_factory=dict)
     packages: Optional[Dict[str, List[str]]] = None
     networking: Dict[str, Any] = field(default_factory=lambda: {"type": "unrestricted"})
+    
+    # ── Safety fields ──
+    host_packages_ok: bool = False  # Allow host package installation (UNSAFE)
 
     # ── Session fields ──
     session_title: str = "PraisonAI local session"
@@ -451,21 +463,89 @@ def _restore_state(self) -> None:
                 self.provider = saved_cfg["provider"]
 
     def _install_packages(self) -> None:
-        """Install packages specified in config before agent starts."""
+        """Install packages specified in config before agent starts.
+        
+        Uses compute provider if available, otherwise installs on host only if
+        host_packages_ok=True to prevent accidental host pollution.
+        """
         packages = self._cfg.get("packages")
         if not packages:
             return
 
         pip_pkgs = packages.get("pip", []) if isinstance(packages, dict) else []
-        if pip_pkgs:
-            cmd = [sys.executable, "-m", "pip", "install", "-q"] + pip_pkgs
-            logger.info("[local_managed] installing pip packages: %s", pip_pkgs)
+        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:
-                subprocess.run(cmd, check=True, capture_output=True, timeout=120)
-            except subprocess.CalledProcessError as e:
-                logger.warning("[local_managed] pip install failed: %s", e.stderr)
-            except subprocess.TimeoutExpired:
-                logger.warning("[local_managed] pip install timed out")
+                try:
+                    # Raises RuntimeError when no event loop is currently running.
+                    asyncio.get_running_loop()
+                    # We're in async context, run in a separate thread loop.
+                    import threading
+                    exception = [None]
+
+                    def run_install():
+                        try:
+                            asyncio.run(self._install_via_compute(pip_pkgs))
+                        except Exception as e:
+                            exception[0] = e
+
+                    thread = threading.Thread(target=run_install)
+                    thread.start()
+                    thread.join()
+
+                    if exception[0]:
+                        raise exception[0]
+                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):
+            raise ManagedSandboxRequired(
+                f"Package installation requires sandbox execution. "
+                f"Either configure a compute provider or set host_packages_ok=True. "
+                f"Packages: {pip_pkgs}"
+            )
+            
+        # Host installation (UNSAFE)
+        cmd = [sys.executable, "-m", "pip", "install", "-q"] + pip_pkgs
+        logger.warning("[local_managed] installing pip packages on HOST (UNSAFE): %s", pip_pkgs)
+        try:
+            subprocess.run(cmd, check=True, capture_output=True, timeout=120)
+        except subprocess.CalledProcessError as e:
+            logger.warning("[local_managed] pip install failed: %s", e.stderr)
+        except subprocess.TimeoutExpired:
+            logger.warning("[local_managed] pip install timed out")
+            
+    async def _install_via_compute(self, pip_pkgs: List[str]) -> None:
+        """Install packages via compute provider."""
+        if not self._compute_instance_id:
+            # Provision compute instance if needed
+            await self.provision_compute(
+                packages={"pip": pip_pkgs},
+                image="python:3.12-slim",
+                working_dir=self._cfg.get("working_dir", "/workspace")
+            )
+        else:
+            # Install on existing instance
+            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:
         """Create or return the inner PraisonAI Agent."""
@@ -497,7 +577,8 @@ def _ensure_agent(self) -> Any:
         self._inner_agent = Agent(**agent_kwargs)
         self.agent_id = self.agent_id or f"agent_{uuid.uuid4().hex[:12]}"
         self.environment_id = self.environment_id or f"env_{uuid.uuid4().hex[:12]}"
-        logger.info("[local_managed] agent created: %s model=%s", self.agent_id, model)
+        logger.info("[local_managed] agent created: %s model=%s sandbox=%s", 
+                   self.agent_id, model, self._cfg.get("sandbox_type", "subprocess"))
         return self._inner_agent
 
     def _ensure_session(self) -> str:
@@ -691,16 +772,34 @@ def interrupt(self) -> None:
     # retrieve_session / list_sessions — ManagedBackendProtocol
     # ------------------------------------------------------------------
     def retrieve_session(self) -> Dict[str, Any]:
-        """Retrieve current session metadata."""
+        """Retrieve current session metadata using unified SessionInfo schema."""
+        if not self._session_id:
+            return {}
+
         self._sync_usage()
-        return {
-            "id": self._session_id,
-            "status": "idle" if self._session_id else "none",
-            "usage": {
-                "input_tokens": self.total_input_tokens,
-                "output_tokens": self.total_output_tokens,
-            },
-        }
+        
+        # Use unified SessionInfo schema for consistency with Anthropic backend
+        try:
+            from praisonaiagents.managed import SessionInfo
+            session_info = SessionInfo(
+                id=self._session_id,
+                status="idle",
+                usage={
+                    "input_tokens": self.total_input_tokens,
+                    "output_tokens": self.total_output_tokens,
+                }
+            )
+            return session_info.to_dict()
+        except ImportError:
+            # Fallback to old format if SessionInfo not available
+            return {
+                "id": self._session_id,
+                "status": "idle",
+                "usage": {
+                    "input_tokens": self.total_input_tokens,
+                    "output_tokens": self.total_output_tokens,
+                },
+            }
 
     def list_sessions(self, **kwargs) -> List[Dict[str, Any]]:
         """List all sessions created in this backend instance."""
diff --git a/src/praisonai/tests/unit/integrations/test_managed_agents.py b/src/praisonai/tests/unit/integrations/test_managed_agents.py
index 4bbf6e395..c266ffbe9 100644
--- a/src/praisonai/tests/unit/integrations/test_managed_agents.py
+++ b/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,66 @@ 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
+@patch("praisonai.integrations.managed_local.subprocess.run")
+async def test_local_install_packages_prefers_compute_inside_running_loop(mock_subprocess_run):
+    """Compute install should also work when called inside an active event loop."""
+    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 +368,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
\ No newline at end of file
+    assert mock_logger is not None