Run the langchain-quickjs code interpreter in-workflow

CHANGELOG.md

@@ -20,6 +20,13 @@ 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
 

pyproject.toml

@@ -98,6 +98,7 @@ dev = [
   "langgraph>=1.1.0",
   "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",

tests/contrib/deepagents/test_quickjs_interpreter.py

@@ -0,0 +1,171 @@
+"""The ``langchain-quickjs`` code interpreter runs in-workflow with durable bridges.
+
+The middleware's ``eval`` tool runs model-written JavaScript in a QuickJS VM.
+Upstream hosts the VM on its own thread and wakes the caller through
+``call_soon_threadsafe``, which the workflow event loop cannot service: without
+the plugin's inline patch the workflow parks forever after the first model
+call. With it, JavaScript is workflow code, a sub-agent dispatched from
+JavaScript via ``task()`` runs in-workflow (so its model call is an
+``invoke_model`` activity), and a PTC tool wrapped with ``tool_as_activity``
+crosses as an ``invoke_tool`` activity. Both scenarios replay from history.
+"""
+
+from __future__ import annotations
+
+import sys
+import uuid
+from datetime import timedelta
+
+import pytest
+
+from temporalio.testing import WorkflowEnvironment
+
+pytestmark = pytest.mark.skipif(
+    sys.version_info < (3, 11), reason="deepagents requires Python >= 3.11"
+)
+pytest.importorskip("deepagents")
+pytest.importorskip("langchain_core")
+pytest.importorskip("langchain_quickjs")
+
+from temporalio import workflow  # noqa: E402
+from temporalio.worker import Replayer, Worker  # noqa: E402
+from tests.contrib.deepagents.helpers import count_scheduled_activities  # noqa: E402
+
+# Bind deepagents symbols off the module importorskip returns: a static
+# `from deepagents import ...` cannot resolve on Python 3.10 (deepagents
+# needs >= 3.11), and with the package absent the type checkers mis-resolve
+# the name against this same-named test directory.
+create_deep_agent = pytest.importorskip("deepagents").create_deep_agent
+CodeInterpreterMiddleware = pytest.importorskip(
+    "langchain_quickjs"
+).CodeInterpreterMiddleware
+
+with workflow.unsafe.imports_passed_through():
+    from langchain_core.messages import AIMessage
+
+    from temporalio.contrib.deepagents import DeepAgentsPlugin, tool_as_activity
+    from temporalio.contrib.deepagents.testing import MockTool, mock_model_provider
+
+INVOKE_MODEL = "deepagents.invoke_model"
+INVOKE_TOOL = "deepagents.invoke_tool"
+
+# One eval: a PTC tool call, then a sub-agent dispatched from JavaScript.
+_JS = (
+    'const found = await tools.lookup({q: "x"}); '
+    'const summary = await task({description: "Summarize", subagentType: "researcher"}); '
+    "`${found}|${summary}`"
+)
+
+
+@workflow.defn
+class InterpreterWorkflow:
+    @workflow.run
+    async def run(self, question: str) -> str:
+        lookup = tool_as_activity(
+            MockTool(name="lookup", description="Look up q.", result="looked-up"),
+            start_to_close_timeout=timedelta(seconds=10),
+        )
+        agent = create_deep_agent(
+            model="anthropic:claude-sonnet-4-5",
+            system_prompt="You coordinate.",
+            tools=[lookup],
+            subagents=[
+                {
+                    "name": "researcher",
+                    "description": "Researches.",
+                    "system_prompt": "You research.",
+                }
+            ],
+            middleware=[CodeInterpreterMiddleware(ptc=["lookup"])],
+        )
+        result = await agent.ainvoke(
+            {"messages": [{"role": "user", "content": question}]}
+        )
+        # The eval tool's result carries what crossed back through JavaScript.
+        return str(result["messages"][-2].content)
+
+
+def _plugin() -> DeepAgentsPlugin:
+    # Main agent asks for one eval; the sub-agent answers; the main agent finishes.
+    return DeepAgentsPlugin(
+        model_provider=mock_model_provider(
+            [
+                AIMessage(
+                    content="",
+                    tool_calls=[{"name": "eval", "args": {"code": _JS}, "id": "c1"}],
+                ),
+                "Researcher findings.",
+                "Final answer.",
+            ]
+        ),
+    )
+
+
+async def _run_and_replay(env: WorkflowEnvironment, **worker_kwargs: object) -> None:
+    plugin = _plugin()
+    task_queue = f"da-quickjs-{uuid.uuid4()}"
+    async with Worker(
+        env.client,
+        task_queue=task_queue,
+        workflows=[InterpreterWorkflow],
+        plugins=[plugin],
+        **worker_kwargs,  # type: ignore[arg-type]
+    ):
+        handle = await env.client.start_workflow(
+            InterpreterWorkflow.run,
+            "Go",
+            id=f"da-quickjs-{uuid.uuid4()}",
+            task_queue=task_queue,
+            execution_timeout=timedelta(seconds=60),
+        )
+        out = await handle.result()
+        history = await handle.fetch_history()
+
+    assert out == "<result>looked-up|Researcher findings.</result>"
+    counts = await count_scheduled_activities(handle)
+    # main -> eval, the sub-agent dispatched from JavaScript, main -> final.
+    assert counts[INVOKE_MODEL] == 3, counts
+    # The PTC call reached the tool_as_activity wrapper, not the tool body.
+    assert counts[INVOKE_TOOL] == 1, counts
+
+    # The JavaScript re-executes on replay against the recorded activity results.
+    await Replayer(workflows=[InterpreterWorkflow], plugins=[plugin]).replay_workflow(
+        history
+    )
+
+
+@pytest.mark.asyncio
+async def test_interpreter_runs_in_workflow_with_durable_bridges(
+    env: WorkflowEnvironment,
+) -> None:
+    await _run_and_replay(env)
+
+
+@pytest.mark.asyncio
+async def test_interpreter_survives_cache_eviction(env: WorkflowEnvironment) -> None:
+    # Every activation replays from scratch, so the eval (which spans a tool
+    # activity and a sub-agent's model activity) is re-driven several times.
+    await _run_and_replay(env, max_cached_workflows=0)
+
+
+def test_inline_patch_is_inert_outside_workflows() -> None:
+    # Activities and clients in the same process keep upstream's thread-hosted
+    # VM; only in-workflow calls run inline.
+    from quickjs_rs.threading import ThreadWorker
+
+    from temporalio.contrib.deepagents import _quickjs
+
+    _quickjs.install_quickjs_inline_patch()
+    try:
+
+        async def probe() -> str:
+            return "ran"
+
+        worker = ThreadWorker(name="probe")
+        try:
+            assert worker.run_sync(probe()) == "ran"
+            assert worker._thread is not None, "expected upstream's worker thread"
+        finally:
+            worker.close()
+    finally:
+        _quickjs.uninstall_quickjs_inline_patch()