diff --git a/backend/app/alembic/versions/075_add_judge_columns_to_evaluation_run.py b/backend/app/alembic/versions/075_add_judge_columns_to_evaluation_run.py
new file mode 100644
index 000000000..bda5cd7e4
--- /dev/null
+++ b/backend/app/alembic/versions/075_add_judge_columns_to_evaluation_run.py
@@ -0,0 +1,57 @@
+"""Add native LLM-as-judge columns to evaluation_run
+
+Revision ID: 075
+Revises: 074
+Create Date: 2026-07-16 00:00:00.000000
+
+The v2 judged fast-eval run stores its judge state on the existing evaluation_run
+row rather than a new table. The chunked aggregate task only knows an eval_run_id,
+so the judge intent (is_judge_run) must be durable on the row for the aggregate to
+read at judge time. per_item_ground_truth mirrors per_item_scores as Kaapi's native
+per-row store (v2 never syncs to Langfuse). Judging is system-config only — always
+the fallback model + built-in prompt — so no per-run judge config is persisted.
+
+Both columns are nullable with default NULL: v1 and pre-feature runs carry no judge
+data, so no backfill is needed.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects.postgresql import JSONB
+
+revision = "075"
+down_revision = "074"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "evaluation_run",
+        sa.Column(
+            "is_judge_run",
+            sa.Boolean(),
+            nullable=True,
+            comment=(
+                "True for v2 runs that run the native LLM-as-judge (and skip the "
+                "Langfuse score sync). NULL/False = v1 run, cosine-only, Langfuse-synced"
+            ),
+        ),
+    )
+    op.add_column(
+        "evaluation_run",
+        sa.Column(
+            "per_item_ground_truth",
+            JSONB(),
+            nullable=True,
+            comment=(
+                "Durable {ref: score} map of the Adherence to Ground Truth judge "
+                "scores (ref = trace_id when traced, else item_id); Kaapi's own store"
+            ),
+        ),
+    )
+
+
+def downgrade():
+    op.drop_column("evaluation_run", "per_item_ground_truth")
+    op.drop_column("evaluation_run", "is_judge_run")
diff --git a/backend/app/api/docs/evaluation/create_evaluation_dataset_v2.md b/backend/app/api/docs/evaluation/create_evaluation_dataset_v2.md
new file mode 100644
index 000000000..6750daeab
--- /dev/null
+++ b/backend/app/api/docs/evaluation/create_evaluation_dataset_v2.md
@@ -0,0 +1,18 @@
+Upload a CSV of golden Q&A pairs for evaluation, stored natively by Kaapi with no
+Langfuse dataset (v2).
+
+Same multipart shape as the v1 dataset upload (`file`, `dataset_name`,
+`description`, `duplication_factor`), but the dataset is stored in object storage
+(S3) only: the row is created with `langfuse_dataset_id` null and the CSV holds
+exactly the **original** rows — no physical duplication. The `duplication_factor`
+is recorded in `dataset_metadata` (with a run-time-duplication marker) and applied
+when the run executes, so an 8-row CSV with factor 5 evaluates 40 rows.
+
+**Response:** `APIResponse[DatasetUploadResponse]`, same shape as v1 with
+`langfuse_dataset_id` null. `original_items` is the stored row count and
+`total_items` is `original_items × duplication_factor` (the count the run will
+produce, not stored rows). `eligible_for_fast` reflects only the unique-row size
+cap (`EVAL_FAST_MAX_UNIQUE_ROWS`).
+
+**CSV format:** required columns `question`, `answer`; optional `category`. Rows
+with a missing question or answer are skipped. Maximum upload size is 1 MB.
diff --git a/backend/app/api/docs/evaluation/create_evaluation_v2.md b/backend/app/api/docs/evaluation/create_evaluation_v2.md
new file mode 100644
index 000000000..2f3bf355c
--- /dev/null
+++ b/backend/app/api/docs/evaluation/create_evaluation_v2.md
@@ -0,0 +1,32 @@
+Start a v2 evaluation run. Replicates the v1 `POST /api/v1/evaluations` request
+body with Kaapi's native LLM-as-Judge built in — v1 is left unchanged.
+
+v2 runs are **always fast** and always judged (there is no `run_mode`; batch is
+deferred to a later phase). Every scoreable row is automatically judged (no opt-in
+flag) on **Adherence to Ground Truth**: an LLM judge scores whether the answer
+conveys the same correct information as the dataset's golden answer (0–1, with
+reasoning). Scores, the durable per-row map (`per_item_ground_truth`), and the
+`ground_truth_judge` cost stage are stored natively by Kaapi. v2 runs compute **no
+cosine similarity** and do **not** touch Langfuse.
+
+Judging is system-config only: the judge always uses the configured model
+(`EVAL_JUDGE_MODEL`, default `gpt-5-mini`) and the built-in ground-truth prompt.
+There is no per-run or ad-hoc judge configuration.
+
+## Example
+
+```json
+{
+  "dataset_id": 123,
+  "experiment_name": "judge-smoke-1",
+  "config_id": "f54f0d67-4817-4103-9fdf-b74b3d46733e",
+  "config_version": 1
+}
+```
+
+## Error responses
+
+| Status | Code | When |
+| --- | --- | --- |
+| 409 | `run_name_already_exists` | A run with the same `experiment_name` already exists for this (organization, project) |
+| 422 | — | An unsupported config type / oversized dataset |
diff --git a/backend/app/api/main.py b/backend/app/api/main.py
index 6e1a0c8d0..a36ebcef1 100644
--- a/backend/app/api/main.py
+++ b/backend/app/api/main.py
@@ -3,7 +3,6 @@
 from app.api.routes import (
     analytics,
     api_keys,
-    assessment as assessment_routes,
     assistants,
     auth,
     collection_job,
@@ -35,7 +34,13 @@
     users,
     utils,
 )
-from app.core.config import settings
+from app.api.routes import (
+    assessment as assessment_routes,
+)
+from app.api.routes.evaluations.dataset_v2 import (
+    router as evaluations_dataset_v2_router,
+)
+from app.api.routes.evaluations.evaluation_v2 import router as evaluations_v2_router
 
 api_router = APIRouter()
 api_router.include_router(analytics.router)
@@ -73,3 +78,10 @@
 api_router.include_router(private.router)
 # if settings.ENVIRONMENT in ["development", "testing"]:
 #     api_router.include_router(private.router)
+
+
+# v2 API surface (mounted at settings.API_V2_STR). Only the endpoints that differ
+# from v1 live here — currently the judged run trigger. Everything else stays v1.
+api_v2_router = APIRouter()
+api_v2_router.include_router(evaluations_v2_router)
+api_v2_router.include_router(evaluations_dataset_v2_router)
diff --git a/backend/app/api/routes/evaluations/dataset_v2.py b/backend/app/api/routes/evaluations/dataset_v2.py
new file mode 100644
index 000000000..46780b43f
--- /dev/null
+++ b/backend/app/api/routes/evaluations/dataset_v2.py
@@ -0,0 +1,62 @@
+"""v2 Langfuse-free evaluation dataset upload route.
+
+Replica of the v1 dataset upload with the same multipart shape and response, but
+it creates no Langfuse dataset and stores only the original items — duplication is
+recorded in metadata and applied at run time.
+"""
+
+import logging
+
+from fastapi import APIRouter, Depends, File, Form, UploadFile
+
+from app.api.deps import AuthContextDep, SessionDep
+from app.api.permissions import Permission, require_permission
+from app.api.routes.evaluations.dataset import _dataset_to_response
+from app.core.rate_monitor import monitor_rate
+from app.models.evaluation import DatasetUploadResponse
+from app.services.evaluations import upload_dataset_v2, validate_csv_file
+from app.utils import APIResponse, load_description
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/evaluations/datasets", tags=["Evaluation v2"])
+
+
+@router.post(
+    "",
+    description=load_description("evaluation/create_evaluation_dataset_v2.md"),
+    response_model=APIResponse[DatasetUploadResponse],
+    dependencies=[
+        Depends(require_permission(Permission.REQUIRE_PROJECT)),
+        Depends(monitor_rate("evaluations")),
+    ],
+)
+async def upload_dataset_v2_route(
+    session: SessionDep,
+    auth_context: AuthContextDep,
+    file: UploadFile = File(
+        ..., description="CSV file with 'question' and 'answer' columns"
+    ),
+    dataset_name: str = Form(..., description="Name for the dataset"),
+    description: str | None = Form(None, description="Optional dataset description"),
+    duplication_factor: int = Form(
+        default=1,
+        ge=1,
+        le=5,
+        description="Run-time duplication per item (min: 1, max: 5)",
+    ),
+) -> APIResponse[DatasetUploadResponse]:
+    """Upload a Langfuse-free evaluation dataset (v2)."""
+    csv_content = await validate_csv_file(file)
+
+    dataset = upload_dataset_v2(
+        session=session,
+        csv_content=csv_content,
+        dataset_name=dataset_name,
+        description=description,
+        duplication_factor=duplication_factor,
+        organization_id=auth_context.organization_.id,
+        project_id=auth_context.project_.id,
+    )
+
+    return APIResponse.success_response(data=_dataset_to_response(dataset))
diff --git a/backend/app/api/routes/evaluations/evaluation_v2.py b/backend/app/api/routes/evaluations/evaluation_v2.py
new file mode 100644
index 000000000..86ef0cc85
--- /dev/null
+++ b/backend/app/api/routes/evaluations/evaluation_v2.py
@@ -0,0 +1,56 @@
+"""v2 evaluation run trigger — replica of the v1 trigger plus native judging."""
+
+import logging
+from uuid import UUID
+
+from asgi_correlation_id import correlation_id
+from fastapi import APIRouter, Body, Depends
+
+from app.api.deps import AuthContextDep, SessionDep
+from app.api.permissions import Permission, require_permission
+from app.core.rate_monitor import monitor_rate
+from app.models.evaluation import EvaluationRunPublic
+from app.services.evaluations.judge import validate_and_start_judged_evaluation
+from app.utils import APIResponse, load_description
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/evaluations", tags=["Evaluation v2"])
+
+
+@router.post(
+    "",
+    description=load_description("evaluation/create_evaluation_v2.md"),
+    response_model=APIResponse[EvaluationRunPublic],
+    dependencies=[
+        Depends(require_permission(Permission.REQUIRE_PROJECT)),
+        Depends(monitor_rate("evaluations")),
+    ],
+)
+def evaluate_v2(
+    session: SessionDep,
+    auth_context: AuthContextDep,
+    dataset_id: int = Body(..., description="ID of the evaluation dataset"),
+    experiment_name: str = Body(
+        ..., description="Name for this evaluation experiment/run"
+    ),
+    config_id: UUID = Body(..., description="Stored config ID"),
+    config_version: int = Body(..., ge=1, description="Stored config version"),
+) -> APIResponse[EvaluationRunPublic]:
+    """Start a v2 evaluation run.
+
+    v2 runs are always fast and judged on Adherence to Ground Truth; there is no
+    `run_mode` — batch judging is deferred to a later phase.
+    """
+
+    eval_run = validate_and_start_judged_evaluation(
+        session=session,
+        dataset_id=dataset_id,
+        run_name=experiment_name,
+        config_id=config_id,
+        config_version=config_version,
+        organization_id=auth_context.organization_.id,
+        project_id=auth_context.project_.id,
+        trace_id=correlation_id.get() or "N/A",
+    )
+    return APIResponse.success_response(data=eval_run)
diff --git a/backend/app/core/config.py b/backend/app/core/config.py
index c9233fd42..24897548d 100644
--- a/backend/app/core/config.py
+++ b/backend/app/core/config.py
@@ -33,6 +33,8 @@ class Settings(BaseSettings):
     )
 
     API_V1_STR: str = "/api/v1"
+    # v2 hosts
+    API_V2_STR: str = "/api/v2"
     SECRET_KEY: str = secrets.token_urlsafe(32)
     # 60 minutes * 24 hours * 1 days = 1 days
     ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 1
@@ -209,6 +211,12 @@ def AWS_S3_BUCKET(self) -> str:
     # task well under CELERY_TASK_SOFT_TIME_LIMIT.
     EVAL_FAST_CHUNK_SIZE: int = 50
 
+    EVAL_JUDGE_MODEL: str = "gpt-5-mini"
+
+    # Reasoning effort for the judge model. "minimal" keeps per-row judging fast.
+    # models. One of: none | minimal | low | medium | high | xhigh.
+    EVAL_JUDGE_REASONING_EFFORT: str = "minimal"
+
     @computed_field  # type: ignore[prop-decorator]
     @property
     def COMPUTED_CELERY_WORKER_CONCURRENCY(self) -> int:
diff --git a/backend/app/core/security.py b/backend/app/core/security.py
index 406f46ed8..fa997c285 100644
--- a/backend/app/core/security.py
+++ b/backend/app/core/security.py
@@ -13,20 +13,20 @@
 import os
 import secrets
 from datetime import UTC, datetime, timedelta
-from typing import Any
+from typing import Any, Tuple
 
 import boto3
 import jwt
+from jwt.exceptions import InvalidTokenError
 from botocore.client import BaseClient
 from cryptography.fernet import Fernet
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
-from jwt.exceptions import InvalidTokenError
 from passlib.context import CryptContext
 from sqlmodel import Session, and_, select
 
 from app.core.config import settings
-from app.models import APIKey, AuthContext, Organization, Project, User
+from app.models import APIKey, User, Organization, Project, AuthContext
 
 logger = logging.getLogger(__name__)
 
@@ -263,7 +263,7 @@ class APIKeyManager:
     pwd_context = CryptContext(schemes=[HASH_ALGORITHM], deprecated="auto")
 
     @classmethod
-    def generate(cls) -> tuple[str, str, str]:
+    def generate(cls) -> Tuple[str, str, str]:
         """
         Generate a new API key with prefix and hashed value.
         Ensures exact lengths: prefix=22 chars, secret=43 chars.
@@ -288,7 +288,7 @@ def generate(cls) -> tuple[str, str, str]:
         return raw_key, key_prefix, key_hash
 
     @classmethod
-    def _extract_key_parts(cls, raw_key: str) -> tuple[str, str] | None:
+    def _extract_key_parts(cls, raw_key: str) -> Tuple[str, str] | None:
         """
         Extract prefix and secret from an API key based on its format.
 
diff --git a/backend/app/crud/evaluations/cost.py b/backend/app/crud/evaluations/cost.py
index 653232dd1..59d5f81d6 100644
--- a/backend/app/crud/evaluations/cost.py
+++ b/backend/app/crud/evaluations/cost.py
@@ -9,12 +9,15 @@
 Persisted shape on `eval_run.cost`:
 
     {
-        "response":  {model, input_tokens, output_tokens, total_tokens, cost_usd},
-        "embedding": {model, input_tokens, output_tokens, total_tokens, cost_usd},
+        "response":          {model, input_tokens, output_tokens, total_tokens, cost_usd},
+        "embedding":         {model, input_tokens, output_tokens, total_tokens, cost_usd},
+        "ground_truth_judge": {model, input_tokens, output_tokens, total_tokens, cost_usd},
         "total_cost_usd": float,
     }
 
-Either stage entry is optional. Embedding entries use output_tokens=0.
+Any stage entry is optional. Embedding entries use output_tokens=0. Judge stages
+(one per metric, keyed by the metric's cost_stage) are priced like the response
+stage from per-row usage.
 """
 
 import logging
