from __future__ import annotations import asyncio import concurrent.futures import json import sys import threading import uuid from inspect import isasyncgenfunction from typing import TYPE_CHECKING, Any, Literal from unittest.mock import MagicMock, patch import pytest from langsmith import Client, RunTree, get_current_run_tree, traceable from langsmith.run_helpers import tracing_context from langsmith.utils import get_env_var from langchain_core.callbacks.base import BaseCallbackHandler from langchain_core.callbacks.manager import CallbackManager from langchain_core.runnables.base import RunnableLambda, RunnableParallel from langchain_core.tracers.langchain import LangChainTracer if TYPE_CHECKING: from collections.abc import AsyncGenerator, Callable, Coroutine, Generator, Mapping from langchain_core.runnables.config import RunnableConfig def _get_posts(client: Client) -> list[dict[str, Any]]: mock_calls = client.session.request.mock_calls # type: ignore[attr-defined] posts = [] for call in mock_calls: if call.args: if call.args[0] != "POST": continue assert call.args[0] == "POST" assert call.args[1].startswith("https://api.smith.langchain.com") body = json.loads(call.kwargs["data"]) if "post" in body: # Batch request assert body["post"] posts.extend(body["post"]) else: posts.append(body) return posts def _create_tracer_with_mocked_client( project_name: str | None = None, tags: list[str] | None = None, metadata: Mapping[str, str] | None = None, ) -> LangChainTracer: mock_session = MagicMock() mock_client_ = Client( session=mock_session, api_key="test", auto_batch_tracing=False ) return LangChainTracer( client=mock_client_, project_name=project_name, tags=tags, metadata=metadata ) def test_tracing_context() -> None: mock_session = MagicMock() mock_client_ = Client( session=mock_session, api_key="test", auto_batch_tracing=False ) @RunnableLambda def my_lambda(a: int) -> int: return a + 1 @RunnableLambda def my_function(a: int) -> int: with tracing_context(enabled=False): return my_lambda.invoke(a) name = uuid.uuid4().hex project_name = f"Some project {name}" with tracing_context(project_name=project_name, client=mock_client_, enabled=True): assert my_function.invoke(1) == 2 posts = _get_posts(mock_client_) assert len(posts) == 1 assert all(post["session_name"] == project_name for post in posts) def test_inheritable_metadata_respects_explicit_metadata_with_tracing_context() -> None: """Tracer defaults fill missing keys while run metadata keeps precedence.""" tracer = _create_tracer_with_mocked_client() @RunnableLambda def my_func(x: int) -> int: return x callbacks = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_metadata={ "tenant": "from_tracer", "shared": "from_tracer", }, ) with tracing_context(enabled=True, client=tracer.client): my_func.invoke( 1, { "callbacks": callbacks, "metadata": {"shared": "from_run", "explicit": "from_run"}, }, ) posts = _get_posts(tracer.client) assert len(posts) == 1 metadata = posts[0].get("extra", {}).get("metadata", {}) assert metadata["tenant"] == "from_tracer" assert metadata["shared"] == "from_run" assert metadata["explicit"] == "from_run" def test_config_traceable_handoff() -> None: if hasattr(get_env_var, "cache_clear"): get_env_var.cache_clear() # type: ignore[attr-defined] tracer = _create_tracer_with_mocked_client( project_name="another-flippin-project", tags=["such-a-tag"] ) @traceable def my_great_great_grandchild_function(a: int) -> int: rt = get_current_run_tree() assert rt assert rt.session_name == "another-flippin-project" return a + 1 @RunnableLambda def my_great_grandchild_function(a: int) -> int: return my_great_great_grandchild_function(a) @RunnableLambda def my_grandchild_function(a: int) -> int: return my_great_grandchild_function.invoke(a) @traceable def my_child_function(a: int) -> int: return my_grandchild_function.invoke(a) * 3 @traceable() def my_function(a: int) -> int: rt = get_current_run_tree() assert rt assert rt.session_name == "another-flippin-project" return my_child_function(a) def my_parent_function(a: int) -> int: rt = get_current_run_tree() assert rt assert rt.session_name == "another-flippin-project" return my_function(a) my_parent_runnable = RunnableLambda(my_parent_function) assert my_parent_runnable.invoke(1, {"callbacks": [tracer]}) == 6 posts = _get_posts(tracer.client) assert all(post["session_name"] == "another-flippin-project" for post in posts) # There should have been 6 runs created, # one for each function invocation assert len(posts) == 6 name_to_body = {post["name"]: post for post in posts} ordered_names = [ "my_parent_function", "my_function", "my_child_function", "my_grandchild_function", "my_great_grandchild_function", "my_great_great_grandchild_function", ] trace_id = posts[0]["trace_id"] last_dotted_order = None parent_run_id = None for name in ordered_names: id_ = name_to_body[name]["id"] parent_run_id_ = name_to_body[name].get("parent_run_id") if parent_run_id_ is not None: assert parent_run_id == parent_run_id_ assert name in name_to_body # All within the same trace assert name_to_body[name]["trace_id"] == trace_id dotted_order: str = name_to_body[name]["dotted_order"] assert dotted_order is not None if last_dotted_order is not None: assert dotted_order > last_dotted_order assert dotted_order.startswith(last_dotted_order), ( "Unexpected dotted order for run" f" {name}\n{dotted_order}\n{last_dotted_order}" ) last_dotted_order = dotted_order parent_run_id = id_ assert "such-a-tag" in name_to_body["my_parent_function"]["tags"] @pytest.mark.skipif( sys.version_info < (3, 11), reason="Asyncio context vars require Python 3.11+" ) async def test_config_traceable_async_handoff() -> None: tracer = _create_tracer_with_mocked_client() @traceable def my_great_great_grandchild_function(a: int) -> int: return a + 1 @RunnableLambda def my_great_grandchild_function(a: int) -> int: return my_great_great_grandchild_function(a) @RunnableLambda async def my_grandchild_function(a: int) -> int: return my_great_grandchild_function.invoke(a) @traceable async def my_child_function(a: int) -> int: return await my_grandchild_function.ainvoke(a) * 3 @traceable() async def my_function(a: int) -> int: return await my_child_function(a) async def my_parent_function(a: int) -> int: return await my_function(a) my_parent_runnable = RunnableLambda(my_parent_function) result = await my_parent_runnable.ainvoke(1, {"callbacks": [tracer]}) assert result == 6 posts = _get_posts(tracer.client) # There should have been 6 runs created, # one for each function invocation assert len(posts) == 6 name_to_body = {post["name"]: post for post in posts} ordered_names = [ "my_parent_function", "my_function", "my_child_function", "my_grandchild_function", "my_great_grandchild_function", "my_great_great_grandchild_function", ] trace_id = posts[0]["trace_id"] last_dotted_order = None parent_run_id = None for name in ordered_names: id_ = name_to_body[name]["id"] parent_run_id_ = name_to_body[name].get("parent_run_id") if parent_run_id_ is not None: assert parent_run_id == parent_run_id_ assert name in name_to_body # All within the same trace assert name_to_body[name]["trace_id"] == trace_id dotted_order: str = name_to_body[name]["dotted_order"] assert dotted_order is not None if last_dotted_order is not None: assert dotted_order > last_dotted_order assert dotted_order.startswith(last_dotted_order), ( "Unexpected dotted order for run" f" {name}\n{dotted_order}\n{last_dotted_order}" ) last_dotted_order = dotted_order parent_run_id = id_ @patch("langchain_core.tracers.langchain.get_client") @pytest.mark.parametrize("enabled", [None, True, False]) @pytest.mark.parametrize("env", ["", "true"]) def test_tracing_enable_disable( mock_get_client: MagicMock, *, enabled: bool | None, env: str ) -> None: mock_session = MagicMock() mock_client_ = Client( session=mock_session, api_key="test", auto_batch_tracing=False ) mock_get_client.return_value = mock_client_ def my_func(a: int) -> int: return a + 1 if hasattr(get_env_var, "cache_clear"): get_env_var.cache_clear() # type: ignore[attr-defined] env_on = env == "true" with ( patch.dict("os.environ", {"LANGSMITH_TRACING": env}), tracing_context(enabled=enabled), ): RunnableLambda(my_func).invoke(1) mock_posts = _get_posts(mock_client_) if enabled is True: assert len(mock_posts) == 1 elif enabled is False: assert not mock_posts elif env_on: assert len(mock_posts) == 1 else: assert not mock_posts class TestRunnableSequenceParallelTraceNesting: @pytest.fixture(autouse=True) def _setup(self) -> None: self.tracer = _create_tracer_with_mocked_client() @staticmethod def _create_parent( other_thing: Callable[ [int], Generator[int, None, None] | AsyncGenerator[int, None] ], ) -> RunnableLambda: @RunnableLambda def my_child_function(a: int) -> int: return a + 2 parallel = RunnableParallel( chain_result=my_child_function.with_config(tags=["atag"]), other_thing=other_thing, ) def before(x: int) -> int: return x def after(x: dict[str, Any]) -> int: return int(x["chain_result"]) sequence = before | parallel | after if isasyncgenfunction(other_thing): @RunnableLambda async def parent(a: int) -> int: return await sequence.ainvoke(a) else: @RunnableLambda def parent(a: int) -> int: return sequence.invoke(a) return parent def _check_posts(self) -> None: posts = _get_posts(self.tracer.client) name_order = [ "parent", "RunnableSequence", "before", "RunnableParallel", ["my_child_function", "other_thing"], "after", ] expected_parents = { "parent": None, "RunnableSequence": "parent", "before": "RunnableSequence", "RunnableParallel": "RunnableSequence", "my_child_function": "RunnableParallel", "other_thing": "RunnableParallel", "after": "RunnableSequence", } assert len(posts) == sum( 1 if isinstance(n, str) else len(n) for n in name_order ) prev_dotted_order = None dotted_order_map = {} id_map = {} parent_id_map = {} i = 0 for name in name_order: if isinstance(name, list): for n in name: matching_post = next( p for p in posts[i : i + len(name)] if p["name"] == n ) assert matching_post dotted_order = matching_post["dotted_order"] if prev_dotted_order is not None: assert dotted_order > prev_dotted_order dotted_order_map[n] = dotted_order id_map[n] = matching_post["id"] parent_id_map[n] = matching_post.get("parent_run_id") i += len(name) continue assert posts[i]["name"] == name dotted_order = posts[i]["dotted_order"] if prev_dotted_order is not None and not str( expected_parents[name] # type: ignore[index] ).startswith("RunnableParallel"): assert dotted_order > prev_dotted_order, ( f"{name} not after {name_order[i - 1]}" ) prev_dotted_order = dotted_order if name in dotted_order_map: msg = f"Duplicate name {name}" raise ValueError(msg) dotted_order_map[name] = dotted_order id_map[name] = posts[i]["id"] parent_id_map[name] = posts[i].get("parent_run_id") i += 1 # Now check the dotted orders for name, parent_ in expected_parents.items(): dotted_order = dotted_order_map[name] if parent_ is not None: parent_dotted_order = dotted_order_map[parent_] assert dotted_order.startswith(parent_dotted_order), ( f"{name}, {parent_dotted_order} not in {dotted_order}" ) assert str(parent_id_map[name]) == str(id_map[parent_]) else: assert dotted_order.split(".")[0] == dotted_order @pytest.mark.parametrize( "method", [ lambda parent, cb: parent.invoke(1, {"callbacks": cb}), lambda parent, cb: list(parent.stream(1, {"callbacks": cb}))[-1], lambda parent, cb: parent.batch([1], {"callbacks": cb})[0], ], ids=["invoke", "stream", "batch"], ) def test_sync( self, method: Callable[[RunnableLambda, list[BaseCallbackHandler]], int] ) -> None: def other_thing(_: int) -> Generator[int, None, None]: yield 1 parent = self._create_parent(other_thing) # Now run the chain and check the resulting posts assert method(parent, [self.tracer]) == 3 self._check_posts() @staticmethod async def ainvoke( parent: RunnableLambda[int, int], cb: list[BaseCallbackHandler] ) -> int: return await parent.ainvoke(1, {"callbacks": cb}) @staticmethod async def astream( parent: RunnableLambda[int, int], cb: list[BaseCallbackHandler] ) -> int: return [res async for res in parent.astream(1, {"callbacks": cb})][-1] @staticmethod async def abatch( parent: RunnableLambda[int, int], cb: list[BaseCallbackHandler] ) -> int: return (await parent.abatch([1], {"callbacks": cb}))[0] @pytest.mark.skipif( sys.version_info < (3, 11), reason="Asyncio context vars require Python 3.11+" ) @pytest.mark.parametrize("method", [ainvoke, astream, abatch]) async def test_async( self, method: Callable[ [RunnableLambda, list[BaseCallbackHandler]], Coroutine[Any, Any, int] ], ) -> None: async def other_thing(_: int) -> AsyncGenerator[int, None]: yield 1 parent = self._create_parent(other_thing) # Now run the chain and check the resulting posts assert await method(parent, [self.tracer]) == 3 self._check_posts() @pytest.mark.parametrize("parent_type", ["ls", "lc"]) def test_tree_is_constructed(parent_type: Literal["ls", "lc"]) -> None: mock_session = MagicMock() mock_client_ = Client( session=mock_session, api_key="test", auto_batch_tracing=False ) grandchild_run = None kitten_run = None @traceable def kitten(x: str) -> str: nonlocal kitten_run kitten_run = get_current_run_tree() return x @RunnableLambda def grandchild(x: str) -> str: nonlocal grandchild_run grandchild_run = get_current_run_tree() return kitten(x) @RunnableLambda def child(x: str) -> str: return grandchild.invoke(x) rid = uuid.uuid4() with tracing_context( client=mock_client_, enabled=True, metadata={"some_foo": "some_bar"}, tags=["afoo"], ): collected: dict[str, RunTree] = {} def collect_langsmith_run(run: RunTree) -> None: collected[str(run.id)] = run def collect_tracer_run(_: LangChainTracer, run: RunTree) -> None: collected[str(run.id)] = run if parent_type == "ls": @traceable def parent() -> str: return child.invoke("foo") assert ( parent(langsmith_extra={"on_end": collect_langsmith_run, "run_id": rid}) == "foo" ) assert collected else: @RunnableLambda def parent(_: Any) -> str: return child.invoke("foo") tracer = LangChainTracer() with patch.object(LangChainTracer, "_persist_run", new=collect_tracer_run): assert ( parent.invoke(..., {"run_id": rid, "callbacks": [tracer]}) == "foo" # type: ignore[attr-defined] ) run = collected.get(str(rid)) assert run is not None assert run.name == "parent" assert run.child_runs child_run = run.child_runs[0] assert child_run.name == "child" assert isinstance(grandchild_run, RunTree) assert grandchild_run.name == "grandchild" assert grandchild_run.metadata.get("some_foo") == "some_bar" assert "afoo" in grandchild_run.tags # type: ignore[operator] assert isinstance(kitten_run, RunTree) assert kitten_run.name == "kitten" assert not kitten_run.child_runs assert kitten_run.metadata.get("some_foo") == "some_bar" assert "afoo" in kitten_run.tags # type: ignore[operator] assert grandchild_run is not None assert kitten_run.dotted_order.startswith(grandchild_run.dotted_order) def test_traceable_parent_run_map_cleanup() -> None: """External RunTree injected into run_map is cleaned up when its child ends. When a `@traceable` function invokes a LangChain `Runnable`, the `RunTree` is added to the tracer's `run_map` so child runs can reference it. Previously the entry was never removed, causing a memory leak that grew with every call. Uses an explicit tracer so we can inspect `run_map` directly after the call — the `_configure` insertion path is identical regardless of whether the tracer was created internally or passed in. """ tracer = _create_tracer_with_mocked_client() @RunnableLambda def child(x: str) -> str: return x with tracing_context(client=tracer.client, enabled=True): @traceable def parent(x: str) -> str: return child.invoke(x, config={"callbacks": [tracer]}) parent("hello") assert tracer.run_map == {}, ( f"run_map should be empty but contains: " f"{[getattr(v, 'name', k) for k, v in tracer.run_map.items()]}" ) def test_traceable_parent_run_map_cleanup_with_sibling_children() -> None: """External parent survives in run_map until ALL its children finish. When a `@traceable` function invokes a chain with multiple steps (e.g. prompt | llm), each step is a sibling child of the same intermediate run. The external parent must stay in `run_map` until the last child completes, not be removed when the first child ends. """ from langchain_core.language_models.fake_chat_models import ( # noqa: PLC0415 FakeListChatModel, ) from langchain_core.prompts import ChatPromptTemplate # noqa: PLC0415 tracer = _create_tracer_with_mocked_client() prompt = ChatPromptTemplate.from_messages([("system", "bot"), ("human", "{input}")]) llm = FakeListChatModel(responses=["hi"]) chain = prompt | llm with tracing_context(client=tracer.client, enabled=True): @traceable def parent(x: dict) -> Any: return chain.invoke(x, config={"callbacks": [tracer]}) result = parent({"input": "hello"}) assert result is not None assert tracer.run_map == {}, ( f"run_map should be empty but contains: " f"{[getattr(v, 'name', k) for k, v in tracer.run_map.items()]}" ) def test_traceable_parent_run_map_no_runttree_accumulation() -> None: """RunTree objects reachable from run_map must not grow across calls. This is the memory-level regression test: a long-lived tracer is reused across many @traceable → Runnable invocations. Without the fix, each call leaves a RunTree (plus its child tree) in run_map, causing unbounded growth. With the fix, run_map is empty after every call, so the count stays flat. """ import gc # noqa: PLC0415 tracer = _create_tracer_with_mocked_client() @RunnableLambda def child(x: str) -> str: return x counts: list[int] = [] with tracing_context(client=tracer.client, enabled=True): @traceable def parent(x: str) -> str: return child.invoke(x, config={"callbacks": [tracer]}) for _ in range(5): parent("hello") gc.collect() # Count RunTree objects reachable from the tracer's run_map. run_map_runtrees = sum( 1 + len(v.child_runs) for v in tracer.run_map.values() ) counts.append(run_map_runtrees) # With the fix every call cleans up → counts are all 0. # Without the fix they grow: [1, 2, 3, 4, 5] (or more with children). assert counts == [0, 0, 0, 0, 0], ( f"RunTree objects in run_map should not accumulate, got counts: {counts}" ) class TestTracerMetadataThroughInvoke: """Tests for tracer metadata merging through invoke calls.""" def test_tracer_metadata_applied_to_all_runs(self) -> None: """Tracer metadata appears on every run when no config metadata is set.""" tracer = _create_tracer_with_mocked_client( metadata={"env": "prod", "service": "api"} ) @RunnableLambda def child(x: int) -> int: return x + 1 @RunnableLambda def parent(x: int) -> int: return child.invoke(x) parent.invoke(1, {"callbacks": [tracer]}) posts = _get_posts(tracer.client) assert len(posts) == 2 for post in posts: md = post.get("extra", {}).get("metadata", {}) assert md.get("env") == "prod", f"run {post['name']} missing env" assert md.get("service") == "api", f"run {post['name']} missing service" def test_config_metadata_takes_precedence(self) -> None: """Config metadata wins over tracer metadata for overlapping keys.""" tracer = _create_tracer_with_mocked_client( metadata={"env": "prod", "tracer_only": "yes"} ) @RunnableLambda def my_func(x: int) -> int: return x my_func.invoke( 1, { "callbacks": [tracer], "metadata": {"env": "staging", "config_only": "yes"}, }, ) posts = _get_posts(tracer.client) assert len(posts) == 1 md = posts[0].get("extra", {}).get("metadata", {}) # Config wins for overlapping key assert md["env"] == "staging" # Both non-overlapping keys are present assert md["tracer_only"] == "yes" assert md["config_only"] == "yes" def test_nested_calls_inherit_config_metadata(self) -> None: """Child runs inherit config metadata; tracer metadata fills gaps.""" tracer = _create_tracer_with_mocked_client( metadata={"tracer_key": "tracer_val"} ) @RunnableLambda def child(x: int) -> int: return x + 1 @RunnableLambda def parent(x: int) -> int: return child.invoke(x) parent.invoke( 1, { "callbacks": [tracer], "metadata": {"config_key": "config_val"}, }, ) posts = _get_posts(tracer.client) assert len(posts) == 2 name_to_md = { post["name"]: post.get("extra", {}).get("metadata", {}) for post in posts } # Both parent and child should have config metadata (inherited) # and tracer metadata (patched in) for name, md in name_to_md.items(): assert md.get("config_key") == "config_val", f"{name} missing config_key" assert md.get("tracer_key") == "tracer_val", f"{name} missing tracer_key" def test_tracer_metadata_not_applied_to_sibling_handlers(self) -> None: """Tracer metadata is not applied to other callback handlers. `_patch_missing_metadata` copies the metadata dict before patching, so the callback manager's shared metadata dict is not mutated. Other handlers should only see config metadata, not tracer metadata. """ tracer = _create_tracer_with_mocked_client( metadata={"tracer_key": "tracer_val"} ) received_metadata: list[dict[str, Any]] = [] class MetadataCapture(BaseCallbackHandler): """Callback handler that records metadata from chain events.""" def on_chain_start(self, *_args: Any, **kwargs: Any) -> None: received_metadata.append(dict(kwargs.get("metadata", {}))) capture = MetadataCapture() @RunnableLambda def my_func(x: int) -> int: return x my_func.invoke( 1, { "callbacks": [tracer, capture], "metadata": {"shared_key": "shared_val"}, }, ) assert len(received_metadata) >= 1 for md in received_metadata: assert md["shared_key"] == "shared_val" assert "tracer_key" not in md # But the posted run DOES have tracer metadata posts = _get_posts(tracer.client) assert len(posts) >= 1 for post in posts: post_md = post.get("extra", {}).get("metadata", {}) assert post_md["shared_key"] == "shared_val" assert post_md["tracer_key"] == "tracer_val" def test_tracer_metadata_with_no_config_metadata(self) -> None: """When no config metadata is set, tracer metadata is the sole source.""" tracer = _create_tracer_with_mocked_client( metadata={"only_from_tracer": "value"} ) @RunnableLambda def my_func(x: int) -> int: return x my_func.invoke(1, {"callbacks": [tracer]}) posts = _get_posts(tracer.client) assert len(posts) == 1 md = posts[0].get("extra", {}).get("metadata", {}) assert md["only_from_tracer"] == "value" def test_empty_tracer_metadata_does_not_interfere(self) -> None: """Tracer with no metadata does not interfere with config metadata.""" tracer = _create_tracer_with_mocked_client(metadata=None) @RunnableLambda def my_func(x: int) -> int: return x my_func.invoke( 1, {"callbacks": [tracer], "metadata": {"config_key": "config_val"}}, ) posts = _get_posts(tracer.client) assert len(posts) == 1 md = posts[0].get("extra", {}).get("metadata", {}) assert md["config_key"] == "config_val" def test_inheritable_metadata_nested_runs_preserve_parent_child_shape() -> None: """Concurrent nested runs keep parent-child linkage within each invocation.""" tracer = _create_tracer_with_mocked_client() barrier = threading.Barrier(2) @RunnableLambda def child(x: int) -> int: barrier.wait() return x + 1 @RunnableLambda def parent(x: int) -> int: return child.invoke(x) def invoke_for_tenant(tenant: str, value: int) -> int: callbacks = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_metadata={"tenant": tenant}, ) return parent.invoke(value, {"callbacks": callbacks}) threads = [ threading.Thread(target=invoke_for_tenant, args=("alpha", 1)), threading.Thread(target=invoke_for_tenant, args=("beta", 2)), ] for thread in threads: thread.start() for thread in threads: thread.join() posts = _get_posts(tracer.client) assert len(posts) == 4 parents = [post for post in posts if post["name"] == "parent"] children = [post for post in posts if post["name"] == "child"] assert len(parents) == 2 assert len(children) == 2 parent_ids = {parent["id"] for parent in parents} assert {child["parent_run_id"] for child in children} == parent_ids assert { post.get("extra", {}).get("metadata", {}).get("tenant") for post in posts } == { "alpha", "beta", } def test_inheritable_metadata_parallel_children_keep_tenant_isolation() -> None: """Concurrent roots with parallel child runs keep tenant metadata isolated.""" tracer = _create_tracer_with_mocked_client() barrier = threading.Barrier(4) @RunnableLambda def add_one(x: int) -> int: barrier.wait() return x + 1 @RunnableLambda def add_two(x: int) -> int: barrier.wait() return x + 2 parallel = RunnableParallel(first=add_one, second=add_two) def invoke_for_tenant(tenant: str, value: int) -> dict[str, int]: callbacks = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_metadata={"tenant": tenant}, ) return parallel.invoke(value, {"callbacks": callbacks}) with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: list(executor.map(invoke_for_tenant, ["alpha", "beta"], [1, 2])) posts = _get_posts(tracer.client) assert len(posts) == 6 assert { post.get("extra", {}).get("metadata", {}).get("tenant") for post in posts } == { "alpha", "beta", } posts_by_trace: dict[str, list[dict[str, Any]]] = {} for post in posts: posts_by_trace.setdefault(post["trace_id"], []).append(post) assert len(posts_by_trace) == 2 assert all(len(trace_posts) == 3 for trace_posts in posts_by_trace.values()) @pytest.mark.skipif( sys.version_info < (3, 11), reason="Asyncio context vars require Python 3.11+" ) async def test_langsmith_inheritable_metadata_mixed_sync_async_managers_isolated() -> ( None ): """Sync and async manager configure paths can overlap without metadata sharing.""" tracer = _create_tracer_with_mocked_client() @RunnableLambda async def async_runnable(x: int) -> int: await asyncio.sleep(0) return x + 1 @RunnableLambda def sync_runnable(x: int) -> int: return x + 1 async def run_sync() -> int: callbacks = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_metadata={"path": "sync"}, ) return await asyncio.to_thread( sync_runnable.invoke, 1, {"callbacks": callbacks} ) async def run_async() -> int: callbacks = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_metadata={"path": "async"}, ) return await async_runnable.ainvoke(1, {"callbacks": callbacks}) await asyncio.gather(run_sync(), run_async()) posts = _get_posts(tracer.client) assert len(posts) == 2 assert { post.get("extra", {}).get("metadata", {}).get("path") for post in posts } == { "sync", "async", } class TestLangsmithInheritableTracingDefaultsInConfigure: """Tests for LangSmith inheritable tracing defaults in configure.""" def test_langsmith_inheritable_metadata_applied_via_configure(self) -> None: """langsmith_inheritable_metadata flows to a copied tracer.""" tracer = _create_tracer_with_mocked_client() cm = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_metadata={"env": "prod", "service": "api"}, ) lc_tracers = [h for h in cm.handlers if isinstance(h, LangChainTracer)] assert len(lc_tracers) == 1 assert lc_tracers[0] is not tracer assert lc_tracers[0].tracing_metadata == {"env": "prod", "service": "api"} assert tracer.tracing_metadata is None def test_langsmith_inheritable_metadata_does_not_overwrite_tracer_metadata( self, ) -> None: """Tracer metadata takes precedence over langsmith_inheritable_metadata.""" tracer = _create_tracer_with_mocked_client(metadata={"env": "staging"}) cm = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_metadata={"env": "prod", "service": "api"}, ) lc_tracer = next(h for h in cm.handlers if isinstance(h, LangChainTracer)) assert tracer.tracing_metadata == {"env": "staging"} assert lc_tracer.tracing_metadata == {"env": "staging", "service": "api"} def test_tracing_context_metadata_merged_into_langsmith_inheritable_metadata( self, ) -> None: """Tracing-context metadata merges into tracer defaults. LangSmith metadata keeps precedence on collisions. """ tracer = _create_tracer_with_mocked_client() with tracing_context( enabled=True, client=tracer.client, metadata={"trace_only": "value", "shared": "trace"}, ): cm = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_metadata={ "shared": "langsmith", "tenant": "alpha", }, ) lc_tracer = next(h for h in cm.handlers if isinstance(h, LangChainTracer)) assert lc_tracer.tracing_metadata == { "trace_only": "value", "shared": "langsmith", "tenant": "alpha", } def test_langsmith_inheritable_metadata_end_to_end(self) -> None: """langsmith_inheritable_metadata in configure propagates to posted runs.""" tracer = _create_tracer_with_mocked_client() @RunnableLambda def my_func(x: int) -> int: return x # Use langsmith_inheritable_metadata through the config callbacks path cm = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_metadata={"env": "prod"}, ) my_func.invoke(1, {"callbacks": cm}) posts = _get_posts(tracer.client) assert len(posts) == 1 md = posts[0].get("extra", {}).get("metadata", {}) assert md["env"] == "prod" def test_runnable_config_copies_configurable_values_to_tracing_metadata( self, ) -> None: tracer = _create_tracer_with_mocked_client() @RunnableLambda def my_func(x: int) -> int: return x config: RunnableConfig = { "callbacks": [tracer], "metadata": { "something": "else", "checkpoint_ns": "from-metadata", "model": "from-metadata", }, "configurable": { "thread_id": "th-123", "checkpoint_id": "ckpt-1", "checkpoint_ns": "from-configurable", "task_id": "task-1", "run_id": "run-456", "assistant_id": "asst-789", "graph_id": "graph-0", "model": "from-configurable", "user_id": "uid-1", "cron_id": "cron-1", "langgraph_auth_user_id": "user-1", "api_key": "should-not-propagate", "__secret_key": "should-not-propagate", "temperature": 0.5, "streaming": True, "custom_setting": {"nested": True}, "none_value": None, }, } my_func.invoke(1, config) posts = _get_posts(tracer.client) assert len(posts) == 1 md = posts[0].get("extra", {}).get("metadata", {}) assert { key: md[key] for key in ( "something", "thread_id", "checkpoint_id", "task_id", "run_id", "assistant_id", "graph_id", "user_id", "cron_id", "langgraph_auth_user_id", "temperature", "streaming", "model", "checkpoint_ns", ) } == { "something": "else", "thread_id": "th-123", "checkpoint_id": "ckpt-1", "task_id": "task-1", "run_id": "run-456", "assistant_id": "asst-789", "graph_id": "graph-0", "user_id": "uid-1", "cron_id": "cron-1", "langgraph_auth_user_id": "user-1", "temperature": 0.5, "streaming": True, "model": "from-metadata", "checkpoint_ns": "from-metadata", } assert "api_key" not in md assert "__secret_key" not in md assert "custom_setting" not in md assert "none_value" not in md def test_langsmith_inheritable_metadata_does_not_affect_non_tracer_handlers( self, ) -> None: """langsmith_inheritable_metadata only applies to tracers.""" tracer = _create_tracer_with_mocked_client() received_metadata: list[dict[str, Any]] = [] class MetadataCapture(BaseCallbackHandler): def on_chain_start(self, *_args: Any, **kwargs: Any) -> None: received_metadata.append(dict(kwargs.get("metadata", {}))) capture = MetadataCapture() cm = CallbackManager.configure( inheritable_callbacks=[tracer, capture], langsmith_inheritable_metadata={"tracer_only": "yes"}, ) @RunnableLambda def my_func(x: int) -> int: return x my_func.invoke(1, {"callbacks": cm}) # Non-tracer handler should NOT see langsmith_inheritable_metadata assert len(received_metadata) >= 1 for md in received_metadata: assert "tracer_only" not in md # But the tracer's posted runs SHOULD have it posts = _get_posts(tracer.client) assert len(posts) >= 1 for post in posts: post_md = post.get("extra", {}).get("metadata", {}) assert post_md["tracer_only"] == "yes" def test_no_langsmith_inheritable_metadata_is_noop(self) -> None: """Passing langsmith_inheritable_metadata=None does not alter tracer state.""" tracer = _create_tracer_with_mocked_client() cm = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_metadata=None, ) lc_tracer = next(h for h in cm.handlers if isinstance(h, LangChainTracer)) assert lc_tracer is tracer assert tracer.tracing_metadata is None def test_langsmith_inheritable_tags_applied_via_configure(self) -> None: """langsmith_inheritable_tags flow to a copied tracer.""" tracer = _create_tracer_with_mocked_client() tracer.tags = ["existing"] cm = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_tags=["tenant:alpha", "existing"], ) lc_tracer = next(h for h in cm.handlers if isinstance(h, LangChainTracer)) assert lc_tracer is not tracer assert lc_tracer.tags == ["existing", "tenant:alpha"] assert tracer.tags == ["existing"] def test_inheritable_tags_do_not_affect_non_tracer_handlers(self) -> None: """langsmith_inheritable_tags only apply to tracers.""" tracer = _create_tracer_with_mocked_client() received_tags: list[list[str]] = [] class TagCapture(BaseCallbackHandler): def on_chain_start(self, *_args: Any, **kwargs: Any) -> None: received_tags.append(list(kwargs.get("tags", []))) capture = TagCapture() cm = CallbackManager.configure( inheritable_callbacks=[tracer, capture], langsmith_inheritable_tags=["tracer-only"], ) @RunnableLambda def my_func(x: int) -> int: return x my_func.invoke(1, {"callbacks": cm}) assert received_tags assert all("tracer-only" not in tags for tags in received_tags) posts = _get_posts(tracer.client) assert posts assert all("tracer-only" in post.get("tags", []) for post in posts) def test_langsmith_inheritable_metadata_copies_handlers_without_mutating_original( self, ) -> None: """Configured manager copies tracers and leaves the original unchanged.""" tracer = _create_tracer_with_mocked_client() cm = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_metadata={"env": "prod"}, ) handler_tracer = next(h for h in cm.handlers if isinstance(h, LangChainTracer)) inheritable_tracer = next( h for h in cm.inheritable_handlers if isinstance(h, LangChainTracer) ) assert handler_tracer is not tracer assert inheritable_tracer is not tracer assert tracer.tracing_metadata is None def test_langsmith_inheritable_metadata_configure_isolated_per_manager( self, ) -> None: """Separate configure calls keep tracer-only defaults isolated.""" tracer = _create_tracer_with_mocked_client() alpha_manager = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_metadata={"tenant": "alpha"}, ) beta_manager = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_metadata={"tenant": "beta"}, ) alpha_tracer = next( handler for handler in alpha_manager.handlers if isinstance(handler, LangChainTracer) ) beta_tracer = next( handler for handler in beta_manager.handlers if isinstance(handler, LangChainTracer) ) assert tracer.tracing_metadata is None assert alpha_tracer is not tracer assert beta_tracer is not tracer assert alpha_tracer is not beta_tracer assert alpha_tracer.tracing_metadata == {"tenant": "alpha"} assert beta_tracer.tracing_metadata == {"tenant": "beta"} assert alpha_tracer.run_map is tracer.run_map assert beta_tracer.run_map is tracer.run_map assert alpha_tracer.order_map is tracer.order_map assert beta_tracer.order_map is tracer.order_map def test_inheritable_metadata_concurrent_invocations_remain_isolated( self, ) -> None: """Parallel invocations through copied tracers keep metadata separated.""" tracer = _create_tracer_with_mocked_client() barrier = threading.Barrier(2) @traceable def traced_leaf(x: int) -> int: barrier.wait() return x @RunnableLambda def my_func(x: int) -> int: return traced_leaf(x) def invoke_for_tenant(tenant: str, value: int) -> int: callbacks = CallbackManager.configure( inheritable_callbacks=[tracer], langsmith_inheritable_metadata={"tenant": tenant}, ) return my_func.invoke(value, {"callbacks": callbacks}) with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: list(executor.map(invoke_for_tenant, ["alpha", "beta"], [1, 2])) posts = _get_posts(tracer.client) assert len(posts) == 4 assert {post["name"] for post in posts} == {"my_func", "traced_leaf"} my_func_posts = [post for post in posts if post["name"] == "my_func"] assert len(my_func_posts) == 2 assert { post.get("extra", {}).get("metadata", {}).get("tenant") for post in my_func_posts } == {"alpha", "beta"} assert tracer.run_map == {} assert len(tracer.order_map) == 2