Add internal shared core for the LangChain-family plugins

temporalio/contrib/_langchain/__init__.py

@@ -0,0 +1,10 @@
+"""Internal shared machinery for the LangChain-family plugins.
+
+This package is shared infrastructure for ``temporalio.contrib.langgraph``,
+``temporalio.contrib.langsmith``, and ``temporalio.contrib.deepagents``. It is
+NOT a public API: names, modules, and behavior may change without notice.
+
+Import discipline: this ``__init__`` performs no imports, and submodules defer
+every third-party import into function bodies, so the package is importable
+with none of the LangChain-family distributions installed.
+"""

temporalio/contrib/_langchain/_activity_helpers.py

@@ -0,0 +1,90 @@
+"""Activity-side helpers shared by the LangChain-family plugins."""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import timedelta
+from functools import wraps
+from typing import Any, Callable
+
+from temporalio import activity
+from temporalio.exceptions import ApplicationError
+
+
+def auto_heartbeater(fn: Callable) -> Callable:
+    """Heartbeat at half the configured ``heartbeat_timeout`` while ``fn`` runs.
+
+    Long LLM calls (thinking mode, long context, streaming accumulation) can run
+    well past a scheduler's patience; without a heartbeat Temporal would cancel
+    them and surface a ``HeartbeatTimeoutError`` instead of the real problem.
+    """
+
+    @wraps(fn)
+    async def wrapped(*args: Any, **kwargs: Any) -> Any:
+        heartbeat_timeout = activity.info().heartbeat_timeout
+        beat_task: asyncio.Task | None = None
+        if heartbeat_timeout:
+            interval = heartbeat_timeout.total_seconds() / 2
+
+            async def beat() -> None:
+                while True:
+                    activity.heartbeat()
+                    await asyncio.sleep(interval)
+
+            beat_task = asyncio.create_task(beat())
+        try:
+            return await fn(*args, **kwargs)
+        finally:
+            if beat_task is not None:
+                beat_task.cancel()
+                # Let the cancellation land before returning so no pending task
+                # outlives the activity (a bare ``cancel()`` leaves the task to
+                # be destroyed while pending if the loop shuts down first).
+                # ``asyncio.wait`` never re-raises the task's CancelledError.
+                await asyncio.wait([beat_task])
+
+    return wrapped
+
+
+def translate_api_error(exc: Exception) -> ApplicationError | None:
+    """Map an LLM SDK HTTP error onto Temporal's retry contract.
+
+    Works by duck typing so neither ``openai`` nor ``anthropic`` needs to be
+    imported here: both expose ``status_code`` and ``response.headers``. Returns
+    ``None`` when ``exc`` is not a recognizable HTTP status error, so the caller
+    can fall through to its generic handling.
+    """
+    status = getattr(exc, "status_code", None)
+    if status is None:
+        return None
+    headers: dict[str, Any] = {}
+    response = getattr(exc, "response", None)
+    if response is not None:
+        headers = dict(getattr(response, "headers", {}) or {})
+    # Case-insensitive header access.
+    lower = {str(k).lower(): v for k, v in headers.items()}
+
+    retryable = status in (408, 409, 429) or 500 <= status < 600
+    should_retry = lower.get("x-should-retry")
+    if should_retry == "false":
+        retryable = False
+    elif should_retry == "true":
+        retryable = True
+
+    delay_ms = lower.get("retry-after-ms")
+    retry_after = lower.get("retry-after")
+    next_delay: timedelta | None = None
+    try:
+        if delay_ms is not None:
+            next_delay = timedelta(milliseconds=int(delay_ms))
+        elif retry_after is not None:
+            next_delay = timedelta(seconds=int(retry_after))
+    except (TypeError, ValueError):
+        next_delay = None
+
+    return ApplicationError(
+        str(exc),
+        type=type(exc).__name__,
+        non_retryable=not retryable,
+        next_retry_delay=next_delay,
+    )

temporalio/contrib/_langchain/_aio_to_thread.py

