From 4a8986dd725b041b1aa3e8f738b66b3a579a656b Mon Sep 17 00:00:00 2001 From: mrinal Date: Tue, 15 Sep 2026 20:38:35 +0000 Subject: [PATCH 1/2] fix(langsmith): keep events appended during an in-flight flush instead of clearing them Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/integrations/langsmith.py | 2 ++ .../integrations/test_langsmith_init.py | 32 +++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/litellm/integrations/langsmith.py b/litellm/integrations/langsmith.py index 9607eccef52..32664ed75d2 100644 --- a/litellm/integrations/langsmith.py +++ b/litellm/integrations/langsmith.py @@ -39,6 +39,8 @@ def is_serializable(value): class LangsmithLogger(CustomBatchLogger): + preserve_events_added_during_flush = True + def __init__( self, langsmith_api_key: str | None = None, diff --git a/tests/test_litellm/integrations/test_langsmith_init.py b/tests/test_litellm/integrations/test_langsmith_init.py index 0bc9e279fbf..4b4b94da22d 100644 --- a/tests/test_litellm/integrations/test_langsmith_init.py +++ b/tests/test_litellm/integrations/test_langsmith_init.py @@ -531,3 +531,35 @@ class TestLangsmithRootRunIdConsistency: assert data["trace_id"] == "trace-1" assert data["dotted_order"] == dotted + + +@pytest.mark.asyncio +async def test_events_appended_during_flush_are_not_dropped(): + logger = LangsmithLogger(langsmith_api_key="test-key", langsmith_project="test-project") + sent_batches: list[list[dict]] = [] + late_event = {"credentials": logger.default_credentials, "data": {"id": "late"}} + + async def fake_post(url, json, headers): + if not sent_batches: + logger.log_queue.append(late_event) + sent_batches.append(json["post"]) + response = MagicMock() + response.status_code = 200 + response.raise_for_status = MagicMock() + return response + + logger.async_httpx_client = MagicMock(post=AsyncMock(side_effect=fake_post)) + logger.log_queue = [ + {"credentials": logger.default_credentials, "data": {"id": "a"}}, + {"credentials": logger.default_credentials, "data": {"id": "b"}}, + ] + + await logger.flush_queue() + + assert [e["id"] for e in sent_batches[0]] == ["a", "b"] + assert logger.log_queue == [late_event] + + await logger.flush_queue() + + assert [e["id"] for e in sent_batches[1]] == ["late"] + assert logger.log_queue == [] From 4c179f2f59375d9c86390ab6b36cf67f10f1e157 Mon Sep 17 00:00:00 2001 From: mrinal Date: Tue, 15 Sep 2026 21:17:36 +0000 Subject: [PATCH 2/2] test(langsmith): type the flush race test and cancel its periodic task Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../integrations/test_langsmith_init.py | 53 +++++++++++-------- 1 file changed, 32 insertions(+), 21 deletions(-) diff --git a/tests/test_litellm/integrations/test_langsmith_init.py b/tests/test_litellm/integrations/test_langsmith_init.py index 4b4b94da22d..f56d2310e73 100644 --- a/tests/test_litellm/integrations/test_langsmith_init.py +++ b/tests/test_litellm/integrations/test_langsmith_init.py @@ -1,5 +1,6 @@ import asyncio import os +from typing import Final from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -7,6 +8,7 @@ import pytest import litellm from litellm.integrations.langsmith import LangsmithLogger +from litellm.types.integrations.langsmith import LangsmithQueueObject @pytest.fixture @@ -536,30 +538,39 @@ class TestLangsmithRootRunIdConsistency: @pytest.mark.asyncio async def test_events_appended_during_flush_are_not_dropped(): logger = LangsmithLogger(langsmith_api_key="test-key", langsmith_project="test-project") - sent_batches: list[list[dict]] = [] - late_event = {"credentials": logger.default_credentials, "data": {"id": "late"}} + try: + sent_batches: Final[list[list[dict[str, str]]]] = [] + late_event: Final = LangsmithQueueObject( + credentials=logger.default_credentials, data={"id": "late"} + ) - async def fake_post(url, json, headers): - if not sent_batches: - logger.log_queue.append(late_event) - sent_batches.append(json["post"]) - response = MagicMock() - response.status_code = 200 - response.raise_for_status = MagicMock() - return response + async def fake_post( + url: str, json: dict[str, list[dict[str, str]]], headers: dict[str, str] + ) -> MagicMock: + if not sent_batches: + logger.log_queue.append(late_event) + sent_batches.append(json["post"]) + response = MagicMock() + response.status_code = 200 + response.raise_for_status = MagicMock() + return response - logger.async_httpx_client = MagicMock(post=AsyncMock(side_effect=fake_post)) - logger.log_queue = [ - {"credentials": logger.default_credentials, "data": {"id": "a"}}, - {"credentials": logger.default_credentials, "data": {"id": "b"}}, - ] + logger.async_httpx_client = MagicMock(post=AsyncMock(side_effect=fake_post)) + logger.log_queue = [ + LangsmithQueueObject(credentials=logger.default_credentials, data={"id": "a"}), + LangsmithQueueObject(credentials=logger.default_credentials, data={"id": "b"}), + ] - await logger.flush_queue() + await logger.flush_queue() - assert [e["id"] for e in sent_batches[0]] == ["a", "b"] - assert logger.log_queue == [late_event] + assert [e["id"] for e in sent_batches[0]] == ["a", "b"] + assert logger.log_queue == [late_event] - await logger.flush_queue() + await logger.flush_queue() - assert [e["id"] for e in sent_batches[1]] == ["late"] - assert logger.log_queue == [] + assert [e["id"] for e in sent_batches[1]] == ["late"] + assert logger.log_queue == [] + finally: + if logger._flush_task is not None: + logger._flush_task.cancel() + await asyncio.gather(logger._flush_task, return_exceptions=True)