fix(resource-manager): prevent stale clients from queueing after shutdown

[trungminhdo4-glitch]

Summary

Fixes a shutdown lifecycle race where a LangfuseResourceManager can accept
new score, trace, or media tasks after its background consumers have stopped.

Those tasks increment Queue.unfinished_tasks, but no live consumer remains to
call task_done(), so a later flush() / shutdown() can block indefinitely
in Queue.join().

This change treats shutdown() as terminal for the shared resource manager:

• score/trace admission is synchronized with shutdown;
• stale clients drop asynchronous work after shutdown instead of queueing it;
• media admission is synchronized with its shutdown boundary;
• pending OTEL spans may still discover/enqueue media during force_flush();
• concurrent shutdown callers wait for the active shutdown to finish;
• fork reinitialization recreates the newly introduced synchronization state.

Why not recreate the manager automatically?

A previous approach in #1800 evicted the shut-down manager so a new same-key
client could create fresh workers.

This PR intentionally does not do that.

LangfuseSpanProcessor instances are registered on the shared OpenTelemetry
TracerProvider, which does not expose a normal processor-removal API.
Automatically recreating managers therefore needs a separate processor-lifetime
design to avoid accumulating registered processors.

The current SDK documentation describes shutdown() as application-terminal,
so this PR keeps the smaller contract: post-shutdown work cannot poison queues
or deadlock later flushes.

Regression coverage

Five durable tests (deterministic Event/Barrier synchronization, no sleep-based timing, assert unfinished_tasks == 0):

• test_1799_stale_client_score_trace_dropped_after_shutdown — stale score/trace clients cannot enqueue after shutdown;
• test_1799_admission_vs_shutdown_atomicity — admission vs shutdown is atomic;
• test_1799_concurrent_shutdown_completion — concurrent shutdown waits for actual completion;
• test_1799_media_admission_toctou_and_begin_shutdown — media admission cannot cross the final shutdown boundary (tests/unit/test_media.py);
• test_1799_media_during_force_flush_not_dropped_and_after_shutdown_dropped — media discovered during OTEL force_flush() is still processed before media shutdown.

Verification

• uv run --frozen pytest tests/unit/test_resource_manager.py — 16 passed
• uv run --frozen pytest tests/unit/test_media.py — 21 passed
• uv run --frozen pytest tests/unit/test_otel.py — 65 passed, 2 skipped
• uv run --frozen ruff check langfuse/_client/resource_manager.py langfuse/_task_manager/media_manager.py tests/unit/test_resource_manager.py tests/unit/test_media.py — passed
• uv run --frozen ruff format --check langfuse/_client/resource_manager.py langfuse/_task_manager/media_manager.py tests/unit/test_resource_manager.py tests/unit/test_media.py — passed
• uv run --frozen mypy langfuse --no-error-summary — passed
• git diff --check — passed

Broader unit run had two pre-existing test_prompt_atexit failures also reproducible on main (not introduced by this change).

Refs #1799

Greptile Summary

This PR makes resource-manager shutdown terminal and synchronizes score, trace, and media admission with consumer teardown. It also makes concurrent shutdown callers wait for completion, recreates lifecycle synchronization after forks, and adds deterministic race regression tests.

• Adds atomic shutdown checks around score and trace queue admission.
• Adds media-manager shutdown state and synchronizes queue insertion with its shutdown boundary.
• Coordinates concurrent shutdown callers through a completion event.
• Adds lifecycle and queue-draining regression coverage.

Confidence Score: 3/5

The PR should not merge until reentrant shutdown and partially admitted media traversal are handled without deadlock or dangling media references.

A callback can re-enter shutdown on the thread that must set its completion event, and an independent media traversal can cross begin_shutdown after rewriting payload values but before their upload jobs are admitted.

Files Needing Attention: langfuse/_client/resource_manager.py and langfuse/_task_manager/media_manager.py

Sequence Diagram

