Run the langchain-quickjs code interpreter in-workflow

CHANGELOG.md

@@ -20,15 +20,32 @@ to include examples, links to docs, or any other relevant information.
 
 ### Added
 
+- `temporalio.contrib.deepagents` runs Deep Agents' `langchain-quickjs` code interpreter
+  in-workflow: the QuickJS REPL executes on the workflow event loop (upstream's dedicated
+  thread cannot wake the deterministic loop, which parked the workflow forever), a sub-agent
+  dispatched from JavaScript via `task()` gets its own `deepagents.invoke_model` Activities,
+  and the interpreter's modules pass through the sandbox. `langchain-quickjs` remains
+  optional; nothing is imported unless the workflow imports it.
+
 - Added the `temporalio.contrib.gcp.cloud_run.id` module with the `CloudRunIdPlugin` client plugin to set the worker identity on Cloud Run.
 ### Changed
 
+- The `deepagents` extra now requires `deepagents>=0.7.12,<0.8` (was `<0.7`). Because
+  deepagents 0.7 requires `langsmith>=0.11.2`, the `langsmith` extra now allows
+  `langsmith<0.13` (was `<0.9`).
+
 ### Deprecated
 
 ### :boom: Breaking Changes
 
 ### Fixed
 
+- `temporalio.contrib.deepagents.TemporalBackend` mirrors the wrapped backend's optional
+  capabilities at class level, where deepagents 0.7 checks them: `delete` / `adelete` are
+  offered (and run as `deepagents.backend_op` Activities) only for a delete-capable backend,
+  and `execute` / `aexecute` only for an execution-capable one, so a wrapped filesystem or
+  store backend no longer advertises a shell tool that could only fail. User subclasses of
+  `TemporalBackend` get the same mirroring.
 - Current workflow and activity payload converter accessors now return the configured converter
   without SDK-internal transfer type conversion.
 

pyproject.toml

@@ -31,9 +31,9 @@ pydantic = ["pydantic>=2.0.0,<3"]
 openai-agents = ["openai-agents>=0.19.2,<0.20", "mcp>=1.9.4, <2"]
 google-adk = ["google-adk>=2.2.0,<3", "mcp>=1.24,<2"]
 langgraph = ["langgraph>=1.1.0"]
-langsmith = ["langsmith>=0.7.34,<0.9"]
+langsmith = ["langsmith>=0.7.34,<0.13"]
 deepagents = [
-  "deepagents>=0.6.12,<0.7; python_version >= '3.11'",
+  "deepagents>=0.7.12,<0.8; python_version >= '3.11'",
   "langchain>=1.3.11,<2; python_version >= '3.11'",
   "langchain-core>=1.4.8,<2; python_version >= '3.11'",
 ]
@@ -96,8 +96,9 @@ dev = [
   "pytest-xdist>=3.6,<4",
   "moto[s3,server]>=5",
   "langgraph>=1.1.0",
-  "langsmith>=0.7.34,<0.9",
-  "deepagents>=0.6.12,<0.7; python_version >= '3.11'",
+  "langsmith>=0.7.34,<0.13",
+  "deepagents>=0.7.12,<0.8; python_version >= '3.11'",
+  "langchain-quickjs>=0.3.5,<0.4; python_version >= '3.11'",
   "langchain>=1.3.11,<2; python_version >= '3.11'",
   "langchain-core>=1.4.8,<2; python_version >= '3.11'",
   "langchain-anthropic>=1.4.7; python_version >= '3.11'",

temporalio/contrib/deepagents/README.md

@@ -315,6 +315,29 @@ worker is unaffected, and the original function is restored when the worker
 stops. If you would rather be explicit, use `create_temporal_deep_agent` or
 pass `TemporalModel("provider:name")` yourself.
 
+## Code interpreter (`langchain-quickjs`)
+
+Deep Agents' [code interpreter](https://docs.langchain.com/oss/python/deepagents/interpreters)
+(`pip install "deepagents[quickjs]"`) runs inside the workflow: add
+`CodeInterpreterMiddleware()` to `create_deep_agent(middleware=[...])` as usual. The
+plugin runs the QuickJS VM on the workflow's own event loop (upstream hosts it on a
+thread the deterministic loop cannot service) and passes `langchain_quickjs`,
+`quickjs_rs`, `wasmtime`, and `bsdiff4` through the sandbox. `langchain-quickjs` stays
+optional: nothing is imported unless your workflow imports it.
+
+What is durable: the JavaScript itself is workflow code and replays; a sub-agent
+dispatched from JavaScript with `task(...)` runs in-workflow, so its model calls are
+`deepagents.invoke_model` Activities like any other sub-agent's; a tool called from
+JavaScript through PTC (`tools.<name>(...)`) follows that tool's own Workflow-vs-Activity
+choice above, so wrap I/O tools with `tool_as_activity`.
+
+Rules for the JavaScript, because it is workflow code: do not read the clock or
+randomness — inside the VM `Date.now()` is wall-clock and `Math.random()` is seeded per
+runtime, so both break replay; set the middleware's `timeout=` generously, since it is
+wall-clock; and prefer `mode="turn"` — a `mode="thread"` heap snapshot lives in graph
+state and is not carried across `run_deep_agent`'s continue-as-new. Upstream's own
+caveat applies too: PTC and `task()` calls do not pass through `interrupt_on` approval.
+
 ## Composing with other plugins
 
 This plugin carries no tracing context of its own. For observability, compose it

