feat(copilot): add local PC executor shim binding

Platform-side implementation for routing copilot execution to the
user's local machine via the autogpt-local-executor shim.

- ShimConnectionManager: in-memory WebSocket registry with wait_for()
- LocalPCShim: duck-type drop-in for E2B AsyncSandbox
  - .commands.run(), .files.read(), .files.write(), .pause(), .kill()
- WebSocket endpoint: /ws/local-executor/{session_id}?token=<access_token>
  - Validates token via introspect_token() before accepting connection
  - HELLO/HELLO_ACK handshake
- Wire into _setup_e2b(): LocalPC branch runs before E2B branch
- Pause guard: skip pause_sandbox_direct for LocalPCShim instances
- Config fields: use_local_pc_executor, allow_computer_use,
  local_pc_executor_ws_path

Requires: autogpt-local-executor shim running on user's machine

Author: ntindle

autogpt_platform/backend/backend/api/features/local_executor/__init__.py

@@ -0,0 +1,83 @@
+"""
+WebSocket endpoint for the autogpt-local-executor shim.
+
+The shim dials in with:
+    ws://<host>/ws/local-executor/<session_id>?token=<access_token>
+
+Auth: the token is validated via introspect_token() before the connection
+is accepted. On success the WebSocket is registered in ShimConnectionManager
+so LocalPCShim.for_session() can find it.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect
+
+from backend.copilot.tools.local_pc_shim import get_shim_manager
+from backend.data.auth.oauth import introspect_token
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+
+@router.websocket("/ws/local-executor/{session_id}")
+async def local_executor_ws(session_id: str, websocket: WebSocket) -> None:
+    token = websocket.query_params.get("token", "")
+    if not token:
+        await websocket.close(code=4401, reason="Missing token")
+        return
+
+    try:
+        token_info = await introspect_token(token, token_type_hint="access_token")
+        if not token_info or not token_info.get("active"):
+            await websocket.close(code=4401, reason="Invalid or expired token")
+            return
+    except Exception:
+        logger.exception("[LocalPC] Token introspection failed for session %s", session_id[:12])
+        await websocket.close(code=4500, reason="Auth error")
+        return
+
+    await websocket.accept()
+
+    # Handshake: expect HELLO, send HELLO_ACK
+    try:
+        import json, time, uuid
+        raw = await websocket.receive_text()
+        hello = json.loads(raw)
+        if hello.get("type") != "HELLO":
+            await websocket.close(code=4400, reason="Expected HELLO")
+            return
+
+        granted = hello.get("payload", {}).get("capabilities", [])
+        ack = {
+            "type": "HELLO_ACK",
+            "id": str(uuid.uuid4()),
+            "ts": time.time(),
+            "payload": {
+                "session_id": session_id,
+                "granted_capabilities": granted,
+                "server_version": "0.0.1",
+            },
+        }
+        await websocket.send_text(json.dumps(ack))
+    except Exception:
+        logger.exception("[LocalPC] Handshake failed for session %s", session_id[:12])
+        await websocket.close(code=4500, reason="Handshake error")
+        return
+
+    manager = get_shim_manager()
+    manager.register(session_id, websocket)
+    logger.info("[LocalPC] Shim connected for session %s", session_id[:12])
+
+    try:
+        # Keep connection alive; LocalPCShim._recv_loop drives the traffic
+        while True:
+            await websocket.receive_text()
+    except (WebSocketDisconnect, Exception):
+        pass
+    finally:
+        manager.unregister(session_id)
+        logger.info("[LocalPC] Shim disconnected for session %s", session_id[:12])

autogpt_platform/backend/backend/api/features/local_executor/routes.py

@@ -33,6 +33,7 @@
 import backend.api.features.oauth
 import backend.api.features.otto.routes
 import backend.api.features.platform_linking.routes
+import backend.api.features.local_executor.routes as local_executor_routes
 import backend.api.features.postmark.postmark
 import backend.api.features.push.routes as push_routes
 import backend.api.features.store.model
@@ -405,6 +406,10 @@ async def validation_error_handler(
     prefix="/api/platform-linking",
 )
 
+app.include_router(
+    local_executor_routes.router,
+    tags=["experimental", "local-executor"],
+)
 app.mount("/external-api", external_api)
 
 

autogpt_platform/backend/backend/api/rest_api.py

@@ -393,6 +393,23 @@ class ChatConfig(BaseSettings):
         description="E2B lifecycle action on timeout: 'pause' (default, free) or 'kill'.",
     )
 
+    # --- Local PC executor ---
+    use_local_pc_executor: bool = Field(
+        default=False,
+        description=(
+            "Route copilot execution to the user's local machine via the "
+            "autogpt-local-executor shim. When True, LocalPCShim is used instead of E2B."
+        ),
+    )
+    allow_computer_use: bool = Field(
+        default=False,
+        description="Allow Claude computer-use beta tools when the local PC shim is active.",
+    )
+    local_pc_executor_ws_path: str = Field(
+        default="/ws/local-executor",
+        description="WebSocket path prefix for the local executor shim endpoint.",
+    )
+
     @property
     def openrouter_active(self) -> bool:
         """True when OpenRouter config is shape-valid (flag + credentials).

