From 3851ac79651cbf33c1c2756b27efd960d2caefec Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 20 Oct 2026 07:59:31 +0000 Subject: [PATCH 1/5] fix(experiments): flush off the event loop in run_experiment The async experiment runner called the blocking flush() at the end of the run, which froze the caller's event loop while scores and spans were sent. Co-authored-by: hassieb.pakzad --- langfuse/_client/client.py | 4 ++-- tests/unit/test_experiment.py | 38 +++++++++++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 2 deletions(-) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index fe80d7e85..00214db20 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -2735,8 +2735,8 @@ async def process_item(item: ExperimentItem) -> ExperimentItemResult: except Exception as e: langfuse_logger.error("Failed to store run evaluation: %s", e) - # Flush scores and traces - self.flush() + # Flush scores and traces off the event loop; flush() blocks on network I/O. + await asyncio.to_thread(self.flush) return ExperimentResult( name=name, diff --git a/tests/unit/test_experiment.py b/tests/unit/test_experiment.py index fea3587b2..deca750c0 100644 --- a/tests/unit/test_experiment.py +++ b/tests/unit/test_experiment.py @@ -1,7 +1,9 @@ """Tests for ``langfuse.experiment`` — ``RunnerContext`` and ``RegressionError``.""" +import asyncio import inspect import json +import time import typing from datetime import datetime, timezone from typing import get_type_hints @@ -779,3 +781,39 @@ def test_tracing_disabled_skips_project_id_lookup(self, monkeypatch): get_project_id.assert_not_called() assert result.experiment_url is None + + +class TestExperimentEventLoop: + @pytest.mark.asyncio + async def test_final_flush_does_not_block_event_loop( + self, langfuse_memory_client, monkeypatch + ): + monkeypatch.setattr(langfuse_memory_client, "_get_project_id", lambda: "p") + monkeypatch.setattr(langfuse_memory_client, "flush", lambda: time.sleep(0.5)) + stop = asyncio.Event() + gaps: list = [] + + async def heartbeat(): + last = time.monotonic() + while not stop.is_set(): + await asyncio.sleep(0.01) + now = time.monotonic() + gaps.append(now - last) + last = now + + beat = asyncio.create_task(heartbeat()) + await langfuse_memory_client._run_experiment_async( + name="exp", + run_name="run", + description=None, + data=[{"input": "a"}], + task=lambda *, item, **kwargs: item["input"], + evaluators=[], + composite_evaluator=None, + run_evaluators=[], + max_concurrency=1, + ) + stop.set() + await beat + + assert max(gaps) < 0.25 From 87f222c30c83a262634d3ec523d3dd5268944aea Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 20 Oct 2026 07:59:52 +0000 Subject: [PATCH 2/5] fix(scores): read LANGFUSE_FLUSH_AT and LANGFUSE_FLUSH_INTERVAL in the score consumer Only the span processor read these variables, so scores kept the 15 / 1s defaults when batching was configured through the environment. Co-authored-by: hassieb.pakzad --- langfuse/_task_manager/score_consumer.py | 10 +++++ tests/unit/test_score_consumer.py | 48 ++++++++++++++++++++++++ 2 files changed, 58 insertions(+) diff --git a/langfuse/_task_manager/score_consumer.py b/langfuse/_task_manager/score_consumer.py index cdeb81404..3e45e74fa 100644 --- a/langfuse/_task_manager/score_consumer.py +++ b/langfuse/_task_manager/score_consumer.py @@ -8,6 +8,10 @@ import backoff import httpx +from langfuse._client.environment_variables import ( + LANGFUSE_FLUSH_AT, + LANGFUSE_FLUSH_INTERVAL, +) from langfuse._utils.parse_error import handle_fern_exception from langfuse._utils.serializer import EventSerializer from langfuse.api import LangfuseAPI @@ -71,6 +75,12 @@ def __init__( self._score_queue = score_queue self._identifier = identifier self._api_client = api_client + env_flush_at = os.environ.get(LANGFUSE_FLUSH_AT) + if flush_at is None and env_flush_at is not None: + flush_at = int(env_flush_at) + env_flush_interval = os.environ.get(LANGFUSE_FLUSH_INTERVAL) + if flush_interval is None and env_flush_interval is not None: + flush_interval = float(env_flush_interval) self._flush_at = flush_at or 15 self._flush_interval = flush_interval or 1 self._max_retries = max_retries or 3 diff --git a/tests/unit/test_score_consumer.py b/tests/unit/test_score_consumer.py index a2818d495..754a53520 100644 --- a/tests/unit/test_score_consumer.py +++ b/tests/unit/test_score_consumer.py @@ -243,6 +243,54 @@ def test_does_not_retry_client_errors(status, caplog): assert caplog.records +def test_reads_flush_at_from_env(monkeypatch): + monkeypatch.setenv("LANGFUSE_FLUSH_AT", "2") + recorder = Recorder(_responder(ACCEPTED)) + queue: Queue = Queue() + for name in ("a", "b", "c"): + queue.put(_score(name)) + + _drain(_consumer(recorder, queue), queue) + + assert [[s["name"] for s in body] for body in recorder.bodies] == [ + ["a", "b"], + ["c"], + ] + + +def test_reads_flush_interval_from_env(monkeypatch): + monkeypatch.setenv("LANGFUSE_FLUSH_INTERVAL", "0.05") + recorder = Recorder(_responder(ACCEPTED)) + consumer = ScoreConsumer( + score_queue=Queue(), + identifier=0, + api_client=LangfuseAPI( + base_url="http://langfuse.test", + username="pk", + password="sk", + httpx_client=httpx.Client(transport=httpx.MockTransport(recorder)), + ), + ) + + started = time.monotonic() + consumer.upload() + + # The default window is 1 second. + assert time.monotonic() - started < 0.5 + + +def test_constructor_flush_settings_win_over_env(monkeypatch): + monkeypatch.setenv("LANGFUSE_FLUSH_AT", "1") + recorder = Recorder(_responder(ACCEPTED)) + queue: Queue = Queue() + for name in ("a", "b", "c"): + queue.put(_score(name)) + + _drain(_consumer(recorder, queue, flush_at=3), queue) + + assert [[s["name"] for s in body] for body in recorder.bodies] == [["a", "b", "c"]] + + def test_pause_sends_carried_over_score(monkeypatch): first_request_started = threading.Event() release_first_request = threading.Event() From 7a56020046b8666a9b6ec63c72ae3081a1d1751d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 20 Oct 2026 07:59:52 +0000 Subject: [PATCH 3/5] fix(scores): send pending scores immediately on flush() flush() waited on the score queue while the consumer kept collecting until flush_interval ran out, so a single queued score made flush() block for up to the full interval. flush() now wakes live consumers so the pending batch is sent right away. Co-authored-by: hassieb.pakzad --- langfuse/_client/resource_manager.py | 5 +++ langfuse/_task_manager/score_consumer.py | 21 ++++++----- tests/unit/test_score_consumer.py | 44 ++++++++++++++++++++++++ 3 files changed, 62 insertions(+), 8 deletions(-) diff --git a/langfuse/_client/resource_manager.py b/langfuse/_client/resource_manager.py index b5e104812..1d62b4703 100644 --- a/langfuse/_client/resource_manager.py +++ b/langfuse/_client/resource_manager.py @@ -593,6 +593,11 @@ def flush(self) -> None: self.tracer_provider.force_flush() langfuse_logger.debug("Successfully flushed OTEL tracer provider") + for score_consumer in self._score_consumers: + # A stopped consumer would leave the signal unacknowledged and + # block join() below. + if score_consumer.is_alive(): + score_consumer.wake() self._score_queue.join() langfuse_logger.debug("Successfully flushed score queue") diff --git a/langfuse/_task_manager/score_consumer.py b/langfuse/_task_manager/score_consumer.py index 3e45e74fa..87d5c7d27 100644 --- a/langfuse/_task_manager/score_consumer.py +++ b/langfuse/_task_manager/score_consumer.py @@ -36,7 +36,8 @@ _SCORES_BODY_LIMIT_BYTES, ) MAX_BATCH_SIZE_BYTES = min(_configured_batch_size_bytes, _SCORES_BODY_LIMIT_BYTES) -_SHUTDOWN_SENTINEL = object() +# Ends the batch being collected so it is sent right away. +_WAKE_SIGNAL = object() def _is_retryable(error: Exception) -> bool: @@ -109,7 +110,7 @@ def _next(self) -> List[Dict[str, Any]]: block=True, timeout=self._flush_interval - elapsed ) - if score is _SHUTDOWN_SENTINEL: + if score is _WAKE_SIGNAL: self._score_queue.task_done() break @@ -209,16 +210,20 @@ def upload(self) -> None: for _ in batch: self._score_queue.task_done() - def pause(self) -> None: - """Pause the consumer.""" - self.running = False + def wake(self) -> None: + """Send the batch being collected now instead of after flush_interval.""" try: - self._score_queue.put(_SHUTDOWN_SENTINEL, block=False) + self._score_queue.put(_WAKE_SIGNAL, block=False) except Full: - # If the queue is full, the consumer will wake up naturally while - # draining items, so a dedicated shutdown signal is not required. + # A full queue fills the batch to flush_at without waiting, so no + # signal is needed. pass + def pause(self) -> None: + """Pause the consumer.""" + self.running = False + self.wake() + def _upload_batch(self, batch: List[Dict[str, Any]]) -> None: logger.debug("API: Sending batch of %s scores to Langfuse API", len(batch)) diff --git a/tests/unit/test_score_consumer.py b/tests/unit/test_score_consumer.py index 754a53520..f23a331fb 100644 --- a/tests/unit/test_score_consumer.py +++ b/tests/unit/test_score_consumer.py @@ -291,6 +291,24 @@ def test_constructor_flush_settings_win_over_env(monkeypatch): assert [[s["name"] for s in body] for body in recorder.bodies] == [["a", "b", "c"]] +def test_wake_sends_pending_batch_before_flush_interval(): + recorder = Recorder(_responder(ACCEPTED)) + queue: Queue = Queue() + consumer = _consumer(recorder, queue, flush_interval=30) + consumer.start() + queue.put(_score("a")) + + consumer.wake() + finished = threading.Event() + threading.Thread(target=lambda: (queue.join(), finished.set()), daemon=True).start() + + assert finished.wait(timeout=5) + assert [[s["name"] for s in body] for body in recorder.bodies] == [["a"]] + assert consumer.is_alive() + consumer.pause() + consumer.join(timeout=5) + + def test_pause_sends_carried_over_score(monkeypatch): first_request_started = threading.Event() release_first_request = threading.Event() @@ -399,6 +417,32 @@ def test_create_score_posts_to_scores_endpoint(client_factory): assert sent[1]["timestamp"].endswith("Z") +def test_flush_sends_pending_score_without_waiting_for_flush_interval( + client_factory, +): + recorder = Recorder(_responder(ACCEPTED)) + client = client_factory(recorder, flush_interval=5) + + client.create_score(name="a", value=1.0, trace_id="a" * 32) + started = time.monotonic() + client.flush() + elapsed = time.monotonic() - started + + assert [s["name"] for body in recorder.bodies for s in body] == ["a"] + assert elapsed < 1 + + +def test_flush_after_shutdown_returns(client_factory): + client = client_factory(Recorder(_responder(ACCEPTED))) + client.shutdown() + + flush = threading.Thread(target=client.flush, daemon=True) + flush.start() + flush.join(timeout=2) + + assert not flush.is_alive() + + def test_create_score_payload_matches_request_model_serialization(client_factory): recorder = Recorder(_responder(ACCEPTED)) client = client_factory(recorder, environment="staging") From 87c50c24f4ff8b1596b2ad0d8cebbb0783574f8a Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 20 Oct 2026 08:04:22 +0000 Subject: [PATCH 4/5] fix(scores): stop waiting in flush() once no score consumer is alive A flush() racing shutdown() could put its wake signal on the queue after the consumer had already exited. Nothing acknowledged it, so Queue.join() never returned. flush() now waits on the queue's condition and stops when every consumer has stopped. Co-authored-by: hassieb.pakzad --- langfuse/_client/resource_manager.py | 21 +++++++++++++++++--- tests/unit/test_score_consumer.py | 29 ++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 3 deletions(-) diff --git a/langfuse/_client/resource_manager.py b/langfuse/_client/resource_manager.py index 1d62b4703..3e4221777 100644 --- a/langfuse/_client/resource_manager.py +++ b/langfuse/_client/resource_manager.py @@ -594,16 +594,31 @@ def flush(self) -> None: langfuse_logger.debug("Successfully flushed OTEL tracer provider") for score_consumer in self._score_consumers: - # A stopped consumer would leave the signal unacknowledged and - # block join() below. if score_consumer.is_alive(): score_consumer.wake() - self._score_queue.join() + self._join_score_queue() langfuse_logger.debug("Successfully flushed score queue") self._media_upload_queue.join() langfuse_logger.debug("Successfully flushed media upload queue") + def _join_score_queue(self) -> None: + """Wait until the score queue is drained or no consumer is left to drain it. + + Queue.join() would block forever when the consumers stop first, e.g. + when shutdown() runs concurrently and a wake signal is never taken. + """ + score_queue = self._score_queue + with score_queue.all_tasks_done: + while score_queue.unfinished_tasks: + if not any(c.is_alive() for c in self._score_consumers): + langfuse_logger.debug( + "Score consumers stopped with %s queue item(s) unprocessed", + score_queue.unfinished_tasks, + ) + return + score_queue.all_tasks_done.wait(timeout=0.1) + def shutdown(self) -> None: self._shutdown = True diff --git a/tests/unit/test_score_consumer.py b/tests/unit/test_score_consumer.py index f23a331fb..cc6611d29 100644 --- a/tests/unit/test_score_consumer.py +++ b/tests/unit/test_score_consumer.py @@ -443,6 +443,35 @@ def test_flush_after_shutdown_returns(client_factory): assert not flush.is_alive() +def test_flush_returns_when_consumer_stops_after_liveness_check(client_factory): + client = client_factory(Recorder(_responder(ACCEPTED))) + consumer = client._resources._score_consumers[0] + wake = consumer.wake + + def stop_then_wake() -> None: + # Shutdown stops the consumer between flush()'s liveness check and its + # wake signal, so nothing acknowledges the signal. + consumer.running = False + wake() + consumer.join(timeout=5) + wake() + + consumer.wake = stop_then_wake # type: ignore[method-assign] + + flush = threading.Thread(target=client.flush, daemon=True) + flush.start() + flush.join(timeout=2) + flush_hung = flush.is_alive() + # Acknowledge the orphaned signal so a hung flush() cannot block teardown. + score_queue = client._resources._score_queue + while not score_queue.empty(): + score_queue.get_nowait() + score_queue.task_done() + + assert not consumer.is_alive() + assert not flush_hung + + def test_create_score_payload_matches_request_model_serialization(client_factory): recorder = Recorder(_responder(ACCEPTED)) client = client_factory(recorder, environment="staging") From ce6abc80706776e85518a06a645949aac1ec64cc Mon Sep 17 00:00:00 2001 From: Hassieb Pakzad <68423100+hassiebp@users.noreply.github.com> Date: Tue, 20 Oct 2026 12:17:11 +0400 Subject: [PATCH 5/5] push --- langfuse/_client/resource_manager.py | 4 ++++ langfuse/_task_manager/score_consumer.py | 7 +++---- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/langfuse/_client/resource_manager.py b/langfuse/_client/resource_manager.py index 3e4221777..e633cdfaa 100644 --- a/langfuse/_client/resource_manager.py +++ b/langfuse/_client/resource_manager.py @@ -596,6 +596,7 @@ def flush(self) -> None: for score_consumer in self._score_consumers: if score_consumer.is_alive(): score_consumer.wake() + self._join_score_queue() langfuse_logger.debug("Successfully flushed score queue") @@ -609,6 +610,7 @@ def _join_score_queue(self) -> None: when shutdown() runs concurrently and a wake signal is never taken. """ score_queue = self._score_queue + with score_queue.all_tasks_done: while score_queue.unfinished_tasks: if not any(c.is_alive() for c in self._score_consumers): @@ -616,7 +618,9 @@ def _join_score_queue(self) -> None: "Score consumers stopped with %s queue item(s) unprocessed", score_queue.unfinished_tasks, ) + return + score_queue.all_tasks_done.wait(timeout=0.1) def shutdown(self) -> None: diff --git a/langfuse/_task_manager/score_consumer.py b/langfuse/_task_manager/score_consumer.py index 87d5c7d27..21d9f5052 100644 --- a/langfuse/_task_manager/score_consumer.py +++ b/langfuse/_task_manager/score_consumer.py @@ -36,7 +36,6 @@ _SCORES_BODY_LIMIT_BYTES, ) MAX_BATCH_SIZE_BYTES = min(_configured_batch_size_bytes, _SCORES_BODY_LIMIT_BYTES) -# Ends the batch being collected so it is sent right away. _WAKE_SIGNAL = object() @@ -76,12 +75,15 @@ def __init__( self._score_queue = score_queue self._identifier = identifier self._api_client = api_client + env_flush_at = os.environ.get(LANGFUSE_FLUSH_AT) if flush_at is None and env_flush_at is not None: flush_at = int(env_flush_at) + env_flush_interval = os.environ.get(LANGFUSE_FLUSH_INTERVAL) if flush_interval is None and env_flush_interval is not None: flush_interval = float(env_flush_interval) + self._flush_at = flush_at or 15 self._flush_interval = flush_interval or 1 self._max_retries = max_retries or 3 @@ -211,12 +213,9 @@ def upload(self) -> None: self._score_queue.task_done() def wake(self) -> None: - """Send the batch being collected now instead of after flush_interval.""" try: self._score_queue.put(_WAKE_SIGNAL, block=False) except Full: - # A full queue fills the batch to flush_at without waiting, so no - # signal is needed. pass def pause(self) -> None: