fix: configurable flush timeout to stop silent event loss on large sessions

.claude-plugin/plugin.json

@@ -63,6 +63,12 @@
       "description": "Upload images to Langfuse and show them in the trace. This includes pasted images and screenshots from tool results. Your Langfuse deployment must support media upload. Set to false to show short text markers instead.",
       "default": true
     },
+    "CC_LANGFUSE_FLUSH_TIMEOUT": {
+      "type": "number",
+      "title": "Flush timeout (seconds)",
+      "description": "How long the hook waits for the SDK to upload the turn's events before giving up. Long sessions need more than a few seconds; giving up early drops the whole session.",
+      "default": 120
+    },
     "CC_LANGFUSE_STATE_DIR": {
       "type": "string",
       "title": "State directory",

README.md

@@ -103,6 +103,7 @@ value is a per-run environment variable.
 | `CC_LANGFUSE_SKILL_TAGS` | Tag traces with `skill:<name>` for every skill invoked in the turn (default `true`). | No |
 | `CC_LANGFUSE_CAPTURE_SKILL_CONTENT` | Include injected skill instruction text in the Skill tool span output (default `false`). | No |
 | `CC_LANGFUSE_CAPTURE_IMAGES` | Upload images to Langfuse and show them in the trace (default `true`). Needs media upload on your deployment (self-hosted: `LANGFUSE_S3_MEDIA_UPLOAD_*`). Set it to `false` if media upload is unavailable: the trace then shows a marker per image, such as `[image image/png ~200KB]`. | No |
+| `CC_LANGFUSE_FLUSH_TIMEOUT` | Seconds the hook waits for the SDK to upload the turn's events before giving up (default `120`). A long session produces a payload that takes far longer than a few seconds to upload, and giving up early drops the whole session silently — the hook now logs when it gives up. Lower it if you would rather an unreachable Langfuse never delay the end of a turn. An unusable value falls back to the default and logs. At `SessionEnd` Claude Code caps the wait at 60s regardless of this value (see below). | No |
 | `CC_LANGFUSE_STATE_DIR` | Absolute directory (`~` is expanded) for the hook's state, lock and log files (default `~/.claude/state`). Set one per `CLAUDE_CONFIG_DIR` installation to keep them apart. An unusable value falls back to the default and logs a warning. | No |
 | `CC_LANGFUSE_TRACE_SEED` | Seed that makes trace IDs predictable, so a headless caller can derive a run's trace ID before the trace exists. Use a unique seed per session, otherwise sessions collide on the same trace IDs. | No |
 | `CC_LANGFUSE_TRACEPARENT` | Per-run environment variable. W3C traceparent of an existing trace to attach to — see [Attach runs to an existing trace](#attach-runs-to-an-existing-trace). | No |
@@ -176,10 +177,27 @@ the newest lines against this table:
 | `Langfuse config incomplete: missing …` | The named keys did not reach the hook. Configure them with `/plugin configure`. If the line also says `loaded under plugin identity '@inline'`, see below. |
 | `Hook started` plus a skip reason | The hook ran and skipped on purpose, which is usual for background sessions. Report it with the log line if real turns are missing. |
 | `Processed N turns …` but nothing in Langfuse | Delivery failed after the SDK took the turns. Check `LANGFUSE_BASE_URL` (EU against US), key validity, and proxy reachability. |
+| `Langfuse flush did not finish within …s` | The upload was still running when the hook gave up, so some or all of that session's events never left the machine. Long sessions and slow links need longer: raise `CC_LANGFUSE_FLUSH_TIMEOUT`. |
 
 `Hook started` and other `[DEBUG]` lines need `CC_LANGFUSE_DEBUG`. The failure
 lines above are `[INFO]` and appear without it.
 
+### Long sessions and the flush window
+
+The hook hands the turn's events to the SDK and then waits for the upload,
+capped by `CC_LANGFUSE_FLUSH_TIMEOUT` (default 120s) so an unreachable Langfuse
+cannot stall Claude Code forever. A few hundred transcript rows produce an OTLP
+payload that takes well over a few seconds to ship, so a short cap loses whole
+sessions — silently before this was logged.
+
+`SessionEnd` has a second, tighter budget that this plugin does not control:
+Claude Code allows its `SessionEnd` hooks the longest `timeout` any of them
+declares, floored at 1.5s and **capped at 60s**. `hooks/hooks.json` therefore
+declares `"timeout": 60` on the `SessionEnd` entry to claim that whole ceiling;
+without it the window is 1.5s and nearly nothing gets uploaded at session end.
+`Stop` has no such ceiling, so it is where large sessions actually finish
+flushing. (Verified against Claude Code 2.1.263.)
+
 ### Desktop app (GUI) sessions
 
 A GUI app does not read your shell profile and resolves `PATH` once at launch, so

hooks/hooks.json

@@ -15,7 +15,8 @@
         "hooks": [
           {
             "type": "command",
-            "command": "if command -v uv >/dev/null 2>&1; then exec uv run --quiet --script \"${CLAUDE_PLUGIN_ROOT}\"/hooks/langfuse_hook.py; else exec python3 \"${CLAUDE_PLUGIN_ROOT}\"/hooks/langfuse_hook.py; fi"
+            "command": "if command -v uv >/dev/null 2>&1; then exec uv run --quiet --script \"${CLAUDE_PLUGIN_ROOT}\"/hooks/langfuse_hook.py; else exec python3 \"${CLAUDE_PLUGIN_ROOT}\"/hooks/langfuse_hook.py; fi",
+            "timeout": 60
           }
         ]
       }

hooks/langfuse_hook.py

@@ -45,6 +45,10 @@ def _opt(name: str) -> str:
 except ValueError:
     MAX_CHARS = 20000
 
+# Seconds to wait for the SDK flush before giving up; see resolve_flush_timeout().
+FLUSH_TIMEOUT_DEFAULT = 120.0
+FLUSH_TIMEOUT_MAX = 3600.0
+
 # Bound for unresolved task notifications kept in the state file between runs.
 MAX_PENDING_TASK_NOTIFICATIONS = 50
 
@@ -3262,11 +3266,46 @@ def emit_new_turns_from_transcript(
     return emitted
 
 
+def resolve_flush_timeout() -> float:
+    """Seconds to wait for the flush thread, from CC_LANGFUSE_FLUSH_TIMEOUT.
+
+    An unusable value falls back to the default and logs, the same way
+    CC_LANGFUSE_STATE_DIR does: silently skipping the wait would reintroduce
+    the very data loss this cap exists to bound.
+    """
+    raw = _opt("CC_LANGFUSE_FLUSH_TIMEOUT").strip()
+    if not raw:
+        return FLUSH_TIMEOUT_DEFAULT
+    try:
+        seconds = float(raw)
+    except ValueError:
+        info(
+            f"CC_LANGFUSE_FLUSH_TIMEOUT {raw!r} is not a number; "
+            f"using the default {FLUSH_TIMEOUT_DEFAULT}s"
+        )
+        return FLUSH_TIMEOUT_DEFAULT
+    # Excludes NaN and inf as well as non-positive values: either would turn the
+    # cap into no cap, and `inf` would hang Claude Code on an unreachable host.
+    if not 0 < seconds <= FLUSH_TIMEOUT_MAX:
+        info(
+            f"CC_LANGFUSE_FLUSH_TIMEOUT {raw!r} is not a number of seconds between "
+            f"0 and {FLUSH_TIMEOUT_MAX}; using the default {FLUSH_TIMEOUT_DEFAULT}s"
+        )
+        return FLUSH_TIMEOUT_DEFAULT
+    return seconds
+
+
 def flush_and_shutdown_langfuse_client(langfuse: Optional[Langfuse]) -> None:
     if langfuse is None:
         return
 
-    # Cap flush+shutdown at 5s so a slow/unreachable Langfuse can't stall Claude Code.
+    # Cap flush+shutdown so a slow or unreachable Langfuse can't stall Claude
+    # Code indefinitely. The cap was a hard 5s, which is shorter than the OTLP
+    # upload of a large session (200+ turns): the join returned, the hook
+    # exited 0 reporting "Processed N turns", and every event of that session
+    # was dropped. Resolve the cap before the try, so a bad value falls back to
+    # the default instead of being swallowed into no wait at all.
+    timeout = resolve_flush_timeout()
     try:
         def _flush_and_shutdown():
             try:
@@ -3277,7 +3316,12 @@ def _flush_and_shutdown():
 
         t = threading.Thread(target=_flush_and_shutdown, daemon=True)
         t.start()
-        t.join(5.0)
+        t.join(timeout)
+        if t.is_alive():
+            info(
+                f"Langfuse flush did not finish within {timeout}s; events from this "
+                "session may be missing. Raise CC_LANGFUSE_FLUSH_TIMEOUT if this recurs."
+            )
     except Exception:
         pass
 

tests/unit/test_flush_timeout.py

@@ -0,0 +1,134 @@
+from __future__ import annotations
+
+import threading
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+
+ENV_NAME = "CC_LANGFUSE_FLUSH_TIMEOUT"
+
+
+@pytest.fixture
+def clean_flush_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv(ENV_NAME, raising=False)
+    monkeypatch.delenv(f"CLAUDE_PLUGIN_OPTION_{ENV_NAME}", raising=False)
+
+
+def _read_log(hook_module: Any) -> str:
+    log_file = Path(hook_module.LOG_FILE)
+    return log_file.read_text(encoding="utf-8") if log_file.exists() else ""
+
+
+# ----------------- resolving the cap -----------------
+
+def test_default_when_unset(hook_module: Any, clean_flush_env: None) -> None:
+    assert hook_module.resolve_flush_timeout() == hook_module.FLUSH_TIMEOUT_DEFAULT
+
+
+def test_env_var_wins(hook_module: Any, clean_flush_env: None, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv(ENV_NAME, "7.5")
+
+    assert hook_module.resolve_flush_timeout() == 7.5
+
+
+def test_plugin_user_config_is_the_fallback(
+    hook_module: Any, clean_flush_env: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The wizard value must reach the hook, like every other CC_LANGFUSE_* option."""
+    monkeypatch.setenv(f"CLAUDE_PLUGIN_OPTION_{ENV_NAME}", "30")
+
+    assert hook_module.resolve_flush_timeout() == 30.0
+
+
+@pytest.mark.parametrize("raw", ["abc", "", "   ", "0", "-1", "nan", "inf", "1e9"])
+def test_unusable_values_fall_back_to_the_default(
+    hook_module: Any, clean_flush_env: None, monkeypatch: pytest.MonkeyPatch, raw: str
+) -> None:
+    monkeypatch.setenv(ENV_NAME, raw)
+
+    assert hook_module.resolve_flush_timeout() == hook_module.FLUSH_TIMEOUT_DEFAULT
+
+
+@pytest.mark.parametrize("raw", ["abc", "0", "-1"])
+def test_unusable_values_are_logged(
+    hook_module: Any, clean_flush_env: None, monkeypatch: pytest.MonkeyPatch, raw: str
+) -> None:
+    monkeypatch.setenv(ENV_NAME, raw)
+
+    hook_module.resolve_flush_timeout()
+
+    assert ENV_NAME in _read_log(hook_module)
+
+
+# ----------------- the cap reaches the join -----------------
+
+class RecordingLangfuse:
+    def __init__(self) -> None:
+        self.flushed = threading.Event()
+        self.shutdown_called = threading.Event()
+
+    def flush(self) -> None:
+        self.flushed.set()
+
+    def shutdown(self) -> None:
+        self.shutdown_called.set()
+
+
+def test_flush_and_shutdown_uses_the_resolved_cap(
+    hook_module: Any, clean_flush_env: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setenv(ENV_NAME, "42")
+    seen: list[float | None] = []
+    real_join = threading.Thread.join
+
+    def recording_join(self: threading.Thread, timeout: float | None = None) -> None:
+        seen.append(timeout)
+        real_join(self, timeout)
+
+    monkeypatch.setattr(threading.Thread, "join", recording_join)
+
+    hook_module.flush_and_shutdown_langfuse_client(RecordingLangfuse())
+
+    assert 42.0 in seen
+
+
+def test_malformed_value_still_waits_for_the_flush(
+    hook_module: Any, clean_flush_env: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Regression: parsing used to happen inside the function's blanket
+    `except Exception`, so a malformed value raised, skipped the join entirely
+    and dropped the session — the exact failure the cap is meant to bound."""
+    monkeypatch.setenv(ENV_NAME, "not-a-number")
+    client = RecordingLangfuse()
+
+    hook_module.flush_and_shutdown_langfuse_client(client)
+
+    assert client.flushed.is_set()
+    assert client.shutdown_called.is_set()
+
+
+def test_none_client_is_a_noop(hook_module: Any, clean_flush_env: None) -> None:
+    hook_module.flush_and_shutdown_langfuse_client(None)
+
+
+def test_unfinished_flush_is_reported(
+    hook_module: Any, clean_flush_env: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A hook that gives up must say so; the silent version is what lost sessions."""
+    release = threading.Event()
+
+    class StuckLangfuse:
+        def flush(self) -> None:
+            release.wait(10)
+
+        def shutdown(self) -> None:
+            pass
+
+    monkeypatch.setenv(ENV_NAME, "0.05")
+    try:
+        hook_module.flush_and_shutdown_langfuse_client(StuckLangfuse())
+        assert "did not finish within" in _read_log(hook_module)
+    finally:
+        release.set()