feat(langfuse): Add tracing opt-in/opt-out support

backend/app/alembic/versions/071_add_settings_to_project.py

@@ -0,0 +1,39 @@
+"""add settings jsonb column to project
+
+Revision ID: 071
+Revises: 070
+Create Date: 2026-06-17
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision = "071"
+down_revision = "070"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "project",
+        sa.Column(
+            "settings",
+            postgresql.JSONB(),
+            nullable=False,
+            server_default=sa.text("'{}'::jsonb"),
+            comment=(
+                "Project-level settings (JSONB). Keys: 'tracing' (bool) — "
+                "Langfuse tracing opt-in, off by default to conserve Langfuse "
+                "rate-limit/credit budget. Gates tracing for both the response "
+                "path and evaluations (which fall back to cosine-only scoring)."
+            ),
+        ),
+    )
+
+
+def downgrade():
+    op.drop_column("project", "settings")

backend/app/api/docs/projects/update_settings.md

@@ -0,0 +1,12 @@
+Update project-level settings.
+
+Patches the `settings` JSONB of the project bound to the authenticating organization
+API key. Only the keys provided in the request body are changed; existing keys are kept.
+
+**Settings**
+
+- `tracing` (bool): enable/disable Langfuse tracing for this project. Off by default to
+  conserve the org's Langfuse rate-limit/credit budget. Gates tracing for both the
+  response path and evaluations; when off, evaluations fall back to cosine-only scoring.
+
+**Scope:** requires an organization API key with project access.

backend/app/api/routes/project.py

@@ -4,7 +4,7 @@
 from sqlalchemy import func
 from sqlmodel import select
 
-from app.api.deps import SessionDep
+from app.api.deps import AuthContextDep, SessionDep
 from app.api.permissions import Permission, require_permission
 from app.crud.organization import get_organization_by_id, validate_organization
 from app.crud.project import (
@@ -14,6 +14,7 @@
     get_projects_by_organization,
     hard_delete_project,
     soft_delete_project,
+    update_project_settings,
 )
 from app.crud.user_project import (
     deactivate_users_without_projects,
@@ -25,6 +26,7 @@
     Project,
     ProjectCreate,
     ProjectPublic,
+    ProjectSettingsUpdate,
     ProjectUpdate,
 )
 from app.utils import APIResponse, load_description
@@ -79,6 +81,34 @@ def create_new_project(*, session: SessionDep, project_in: ProjectCreate):
     return APIResponse.success_response(project)
 
 
+@router.patch(
+    "/settings",
+    dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))],
+    response_model=APIResponse[ProjectPublic],
+    description=load_description("projects/update_settings.md"),
+)
+def update_project_settings_route(
+    *,
+    session: SessionDep,
+    auth_context: AuthContextDep,
+    settings_in: ProjectSettingsUpdate,
+) -> APIResponse[ProjectPublic]:
+    settings_patch = settings_in.model_dump(exclude_unset=True)
+    if not settings_patch:
+        raise HTTPException(status_code=400, detail="No settings provided")
+
+    project = update_project_settings(
+        session=session,
+        project_id=auth_context.project.id,
+        settings_patch=settings_patch,
+    )
+    logger.info(
+        f"[update_project_settings_route] Settings updated | project_id={project.id}, "
+        f"keys={list(settings_patch.keys())}"
+    )
+    return APIResponse.success_response(project)
+
+
 @router.get(
     "/{project_id}",
     dependencies=[Depends(require_permission(Permission.SUPERUSER))],

backend/app/api/routes/responses.py

@@ -7,7 +7,7 @@
 from app.api.deps import AuthContextDep, SessionDep
 from app.api.permissions import Permission, require_permission
 from app.core.langfuse.langfuse import LangfuseTracer
-from app.crud.credentials import get_provider_credential
+from app.crud.credentials import get_tracing_credential
 from app.models import (
     CallbackResponse,
     Diagnostics,
@@ -97,10 +97,9 @@ async def responses_sync(
             },
         )
 
-    langfuse_credentials = get_provider_credential(
+    langfuse_credentials = get_tracing_credential(
         session=session,
         org_id=organization_id,
-        provider="langfuse",
         project_id=project_id,
     )
     tracer = LangfuseTracer(

backend/app/api/routes/threads.py

@@ -9,11 +9,11 @@
 
 from app.api.deps import AuthContextDep, SessionDep
 from app.api.permissions import Permission, require_permission
-from app.core import logging, settings
+from app.core import logging
 from app.models import OpenAIThreadCreate
 from app.crud import upsert_thread_result, get_thread_result
 from app.utils import APIResponse, mask_string
-from app.crud.credentials import get_provider_credential
+from app.crud.credentials import get_provider_credential, get_tracing_credential
 from app.core.util import configure_openai
 from app.core.langfuse.langfuse import LangfuseTracer
 
@@ -312,10 +312,9 @@ async def threads(
             error="OpenAI API key not configured for this organization."
         )
 
-    langfuse_credentials = get_provider_credential(
+    langfuse_credentials = get_tracing_credential(
         session=session,
         org_id=_current_user.organization_.id,
-        provider="langfuse",
         project_id=request.get("project_id"),
     )
 
@@ -387,10 +386,9 @@ async def threads_sync(
         )
 
     # Get Langfuse credentials
-    langfuse_credentials = get_provider_credential(
+    langfuse_credentials = get_tracing_credential(
         session=session,
         org_id=_current_user.organization_.id,
-        provider="langfuse",
         project_id=request.get("project_id"),
     )
 

backend/app/crud/__init__.py

@@ -36,6 +36,7 @@
     update_creds_for_org,
     remove_creds_for_org,
     get_provider_credential,
+    get_tracing_credential,
     remove_provider_credential,
 )
 

backend/app/crud/credentials.py

@@ -9,6 +9,7 @@
 from app.core.providers import validate_provider, validate_provider_credentials
 from app.core.security import decrypt_credentials, encrypt_credentials
 from app.core.util import now
+from app.crud.project import get_project_by_id
 from app.models import Credential, CredsCreate, CredsUpdate
 
 
@@ -167,6 +168,41 @@ def get_provider_credential(
     return None
 
 
+def get_tracing_credential(
+    *,
+    session: Session,
+    org_id: int,
+    project_id: int | str,
+) -> dict[str, Any] | None:
+    """Return langfuse credentials only when the project opted into tracing.
+
+    Tracing is gated by the project's `settings["tracing"]` flag (off by
+    default). When disabled, returns None so LangfuseTracer /
+    observe_llm_execution degrade to a no-op and evaluations (via
+    get_tracing_client) fall back to cosine-only scoring.
+    """
+    try:
+        pid = int(project_id)
+    except (TypeError, ValueError):
+        logger.info(
+            f"[get_tracing_credential] Invalid project_id; tracing off | "
+            f"project_id={project_id}"
+        )
+        return None
+
+    project = get_project_by_id(session=session, project_id=pid)
+    if not project or not (project.settings or {}).get("tracing", False):
+        logger.info(f"[get_tracing_credential] Tracing disabled | project_id={pid}")
+        return None
+
+    return get_provider_credential(
+        session=session,
+        org_id=org_id,
+        project_id=pid,
+        provider="langfuse",
+    )
+
+
 def get_providers(*, session: Session, org_id: int, project_id: int) -> list[str]:
     """Returns a list of all active providers for which credentials are stored."""
     creds = get_creds_by_org(session=session, org_id=org_id, project_id=project_id)

backend/app/crud/evaluations/batch.py

@@ -20,6 +20,12 @@
     start_batch_job,
 )
 from app.core.batch.client import GeminiClient
+from app.core.cloud import get_cloud_storage
+from app.crud.evaluations.dataset import (
+    download_csv_from_object_store,
+    get_dataset_by_id,
+)
+from app.crud.evaluations.score import DEFAULT_CATEGORY
 from app.models import EvaluationRun
 from app.models.batch_job import BatchJobType
 from app.services.llm.mappers import (
@@ -70,6 +76,84 @@ def fetch_dataset_items(langfuse: Langfuse, dataset_name: str) -> list[dict[str,
     return items
 
 
+def use_langfuse_client(
+    session: Session,
+    eval_run: EvaluationRun,
+    langfuse: Langfuse | None,
+) -> Langfuse | None:
+    """Return the live client only if the run's dataset is Langfuse-backed, else
+    None, so an opt-out dataset is never traced even if the flag is later on."""
+    if langfuse is None:
+        return None
+
+    dataset = get_dataset_by_id(
+        session=session,
+        dataset_id=eval_run.dataset_id,
+        organization_id=eval_run.organization_id,
+        project_id=eval_run.project_id,
+    )
+    return langfuse if (dataset and dataset.langfuse_dataset_id) else None
+
+
+def load_evaluation_dataset_items(
+    session: Session,
+    eval_run: EvaluationRun,
+    langfuse: Langfuse | None,
+) -> list[dict[str, Any]]:
+    """Load dataset items from Langfuse when a (reconciled) client is present,
+    else from the dataset's object-store CSV."""
+    dataset = get_dataset_by_id(
+        session=session,
+        dataset_id=eval_run.dataset_id,
+        organization_id=eval_run.organization_id,
+        project_id=eval_run.project_id,
+    )
+    if not dataset:
+        raise ValueError(f"Dataset {eval_run.dataset_id} not found")
+
+    if langfuse is not None:
+        return fetch_dataset_items(
+            langfuse=langfuse, dataset_name=eval_run.dataset_name
+        )
+
+    return _load_items_from_object_store(session=session, dataset=dataset)
+
+
+def _load_items_from_object_store(
+    session: Session, dataset: Any
+) -> list[dict[str, Any]]:
+    """Load items from the dataset's object-store CSV with deterministic ids."""
+    from app.services.evaluations.validators import parse_csv_items
+
+    if not dataset.object_store_url:
+        raise ValueError(f"Dataset {dataset.id} has no object-store backing")
+
+    storage = get_cloud_storage(session=session, project_id=dataset.project_id)
+    csv_content = download_csv_from_object_store(
+        storage=storage, object_store_url=dataset.object_store_url
+    )
+    original_items = parse_csv_items(csv_content)
+    duplication_factor = max(
+        1, int((dataset.dataset_metadata or {}).get("duplication_factor", 1))
+    )
+
+    items: list[dict[str, Any]] = []
+    for row_idx, item in enumerate(original_items):
+        for dup_idx in range(duplication_factor):
+            items.append(
+                {
+                    "id": f"item_{row_idx}_{dup_idx}",
+                    "input": {"question": item["question"]},
+                    "expected_output": {"answer": item["answer"]},
+                    "metadata": {
+                        "category": item.get("category") or DEFAULT_CATEGORY,
+                        "question_id": f"item_{row_idx}",
+                    },
+                }
+            )
+    return items
+
+
 def build_openai_evaluation_jsonl(
     dataset_items: list[dict[str, Any]], openai_params: dict[str, Any]
 ) -> list[dict[str, Any]]:
