fix(experiments)!: run synchronous tasks in worker threads

MIGRATION_V5_SYNC_TASKS.md

@@ -0,0 +1,55 @@
+# Synchronous experiment tasks in v5
+
+This note describes the execution-context change on the v5 preparation branch;
+it does not announce a released SDK version.
+
+Blocking synchronous experiment tasks now run in worker threads instead of
+blocking the experiment's event loop. They can overlap up to the experiment's
+`max_concurrency`. Async tasks, including recognized decorated wrappers and
+callable objects, continue to run on the experiment's event loop.
+
+## Audit captured resources before upgrading
+
+A synchronous task must not assume it runs on the caller's thread. A SQLite
+connection created by the caller uses thread checking by default and fails when
+used from a worker. The same concern applies to other resources that require a
+particular thread. Even `max_concurrency=1` executes the synchronous task on a
+worker and does not restore the previous execution context.
+
+Shared task state may now be accessed concurrently. Use a client that supports
+concurrent use, synchronize shared state, or create each task's resource where
+it is used. Turning off SQLite's thread check alone does not establish safe
+concurrent ownership.
+
+## Create and close the resource in the synchronous task
+
+For string dataset inputs, this task opens, uses and closes its connection on
+the same worker thread:
+
+```python
+import sqlite3
+from contextlib import closing
+
+
+def task(*, item, **kwargs):
+    with closing(sqlite3.connect("experiment.sqlite")) as connection:
+        return connection.execute("SELECT ?", (item["input"],)).fetchone()[0]
+```
+
+The unit test exercises this pattern through `run_experiment` with two items
+and `max_concurrency=2`.
+
+An alternative is to use an asynchronous client and create/use its resources
+on the experiment's event loop. Merely wrapping blocking work in `async def`
+can keep that work on the loop and serialize it again. The loop itself is not
+guaranteed to be on the original caller's thread, including notebook use.
+
+## Inspect failed items as well as successful outputs
+
+The existing item-error policy is unchanged: task exceptions are logged and
+recorded in the trace, and failed items do not appear in `item_results`.
+`run_experiment` does not re-raise each task's exception. A plain script that
+captures a caller-owned SQLite connection can therefore return zero item
+results without raising that SQLite exception. Check the logged failures and
+returned items during migration; successful completion of the runner alone
+does not prove that every task succeeded.

README.md

@@ -22,6 +22,10 @@
 
 The [Langfuse](https://langfuse.com) Python SDK covers the full platform: **observability/tracing** (OpenTelemetry-based, with OpenAI and LangChain integrations), **datasets & experiments** (offline evaluation and regression testing of prompt/model changes, including [CI via GitHub Actions](https://github.com/langfuse/experiment-action)), **LLM-as-a-judge and custom evaluations/scores**, **prompt management**, and a **full REST API client**.
 
+On the v5 preparation branch, synchronous experiment tasks run in worker threads.
+This changes resource ownership and concurrency requirements; see the
+[synchronous task migration notes](MIGRATION_V5_SYNC_TASKS.md) before upgrading.
+
 ## Installation
 
 ```

langfuse/experiment.py

@@ -6,8 +6,10 @@
 """
 
 import asyncio
+import inspect
 import warnings
 from datetime import datetime
+from functools import partial
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -642,6 +644,16 @@ class TaskFunction(Protocol):
     - Return any type of output (will be passed to evaluators)
     - Can be either synchronous or asynchronous
     - Should handle their own errors gracefully (exceptions will be logged)
+
+    Synchronous task functions run in worker threads so blocking calls can run
+    concurrently. Asynchronous task functions, including their decorated wrappers,
+    run on the experiment's event loop.
+
+    This is a breaking execution-context change in v5. Do not capture a resource
+    that can only be used on the caller's thread, such as a default sqlite3
+    connection, in a synchronous task. Create and close the resource inside the
+    task or use an asynchronous client on the experiment's event loop. Setting
+    max_concurrency to 1 does not restore caller-thread execution.
     """
 
     def __call__(
@@ -1056,9 +1068,40 @@ async def _run_evaluator(
         return []
 
 
+def _is_async_task(task: Any) -> bool:
+    """Inspect async task metadata without replacing the callable we invoke."""
+    seen: set[int] = set()
+    check_call = True
+    for _ in range(100):
+        if id(task) in seen:
+            return False
+        seen.add(id(task))
+        if isinstance(task, partial):
+            task = task.func
+            continue
+        try:
+            unwrapped = inspect.unwrap(task, stop=inspect.iscoroutinefunction)
+        except ValueError:
+            return False
+        if inspect.iscoroutinefunction(unwrapped):
+            return True
+        if unwrapped is not task:
+            task = unwrapped
+        elif check_call:
+            task = getattr(task, "__call__", None)
+            check_call = False
+        else:
+            return False
+    return False
+
+
 async def _run_task(task: TaskFunction, item: ExperimentItem) -> Any:
     """Run a task function and handle sync/async."""
-    result = task(item=item)
+    result: Any
+    if _is_async_task(task):
+        result = task(item=item)
+    else:
+        result = await asyncio.to_thread(task, item=item)
 
     # Handle async tasks
     if asyncio.iscoroutine(result):