feat(quickjs): add swarm task tool (#3472)

Port of https://github.com/langchain-ai/deepagentsjs/pull/500

libs/deepagents/deepagents/middleware/skills.py

@@ -350,7 +350,20 @@ def _validate_skill_name(name: str, directory_name: str) -> tuple[bool, str]:
     return True, ""
 
 
-def _parse_skill_metadata(  # noqa: C901
+def _parse_allowed_tools(raw_tools: object, skill_path: str) -> list[str]:
+    """Parse the `allowed-tools` frontmatter value into a list of tool names."""
+    if isinstance(raw_tools, str):
+        return [t.strip(",").strip() for t in raw_tools.split() if t.strip(",").strip()]
+    if raw_tools is not None:
+        logger.warning(
+            "Ignoring non-string 'allowed-tools' in %s (got %s)",
+            skill_path,
+            type(raw_tools).__name__,
+        )
+    return []
+
+
+def _parse_skill_metadata(
     content: str,
     skill_path: str,
     directory_name: str,
@@ -419,21 +432,7 @@ def _parse_skill_metadata(  # noqa: C901
         )
         description_str = description_str[:MAX_SKILL_DESCRIPTION_LENGTH]
 
-    raw_tools = frontmatter_data.get("allowed-tools")
-    if isinstance(raw_tools, str):
-        allowed_tools = [
-            t.strip(",")  # Support commas for compatibility with skills created for Claude Code.
-            for t in raw_tools.split()
-            if t.strip(",")
-        ]
-    else:
-        if raw_tools is not None:
-            logger.warning(
-                "Ignoring non-string 'allowed-tools' in %s (got %s)",
-                skill_path,
-                type(raw_tools).__name__,
-            )
-        allowed_tools = []
+    allowed_tools = _parse_allowed_tools(frontmatter_data.get("allowed-tools"), skill_path)
 
     compatibility_str = str(frontmatter_data.get("compatibility", "")).strip() or None
     if compatibility_str and len(compatibility_str) > MAX_SKILL_COMPATIBILITY_LENGTH:

libs/deepagents/tests/unit_tests/middleware/test_skills_middleware.py

@@ -2102,3 +2102,41 @@ def test_modify_request_uses_custom_template() -> None:
     assert "WARN:" in appended
     assert "LIST:" in appended
     assert "## Skills System" not in appended  # default template marker absent
+
+
+# --- required-ptc-tools stays in metadata ----------------------------------
+
+
+def test_parse_skill_metadata_required_ptc_tools_in_metadata() -> None:
+    """required-ptc-tools is preserved in the metadata dict, not promoted."""
+    content = """---
+name: test-skill
+description: A test skill
+metadata:
+  required-ptc-tools: swarm_task read_file write_file glob
+---
+
+Content
+"""
+
+    result = _parse_skill_metadata(content, "/skills/test-skill/SKILL.md", "test-skill")
+    assert result is not None
+    assert "required_ptc_tools" not in result
+    assert result["metadata"]["required-ptc-tools"] == "swarm_task read_file write_file glob"
+
+
+def test_parse_skill_metadata_entrypoint_stays_in_metadata() -> None:
+    """metadata.entrypoint is preserved in metadata, not promoted to a top-level field."""
+    content = """---
+name: swarm
+description: Dispatch tasks in parallel
+metadata:
+  entrypoint: scripts/index.ts
+---
+
+# Swarm
+"""
+    result = _parse_skill_metadata(content, "/skills/swarm/SKILL.md", "swarm")
+    assert result is not None
+    assert "module" not in result
+    assert result["metadata"]["entrypoint"] == "scripts/index.ts"

libs/partners/quickjs/langchain_quickjs/__init__.py

@@ -1,6 +1,19 @@
 """langchain-quickjs: persistent JS REPL middleware for agents."""
 
 from langchain_quickjs._ptc import PTCOption
+from langchain_quickjs._swarm_task import (
+    SwarmSubAgent,
+    SwarmTaskMode,
+    VariantCache,
+    create_swarm_task_tool,
+)
 from langchain_quickjs.middleware import CodeInterpreterMiddleware
 
-__all__ = ["CodeInterpreterMiddleware", "PTCOption"]
+__all__ = [
+    "CodeInterpreterMiddleware",
+    "PTCOption",
+    "SwarmSubAgent",
+    "SwarmTaskMode",
+    "VariantCache",
+    "create_swarm_task_tool",
+]

libs/partners/quickjs/langchain_quickjs/_skills.py

@@ -1,14 +1,14 @@
 """Skill module loader for the REPL.
 
-Turns a skill's ``SkillMetadata`` (parsed by ``SkillsMiddleware``) plus a
-``BackendProtocol`` into a ``ModuleScope`` installable on a QuickJS
-context under the bare specifier ``@/skills/<name>``. The guest can
-then ``await import("@/skills/<name>")`` to pull the skill's entrypoint
+Turns a skill's `SkillMetadata` (parsed by `SkillsMiddleware`) plus a
+`BackendProtocol` into a `ModuleScope` installable on a QuickJS
+context under the bare specifier `@/skills/<name>`. The guest can
+then `await import("@/skills/<name>")` to pull the skill's entrypoint
 exports.
 
 This module is the enumeration + scope-build half; the install-cache
-and the pre-eval specifier scan live on ``_ThreadREPL`` in
-``_repl.py`` so they share the Context / Runtime locks.
+and the pre-eval specifier scan live on `_ThreadREPL` in
+`_repl.py` so they share the Context / Runtime locks.
 """
 
 from __future__ import annotations
@@ -66,7 +66,7 @@ class InvalidSkillScopeError(SkillLoadError):
     """Skill directory contains nothing installable.
 
     Either no module-extension files were found, or the frontmatter
-    ``module`` path doesn't match any of them.
+    `module` path doesn't match any of them.
     """
 
 
@@ -81,10 +81,10 @@ class LoadedSkill:
     Attributes:
         name: Spec-validated skill name (kebab-case).
         specifier: The bare specifier we install under. Always
-            ``"@/skills/<name>"``.
-        scope: A ``ModuleScope`` carrying every code file from the
+            `"@/skills/<name>"`.
+        scope: A `ModuleScope` carrying every code file from the
             skill directory, with the entrypoint renamed to
-            ``index.<ext>`` if the author picked a different name.
+            `index.<ext>` if the author picked a different name.
     """
 
     name: str
@@ -107,9 +107,9 @@ def _skill_specifier(name: str) -> str:
 
 
 def _skill_dir_from_metadata(metadata: SkillMetadata) -> str:
-    """Extract the skill directory from a ``SkillMetadata``.
+    """Extract the skill directory from a `SkillMetadata`.
 
-    ``path`` is the SKILL.md path; the skill dir is its parent.
+    `path` is the SKILL.md path; the skill dir is its parent.
     """
     return str(PurePosixPath(metadata["path"]).parent)
 
@@ -118,11 +118,11 @@ def _enumerate_code_files(
     backend: BackendProtocol,
     skill_dir: str,
 ) -> list[str]:
-    """List every code-extension file under ``skill_dir`` (recursive).
+    """List every code-extension file under `skill_dir` (recursive).
 
-    Makes one ``glob`` call per extension. We'd prefer one call with a
+    Makes one `glob` call per extension. We'd prefer one call with a
     brace-expansion pattern, but the backend protocol doesn't require
-    brace support and ``FilesystemBackend.glob`` uses ``pathlib.rglob``
+    brace support and `FilesystemBackend.glob` uses `pathlib.rglob`
     which doesn't expand them. Per-extension calls sidestep that cross-
     backend inconsistency. Paths are deduped (a single file can only
     match one extension) and sorted for determinism so the model's
@@ -216,7 +216,7 @@ def _build_scope_modules(
     file_pairs: list[tuple[str, bytes]],
     skill_name: str,
 ) -> dict[str, str | ModuleScope]:
-    """Build the ``ModuleScope`` contents dict for one skill.
+    """Build the `ModuleScope` contents dict for one skill.
 
     The dict has only `str` entries — no nested subscopes. That's the
     isolation guarantee from the spec: a skill scope can see only its
@@ -230,6 +230,14 @@ def _build_scope_modules(
     installed_index = _pick_installed_index_name(entry_rel)
     entry_installed = False
 
+    # When the entrypoint lives in a subdirectory (e.g. "scripts/index.ts"),
+    # it gets flattened to root-level "index.<ext>" for quickjs-rs bare-
+    # specifier resolution.  All sibling files must be relocated the same
+    # way so that relative imports (e.g. `import "./table.js"` from the
+    # now-root-level index) still resolve.
+    entry_parent = str(PurePosixPath(entry_rel).parent)
+    strip_prefix = entry_parent + "/" if entry_parent != "." else ""
+
     for abs_path, raw in file_pairs:
         rel = _relative(skill_dir, abs_path)
         try:
@@ -239,19 +247,20 @@ def _build_scope_modules(
             raise SkillInstallError(msg) from exc
 
         if rel == entry_rel:
-            # Always install the entrypoint under the canonical
-            # `index.<ext>` key. If the author already named it
-            # `index.<ext>` at the root, this is a no-op rename.
             files[installed_index] = source
             entry_installed = True
-            # Also expose the file at its original key so relative
-            # imports (e.g. `import "./index.ts"` from inside a sub-
-            # directory) still resolve. Skipped when entry_rel already
-            # equals installed_index.
             if rel != installed_index:
                 files[rel] = source
         else:
-            files[rel] = source
+            relocated = rel.removeprefix(strip_prefix) if strip_prefix else rel
+            if relocated.startswith(("/", "../")) or "/../" in relocated:
+                msg = (
+                    f"skill {skill_name!r}: relocated path {relocated!r} escapes scope"
+                )
+                raise SkillInstallError(msg)
+            files[relocated] = source
+            if relocated != rel:
+                files[rel] = source
 
     if not entry_installed:
         msg = (
@@ -259,14 +268,57 @@ def _build_scope_modules(
             "did not match any file in the skill directory"
         )
         raise InvalidSkillScopeError(msg)
+
+    _rewrite_js_imports_to_ts(files)
     return files
 
 
+# Matches relative import specifiers ending in .js:
+#   from "./table.js"  /  from '../lib/utils.js'  /  import("./foo.js")
+_JS_IMPORT_RE = re.compile(
+    r"""((?:from\s+|import\s*\()\s*["'])(\.\.?/[^"']*?)\.js(["'])"""
+)
+
+
+def _rewrite_js_imports_to_ts(files: dict[str, str | ModuleScope]) -> None:
+    """Rewrite `.js` import specifiers to `.ts` when only the `.ts` key exists.
+
+    TypeScript convention uses `.js` extensions in import specifiers even
+    when the source files are `.ts`. Quickjs-rs does exact key matching,
+    so `import "./table.js"` won't find `table.ts`. This rewrites
+    specifiers in-place for cases where the `.js` key is missing but the
+    `.ts` key is present.
+    """
+    all_keys = set(files)
+
+    for key in list(files):
+        source = files[key]
+        if not isinstance(source, str):
+            continue
+
+        def _replace(
+            m: re.Match[str], _dir: str = str(PurePosixPath(key).parent)
+        ) -> str:
+            prefix, rel_stem, suffix = m.group(1), m.group(2), m.group(3)
+            if _dir == ".":
+                resolved = rel_stem.lstrip("./") + ".js"
+            else:
+                resolved = str(PurePosixPath(_dir) / (rel_stem + ".js"))
+            ts_resolved = resolved[:-3] + ".ts"
+            if resolved not in all_keys and ts_resolved in all_keys:
+                return f"{prefix}{rel_stem}.ts{suffix}"
+            return m.group(0)
+
+        new_source = _JS_IMPORT_RE.sub(_replace, source)
+        if new_source is not source:
+            files[key] = new_source
+
+
 def load_skill(
     metadata: SkillMetadata,
     backend: BackendProtocol,
 ) -> LoadedSkill:
-    """Load one skill into a ``LoadedSkill``.
+    """Load one skill into a `LoadedSkill`.
 
     Raises:
         InvalidSkillScopeError: Metadata has no `entrypoint`, or the
@@ -354,9 +406,9 @@ _SKILL_SPECIFIER_RE = re.compile(
 def scan_skill_references(source: str) -> frozenset[str]:
     """Return the set of skill names the source imports from.
 
-    Extracts every literal ``"@/skills/<name>"`` specifier the source
+    Extracts every literal `"@/skills/<name>"` specifier the source
     contains. The caller is responsible for rejecting unknown names
-    with a ``SkillNotAvailable``-style error — this is a scan, not a
+    with a `SkillNotAvailable`-style error — this is a scan, not a
     validator.
 
     A returned name is not proof the skill exists or installs cleanly.

libs/partners/quickjs/langchain_quickjs/_swarm_task.py

@@ -0,0 +1,340 @@
+from __future__ import annotations
+
+import json
+import logging
+import time
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Any, Literal
+
+from deepagents._models import resolve_model
+from langchain.agents import create_agent
+from langchain_core.language_models import BaseChatModel  # noqa: TC002
+from langchain_core.messages import AIMessage, HumanMessage
+from langchain_core.tools import BaseTool, StructuredTool
+from pydantic import BaseModel, Field
+
+if TYPE_CHECKING:
+    from langchain_core.runnables import Runnable
+
+logger = logging.getLogger(__name__)
+
+SwarmTaskMode = Literal["agent", "invoke"]
+
+_VARIANT_TTL_S = 60.0
+_VARIANT_MAX_ENTRIES = 64
+
+_SCHEMA_MAX_BYTES = 4096
+_SCHEMA_MAX_DEPTH = 5
+_SCHEMA_MAX_PROPERTIES = 32
+
+
+@dataclass
+class SwarmSubAgent:
+    """Subagent specification for swarm dispatch targets."""
+
+    name: str
+    """Identifier used to select this subagent in swarm dispatch calls."""
+
+    description: str
+    """Human-readable description of what this subagent does."""
+
+    system_prompt: str
+    """System prompt injected at the start of the subagent's conversation."""
+
+    tools: list[BaseTool] = field(default_factory=list)
+    """Tools available to this subagent."""
+
+    model: str | BaseChatModel | None = None
+    """Model override for this subagent. Falls back to the tool's `default_model`."""
+
+
+def _validate_response_schema(schema: dict[str, Any]) -> None:
+    """Reject schemas that exceed size, depth, or property-count limits."""
+    serialized = json.dumps(schema)
+    if len(serialized) > _SCHEMA_MAX_BYTES:
+        msg = (
+            f"response_schema exceeds {_SCHEMA_MAX_BYTES}"
+            f" byte limit ({len(serialized)} bytes)"
+        )
+        raise ValueError(msg)
+
+    def _check(node: dict[str, Any], depth: int, prop_count: list[int]) -> None:
+        if depth > _SCHEMA_MAX_DEPTH:
+            msg = (
+                f"response_schema exceeds maximum nesting depth of {_SCHEMA_MAX_DEPTH}"
+            )
+            raise ValueError(msg)
+        props = node.get("properties")
+        if isinstance(props, dict):
+            prop_count[0] += len(props)
+            if prop_count[0] > _SCHEMA_MAX_PROPERTIES:
+                msg = (
+                    "response_schema exceeds maximum of"
+                    f" {_SCHEMA_MAX_PROPERTIES} properties"
+                )
+                raise ValueError(msg)
+            for v in props.values():
+                if isinstance(v, dict):
+                    _check(v, depth + 1, prop_count)
+        items = node.get("items")
+        if isinstance(items, dict):
+            _check(items, depth + 1, prop_count)
+
+    _check(schema, 0, [0])
+
+
+class VariantCache:
+    """TTL cache for compiled agent variants.
+
+    Stores values keyed by string with a last-accessed timestamp. On every
+    `get_or_create` call, entries that haven't been accessed within `ttl_s`
+    are swept first. Cache hits refresh the timestamp.
+    """
+
+    def __init__(
+        self,
+        ttl_s: float = _VARIANT_TTL_S,
+        max_entries: int = _VARIANT_MAX_ENTRIES,
+    ) -> None:
+        self._entries: dict[str, tuple[Any, float]] = {}
+        self._ttl_s = ttl_s
+        self._max_entries = max_entries
+
+    def get_or_create(self, key: str, factory: Any) -> Any:
+        """Return a cached value or create one via `factory` on cache miss."""
+        self._sweep()
+        entry = self._entries.get(key)
+        if entry is not None:
+            value, _ = entry
+            self._entries[key] = (value, time.monotonic())
+            return value
+        if len(self._entries) >= self._max_entries:
+            self._evict_lru()
+        value = factory()
+        self._entries[key] = (value, time.monotonic())
+        return value
+
+    @property
+    def size(self) -> int:
+        return len(self._entries)
+
+    def _sweep(self) -> None:
+        now = time.monotonic()
+        expired = [k for k, (_, ts) in self._entries.items() if now - ts > self._ttl_s]
+        for k in expired:
+            del self._entries[k]
+
+    def _evict_lru(self) -> None:
+        if not self._entries:
+            return
+        oldest_key = min(self._entries, key=lambda k: self._entries[k][1])
+        del self._entries[oldest_key]
+
+
+async def _invoke_model(
+    model: str | BaseChatModel,
+    description: str,
+    response_schema: dict[str, Any] | None,
+) -> str:
+    """Direct model invocation — single LLM call with optional structured output."""
+    resolved = resolve_model(model) if isinstance(model, str) else model
+    messages = [HumanMessage(content=description)]
+
+    if response_schema is not None:
+        return await _invoke_with_structured_output(resolved, messages, response_schema)
+
+    result = await resolved.ainvoke(messages)
+    if isinstance(result, str):
+        return result
+    if isinstance(result, AIMessage):
+        return str(result.text)
+    return json.dumps(result)
+
+
+async def _invoke_with_structured_output(
+    model: BaseChatModel,
+    messages: list[HumanMessage],
+    response_schema: dict[str, Any],
+) -> str:
+    """Use the model's structured output support to return validated JSON."""
+    schema = response_schema
+    if "title" not in schema:
+        schema = {**schema, "title": "structured_output"}
+    structured_model = model.with_structured_output(schema)
+    result = await structured_model.ainvoke(messages)
+    return json.dumps(result)
+
+
+class _AgentSpec:
+    """Minimal agent creation parameters preserved for recompilation."""
+
+    def __init__(
+        self,
+        *,
+        model: str | BaseChatModel,
+        system_prompt: str,
+        tools: list[BaseTool],
+        name: str,
+    ) -> None:
+        self.model = model
+        self.system_prompt = system_prompt
+        self.tools = tools
+        self.name = name
+
+
+class _CompiledAgent:
+    """Compiled agent alongside its creation spec."""
+
+    def __init__(self, agent: Runnable, spec: _AgentSpec) -> None:
+        self.agent = agent
+        self.spec = spec
+
+
+async def _invoke_agent(
+    entry: _CompiledAgent,
+    description: str,
+    response_schema: dict[str, Any] | None,
+    variant_cache: VariantCache,
+) -> str:
+    """Full agentic invocation with optional schema-constrained variant caching."""
+    agent = entry.agent
+
+    if response_schema is not None:
+        cache_key = f"{entry.spec.name}::{json.dumps(response_schema, sort_keys=True)}"
+        agent = variant_cache.get_or_create(
+            cache_key,
+            lambda: create_agent(
+                model=entry.spec.model,
+                system_prompt=entry.spec.system_prompt,
+                tools=entry.spec.tools,
+                name=entry.spec.name,
+                response_format=response_schema,
+            ),
+        )
+
+    state = {"messages": [HumanMessage(content=description)]}
+    result = await agent.ainvoke(state)
+
+    if isinstance(result, dict):
+        structured = result.get("structured_response")
+        if structured is not None:
+            return json.dumps(structured)
+
+        messages = result.get("messages", [])
+        last = messages[-1] if messages else None
+        if last is None:
+            return "Task completed"
+        return str(last.text) if hasattr(last, "text") else str(last)
+
+    return str(result)
+
+
+class _SwarmTaskInput(BaseModel):
+    """Input schema for the swarm_task tool."""
+
+    description: str = Field(
+        description="The task to execute with the selected subagent."
+    )
+    subagent_type: str | None = Field(
+        default=None,
+        description="Name of the swarm subagent to use.",
+    )
+    response_schema: dict[str, Any] | None = Field(
+        default=None,
+        description="JSON Schema (type: 'object') for structured output.",
+    )
+    mode: SwarmTaskMode | None = Field(
+        default=None,
+        description=(
+            'Dispatch mode. "agent" (default) runs a full'
+            ' agentic loop. "invoke" makes a single model call.'
+        ),
+    )
+
+
+def create_swarm_task_tool(
+    *,
+    subagents: list[SwarmSubAgent] | None = None,
+    default_model: str | BaseChatModel,
+) -> BaseTool:
+    """Create a PTC-only tool for swarm subagent dispatch.
+
+    The returned tool is designed to be passed into the CodeInterpreterMiddleware's
+    `ptc` list. It is never exposed to the LLM — only callable from QuickJS skill
+    code via `tools.swarmTask()`.
+
+    Supports two dispatch modes:
+
+    - `agent` (default): Full agentic loop with tools. Schema-constrained
+      agents are cached with a TTL.
+    - `invoke`: Direct model call with structured output. No tools, no iteration.
+
+    Args:
+        subagents: Subagent specifications for swarm dispatch targets.
+        default_model: Default model used for subagents that don't specify their own,
+            and for `invoke` mode direct model calls.
+
+    Returns:
+        A `BaseTool` suitable for the `ptc` config.
+    """
+    subs = subagents or []
+    compiled: dict[str, _CompiledAgent] = {}
+
+    for sub in subs:
+        model = sub.model if sub.model is not None else default_model
+        spec = _AgentSpec(
+            model=model,
+            system_prompt=sub.system_prompt,
+            tools=sub.tools,
+            name=sub.name,
+        )
+        agent = create_agent(
+            model=model,
+            system_prompt=sub.system_prompt,
+            tools=sub.tools,
+            name=sub.name,
+        )
+        compiled[sub.name] = _CompiledAgent(agent=agent, spec=spec)
+
+    subagent_names = [s.name for s in subs]
+    variant_cache = VariantCache()
+
+    async def _run(
+        description: str,
+        subagent_type: str | None = None,
+        response_schema: dict[str, Any] | None = None,
+        mode: SwarmTaskMode | None = None,
+    ) -> str:
+        if response_schema is not None:
+            _validate_response_schema(response_schema)
+        effective_mode = mode or "agent"
+
+        if effective_mode == "invoke":
+            return await _invoke_model(default_model, description, response_schema)
+
+        if subagent_type is None:
+            available = (
+                ", ".join(subagent_names) if subagent_names else "(none configured)"
+            )
+            msg = f"agent mode requires subagent_type. Available: {available}"
+            raise ValueError(msg)
+
+        entry = compiled.get(subagent_type)
+        if entry is None:
+            available = ", ".join(subagent_names)
+            msg = (
+                f'Unknown swarm subagent type "{subagent_type}". Available: {available}'
+            )
+            raise ValueError(msg)
+
+        return await _invoke_agent(entry, description, response_schema, variant_cache)
+
+    return StructuredTool.from_function(
+        name="swarm_task",
+        description=(
+            "Dispatch a task to a swarm subagent. Supports agent mode "
+            "(full agentic loop) and invoke mode (direct model call)."
+        ),
+        coroutine=_run,
+        args_schema=_SwarmTaskInput,
+    )

libs/partners/quickjs/langchain_quickjs/middleware.py

@@ -299,6 +299,16 @@ class CodeInterpreterMiddleware(AgentMiddleware[REPLState, ContextT, ResponseT])
             metadata={"ls_code_input_language": "javascript"},
         )
 
+    def _ptc_tool_names(self) -> set[str]:
+        """Collect tool names from the PTC configuration."""
+        names: set[str] = set()
+        for entry in self._ptc or []:
+            if isinstance(entry, str):
+                names.add(entry)
+            elif isinstance(entry, BaseTool):
+                names.add(entry.name)
+        return names
+
     def _skills_for_eval(
         self,
         runtime: ToolRuntime[None, Any],
@@ -309,7 +319,40 @@ class CodeInterpreterMiddleware(AgentMiddleware[REPLState, ContextT, ResponseT])
         metadata_list = (
             runtime.state.get("skills_metadata", []) if runtime.state else []
         )
-        return {m["name"]: m for m in metadata_list}
+        ptc_names = self._ptc_tool_names()
+        result: dict[str, SkillMetadata] = {}
+        for m in metadata_list:
+            raw = m.get("metadata", {}).get("required-ptc-tools", "")
+            required = str(raw).split() if raw else []
+            missing = [t for t in required if t not in ptc_names]
+            if missing:
+                logger.warning(
+                    "Skill '%s' requires PTC tools not in ptc config: %s",
+                    m["name"],
+                    ", ".join(missing),
+                )
+                continue
+            result[m["name"]] = m
+        return result
+
+    def _validate_required_ptc_tools(self, state: REPLState) -> None:
+        """Raise if any skill requires PTC tools not in the config."""
+        if self._skills_backend is None:
+            return
+        metadata_list: list[SkillMetadata] = state.get("skills_metadata", [])  # type: ignore[assignment]
+        ptc_names = self._ptc_tool_names()
+        for skill in metadata_list:
+            raw = skill.get("metadata", {}).get("required-ptc-tools", "")
+            required = str(raw).split() if raw else []
+            missing = [t for t in required if t not in ptc_names]
+            if missing:
+                msg = (
+                    f"Skill '{skill['name']}' requires PTC tools"
+                    " that are not configured: "
+                    f"{', '.join(missing)}. "
+                    f"Add them to CodeInterpreterMiddleware(ptc=[...])."
+                )
+                raise ValueError(msg)
 
     def before_agent(
         self,
@@ -317,6 +360,7 @@ class CodeInterpreterMiddleware(AgentMiddleware[REPLState, ContextT, ResponseT])
         runtime: "Runtime[ContextT]",  # noqa: ARG002
     ) -> dict[str, Any] | None:
         """Restore REPL snapshot bytes into the current thread slot."""
+        self._validate_required_ptc_tools(state)
         if not self._snapshot_between_turns:
             return None
         payload = state.get("_quickjs_snapshot_payload")
@@ -341,6 +385,7 @@ class CodeInterpreterMiddleware(AgentMiddleware[REPLState, ContextT, ResponseT])
         runtime: "Runtime[ContextT]",  # noqa: ARG002
     ) -> dict[str, Any] | None:
         """Async variant of `before_agent` snapshot restore."""
+        self._validate_required_ptc_tools(state)
         if not self._snapshot_between_turns:
             return None
         payload = state.get("_quickjs_snapshot_payload")

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

@@ -769,3 +769,222 @@ def test_sync_path_still_works(repl: _ThreadREPL) -> None:
     """After the v0.2 split, the sync path continues to use ``ctx.eval``."""
     repl.eval_sync("let n = 7")
     assert repl.eval_sync("n * 6").result == "42"
+
+
+# ---------------------------------------------------------------------------
+# required_ptc_tools validation
+# ---------------------------------------------------------------------------
+
+
+def _make_skill_metadata(
+    name: str,
+    *,
+    required_ptc_tools: list[str] | None = None,
+) -> dict[str, Any]:
+    """Build a minimal SkillMetadata dict for testing."""
+    inner_metadata: dict[str, str] = {}
+    if required_ptc_tools is not None:
+        inner_metadata["required-ptc-tools"] = " ".join(required_ptc_tools)
+    return {
+        "name": name,
+        "description": "test",
+        "path": f"/skills/{name}/SKILL.md",
+        "metadata": inner_metadata,
+        "license": None,
+        "compatibility": None,
+        "allowed_tools": [],
+    }
+
+
+def _make_ptc_tool(name: str) -> StructuredTool:
+    """Create a minimal StructuredTool for PTC config."""
+    return StructuredTool.from_function(
+        name=name,
+        description=f"{name} tool",
+        func=lambda: "ok",
+    )
+
+
+def test_before_agent_raises_when_required_ptc_tools_missing() -> None:
+    """``before_agent`` raises when a skill needs PTC tools not in config."""
+    mw = CodeInterpreterMiddleware(
+        skills_backend=MagicMock(),
+        ptc=[_make_ptc_tool("read_file")],
+    )
+    try:
+        state = {
+            "skills_metadata": [
+                _make_skill_metadata(
+                    "swarm",
+                    required_ptc_tools=["swarm_task", "read_file", "write_file"],
+                ),
+            ],
+        }
+        with pytest.raises(ValueError, match="swarm_task"):
+            mw.before_agent(state=state, runtime=MagicMock())  # type: ignore[arg-type]
+    finally:
+        mw._registry.close()
+
+
+async def test_abefore_agent_raises_when_required_ptc_tools_missing() -> None:
+    """``abefore_agent`` raises when a skill needs PTC tools not in config."""
+    mw = CodeInterpreterMiddleware(
+        skills_backend=MagicMock(),
+        ptc=[_make_ptc_tool("read_file")],
+    )
+    try:
+        state = {
+            "skills_metadata": [
+                _make_skill_metadata(
+                    "swarm", required_ptc_tools=["swarm_task", "read_file"]
+                ),
+            ],
+        }
+        with pytest.raises(ValueError, match="swarm_task"):
+            await mw.abefore_agent(state=state, runtime=MagicMock())  # type: ignore[arg-type]
+    finally:
+        mw._registry.close()
+
+
+def test_before_agent_passes_when_all_required_ptc_tools_present() -> None:
+    """``before_agent`` succeeds when all required PTC tools are configured."""
+    mw = CodeInterpreterMiddleware(
+        skills_backend=MagicMock(),
+        ptc=[
+            _make_ptc_tool("swarm_task"),
+            _make_ptc_tool("read_file"),
+            _make_ptc_tool("write_file"),
+        ],
+    )
+    try:
+        state = {
+            "skills_metadata": [
+                _make_skill_metadata(
+                    "swarm", required_ptc_tools=["swarm_task", "read_file"]
+                ),
+            ],
+        }
+        result = mw.before_agent(state=state, runtime=MagicMock())  # type: ignore[arg-type]
+        assert result is None
+    finally:
+        mw._registry.close()
+
+
+def test_before_agent_passes_when_no_required_ptc_tools() -> None:
+    """``before_agent`` succeeds when skills have no required_ptc_tools."""
+    mw = CodeInterpreterMiddleware(skills_backend=MagicMock())
+    try:
+        state = {
+            "skills_metadata": [_make_skill_metadata("simple-skill")],
+        }
+        result = mw.before_agent(state=state, runtime=MagicMock())  # type: ignore[arg-type]
+        assert result is None
+    finally:
+        mw._registry.close()
+
+
+def test_before_agent_skips_validation_when_no_skills_backend() -> None:
+    """``before_agent`` skips PTC validation when ``skills_backend`` is not set."""
+    mw = CodeInterpreterMiddleware()
+    try:
+        state = {
+            "skills_metadata": [
+                _make_skill_metadata("swarm", required_ptc_tools=["swarm_task"]),
+            ],
+        }
+        result = mw.before_agent(state=state, runtime=MagicMock())  # type: ignore[arg-type]
+        assert result is None
+    finally:
+        mw._registry.close()
+
+
+def test_before_agent_ptc_validation_accepts_string_entries() -> None:
+    """``_ptc_tool_names`` collects names from string entries in PTC config."""
+    mw = CodeInterpreterMiddleware(
+        skills_backend=MagicMock(),
+        ptc=["swarm_task", "read_file"],
+    )
+    try:
+        state = {
+            "skills_metadata": [
+                _make_skill_metadata(
+                    "swarm", required_ptc_tools=["swarm_task", "read_file"]
+                ),
+            ],
+        }
+        result = mw.before_agent(state=state, runtime=MagicMock())  # type: ignore[arg-type]
+        assert result is None
+    finally:
+        mw._registry.close()
+
+
+def test_before_agent_error_message_includes_all_missing_tools() -> None:
+    """Error message lists all missing tools, not just the first."""
+    mw = CodeInterpreterMiddleware(
+        skills_backend=MagicMock(),
+        ptc=[_make_ptc_tool("read_file")],
+    )
+    try:
+        state = {
+            "skills_metadata": [
+                _make_skill_metadata(
+                    "swarm",
+                    required_ptc_tools=["swarm_task", "write_file", "read_file"],
+                ),
+            ],
+        }
+        with pytest.raises(ValueError, match="swarm_task, write_file"):
+            mw.before_agent(state=state, runtime=MagicMock())  # type: ignore[arg-type]
+    finally:
+        mw._registry.close()
+
+
+def test_skills_for_eval_skips_skills_with_missing_ptc_tools(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """``_skills_for_eval`` filters skills with unsatisfied PTC requirements."""
+    mw = CodeInterpreterMiddleware(
+        skills_backend=MagicMock(),
+        ptc=[_make_ptc_tool("read_file")],
+    )
+    try:
+        runtime = MagicMock()
+        runtime.state = {
+            "skills_metadata": [
+                _make_skill_metadata(
+                    "needs-swarm", required_ptc_tools=["swarm_task", "read_file"]
+                ),
+                _make_skill_metadata("plain-skill"),
+            ],
+        }
+        with caplog.at_level("WARNING"):
+            result = mw._skills_for_eval(runtime)
+
+        assert result is not None
+        assert "needs-swarm" not in result
+        assert "plain-skill" in result
+        assert "swarm_task" in caplog.text
+    finally:
+        mw._registry.close()
+
+
+def test_skills_for_eval_includes_skills_when_all_ptc_tools_present() -> None:
+    """``_skills_for_eval`` includes skills when all PTC tools present."""
+    mw = CodeInterpreterMiddleware(
+        skills_backend=MagicMock(),
+        ptc=[_make_ptc_tool("swarm_task"), _make_ptc_tool("read_file")],
+    )
+    try:
+        runtime = MagicMock()
+        runtime.state = {
+            "skills_metadata": [
+                _make_skill_metadata(
+                    "swarm", required_ptc_tools=["swarm_task", "read_file"]
+                ),
+            ],
+        }
+        result = mw._skills_for_eval(runtime)
+        assert result is not None
+        assert "swarm" in result
+    finally:
+        mw._registry.close()

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

@@ -359,4 +359,36 @@ def test_loaded_scope_installs_on_context(tmp_path: Path) -> None:
         rt.install(ModuleScope({loaded.specifier: loaded.scope}))
 
 
+def test_load_skill_subdirectory_entrypoint_relocates_siblings(tmp_path: Path) -> None:
+    """When the entrypoint is in a subdirectory, sibling files are relocated
+    to the scope root so relative imports from the flattened index resolve."""
+    backend = FilesystemBackend(root_dir=str(tmp_path), virtual_mode=False)
+    skill_dir = str(tmp_path / "skills" / "swarm")
+    _write(
+        backend,
+        {
+            f"{skill_dir}/SKILL.md": "---\nname: swarm\ndescription: x\n---\n",
+            f"{skill_dir}/scripts/index.ts": 'import { create } from "./table.js";',
+            f"{skill_dir}/scripts/table.ts": "export function create() {}",
+            f"{skill_dir}/scripts/lib/utils.ts": "export function read() {}",
+        },
+    )
+    meta = _metadata(
+        "swarm",
+        path=f"{skill_dir}/SKILL.md",
+        entrypoint="scripts/index.ts",
+    )
+
+    loaded = load_skill(meta, backend)
+
+    keys = set(loaded.scope.modules.keys())
+    assert "index.ts" in keys
+    assert "table.ts" in keys
+    assert "lib/utils.ts" in keys
+    # Originals kept for backwards compatibility
+    assert "scripts/index.ts" in keys
+    assert "scripts/table.ts" in keys
+    assert "scripts/lib/utils.ts" in keys
+
+
 __all__: list[Any] = []

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

@@ -171,6 +171,71 @@ async def test_multi_file_skill_relative_import(
     assert after.result == "14"
 
 
+async def test_multi_file_skill_js_import_specifiers_resolve_to_ts(
+    registry: _Registry, tmp_path: Path
+) -> None:
+    """TS files imported via .js specifiers (Node convention) resolve correctly."""
+    backend = FilesystemBackend(root_dir=str(tmp_path), virtual_mode=False)
+    skill_dir = str(tmp_path / "skills" / "jsimport")
+    _write(
+        backend,
+        {
+            f"{skill_dir}/SKILL.md": "---\nname: jsimport\ndescription: x\n---\n",
+            f"{skill_dir}/index.ts": (
+                'import { value } from "./util.js";\n'
+                "export const doubled = value * 2;\n"
+            ),
+            f"{skill_dir}/util.ts": "export const value: number = 7;\n",
+        },
+    )
+    meta = _metadata("jsimport", path=f"{skill_dir}/SKILL.md", entrypoint="index.ts")
+
+    repl = registry.get("t1")
+    import_outcome = await repl.eval_async(
+        'const m = await import("@/skills/jsimport"); globalThis.r = m.doubled;',
+        skills={"jsimport": meta},
+        skills_backend=backend,
+    )
+    assert import_outcome.error_type is None, import_outcome.error_message
+    after = await repl.eval_async("globalThis.r")
+    assert after.result == "14"
+
+
+async def test_subdirectory_entrypoint_with_js_imports(
+    registry: _Registry, tmp_path: Path
+) -> None:
+    """Entrypoint in a subdirectory with .js import specifiers resolves correctly."""
+    backend = FilesystemBackend(root_dir=str(tmp_path), virtual_mode=False)
+    skill_dir = str(tmp_path / "skills" / "subdir")
+    _write(
+        backend,
+        {
+            f"{skill_dir}/SKILL.md": "---\nname: subdir\ndescription: x\n---\n",
+            f"{skill_dir}/scripts/index.ts": (
+                'import { compute } from "./math.js";\n'
+                "export const result = compute(3, 4);\n"
+            ),
+            f"{skill_dir}/scripts/math.ts": "export function compute("
+            "a: number, b: number): number { return a + b; }\n",
+        },
+    )
+    meta = _metadata(
+        "subdir",
+        path=f"{skill_dir}/SKILL.md",
+        entrypoint="scripts/index.ts",
+    )
+
+    repl = registry.get("t1")
+    import_outcome = await repl.eval_async(
+        'const m = await import("@/skills/subdir"); globalThis.r = m.result;',
+        skills={"subdir": meta},
+        skills_backend=backend,
+    )
+    assert import_outcome.error_type is None, import_outcome.error_message
+    after = await repl.eval_async("globalThis.r")
+    assert after.result == "7"
+
+
 async def test_install_cache_avoids_second_fetch(
     registry: _Registry, tmp_path: Path
 ) -> None:

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

@@ -0,0 +1,717 @@
+"""Unit tests for the swarm_task PTC tool.
+
+Covers VariantCache and the tool created by create_swarm_task_tool
+(subagent validation, agent mode dispatch, invoke mode,
+schema-constrained variant caching).
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Iterator  # noqa: TC003
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
+from langchain_core.messages import AIMessage, HumanMessage
+from langchain_core.runnables import RunnableLambda
+from pydantic import Field
+
+from langchain_quickjs._swarm_task import (
+    SwarmSubAgent,
+    VariantCache,
+    create_swarm_task_tool,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+class _FakeModel(GenericFakeChatModel):
+    """Fake model that passes Pydantic validation on SwarmSubAgent.model."""
+
+    messages: Iterator[AIMessage | str] = Field(
+        default_factory=lambda: iter([AIMessage(content="")]), exclude=True
+    )
+
+    def bind_tools(self, tools: Any, **_: Any) -> _FakeModel:
+        return self
+
+
+def _make_mock_model(response: str = "model response") -> MagicMock:
+    """Create a mock model that returns an AIMessage with the given content."""
+    model = MagicMock()
+    model.ainvoke = AsyncMock(return_value=AIMessage(content=response))
+    return model
+
+
+def _make_runnable(afunc: Any) -> RunnableLambda:
+    """Wrap an async function in a RunnableLambda (func arg is required)."""
+    return RunnableLambda(func=lambda _: None, afunc=afunc)
+
+
+def _make_fake_agent(
+    content: str = "agent response",
+    structured_response: Any = None,
+) -> RunnableLambda:
+    """Create a fake agent that returns a state dict with messages."""
+    result: dict[str, Any] = {
+        "messages": [AIMessage(content=content)],
+    }
+    if structured_response is not None:
+        result["structured_response"] = structured_response
+
+    async def _afunc(_: Any) -> dict[str, Any]:
+        return result
+
+    return _make_runnable(_afunc)
+
+
+# ---------------------------------------------------------------------------
+# VariantCache (isolated unit tests)
+# ---------------------------------------------------------------------------
+
+
+class TestVariantCache:
+    """Tests for the VariantCache class."""
+
+    def test_returns_factory_value_on_cache_miss(self) -> None:
+        cache = VariantCache(ttl_s=60.0)
+        value = cache.get_or_create("key1", lambda: "created")
+        assert value == "created"
+        assert cache.size == 1
+
+    def test_returns_cached_value_on_hit_without_calling_factory(self) -> None:
+        cache = VariantCache(ttl_s=60.0)
+        cache.get_or_create("key1", lambda: "first")
+
+        factory = MagicMock(return_value="second")
+        value = cache.get_or_create("key1", factory)
+
+        assert value == "first"
+        factory.assert_not_called()
+
+    def test_stores_separate_entries_for_different_keys(self) -> None:
+        cache = VariantCache(ttl_s=60.0)
+        cache.get_or_create("a", lambda: "alpha")
+        cache.get_or_create("b", lambda: "beta")
+
+        assert cache.size == 2
+        assert cache.get_or_create("a", lambda: "unused") == "alpha"
+        assert cache.get_or_create("b", lambda: "unused") == "beta"
+
+    def test_evicts_entries_that_exceed_ttl(self) -> None:
+        cache = VariantCache(ttl_s=1.0)
+
+        with patch("langchain_quickjs._swarm_task.time") as mock_time:
+            mock_time.monotonic = MagicMock(side_effect=[0.0, 0.0, 1.5, 1.5, 1.5])
+            cache.get_or_create("key1", lambda: "value1")
+
+            factory = MagicMock(return_value="value2")
+            value = cache.get_or_create("key1", factory)
+
+        assert value == "value2"
+        factory.assert_called_once()
+
+    def test_keeps_entries_alive_when_accessed_within_ttl(self) -> None:
+        cache = VariantCache(ttl_s=1.0)
+
+        with patch("langchain_quickjs._swarm_task.time") as mock_time:
+            mock_time.monotonic = MagicMock(
+                side_effect=[
+                    0.0,
+                    0.0,  # create key1 (sweep reads 0.0, set reads 0.0)
+                    0.8,
+                    0.8,
+                    0.8,  # access at 0.8 — not expired
+                    1.6,
+                    1.6,
+                    1.6,  # access at 1.6 — diff=0.8 < 1.0, alive
+                ]
+            )
+            cache.get_or_create("key1", lambda: "value1")
+            cache.get_or_create("key1", lambda: "unused")
+
+            factory = MagicMock(return_value="replaced")
+            value = cache.get_or_create("key1", factory)
+
+        assert value == "value1"
+        factory.assert_not_called()
+
+    def test_sweeps_multiple_expired_entries_in_one_call(self) -> None:
+        cache = VariantCache(ttl_s=1.0)
+
+        with patch("langchain_quickjs._swarm_task.time") as mock_time:
+            mock_time.monotonic = MagicMock(
+                side_effect=[
+                    0.0,
+                    0.0,  # create a
+                    0.0,
+                    0.0,  # create b
+                    0.0,
+                    0.0,  # create c
+                    1.5,
+                    1.5,  # create d (sweep evicts a, b, c)
+                ]
+            )
+            cache.get_or_create("a", lambda: "1")
+            cache.get_or_create("b", lambda: "2")
+            cache.get_or_create("c", lambda: "3")
+            assert cache.size == 3
+
+            cache.get_or_create("d", lambda: "4")
+
+        assert cache.size == 1
+
+    def test_only_evicts_expired_entries_not_active_ones(self) -> None:
+        cache = VariantCache(ttl_s=1.0)
+
+        with patch("langchain_quickjs._swarm_task.time") as mock_time:
+            mock_time.monotonic = MagicMock(
+                side_effect=[
+                    0.0,
+                    0.0,  # create "old": sweep(now=0.0) + set(0.0)
+                    0.8,
+                    0.8,  # create "new": old@0.0 diff<1.0, ok
+                    1.1,
+                    1.1,  # create "trigger": old evicted, new ok
+                ]
+            )
+            cache.get_or_create("old", lambda: "stale")
+            cache.get_or_create("new", lambda: "fresh")
+            cache.get_or_create("trigger", lambda: "sweep")
+
+        assert cache.size == 2
+
+
+# ---------------------------------------------------------------------------
+# Subagent validation
+# ---------------------------------------------------------------------------
+
+
+class TestSubagentValidation:
+    """Tests for subagent validation in create_swarm_task_tool."""
+
+    async def test_throws_when_subagent_type_not_in_configured_list(self) -> None:
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent",
+            return_value=_make_fake_agent(),
+        ):
+            tool = create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(
+                        name="screener",
+                        description="A screener",
+                        system_prompt="Screen.",
+                    ),
+                ],
+                default_model=_make_mock_model(),
+            )
+
+        with pytest.raises(
+            Exception, match='Unknown swarm subagent type "nonexistent"'
+        ):
+            await tool.ainvoke(
+                {"description": "do work", "subagent_type": "nonexistent"}
+            )
+
+    async def test_includes_available_subagent_names_in_error_message(self) -> None:
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent",
+            return_value=_make_fake_agent(),
+        ):
+            tool = create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(name="alpha", description="A", system_prompt="A."),
+                    SwarmSubAgent(name="beta", description="B", system_prompt="B."),
+                ],
+                default_model=_make_mock_model(),
+            )
+
+        with pytest.raises(Exception, match="alpha, beta"):
+            await tool.ainvoke({"description": "do work", "subagent_type": "gamma"})
+
+    async def test_agent_mode_requires_subagent_type(self) -> None:
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent",
+            return_value=_make_fake_agent(),
+        ):
+            tool = create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(name="worker", description="W", system_prompt="W."),
+                ],
+                default_model=_make_mock_model(),
+            )
+
+        with pytest.raises(Exception, match="agent mode requires subagent_type"):
+            await tool.ainvoke({"description": "do work"})
+
+    async def test_agent_mode_none_configured_message(self) -> None:
+        tool = create_swarm_task_tool(
+            subagents=[],
+            default_model=_make_mock_model(),
+        )
+
+        with pytest.raises(Exception, match=r"\(none configured\)"):
+            await tool.ainvoke({"description": "do work"})
+
+
+# ---------------------------------------------------------------------------
+# Agent mode
+# ---------------------------------------------------------------------------
+
+
+class TestAgentMode:
+    """Tests for agent dispatch mode (default)."""
+
+    async def test_dispatches_to_correct_subagent_by_name(self) -> None:
+        alpha_agent = _make_fake_agent("alpha result")
+        beta_agent = _make_fake_agent("beta result")
+
+        agents = [alpha_agent, beta_agent]
+        agent_idx = 0
+
+        def mock_create_agent(**_: Any) -> RunnableLambda:
+            nonlocal agent_idx
+            agent = agents[agent_idx]
+            agent_idx += 1
+            return agent
+
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent", side_effect=mock_create_agent
+        ):
+            tool = create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(name="alpha", description="A", system_prompt="A."),
+                    SwarmSubAgent(name="beta", description="B", system_prompt="B."),
+                ],
+                default_model=_make_mock_model(),
+            )
+
+        result = await tool.ainvoke({"description": "do work", "subagent_type": "beta"})
+        assert result == "beta result"
+
+    async def test_passes_description_as_human_message_content(self) -> None:
+        invoked_with: list[Any] = []
+
+        def capture_agent(**_: Any) -> RunnableLambda:
+            async def run(state: dict[str, Any]) -> dict[str, Any]:
+                invoked_with.append(state)
+                return {"messages": [AIMessage(content="done")]}
+
+            return _make_runnable(run)
+
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent", side_effect=capture_agent
+        ):
+            tool = create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(name="worker", description="W", system_prompt="W.")
+                ],
+                default_model=_make_mock_model(),
+            )
+
+        await tool.ainvoke(
+            {"description": "classify this trace", "subagent_type": "worker"}
+        )
+
+        state = invoked_with[-1]
+        messages = state["messages"]
+        assert len(messages) == 1
+        assert isinstance(messages[0], HumanMessage)
+        assert messages[0].content == "classify this trace"
+
+    async def test_returns_structured_response_as_json_when_present(self) -> None:
+        agent = _make_fake_agent(
+            content="ignored",
+            structured_response={"label": "positive", "score": 0.95},
+        )
+
+        with patch("langchain_quickjs._swarm_task.create_agent", return_value=agent):
+            tool = create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(name="worker", description="W", system_prompt="W.")
+                ],
+                default_model=_make_mock_model(),
+            )
+
+        result = await tool.ainvoke(
+            {"description": "classify", "subagent_type": "worker"}
+        )
+        assert json.loads(result) == {"label": "positive", "score": 0.95}
+
+    async def test_returns_last_message_content_when_no_structured_response(
+        self,
+    ) -> None:
+        async def run(_: Any) -> dict[str, Any]:
+            return {
+                "messages": [
+                    AIMessage(content="first"),
+                    AIMessage(content="last message"),
+                ],
+            }
+
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent",
+            return_value=_make_runnable(run),
+        ):
+            tool = create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(name="worker", description="W", system_prompt="W.")
+                ],
+                default_model=_make_mock_model(),
+            )
+
+        result = await tool.ainvoke({"description": "work", "subagent_type": "worker"})
+        assert result == "last message"
+
+    async def test_returns_task_completed_when_no_messages(self) -> None:
+        async def run(_: Any) -> dict[str, Any]:
+            return {"messages": []}
+
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent",
+            return_value=_make_runnable(run),
+        ):
+            tool = create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(name="worker", description="W", system_prompt="W.")
+                ],
+                default_model=_make_mock_model(),
+            )
+
+        result = await tool.ainvoke({"description": "work", "subagent_type": "worker"})
+        assert result == "Task completed"
+
+    async def test_compiles_new_agent_with_response_format_when_schema_provided(
+        self,
+    ) -> None:
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent",
+            return_value=_make_fake_agent(),
+        ) as mock_create:
+            tool = create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(
+                        name="worker", description="W", system_prompt="Analyze."
+                    )
+                ],
+                default_model=_make_mock_model(),
+            )
+
+            assert mock_create.call_count == 1
+
+            schema = {
+                "type": "object",
+                "properties": {"label": {"type": "string"}},
+                "required": ["label"],
+            }
+
+            await tool.ainvoke(
+                {
+                    "description": "classify",
+                    "subagent_type": "worker",
+                    "response_schema": schema,
+                }
+            )
+
+            assert mock_create.call_count == 2
+            last_call_kwargs = mock_create.call_args_list[-1]
+            assert last_call_kwargs.kwargs["response_format"] == schema
+
+    async def test_does_not_compile_new_agent_when_schema_omitted(self) -> None:
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent",
+            return_value=_make_fake_agent(),
+        ) as mock_create:
+            tool = create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(name="worker", description="W", system_prompt="W.")
+                ],
+                default_model=_make_mock_model(),
+            )
+            assert mock_create.call_count == 1
+
+            await tool.ainvoke({"description": "work", "subagent_type": "worker"})
+            assert mock_create.call_count == 1
+
+
+# ---------------------------------------------------------------------------
+# Invoke mode
+# ---------------------------------------------------------------------------
+
+
+class TestInvokeMode:
+    """Tests for invoke dispatch mode."""
+
+    async def test_calls_model_directly_with_human_message(self) -> None:
+        model = _make_mock_model("classified: positive")
+
+        tool = create_swarm_task_tool(subagents=[], default_model=model)
+
+        await tool.ainvoke({"description": "classify this", "mode": "invoke"})
+
+        model.ainvoke.assert_awaited_once()
+        messages = model.ainvoke.call_args[0][0]
+        assert len(messages) == 1
+        assert isinstance(messages[0], HumanMessage)
+        assert messages[0].content == "classify this"
+
+    async def test_does_not_call_create_agent_in_invoke_mode(self) -> None:
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent",
+            return_value=_make_fake_agent(),
+        ) as mock_create:
+            tool = create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(
+                        name="worker",
+                        description="W",
+                        system_prompt="W.",
+                        model=_FakeModel(),
+                    ),
+                ],
+                default_model=_make_mock_model(),
+            )
+            after_construction = mock_create.call_count
+
+            await tool.ainvoke({"description": "work", "mode": "invoke"})
+            assert mock_create.call_count == after_construction
+
+    async def test_uses_with_structured_output_when_schema_provided(
+        self,
+    ) -> None:
+        structured_result = {"label": "positive"}
+        structured_model = AsyncMock()
+        structured_model.ainvoke = AsyncMock(return_value=structured_result)
+
+        model = MagicMock()
+        model.with_structured_output = MagicMock(return_value=structured_model)
+
+        schema = {
+            "type": "object",
+            "properties": {"label": {"type": "string"}},
+        }
+
+        tool = create_swarm_task_tool(subagents=[], default_model=model)
+
+        result = await tool.ainvoke(
+            {
+                "description": "work",
+                "mode": "invoke",
+                "response_schema": schema,
+            }
+        )
+
+        model.with_structured_output.assert_called_once_with(
+            {**schema, "title": "structured_output"}
+        )
+        structured_model.ainvoke.assert_awaited_once()
+        assert result == json.dumps(structured_result)
+
+    async def test_returns_string_content_from_model_response(self) -> None:
+        model = _make_mock_model("the answer")
+        tool = create_swarm_task_tool(subagents=[], default_model=model)
+
+        result = await tool.ainvoke({"description": "work", "mode": "invoke"})
+        assert result == "the answer"
+
+    async def test_works_without_response_schema(self) -> None:
+        model = _make_mock_model("plain response")
+        tool = create_swarm_task_tool(subagents=[], default_model=model)
+
+        result = await tool.ainvoke({"description": "work", "mode": "invoke"})
+
+        assert result == "plain response"
+        model.ainvoke.assert_awaited_once()
+
+
+# ---------------------------------------------------------------------------
+# Mode defaulting
+# ---------------------------------------------------------------------------
+
+
+class TestModeDefaulting:
+    """Tests for mode default behavior."""
+
+    async def test_defaults_to_agent_mode_when_mode_not_provided(self) -> None:
+        subagent_model = _FakeModel()
+        default_model = _make_mock_model()
+
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent",
+            return_value=_make_fake_agent(),
+        ):
+            tool = create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(
+                        name="worker",
+                        description="W",
+                        system_prompt="W.",
+                        model=subagent_model,
+                    )
+                ],
+                default_model=default_model,
+            )
+
+        await tool.ainvoke({"description": "work", "subagent_type": "worker"})
+
+        # default_model.ainvoke should NOT have been called (that's invoke mode)
+        default_model.ainvoke.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# Multiple subagents
+# ---------------------------------------------------------------------------
+
+
+class TestMultipleSubagents:
+    """Tests for multiple subagent configurations."""
+
+    def test_uses_subagent_own_model_when_specified(self) -> None:
+        screener_model = _FakeModel()
+        default_model = _make_mock_model()
+
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent",
+            return_value=_make_fake_agent(),
+        ) as mock_create:
+            create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(
+                        name="screener",
+                        description="S",
+                        system_prompt="Screen.",
+                        model=screener_model,
+                    ),
+                ],
+                default_model=default_model,
+            )
+
+        mock_create.assert_called_once()
+        assert mock_create.call_args.kwargs["model"] is screener_model
+
+    def test_falls_back_to_default_model_when_subagent_has_no_model(self) -> None:
+        default_model = _make_mock_model()
+
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent",
+            return_value=_make_fake_agent(),
+        ) as mock_create:
+            create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(name="worker", description="W", system_prompt="W.")
+                ],
+                default_model=default_model,
+            )
+
+        mock_create.assert_called_once()
+        assert mock_create.call_args.kwargs["model"] is default_model
+
+
+# ---------------------------------------------------------------------------
+# TTL variant cache integration
+# ---------------------------------------------------------------------------
+
+
+class TestTTLVariantCacheIntegration:
+    """Tests for TTL-based variant caching in agent mode."""
+
+    async def test_reuses_compiled_agent_on_repeated_calls_with_same_schema(
+        self,
+    ) -> None:
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent",
+            return_value=_make_fake_agent(),
+        ) as mock_create:
+            tool = create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(name="worker", description="W", system_prompt="W.")
+                ],
+                default_model=_make_mock_model(),
+            )
+
+            schema = {
+                "type": "object",
+                "properties": {"label": {"type": "string"}},
+            }
+
+            await tool.ainvoke(
+                {
+                    "description": "row 1",
+                    "subagent_type": "worker",
+                    "response_schema": schema,
+                }
+            )
+            await tool.ainvoke(
+                {
+                    "description": "row 2",
+                    "subagent_type": "worker",
+                    "response_schema": schema,
+                }
+            )
+            await tool.ainvoke(
+                {
+                    "description": "row 3",
+                    "subagent_type": "worker",
+                    "response_schema": schema,
+                }
+            )
+
+            # 1 at construction + 1 for the schema variant (reused for rows 2 and 3)
+            assert mock_create.call_count == 2
+
+    async def test_compiles_separate_variants_for_distinct_schemas(self) -> None:
+        with patch(
+            "langchain_quickjs._swarm_task.create_agent",
+            return_value=_make_fake_agent(),
+        ) as mock_create:
+            tool = create_swarm_task_tool(
+                subagents=[
+                    SwarmSubAgent(name="worker", description="W", system_prompt="W.")
+                ],
+                default_model=_make_mock_model(),
+            )
+
+            await tool.ainvoke(
+                {
+                    "description": "work",
+                    "subagent_type": "worker",
+                    "response_schema": {
+                        "type": "object",
+                        "properties": {"a": {"type": "string"}},
+                    },
+                }
+            )
+            await tool.ainvoke(
+                {
+                    "description": "work",
+                    "subagent_type": "worker",
+                    "response_schema": {
+                        "type": "object",
+                        "properties": {"b": {"type": "number"}},
+                    },
+                }
+            )
+
+            # 1 at construction + 2 schema variants
+            assert mock_create.call_count == 3
+
+
+# ---------------------------------------------------------------------------
+# Tool metadata
+# ---------------------------------------------------------------------------
+
+
+class TestToolMetadata:
+    """Tests for the tool's name and schema."""
+
+    def test_tool_name_is_swarm_task(self) -> None:
+        tool = create_swarm_task_tool(subagents=[], default_model=_make_mock_model())
+        assert tool.name == "swarm_task"
+
+    def test_tool_has_description(self) -> None:
+        tool = create_swarm_task_tool(subagents=[], default_model=_make_mock_model())
+        assert "swarm subagent" in tool.description.lower()