sequenceDiagram
    participant App as Application thread
    participant RM as Resource manager
    participant OTel as Tracer provider
    participant MM as Media manager
    participant Q as Background queues
    App->>RM: shutdown()
    RM->>RM: set terminal shutdown state
    RM->>OTel: force_flush()
    OTel->>MM: discover and enqueue media
    MM->>Q: enqueue upload jobs
    RM->>Q: join queues
    RM->>MM: begin_shutdown()
    RM->>RM: stop and join consumers
    RM->>RM: set shutdown completion event
    Note over App,RM: Concurrent callers wait for completion

Loading

Prompt To Fix All With AI

### Issue 1
langfuse/_client/resource_manager.py:650-655
**Reentrant shutdown self-deadlocks**

When a user-supplied span-export callback synchronously calls `shutdown()` during the initial shutdown's `force_flush()`, this branch waits on `_shutdown_event` from the same thread responsible for eventually setting it, causing shutdown to block indefinitely.

### Issue 2
langfuse/_task_manager/media_manager.py:297-302
**Shutdown creates dangling media references**

When an application thread is traversing a multi-media payload as another thread reaches `begin_shutdown()`, later `_process_media` calls return without queueing uploads while their callers still replace the payload values with media references, causing those referenced media objects never to be uploaded.

### Issue 3
tests/unit/test_resource_manager.py:457
**Test imports remain function-local**

The new regression tests place imports inside their test functions, with the same pattern repeated in the new media test and throughout this file. Move these imports to module scope to keep dependencies discoverable and follow the repository's import-placement rule.

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

_{Reviews (1): Last reviewed commit: "test(resource-manager): harden #1799 shu..." | Re-trigger Greptile}

Greptile also left 2 inline comments on this PR.

Context used (3)

• Rule used - Move imports to the top of the module instead of p... (source)

Learned From
langfuse/langfuse-python#1387

• Knowledge Base — Client Core
• Knowledge Base — Task Manager: Media Upload and Score Ingestion

[CLAassistant]

All committers have signed the CLA.

[claude]

Claude Code Review

This pull request is from a fork — automated review is disabled. A repository maintainer can comment @claude review to run a one-time review.

[greptile-apps]

Reentrant shutdown self-deadlocks

When a user-supplied span-export callback synchronously calls shutdown() during the initial shutdown's force_flush(), this branch waits on _shutdown_event from the same thread responsible for eventually setting it, causing shutdown to block indefinitely.

Knowledge Base Used: Client Core

Prompt To Fix With AI

This is a comment left during a code review.
Path: langfuse/_client/resource_manager.py
Line: 650-655

Comment:
**Reentrant shutdown self-deadlocks**

When a user-supplied span-export callback synchronously calls `shutdown()` during the initial shutdown's `force_flush()`, this branch waits on `_shutdown_event` from the same thread responsible for eventually setting it, causing shutdown to block indefinitely.

**Knowledge Base Used:** [Client Core](https://app.greptile.com/personal-org-4986/-/custom-context/knowledge-base/langfuse/langfuse-python/-/docs/client-core.md)

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

[greptile-apps]

Shutdown creates dangling media references

When an application thread is traversing a multi-media payload as another thread reaches begin_shutdown(), later _process_media calls return without queueing uploads while their callers still replace the payload values with media references, causing those referenced media objects never to be uploaded.

Knowledge Base Used:

• Client Core
• Task Manager: Media Upload and Score Ingestion

Prompt To Fix With AI

This is a comment left during a code review.
Path: langfuse/_task_manager/media_manager.py
Line: 297-302

Comment:
**Shutdown creates dangling media references**

When an application thread is traversing a multi-media payload as another thread reaches `begin_shutdown()`, later `_process_media` calls return without queueing uploads while their callers still replace the payload values with media references, causing those referenced media objects never to be uploaded.

**Knowledge Base Used:**
- [Client Core](https://app.greptile.com/personal-org-4986/-/custom-context/knowledge-base/langfuse/langfuse-python/-/docs/client-core.md)
- [Task Manager: Media Upload and Score Ingestion](https://app.greptile.com/personal-org-4986/-/custom-context/knowledge-base/langfuse/langfuse-python/-/docs/task-manager.md)

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.