temporalio/contrib/deepagents/_plugin.py

@@ -205,9 +205,16 @@ async def _run_context(self) -> AsyncIterator[None]:
         # Installed for the life of the process, never uninstalled — see
         # _install_langsmith_temporal_override for why.
         _install_langsmith_temporal_override()
+        # No-op unless quickjs-rs (the langchain-quickjs code interpreter's VM)
+        # is installed; see _quickjs for why its worker thread cannot be used
+        # from a workflow.
+        from temporalio.contrib.deepagents import _quickjs
+
+        _quickjs.install_quickjs_inline_patch()
         try:
             yield
         finally:
+            _quickjs.uninstall_quickjs_inline_patch()
             if patched:
                 # Import is cached: patched=True implies the import above succeeded.
                 from temporalio.contrib.deepagents import _model, _tools

temporalio/contrib/deepagents/_quickjs.py

@@ -0,0 +1,87 @@
+"""In-workflow execution of the ``langchain-quickjs`` code interpreter.
+
+``langchain_quickjs.CodeInterpreterMiddleware`` gives a Deep Agent an ``eval``
+tool that runs model-written JavaScript in a QuickJS VM. Upstream hosts that VM
+on a dedicated OS thread per LangGraph thread
+(``quickjs_rs.threading.ThreadWorker``) and hands results back to the caller's
+loop through ``asyncio.wrap_future`` — that is, ``call_soon_threadsafe``. The
+deterministic workflow event loop neither implements that nor runs outside an
+activation, so an unmodified ``eval`` parks the workflow forever: no failure,
+no deadlock report, the execution just never wakes up.
+
+Nothing about the REPL needs the thread. ``quickjs_rs`` drives the VM
+cooperatively on whatever loop awaits it, and the middleware's ``task()`` /
+``tools.*`` bridges already run their callbacks directly when the calling loop
+is the loop that invoked ``eval``. So inside a workflow the worker's hops are
+made inline: JavaScript runs as workflow code, a sub-agent dispatched from
+JavaScript runs in-workflow (its model calls are ``deepagents.invoke_model``
+activities like any other sub-agent's), and a PTC tool call follows the tool's
+own Workflow-vs-Activity wrapping. Outside a workflow upstream behavior is
+untouched, and when ``quickjs_rs`` is not installed there is nothing to patch.
+"""
+
+from __future__ import annotations
+
+import importlib
+from typing import Any
+
+from temporalio import workflow
+
+_originals: dict[str, Any] = {}
+
+
+def install_quickjs_inline_patch() -> None:
+    """Run ``quickjs_rs`` worker hops inline while in a workflow. Idempotent."""
+    try:
+        threading_mod = importlib.import_module("quickjs_rs.threading")
+    except ImportError:
+        return
+    if _originals:
+        return
+    worker_cls = threading_mod.ThreadWorker
+    orig_run_sync = worker_cls.run_sync
+    orig_run_async = worker_cls.run_async
+    orig_ensure_started = worker_cls._ensure_started
+
+    def run_sync(self: Any, coro: Any) -> Any:
+        if not workflow.in_workflow():
+            return orig_run_sync(self, coro)
+        # The REPL's synchronous hops (context creation, snapshot, close) never
+        # suspend; drive the coroutine to completion right here.
+        try:
+            coro.send(None)
+        except StopIteration as done:
+            return done.value
+        coro.close()
+        raise RuntimeError(
+            "quickjs-rs REPL work suspended on the synchronous in-workflow path"
+        )
+
+    async def run_async(self: Any, coro: Any) -> Any:
+        if not workflow.in_workflow():
+            return await orig_run_async(self, coro)
+        return await coro
+
+    def ensure_started(self: Any) -> None:
+        if workflow.in_workflow():
+            return
+        orig_ensure_started(self)
+
+    _originals.update(
+        run_sync=orig_run_sync,
+        run_async=orig_run_async,
+        _ensure_started=orig_ensure_started,
+    )
+    worker_cls.run_sync = run_sync
+    worker_cls.run_async = run_async
+    worker_cls._ensure_started = ensure_started
+
+
+def uninstall_quickjs_inline_patch() -> None:
+    """Restore ``quickjs_rs``'s thread-based worker hops."""
+    if not _originals:
+        return
+    worker_cls = importlib.import_module("quickjs_rs.threading").ThreadWorker
+    for name, original in _originals.items():
+        setattr(worker_cls, name, original)
+    _originals.clear()