@@ -0,0 +1,59 @@
+"""LangSmith ``aio_to_thread`` override shared by the LangChain-family plugins."""
+
+from __future__ import annotations
+
+from typing import Any, Callable
+
+import temporalio.workflow
+
+_installed = False
+
+
+async def _temporal_aio_to_thread(
+    default_aio_to_thread: Callable[..., Any],
+    ctx: Any,
+    func: Callable[..., Any],
+    /,
+    *args: Any,
+    **kwargs: Any,
+) -> Any:
+    """Run LangSmith's ``aio_to_thread`` synchronously inside Temporal workflows.
+
+    The ``@traceable`` decorator on async functions uses ``aio_to_thread()`` →
+    ``loop.run_in_executor()`` for run setup/teardown. The Temporal workflow
+    event loop does not support ``run_in_executor``. This override runs those
+    functions synchronously on the workflow thread when inside a workflow,
+    and delegates to the default implementation outside workflows.
+
+    Registered via ``langsmith.set_runtime_overrides(aio_to_thread=...)``.
+    """
+    if not temporalio.workflow.in_workflow():
+        return await default_aio_to_thread(ctx, func, *args, **kwargs)
+    with temporalio.workflow.unsafe.sandbox_unrestricted():
+        return ctx.run(func, *args, **kwargs)
+
+
+def install_aio_to_thread_override() -> None:
+    """Install the ``aio_to_thread`` override via LangSmith's official API.
+
+    Safe to call multiple times and from multiple plugins; the override is
+    installed once per process. It is deliberately never uninstalled:
+    LangSmith exposes a single process-wide override slot (each
+    ``set_runtime_overrides`` call replaces it wholesale), so resetting it on
+    one worker's shutdown would strip a composed plugin's still-needed
+    override. Leaving it installed is safe — the override defers to
+    LangSmith's default thread hop whenever ``workflow.in_workflow()`` is
+    false, so it is inert outside workflows.
+
+    Raises whatever the lazy ``langsmith`` import or
+    ``set_runtime_overrides`` call raises (e.g. ``ImportError`` when
+    LangSmith is absent); the installed flag stays unset on failure so a
+    later call can retry.
+    """
+    global _installed  # noqa: PLW0603
+    if _installed:
+        return
+    import langsmith
+
+    langsmith.set_runtime_overrides(aio_to_thread=_temporal_aio_to_thread)
+    _installed = True

temporalio/contrib/_langchain/_converter.py

@@ -0,0 +1,61 @@
+"""Payload converter shared by the LangChain-family plugins.
+
+This module eagerly imports ``temporalio.contrib.pydantic``; only plugins
+that actually wire the converter import it, so the core package itself stays
+importable without pydantic installed.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+
+from temporalio.contrib.pydantic import PydanticPayloadConverter, ToJsonOptions
+from temporalio.converter import DataConverter
+
+
+class LangChainPayloadConverter(PydanticPayloadConverter):
+    """Pydantic payload converter pinned to ``exclude_unset=True``.
+
+    LangChain request/response types are deeply nested with many
+    ``Optional[...] = None`` fields. Shipping every unset default inflates
+    payloads several-fold and some peers reject the explicit nulls on
+    round-trip, so unset fields are excluded by convention.
+    """
+
+    def __init__(self) -> None:
+        """Construct the converter with ``exclude_unset`` serialization."""
+        super().__init__(ToJsonOptions(exclude_unset=True))
+
+
+data_converter = DataConverter(payload_converter_class=LangChainPayloadConverter)
+"""The family default data converter (LangChain messages are shipped as their
+``dumpd`` JSON form, so the Pydantic converter only ever sees plain
+containers)."""
+
+
+def build_data_converter(
+    user_converter: DataConverter | None,
+    *,
+    plugin_name: str,
+) -> DataConverter:
+    """Compose the family converter with whatever the caller already set.
+
+    * ``None`` — install the family default.
+    * the SDK default converter — swap in the LangChain-aware Pydantic
+      converter via :func:`dataclasses.replace`.
+    * a custom converter — refuse rather than silently clobber it; the caller
+      must fold ``LangChainPayloadConverter`` into their own converter.
+      ``plugin_name`` names the refusing plugin in the error.
+    """
+    if user_converter is None:
+        return data_converter
+    if user_converter is DataConverter.default:
+        return dataclasses.replace(
+            user_converter, payload_converter_class=LangChainPayloadConverter
+        )
+    raise ValueError(
+        f"{plugin_name} cannot compose with a custom data_converter "
+        "automatically. Set payload_converter_class=LangChainPayloadConverter "
+        "on your own DataConverter (so LangChain messages serialize with "
+        "exclude_unset=True), or omit data_converter to use the plugin default."
+    )

temporalio/contrib/_langchain/_messages.py

@@ -0,0 +1,44 @@
+"""LangChain object serialization shared by the LangChain-family plugins."""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def dump_object(obj: Any) -> Any:
+    """Serialize a single LangChain ``Serializable`` (message, tool call, …)."""
+    from langchain_core.load import dumpd
+
+    return dumpd(obj)
+
+
+def load_object(data: Any) -> Any:
+    """Rehydrate a value produced by :func:`dump_object`, preserving subtype."""
+    from langchain_core.load import load
+
+    return load(data)
+
+
+def dump_messages(messages: Any) -> list[Any]:
+    """Serialize a sequence of LangChain messages to their ``dumpd`` form."""
+    from langchain_core.load import dumpd
+
+    return [dumpd(m) for m in messages]
+
+
+def load_messages(dumped: list[Any]) -> list[Any]:
+    """Rehydrate messages serialized by :func:`dump_messages`."""
+    from langchain_core.load import load
+
+    return [load(d) for d in dumped]
+
+
+def tool_to_schema(tool: Any) -> dict[str, Any]:
+    """Advertise a tool to the model as a full OpenAI tool schema.
+
+    Carries name + description + argument JSON schema so the model can build
+    valid arguments, not just select the tool by name.
+    """
+    from langchain_core.utils.function_calling import convert_to_openai_tool
+
+    return convert_to_openai_tool(tool)

temporalio/contrib/_langchain/_passthrough.py

@@ -0,0 +1,22 @@
+"""Sandbox passthrough-list merging shared by the LangChain-family plugins.
+
+Only the MECHANISM lives here; each plugin owns its module list (the lists
+are behavior for released plugins and must not drift by sharing).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable, Sequence
+
+
+def merge_passthrough_modules(
+    defaults: Sequence[str], user: Iterable[str] | None
+) -> tuple[str, ...]:
+    """Merge caller-supplied passthrough modules with a plugin's defaults.
+
+    Order-preserving: defaults first, then user additions, first occurrence
+    wins on duplicates.
+    """
+    merged = [*defaults, *(user or ())]
+    # dict.fromkeys preserves order while de-duplicating.
+    return tuple(dict.fromkeys(merged))