autogpt_platform/backend/backend/copilot/config.py

@@ -120,6 +120,7 @@
 from ..token_tracking import persist_and_record_usage
 from ..tools import ToolGroup
 from ..tools.e2b_sandbox import get_or_create_sandbox, pause_sandbox_direct
+from ..tools.local_pc_shim import LocalPCShim, get_shim_manager
 from ..tools.sandbox import WORKSPACE_PREFIX, make_session_path
 from ..tracking import track_user_message
 from ..transcript import (
@@ -3812,6 +3813,19 @@ async def stream_chat_completion_sdk(  # pyright: ignore[reportGeneralTypeIssues
 
         async def _setup_e2b():
             """Set up E2B sandbox if configured, return sandbox or None."""
+            if config.use_local_pc_executor:
+                try:
+                    shim = await LocalPCShim.for_session(
+                        session_id, manager=get_shim_manager(), connect_timeout=30.0
+                    )
+                    return shim
+                except Exception as shim_err:
+                    logger.error(
+                        "[LocalPC] [%s] Shim connection failed: %s",
+                        session_id[:12],
+                        shim_err,
+                    )
+                    return None
             if not (e2b_api_key := config.active_e2b_api_key):
                 if config.use_e2b_sandbox:
                     logger.warning(
@@ -4919,7 +4933,7 @@ def _on_stderr(line: str) -> None:
         # _background_tasks to prevent garbage collection.
         # Use pause_sandbox_direct to skip the Redis lookup and reconnect
         # round-trip — e2b_sandbox is the live object from this turn.
-        if e2b_sandbox is not None:
+        if e2b_sandbox is not None and not isinstance(e2b_sandbox, LocalPCShim):
             task = asyncio.create_task(pause_sandbox_direct(e2b_sandbox, session_id))
             _background_tasks.add(task)
             task.add_done_callback(_background_tasks.discard)

autogpt_platform/backend/backend/copilot/sdk/service.py

@@ -0,0 +1,178 @@
+"""
+Platform-side binding for the autogpt-local-executor shim.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import time
+import uuid
+from typing import Any
+
+from fastapi import WebSocket
+
+logger = logging.getLogger(__name__)
+
+_shim_manager: "ShimConnectionManager | None" = None
+
+
+def get_shim_manager() -> "ShimConnectionManager":
+    global _shim_manager
+    if _shim_manager is None:
+        _shim_manager = ShimConnectionManager()
+    return _shim_manager
+
+
+class ShimConnectionManager:
+    def __init__(self) -> None:
+        self._connections: dict[str, WebSocket] = {}
+        self._waiters: dict[str, list[asyncio.Future[WebSocket]]] = {}
+
+    def register(self, session_id: str, ws: WebSocket) -> None:
+        self._connections[session_id] = ws
+        for fut in self._waiters.pop(session_id, []):
+            if not fut.done():
+                fut.set_result(ws)
+        logger.info("[LocalPC] Shim registered for session %s", session_id[:12])
+
+    def unregister(self, session_id: str) -> None:
+        self._connections.pop(session_id, None)
+        logger.info("[LocalPC] Shim unregistered for session %s", session_id[:12])
+
+    async def wait_for(self, session_id: str, timeout: float = 30.0) -> WebSocket:
+        if session_id in self._connections:
+            return self._connections[session_id]
+        loop = asyncio.get_event_loop()
+        fut: asyncio.Future[WebSocket] = loop.create_future()
+        self._waiters.setdefault(session_id, []).append(fut)
+        try:
+            return await asyncio.wait_for(asyncio.shield(fut), timeout=timeout)
+        except asyncio.TimeoutError:
+            raise TimeoutError(
+                f"[LocalPC] Shim for session {session_id[:12]} did not connect within {timeout}s"
+            )
+
+    def get(self, session_id: str) -> WebSocket | None:
+        return self._connections.get(session_id)
+
+
+class _FilesProxy:
+    def __init__(self, shim: "LocalPCShim") -> None:
+        self._shim = shim
+
+    async def read(self, path: str, *, encoding: str = "utf-8") -> str:
+        resp = await self._shim._rpc("FILE_READ", {"path": path, "encoding": encoding})
+        if resp.get("type") == "ERROR":
+            raise OSError(resp["payload"].get("message", "FILE_READ failed"))
+        return resp["payload"]["content"]
+
+    async def write(self, path: str, content: str, *, encoding: str = "utf-8") -> None:
+        resp = await self._shim._rpc(
+            "FILE_WRITE",
+            {"path": path, "content": content, "encoding": encoding, "create_parents": True},
+        )
+        if resp.get("type") == "ERROR":
+            raise OSError(resp["payload"].get("message", "FILE_WRITE failed"))
+
+
+class _CommandsProxy:
+    def __init__(self, shim: "LocalPCShim") -> None:
+        self._shim = shim
+
+    async def run(
+        self,
+        command: str,
+        *,
+        cwd: str | None = None,
+        timeout: int | None = None,
+        envs: dict[str, str] | None = None,
+    ) -> Any:
+        payload: dict[str, Any] = {"command": command}
+        if cwd:
+            payload["cwd"] = cwd
+        if timeout:
+            payload["timeout_seconds"] = timeout
+        if envs:
+            payload["env"] = envs
+        resp = await self._shim._rpc("EXECUTE_COMMAND", payload)
+        if resp.get("type") == "ERROR":
+            raise RuntimeError(resp["payload"].get("message", "EXECUTE_COMMAND failed"))
+        return _CommandResult(resp["payload"])
+
+
+class _CommandResult:
+    def __init__(self, payload: dict) -> None:
+        self.stdout = payload.get("stdout", "")
+        self.stderr = payload.get("stderr", "")
+        self.exit_code = payload.get("exit_code", -1)
+        self.timed_out = payload.get("timed_out", False)
+
+
+class LocalPCShim:
+    """
+    Drop-in replacement for E2B AsyncSandbox that routes execution to the
+    user's local machine via the autogpt-local-executor shim.
+
+    Duck-type contract: .commands.run(), .files.read(), .files.write(),
+                        .pause(), .kill(), .sandbox_id
+    """
+
+    def __init__(self, session_id: str, ws: WebSocket) -> None:
+        self.sandbox_id = session_id
+        self._ws = ws
+        self._pending: dict[str, asyncio.Future[dict]] = {}
+        self.files = _FilesProxy(self)
+        self.commands = _CommandsProxy(self)
+        self._recv_task = asyncio.create_task(self._recv_loop())
+
+    @classmethod
+    async def for_session(
+        cls,
+        session_id: str,
+        *,
+        manager: ShimConnectionManager,
+        connect_timeout: float = 30.0,
+    ) -> "LocalPCShim":
+        ws = await manager.wait_for(session_id, timeout=connect_timeout)
+        return cls(session_id, ws)
+
+    async def _rpc(self, msg_type: str, payload: dict, *, timeout: float = 30.0) -> dict:
+        msg_id = str(uuid.uuid4())
+        msg = {"type": msg_type, "id": msg_id, "ts": time.time(), "payload": payload}
+        loop = asyncio.get_event_loop()
+        fut: asyncio.Future[dict] = loop.create_future()
+        self._pending[msg_id] = fut
+        try:
+            await self._ws.send_text(json.dumps(msg))
+            return await asyncio.wait_for(asyncio.shield(fut), timeout=timeout)
+        except asyncio.TimeoutError:
+            self._pending.pop(msg_id, None)
+            raise TimeoutError(f"[LocalPC] RPC {msg_type} timed out after {timeout}s")
+
+    async def _recv_loop(self) -> None:
+        try:
+            async for raw in self._ws.iter_text():
+                try:
+                    msg = json.loads(raw)
+                    msg_id = msg.get("id")
+                    if msg_id and msg_id in self._pending:
+                        fut = self._pending.pop(msg_id)
+                        if not fut.done():
+                            fut.set_result(msg)
+                except Exception:
+                    logger.exception("[LocalPC] Error processing shim message")
+        except Exception:
+            logger.debug("[LocalPC] Shim recv loop ended for %s", self.sandbox_id[:12])
+
+    async def pause(self) -> None:
+        pass  # no billing on local machine
+
+    async def kill(self) -> None:
+        try:
+            await self._ws.close()
+        except Exception:
+            pass
+        if not self._recv_task.done():
+            self._recv_task.cancel()