temporalio/contrib/deepagents/_serde.py

@@ -383,6 +383,12 @@ def cache_put(key: str, value: Any) -> None:
     "langchain_anthropic",
     "langgraph",
     "deepagents",
+    # The langchain-quickjs code interpreter and its VM; a name-only allowlist,
+    # so nothing here is imported unless the workflow imports it.
+    "langchain_quickjs",
+    "quickjs_rs",
+    "wasmtime",
+    "bsdiff4",
     "langsmith",
     "numpy",
     "pydantic",

temporalio/contrib/deepagents/_tools.py

@@ -386,6 +386,7 @@ def _is_coroutine(fn: Any) -> bool:
     "adownload_files": "download_files",
     "aupload_files": "upload_files",
     "aexecute": "execute",
+    "adelete": "delete",
 }
 
 _original_backend_async_defaults: dict[str, Any] = {}
@@ -462,6 +463,7 @@ def uninstall_backend_async_patch() -> None:
     "download_files",
     "upload_files",
     "execute",
+    "delete",
     # Async twins (what FilesystemMiddleware actually calls).
     "als",
     "als_info",
@@ -475,8 +477,17 @@ def uninstall_backend_async_patch() -> None:
     "adownload_files",
     "aupload_files",
     "aexecute",
+    "adelete",
 )
 
+# Capability-gated ops. deepagents decides whether a backend has ``delete`` from
+# the CLASS (``type(backend).delete`` against the protocol default) and whether
+# it can ``execute`` from a nominal ``isinstance`` check against
+# ``SandboxBackendProtocol`` (an ABC subclass, not a structural Protocol).
+# Neither can be answered per instance, so the wrapper mirrors the inner
+# backend at class level; see :func:`_wrapper_class_for`.
+_OPTIONAL_OPS = frozenset({"delete", "adelete", "execute", "aexecute"})
+
 
 class TemporalBackend:
     """Route a real-I/O backend's operations through Temporal activities.
@@ -491,6 +502,25 @@ class TemporalBackend:
     metadata / configuration the agent reads (but that does no I/O) still works.
     """
 
