[DavidTraina]

What does this PR do?

Media logged after shutdown() makes flush() block forever, and the next shutdown() too, since it calls flush() first.

shutdown() stops the media upload threads but keeps the resource manager registered and the tracer provider running. Langfuse(...) with the same public key and get_client() keep using it, and spans keep exporting. Media in their input, output or metadata still goes on the upload queue, where no consumer is left to take it, and flush() ends with self._media_upload_queue.join(). Media logged while shutdown() stops the consumers can be stranded the same way: a consumer that is busy when it is paused exits after its current upload, and one that takes its shutdown signal first never reaches an upload queued behind it.

This hits code that creates a client and calls shutdown() at the end of each request or job: the first later request that logs media (a LangfuseMedia, or a base64 data URI in span input) hangs.

import threading

from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter

from langfuse import Langfuse
from langfuse.media import LangfuseMedia

langfuse = Langfuse(
    public_key="pk-lf-test",
    secret_key="sk-lf-test",
    span_exporter=InMemorySpanExporter(),
)
langfuse.shutdown()

# e.g. the next request; Langfuse(...) with the same key and get_client()
# return clients backed by the same shut-down resources
with langfuse.start_as_current_observation(
    name="request",
    input=LangfuseMedia(content_bytes=b"image", content_type="image/png"),
):
    pass

flush = threading.Thread(target=langfuse.flush, daemon=True)
flush.start()
flush.join(timeout=5)
print("flush() returned:", not flush.is_alive())  # main: False, this PR: True

Fix: as with scores created after shutdown(), media found once shutdown() has finished its own flush() is dropped with a warning instead of queued, so join() does not wait on uploads that no consumer will take.

• MediaManager.stop_accepting_uploads() flips a flag, which _process_media() checks right before put().
• shutdown() flips it right after its own flush(), not when shutdown() starts, so media that flush finds is still uploaded. It then joins the media queue once more before stopping the consumers, so media queued between that flush() and the stop is uploaded too.
• After the consumers have stopped, shutdown() calls MediaManager.drop_pending_uploads(), which marks done anything still on the queue. That covers a thread that read the flag just before it flipped and calls put() while the consumers stop.

Unchanged:

• flush(). Its latency matches main (below).
• Media logged before shutdown(), or found while shutdown() exports pending spans, is uploaded (POST, PUT, PATCH), along with pending spans and scores.
• A later shutdown() still force-flushes spans created after the first one.
• The atexit shutdown(), reset(), fork reinit, and Langfuse(...) / get_client() reusing the shut-down resources.
• No new lock: the change only uses the queue's put(), get_nowait(), task_done() and join().

Media logged after shutdown() is still not uploaded, as on main, where nothing takes it off the queue either. Its spans still export with the media reference.

Not covered: a thread that reads the flag before it flips but calls put() only after shutdown() has returned, for example when a signal handler runs shutdown() between the two. Its upload is left on the queue, and the next flush() hangs, as on main.

Considered and not done

• Stopping intake when shutdown() starts, as for scores. The exporter finds base64 data URIs in span attributes only when it exports the span, which for the last spans happens during shutdown()'s own flush(). Stopping intake at the start skipped such a data URI (0 uploads, against 1 on main), and also media logged between that flush() and the stop.
• A lock around the flag check and put(), instead of the drain. It also closes the case above, but under gevent with the stdlib queue.Queue (gevent before 25.4.1, or GEVENT_MONKEY_DISABLE_QUEUE_QUEUE=true), and with media logged both from greenlets and from native threads (the hub threadpool or OS threads), a threading.RLock there made media logging hang in 8 of 8 runs, against 0 of 8 on main. Without the lock and without the drain, a stress run (8 threads logging media while shutdown() runs) left a later flush() hung in 1 of 1200 iterations on Python 3.14 and 1 of 300 on 3.11. With the drain: none of 1200, 300 and 300 (3.14, free-threaded 3.14t, 3.11).
• Bounding flush()'s media wait by consumer liveness instead. This fixes the hang too, including the case this PR does not cover, but every media job logged after shutdown() then stays on the queue, bytes included, for the life of the process: 1000 spans with a 100 KB image each, logged after shutdown(), kept 102 MB allocated (RSS 65 to 178 MB), against 1.4 MB with this PR. With a client and a shutdown() per request, that grows with every request until the 100,000-item queue cap. An event-driven wait also needs stdlib's undocumented Queue.all_tasks_done plus a separate path for gevent's patched Queue, which lacks it. Using only the public queue API means polling, and the poll interval sets the cost: in one run with one pending 20 ms upload, the median flush() was 28.5 ms on main, 109 ms with a 0.1 s poll and 38 ms with a 10 ms poll.