@@ -153,7 +237,7 @@ def build_google_evaluation_jsonl(
 
 
 def start_evaluation_batch(
-    langfuse: Langfuse,
+    langfuse: Langfuse | None,
     session: Session,
     eval_run: EvaluationRun,
     params: dict[str, Any],
@@ -176,8 +260,8 @@ def start_evaluation_batch(
         logger.info(
             f"[start_evaluation_batch] Starting evaluation batch | run={eval_run.run_name} | provider={provider}"
         )
-        dataset_items = fetch_dataset_items(
-            langfuse=langfuse, dataset_name=eval_run.dataset_name
+        dataset_items = load_evaluation_dataset_items(
+            session=session, eval_run=eval_run, langfuse=langfuse
         )
 
         base_provider = provider.replace("-native", "")

backend/app/crud/evaluations/core.py

@@ -271,7 +271,7 @@ def _enqueue_eval_completion_notification(eval_run: EvaluationRun) -> None:
 def get_or_fetch_score(
     session: Session,
     eval_run: EvaluationRun,
-    langfuse: Langfuse,
+    langfuse: Langfuse | None,
     force_refetch: bool = False,
 ) -> EvaluationScore:
     """
@@ -295,6 +295,15 @@ def get_or_fetch_score(
         ValueError: If the run is not found in Langfuse
         Exception: If Langfuse API calls fail
     """
+    # Tracing disabled: no Langfuse traces exist, so return whatever
+    # cosine-based summary_scores were already computed.
+    if langfuse is None:
+        logger.info(
+            f"[get_or_fetch_score] Tracing off; returning cosine-only score | "
+            f"evaluation_id={eval_run.id}"
+        )
+        return eval_run.score or {"summary_scores": [], "traces": []}
+
     # Check if score already exists with traces
     has_traces = eval_run.score is not None and "traces" in eval_run.score
     if not force_refetch and has_traces: