[hassiebp]

What does this PR do?

flush() with a single queued score blocked for most of the score consumer's flush_interval, and longer against a slow server. The async experiment runner also called that blocking flush() on the event loop. This PR fixes both, with no public API changes:

1. flush() wakes the score consumer. Before waiting on the score queue, flush() puts a wake signal on it, so the consumer stops collecting and sends the pending batch right away. This reuses the signal pause() already sent, renamed to _WAKE_SIGNAL, and adds ScoreConsumer.wake().
2. flush() no longer hangs when consumers stop under it. If flush() runs concurrently with shutdown(), the consumer can exit before it takes the wake signal. Nothing acknowledges the signal, so Queue.join() would never return. flush() now waits on the queue's all_tasks_done condition, with a 0.1 s timeout as a backstop, and stops waiting once no consumer is alive (logged at debug). The wake-up still comes from the task_done() notify, so normal latency doesn't change.
3. _run_experiment_async flushes off the loop with await asyncio.to_thread(self.flush).
4. The score consumer reads LANGFUSE_FLUSH_AT / LANGFUSE_FLUSH_INTERVAL. Previously only the span processor read them, so scores stayed at 15 / 1 s when batching was configured through the environment. Constructor arguments still take precedence. Defaults are unchanged when nothing is set.

Measured (local fake server; one score, then flush(), in a loop)

setup	before (main 685dc884)	after
defaults	1003 ms	2 ms
flush_interval=5	5003 ms	2 ms
env LANGFUSE_FLUSH_INTERVAL=5	1003 ms	2 ms
server takes 2 s per score POST	3003 ms	2003 ms (upload time is still waited on, as intended)
_run_experiment_async (5 items) awaited on the caller's loop, max loop gap	884 ms	9 ms
same, with a 2 s score server	2996 ms	13 ms

The public run_experiment() called from inside a running loop still blocks that loop for the whole run, because run_async_safely joins a thread. Fixing that needs a public async entry point and is out of scope here.

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 .             # 575 files already formatted
uv run --frozen mypy langfuse --no-error-summary  # exit 0
uv run --frozen pytest -n auto --dist worksteal tests/unit   # 941 passed, 2 skipped
# score consumer / experiment / resource manager suites, 3 repeated runs: 75 passed each

New tests were written first; six of them failed before their fix:

• test_reads_flush_at_from_env: got one batch [a, b, c] instead of [a, b], [c]
• test_reads_flush_interval_from_env: waited 1.0 s, expected under 0.5 s
• test_wake_sends_pending_batch_before_flush_interval: wake did not exist
• test_flush_sends_pending_score_without_waiting_for_flush_interval: 5.0 s, expected under 1 s
• test_final_flush_does_not_block_event_loop: loop gap 0.50 s, expected under 0.25 s
• test_flush_returns_when_consumer_stops_after_liveness_check: flush() still blocked after 2 s. The test stops the consumer between the liveness check and the wake signal, and acknowledges the orphaned signal itself so a hang can't block teardown.

test_constructor_flush_settings_win_over_env and test_flush_after_shutdown_returns pass on main as well. They guard behavior that this change could break.

Skipped: tests/e2e and tests/live_provider need a Langfuse server and provider keys. The behavior is covered by unit tests using a mock transport.

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: the env vars are already documented, and the score consumer now matches those docs.)
• I did not hand-edit generated files; if generated files changed, I used the upstream regeneration path.
• I did not commit secrets or credentials.

  

Confidence Score: 4/5

The PR appears safe to merge, with a non-blocking gap in flush latency when the score queue is full.

Summary

This PR makes flush() wake the score consumer and moves the experiment runner's final flush off the event loop.

• Flushing sends queued scores without waiting for the batch timer.
• Async experiment runs leave the event loop free during their final flush.
• Score batches follow the flush settings in your environment.

Acknowledged behavior: hassiebp explicitly describes returning when consumers stop as intentional. This prevents an orphaned wake signal from hanging flush() during concurrent shutdown().

Diagram

sequenceDiagram
    participant E as Async experiment
    participant F as Flush worker
    participant Q as Score queue
    participant C as Score consumer
    E->>F: Run flush off the event loop
    F->>Q: Add wake signal
    Q->>C: Scores, then wake signal
    C->>C: Send pending batch
    C->>Q: Acknowledge completed work
    F->>Q: Wait for completion or stopped consumer
    F-->>E: Flush returns

Loading

_{Reviews (1) · Last reviewed commit: "fix(scores): stop waiting in flush() onc..." · Reviewed by Greptile}

[github-actions]

@claude review

[claude]

Claude Code Review

No review was started: this request came from a bot account. Manual reviews can only be requested by someone with write access to this repository. Ask a maintainer to comment @claude review, or have your automation post the comment from a user account with write access.

_{Tip: disable this comment in your organization's Code Review settings.}

[chatgpt-codex-connector]

Codex Review Summary

This comment shows the latest Codex review activity on this pull request.

Review	Status	Commit	Review trigger
📝 Code Review	✅ Completed 2026-10-20T08:19:31.663138Z	ce6abc8	New commits

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review" or "@codex security review".

Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 87c50c24f4

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[claude]

Findings marked 🟡 are optional suggestions and need no follow-up push.