temporalio/contrib/_langchain/_runnable_config.py

@@ -0,0 +1,103 @@
+"""``RunnableConfig`` strip/rebuild shared by the LangChain-family plugins."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping, Sequence
+from typing import TYPE_CHECKING, Any, cast
+
+if TYPE_CHECKING:
+    from langchain_core.runnables import RunnableConfig
+
+
+def is_jsonish(value: Any) -> bool:
+    """Return True for values representable as plain JSON.
+
+    Containers are checked recursively. Tuples are accepted — they encode
+    as JSON arrays and rehydrate as lists, so a round-trip preserves
+    content but may change the container type.
+    """
+    if value is None or isinstance(value, (str, int, float, bool)):
+        return True
+    if isinstance(value, (list, tuple)):
+        return all(is_jsonish(v) for v in value)
+    if isinstance(value, dict):
+        return all(isinstance(k, str) and is_jsonish(v) for k, v in value.items())
+    return False
+
+
+def strip_runnable_config(
+    config: Mapping[str, Any] | None,
+    *,
+    configurable_keys: Sequence[str] | None = None,
+    configurable_filter: Callable[[str, Any], bool] | None = None,
+    metadata_filter: Callable[[Any], bool] | None = None,
+) -> dict[str, Any]:
+    """Return a serializable subset of a ``RunnableConfig``.
+
+    The full object holds non-serializable things (callbacks, checkpointer /
+    store / cache handles, pregel send/read callables) that cannot cross an
+    activity boundary, so only primitive fields and a caller-selected subset
+    of ``configurable`` are kept. The output shape follows the langgraph
+    plugin's released behavior: ``tags`` and ``metadata`` are always present
+    (empty when absent), ``run_name`` / ``run_id`` are kept when truthy,
+    ``recursion_limit`` when not None, and ``configurable`` only when the
+    selection is non-empty.
+
+    Exactly one of ``configurable_keys`` (keys kept in whitelist order) or
+    ``configurable_filter`` (entries kept in source order) must be provided.
+    ``metadata_filter`` optionally drops metadata VALUES (default: keep all,
+    matching langgraph).
+    """
+    if (configurable_keys is None) == (configurable_filter is None):
+        raise ValueError(
+            "exactly one of configurable_keys or configurable_filter is required"
+        )
+    orig: Mapping[str, Any] = config or {}
+    configurable: Mapping[str, Any] = orig.get("configurable") or {}
+
+    metadata = dict(orig.get("metadata") or {})
+    if metadata_filter is not None:
+        metadata = {k: v for k, v in metadata.items() if metadata_filter(v)}
+    result: dict[str, Any] = {
+        "tags": list(orig.get("tags") or []),
+        "metadata": metadata,
+    }
+    if run_name := orig.get("run_name"):
+        result["run_name"] = run_name
+    if run_id := orig.get("run_id"):
+        result["run_id"] = run_id
+    if (recursion_limit := orig.get("recursion_limit")) is not None:
+        result["recursion_limit"] = recursion_limit
+
+    if configurable_keys is not None:
+        stripped_configurable: dict[str, Any] = {
+            key: configurable[key] for key in configurable_keys if key in configurable
+        }
+    else:
+        stripped_configurable = {
+            k: v
+            for k, v in configurable.items()
+            # The guard above makes configurable_filter non-None on this
+            # branch; the re-check narrows for type checkers.
+            if configurable_filter is not None and configurable_filter(k, v)
+        }
+    if stripped_configurable:
+        result["configurable"] = stripped_configurable
+    return result
+
+
+def rebuild_runnable_config(data: dict[str, Any]) -> "RunnableConfig":
+    """Reconstruct a minimal ``RunnableConfig`` from :func:`strip_runnable_config`."""
+    config: dict[str, Any] = {"metadata": dict(data.get("metadata", {}))}
+    if data.get("tags"):
+        config["tags"] = list(data["tags"])
+    for key in ("run_id", "run_name"):
+        if data.get(key) is not None:
+            config[key] = data[key]
+    if data.get("recursion_limit") is not None:
+        config["recursion_limit"] = data["recursion_limit"]
+    if data.get("configurable"):
+        config["configurable"] = dict(data["configurable"])
+    # Double cast: a TypedDict and dict[str, Any] "insufficiently overlap"
+    # for basedpyright's reportInvalidCast; object is the sanctioned bridge.
+    return cast("RunnableConfig", cast(object, config))