@@ -112,9 +115,29 @@ def _build_embedding_cost_entry(
     return _build_cost_entry(session=session, model=model, totals=totals)
 
 
+# Fixed top-level keys on eval_run.cost that are NOT per-metric judge stages;
+# everything else at the top level is a judge stage preserved across partial updates.
+_NON_JUDGE_COST_KEYS: frozenset[str] = frozenset(
+    {"response", "embedding", "total_cost_usd"}
+)
+
+
+def _build_judge_cost_entry(
+    session: Session, model: str, results: list[dict[str, Any]]
+) -> dict[str, Any]:
+    """Build a judge-stage cost entry from per-row judge usage results."""
+    totals = _sum_tokens(
+        items=results,
+        usage_extractor=lambda r: r.get("usage"),
+        input_key="input_tokens",
+    )
+    return _build_cost_entry(session=session, model=model, totals=totals)
+
+
 def _build_cost_dict(
     response_entry: dict[str, Any] | None,
     embedding_entry: dict[str, Any] | None,
+    judge_entries: dict[str, dict[str, Any]],
 ) -> dict[str, Any]:
     """Combine per-stage entries into the `eval_run.cost` payload with a grand total."""
     cost: dict[str, Any] = {}
@@ -128,6 +151,11 @@ def _build_cost_dict(
         cost["embedding"] = embedding_entry
         total += embedding_entry.get("cost_usd", 0.0)
 
+    for stage, entry in judge_entries.items():
+        if entry:
+            cost[stage] = entry
+            total += entry.get("cost_usd", 0.0)
+
     cost["total_cost_usd"] = round(total, COST_USD_DECIMALS)
     return cost
 
@@ -141,12 +169,16 @@ def attach_cost(
     response_results: list[dict[str, Any]] | None = None,
     embedding_model: str | None = None,
     embedding_raw_results: list[dict[str, Any]] | None = None,
+    judge_stage: str | None = None,
+    judge_model: str | None = None,
+    judge_results: list[dict[str, Any]] | None = None,
 ) -> None:
     """Compute cost for the given stage(s) and attach to `eval_run.cost`, never raising.
 
-    Caller is responsible for persisting `eval_run` afterwards. Either stage's
+    Caller is responsible for persisting `eval_run` afterwards. Any stage's
     previously-computed entry on `eval_run.cost` is preserved when that stage's
-    inputs are not supplied, so partial updates never clobber prior data.
+    inputs are not supplied, so partial updates never clobber prior data — including
+    prior per-metric judge stages, which are carried forward untouched.
     """
     try:
         existing_cost = eval_run.cost or {}
@@ -167,9 +199,25 @@ def attach_cost(
         else:
             embedding_entry = existing_cost.get("embedding")
 
+        # Carry forward prior judge stages; recompute only the one supplied this call.
+        judge_entries: dict[str, dict[str, Any]] = {
+            k: v
+            for k, v in existing_cost.items()
+            if k not in _NON_JUDGE_COST_KEYS and isinstance(v, dict)
+        }
+        if (
+            judge_stage is not None
+            and judge_model is not None
+            and judge_results is not None
+        ):
+            judge_entries[judge_stage] = _build_judge_cost_entry(
+                session=session, model=judge_model, results=judge_results
+            )
+
         eval_run.cost = _build_cost_dict(
             response_entry=response_entry,
             embedding_entry=embedding_entry,
+            judge_entries=judge_entries,
         )
     except Exception as cost_err:
         logger.warning(
diff --git a/backend/app/crud/evaluations/dataset.py b/backend/app/crud/evaluations/dataset.py
index d0cf29930..80efae58c 100644
--- a/backend/app/crud/evaluations/dataset.py
+++ b/backend/app/crud/evaluations/dataset.py
@@ -29,6 +29,16 @@
 logger = logging.getLogger(__name__)
 
 
+# dataset_metadata keys, shared by the upload services and the run-time loader so
+DATASET_META_ORIGINAL_ITEMS = "original_items_count"
+DATASET_META_TOTAL_ITEMS = "total_items_count"
+DATASET_META_DUPLICATION_FACTOR = "duplication_factor"
+# v2 marker: the stored CSV holds only the original rows and duplication is applied
+# at run time. Absent/false means the S3 data is already physically duplicated (v1),
+# so the run reads it as-is and must not multiply again.
+DATASET_META_DUPLICATE_AT_RUNTIME = "duplicate_at_runtime"
+
+
 def create_evaluation_dataset(
     session: Session,
     name: str,
diff --git a/backend/app/crud/evaluations/fast.py b/backend/app/crud/evaluations/fast.py
index 7d77a3542..24d8f364d 100644
--- a/backend/app/crud/evaluations/fast.py
+++ b/backend/app/crud/evaluations/fast.py
@@ -51,6 +51,13 @@
     EMBEDDING_MODEL,
     calculate_cosine_similarity,
 )
+from app.crud.evaluations.judge import (
+    JudgeMetricSpec,
+    JudgeResult,
+    build_judge_params,
+    enabled_metric_specs,
+    judge_row,
+)
 from app.crud.evaluations.langfuse import (
     create_langfuse_dataset_run,
     update_traces_with_cosine_scores,
@@ -59,6 +66,7 @@
 from app.crud.evaluations.score import (
     COSINE_SCORE_COMMENT,
     COSINE_SCORE_NAME,
+    JUDGE_FAILED_REASON,
     EvaluationScore,
     TraceData,
     TraceScore,
@@ -87,6 +95,12 @@
 CHUNK_CONFIG_RUN_ID = "eval_run_id"
 CHUNK_CONFIG_INDEX = "chunk_index"
 
+# Reasons a row cannot be scored. embedding_failed is v1-only
+# (cosine); v2 judged runs never embed, so only the empty-side reasons apply.
+UNSCOREABLE_EMPTY_OUTPUT = "empty_output"
+UNSCOREABLE_EMPTY_GROUND_TRUTH = "empty_ground_truth"
+UNSCOREABLE_EMBEDDING_FAILED = "embedding_failed"
+
 
 # Per-call retry policy for Stage 1 / Stage 2.
 _RETRY_MAX_ATTEMPTS = 3
@@ -727,29 +741,139 @@ def _stage2_embeddings(
     return eval_run, embedding_results
 
 
+def _judge_rows(
+    *,
+    session: Session,
+    openai_client: OpenAI,
+    eval_run: EvaluationRun,
+    judgeable: list[tuple[str, str, dict[str, Any]]],
+    log_prefix: str,
+) -> tuple[dict[str, JudgeResult], set[str], str | None]:
+    """Run one combined judge completion per judgeable row, isolated per row."""
+    results: dict[str, JudgeResult] = {}
+    failed_refs: set[str] = set()
+    if not judgeable:
+        return results, failed_refs, None
+
+    metrics = enabled_metric_specs()
+
+    # Build base params once per run; judging is system-config only, so every metric
+    # uses its built-in prompt + fallback model.
+    try:
+        base_params, _system_prompt = build_judge_params(
+            session=session, metrics=metrics
+        )
+    except Exception as exc:
+        logger.error(
+            f"[_judge_rows] {log_prefix} Judge setup failed; leaving all rows "
+            f"unjudged | error={exc}",
+            exc_info=True,
+        )
+        return results, {ref for _item_id, ref, _r in judgeable}, None
+
+    judge_model = base_params.get("model")
+
+    max_workers = max(1, min(settings.EVAL_FAST_API_CONCURRENCY, len(judgeable)))
+    with ThreadPoolExecutor(max_workers=max_workers) as executor:
+        future_map = {
+            executor.submit(
+                judge_row,
+                openai_client=openai_client,
+                base_params=base_params,
+                metrics=metrics,
+                question=response.get("question", ""),
+                generated_answer=response.get("generated_output", ""),
+                golden_answer=response.get("ground_truth", ""),
+            ): (item_id, ref)
+            for item_id, ref, response in judgeable
+        }
+        for future in as_completed(future_map):
+            item_id, ref = future_map[future]
+            try:
+                results[item_id] = future.result()
+            except Exception as exc:
+                failed_refs.add(ref)
+                logger.warning(
+                    f"[_judge_rows] {log_prefix} Judge failed for row; flagged "
+                    f"unscoreable | item_id={item_id} | ref={ref} | error={exc}"
+                )
+
+    return results, failed_refs, judge_model
+
+
+def _attach_metric_scores(
+    *,
+    spec: JudgeMetricSpec,
+    eval_run: EvaluationRun,
+    judge_results: dict[str, JudgeResult],
+    item_id_to_ref: dict[str, str],
+    summary_scores: list[dict[str, Any]],
+) -> None:
+    """Persist one metric's per-row map + summary score from the combined results.
+
+    Iterates the registry so knowledge_base / prompt need only add a spec + column.
+    Per-trace score comments are attached separately in the trace-build loop.
+    """
+    per_item: dict[str, float] = {}
+    values: list[float] = []
+    for item_id, result in judge_results.items():
+        metric_score = result.metrics.get(spec.key)
+        if metric_score is None:
+            continue
+        ref = item_id_to_ref[item_id]
+        per_item[ref] = round(float(metric_score.score), 6)
+        values.append(metric_score.score)
+
+    # Durable {ref: score} map on the metric's own column (Kaapi-native store).
+    setattr(eval_run, spec.per_item_column, per_item or None)
+
+    if values:
+        arr = np.array(values)
+        summary_scores.append(
+            {
+                "name": spec.score_name,
+                "avg": round(float(np.mean(arr)), 2),
+                "std": round(float(np.std(arr)), 2),
+                "total_pairs": len(values),
+                "data_type": "NUMERIC",
+            }
+        )
+
+
 def _stage3_score_and_trace(
     *,
     session: Session,
+    openai_client: OpenAI,
     eval_run: EvaluationRun,
-    langfuse: Langfuse,
+    langfuse: Langfuse | None,
     response_results: list[dict[str, Any]],
-    embedding_results: list[dict[str, Any]],
+    embedding_results: list[dict[str, Any]] | None,
     log_prefix: str,
 ) -> tuple[EvaluationRun, EvaluationScore, list[dict[str, Any]]]:
-    """Stage 3 — compute cosine, create Langfuse traces, attach costs.
+    """Stage 3 — cosine (v1) or judge (v2), create traces, attach costs.
 
     Returns the run, the full score unit (summary_scores + per-trace records, in
     the batch path's shape), and the Langfuse `write_items` (per-item cosine +
-    unscoreable placeholders). The caller completes the run before writing those
-    items, so completion is not gated on Langfuse calls. Idempotent, no stage
-    marker.
+    unscoreable placeholders; empty for v2). Everything is keyed by `ref` (trace_id
+    when traced, else item_id) so it works with or without Langfuse.
+
+    Two mutually-exclusive scoring paths, gated on `eval_run.is_judge_run`:
+      - v1 (`is_judge_run` false): cosine over the embedded pairs, per_item_scores,
+        the Cosine summary score, and Langfuse cosine sync — unchanged.
+      - v2 (`is_judge_run` true): no embeddings ran, so cosine is skipped entirely
+        (`embedding_results` is None/empty); only the ground-truth judge scores each
+        row, per_item_scores stays NULL, and there are no Langfuse writes.
+    A judge failure can never block a v1 cosine score because v1 never judges.
+    Idempotent, no stage marker.
     """
+    is_judge_run = eval_run.is_judge_run
     logger.info(
-        f"[_stage3_score_and_trace] {log_prefix} Computing cosine + creating traces"
+        f"[_stage3_score_and_trace] {log_prefix} Scoring stage 3 | "
+        f"judge_run={is_judge_run}"
     )
 
     item_id_to_pair = {
-        r["item_id"]: r for r in embedding_results if not r.get("failed")
+        r["item_id"]: r for r in (embedding_results or []) if not r.get("failed")
     }
 
     model = resolve_model_from_config(session=session, eval_run=eval_run)
@@ -761,68 +885,88 @@ def _stage3_score_and_trace(
         model=model,
     )
 
-    # Per-item cosine scores, keyed by trace_id (Langfuse) and item_id (persisted
-    # records). Items with a trace but no computable score are flagged unscoreable
-    # so they're kept out of avg/std/total_pairs.
+    # Scoring accumulators keyed by ref (trace_id when traced, else item_id) so they
+    # persist with tracing off. Unscoreable rows stay out of avg/std/total_pairs. The
+    # cosine-only fields (per_item_scores, similarities, write_items) stay empty for v2.
     per_item_scores: list[dict[str, Any]] = []
     item_id_to_score: dict[str, float] = {}
+    item_id_to_ref: dict[str, str] = {}
     similarities: list[float] = []
-    unscoreable: dict[str, str] = {}  # {trace_id: reason}
-    for response in response_results:
-        item_id = response["item_id"]
-        trace_id = trace_id_mapping.get(item_id)
-        if not trace_id:
-            continue
-        embedding_pair = item_id_to_pair.get(item_id)
-        has_embeddings = (
-            embedding_pair is not None
-            and embedding_pair.get("output_embedding") is not None
-            and embedding_pair.get("ground_truth_embedding") is not None
-        )
-        if not has_embeddings:
-            # Classify why this item cannot be scored, for the UI flag.
+    unscoreable: dict[str, str] = {}  # {ref: reason}
+    write_items: list[dict[str, Any]] = []
+    summary_scores: list[dict[str, Any]] = []
+
+    if is_judge_run:
+        # v2: no cosine. A row is judgeable only with a non-empty generated AND
+        # golden answer; empty sides are unscoreable and skip the judge below.
+        for response in response_results:
+            item_id = response["item_id"]
+            ref = trace_id_mapping.get(item_id) or item_id
+            item_id_to_ref[item_id] = ref
             if not response.get("generated_output"):
-                unscoreable[trace_id] = "empty_output"
+                unscoreable[ref] = UNSCOREABLE_EMPTY_OUTPUT
             elif not response.get("ground_truth"):
-                unscoreable[trace_id] = "empty_ground_truth"
-            else:
-                unscoreable[trace_id] = "embedding_failed"
-            continue
-        cosine = calculate_cosine_similarity(
-            embedding_pair["output_embedding"],
-            embedding_pair["ground_truth_embedding"],
-        )
-        similarities.append(cosine)
-        item_id_to_score[item_id] = cosine
-        per_item_scores.append({"trace_id": trace_id, "cosine_similarity": cosine})
-
-    # Langfuse write list (cosine + 0-scores for unscoreable items); written
-    # after completion in run_fast_evaluation.
-    unscoreable_writes = [
-        {"trace_id": trace_id, "unscoreable": True, "reason": reason}
-        for trace_id, reason in unscoreable.items()
-    ]
-    write_items = per_item_scores + unscoreable_writes
-
-    # Durable source of truth, persisted by the commit below.
-    eval_run.per_item_scores = {
-        trace_id_mapping[item_id]: round(float(score), 6)
-        for item_id, score in item_id_to_score.items()
-        if item_id in trace_id_mapping
-    }
-    eval_run.unscoreable = unscoreable or None
-
-    # Aggregate similarity stats, in the batch path's summary_scores shape.
-    if similarities:
-        sim_array = np.array(similarities)
-        avg = float(np.mean(sim_array))
-        std = float(np.std(sim_array))
+                unscoreable[ref] = UNSCOREABLE_EMPTY_GROUND_TRUTH
     else:
-        avg = 0.0
-        std = 0.0
+        for response in response_results:
+            item_id = response["item_id"]
+            ref = trace_id_mapping.get(item_id) or item_id
+            item_id_to_ref[item_id] = ref
+            embedding_pair = item_id_to_pair.get(item_id)
+            has_embeddings = (
+                embedding_pair is not None
+                and embedding_pair.get("output_embedding") is not None
+                and embedding_pair.get("ground_truth_embedding") is not None
+            )
+            if not has_embeddings:
+                # Classify why this item cannot be scored, for the UI flag.
+                if not response.get("generated_output"):
+                    unscoreable[ref] = UNSCOREABLE_EMPTY_OUTPUT
+                elif not response.get("ground_truth"):
+                    unscoreable[ref] = UNSCOREABLE_EMPTY_GROUND_TRUTH
+                else:
+                    unscoreable[ref] = UNSCOREABLE_EMBEDDING_FAILED
+                continue
+            cosine = calculate_cosine_similarity(
+                embedding_pair["output_embedding"],
+                embedding_pair["ground_truth_embedding"],
+            )
+            similarities.append(cosine)
+            item_id_to_score[item_id] = cosine
+            per_item_scores.append(
+                {"trace_id": trace_id_mapping.get(item_id), "cosine_similarity": cosine}
+            )
+
+        # Langfuse write list, filtered to real trace_ids (empty when untraced).
+        unscoreable_writes = [
+            {
+                "trace_id": trace_id_mapping[item_id],
+                "unscoreable": True,
+                "reason": reason,
+            }
+            for item_id, ref in item_id_to_ref.items()
+            if item_id in trace_id_mapping
+            and (reason := unscoreable.get(ref)) is not None
+        ]
+        scored_writes = [w for w in per_item_scores if w["trace_id"] is not None]
+        write_items = scored_writes + unscoreable_writes
+
+        # Durable source of truth, keyed by ref, persisted by the commit below.
+        eval_run.per_item_scores = {
+            item_id_to_ref[item_id]: round(float(score), 6)
+            for item_id, score in item_id_to_score.items()
+        }
 
-    score_payload = {
-        "summary_scores": apply_cosine_breakdown(
+        # Aggregate similarity stats, in the batch path's summary_scores shape.
+        if similarities:
+            sim_array = np.array(similarities)
+            avg = float(np.mean(sim_array))
+            std = float(np.std(sim_array))
+        else:
+            avg = 0.0
+            std = 0.0
+
+        summary_scores = apply_cosine_breakdown(
             [
                 {
                     "name": COSINE_SCORE_NAME,
@@ -833,9 +977,8 @@ def _stage3_score_and_trace(
                 }
             ],
             total_items=eval_run.total_items,
-            unscoreable=eval_run.unscoreable,
+            unscoreable=unscoreable or None,
         )
-    }
 
     # Attach response- and embedding-stage costs (attach_cost is idempotent per stage).
     if response_results:
@@ -870,40 +1013,103 @@ def _stage3_score_and_trace(
                 embedding_raw_results=embedding_raw,
             )
 
-    # Build the per-trace records in the same shape the batch path persists (via
-    # fetch_trace_scores_from_langfuse). One record per response that has a
-    # Langfuse trace; the cosine score is attached when its embedding succeeded.
+    judge_results: dict[str, JudgeResult] = {}
+    if eval_run.is_judge_run:
+        judgeable = [
+            (response["item_id"], item_id_to_ref[response["item_id"]], response)
+            for response in response_results
+            if response.get("generated_output") and response.get("ground_truth")
+        ]
+        judge_results, judge_failed_refs, judge_model = _judge_rows(
+            session=session,
+            openai_client=openai_client,
+            eval_run=eval_run,
+            judgeable=judgeable,
+            log_prefix=log_prefix,
+        )
+
+        # Flag judge-failed rows unscoreable WITHOUT clobbering an empty-side reason
+        # (setdefault): a row already flagged empty_output/empty_ground_truth keeps it.
+        for ref in judge_failed_refs:
+            unscoreable.setdefault(ref, JUDGE_FAILED_REASON)
+
+        for spec in enabled_metric_specs():
+            _attach_metric_scores(
+                spec=spec,
+                eval_run=eval_run,
+                judge_results=judge_results,
+                item_id_to_ref=item_id_to_ref,
+                summary_scores=summary_scores,
+            )
+
+        # One combined call per row; attribute its usage to the primary metric's
+        # cost stage (single-metric slice — per-metric cost split is deferred).
+        if judge_results and judge_model:
+            attach_cost(
+                session=session,
+                eval_run=eval_run,
+                log_prefix=log_prefix,
+                judge_stage=enabled_metric_specs()[0].cost_stage,
+                judge_model=judge_model,
+                judge_results=[
+                    {"usage": result.usage} for result in judge_results.values()
+                ],
+            )
+
+    eval_run.unscoreable = unscoreable or None
+
+    # Per-trace records, in the batch path's shape. Keyed by ref so untraced runs
+    # persist too. Judge metric scores carry their reasoning in the score comment.
     traces: list[TraceData] = []
     for response in response_results:
         item_id = response["item_id"]
-        trace_id = trace_id_mapping.get(item_id)
-        if not trace_id:
-            continue
+        ref = item_id_to_ref.get(item_id, item_id)
         trace_scores: list[TraceScore] = []
-        cosine = item_id_to_score.get(item_id)
-        if cosine is not None:
-            trace_scores.append(
-                {
-                    "name": COSINE_SCORE_NAME,
-                    "value": round(cosine, 2),
-                    "data_type": "NUMERIC",
-                    "comment": COSINE_SCORE_COMMENT,
-                }
-            )
-        elif trace_id in unscoreable:
-            # Placeholder 0-score, excluded from summary stats via the marker.
-            trace_scores.append(
-                {
-                    "name": COSINE_SCORE_NAME,
-                    "value": 0,
-                    "data_type": "NUMERIC",
-                    "comment": f"Cannot compute: {unscoreable[trace_id]}",
-                    "unscoreable": True,
-                }
-            )
+        # v2 judged runs carry no cosine score or cosine placeholder — only the
+        # ground-truth judge score below.
+        if not is_judge_run:
+            cosine = item_id_to_score.get(item_id)
+            if cosine is not None:
+                trace_scores.append(
+                    {
+                        "name": COSINE_SCORE_NAME,
+                        "value": round(cosine, 2),
+                        "data_type": "NUMERIC",
+                        "comment": COSINE_SCORE_COMMENT,
+                    }
+                )
+            elif ref in unscoreable and unscoreable[ref] != JUDGE_FAILED_REASON:
+                # Placeholder 0-score, excluded from summary stats via the marker. A
+                # judge_failed-only reason is about the judge, not cosine, so it gets
+                # no cosine placeholder.
+                trace_scores.append(
+                    {
+                        "name": COSINE_SCORE_NAME,
+                        "value": 0,
+                        "data_type": "NUMERIC",
+                        "comment": f"Cannot compute: {unscoreable[ref]}",
+                        "unscoreable": True,
+                    }
+                )
+
+        judge_result = judge_results.get(item_id)
+        if judge_result is not None:
+            for spec in enabled_metric_specs():
+                metric_score = judge_result.metrics.get(spec.key)
+                if metric_score is None:
+                    continue
+                trace_scores.append(
+                    {
+                        "name": spec.score_name,
+                        "value": round(metric_score.score, 2),
+                        "data_type": "NUMERIC",
+                        "comment": metric_score.reasoning,
+                    }
+                )
+
         traces.append(
             {
-                "trace_id": trace_id,
+                "trace_id": ref,
                 "question": response.get("question", ""),
                 "llm_answer": response.get("generated_output", ""),
                 "ground_truth_answer": response.get("ground_truth", ""),
@@ -912,16 +1118,20 @@ def _stage3_score_and_trace(
             }
         )
 
-    # Persist cost here; the score unit (summary + traces) is persisted by the
-    # caller via save_score so it lands in S3 (score_trace_url) like the batch path.
+    # Persist cost + the durable judge maps here; the score unit (summary + traces)
+    # is persisted by the caller via save_score so it lands in S3 like the batch path.
     eval_run = update_evaluation_run(
         session=session,
         eval_run=eval_run,
-        update=EvaluationRunUpdate(cost=eval_run.cost),
+        update=EvaluationRunUpdate(
+            cost=eval_run.cost,
+            unscoreable=eval_run.unscoreable,
+            per_item_ground_truth=eval_run.per_item_ground_truth,
+        ),
     )
 
     score: EvaluationScore = {
-        "summary_scores": score_payload["summary_scores"],
+        "summary_scores": summary_scores,
         "traces": traces,
     }
     return eval_run, score, write_items
@@ -931,7 +1141,7 @@ def run_fast_evaluation(
     *,
     session: Session,
     openai_client: OpenAI,
-    langfuse: Langfuse,
+    langfuse: Langfuse | None,
     eval_run: EvaluationRun,
 ) -> EvaluationRun:
     """Merge the response chunks, then run embeddings + scoring + completion.
@@ -940,6 +1150,10 @@ def run_fast_evaluation(
     only after every chunk has a raw_output_url). Stages are skipped on retry
     when their batch_job marker is set. Raises on terminal failure (run marked
     failed).
+
+    `langfuse` is None for v2 judged runs (fully Kaapi-native, no trace creation
+    or score sync) and for tracing-opted-out projects; scoring falls back to
+    keying by item_id. Whether the run judges is read from `eval_run.is_judge_run`.
     """
     log_prefix = (
         f"[org={eval_run.organization_id}]"
@@ -972,18 +1186,22 @@ def run_fast_evaluation(
             f"threshold={settings.EVAL_FAST_FAILURE_THRESHOLD}"
         )
 
-    # Stage 2
-    eval_run, embedding_results = _stage2_embeddings(
-        session=session,
-        openai_client=openai_client,
-        eval_run=eval_run,
-        response_results=response_results,
-        log_prefix=log_prefix,
-    )
+    # Stage 2 — embeddings feed only cosine, so v2 judged runs skip it entirely
+    # (no embedding API calls, no embedding_batch_job). v1 embeds exactly as before.
+    embedding_results: list[dict[str, Any]] | None = None
+    if not eval_run.is_judge_run:
+        eval_run, embedding_results = _stage2_embeddings(
+            session=session,
+            openai_client=openai_client,
+            eval_run=eval_run,
+            response_results=response_results,
+            log_prefix=log_prefix,
+        )
 
     # Stage 3
     eval_run, score, write_items = _stage3_score_and_trace(
         session=session,
+        openai_client=openai_client,
         eval_run=eval_run,
         langfuse=langfuse,
         response_results=response_results,
@@ -1007,9 +1225,10 @@ def run_fast_evaluation(
 
     # Stage 5a — write cosine scores to Langfuse after completion (mirrors the
     # batch path). is_score_updated tracks the outcome so a cron can retry the
-    # gap from per_item_scores.
+    # gap from per_item_scores. Skipped entirely when langfuse is None — v2 judged
+    # runs are Kaapi-native and never sync scores to Langfuse.
     is_score_updated = True
-    if write_items:
+    if langfuse is not None and write_items:
         try:
             failed_trace_ids = update_traces_with_cosine_scores(
                 langfuse=langfuse, per_item_scores=write_items
diff --git a/backend/app/crud/evaluations/judge.py b/backend/app/crud/evaluations/judge.py
new file mode 100644
index 000000000..f2fe979d9
--- /dev/null
+++ b/backend/app/crud/evaluations/judge.py
@@ -0,0 +1,323 @@
+"""Native LLM-as-a-judge for v2 fast evaluations.
+
+Runs one OpenAI judge call per row (after the answers are generated) to score all
+enabled metrics together. v2 runs are judge-only — no cosine, no embeddings — and
+v1 never invokes this judge. A per-row judge failure is isolated to that row.
+Metrics are registry-driven and easily extensible.
+"""
+
+import json
+import logging
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+import openai
+from openai import OpenAI
+from sqlmodel import Session
+from tenacity import (
+    before_sleep_log,
+    retry,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_random_exponential,
+)
+
+from app.core.config import settings
+from app.crud.evaluations.score import (
+    GROUND_TRUTH_JUDGE_PROMPT,
+    GROUND_TRUTH_SCORE_NAME,
+    JUDGE_OUTPUT_INSTRUCTION,
+    JUDGE_SYSTEM_PREAMBLE,
+)
+from app.services.llm.mappers import map_kaapi_to_openai_params
+
+logger = logging.getLogger(__name__)
+
+
+class JudgeMetricEnum(str, Enum):
+    """Registry keys — also the JSON keys the combined judge returns per metric."""
+
+    GROUND_TRUTH = "ground_truth"
+
+
+class JudgeInputEnum(str, Enum):
+    """Per-row inputs a metric may require in the composed judge prompt."""
+
+    QUESTION = "question"
+    GENERATED_ANSWER = "generated_answer"
+    GOLDEN_ANSWER = "golden_answer"
+
+
+_INPUT_LABELS: dict[JudgeInputEnum, str] = {
+    JudgeInputEnum.QUESTION: "Question",
+    JudgeInputEnum.GENERATED_ANSWER: "Generated answer",
+    JudgeInputEnum.GOLDEN_ANSWER: "Golden (reference) answer",
+}
+
+
+@dataclass(frozen=True)
+class JudgeMetricSpec:
+    """Everything the pipeline needs to run and persist one judge metric."""
+
+    key: JudgeMetricEnum
+    score_name: str
+    prompt_fragment: str
+    required_inputs: tuple[JudgeInputEnum, ...]
+    per_item_column: str
+    cost_stage: str
+
+
+# Phase 1: only ground_truth, knowledge_base / prompt slot in here. All
+# metrics are graded by one combined call, so they share a single judge model
+# (settings.EVAL_JUDGE_MODEL); there is no per-metric model.
+METRIC_REGISTRY: dict[JudgeMetricEnum, JudgeMetricSpec] = {
+    JudgeMetricEnum.GROUND_TRUTH: JudgeMetricSpec(
+        key=JudgeMetricEnum.GROUND_TRUTH,
+        score_name=GROUND_TRUTH_SCORE_NAME,
+        prompt_fragment=GROUND_TRUTH_JUDGE_PROMPT,
+        required_inputs=(
+            JudgeInputEnum.QUESTION,
+            JudgeInputEnum.GENERATED_ANSWER,
+            JudgeInputEnum.GOLDEN_ANSWER,
+        ),
+        per_item_column="per_item_ground_truth",
+        cost_stage="ground_truth_judge",
+    ),
+}
+
+
+def enabled_metric_specs() -> list[JudgeMetricSpec]:
+    """The metrics a judged run scores. Phase 1 always scores every registry entry."""
+    return list(METRIC_REGISTRY.values())
+
+
+# Per-call retry mechanism (mirrors the fast-eval Responses/Embeddings stages).
+_RETRY_MAX_ATTEMPTS = 3
+_RETRY_BASE_DELAY_SECONDS = 1.0
+_RETRY_MAX_DELAY_SECONDS = 30.0
+
+_RETRYABLE_OPENAI_ERRORS: tuple[type[Exception], ...] = (
+    openai.RateLimitError,
+    openai.APITimeoutError,
+    openai.APIConnectionError,
+    openai.InternalServerError,
+)
+
+# reraise=True so the call-site handler sees the original OpenAIError.
+_retry_judge_call = retry(
+    retry=retry_if_exception_type(_RETRYABLE_OPENAI_ERRORS),
+    wait=wait_random_exponential(
+        multiplier=_RETRY_BASE_DELAY_SECONDS, max=_RETRY_MAX_DELAY_SECONDS
+    ),
+    stop=stop_after_attempt(_RETRY_MAX_ATTEMPTS),
+    before_sleep=before_sleep_log(logger, logging.INFO),
+    reraise=True,
+)
+
+
+@dataclass
+class MetricScore:
+    """One metric's outcome for a row: score in [0, 1] and its reasoning."""
+
+    score: float
+    reasoning: str
+
+
+@dataclass
+class JudgeResult:
+    """One row's combined judge outcome across all enabled metrics, plus usage."""
+
+    metrics: dict[JudgeMetricEnum, MetricScore]
+    usage: dict[str, int]
+
+
+def build_judge_params(
+    *,
+    session: Session,
+    metrics: list[JudgeMetricSpec],
+) -> tuple[dict[str, Any], str]:
+    """Build the model-independent OpenAI body and the combined judge system prompt.
+
+    Judging is system-config only: every metric uses its built-in rubric fragment,
+    and the single combined call runs on one shared judge model
+    (settings.EVAL_JUDGE_MODEL) for all metrics. The system prompt is the shared
+    preamble followed by each enabled metric's fragment.
+    """
+    judge_params: dict[str, Any] = {
+        "model": settings.EVAL_JUDGE_MODEL,
+        "effort": settings.EVAL_JUDGE_REASONING_EFFORT,
+    }
+
+    base_params, mapper_warnings = map_kaapi_to_openai_params(
+        session=session, kaapi_params=judge_params
+    )
+    if mapper_warnings:
+        logger.warning(f"[build_judge_params] Mapper warnings: {mapper_warnings}")
+
+    fragments = "\n\n".join(spec.prompt_fragment for spec in metrics)
+    system_prompt = f"{JUDGE_SYSTEM_PREAMBLE}\n\n{fragments}"
+    # The judge prompt IS the instructions; overwrite anything the mapper carried.
+    base_params["instructions"] = system_prompt
+    return base_params, system_prompt
+
+
+def _compose_judge_input(
+    *, metrics: list[JudgeMetricSpec], inputs: dict[JudgeInputEnum, str]
+) -> str:
+    """Append the row's inputs (union of enabled metrics' needs) + the output contract.
+
+    Metric prompts carry no interpolation placeholder — inputs are appended here.
+    """
+    required: list[JudgeInputEnum] = []
+    for spec in metrics:
+        for key in spec.required_inputs:
+            if key not in required:
+                required.append(key)
+
+    # Render in stable enum order regardless of registry declaration order.
+    blocks = [
+        f"{_INPUT_LABELS[key]}:\n{inputs.get(key, '')}"
+        for key in JudgeInputEnum
+        if key in required
+    ]
+    metric_keys = ", ".join(spec.key.value for spec in metrics)
+    contract = JUDGE_OUTPUT_INSTRUCTION.format(metric_keys=metric_keys)
+    return "\n\n".join(blocks) + "\n\n" + contract
+
+
+def _extract_response_text(response: Any) -> str:
+    """Extract generated text, preferring `output_text` then walking `output`."""
+    output_text = getattr(response, "output_text", None)
+    if output_text:
+        return output_text
+
+    output = getattr(response, "output", None) or []
+    for item in output:
+        if getattr(item, "type", None) != "message":
+            continue
+        for content in getattr(item, "content", None) or []:
+            if getattr(content, "type", None) == "output_text":
+                text = getattr(content, "text", None)
+                if text:
+                    return text
+    return ""
+
+
+def _parse_metric_score(key: JudgeMetricEnum, raw: Any) -> MetricScore:
+    """Parse one metric's {"score", "reasoning"} object. Raises ValueError if malformed."""
+    if not isinstance(raw, dict):
+        raise ValueError(f"metric '{key.value}' is not a JSON object: {raw!r}")
+    if "score" not in raw:
+        raise ValueError(f"metric '{key.value}' missing 'score': {raw}")
+    try:
+        score = float(raw["score"])
+    except (TypeError, ValueError) as exc:
+        raise ValueError(
+            f"metric '{key.value}' score is not a number: {raw.get('score')!r}"
+        ) from exc
+    if not 0.0 <= score <= 1.0:
+        raise ValueError(f"metric '{key.value}' score out of [0, 1]: {score}")
+
+    reasoning = str(raw.get("reasoning") or "").strip()
+    if not reasoning:
+        raise ValueError(f"metric '{key.value}' has empty 'reasoning'")
+    return MetricScore(score=score, reasoning=reasoning)
+
+
+def _parse_judge_output(
+    text: str, metrics: list[JudgeMetricSpec]
+) -> dict[JudgeMetricEnum, MetricScore]:
+    """Parse the combined reply into a per-metric score map.
+
+    Leniently extracts the outermost JSON object so a stray prose wrapper doesn't
+    break parsing. A well-formed object missing one metric leaves only that metric
+    unscoreable (dropped from the map); a fully malformed reply raises so the caller
+    isolates the whole row. Raises ValueError on anything unparseable.
+    """
+    if not text or not text.strip():
+        raise ValueError("empty judge response")
+
+    start = text.find("{")
+    end = text.rfind("}")
+    if start == -1 or end == -1 or end < start:
+        raise ValueError(f"no JSON object in judge response: {text[:200]!r}")
+
+    try:
+        data = json.loads(text[start : end + 1])
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"invalid JSON in judge response: {exc}") from exc
+    if not isinstance(data, dict):
+        raise ValueError(f"judge response is not a JSON object: {data!r}")
+
+    results: dict[JudgeMetricEnum, MetricScore] = {}
+    for spec in metrics:
+        raw = data.get(spec.key.value)
+        if raw is None:
+            # Well-formed but missing this metric: only this metric is unscoreable.
+            continue
+        results[spec.key] = _parse_metric_score(spec.key, raw)
+
+    if not results:
+        raise ValueError(f"judge response scored no enabled metric: {data}")
+    return results
+
+
+@_retry_judge_call
+def _create_judge_response(openai_client: OpenAI, params: dict[str, Any]) -> Any:
+    return openai_client.responses.create(**params)
+
+
+def judge_row(
+    *,
+    openai_client: OpenAI,
+    base_params: dict[str, Any],
+    metrics: list[JudgeMetricSpec],
+    question: str,
+    generated_answer: str,
+    golden_answer: str,
+) -> JudgeResult:
+    """Run one combined judge completion for a row and return its per-metric result.
+
+    `base_params` is the model-independent body from `build_judge_params`, built
+    once per run; only `input` varies per row. Raises on retry-exhausted OpenAI
+    errors or malformed output so the caller can flag the whole row unscoreable.
+    """
+    model = base_params.get("model")
+    params = {
+        **base_params,
+        "input": _compose_judge_input(
+            metrics=metrics,
+            inputs={
+                JudgeInputEnum.QUESTION: question,
+                JudgeInputEnum.GENERATED_ANSWER: generated_answer,
+                JudgeInputEnum.GOLDEN_ANSWER: golden_answer,
+            },
+        ),
+    }
+
+    try:
+        response = _create_judge_response(openai_client, params)
+    except openai.OpenAIError as exc:
+        status = getattr(exc, "status_code", None)
+        # 5xx is provider-side (alert-worthy); 4xx/None is caller/Kaapi-side noise.
+        log = logger.error if (status and status >= 500) else logger.warning
+        tag = "[OPENAI]" if status else "[KAAPI]"
+        request_id = getattr(exc, "request_id", None)
+        log(
+            f"[judge_row] {tag} Judge completion failed "
+            f"(code: {status or type(exc).__name__}) | model={model} | "
+            f"request_id={request_id} | {exc}",
+            exc_info=True,
+        )
+        raise
+
+    usage_obj = getattr(response, "usage", None)
+    usage = {
+        "input_tokens": int(getattr(usage_obj, "input_tokens", 0) or 0),
+        "output_tokens": int(getattr(usage_obj, "output_tokens", 0) or 0),
+        "total_tokens": int(getattr(usage_obj, "total_tokens", 0) or 0),
+    }
+
+    metric_scores = _parse_judge_output(_extract_response_text(response), metrics)
+    return JudgeResult(metrics=metric_scores, usage=usage)
diff --git a/backend/app/crud/evaluations/langfuse.py b/backend/app/crud/evaluations/langfuse.py
index ad2a52f70..53f6f2473 100644
--- a/backend/app/crud/evaluations/langfuse.py
+++ b/backend/app/crud/evaluations/langfuse.py
@@ -46,7 +46,7 @@ def _write_trace_score(
 
 
 def create_langfuse_dataset_run(
-    langfuse: Langfuse,
+    langfuse: Langfuse | None,
     dataset_name: str,
     run_name: str,
     results: list[dict[str, Any]],
@@ -93,6 +93,15 @@ def create_langfuse_dataset_run(
     Raises:
         Exception: If Langfuse operations fail
     """
+    # v2 native (judged) runs and tracing-opted-out projects pass langfuse=None:
+    # no traces are created, and callers fall back to keying scores by item_id.
+    if langfuse is None:
+        logger.info(
+            "[create_langfuse_dataset_run] No Langfuse client; skipping trace "
+            f"creation | run_name={run_name} | dataset={dataset_name}"
+        )
+        return {}
+
     logger.info(
         f"[create_langfuse_dataset_run] Creating Langfuse dataset run | "
         f"run_name={run_name} | dataset={dataset_name} | items={len(results)}"
diff --git a/backend/app/crud/evaluations/score.py b/backend/app/crud/evaluations/score.py
index bc484b03d..bd00782d8 100644
--- a/backend/app/crud/evaluations/score.py
+++ b/backend/app/crud/evaluations/score.py
@@ -7,7 +7,6 @@
 
 from typing import NotRequired, TypedDict
 
-
 DEFAULT_CATEGORY: str = "Other"
 
 # Canonical name/comment for the cosine-similarity score, centralized to avoid
@@ -17,6 +16,9 @@
     "Cosine similarity between generated output and ground truth embeddings"
 )
 
+GROUND_TRUTH_SCORE_NAME: str = "Adherence to Ground Truth"
+JUDGE_FAILED_REASON: str = "judge_failed"
+
 # Reasons an item cannot be scored, recorded in EvaluationRun.unscoreable.
 # "missing_trace_id" appears only in build_embedding_jsonl's internal skipped list.
 UNSCOREABLE_REASONS: tuple[str, ...] = (
@@ -24,6 +26,37 @@
     "empty_ground_truth",
     "embedding_failed",
     "missing_trace_id",
+    JUDGE_FAILED_REASON,
+)
+
+JUDGE_SYSTEM_PREAMBLE: str = (
+    "You are a strict, impartial evaluator. You score an assistant's answer on the "
+    "independent metrics listed below in a single pass. Each metric is a float in "
+    "[0.0, 1.0] with one or two sentences of reasoning. The metrics are independent "
+    "— judge each only against its own inputs and rules; do not let one metric's "
+    "verdict bleed into another."
+)
+
+GROUND_TRUTH_JUDGE_PROMPT: str = (
+    'Adherence to Ground Truth (score key "ground_truth"):\n'
+    "Judge ONLY whether the assistant's answer conveys the same correct information "
+    "as the golden answer.\n"
+    "- Judge meaning, not wording. A correct paraphrase, a different order, or extra "
+    "detail that is also correct must score high.\n"
+    "- Lower the score for information that is missing, incomplete, or contradicts "
+    "the golden answer. An answer that states something the golden answer does not, "
+    "and that would be wrong, is a factual error.\n"
+    "- Do NOT reward or penalize style, tone, length, or language.\n"
+    "- Do NOT use any outside knowledge; the golden answer is the source of truth.\n"
+    "- Do NOT answer the question yourself.\n"
+    "Reasoning: name what was correct or what was missing/contradicted."
+)
+
+JUDGE_OUTPUT_INSTRUCTION: str = (
+    "Respond with ONLY a single JSON object mapping each metric key to its result, of "
+    'the form {{"<metric_key>": {{"score": <float between 0 and 1>, "reasoning": '
+    '"<one or two sentences>"}}}}. Include exactly these metric keys: {metric_keys}. '
+    "Output nothing else."
 )
 
 
diff --git a/backend/app/main.py b/backend/app/main.py
index 474837759..5d2a09cf0 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -1,6 +1,10 @@
 import logging
 
 import sentry_sdk
+from asgi_correlation_id.middleware import CorrelationIdMiddleware
+from fastapi import FastAPI
+from fastapi.openapi.utils import get_openapi
+from fastapi.routing import APIRoute
 from sentry_sdk.integrations.celery import CeleryIntegration
 from sentry_sdk.integrations.fastapi import FastApiIntegration
 from sentry_sdk.integrations.httpx import HttpxIntegration
@@ -8,19 +12,14 @@
 from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
 from sentry_sdk.integrations.starlette import StarletteIntegration
 
-from fastapi import FastAPI
-from fastapi.routing import APIRoute
-from fastapi.openapi.utils import get_openapi
-from asgi_correlation_id.middleware import CorrelationIdMiddleware
-from app.api.main import api_router
-from app.api.docs.openapi_config import tags_metadata, customize_openapi_schema
+from app.api.docs.openapi_config import customize_openapi_schema, tags_metadata
+from app.api.main import api_router, api_v2_router
 from app.core.config import settings
 from app.core.exception_handlers import register_exception_handlers
 from app.core.logger import configure_logging
 from app.core.middleware import StripTrailingSlashMiddleware, http_request_logger
 from app.core.sentry_filters import before_send_transaction_filter
 from app.core.telemetry import instrument_app, setup_telemetry
-
 from app.load_env import load_environment
 
 # Load environment variables
@@ -94,6 +93,7 @@ def custom_openapi():
 app.add_middleware(StripTrailingSlashMiddleware)
 
 app.include_router(api_router, prefix=settings.API_V1_STR)
+app.include_router(api_v2_router, prefix=settings.API_V2_STR)
 
 register_exception_handlers(app)
 
diff --git a/backend/app/models/evaluation.py b/backend/app/models/evaluation.py
index 92ef8770d..9f07fe9b3 100644
--- a/backend/app/models/evaluation.py
+++ b/backend/app/models/evaluation.py
@@ -6,7 +6,8 @@
 from pydantic import BaseModel, Field, HttpUrl
 from sqlalchemy import Boolean, Column, Index, Text, UniqueConstraint, text
 from sqlalchemy.dialects.postgresql import ENUM, JSONB
-from sqlmodel import Field as SQLField, Relationship, SQLModel
+from sqlmodel import Field as SQLField
+from sqlmodel import Relationship, SQLModel
 
 from app.core.util import now
 from app.models.config.version import ConfigVersionPublic
@@ -371,12 +372,37 @@ class EvaluationRun(SQLModel, table=True):
             nullable=True,
             comment=(
                 "{trace_id: reason} for items that cannot be scored "
-                "(empty_output / empty_ground_truth / embedding_failed)"
+                "(empty_output / empty_ground_truth / embedding_failed / judge_failed)"
             ),
         ),
         description="Map of trace_id to the reason the item cannot be scored",
     )
 
+    is_judge_run: bool | None = SQLField(
+        default=None,
+        sa_column=Column(
+            Boolean,
+            nullable=True,
+            comment=(
+                "True for v2 runs that run the native LLM-as-judge (and skip the "
+                "Langfuse score sync). NULL/False = v1 run, cosine-only, Langfuse-synced"
+            ),
+        ),
+        description="Marks a v2 judged run; gates judging and the Langfuse-sync skip",
+    )
+    per_item_ground_truth: dict[str, Any] | None = SQLField(
+        default=None,
+        sa_column=Column(
+            JSONB,
+            nullable=True,
+            comment=(
+                "Durable {ref: score} map of the Adherence to Ground Truth judge "
+                "scores (ref = trace_id when traced, else item_id); Kaapi's own store"
+            ),
+        ),
+        description="Durable map of per-row ground-truth judge scores keyed by ref",
+    )
+
     is_score_updated: bool | None = SQLField(
         default=None,
         sa_column=Column(
@@ -488,6 +514,8 @@ class EvaluationRunUpdate(SQLModel):
     is_score_updated: bool | None = None
     cost: dict[str, Any] | None = None
     embedding_batch_job_id: int | None = None
+    is_judge_run: bool | None = None
+    per_item_ground_truth: dict[str, Any] | None = None
 
 
 class EvaluationRunPublic(SQLModel):
@@ -509,6 +537,8 @@ class EvaluationRunPublic(SQLModel):
     score: dict[str, Any] | None
     unscoreable: dict[str, Any] | None = None
     is_score_updated: bool | None = None
+    is_judge_run: bool | None = None
+    per_item_ground_truth: dict[str, Any] | None = None
     cost: dict[str, Any] | None
     error_message: str | None
     organization_id: int
diff --git a/backend/app/services/evaluations/__init__.py b/backend/app/services/evaluations/__init__.py
index 5a06cb7dc..192d181e2 100644
--- a/backend/app/services/evaluations/__init__.py
+++ b/backend/app/services/evaluations/__init__.py
@@ -1,6 +1,6 @@
 """Evaluation services."""
 
-from app.services.evaluations.dataset import upload_dataset
+from app.services.evaluations.dataset import upload_dataset, upload_dataset_v2
 from app.services.evaluations.evaluation import (
     get_evaluation_with_scores,
     validate_and_start_batch_evaluation,
@@ -24,3 +24,21 @@
     sanitize_dataset_name,
     validate_csv_file,
 )
+
+__all__ = [
+    "ALLOWED_EXTENSIONS",
+    "ALLOWED_MIME_TYPES",
+    "MAX_FILE_SIZE",
+    "execute_fast_evaluation_aggregate",
+    "execute_fast_evaluation_chunk",
+    "get_evaluation_with_scores",
+    "improve_prompt",
+    "is_dataset_fast_eligible",
+    "parse_csv_items",
+    "sanitize_dataset_name",
+    "upload_dataset",
+    "upload_dataset_v2",
+    "validate_and_start_batch_evaluation",
+    "validate_and_start_fast_evaluation",
+    "validate_csv_file",
+]
diff --git a/backend/app/services/evaluations/dataset.py b/backend/app/services/evaluations/dataset.py
index fe0e89249..e1603a605 100644
--- a/backend/app/services/evaluations/dataset.py
+++ b/backend/app/services/evaluations/dataset.py
@@ -11,6 +11,12 @@
     upload_csv_to_object_store,
     upload_dataset_to_langfuse,
 )
+from app.crud.evaluations.dataset import (
+    DATASET_META_DUPLICATE_AT_RUNTIME,
+    DATASET_META_DUPLICATION_FACTOR,
+    DATASET_META_ORIGINAL_ITEMS,
+    DATASET_META_TOTAL_ITEMS,
+)
 from app.models.evaluation import EvaluationDataset
 from app.services.evaluations.validators import (
     parse_csv_items,
@@ -161,3 +167,65 @@ def upload_dataset(
     )
 
     return dataset
+
+
+def upload_dataset_v2(
+    *,
+    session: Session,
+    csv_content: bytes,
+    dataset_name: str,
+    description: str | None,
+    duplication_factor: int,
+    organization_id: int,
+    project_id: int,
+) -> EvaluationDataset:
+    """Uploads a v2 dataset without Langfuse. Stores the original CSV in S3, records
+    `duplication_factor` for run-time expansion, and requires a successful S3 upload.
+    """
+    original_name = dataset_name
+    try:
+        dataset_name = sanitize_dataset_name(dataset_name)
+    except ValueError as e:
+        raise HTTPException(status_code=422, detail=f"Invalid dataset name: {str(e)}")
+
+    if original_name != dataset_name:
+        logger.info(
+            f"[upload_dataset_v2] Dataset name sanitized | "
+            f"'{original_name}' -> '{dataset_name}'"
+        )
+
+    original_items = parse_csv_items(csv_content)
+    original_items_count = len(original_items)
+    total_items_count = original_items_count * duplication_factor
+
+    storage = get_cloud_storage(session=session, project_id=project_id)
+    object_store_url = upload_csv_to_object_store(
+        storage=storage, csv_content=csv_content, dataset_name=dataset_name
+    )
+    if not object_store_url:
+        # No Langfuse fallback in v2: the run loads items from this CSV, so a
+        # missing object-store URL leaves the dataset unrunnable.
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to store dataset CSV in object store",
+        )
+
+    metadata = {
+        DATASET_META_ORIGINAL_ITEMS: original_items_count,
+        DATASET_META_TOTAL_ITEMS: total_items_count,
+        DATASET_META_DUPLICATION_FACTOR: duplication_factor,
+        DATASET_META_DUPLICATE_AT_RUNTIME: True,
+    }
+
+    dataset = create_evaluation_dataset(
+        session=session,
+        name=dataset_name,
+        description=description,
+        dataset_metadata=metadata,
+        object_store_url=object_store_url,
+        langfuse_dataset_id=None,
+        organization_id=organization_id,
+        project_id=project_id,
+    )
+
+    return dataset
diff --git a/backend/app/services/evaluations/fast.py b/backend/app/services/evaluations/fast.py
index c0951de30..20b0e94c6 100644
--- a/backend/app/services/evaluations/fast.py
+++ b/backend/app/services/evaluations/fast.py
@@ -9,6 +9,7 @@
 
 import logging
 import math
+from typing import Any
 from uuid import UUID
 
 from fastapi import HTTPException
@@ -17,6 +18,7 @@
 from sqlmodel import Session
 
 from app.celery.utils import start_fast_evaluation_chunk
+from app.core.cloud import get_cloud_storage
 from app.core.config import settings
 from app.core.db import engine
 from app.crud.evaluations import (
@@ -26,10 +28,22 @@
 )
 from app.crud.evaluations.batch import fetch_dataset_items
 from app.crud.evaluations.core import update_evaluation_run
+from app.crud.evaluations.dataset import (
+    DATASET_META_DUPLICATE_AT_RUNTIME,
+    DATASET_META_DUPLICATION_FACTOR,
+    download_csv_from_object_store,
+)
 from app.crud.evaluations.fast import run_response_chunk
-from app.models.evaluation import EvaluationRun, EvaluationRunUpdate, RunModeEnum
+from app.crud.evaluations.score import DEFAULT_CATEGORY
+from app.models.evaluation import (
+    EvaluationDataset,
+    EvaluationRun,
+    EvaluationRunUpdate,
+    RunModeEnum,
+)
 from app.models.llm.request import TextLLMParams
 from app.services.evaluations.evaluation import create_evaluation_run_or_409
+from app.services.evaluations.validators import parse_csv_items
 from app.services.llm.providers import LLMProvider
 from app.utils import get_langfuse_client, get_openai_client
 
@@ -46,6 +60,82 @@ def is_dataset_fast_eligible(*, original_items_count: int) -> bool:
     return original_items_count <= settings.EVAL_FAST_MAX_UNIQUE_ROWS
 
 
+def load_run_dataset_items(
+    *,
+    session: Session,
+    dataset: EvaluationDataset,
+    langfuse: Langfuse | None,
+) -> list[dict[str, Any]]:
+    """Load a run's dataset items, choosing the source by the dataset row.
+
+    - Langfuse-backed dataset (v1): items are already physically duplicated in
+      Langfuse; read as-is via `fetch_dataset_items`, never re-multiplied.
+    - S3-only dataset (v2, `langfuse_dataset_id` NULL): download the original-items
+      CSV and, when the dataset is marked for run-time duplication, expand each row
+      ×duplication_factor with a unique item id per copy.
+
+    Both the fan-out sizing and the per-chunk load call this, so they agree on the
+    same expanded item set.
+    """
+    if dataset.langfuse_dataset_id:
+        if langfuse is None:
+            raise ValueError(
+                f"Dataset {dataset.id} is Langfuse-backed but no Langfuse client "
+                "is available to load its items"
+            )
+        return fetch_dataset_items(langfuse=langfuse, dataset_name=dataset.name)
+
+    return _load_items_from_object_store(session=session, dataset=dataset)
+
+
+def _load_items_from_object_store(
+    *, session: Session, dataset: EvaluationDataset
+) -> list[dict[str, Any]]:
+    """Parse the dataset's original-items CSV from S3 into fast-pipeline items.
+
+    When the dataset is marked for run-time duplication (v2), each original row is
+    emitted `duplication_factor` times with a distinct item id (`item_{row}_{dup}`)
+    so per-row score keys stay unique. A v1 dataset's S3 CSV is already physically
+    duplicated, so it loads as-is (factor forced to 1)."""
+    if not dataset.object_store_url:
+        raise ValueError(f"Dataset {dataset.id} has no object-store CSV to load")
+
+    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)
+
+    metadata = dataset.dataset_metadata or {}
+    duplicate_at_runtime = bool(metadata.get(DATASET_META_DUPLICATE_AT_RUNTIME, False))
+    duplication_factor = (
+        max(1, int(metadata.get(DATASET_META_DUPLICATION_FACTOR, 1)))
+        if duplicate_at_runtime
+        else 1
+    )
+
+    items: list[dict[str, Any]] = []
+    for row_idx, item in enumerate(original_items):
+        for dup_idx in range(duplication_factor):
+            item_metadata: dict[str, Any] = {
+                # 1-based, shared across a row's duplicates so the Q.ID column
+                # groups by original question (mirrors the Langfuse upload path).
+                "question_id": row_idx
+                + 1,
+            }
+            if "category" in item:
+                item_metadata["category"] = item["category"] or DEFAULT_CATEGORY
+            items.append(
+                {
+                    "id": f"item_{row_idx}_{dup_idx}",
+                    "input": {"question": item["question"]},
+                    "expected_output": {"answer": item["answer"]},
+                    "metadata": item_metadata,
+                }
+            )
+    return items
+
+
 def validate_and_start_fast_evaluation(
     *,
     session: Session,
@@ -56,11 +146,13 @@ def validate_and_start_fast_evaluation(
     organization_id: int,
     project_id: int,
     trace_id: str = "N/A",
+    is_judge_run: bool = False,
 ) -> EvaluationRun:
     """Validate + create + dispatch a fast evaluation run.
 
     Validation (in order):
-    1. Dataset exists and has a Langfuse id.
+    1. Dataset exists; v1 runs also require a Langfuse id, v2 judged runs don't
+       (they load items from S3).
     2. Config resolves to a text-type OpenAI config.
     3. Dataset's original_items_count <= EVAL_FAST_MAX_UNIQUE_ROWS.
     4. (organization_id, project_id, run_name) is unique — enforced by the DB
@@ -69,6 +161,12 @@ def validate_and_start_fast_evaluation(
     On success the function creates the EvaluationRun row with
     `run_mode="fast"`, `status="processing"`, and enqueues the orchestrator
     task. The caller (route) returns the row immediately.
+
+    `is_judge_run` is the v2 native-judge marker, persisted on the run before
+    dispatch so the aggregate (which only knows eval_run_id) reads it at judge
+    time. It defaults to the v1 behavior — no judging, Langfuse sync as today —
+    so the v1 call path is unchanged. Judging is system-config only: the judge
+    always uses the fallback model + built-in prompt, so there is no per-run config.
     """
     logger.info(
         f"[validate_and_start_fast_evaluation] Starting fast eval | "
@@ -76,7 +174,7 @@ def validate_and_start_fast_evaluation(
         f"org_id={organization_id} | project_id={project_id}"
     )
 
-    # 1. Dataset must exist + have a Langfuse id (same as batch path).
+    # 1. Dataset must exist (Langfuse id required for v1 runs only; see below).
     dataset = get_dataset_by_id(
         session=session,
         dataset_id=dataset_id,
@@ -91,7 +189,9 @@ def validate_and_start_fast_evaluation(
                 "organization/project"
             ),
         )
-    if not dataset.langfuse_dataset_id:
+    # v1 runs still require a Langfuse-backed dataset. v2 judged runs are
+    # Langfuse-free and load items from S3, so a NULL langfuse id is allowed there.
+    if not dataset.langfuse_dataset_id and not is_judge_run:
         raise HTTPException(
             status_code=400,
             detail=(
@@ -158,17 +258,32 @@ def validate_and_start_fast_evaluation(
         log_context="validate_and_start_fast_evaluation",
     )
 
+    # Persist the judge marker before dispatch so the post-barrier aggregate reads
+    # it. Skipped for v1 runs (is_judge_run=False), keeping that path unchanged.
+    if is_judge_run:
+        eval_run = update_evaluation_run(
+            session=session,
+            eval_run=eval_run,
+            update=EvaluationRunUpdate(is_judge_run=True),
+        )
+
     # Fetch the dataset items now to size the fan-out: ceil(total / chunk_size)
     # parallel chunk tasks drain the responses stage across workers. Any failure
     # here marks the run failed so it never lingers in `processing`.
     try:
-        langfuse_client = get_langfuse_client(
-            session=session,
-            org_id=organization_id,
-            project_id=project_id,
+        # Only Langfuse-backed (v1) datasets need a client; a v2 dataset loads from
+        # S3, so we skip the client rather than require Langfuse for a native run.
+        langfuse_client = (
+            get_langfuse_client(
+                session=session,
+                org_id=organization_id,
+                project_id=project_id,
+            )
+            if dataset.langfuse_dataset_id
+            else None
         )
-        dataset_items = fetch_dataset_items(
-            langfuse=langfuse_client, dataset_name=eval_run.dataset_name
+        dataset_items = load_run_dataset_items(
+            session=session, dataset=dataset, langfuse=langfuse_client
         )
         total_items = len(dataset_items)
         if total_items == 0:
@@ -231,12 +346,13 @@ def _get_fast_run(*, session: Session, eval_run_id: int) -> EvaluationRun:
 
 
 def _resolve_config_and_clients(
-    *, session: Session, eval_run: EvaluationRun
-) -> tuple[TextLLMParams, OpenAI, Langfuse]:
-    """Resolve the run's text config and build its OpenAI + Langfuse clients.
+    *, session: Session, eval_run: EvaluationRun, dataset: EvaluationDataset
+) -> tuple[TextLLMParams, OpenAI, Langfuse | None]:
+    """Resolve the run's text config and build its OpenAI + (optional) Langfuse clients.
 
-    The Langfuse client is None when the project opted out of tracing (#996) so
-    the run degrades to cosine-only instead of failing."""
+    Only a Langfuse-backed (v1) dataset needs a Langfuse client — its items live in
+    Langfuse. A v2 dataset loads from S3, so we skip the client (and its credential
+    requirement) rather than fail a Langfuse-free run. Mirrors the fan-out sizing."""
     config_blob, error = resolve_evaluation_config(
         session=session,
         config_id=eval_run.config_id,
@@ -252,10 +368,14 @@ def _resolve_config_and_clients(
         org_id=eval_run.organization_id,
         project_id=eval_run.project_id,
     )
-    langfuse_client = get_langfuse_client(
-        session=session,
-        org_id=eval_run.organization_id,
-        project_id=eval_run.project_id,
+    langfuse_client = (
+        get_langfuse_client(
+            session=session,
+            org_id=eval_run.organization_id,
+            project_id=eval_run.project_id,
+        )
+        if dataset.langfuse_dataset_id
+        else None
     )
     return text_params, openai_client, langfuse_client
 
@@ -283,11 +403,21 @@ def execute_fast_evaluation_chunk(*, eval_run_id: int, chunk_index: int) -> None
             return
 
         try:
+            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 dataset is None:
+                raise ValueError(
+                    f"Dataset {eval_run.dataset_id} not found for run {eval_run_id}"
+                )
             text_params, openai_client, langfuse_client = _resolve_config_and_clients(
-                session=session, eval_run=eval_run
+                session=session, eval_run=eval_run, dataset=dataset
             )
-            dataset_items = fetch_dataset_items(
-                langfuse=langfuse_client, dataset_name=eval_run.dataset_name
+            dataset_items = load_run_dataset_items(
+                session=session, dataset=dataset, langfuse=langfuse_client
             )
             # Same order across every chunk task, so slices never overlap or miss.
             dataset_items.sort(key=lambda item: item["id"])
@@ -350,10 +480,16 @@ def execute_fast_evaluation_aggregate(*, eval_run_id: int) -> None:
                 org_id=eval_run.organization_id,
                 project_id=eval_run.project_id,
             )
-            langfuse_client = get_langfuse_client(
-                session=session,
-                org_id=eval_run.organization_id,
-                project_id=eval_run.project_id,
+            # v2 judged runs are fully Kaapi-native: no Langfuse client, so no
+            # traces are created and no scores are synced. v1 keeps syncing.
+            langfuse_client = (
+                None
+                if eval_run.is_judge_run
+                else get_langfuse_client(
+                    session=session,
+                    org_id=eval_run.organization_id,
+                    project_id=eval_run.project_id,
+                )
             )
             run_fast_evaluation(
                 session=session,
diff --git a/backend/app/services/evaluations/judge.py b/backend/app/services/evaluations/judge.py
new file mode 100644
index 000000000..815982527
--- /dev/null
+++ b/backend/app/services/evaluations/judge.py
@@ -0,0 +1,43 @@
+"""Starts a v2 native LLM-as-judge run using the shared fast-eval pipeline.
+Judging runs in the aggregate with the built-in prompt and fallback model; v1 remains unchanged.
+"""
+
+import logging
+from uuid import UUID
+
+from sqlmodel import Session
+
+from app.models.evaluation import EvaluationRun
+from app.services.evaluations.fast import validate_and_start_fast_evaluation
+
+logger = logging.getLogger(__name__)
+
+
+def validate_and_start_judged_evaluation(
+    *,
+    session: Session,
+    dataset_id: int,
+    run_name: str,
+    config_id: UUID,
+    config_version: int,
+    organization_id: int,
+    project_id: int,
+    trace_id: str = "N/A",
+) -> EvaluationRun:
+    """Start a v2 judged fast evaluation run.
+
+    Delegates dataset/config validation, run creation, and chunk dispatch to the
+    shared v1 fast trigger with the native-judge marker set. Judging always runs
+    for v2 fast runs — there is no opt-in flag and no per-run judge config.
+    """
+    return validate_and_start_fast_evaluation(
+        session=session,
+        dataset_id=dataset_id,
+        run_name=run_name,
+        config_id=config_id,
+        config_version=config_version,
+        organization_id=organization_id,
+        project_id=project_id,
+        trace_id=trace_id,
+        is_judge_run=True,
+    )
diff --git a/backend/app/tests/api/routes/test_evaluation_dataset_v2.py b/backend/app/tests/api/routes/test_evaluation_dataset_v2.py
new file mode 100644
index 000000000..54ab0e643
--- /dev/null
+++ b/backend/app/tests/api/routes/test_evaluation_dataset_v2.py
@@ -0,0 +1,174 @@
+"""Tests for the v2 Langfuse-free dataset upload route.
+
+`POST {API_V2_STR}/evaluations/datasets` — the three-metric SRD's Langfuse-free
+upload (docs/srd-three-metric-evaluation-verdict.md, FR-19/FR-20):
+
+- FR-19: 200, row created with null langfuse id, CSV stored in S3, Langfuse never
+  called.
+- FR-20: response + persisted metadata carry the run-time-duplication marker and
+  original/total item counts.
+
+Object storage and Langfuse are the external boundaries and are mocked; the
+dataset row lands in the real (transactional) DB.
+"""
+
+import io
+from unittest.mock import MagicMock, patch
+
+from fastapi.testclient import TestClient
+from sqlmodel import Session
+
+from app.core.config import settings
+from app.crud.evaluations.dataset import (
+    DATASET_META_DUPLICATE_AT_RUNTIME,
+    DATASET_META_DUPLICATION_FACTOR,
+    DATASET_META_ORIGINAL_ITEMS,
+    DATASET_META_TOTAL_ITEMS,
+)
+from app.models import EvaluationDataset
+from app.tests.utils.utils import random_lower_string
+
+V2_DATASETS = f"{settings.API_V2_STR}/evaluations/datasets"
+_DATASET = "app.services.evaluations.dataset"
+
+_CSV_TEXT = "question,answer\nQ0?,A0\nQ1?,A1\nQ2?,A2\n"  # 3 original rows
+
+
+def _csv_upload() -> tuple[str, io.BytesIO, str]:
+    return ("dataset.csv", io.BytesIO(_CSV_TEXT.encode("utf-8")), "text/csv")
+
+
+class TestUploadDatasetV2Route:
+    def test_upload_creates_langfuse_free_dataset(
+        self,
+        client: TestClient,
+        user_api_key_header: dict[str, str],
+        db: Session,
+    ) -> None:
+        """FR-19/FR-20: 200, null langfuse id, S3 stored, run-time-dup metadata."""
+        name = f"v2-route-{random_lower_string()}"
+        with (
+            patch(f"{_DATASET}.get_cloud_storage", return_value=MagicMock()),
+            patch(
+                f"{_DATASET}.upload_csv_to_object_store",
+                return_value="s3://bucket/datasets/v2.csv",
+            ) as mock_upload,
+            patch(f"{_DATASET}.get_langfuse_client") as mock_langfuse_client,
+            patch(f"{_DATASET}.upload_dataset_to_langfuse") as mock_langfuse_upload,
+        ):
+            resp = client.post(
+                V2_DATASETS,
+                files={"file": _csv_upload()},
+                data={"dataset_name": name, "duplication_factor": 5},
+                headers=user_api_key_header,
+            )
+
+        assert resp.status_code == 200, resp.text
+        body = resp.json()["data"]
+        assert body["langfuse_dataset_id"] is None
+        assert body["original_items"] == 3
+        assert body["total_items"] == 15  # 3 rows × factor 5
+        assert body["duplication_factor"] == 5
+        assert body["object_store_url"] == "s3://bucket/datasets/v2.csv"
+
+        mock_upload.assert_called_once()
+        mock_langfuse_client.assert_not_called()
+        mock_langfuse_upload.assert_not_called()
+
+        persisted = db.get(EvaluationDataset, body["dataset_id"])
+        assert persisted is not None
+        assert persisted.langfuse_dataset_id is None
+        meta = persisted.dataset_metadata
+        assert meta[DATASET_META_DUPLICATE_AT_RUNTIME] is True
+        assert meta[DATASET_META_DUPLICATION_FACTOR] == 5
+        assert meta[DATASET_META_ORIGINAL_ITEMS] == 3
+        assert meta[DATASET_META_TOTAL_ITEMS] == 15
+
+    def test_upload_defaults_duplication_factor_to_one(
+        self,
+        client: TestClient,
+        user_api_key_header: dict[str, str],
+        db: Session,
+    ) -> None:
+        """An omitted duplication_factor defaults to 1 — total equals original."""
+        name = f"v2-default-{random_lower_string()}"
+        with (
+            patch(f"{_DATASET}.get_cloud_storage", return_value=MagicMock()),
+            patch(
+                f"{_DATASET}.upload_csv_to_object_store",
+                return_value="s3://bucket/datasets/v2.csv",
+            ),
+            patch(f"{_DATASET}.get_langfuse_client"),
+            patch(f"{_DATASET}.upload_dataset_to_langfuse"),
+        ):
+            resp = client.post(
+                V2_DATASETS,
+                files={"file": _csv_upload()},
+                data={"dataset_name": name},
+                headers=user_api_key_header,
+            )
+
+        assert resp.status_code == 200, resp.text
+        body = resp.json()["data"]
+        assert body["duplication_factor"] == 1
+        assert body["original_items"] == 3
+        assert body["total_items"] == 3
+
+    def test_object_store_no_url_returns_500(
+        self,
+        client: TestClient,
+        user_api_key_header: dict[str, str],
+    ) -> None:
+        """FR-19: with no Langfuse fallback, a failed S3 store is a 500."""
+        with (
+            patch(f"{_DATASET}.get_cloud_storage", return_value=MagicMock()),
+            patch(f"{_DATASET}.upload_csv_to_object_store", return_value=None),
+            patch(f"{_DATASET}.get_langfuse_client"),
+        ):
+            resp = client.post(
+                V2_DATASETS,
+                files={"file": _csv_upload()},
+                data={
+                    "dataset_name": f"v2-nourl-{random_lower_string()}",
+                    "duplication_factor": 2,
+                },
+                headers=user_api_key_header,
+            )
+
+        assert resp.status_code == 500
+
+    def test_duplication_factor_above_max_rejected(
+        self,
+        client: TestClient,
+        user_api_key_header: dict[str, str],
+    ) -> None:
+        """The route caps duplication_factor at 5 (ge=1, le=5)."""
+        resp = client.post(
+            V2_DATASETS,
+            files={"file": _csv_upload()},
+            data={
+                "dataset_name": f"v2-toobig-{random_lower_string()}",
+                "duplication_factor": 6,
+            },
+            headers=user_api_key_header,
+        )
+
+        assert resp.status_code == 422
+
+    def test_duplication_factor_below_min_rejected(
+        self,
+        client: TestClient,
+        user_api_key_header: dict[str, str],
+    ) -> None:
+        """duplication_factor below 1 is rejected before any upload."""
+        resp = client.post(
+            V2_DATASETS,
+            files={"file": _csv_upload()},
+            data={
+                "dataset_name": f"v2-toosmall-{random_lower_string()}",
+                "duplication_factor": 0,
+            },
+            headers=user_api_key_header,
+        )
+
+        assert resp.status_code == 422
diff --git a/backend/app/tests/api/routes/test_evaluation_v2.py b/backend/app/tests/api/routes/test_evaluation_v2.py
new file mode 100644
index 000000000..383df2b06
--- /dev/null
+++ b/backend/app/tests/api/routes/test_evaluation_v2.py
@@ -0,0 +1,178 @@
+"""Tests for the v2 judged evaluation run trigger (`POST /api/v2/evaluations`).
+
+Covers the ground-truth slice of the three-metric SRD at the route boundary:
+a v2 run is always fast and always judged (FR-9 no-flag; there is no run_mode —
+batch is deferred), and the v1 trigger stays cosine-only (FR-18). Judging is
+system-config only — there is no per-run judge_config in the v2 body. External
+dispatch boundaries (Langfuse, dataset fetch, chunk enqueue) are mocked; the DB is
+real.
+"""
+
+from collections.abc import Iterator
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlmodel import Session
+
+from app.core.config import settings
+from app.models import Config, EvaluationDataset, EvaluationRun
+from app.models.llm.request import ConfigBlob, KaapiCompletionConfig
+from app.tests.utils.auth import TestAuthContext
+from app.tests.utils.test_data import (
+    create_test_config,
+    create_test_evaluation_dataset,
+)
+from app.tests.utils.utils import random_lower_string
+
+V2_EVALS = f"{settings.API_V2_STR}/evaluations"
+
+
+def _make_dataset(*, db: Session, user_api_key: TestAuthContext) -> EvaluationDataset:
+    return create_test_evaluation_dataset(
+        db=db,
+        organization_id=user_api_key.organization_id,
+        project_id=user_api_key.project_id,
+        original_items_count=3,
+        duplication_factor=1,
+    )
+
+
+def _make_text_config(db: Session, project_id: int) -> Config:
+    blob = ConfigBlob(
+        completion=KaapiCompletionConfig(
+            provider="openai",
+            type="text",
+            params={"model": "gpt-4o-fast-eval-test", "temperature": 0.7},
+        )
+    )
+    return create_test_config(
+        db=db, project_id=project_id, use_kaapi_schema=True, config_blob=blob
+    )
+
+
+def _dataset_item(item_id: str) -> dict[str, Any]:
+    return {
+        "id": item_id,
+        "input": {"question": "Q"},
+        "expected_output": {"answer": "A"},
+        "metadata": {"question_id": item_id},
+    }
+
+
+@pytest.fixture
+def _patch_dispatch() -> Iterator[MagicMock]:
+    """Stub the synchronous request-path boundaries of validate_and_start.
+
+    Same boundaries the v1 fast route stubs — Langfuse client, dataset fetch, and
+    the chunk enqueue — so a v2 run reaches dispatch without real I/O. Yields the
+    chunk-enqueue mock (call count == number of dispatched chunks).
+    """
+    with (
+        patch("app.services.evaluations.fast.get_langfuse_client"),
+        patch(
+            "app.services.evaluations.fast.fetch_dataset_items",
+            return_value=[_dataset_item(f"item-{i}") for i in range(3)],
+        ),
+        patch(
+            "app.services.evaluations.fast.start_fast_evaluation_chunk",
+            return_value="fake-task-id",
+        ) as mock_start_chunk,
+    ):
+        yield mock_start_chunk
+
+
+class TestV2JudgedRunTrigger:
+    def test_fast_run_marks_judge_run_and_dispatches(
+        self,
+        client: TestClient,
+        user_api_key_header: dict[str, str],
+        db: Session,
+        user_api_key: TestAuthContext,
+        _patch_dispatch,
+    ):
+        """FR-9: a v2 fast run is always a judged run — no flag, dispatched."""
+        dataset = _make_dataset(db=db, user_api_key=user_api_key)
+        config = _make_text_config(db, user_api_key.project_id)
+
+        resp = client.post(
+            V2_EVALS,
+            json={
+                "experiment_name": "v2-judged",
+                "dataset_id": dataset.id,
+                "config_id": str(config.id),
+                "config_version": 1,
+            },
+            headers=user_api_key_header,
+        )
+
+        assert resp.status_code == 200, resp.text
+        body = resp.json()["data"]
+        assert body["status"] == "processing"
+        assert body["is_judge_run"] is True
+        _patch_dispatch.assert_called_once()
+
+        run = db.get(EvaluationRun, body["id"])
+        assert run is not None
+        assert run.is_judge_run is True
+
+    def test_v2_run_is_always_fast_and_judged(
+        self,
+        client: TestClient,
+        user_api_key_header: dict[str, str],
+        db: Session,
+        user_api_key: TestAuthContext,
+        _patch_dispatch,
+    ):
+        """A v2 run carries no run_mode; it is always created fast and judged."""
+        dataset = _make_dataset(db=db, user_api_key=user_api_key)
+        config = _make_text_config(db, user_api_key.project_id)
+
+        resp = client.post(
+            V2_EVALS,
+            json={
+                "experiment_name": "v2-default-mode",
+                "dataset_id": dataset.id,
+                "config_id": str(config.id),
+                "config_version": 1,
+            },
+            headers=user_api_key_header,
+        )
+
+        assert resp.status_code == 200, resp.text
+        body = resp.json()["data"]
+        assert body["run_mode"] == "fast"
+        assert body["is_judge_run"] is True
+
+
+class TestV1TriggerUnchanged:
+    def test_v1_fast_run_is_not_a_judge_run(
+        self,
+        client: TestClient,
+        user_api_key_header: dict[str, str],
+        db: Session,
+        user_api_key: TestAuthContext,
+        _patch_dispatch,
+    ):
+        """FR-18: the v1 trigger produces a cosine-only run — never a judged one."""
+        dataset = _make_dataset(db=db, user_api_key=user_api_key)
+        config = _make_text_config(db, user_api_key.project_id)
+
+        resp = client.post(
+            f"{settings.API_V1_STR}/evaluations",
+            json={
+                "experiment_name": f"v1-plain-{random_lower_string()}",
+                "dataset_id": dataset.id,
+                "config_id": str(config.id),
+                "config_version": 1,
+                "run_mode": "fast",
+            },
+            headers=user_api_key_header,
+        )
+
+        assert resp.status_code == 200, resp.text
+        body = resp.json()["data"]
+        run = db.get(EvaluationRun, body["id"])
+        assert not run.is_judge_run
+        assert run.per_item_ground_truth is None
diff --git a/backend/app/tests/crud/evaluations/test_fast_judge.py b/backend/app/tests/crud/evaluations/test_fast_judge.py
new file mode 100644
index 000000000..0e3a2ff82
--- /dev/null
+++ b/backend/app/tests/crud/evaluations/test_fast_judge.py
@@ -0,0 +1,461 @@
+"""End-to-end judge scoring on the v2 fast pipeline (`run_fast_evaluation`).
+
+Drives the ground-truth slice of the three-metric SRD through the real fast
+pipeline with a judged run (`is_judge_run=True`, `langfuse=None` as v2 dispatches):
+FR-2 (a ground-truth trace score in [0,1] + reasoning), FR-9 (zero-config uses
+the fallback model + built-in prompt), FR-14 (summary + durable per-row map),
+FR-15 (per-row isolation), FR-16 (judge cost stage), FR-18 (v1 never judges).
+
+A v2 judged run drops cosine + embeddings entirely: no embedding API calls, no
+"Cosine Similarity" trace/summary score, and `per_item_scores` stays NULL. The
+"Adherence to Ground Truth" judge score is the only scorer. v1 (`is_judge_run`
+False) is unchanged — cosine only, no judge.
+
+External boundaries mocked: OpenAI (embeddings + the judge completion at
+`_create_judge_response`), S3, `save_score`, model/cost resolution. DB is real.
+"""
+
+import json
+from collections.abc import Iterator
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+from sqlmodel import Session
+
+from app.core.config import settings
+from app.crud.evaluations.fast import (
+    CHUNK_CONFIG_INDEX,
+    CHUNK_CONFIG_RUN_ID,
+    JOB_TYPE_EVALUATION_FAST_CHUNK,
+    run_fast_evaluation,
+)
+from app.crud.evaluations.score import GROUND_TRUTH_SCORE_NAME, JUDGE_FAILED_REASON
+from app.models import Config, EvaluationDataset, EvaluationRun
+from app.models.batch_job import BatchJob
+from app.models.evaluation import RunModeEnum
+from app.models.llm.request import ConfigBlob, KaapiCompletionConfig
+from app.tests.utils.auth import TestAuthContext
+from app.tests.utils.test_data import (
+    create_test_config,
+    create_test_evaluation_dataset,
+)
+from app.tests.utils.utils import random_lower_string
+
+COSINE_SCORE_NAME = "Cosine Similarity"
+
+
+def _make_dataset(*, db: Session, user_api_key: TestAuthContext) -> EvaluationDataset:
+    return create_test_evaluation_dataset(
+        db=db,
+        organization_id=user_api_key.organization_id,
+        project_id=user_api_key.project_id,
+        original_items_count=3,
+        duplication_factor=1,
+    )
+
+
+def _make_text_config(db: Session, project_id: int) -> Config:
+    blob = ConfigBlob(
+        completion=KaapiCompletionConfig(
+            provider="openai",
+            type="text",
+            params={"model": "gpt-4o-fast-eval-test", "temperature": 0.7},
+        )
+    )
+    return create_test_config(
+        db=db, project_id=project_id, use_kaapi_schema=True, config_blob=blob
+    )
+
+
+def _make_run(
+    *,
+    db: Session,
+    user_api_key: TestAuthContext,
+    is_judge_run: bool,
+) -> EvaluationRun:
+    dataset = _make_dataset(db=db, user_api_key=user_api_key)
+    config = _make_text_config(db, user_api_key.project_id)
+    run = EvaluationRun(
+        run_name=f"run-{random_lower_string()}",
+        dataset_name=dataset.name,
+        dataset_id=dataset.id,
+        config_id=config.id,
+        config_version=1,
+        status="pending",
+        run_mode=RunModeEnum.FAST.value,
+        total_items=0,
+        is_judge_run=is_judge_run or None,
+        organization_id=user_api_key.organization_id,
+        project_id=user_api_key.project_id,
+    )
+    db.add(run)
+    db.commit()
+    db.refresh(run)
+    return run
+
+
+def _resp_result(
+    item_id: str, question: str, ground_truth: str = "golden"
+) -> dict[str, Any]:
+    return {
+        "item_id": item_id,
+        "question": question,
+        "generated_output": f"generated for {question}",
+        "ground_truth": ground_truth,
+        "response_id": f"resp_{item_id}",
+        "usage": {"input_tokens": 5, "output_tokens": 5, "total_tokens": 10},
+        "question_id": item_id,
+        "failed": False,
+    }
+
+
+def _fake_embedding_response():
+    """Two identical unit vectors → cosine ≈ 1.0."""
+    return SimpleNamespace(
+        data=[
+            SimpleNamespace(index=0, embedding=[1.0, 0.0, 0.0]),
+            SimpleNamespace(index=1, embedding=[1.0, 0.0, 0.0]),
+        ],
+        usage=SimpleNamespace(prompt_tokens=5, total_tokens=5),
+    )
+
+
+def _judge_response(score: float, reasoning: str, *, usage=(12, 6, 18)):
+    body = json.dumps({"ground_truth": {"score": score, "reasoning": reasoning}})
+    return _raw_judge_response(body, usage=usage)
+
+
+def _raw_judge_response(text: str, *, usage=(12, 6, 18)):
+    i, o, t = usage
+    return SimpleNamespace(
+        output_text=text,
+        output=[],
+        usage=SimpleNamespace(input_tokens=i, output_tokens=o, total_tokens=t),
+    )
+
+
+@pytest.fixture
+def _s3_store() -> Iterator[dict[str, list[dict[str, Any]]]]:
+    store: dict[str, list[dict[str, Any]]] = {}
+
+    def _upload(*, filename, results, **_):
+        url = f"s3://bucket/{filename}"
+        store[url] = list(results)
+        return url
+
+    def _load(*, url, **_):
+        return store[url]
+
+    with (
+        patch("app.crud.evaluations.fast._upload_unit_to_s3", side_effect=_upload),
+        patch("app.crud.evaluations.fast._load_unit_from_s3", side_effect=_load),
+    ):
+        yield store
+
+
+def _seed_chunk(
+    *,
+    db: Session,
+    eval_run: EvaluationRun,
+    results: list[dict[str, Any]],
+    store: dict[str, list[dict[str, Any]]],
+) -> None:
+    url = f"s3://bucket/responses_{eval_run.id}_0.json"
+    store[url] = results
+    job = BatchJob(
+        provider="openai",
+        job_type=JOB_TYPE_EVALUATION_FAST_CHUNK,
+        config={
+            "model": "gpt-4o",
+            CHUNK_CONFIG_RUN_ID: eval_run.id,
+            CHUNK_CONFIG_INDEX: 0,
+        },
+        raw_output_url=url,
+        total_items=len(results),
+        organization_id=eval_run.organization_id,
+        project_id=eval_run.project_id,
+    )
+    db.add(job)
+    db.commit()
+
+
+def _persist_score_into(db: Session):
+    def _fake_save_score(*, eval_run_id, score, **_):
+        run = db.get(EvaluationRun, eval_run_id)
+        run.score = {"summary_scores": score["summary_scores"]}
+        run.score_trace_url = f"s3://bucket/traces_{eval_run_id}.json"
+        db.add(run)
+        db.commit()
+        db.refresh(run)
+        return run
+
+    return _fake_save_score
+
+
+def _run_pipeline(
+    *,
+    db: Session,
+    eval_run: EvaluationRun,
+    judge_side_effect,
+    mock_cost: bool = False,
+) -> tuple[EvaluationRun, MagicMock]:
+    """Run `run_fast_evaluation` for a judged run with all externals stubbed.
+
+    `langfuse=None` mirrors what the v2 aggregate passes for a judged run, so refs
+    key by item_id. The judge completion is driven by `judge_side_effect(params)`.
+    Returns the run plus the OpenAI mock so callers can assert the embedding path
+    was (v1) or was not (v2 judge) exercised.
+    """
+    fake_openai = MagicMock()
+    fake_openai.embeddings.create.return_value = _fake_embedding_response()
+
+    def _judge(_client, params):
+        return judge_side_effect(params)
+
+    ctx = [
+        patch(
+            "app.crud.evaluations.fast.resolve_model_from_config", return_value="gpt-4o"
+        ),
+        patch(
+            "app.crud.evaluations.fast.save_score", side_effect=_persist_score_into(db)
+        ),
+        patch("app.crud.evaluations.judge._create_judge_response", side_effect=_judge),
+    ]
+    if mock_cost:
+        ctx.append(
+            patch(
+                "app.crud.evaluations.cost.estimate_model_cost",
+                return_value={"input_cost": 0.001, "output_cost": 0.002},
+            )
+        )
+
+    with ctx[0], ctx[1], ctx[2]:
+        if mock_cost:
+            with ctx[3]:
+                result = run_fast_evaluation(
+                    session=db,
+                    openai_client=fake_openai,
+                    langfuse=None,
+                    eval_run=eval_run,
+                )
+        else:
+            result = run_fast_evaluation(
+                session=db, openai_client=fake_openai, langfuse=None, eval_run=eval_run
+            )
+    return result, fake_openai
+
+
+def _trace_by_ref(result: EvaluationRun) -> dict[str, dict[str, Any]]:
+    return {t["trace_id"]: t for t in result.score["traces"]}
+
+
+def _score_named(trace: dict[str, Any], name: str) -> dict[str, Any] | None:
+    for s in trace["scores"]:
+        if s["name"] == name:
+            return s
+    return None
+
+
+class TestGroundTruthScoring:
+    def test_ground_truth_score_is_the_only_scorer_no_cosine(
+        self, db: Session, user_api_key: TestAuthContext, _s3_store
+    ):
+        """FR-2/FR-14 + v2 contract: each row carries ONLY a ground-truth score (no
+        cosine); summary + durable per-row map populated; per_item_scores stays NULL
+        and no embeddings are computed for a judged run."""
+        eval_run = _make_run(db=db, user_api_key=user_api_key, is_judge_run=True)
+        _seed_chunk(
+            db=db,
+            eval_run=eval_run,
+            results=[
+                _resp_result("item-1", "Q1", "golden-1"),
+                _resp_result("item-2", "Q2", "golden-2"),
+            ],
+            store=_s3_store,
+        )
+
+        result, fake_openai = _run_pipeline(
+            db=db,
+            eval_run=eval_run,
+            judge_side_effect=lambda _p: _judge_response(0.8, "conveys the same facts"),
+        )
+
+        assert result.status == "completed"
+        # A judged run never embeds — cosine is gone entirely.
+        fake_openai.embeddings.create.assert_not_called()
+
+        traces = _trace_by_ref(result)
+        assert set(traces) == {"item-1", "item-2"}
+        for ref in ("item-1", "item-2"):
+            gt = _score_named(traces[ref], GROUND_TRUTH_SCORE_NAME)
+            assert _score_named(traces[ref], COSINE_SCORE_NAME) is None
+            assert gt is not None
+            assert 0.0 <= gt["value"] <= 1.0
+            assert gt["value"] == pytest.approx(0.8, abs=0.01)
+            assert gt["comment"] == "conveys the same facts"
+
+        summary_names = {s["name"] for s in result.score["summary_scores"]}
+        assert GROUND_TRUTH_SCORE_NAME in summary_names
+        assert COSINE_SCORE_NAME not in summary_names
+        gt_summary = next(
+            s
+            for s in result.score["summary_scores"]
+            if s["name"] == GROUND_TRUTH_SCORE_NAME
+        )
+        assert gt_summary["avg"] == pytest.approx(0.8, abs=0.01)
+        assert gt_summary["total_pairs"] == 2
+
+        run = db.get(EvaluationRun, result.id)
+        assert run.per_item_ground_truth == {"item-1": 0.8, "item-2": 0.8}
+        # Cosine's durable per-row map is a v1-only artifact; a judged run leaves it NULL.
+        assert run.per_item_scores is None
+
+    def test_zero_config_judges_with_fallback_model_and_builtin_prompt(
+        self, db: Session, user_api_key: TestAuthContext, _s3_store
+    ):
+        """FR-9: a judged run uses the fallback model and the built-in ground-truth
+        prompt (as its instructions) — judging is system-config only."""
+        eval_run = _make_run(db=db, user_api_key=user_api_key, is_judge_run=True)
+        _seed_chunk(
+            db=db,
+            eval_run=eval_run,
+            results=[_resp_result("item-1", "Q1", "golden-1")],
+            store=_s3_store,
+        )
+
+        captured: dict = {}
+
+        def _capture(params):
+            captured.update(params)
+            return _judge_response(0.6, "partially correct")
+
+        result, _ = _run_pipeline(db=db, eval_run=eval_run, judge_side_effect=_capture)
+
+        assert result.status == "completed"
+        assert captured["model"] == settings.EVAL_JUDGE_MODEL
+        # The judge is a reasoning model that rejects a custom temperature.
+        assert "temperature" not in captured
+        # The built-in ground-truth rubric drives the grade (as the call instructions).
+        assert "Adherence to Ground Truth" in captured["instructions"]
+        # The golden answer reaches the judge (FR-3).
+        assert "golden-1" in captured["input"]
+
+    def test_per_row_isolation_leaves_failed_row_unscoreable(
+        self, db: Session, user_api_key: TestAuthContext, _s3_store
+    ):
+        """FR-15: a malformed judge reply for one row flags that row judge_failed and
+        drops its ground-truth score, while its siblings and the run survive. Neither
+        row carries a cosine score (v2 never embeds)."""
+        eval_run = _make_run(db=db, user_api_key=user_api_key, is_judge_run=True)
+        _seed_chunk(
+            db=db,
+            eval_run=eval_run,
+            results=[
+                _resp_result("item-good", "Q-good", "golden-good"),
+                _resp_result("item-bad", "Q-bad", "golden-bad"),
+            ],
+            store=_s3_store,
+        )
+
+        def _judge(params):
+            if "Q-bad" in params["input"]:
+                return _raw_judge_response("totally not json")
+            return _judge_response(0.9, "correct")
+
+        result, fake_openai = _run_pipeline(
+            db=db, eval_run=eval_run, judge_side_effect=_judge
+        )
+
+        assert result.status == "completed"
+        fake_openai.embeddings.create.assert_not_called()
+        traces = _trace_by_ref(result)
+
+        # Good row: ground truth only, no cosine.
+        assert _score_named(traces["item-good"], GROUND_TRUTH_SCORE_NAME) is not None
+        assert _score_named(traces["item-good"], COSINE_SCORE_NAME) is None
+
+        # Bad row: ground truth dropped, and still no cosine placeholder for a judged run.
+        assert _score_named(traces["item-bad"], COSINE_SCORE_NAME) is None
+        assert _score_named(traces["item-bad"], GROUND_TRUTH_SCORE_NAME) is None
+
+        run = db.get(EvaluationRun, result.id)
+        assert run.unscoreable.get("item-bad") == JUDGE_FAILED_REASON
+        assert set(run.per_item_ground_truth) == {"item-good"}
+        assert run.per_item_scores is None
+
+    def test_judge_cost_stage_tracked(
+        self, db: Session, user_api_key: TestAuthContext, _s3_store
+    ):
+        """FR-16: EvaluationRun.cost gains a ground_truth_judge stage with tokens + USD."""
+        eval_run = _make_run(db=db, user_api_key=user_api_key, is_judge_run=True)
+        _seed_chunk(
+            db=db,
+            eval_run=eval_run,
+            results=[
+                _resp_result("item-1", "Q1", "golden-1"),
+                _resp_result("item-2", "Q2", "golden-2"),
+            ],
+            store=_s3_store,
+        )
+
+        result, _ = _run_pipeline(
+            db=db,
+            eval_run=eval_run,
+            judge_side_effect=lambda _p: _judge_response(
+                0.7, "close", usage=(12, 6, 18)
+            ),
+            mock_cost=True,
+        )
+
+        run = db.get(EvaluationRun, result.id)
+        stage = run.cost["ground_truth_judge"]
+        # Two rows × (12 in, 6 out, 18 total) summed for the combined judge call.
+        assert stage["input_tokens"] == 24
+        assert stage["output_tokens"] == 12
+        assert stage["total_tokens"] == 36
+        # cost_usd from the mocked estimate: (0.001 + 0.002) rounded.
+        assert stage["cost_usd"] == pytest.approx(0.003, abs=1e-9)
+
+
+class TestV1PipelineUnchanged:
+    def test_v1_run_produces_no_judge_metrics(
+        self, db: Session, user_api_key: TestAuthContext, _s3_store
+    ):
+        """FR-18: a non-judge run scores cosine only — no ground-truth score, no
+        per-row ground-truth map, and the judge completion is never called. Embeddings
+        run and per_item_scores is populated exactly as before."""
+        eval_run = _make_run(db=db, user_api_key=user_api_key, is_judge_run=False)
+        _seed_chunk(
+            db=db,
+            eval_run=eval_run,
+            results=[_resp_result("item-1", "Q1", "golden-1")],
+            store=_s3_store,
+        )
+
+        judge_calls: list = []
+
+        def _judge(params):
+            judge_calls.append(params)
+            return _judge_response(0.9, "should never run")
+
+        result, fake_openai = _run_pipeline(
+            db=db, eval_run=eval_run, judge_side_effect=_judge
+        )
+
+        assert result.status == "completed"
+        assert judge_calls == []
+        # v1 embeds every row (cosine input); the pair is identical → cosine ≈ 1.0.
+        fake_openai.embeddings.create.assert_called()
+        traces = _trace_by_ref(result)
+        assert _score_named(traces["item-1"], GROUND_TRUTH_SCORE_NAME) is None
+        assert _score_named(traces["item-1"], COSINE_SCORE_NAME) is not None
+
+        run = db.get(EvaluationRun, result.id)
+        assert run.per_item_ground_truth is None
+        # v1 keeps its durable cosine per-row map.
+        assert run.per_item_scores == {"item-1": pytest.approx(1.0, abs=0.01)}
+        summary_names = {s["name"] for s in result.score["summary_scores"]}
+        assert GROUND_TRUTH_SCORE_NAME not in summary_names
+        assert COSINE_SCORE_NAME in summary_names
diff --git a/backend/app/tests/crud/evaluations/test_judge.py b/backend/app/tests/crud/evaluations/test_judge.py
new file mode 100644
index 000000000..95bff9046
--- /dev/null
+++ b/backend/app/tests/crud/evaluations/test_judge.py
@@ -0,0 +1,261 @@
+"""Unit tests for the native LLM-as-judge scoring primitives (`crud/evaluations/judge.py`).
+
+Covers the ground-truth judge slice of the three-metric SRD:
+FR-2 (score in [0,1] + reasoning), FR-3 (ground truth judged against the golden
+answer), FR-9 (zero-config default prompt), FR-15 (malformed → raises so the row
+isolates). The single external boundary — the OpenAI judge completion — is mocked
+at `_create_judge_response`; parsing/prompt-composition helpers are pure.
+"""
+
+import json
+from dataclasses import replace
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import openai
+import pytest
+from sqlmodel import Session
+
+from app.core.config import settings
+from app.crud.evaluations.judge import (
+    JudgeInputEnum,
+    JudgeMetricEnum,
+    MetricScore,
+    _compose_judge_input,
+    _parse_judge_output,
+    build_judge_params,
+    enabled_metric_specs,
+    judge_row,
+)
+from app.crud.evaluations.score import (
+    GROUND_TRUTH_JUDGE_PROMPT,
+    GROUND_TRUTH_SCORE_NAME,
+    JUDGE_SYSTEM_PREAMBLE,
+)
+
+
+def _judge_response(payload: dict | str, *, usage=(15, 8, 23)):
+    """Mimic the OpenAI Responses SDK object the judge parses (`output_text` + usage)."""
+    text = payload if isinstance(payload, str) else json.dumps(payload)
+    input_tokens, output_tokens, total_tokens = usage
+    return SimpleNamespace(
+        output_text=text,
+        output=[],
+        usage=SimpleNamespace(
+            input_tokens=input_tokens,
+            output_tokens=output_tokens,
+            total_tokens=total_tokens,
+        ),
+    )
+
+
+class TestParseJudgeOutput:
+    """FR-2 / FR-15: well-formed replies parse; malformed ones raise."""
+
+    def test_parses_well_formed_ground_truth(self) -> None:
+        specs = enabled_metric_specs()
+        result = _parse_judge_output(
+            json.dumps(
+                {"ground_truth": {"score": 0.75, "reasoning": "close paraphrase"}}
+            ),
+            specs,
+        )
+        assert set(result) == {JudgeMetricEnum.GROUND_TRUTH}
+        score = result[JudgeMetricEnum.GROUND_TRUTH]
+        assert score.score == 0.75
+        assert score.reasoning == "close paraphrase"
+
+    def test_extracts_json_from_prose_wrapper(self) -> None:
+        specs = enabled_metric_specs()
+        wrapped = (
+            'Here is my grade:\n{"ground_truth": {"score": 0.4, "reasoning": '
+            '"missing a key fact"}}\nThanks.'
+        )
+        result = _parse_judge_output(wrapped, specs)
+        assert result[JudgeMetricEnum.GROUND_TRUTH].score == 0.4
+
+    def test_drops_only_the_missing_metric_when_others_present(self) -> None:
+        # Grade against two specs (ground_truth + a synthetic sibling); a reply that
+        # scores only ground_truth must drop the sibling from the map, not raise.
+        gt_spec = enabled_metric_specs()[0]
+        sibling_key = SimpleNamespace(value="sibling_metric")
+        sibling = replace(gt_spec, key=sibling_key)
+
+        result = _parse_judge_output(
+            json.dumps({"ground_truth": {"score": 0.9, "reasoning": "correct"}}),
+            [gt_spec, sibling],
+        )
+        assert list(result) == [JudgeMetricEnum.GROUND_TRUTH]
+
+    def test_raises_when_no_enabled_metric_scored(self) -> None:
+        specs = enabled_metric_specs()
+        with pytest.raises(ValueError, match="scored no enabled metric"):
+            _parse_judge_output(json.dumps({"unrelated": {"score": 0.5}}), specs)
+
+    def test_raises_on_empty_response(self) -> None:
+        with pytest.raises(ValueError, match="empty judge response"):
+            _parse_judge_output("   ", enabled_metric_specs())
+
+    def test_raises_on_non_json(self) -> None:
+        with pytest.raises(ValueError, match="no JSON object"):
+            _parse_judge_output("the answer is basically fine", enabled_metric_specs())
+
+    def test_raises_on_score_out_of_range(self) -> None:
+        with pytest.raises(ValueError, match="out of .0, 1."):
+            _parse_judge_output(
+                json.dumps({"ground_truth": {"score": 1.4, "reasoning": "x"}}),
+                enabled_metric_specs(),
+            )
+
+    def test_raises_on_empty_reasoning(self) -> None:
+        with pytest.raises(ValueError, match="empty 'reasoning'"):
+            _parse_judge_output(
+                json.dumps({"ground_truth": {"score": 0.8, "reasoning": "  "}}),
+                enabled_metric_specs(),
+            )
+
+    def test_raises_on_non_numeric_score(self) -> None:
+        with pytest.raises(ValueError, match="not a number"):
+            _parse_judge_output(
+                json.dumps({"ground_truth": {"score": "high", "reasoning": "x"}}),
+                enabled_metric_specs(),
+            )
+
+
+class TestComposeJudgeInput:
+    """FR-3: the ground-truth judge input carries question + answer + golden answer."""
+
+    def test_input_contains_all_three_ground_truth_inputs(self) -> None:
+        composed = _compose_judge_input(
+            metrics=enabled_metric_specs(),
+            inputs={
+                JudgeInputEnum.QUESTION: "What is the capital of France?",
+                JudgeInputEnum.GENERATED_ANSWER: "Paris is the capital.",
+                JudgeInputEnum.GOLDEN_ANSWER: "Paris",
+            },
+        )
+        assert "What is the capital of France?" in composed
+        assert "Paris is the capital." in composed
+        assert "Golden (reference) answer:\nParis" in composed
+        # The output contract names the metric key the judge must return.
+        assert "ground_truth" in composed
+
+
+class TestBuildJudgeParams:
+    """FR-9: system-config judging uses the fallback model + built-in ground-truth prompt.
+
+    The judge is a reasoning model (gpt-5-mini) that rejects a custom temperature, so
+    the request never carries one.
+    """
+
+    def test_defaults_to_fallback_model_and_builtin_prompt(self, db: Session) -> None:
+        specs = enabled_metric_specs()
+        base_params, system_prompt = build_judge_params(session=db, metrics=specs)
+
+        assert base_params["model"] == settings.EVAL_JUDGE_MODEL
+        assert base_params["model"] == "gpt-5-mini"
+        assert "temperature" not in base_params
+        assert JUDGE_SYSTEM_PREAMBLE in system_prompt
+        assert GROUND_TRUTH_JUDGE_PROMPT in system_prompt
+        # The judge prompt IS the instructions; a bot's own instructions never leak.
+        assert base_params["instructions"] == system_prompt
+
+
+class TestJudgeRow:
+    """FR-3 / FR-15: one combined call per row, returning scores + usage; raises isolate."""
+
+    def _base_params(self, db: Session) -> dict:
+        specs = enabled_metric_specs()
+        base_params, _ = build_judge_params(session=db, metrics=specs)
+        return base_params
+
+    def test_returns_metric_score_and_usage(self, db: Session) -> None:
+        specs = enabled_metric_specs()
+        with patch(
+            "app.crud.evaluations.judge._create_judge_response",
+            return_value=_judge_response(
+                {"ground_truth": {"score": 0.9, "reasoning": "same meaning"}}
+            ),
+        ):
+            result = judge_row(
+                openai_client=SimpleNamespace(),
+                base_params=self._base_params(db),
+                metrics=specs,
+                question="Q",
+                generated_answer="A",
+                golden_answer="A-golden",
+            )
+
+        assert result.metrics[JudgeMetricEnum.GROUND_TRUTH] == MetricScore(
+            score=0.9, reasoning="same meaning"
+        )
+        assert result.usage == {
+            "input_tokens": 15,
+            "output_tokens": 8,
+            "total_tokens": 23,
+        }
+
+    def test_judge_call_receives_question_answer_and_golden(self, db: Session) -> None:
+        """FR-3: the row's golden answer reaches the judge completion input."""
+        captured: dict = {}
+
+        def _capture(_client, params):
+            captured.update(params)
+            return _judge_response(
+                {"ground_truth": {"score": 0.5, "reasoning": "partial"}}
+            )
+
+        with patch(
+            "app.crud.evaluations.judge._create_judge_response", side_effect=_capture
+        ):
+            judge_row(
+                openai_client=SimpleNamespace(),
+                base_params=self._base_params(db),
+                metrics=enabled_metric_specs(),
+                question="Who wrote Hamlet?",
+                generated_answer="Shakespeare wrote it.",
+                golden_answer="William Shakespeare",
+            )
+
+        assert "Who wrote Hamlet?" in captured["input"]
+        assert "Shakespeare wrote it." in captured["input"]
+        assert "William Shakespeare" in captured["input"]
+
+    def test_malformed_output_raises_for_row_isolation(self, db: Session) -> None:
+        with patch(
+            "app.crud.evaluations.judge._create_judge_response",
+            return_value=_judge_response("not json at all"),
+        ):
+            with pytest.raises(ValueError):
+                judge_row(
+                    openai_client=SimpleNamespace(),
+                    base_params=self._base_params(db),
+                    metrics=enabled_metric_specs(),
+                    question="Q",
+                    generated_answer="A",
+                    golden_answer="G",
+                )
+
+    def test_openai_error_propagates_for_row_isolation(self, db: Session) -> None:
+        with patch(
+            "app.crud.evaluations.judge._create_judge_response",
+            side_effect=openai.OpenAIError("judge provider down"),
+        ):
+            with pytest.raises(openai.OpenAIError):
+                judge_row(
+                    openai_client=SimpleNamespace(),
+                    base_params=self._base_params(db),
+                    metrics=enabled_metric_specs(),
+                    question="Q",
+                    generated_answer="A",
+                    golden_answer="G",
+                )
+
+
+def test_ground_truth_is_the_only_enabled_metric() -> None:
+    """Phase 1 ships exactly one metric; guards against silently enabling more."""
+    specs = enabled_metric_specs()
+    assert [s.key for s in specs] == [JudgeMetricEnum.GROUND_TRUTH]
+    assert specs[0].score_name == GROUND_TRUTH_SCORE_NAME
+    assert specs[0].per_item_column == "per_item_ground_truth"
+    assert specs[0].cost_stage == "ground_truth_judge"
diff --git a/backend/app/tests/services/evaluations/test_dataset_v2.py b/backend/app/tests/services/evaluations/test_dataset_v2.py
new file mode 100644
index 000000000..dfeeabdef
--- /dev/null
+++ b/backend/app/tests/services/evaluations/test_dataset_v2.py
@@ -0,0 +1,123 @@
+"""Tests for the Langfuse-free dataset upload service (`upload_dataset_v2`).
+
+Covers the v2 upload slice of the three-metric SRD
+(docs/srd-three-metric-evaluation-verdict.md, FR-19/FR-20):
+
+- FR-19: creates the `evaluation_dataset` row with `langfuse_dataset_id` null and
+  never touches the Langfuse client.
+- FR-20: stores only the original rows (no physical duplication) and records the
+  run-time-duplication metadata.
+
+Object storage is the external boundary and is mocked; the dataset row lands in
+the real (transactional) DB.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+from sqlmodel import Session
+
+from app.crud.evaluations.dataset import (
+    DATASET_META_DUPLICATE_AT_RUNTIME,
+    DATASET_META_DUPLICATION_FACTOR,
+    DATASET_META_ORIGINAL_ITEMS,
+    DATASET_META_TOTAL_ITEMS,
+)
+from app.models import EvaluationDataset
+from app.services.evaluations.dataset import upload_dataset_v2
+from app.tests.utils.auth import TestAuthContext
+from app.tests.utils.utils import random_lower_string
+
+_DATASET = "app.services.evaluations.dataset"
+
+_CSV = b"question,answer\nQ0?,A0\nQ1?,A1\nQ2?,A2\nQ3?,A3\n"  # 4 original rows
+
+
+class TestUploadDatasetV2:
+    def test_creates_langfuse_free_row_without_calling_langfuse(
+        self, db: Session, user_api_key: TestAuthContext
+    ) -> None:
+        """FR-19: row has null langfuse id; no Langfuse client/upload is called."""
+        name = f"v2-upload-{random_lower_string()}"
+        with (
+            patch(f"{_DATASET}.get_cloud_storage", return_value=MagicMock()),
+            patch(
+                f"{_DATASET}.upload_csv_to_object_store",
+                return_value="s3://bucket/datasets/v2.csv",
+            ),
+            patch(f"{_DATASET}.get_langfuse_client") as mock_langfuse_client,
+            patch(f"{_DATASET}.upload_dataset_to_langfuse") as mock_langfuse_upload,
+        ):
+            dataset = upload_dataset_v2(
+                session=db,
+                csv_content=_CSV,
+                dataset_name=name,
+                description=None,
+                duplication_factor=5,
+                organization_id=user_api_key.organization_id,
+                project_id=user_api_key.project_id,
+            )
+
+        mock_langfuse_client.assert_not_called()
+        mock_langfuse_upload.assert_not_called()
+
+        persisted = db.get(EvaluationDataset, dataset.id)
+        assert persisted is not None
+        assert persisted.langfuse_dataset_id is None
+        assert persisted.object_store_url == "s3://bucket/datasets/v2.csv"
+
+    def test_stores_original_rows_and_runtime_dup_metadata(
+        self, db: Session, user_api_key: TestAuthContext
+    ) -> None:
+        """FR-20: original CSV stored verbatim; metadata records run-time dup."""
+        name = f"v2-meta-{random_lower_string()}"
+        with (
+            patch(f"{_DATASET}.get_cloud_storage", return_value=MagicMock()),
+            patch(
+                f"{_DATASET}.upload_csv_to_object_store",
+                return_value="s3://bucket/datasets/v2.csv",
+            ) as mock_upload,
+            patch(f"{_DATASET}.get_langfuse_client"),
+            patch(f"{_DATASET}.upload_dataset_to_langfuse"),
+        ):
+            dataset = upload_dataset_v2(
+                session=db,
+                csv_content=_CSV,
+                dataset_name=name,
+                description=None,
+                duplication_factor=5,
+                organization_id=user_api_key.organization_id,
+                project_id=user_api_key.project_id,
+            )
+
+        # The bytes handed to S3 are the original rows, not a duplicated payload.
+        assert mock_upload.call_args.kwargs["csv_content"] == _CSV
+
+        meta = dataset.dataset_metadata
+        assert meta[DATASET_META_DUPLICATE_AT_RUNTIME] is True
+        assert meta[DATASET_META_DUPLICATION_FACTOR] == 5
+        assert meta[DATASET_META_ORIGINAL_ITEMS] == 4
+        assert meta[DATASET_META_TOTAL_ITEMS] == 20  # 4 rows × factor 5
+
+    def test_object_store_returns_no_url_raises_500(
+        self, db: Session, user_api_key: TestAuthContext
+    ) -> None:
+        """S3 is the sole source in v2, so a missing url is fatal (no Langfuse fallback)."""
+        with (
+            patch(f"{_DATASET}.get_cloud_storage", return_value=MagicMock()),
+            patch(f"{_DATASET}.upload_csv_to_object_store", return_value=None),
+            patch(f"{_DATASET}.get_langfuse_client"),
+        ):
+            with pytest.raises(HTTPException) as exc:
+                upload_dataset_v2(
+                    session=db,
+                    csv_content=_CSV,
+                    dataset_name=f"v2-nourl-{random_lower_string()}",
+                    description=None,
+                    duplication_factor=2,
+                    organization_id=user_api_key.organization_id,
+                    project_id=user_api_key.project_id,
+                )
+
+        assert exc.value.status_code == 500
diff --git a/backend/app/tests/services/evaluations/test_load_run_dataset_items.py b/backend/app/tests/services/evaluations/test_load_run_dataset_items.py
new file mode 100644
index 000000000..b4908b4b4
--- /dev/null
+++ b/backend/app/tests/services/evaluations/test_load_run_dataset_items.py
@@ -0,0 +1,201 @@
+"""Tests for the run-time dataset item loader (`load_run_dataset_items`).
+
+Covers the v2 run-time-duplication slice of the three-metric SRD
+(docs/srd-three-metric-evaluation-verdict.md, FR-21/FR-22):
+
+- FR-21: a v2 dataset (null Langfuse id, run-time-duplication marker, factor N)
+  expands each original row ×N with unique ids at run time.
+- FR-22: a v1 dataset (Langfuse-backed) is read from Langfuse as-is, never
+  re-multiplied, and its S3 CSV is not touched.
+
+The S3 download and the Langfuse fetch are the external boundaries and are
+mocked; the dataset row and its metadata live in the real (transactional) DB.
+"""
+
+from unittest.mock import MagicMock, patch
+from uuid import uuid4
+
+import pytest
+from fastapi import HTTPException
+from sqlmodel import Session
+
+from app.crud.evaluations.dataset import (
+    DATASET_META_DUPLICATE_AT_RUNTIME,
+    DATASET_META_DUPLICATION_FACTOR,
+    DATASET_META_ORIGINAL_ITEMS,
+    DATASET_META_TOTAL_ITEMS,
+    create_evaluation_dataset,
+)
+from app.models import EvaluationDataset
+from app.services.evaluations.fast import (
+    load_run_dataset_items,
+    validate_and_start_fast_evaluation,
+)
+from app.tests.utils.auth import TestAuthContext
+from app.tests.utils.utils import random_lower_string
+
+_FAST = "app.services.evaluations.fast"
+
+
+def _csv_bytes(n_rows: int) -> bytes:
+    lines = ["question,answer"]
+    for i in range(n_rows):
+        lines.append(f"Q{i}?,A{i}")
+    return ("\n".join(lines) + "\n").encode("utf-8")
+
+
+def _make_v2_dataset(
+    *,
+    db: Session,
+    auth: TestAuthContext,
+    original_items_count: int,
+    duplication_factor: int,
+    duplicate_at_runtime: bool = True,
+) -> EvaluationDataset:
+    """A Langfuse-free dataset: null langfuse id, S3 url, run-time-dup metadata."""
+    return create_evaluation_dataset(
+        session=db,
+        name=f"v2_ds_{random_lower_string()}",
+        dataset_metadata={
+            DATASET_META_ORIGINAL_ITEMS: original_items_count,
+            DATASET_META_TOTAL_ITEMS: original_items_count * duplication_factor,
+            DATASET_META_DUPLICATION_FACTOR: duplication_factor,
+            DATASET_META_DUPLICATE_AT_RUNTIME: duplicate_at_runtime,
+        },
+        object_store_url="s3://bucket/datasets/v2.csv",
+        langfuse_dataset_id=None,
+        organization_id=auth.organization_id,
+        project_id=auth.project_id,
+    )
+
+
+class TestV2RunTimeDuplication:
+    def test_expands_each_row_by_duplication_factor(
+        self, db: Session, user_api_key: TestAuthContext
+    ) -> None:
+        """FR-21: 8 original rows × factor 5 → 40 items with unique ids."""
+        dataset = _make_v2_dataset(
+            db=db, auth=user_api_key, original_items_count=8, duplication_factor=5
+        )
+
+        with (
+            patch(f"{_FAST}.get_cloud_storage", return_value=MagicMock()),
+            patch(
+                f"{_FAST}.download_csv_from_object_store",
+                return_value=_csv_bytes(8),
+            ) as mock_download,
+        ):
+            items = load_run_dataset_items(session=db, dataset=dataset, langfuse=None)
+
+        assert len(items) == 40
+        assert len({item["id"] for item in items}) == 40
+        mock_download.assert_called_once()
+
+    def test_duplicates_of_a_row_share_question_id(
+        self, db: Session, user_api_key: TestAuthContext
+    ) -> None:
+        """FR-21: the N copies of one original row carry the same question_id."""
+        dataset = _make_v2_dataset(
+            db=db, auth=user_api_key, original_items_count=3, duplication_factor=4
+        )
+
+        with (
+            patch(f"{_FAST}.get_cloud_storage", return_value=MagicMock()),
+            patch(
+                f"{_FAST}.download_csv_from_object_store",
+                return_value=_csv_bytes(3),
+            ),
+        ):
+            items = load_run_dataset_items(session=db, dataset=dataset, langfuse=None)
+
+        # Group ids by their (shared) question_id; each original row → one group of 4.
+        groups: dict[int, set[str]] = {}
+        for item in items:
+            groups.setdefault(item["metadata"]["question_id"], set()).add(item["id"])
+
+        assert sorted(groups) == [1, 2, 3]
+        assert all(len(ids) == 4 for ids in groups.values())
+
+    def test_marker_absent_does_not_multiply(
+        self, db: Session, user_api_key: TestAuthContext
+    ) -> None:
+        """A null-langfuse dataset without the run-time-dup marker loads as-is."""
+        dataset = _make_v2_dataset(
+            db=db,
+            auth=user_api_key,
+            original_items_count=5,
+            duplication_factor=5,
+            duplicate_at_runtime=False,
+        )
+
+        with (
+            patch(f"{_FAST}.get_cloud_storage", return_value=MagicMock()),
+            patch(
+                f"{_FAST}.download_csv_from_object_store",
+                return_value=_csv_bytes(5),
+            ),
+        ):
+            items = load_run_dataset_items(session=db, dataset=dataset, langfuse=None)
+
+        assert len(items) == 5
+
+
+class TestV1DatasetReadAsIs:
+    def test_langfuse_backed_dataset_is_not_re_multiplied(
+        self, db: Session, user_api_key: TestAuthContext
+    ) -> None:
+        """FR-22: a v1 dataset is read from Langfuse verbatim; S3 untouched."""
+        dataset = create_evaluation_dataset(
+            session=db,
+            name=f"v1_ds_{random_lower_string()}",
+            dataset_metadata={
+                DATASET_META_ORIGINAL_ITEMS: 3,
+                DATASET_META_TOTAL_ITEMS: 15,
+                DATASET_META_DUPLICATION_FACTOR: 5,
+            },
+            object_store_url="s3://bucket/datasets/v1.csv",
+            langfuse_dataset_id="langfuse_abc",
+            organization_id=user_api_key.organization_id,
+            project_id=user_api_key.project_id,
+        )
+        # v1 datasets are already physically duplicated in Langfuse: 3 rows × 5 = 15.
+        langfuse_items = [{"id": f"lf_{i}"} for i in range(15)]
+
+        with (
+            patch(
+                f"{_FAST}.fetch_dataset_items", return_value=langfuse_items
+            ) as mock_fetch,
+            patch(f"{_FAST}.download_csv_from_object_store") as mock_download,
+        ):
+            items = load_run_dataset_items(
+                session=db, dataset=dataset, langfuse=MagicMock()
+            )
+
+        assert len(items) == 15
+        mock_fetch.assert_called_once()
+        mock_download.assert_not_called()
+
+
+class TestV1RunStillRequiresLangfuseId:
+    def test_non_judge_run_on_null_langfuse_dataset_400s(
+        self, db: Session, user_api_key: TestAuthContext
+    ) -> None:
+        """The relaxed langfuse-id check must not leak to v1 (is_judge_run=False)."""
+        dataset = _make_v2_dataset(
+            db=db, auth=user_api_key, original_items_count=3, duplication_factor=1
+        )
+
+        with pytest.raises(HTTPException) as exc:
+            validate_and_start_fast_evaluation(
+                session=db,
+                dataset_id=dataset.id,
+                run_name=f"v1-null-lf-{random_lower_string()}",
+                config_id=uuid4(),
+                config_version=1,
+                organization_id=user_api_key.organization_id,
+                project_id=user_api_key.project_id,
+                is_judge_run=False,
+            )
+
+        assert exc.value.status_code == 400
+        assert "langfuse" in exc.value.detail.lower()
diff --git a/docs/wiki/modules/evaluations.md b/docs/wiki/modules/evaluations.md
index 7f296c6e9..eaea4127e 100644
--- a/docs/wiki/modules/evaluations.md
+++ b/docs/wiki/modules/evaluations.md
@@ -6,7 +6,8 @@ Deep dive: `docs/architecture/kaapi-evaluations-ARCHITECTURE.md` (§3 data model
 All paths relative to `backend/app/`.
 
 ## Routes
-- `api/routes/evaluations/dataset.py`, `api/routes/evaluations/evaluation.py` — text datasets + runs
+- `api/routes/evaluations/dataset.py`, `api/routes/evaluations/evaluation.py` — text datasets + runs (v1, `/api/v1`)
+- `api/routes/evaluations/evaluation_v2.py` — `POST /api/v2/evaluations`, replica of v1 run trigger + native ground-truth LLM judge (Langfuse-free); mounted under `settings.API_V2_STR`
 - `api/routes/stt_evaluations/`, `api/routes/tts_evaluations/` — STT/TTS
 - `api/routes/cron.py` — batch polling trigger
 
@@ -19,11 +20,12 @@ All paths relative to `backend/app/`.
 | `batch_job` (BatchJob) | `models/batch_job.py` |
 
 Key `EvaluationRun` JSONB fields: `score` (per-trace `scores` + `summary_scores`), `per_item_scores`, `cost` (per-stage), `unscoreable`. References `config_id`, `batch_job_id`.
+v2 judge fields on `EvaluationRun`: `is_judge_run` (bool marker gating native judging + Langfuse skip), `per_item_ground_truth` (JSONB `{ref: score}`, Kaapi-native, not synced to Langfuse). Judging is system-config only — always the fallback model (`gpt-5-mini`) + built-in prompt, no per-run config.
 
 ## Services / CRUD
-- `services/evaluations/` — `evaluation.py`, `dataset.py`, `fast.py`, `batch_job.py`, `validators.py`, `prompt_improvement.py`
+- `services/evaluations/` — `evaluation.py`, `dataset.py`, `fast.py`, `judge.py` (v2 judged run trigger), `batch_job.py`, `validators.py`, `prompt_improvement.py`
 - `services/stt_evaluations/`, `services/tts_evaluations/`
-- `crud/evaluations/` — `core.py`, `batch.py`, `fast.py`, `score.py`, `embeddings.py`, `cost.py`, `langfuse.py`, `merge.py`, `processing.py`, `cron.py`
+- `crud/evaluations/` — `core.py`, `batch.py`, `fast.py`, `judge.py` (metric registry + combined judge call), `score.py`, `embeddings.py`, `cost.py`, `langfuse.py`, `merge.py`, `processing.py`, `cron.py`
 - `core/batch/` — shared provider batch clients: `openai.py`, `gemini.py`, `anthropic.py`, `polling.py`, `operations.py`
 
 ## Async
@@ -36,3 +38,4 @@ Key `EvaluationRun` JSONB fields: `score` (per-trace `scores` + `summary_scores`
 ## Gotchas
 - Eval traffic deliberately bypasses `/llm/call` (separate code path from production).
 - Scores sync to Langfuse; durable per-row maps on `evaluation_run` are the resync source of truth.
+- v2 (`is_judge_run`) is fully Kaapi-native: no Langfuse dataset/trace/score sync; the native ground-truth judge is one combined `responses.create` per row, structured via `crud/evaluations/judge.py` `METRIC_REGISTRY` so knowledge_base/prompt metrics + verdict slot in later. v1 path is unchanged.