feat(langchain): add per-thread sandbox isolation for chat applications

Add ThreadedSandboxManager to map conversation thread IDs to isolated
sandbox environments with persistent filesystems:

- ThreadedSandboxManager class with get_backend(), delete_thread(),
  cleanup_idle(), and close() methods for lifecycle management
- create_threaded_backend_factory() for DeepAgents integration that
  reads thread_id from LangGraph config
- SandboxClient enhancements: claim_name parameter for custom/deterministic
  names, delete_on_exit flag, connect() class method, delete() method,
  and was_reconnected property

Includes proper error handling:
- Resource cleanup on backend creation failure
- Continues delete() even if __exit__() fails
- Logs kubectl commands for manual cleanup on failures

New tests cover creation failure, deletion exceptions, concurrent access,
and call ordering (25 tests total for thread manager).

Example multi_thread_chat.py demonstrates multi-thread isolation with
LangGraph checkpointer for message history.

Author: johannhartmann

clients/python/agentic-sandbox-client/agentic_sandbox/sandbox_client.py

@@ -67,6 +67,24 @@ class ExecutionResult:
 class SandboxClient:
     """
     A client for creating and interacting with a stateful Sandbox via a router.
+
+    Args:
+        template_name: Name of the SandboxTemplate to claim.
+        namespace: Kubernetes namespace for the sandbox (default: "default").
+        gateway_name: Optional gateway name for production mode.
+        gateway_namespace: Namespace where the gateway lives (default: "default").
+        api_url: Direct API URL bypassing gateway discovery.
+        server_port: Port the sandbox runtime listens on (default: 8888).
+        sandbox_ready_timeout: Timeout waiting for sandbox readiness (default: 180s).
+        gateway_ready_timeout: Timeout waiting for gateway IP (default: 180s).
+        port_forward_ready_timeout: Timeout waiting for port-forward (default: 30s).
+        enable_tracing: Enable OpenTelemetry tracing.
+        trace_service_name: Service name for tracing (default: "sandbox-client").
+        claim_name: Optional custom claim name. If provided and the claim already
+            exists, the client reconnects to it (check was_reconnected property).
+            If the claim does not exist, a new one is created with this name.
+        delete_on_exit: Whether to delete the sandbox on context exit (default: True).
+            Set to False for persistent sandboxes that survive across sessions.
     """
 
     def __init__(
@@ -82,6 +100,8 @@ def __init__(
         port_forward_ready_timeout: int = 30,
         enable_tracing: bool = False,
         trace_service_name: str = "sandbox-client",
+        claim_name: str | None = None,
+        delete_on_exit: bool = True,
     ):
         self.trace_service_name = trace_service_name
         self.tracing_manager = None
@@ -105,11 +125,14 @@ def __init__(
         self.sandbox_ready_timeout = sandbox_ready_timeout
         self.gateway_ready_timeout = gateway_ready_timeout
         self.port_forward_ready_timeout = port_forward_ready_timeout
+        self.delete_on_exit = delete_on_exit
+        self._custom_claim_name = claim_name
 
         self.port_forward_process: subprocess.Popen | None = None
 
         self.claim_name: str | None = None
         self.sandbox_name: str | None = None
+        self._reconnected: bool = False
         self.pod_name: str | None = None
         self.annotations: dict | None = None
 
@@ -135,10 +158,133 @@ def is_ready(self) -> bool:
         """Returns True if the sandbox is ready and the Gateway IP has been found."""
         return self.base_url is not None
 
+    @property
+    def was_reconnected(self) -> bool:
+        """Returns True if this client reconnected to an existing sandbox.
+
+        Only valid after entering the context manager. Returns True if a
+        custom claim_name was provided and the claim already existed.
+        """
+        return self._reconnected
+
+    @classmethod
+    def connect(
+        cls,
+        claim_name: str,
+        template_name: str,
+        namespace: str = "default",
+        **kwargs,
+    ) -> "SandboxClient":
+        """Connect to an existing sandbox by claim name.
+
+        This is a convenience method for reconnecting to a sandbox that was
+        created with delete_on_exit=False. If the claim doesn't exist, it will
+        be created.
+
+        Note:
+            This method automatically sets delete_on_exit=False, so the sandbox
+            will persist after the context manager exits.
+
+        Args:
+            claim_name: The name of the existing SandboxClaim.
+            template_name: The template name (required for creating if missing).
+            namespace: Kubernetes namespace (default: "default").
+            **kwargs: Additional arguments passed to SandboxClient.
+
+        Returns:
+            SandboxClient configured to connect to the existing sandbox.
+
+        Example:
+            # First session - create persistent sandbox
+            with SandboxClient(
+                template_name="python",
+                claim_name="my-sandbox",
+                delete_on_exit=False
+            ) as client:
+                client.run("echo 'hello' > /app/state.txt")
+
+            # Later session - reconnect (delete_on_exit=False is set automatically)
+            with SandboxClient.connect(
+                claim_name="my-sandbox",
+                template_name="python"
+            ) as client:
+                result = client.run("cat /app/state.txt")
+        """
+        return cls(
+            template_name=template_name,
+            namespace=namespace,
+            claim_name=claim_name,
+            delete_on_exit=False,
+            **kwargs,
+        )
+
+    def delete(self) -> bool:
+        """Explicitly delete the sandbox claim.
+
+        Useful when delete_on_exit=False but you want to clean up manually.
+
+        Returns:
+            True if deleted successfully, False if not found or error.
+        """
+        if not self.claim_name:
+            logging.warning("No claim name set, nothing to delete")
+            return False
+
+        logging.info(f"Explicitly deleting SandboxClaim: {self.claim_name}")
+        try:
+            self.custom_objects_api.delete_namespaced_custom_object(
+                group=CLAIM_API_GROUP,
+                version=CLAIM_API_VERSION,
+                namespace=self.namespace,
+                plural=CLAIM_PLURAL_NAME,
+                name=self.claim_name
+            )
+            return True
+        except client.ApiException as e:
+            if e.status == 404:
+                logging.warning(f"SandboxClaim '{self.claim_name}' not found")
+                return False
+            logging.error(f"Error deleting sandbox claim: {e}", exc_info=True)
+            return False
+        except Exception as e:
+            logging.error(f"Unexpected error deleting sandbox claim: {e}", exc_info=True)
+            return False
+
+    def _claim_exists(self, claim_name: str) -> bool:
+        """Check if a SandboxClaim with the given name exists."""
+        try:
+            self.custom_objects_api.get_namespaced_custom_object(
+                group=CLAIM_API_GROUP,
+                version=CLAIM_API_VERSION,
+                namespace=self.namespace,
+                plural=CLAIM_PLURAL_NAME,
+                name=claim_name,
+            )
+            return True
+        except client.ApiException as e:
+            if e.status == 404:
+                return False
+            raise
+
     @trace_span("create_claim")
     def _create_claim(self, trace_context_str: str = ""):
-        """Creates the SandboxClaim custom resource in the Kubernetes cluster."""
-        self.claim_name = f"sandbox-claim-{os.urandom(4).hex()}"
+        """Creates the SandboxClaim custom resource in the Kubernetes cluster.
+
+        If a custom claim_name was provided and that claim already exists,
+        this method will reconnect to it instead of creating a new one.
+        """
+        if self._custom_claim_name:
+            self.claim_name = self._custom_claim_name
+            # Check if the claim already exists
+            if self._claim_exists(self.claim_name):
+                logging.info(
+                    f"Reconnecting to existing SandboxClaim '{self.claim_name}' "
+                    f"in namespace '{self.namespace}'..."
+                )
+                self._reconnected = True
+                return
+        else:
+            self.claim_name = f"sandbox-claim-{os.urandom(4).hex()}"
 
         span = trace.get_current_span()
         if span.is_recording():
@@ -363,8 +509,8 @@ def __exit__(self, exc_type, exc_val, exc_tb):
             except Exception as e:
                 logging.error(f"Failed to stop port-forwarding: {e}")
 
-        # Delete the SandboxClaim
-        if self.claim_name:
+        # Delete the SandboxClaim only if delete_on_exit is True
+        if self.claim_name and self.delete_on_exit:
             logging.info(f"Deleting SandboxClaim: {self.claim_name}")
             try:
                 self.custom_objects_api.delete_namespaced_custom_object(
@@ -381,6 +527,10 @@ def __exit__(self, exc_type, exc_val, exc_tb):
             except Exception as e:
                 logging.error(
                     f"Unexpected error deleting sandbox claim: {e}", exc_info=True)
+        elif self.claim_name and not self.delete_on_exit:
+            logging.info(
+                f"Keeping SandboxClaim '{self.claim_name}' alive (delete_on_exit=False)"
+            )
 
         # Cleanup Trace if it exists
         if self.tracing_manager:

clients/python/langchain-agent-sandbox/langchain_agent_sandbox/__init__.py

@@ -3,11 +3,15 @@
     SandboxPolicyWrapper,
     WarmPoolBackend,
     create_sandbox_backend_factory,
+    create_threaded_backend_factory,
 )
+from .thread_manager import ThreadedSandboxManager
 
 __all__ = [
     "AgentSandboxBackend",
     "SandboxPolicyWrapper",
     "WarmPoolBackend",
+    "ThreadedSandboxManager",
     "create_sandbox_backend_factory",
+    "create_threaded_backend_factory",
 ]

clients/python/langchain-agent-sandbox/langchain_agent_sandbox/backend.py

@@ -669,6 +669,97 @@ def factory(_runtime: Any) -> AgentSandboxBackend:
     return factory
 
 
+def _get_thread_manager_class() -> type:
+    """Lazy import of ThreadedSandboxManager to avoid circular imports."""
+    from .thread_manager import ThreadedSandboxManager
+    return ThreadedSandboxManager
+
+
+def create_threaded_backend_factory(
+    template_name: str,
+    manager: Optional[Any] = None,
+    namespace: str = "default",
+    **kwargs: Any,
+) -> Callable[[Any], AgentSandboxBackend]:
+    """Create a BackendFactory that provides per-thread sandbox isolation.
+
+    This factory reads the thread_id from the LangGraph config and uses a
+    ThreadedSandboxManager to provide isolated sandboxes per conversation thread.
+    Each thread gets its own sandbox with persistent filesystem.
+
+    Usage:
+        from deepagents import create_deep_agent
+        from langgraph.checkpoint.memory import MemorySaver
+        from langchain_agent_sandbox import (
+            ThreadedSandboxManager,
+            create_threaded_backend_factory,
+        )
+
+        # Create manager for lifecycle control
+        manager = ThreadedSandboxManager(
+            template_name="python-deepagent",
+            idle_ttl=timedelta(hours=1),
+        )
+
+        # Create agent with threaded backend
+        agent = create_deep_agent(
+            model=model,
+            backend=create_threaded_backend_factory("python-deepagent", manager=manager),
+            checkpointer=MemorySaver(),  # For message history
+        )
+
+        # Same thread = same sandbox (filesystem persists)
+        agent.invoke(msg1, config={"configurable": {"thread_id": "user-123"}})
+        agent.invoke(msg2, config={"configurable": {"thread_id": "user-123"}})
+
+        # Different thread = different sandbox
+        agent.invoke(msg3, config={"configurable": {"thread_id": "user-456"}})
+
+        # Cleanup when done
+        manager.close()
+
+    Args:
+        template_name: Name of the SandboxTemplate to claim.
+        manager: Optional ThreadedSandboxManager instance. If not provided,
+            one will be created (but you won't have lifecycle control).
+        namespace: Kubernetes namespace for the sandbox.
+        **kwargs: Additional arguments passed to ThreadedSandboxManager.
+
+    Returns:
+        A factory callable that accepts a ToolRuntime and returns an
+        AgentSandboxBackend for the current thread.
+
+    Note:
+        The thread_id is read from `runtime.config.get("configurable", {}).get("thread_id")`.
+        If no thread_id is found, a default "default-thread" is used.
+
+    Warning:
+        If no manager is provided, an internal manager is created but you will
+        have no way to call close() or delete_thread(). For production use,
+        always pass an explicit manager instance for lifecycle control.
+    """
+    # Create manager if not provided
+    _manager = manager
+    if _manager is None:
+        ThreadedSandboxManager = _get_thread_manager_class()
+        _manager = ThreadedSandboxManager(
+            template_name=template_name,
+            namespace=namespace,
+            **kwargs,
+        )
+
+    def factory(runtime: Any) -> AgentSandboxBackend:
+        # Extract thread_id from LangGraph config
+        config = getattr(runtime, "config", {}) or {}
+        configurable = config.get("configurable", {}) or {}
+        thread_id = configurable.get("thread_id", "default-thread")
+
+        logger.debug("Creating backend for thread_id: %s", thread_id)
+        return _manager.get_backend(thread_id)
+
+    return factory
+
+
 class SandboxPolicyWrapper:
     """Wraps AgentSandboxBackend with policy enforcement.