fix(quickjs): propagate JS task() subagent interrupts (#4401)

QuickJS exposes a JavaScript `task()` helper so an eval can fan out work
to Deep Agents subagents, but that helper was not quite equivalent to
the normal `task` tool path. When a JS-launched subagent hit HITL, the
nested `GraphInterrupt` was caught and rendered as an eval error instead
of reaching LangGraph’s normal interrupt machinery; the bridge also
skipped the parent runtime config and child tool call id when invoking
the underlying task tool.

## Changes

- Preserve the parent `ToolRuntime.config` and pass a synthetic child
`tool_call_id` when `call_subagent_task_tool` invokes the Deep Agents
task tool. This keeps subagent dispatch closer to the normal `ToolNode`
path, including config-driven behavior such as thread/run metadata and
runtime context propagation.
- Let `GraphInterrupt` bubble through both the subagent helper and the
QuickJS `task()` bridge. Other validation/runtime failures are still
wrapped as eval errors, but LangGraph interrupts remain control-flow
signals rather than being converted into `<error
type="GraphInterrupt">...`.

libs/partners/quickjs/langchain_quickjs/_repl.py

@@ -16,6 +16,7 @@ import uuid
 from dataclasses import dataclass, field, replace
 from typing import TYPE_CHECKING, Any
 
+from langgraph.errors import GraphInterrupt
 from quickjs_rs import (
     UNDEFINED,
     ConcurrentEvalError,
@@ -633,6 +634,8 @@ class _ThreadREPL:
                         payload,
                         state=state,
                     )
+                except GraphInterrupt:
+                    raise
                 except Exception as e:
                     # Subagent dispatches are part of the eval language, not
                     # PTC calls. Surface their validation/runtime failures as

libs/partners/quickjs/langchain_quickjs/_subagent.py

@@ -10,6 +10,7 @@ from dataclasses import replace
 from typing import TYPE_CHECKING, Any, Final, Literal, NotRequired, TypedDict
 
 from langchain.agents.structured_output import AutoStrategy
+from langgraph.errors import GraphInterrupt
 
 from langchain_quickjs._format import coerce_tool_output_for_ptc
 
@@ -239,8 +240,12 @@ async def call_subagent_task_tool(
                 "description": description,
                 "subagent_type": subagent_type,
                 "runtime": runtime,
-            }
+            },
+            config=getattr(runtime, "config", None),
+            tool_call_id=subagent_id,
         )
+    except GraphInterrupt:
+        raise
     except Exception as e:
         error_event: SubagentErrorEvent = {
             "type": SUBAGENT_STREAM_EVENT_TYPE,

libs/partners/quickjs/tests/unit_tests/test_repl_middleware.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import threading
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -22,6 +22,8 @@ from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
 from langchain_core.messages import AIMessage, SystemMessage
 from langchain_core.runnables import RunnableLambda
 from langchain_core.tools import BaseTool, StructuredTool
+from langgraph.errors import GraphInterrupt
+from langgraph.types import Interrupt
 from pydantic import BaseModel, Field
 from quickjs_rs import Runtime, ThreadWorker
 
@@ -31,6 +33,7 @@ from langchain_quickjs._repl import _clear_exception_references, _Registry, _Thr
 from langchain_quickjs._subagent import (
     _ensure_schema_title,
     _runtime_with_response_format,
+    call_subagent_task_tool,
 )
 
 if TYPE_CHECKING:
@@ -872,6 +875,74 @@ def _subagent_runtime(
     return _subagent_runtime_from_task_tool(_task_tool_for_runnable(runnable))
 
 
+async def test_call_subagent_task_tool_forwards_config_and_tool_call_id() -> None:
+    calls: list[dict[str, Any]] = []
+
+    class _TaskTool:
+        name = "task"
+
+        async def arun(
+            self,
+            tool_input: dict[str, Any],
+            *,
+            config: dict[str, Any] | None = None,
+            tool_call_id: str | None = None,
+        ) -> str:
+            calls.append(
+                {
+                    "tool_input": tool_input,
+                    "config": config,
+                    "tool_call_id": tool_call_id,
+                }
+            )
+            return "ok"
+
+    runtime = ToolRuntime(
+        state={},
+        context={},
+        config={"configurable": {"thread_id": "parent-thread"}},
+        stream_writer=lambda _chunk: None,
+        tools=[],
+        tool_call_id="outer_eval_call",
+        store=None,
+    )
+
+    result = await call_subagent_task_tool(
+        cast("BaseTool", _TaskTool()),
+        description="work",
+        subagent_type="worker",
+        response_schema=None,
+        runtime=runtime,
+    )
+
+    assert result == "ok"
+    assert calls
+    assert calls[0]["config"] == runtime.config
+    assert calls[0]["tool_call_id"].startswith("ptc_task_")
+    assert calls[0]["tool_input"]["runtime"].tool_call_id == calls[0]["tool_call_id"]
+
+
+async def test_async_task_global_propagates_graph_interrupt(repl: _ThreadREPL) -> None:
+    interrupt = GraphInterrupt([Interrupt(value={"action_requests": []})])
+
+    async def _async(state: dict[str, Any], config: Any) -> dict[str, Any]:
+        del state, config
+        raise interrupt
+
+    runnable = RunnableLambda(
+        lambda state, config: {"messages": [AIMessage(content="sync")]},
+        afunc=_async,
+    )
+
+    with pytest.raises(GraphInterrupt) as exc_info:
+        await repl.eval_async(
+            "await task({description: 'work', subagentType: 'worker'})",
+            outer_runtime=_subagent_runtime(runnable),
+        )
+
+    assert exc_info.value is interrupt
+
+
 def test_runtime_with_response_format_uses_configurable() -> None:
     runtime = _subagent_runtime(
         RunnableLambda(lambda _state, _config: {"messages": [AIMessage(content="ok")]})