Type of change

• Bug fix
• New feature
• Breaking change
• Refactor
• Documentation update
• Tooling, CI, or repo maintenance

Verification

uv run --frozen ruff check .                         # All checks passed!
uv run --frozen ruff format --check <changed files>  # 3 files already formatted
uv run --frozen mypy langfuse --no-error-summary     # exit 0
uv run --frozen pytest -n auto --dist worksteal tests/unit   # 963 passed, 2 skipped (Python 3.14, includes the gevent tests)
UV_PROJECT_ENVIRONMENT=/tmp/lf-py311 uv run --frozen --python 3.11 pytest tests/unit/test_media_manager.py tests/unit/test_resource_manager.py tests/unit/test_score_consumer.py   # 70 passed

New tests in tests/unit/test_resource_manager.py, all failing on main:

• test_flush_returns_when_media_is_logged_after_shutdown: on main, flush() was still blocked after 2 s.
• test_shutdown_uploads_media_logged_after_its_flush: logs two media items right after shutdown()'s own flush(), while the single consumer is busy with the first. On main, one was processed and the other was left on the queue.
• test_flush_returns_when_media_is_queued_while_consumers_stop: queues an upload right after the consumers stop, as a thread that read the flag just before it flipped would. On main, the next flush() was still blocked after 2 s.

Each runs the call on a thread with a timeout and marks leftover queue items done before asserting, so a regression fails instead of hanging the test or the autouse reset(). Removing the flag check, the second join() or the drain each fails exactly one of them.

flush() latency, median of 200 calls, 3 interleaved runs (Python 3.14):

	main	this PR
empty	1 µs	1 µs
one pending media upload	0.29-0.33 ms	0.27-0.31 ms
one pending 20 ms upload	29.3-29.6 ms	29.4-30.0 ms

Also checked with scripts that are not part of this PR, on Python 3.14, 3.14t (free-threaded) and 3.11 (gevent on 3.14 only):

• Media logged from another thread at each step of shutdown(), alone and with a concurrent flush() or second shutdown(): nothing hung, and every upload queued before the stop was sent exactly once (150 media: 150 POST, 150 PUT, 150 PATCH). On main, the cases with media hung at 4 of the steps.
• Fork children of a live parent (upload in flight) and of a shut-down parent: the child logs media, flushes, shuts down and exits 0.
• atexit: a script that exits without calling shutdown() delivers its media and score.
• Real gevent with monkey.patch_all():
  • Media logged after shutdown() hangs the next flush() and shutdown() on main and not here (gevent 24.11.1, 25.9.1 and 26.9.0; 26.9.0 also with GEVENT_MONKEY_DISABLE_QUEUE_QUEUE=true).
  • Greenlets, hub-threadpool threads and OS threads logging media together finish in every run, as on main (24.11.1; 25.9.1 and 26.9.0, each with and without that variable).
  • 8 greenlets logging media while another runs shutdown() uploaded everything queued before the stop, with no duplicates (25.9.1).

Skipped: tests/e2e and tests/live_provider need a Langfuse server and provider keys; the behavior is covered by the unit tests.

Checklist

• I self-reviewed the diff using code_review.md.
• I added or updated tests for behavior changes.
• I updated docs, examples, or .env.template if needed. (Not needed.)
• I did not hand-edit generated files; if generated files changed, I used the upstream regeneration path.
• I did not commit secrets or credentials.