+    def __new__(
+        cls,
+        inner: Any,
+        *,
+        activity_options: Mapping[str, Any] | None = None,
+    ) -> "TemporalBackend":
+        """Pick the wrapper class that mirrors ``inner``'s optional capabilities.
+
+        Applies to subclasses too: a user's ``class MyBackend(TemporalBackend)``
+        gets a mirrored subclass of ``MyBackend``, so ``isinstance`` and any
+        methods it defines are preserved.
+        """
+        target: type[TemporalBackend] = (
+            cls
+            if getattr(cls, "_temporal_mirror_of", None) is not None
+            else _wrapper_class_for(inner, cls)
+        )
+        return object.__new__(target)
+
     def __init__(
         self,
         inner: Any,
@@ -533,14 +563,80 @@ async def _dispatch(self, op: str, *args: Any, **kwargs: Any) -> Any:
         return _serde.load_backend_result(output.result)
 
     def __getattr__(self, name: str) -> Any:
-        """Bound-method access for a known I/O op returns an activity dispatcher.
+        """Forward anything that is not an I/O op to the inner backend unchanged."""
+        return getattr(self._inner, name)
 
-        Everything else forwards to the inner backend unchanged.
-        """
-        if name in _BACKEND_OPS:
 
-            async def _op(*args: Any, **kwargs: Any) -> Any:
-                return await self._dispatch(name, *args, **kwargs)
+def _make_backend_op(name: str) -> Callable[..., Any]:
+    async def _op(self: TemporalBackend, *args: Any, **kwargs: Any) -> Any:
+        return await self._dispatch(name, *args, **kwargs)
 
-            return _op
-        return getattr(self._inner, name)
+    _op.__name__ = _op.__qualname__ = name
+    return _op
+
+
+# The I/O ops are real class-level methods, not ``__getattr__`` products, so
+# class-based capability checks in deepagents see them.
+for _op_name in _BACKEND_OPS:
+    if _op_name not in _OPTIONAL_OPS:
+        setattr(TemporalBackend, _op_name, _make_backend_op(_op_name))
+
+_wrapper_classes: dict[tuple[type, bool, bool], type[TemporalBackend]] = {}
+
+
+def _wrapper_class_for(
+    inner: Any, base: type[TemporalBackend]
+) -> type[TemporalBackend]:
+    """Return the subclass of ``base`` whose optional capabilities mirror ``inner``.
+
+    ``delete`` / ``adelete``: a delete-capable inner backend gets dispatchers
+    like every other op; any other inner backend keeps the protocol defaults,
+    so deepagents disables its delete tool exactly as it would for the
+    unwrapped backend. Execution: deepagents offers its shell tool only to a
+    nominal ``SandboxBackendProtocol`` instance, so when deepagents reports the
+    inner backend as execution-capable the mirror also derives from that class
+    (with ``execute`` / ``aexecute`` dispatchers and ``id`` forwarded); a wrapper
+    around a plain filesystem or store backend stays a non-sandbox class. An op
+    that ``base`` already defines (a user subclass) is left alone. Without
+    deepagents installed there is nothing to mirror and ``base`` is used as-is.
+    """
+    try:
+        protocol = importlib.import_module("deepagents.backends.protocol")
+        filesystem = importlib.import_module("deepagents.middleware.filesystem")
+    except ImportError:
+        return base
+    default_delete = protocol.BackendProtocol.delete
+    has_delete = getattr(type(inner), "delete", default_delete) is not default_delete
+    has_execute = bool(filesystem.supports_execution(inner))
+    key = (base, has_delete, has_execute)
+    cls = _wrapper_classes.get(key)
+    if cls is None:
+        namespace: dict[str, Any] = {
+            "__module__": base.__module__,
+            "__qualname__": base.__qualname__,
+            "__doc__": base.__doc__,
+            "_temporal_mirror_of": base,
+        }
+        for op in ("delete", "adelete"):
+            if hasattr(base, op):
+                continue
+            namespace[op] = (
+                _make_backend_op(op)
+                if has_delete
+                else getattr(protocol.BackendProtocol, op)
+            )
+        bases: tuple[type, ...] = (base,)
+        if has_execute:
+            sandbox_cls = protocol.SandboxBackendProtocol
+            if not issubclass(base, sandbox_cls):
+                bases = (base, sandbox_cls)
+            for op in ("execute", "aexecute"):
+                if not hasattr(base, op):
+                    namespace[op] = _make_backend_op(op)
+            if not hasattr(base, "id"):
+                # The sandbox class defines ``id``; keep reading the inner
+                # backend's, which ``__getattr__`` would otherwise have served.
+                namespace["id"] = property(lambda self: self._inner.id)
+        cls = type(base.__name__, bases, namespace)
+        _wrapper_classes[key] = cls
+    return cls

temporalio/contrib/langsmith/_interceptor.py

@@ -336,8 +336,13 @@ def end(self, **kwargs: Any) -> None:
             kwargs.setdefault("end_time", temporalio.workflow.now())
         self._run.end(**kwargs)
 
-    def patch(self, *, exclude_inputs: bool = False) -> None:
-        """Patch the run to LangSmith, skipping during replay."""
+    def patch(self, *, exclude_inputs: bool | None = None) -> None:
+        """Patch the run to LangSmith, skipping during replay.
+
+        ``exclude_inputs`` is forwarded untouched so LangSmith's own default
+        applies (a plain ``False`` before 0.12; the
+        ``LANGSMITH_EXCLUDE_INPUTS_ON_PATCH`` setting from 0.12 on).
+        """
         if temporalio.workflow.in_workflow():
             if _is_replaying():
                 return
@@ -394,7 +399,7 @@ def post(self, exclude_child_runs: bool = True) -> NoReturn:
         """Factory must never be posted."""
         raise RuntimeError("_RootReplaySafeRunTreeFactory must never be posted")
 
-    def patch(self, *, exclude_inputs: bool = False) -> NoReturn:
+    def patch(self, *, exclude_inputs: bool | None = None) -> NoReturn:
         """Factory must never be patched."""
         raise RuntimeError("_RootReplaySafeRunTreeFactory must never be patched")