import concurrent.futures import threading import time import unittest.mock import uuid from typing import Any from uuid import UUID import pytest from langsmith import Client from langsmith.run_trees import RunTree from langsmith.utils import get_env_var, get_tracer_project from langchain_core.messages import AIMessage from langchain_core.outputs import ChatGeneration, LLMResult from langchain_core.tracers.langchain import ( LangChainTracer, _get_usage_metadata_from_generations, _patch_missing_metadata, ) from langchain_core.tracers.schemas import Run def test_example_id_assignment_threadsafe() -> None: """Test that example assigned at callback start/end is honored.""" example_ids = {} def mock_create_run(**kwargs: Any) -> Any: example_ids[kwargs.get("id")] = kwargs.get("reference_example_id") return unittest.mock.MagicMock() client = unittest.mock.MagicMock(spec=Client) client.tracing_queue = None client.create_run = mock_create_run tracer = LangChainTracer(client=client) old_persist_run_single = tracer._persist_run_single def new_persist_run_single(run: Run) -> None: time.sleep(0.01) old_persist_run_single(run) with unittest.mock.patch.object( tracer, "_persist_run_single", new=new_persist_run_single ): run_id_1 = UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160a") run_id_2 = UUID("f1f9fa53-8b2f-4742-bdbc-38215f7bd1e1") run_id_3 = UUID("f1f9fa53-8b2f-4742-bdbc-38215f7cd1e1") example_id_1 = UUID("57e42c57-8c79-4d9f-8765-bf6cd3a98055") tracer.example_id = example_id_1 tracer.on_llm_start({"name": "example_1"}, ["foo"], run_id=run_id_1) tracer.on_llm_end(LLMResult(generations=[], llm_output={}), run_id=run_id_1) example_id_2 = UUID("4f31216e-7c26-4027-a5fd-0bbf9ace17dc") tracer.example_id = example_id_2 tracer.on_llm_start({"name": "example_2"}, ["foo"], run_id=run_id_2) tracer.on_llm_end(LLMResult(generations=[], llm_output={}), run_id=run_id_2) tracer.example_id = None tracer.on_chain_start( {"name": "no_examples"}, {"inputs": (i for i in range(10))}, run_id=run_id_3 ) tracer.on_chain_error(ValueError("Foo bar"), run_id=run_id_3) expected_example_ids = { run_id_1: example_id_1, run_id_2: example_id_2, run_id_3: None, } tracer.wait_for_futures() assert example_ids == expected_example_ids def test_tracer_with_run_tree_parent() -> None: mock_session = unittest.mock.MagicMock() client = Client(session=mock_session, api_key="test") parent = RunTree(name="parent", inputs={"input": "foo"}, ls_client=client) run_id = uuid.uuid4() tracer = LangChainTracer(client=client) tracer.order_map[parent.id] = (parent.trace_id, parent.dotted_order) tracer.run_map[str(parent.id)] = parent tracer.on_chain_start( {"name": "child"}, {"input": "bar"}, run_id=run_id, parent_run_id=parent.id ) tracer.on_chain_end({}, run_id=run_id) assert parent.child_runs assert len(parent.child_runs) == 1 assert parent.child_runs[0].id == run_id assert parent.child_runs[0].trace_id == parent.id assert parent.child_runs[0].parent_run_id == parent.id def test_log_lock() -> None: """Test that example assigned at callback start/end is honored.""" client = unittest.mock.MagicMock(spec=Client) tracer = LangChainTracer(client=client) with unittest.mock.patch.object(tracer, "_persist_run_single", new=lambda _: _): run_id_1 = UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160a") lock = threading.Lock() tracer.on_chain_start({"name": "example_1"}, {"input": lock}, run_id=run_id_1) tracer.on_chain_end({}, run_id=run_id_1) tracer.wait_for_futures() @pytest.mark.parametrize( ("envvars", "expected_project_name"), [ ( {}, "default", ), ( {"LANGCHAIN_SESSION": "old_timey_session"}, "old_timey_session", ), ( { "LANGCHAIN_SESSION": "old_timey_session", "LANGCHAIN_PROJECT": "modern_session", }, "modern_session", ), ], ids=[ "default to 'default' when no project provided", "use session_name for legacy tracers", "use LANGCHAIN_PROJECT over SESSION_NAME", ], ) def test_correct_get_tracer_project( envvars: dict[str, str], expected_project_name: str ) -> None: if hasattr(get_env_var, "cache_clear"): get_env_var.cache_clear() # type: ignore[attr-defined] if hasattr(get_tracer_project, "cache_clear"): get_tracer_project.cache_clear() with pytest.MonkeyPatch.context() as mp: for k, v in envvars.items(): mp.setenv(k, v) client = unittest.mock.MagicMock(spec=Client) tracer = LangChainTracer(client=client) projects = [] def mock_create_run(**kwargs: Any) -> Any: projects.append(kwargs.get("session_name")) return unittest.mock.MagicMock() client.create_run = mock_create_run tracer.on_llm_start( {"name": "example_1"}, ["foo"], run_id=UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160a"), ) tracer.wait_for_futures() assert projects == [expected_project_name] @pytest.mark.parametrize( ("generations", "expected"), [ # Returns None for non-serialized message usage_metadata shape # (earlier regression) ( [ [ { "text": "Hello!", "message": { "content": "Hello!", "usage_metadata": { "input_tokens": 10, "output_tokens": 20, "total_tokens": 30, }, }, } ] ], None, ), # Returns usage_metadata when message is serialized via dumpd ( [ [ { "text": "Hello!", "message": { "lc": 1, "type": "constructor", "id": ["langchain", "schema", "messages", "AIMessage"], "kwargs": { "content": "Hello!", "type": "ai", "usage_metadata": { "input_tokens": 10, "output_tokens": 20, "total_tokens": 30, }, "tool_calls": [], "invalid_tool_calls": [], }, }, } ] ], {"input_tokens": 10, "output_tokens": 20, "total_tokens": 30}, ), # Returns None when no usage_metadata ([[{"text": "Hello!", "message": {"content": "Hello!"}}]], None), # Returns None when no message ([[{"text": "Hello!"}]], None), # Returns None for empty generations ([], None), ([[]], None), # Aggregates usage_metadata across multiple generations ( [ [ { "text": "First", "message": { "lc": 1, "type": "constructor", "id": ["langchain", "schema", "messages", "AIMessage"], "kwargs": { "content": "First", "type": "ai", "usage_metadata": { "input_tokens": 5, "output_tokens": 10, "total_tokens": 15, }, "tool_calls": [], "invalid_tool_calls": [], }, }, }, { "text": "Second", "message": { "lc": 1, "type": "constructor", "id": ["langchain", "schema", "messages", "AIMessage"], "kwargs": { "content": "Second", "type": "ai", "usage_metadata": { "input_tokens": 50, "output_tokens": 100, "total_tokens": 150, }, "tool_calls": [], "invalid_tool_calls": [], }, }, }, ] ], {"input_tokens": 55, "output_tokens": 110, "total_tokens": 165}, ), # Finds usage_metadata across multiple batches ( [ [{"text": "No message here"}], [ { "text": "Has message", "message": { "lc": 1, "type": "constructor", "id": ["langchain", "schema", "messages", "AIMessage"], "kwargs": { "content": "Has message", "type": "ai", "usage_metadata": { "input_tokens": 10, "output_tokens": 20, "total_tokens": 30, }, "tool_calls": [], "invalid_tool_calls": [], }, }, } ], ], {"input_tokens": 10, "output_tokens": 20, "total_tokens": 30}, ), ], ids=[ "returns_none_when_non_serialized_message_shape", "returns_usage_metadata_when_message_serialized", "returns_none_when_no_usage_metadata", "returns_none_when_no_message", "returns_none_for_empty_list", "returns_none_for_empty_batch", "aggregates_across_multiple_generations", "finds_across_multiple_batches", ], ) def test_get_usage_metadata_from_generations( generations: list[list[dict[str, Any]]], expected: dict[str, Any] | None ) -> None: """Test `_get_usage_metadata_from_generations` utility function.""" result = _get_usage_metadata_from_generations(generations) assert result == expected def test_on_llm_end_stores_usage_metadata_in_run_extra() -> None: """Test that `usage_metadata` is stored in `run.extra.metadata` on llm end.""" client = unittest.mock.MagicMock(spec=Client) client.tracing_queue = None tracer = LangChainTracer(client=client) run_id = UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160a") tracer.on_llm_start({"name": "test_llm"}, ["foo"], run_id=run_id) run = tracer.run_map[str(run_id)] usage_metadata = {"input_tokens": 100, "output_tokens": 200, "total_tokens": 300} run.outputs = { "generations": [ [ { "text": "Hello!", "message": { "lc": 1, "type": "constructor", "id": ["langchain", "schema", "messages", "AIMessage"], "kwargs": { "content": "Hello!", "type": "ai", "usage_metadata": usage_metadata, "tool_calls": [], "invalid_tool_calls": [], }, }, } ] ] } captured_run = None def capture_run(r: Run) -> None: nonlocal captured_run captured_run = r with unittest.mock.patch.object(tracer, "_update_run_single", capture_run): tracer._on_llm_end(run) assert captured_run is not None assert "metadata" in captured_run.extra assert captured_run.extra["metadata"]["usage_metadata"] == usage_metadata def test_on_llm_end_stores_usage_metadata_from_serialized_outputs() -> None: """Store `usage_metadata` from serialized generation message outputs.""" client = unittest.mock.MagicMock(spec=Client) client.tracing_queue = None tracer = LangChainTracer(client=client) run_id = UUID("d94d0ff8-cf5a-4100-ab11-1a0efaa8d8d0") tracer.on_llm_start({"name": "test_llm"}, ["foo"], run_id=run_id) usage_metadata = {"input_tokens": 100, "output_tokens": 200, "total_tokens": 300} response = LLMResult( generations=[ [ ChatGeneration( message=AIMessage(content="Hello!", usage_metadata=usage_metadata) ) ] ] ) run = tracer._complete_llm_run(response=response, run_id=run_id) captured_run = None def capture_run(r: Run) -> None: nonlocal captured_run captured_run = r with unittest.mock.patch.object(tracer, "_update_run_single", capture_run): tracer._on_llm_end(run) assert captured_run is not None assert "metadata" in captured_run.extra assert captured_run.extra["metadata"]["usage_metadata"] == usage_metadata def test_on_llm_end_no_usage_metadata_when_not_present() -> None: """Test that no `usage_metadata` is added when not present in outputs.""" client = unittest.mock.MagicMock(spec=Client) client.tracing_queue = None tracer = LangChainTracer(client=client) run_id = UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160a") tracer.on_llm_start({"name": "test_llm"}, ["foo"], run_id=run_id) run = tracer.run_map[str(run_id)] run.outputs = { "generations": [ [ { "text": "Hello!", "message": { "lc": 1, "type": "constructor", "id": ["langchain", "schema", "messages", "AIMessage"], "kwargs": { "content": "Hello!", "type": "ai", "tool_calls": [], "invalid_tool_calls": [], }, }, } ] ] } captured_run = None def capture_run(r: Run) -> None: nonlocal captured_run captured_run = r with unittest.mock.patch.object(tracer, "_update_run_single", capture_run): tracer._on_llm_end(run) assert captured_run is not None extra_metadata = captured_run.extra.get("metadata", {}) assert "usage_metadata" not in extra_metadata def test_on_llm_end_preserves_existing_metadata() -> None: """Test that existing metadata is preserved when adding `usage_metadata`.""" client = unittest.mock.MagicMock(spec=Client) client.tracing_queue = None tracer = LangChainTracer(client=client) run_id = UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160a") tracer.on_llm_start( {"name": "test_llm"}, ["foo"], run_id=run_id, metadata={"existing_key": "existing_value"}, ) run = tracer.run_map[str(run_id)] usage_metadata = {"input_tokens": 10, "output_tokens": 20, "total_tokens": 30} run.outputs = { "generations": [ [ { "text": "Hello!", "message": { "lc": 1, "type": "constructor", "id": ["langchain", "schema", "messages", "AIMessage"], "kwargs": { "content": "Hello!", "type": "ai", "usage_metadata": usage_metadata, "tool_calls": [], "invalid_tool_calls": [], }, }, } ] ] } captured_run = None def capture_run(r: Run) -> None: nonlocal captured_run captured_run = r with unittest.mock.patch.object(tracer, "_update_run_single", capture_run): tracer._on_llm_end(run) assert captured_run is not None assert "metadata" in captured_run.extra assert captured_run.extra["metadata"]["usage_metadata"] == usage_metadata assert captured_run.extra["metadata"]["existing_key"] == "existing_value" def test_on_chain_start_skips_persist_when_defers_inputs() -> None: """Test that `_on_chain_start` skips persist when `defers_inputs` is set.""" client = unittest.mock.MagicMock(spec=Client) client.tracing_queue = None tracer = LangChainTracer(client=client) run_id = UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160a") # Pass defers_inputs=True to signal deferred inputs tracer.on_chain_start( {"name": "test_chain"}, {"input": ""}, run_id=run_id, defers_inputs=True, ) run = tracer.run_map[str(run_id)] persist_called = False def mock_persist() -> None: nonlocal persist_called persist_called = True with unittest.mock.patch.object(tracer, "_persist_run_single", mock_persist): tracer._on_chain_start(run) # Should NOT call persist when defers_inputs is set assert not persist_called def test_on_chain_start_persists_when_not_defers_inputs() -> None: """Test that `_on_chain_start` persists when `defers_inputs` is not set.""" client = unittest.mock.MagicMock(spec=Client) client.tracing_queue = None tracer = LangChainTracer(client=client) run_id = UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160a") # Normal chain start without defers_inputs tracer.on_chain_start( {"name": "test_chain"}, {"input": "hello"}, run_id=run_id, ) run = tracer.run_map[str(run_id)] persist_called = False def mock_persist(_: Any) -> None: nonlocal persist_called persist_called = True with unittest.mock.patch.object(tracer, "_persist_run_single", mock_persist): tracer._on_chain_start(run) # Should call persist when defers_inputs is not set assert persist_called def test_on_chain_end_persists_when_defers_inputs() -> None: """Test that `_on_chain_end` calls persist (POST) when `defers_inputs` is set.""" client = unittest.mock.MagicMock(spec=Client) client.tracing_queue = None tracer = LangChainTracer(client=client) run_id = UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160a") tracer.on_chain_start( {"name": "test_chain"}, {"input": ""}, run_id=run_id, defers_inputs=True, ) run = tracer.run_map[str(run_id)] run.outputs = {"output": "result"} run.inputs = {"input": "realized input"} persist_called = False update_called = False def mock_persist(_: Any) -> None: nonlocal persist_called persist_called = True def mock_update(_: Any) -> None: nonlocal update_called update_called = True with ( unittest.mock.patch.object(tracer, "_persist_run_single", mock_persist), unittest.mock.patch.object(tracer, "_update_run_single", mock_update), ): tracer._on_chain_end(run) # Should call persist (POST), not update (PATCH) for deferred inputs assert persist_called assert not update_called def test_on_chain_end_updates_when_not_defers_inputs() -> None: """Tests `_on_chain_end` calls update (PATCH) when `defers_inputs` is not set.""" client = unittest.mock.MagicMock(spec=Client) client.tracing_queue = None tracer = LangChainTracer(client=client) run_id = UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160a") tracer.on_chain_start( {"name": "test_chain"}, {"input": "hello"}, run_id=run_id, ) run = tracer.run_map[str(run_id)] run.outputs = {"output": "result"} persist_called = False update_called = False def mock_persist(_: Any) -> None: nonlocal persist_called persist_called = True def mock_update(_: Any) -> None: nonlocal update_called update_called = True with ( unittest.mock.patch.object(tracer, "_persist_run_single", mock_persist), unittest.mock.patch.object(tracer, "_update_run_single", mock_update), ): tracer._on_chain_end(run) # Should call update (PATCH), not persist (POST) for normal inputs assert not persist_called assert update_called def test_on_chain_error_persists_when_defers_inputs() -> None: """Test that `_on_chain_error` calls persist (POST) when `defers_inputs` is set.""" client = unittest.mock.MagicMock(spec=Client) client.tracing_queue = None tracer = LangChainTracer(client=client) run_id = UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160a") tracer.on_chain_start( {"name": "test_chain"}, {"input": ""}, run_id=run_id, defers_inputs=True, ) run = tracer.run_map[str(run_id)] run.error = "Test error" run.inputs = {"input": "realized input"} persist_called = False update_called = False def mock_persist(_: Any) -> None: nonlocal persist_called persist_called = True def mock_update(_: Any) -> None: nonlocal update_called update_called = True with ( unittest.mock.patch.object(tracer, "_persist_run_single", mock_persist), unittest.mock.patch.object(tracer, "_update_run_single", mock_update), ): tracer._on_chain_error(run) # Should call persist (POST), not update (PATCH) for deferred inputs assert persist_called assert not update_called def test_on_chain_error_updates_when_not_defers_inputs() -> None: """Tests `_on_chain_error` calls update (PATCH) when `defers_inputs` is not set.""" client = unittest.mock.MagicMock(spec=Client) client.tracing_queue = None tracer = LangChainTracer(client=client) run_id = UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160a") tracer.on_chain_start( {"name": "test_chain"}, {"input": "hello"}, run_id=run_id, ) run = tracer.run_map[str(run_id)] run.error = "Test error" persist_called = False update_called = False def mock_persist(_: Any) -> None: nonlocal persist_called persist_called = True def mock_update(_: Any) -> None: nonlocal update_called update_called = True with ( unittest.mock.patch.object(tracer, "_persist_run_single", mock_persist), unittest.mock.patch.object(tracer, "_update_run_single", mock_update), ): tracer._on_chain_error(run) # Should call update (PATCH), not persist (POST) for normal inputs assert not persist_called assert update_called class TestPatchMissingMetadata: """Tests for `_patch_missing_metadata` and tracer metadata behavior.""" @staticmethod def _make_tracer( metadata: dict[str, str] | None = None, ) -> LangChainTracer: client = unittest.mock.MagicMock(spec=Client) client.tracing_queue = None return LangChainTracer(client=client, metadata=metadata) @staticmethod def _make_run( metadata: dict[str, Any] | None = None, ) -> Run: return Run( id=uuid.uuid4(), name="test", inputs={}, run_type="chain", extra={"metadata": metadata or {}}, ) def test_adds_metadata_when_run_has_none(self) -> None: """Tracer metadata fills in when the run has no matching keys.""" tracer = self._make_tracer(metadata={"env": "prod", "service": "api"}) run = self._make_run() _patch_missing_metadata(tracer, run) assert run.metadata["env"] == "prod" assert run.metadata["service"] == "api" def test_does_not_overwrite_existing_keys(self) -> None: """Config metadata takes precedence over tracer metadata.""" tracer = self._make_tracer(metadata={"env": "prod", "service": "api"}) run = self._make_run(metadata={"env": "staging"}) _patch_missing_metadata(tracer, run) assert run.metadata["env"] == "staging" assert run.metadata["service"] == "api" def test_noop_when_tracer_has_no_metadata(self) -> None: """No-op when the tracer has no metadata configured.""" tracer = self._make_tracer(metadata=None) run = self._make_run(metadata={"existing": "value"}) _patch_missing_metadata(tracer, run) assert run.metadata == {"existing": "value"} def test_noop_when_all_keys_already_present(self) -> None: """No-op when every tracer key already exists in the run.""" tracer = self._make_tracer(metadata={"env": "prod"}) run = self._make_run(metadata={"env": "dev"}) _patch_missing_metadata(tracer, run) assert run.metadata == {"env": "dev"} def test_merges_disjoint_keys(self) -> None: """Disjoint keys from tracer and config are all present after patching.""" tracer = self._make_tracer(metadata={"tracer_key": "tracer_val"}) run = self._make_run(metadata={"config_key": "config_val"}) _patch_missing_metadata(tracer, run) assert run.metadata == { "tracer_key": "tracer_val", "config_key": "config_val", } def test_persist_run_single_applies_tracer_metadata(self) -> None: """End-to-end: `_persist_run_single` calls `_patch_missing_metadata`.""" tracer = self._make_tracer(metadata={"env": "prod"}) run_id = UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160a") tracer.on_chain_start( {"name": "test_chain"}, {"input": "hello"}, run_id=run_id, ) run = tracer.run_map[str(run_id)] with unittest.mock.patch.object(Run, "post"): tracer._persist_run_single(run) assert run.metadata.get("env") == "prod" def test_persist_run_single_config_metadata_wins(self) -> None: """Config metadata is not overwritten by tracer metadata during persist.""" tracer = self._make_tracer(metadata={"env": "prod", "extra": "from_tracer"}) run_id = UUID("9d878ab3-e5ca-4218-aef6-44cbdc90160b") tracer.on_chain_start( {"name": "test_chain"}, {"input": "hello"}, run_id=run_id, metadata={"env": "staging"}, ) run = tracer.run_map[str(run_id)] with unittest.mock.patch.object(Run, "post"): tracer._persist_run_single(run) assert run.metadata["env"] == "staging" assert run.metadata["extra"] == "from_tracer" def test_allowlisted_key_overrides_existing_run_metadata(self) -> None: """Allowlisted LangSmith keys override existing run metadata.""" tracer = self._make_tracer(metadata={"ls_agent_type": "subagent"}) run = self._make_run(metadata={"ls_agent_type": "root", "other": "keep"}) _patch_missing_metadata(tracer, run) assert run.metadata["ls_agent_type"] == "subagent" assert run.metadata["other"] == "keep" def test_allowlisted_key_noop_when_values_match(self) -> None: """Allowlisted keys do not clone run metadata when the value is unchanged.""" original = {"ls_agent_type": "root"} tracer = self._make_tracer(metadata={"ls_agent_type": "root"}) run = self._make_run(metadata=original) _patch_missing_metadata(tracer, run) # No-op: the shared dict should not be replaced with a copy. assert run.extra["metadata"] is original assert run.metadata == {"ls_agent_type": "root"} class TestTracerMetadataCloning: """Tests for LangChainTracer metadata cloning helpers.""" @staticmethod def _make_tracer( metadata: dict[str, str] | None = None, ) -> LangChainTracer: client = unittest.mock.MagicMock(spec=Client) client.tracing_queue = None return LangChainTracer(client=client, metadata=metadata) def test_copy_with_metadata_defaults_copies_configuration(self) -> None: """Copied tracer keeps stable configuration but not identity.""" tracer = self._make_tracer(metadata={"env": "staging"}) tracer.project_name = "project" tracer.tags = ["tag"] copied = tracer.copy_with_metadata_defaults(metadata={"service": "api"}) assert copied is not tracer assert copied.client is tracer.client assert copied.project_name == "project" assert copied.tags == ["tag"] assert copied.tags is tracer.tags assert copied.tracing_metadata == {"env": "staging", "service": "api"} assert copied.run_map is tracer.run_map assert copied.order_map is tracer.order_map assert copied.run_has_token_event_map == {} def test_copy_with_metadata_defaults_does_not_mutate_original(self) -> None: """Metadata-default cloning leaves the source tracer unchanged.""" tracer = self._make_tracer(metadata={"env": "staging"}) copied = tracer.copy_with_metadata_defaults(metadata={"service": "api"}) assert tracer.tracing_metadata == {"env": "staging"} assert copied.tracing_metadata == {"env": "staging", "service": "api"} def test_copy_with_metadata_defaults_none_preserves_configuration(self) -> None: """Copying without new metadata preserves metadata and shared run state.""" tracer = self._make_tracer(metadata={"env": "staging"}) copied = tracer.copy_with_metadata_defaults(metadata=None) assert copied is not tracer assert copied.tracing_metadata == {"env": "staging"} assert copied.run_map is tracer.run_map assert copied.order_map is tracer.order_map def test_copy_with_metadata_defaults_threadsafe(self) -> None: """Concurrent metadata-default copies do not mutate each other or the source.""" tracer = self._make_tracer(metadata={"env": "staging"}) def copy_for_service(service: str) -> dict[str, str]: copied = tracer.copy_with_metadata_defaults(metadata={"service": service}) assert copied is not tracer return copied.tracing_metadata or {} with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: metadata_values = list(executor.map(copy_for_service, ["api", "worker"])) assert tracer.tracing_metadata == {"env": "staging"} assert {metadata["service"] for metadata in metadata_values} == { "api", "worker", } assert all(metadata["env"] == "staging" for metadata in metadata_values) def test_copy_with_metadata_defaults_threadsafe_with_existing_shared_state( self, ) -> None: """Concurrent copies preserve pre-populated shared run state.""" tracer = self._make_tracer(metadata={"env": "staging"}) run_id = uuid.uuid4() tracer.run_map["existing"] = unittest.mock.MagicMock() tracer.order_map[run_id] = (run_id, f"prefix.{run_id}") def copy_for_service(service: str) -> LangChainTracer: copied = tracer.copy_with_metadata_defaults(metadata={"service": service}) assert copied.run_map is tracer.run_map assert copied.order_map is tracer.order_map return copied with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: copied_tracers = list(executor.map(copy_for_service, ["api", "worker"])) assert tracer.run_map.keys() == {"existing"} assert tracer.order_map == {run_id: (run_id, f"prefix.{run_id}")} copied_services = { copied.tracing_metadata["service"] for copied in copied_tracers if copied.tracing_metadata is not None } assert copied_services == {"api", "worker"} def test_copy_with_metadata_defaults_regular_keys_first_wins(self) -> None: """Regular (non-allowlisted) metadata keys keep "first wins" semantics.""" tracer = self._make_tracer(metadata={"env": "staging", "service": "orig"}) copied = tracer.copy_with_metadata_defaults( metadata={"env": "prod", "service": "new"}, ) assert copied.tracing_metadata == {"env": "staging", "service": "orig"} def test_copy_with_metadata_defaults_allowlisted_key_overrides(self) -> None: """Allowlisted LangSmith keys are overridden by nested caller metadata.""" tracer = self._make_tracer( metadata={"ls_agent_type": "root", "env": "staging"}, ) copied = tracer.copy_with_metadata_defaults( metadata={"ls_agent_type": "subagent", "env": "prod"}, ) # Allowlisted key is overridden, non-allowlisted keeps first-wins. assert copied.tracing_metadata == { "ls_agent_type": "subagent", "env": "staging", }