kiln_ai.adapters.eval.eval_runner

  1import logging
  2from dataclasses import dataclass
  3from typing import Any, AsyncGenerator, Dict, List, Literal, Set
  4
  5from kiln_ai.adapters.adapter_registry import load_skills_for_task
  6from kiln_ai.adapters.chat.chat_formatter import (
  7    chat_strategy_for_run,
  8    is_two_message_cot_strategy,
  9)
 10from kiln_ai.adapters.eval.base_eval import BaseEval, BaseV2EvalBridge
 11from kiln_ai.adapters.eval.registry import legacy_eval_adapter_from_type
 12from kiln_ai.adapters.eval.trace_index import TraceIndex, TraceKey, trace_key
 13from kiln_ai.adapters.ml_model_list import built_in_models_from_provider
 14from kiln_ai.adapters.model_adapters.base_adapter import SkillsDict
 15from kiln_ai.adapters.prompt_builders import prompt_builder_from_id
 16from kiln_ai.adapters.provider_tools import kiln_model_provider_from
 17from kiln_ai.adapters.retry_classification import (
 18    is_retryable_error,
 19    unwrap_kiln_run_error,
 20)
 21from kiln_ai.datamodel.basemodel import ID_TYPE, generate_model_id
 22from kiln_ai.datamodel.datamodel_enums import ModelProviderName
 23from kiln_ai.datamodel.dataset_filters import (
 24    DatasetFilterId,
 25    dataset_filter_from_id,
 26)
 27from kiln_ai.datamodel.eval import (
 28    Eval,
 29    EvalConfig,
 30    EvalConfigType,
 31    EvalDataType,
 32    EvalInput,
 33    EvalRun,
 34    EvalScores,
 35    EvalTaskInput,
 36    MultiTurnSyntheticEvalInputData,
 37    SkippedReason,
 38    V2EvalResult,
 39)
 40from kiln_ai.datamodel.eval_splits import (
 41    ItemKey,
 42    ResolvedSplit,
 43    eval_run_item_key,
 44    item_key,
 45)
 46from kiln_ai.datamodel.run_config import (
 47    KilnAgentRunConfigProperties,
 48    as_kiln_agent_run_config,
 49)
 50from kiln_ai.datamodel.task import Task, TaskRunConfig
 51from kiln_ai.datamodel.task_run import EvalItemSource, TaskRun, Usage
 52from kiln_ai.datamodel.usage import MessageUsage
 53from kiln_ai.synthetic_user import drive_case_for_eval
 54from kiln_ai.synthetic_user.drive_loop import DriveCaseResult
 55from kiln_ai.synthetic_user.models import (
 56    TAG_SU_ENDED_CONVERSATION,
 57    SyntheticUserDriverConfig,
 58)
 59from kiln_ai.utils.async_job_runner import AsyncJobRunner, Progress, RetryableError
 60from kiln_ai.utils.git_sync_protocols import SaveContext, default_save_context
 61from kiln_ai.utils.open_ai_types import ChatCompletionMessageParam, serialize_trace
 62from kiln_ai.utils.slow_operation import log_if_slow
 63
 64logger = logging.getLogger(__name__)
 65
 66
 67@dataclass
 68class EvalJob:
 69    item: TaskRun | EvalInput
 70    type: Literal["task_run_eval", "eval_config_eval"]
 71    eval_config: EvalConfig
 72    task_run_config: TaskRunConfig | None = None
 73
 74
 75def _calibration_item(job: EvalJob) -> TaskRun | None:
 76    """The golden TaskRun a calibration job scores, or None if this isn't calibration.
 77
 78    Calibration generates nothing: the human-rated dataset item *is* what gets scored
 79    (functional spec 4.5). So the trace is known before the job starts, and stays known
 80    even for a job that is skipped before it reaches the judge — which is why both the
 81    scoring path and the skip path ask here rather than each deciding for themselves.
 82    """
 83    if job.type != "eval_config_eval":
 84        return None
 85    if not isinstance(job.item, TaskRun):
 86        raise ValueError("Calibration items are always TaskRuns")
 87    return job.item
 88
 89
 90def _has_text_content(content: Any) -> bool:
 91    """Whether a message's content carries any text.
 92
 93    Content is either a plain string or a list of content parts, and both shapes reach
 94    here — traces are written by adapters and read back through pydantic.
 95    """
 96    if isinstance(content, str):
 97        return bool(content.strip())
 98    if isinstance(content, list):
 99        return any(
100            isinstance(part, dict) and bool(str(part.get("text") or "").strip())
101            for part in content
102        )
103    return False
104
105
106def conversation_health_problem(
107    trace: list[ChatCompletionMessageParam] | None,
108    required_turns: int,
109) -> str | None:
110    """Why `trace` is not a complete conversation of `required_turns` turns, or None when
111    it is complete.
112
113    Structural completeness only, never error-freeness: a conversation whose tool calls
114    failed is a legitimate thing to evaluate (judging how an agent handles errors is a
115    first-class eval), so error-bearing tool messages say nothing about health here. What
116    it does catch is a conversation that stopped short, which would otherwise be judged
117    as if the agent had simply finished.
118    """
119    messages = trace or []
120    user_turns = sum(1 for message in messages if message.get("role") == "user")
121    if user_turns != required_turns:
122        return f"expected {required_turns} user turns, found {user_turns}"
123    if not messages:
124        return "the conversation is empty"
125    last_role = messages[-1].get("role")
126    if last_role != "assistant":
127        return f"the conversation ends with a '{last_role}' message, not an assistant reply"
128    if not _has_text_content(messages[-1].get("content")):
129        return "the final assistant message has no text content"
130    return None
131
132
133def _splits_a_turn_into_two_messages(
134    properties: KilnAgentRunConfigProperties, task: Task
135) -> bool:
136    """Whether this run config answers one turn with two user-role messages.
137
138    A chain of thought prompt on a model with no reasoning step of its own is served by
139    asking the model to think, then sending a second user message asking for the final
140    answer. That extra message is indistinguishable from a real user turn, so a driven
141    conversation both miscounts its turns and feeds the injected message back to the
142    synthetic user as if the user had written it.
143
144    The strategy is resolved from the same prompt builder and model provider the adapter
145    reads, so the answer here is the one the drive would get. Built-in models are read
146    straight from the model table rather than through the credentialed provider lookup:
147    this preflight runs before any paid call and must give the same answer on a machine
148    with no provider keys, and the two fields it needs are static entries in that table.
149    Custom, fine-tuned and litellm models are not in the table, so they keep the
150    credentialed lookup. A config that cannot be resolved answers False: the drive
151    fails on the identical lookup before it spends anything, and one unresolvable
152    config must not block the run configs beside it.
153    """
154    try:
155        cot_prompt = prompt_builder_from_id(
156            properties.prompt_id, task
157        ).chain_of_thought_prompt()
158        provider = None
159        if cot_prompt:
160            provider = built_in_models_from_provider(
161                properties.model_provider_name, properties.model_name
162            ) or kiln_model_provider_from(
163                properties.model_name, properties.model_provider_name
164            )
165    except Exception as error:
166        logger.warning(
167            "Could not resolve the chat strategy for model '%s' with prompt '%s' (%s); "
168            "leaving its multi-turn compatibility to the per-job checks",
169            properties.model_name,
170            properties.prompt_id,
171            error,
172        )
173        return False
174    return is_two_message_cot_strategy(
175        chat_strategy_for_run(
176            cot_prompt=cot_prompt,
177            tuned_chat_strategy=provider.tuned_chat_strategy if provider else None,
178            reasoning_capable=provider.reasoning_capable if provider else False,
179        )
180    )
181
182
183def no_golden_set_message(eval: Eval) -> str:
184    """Why judge comparison can't run without a golden set. One wording, two raisers.
185
186    Shared with the API layer so the 4xx a user sees and the ValueError a library caller
187    sees say the same thing (functional spec 9, architecture 4.3).
188    """
189    return (
190        f"Eval '{eval.id}' has no golden set configured. Comparing judges scores them "
191        "against a set of human-rated dataset items, so the eval needs one before a "
192        "comparison can run."
193    )
194
195
196class EvalRunner:
197    """
198    Runs an eval. Async execution is supported to make it faster when using remote/fast model providers.
199
200    Can run an eval in 2 modes:
201    1) eval_config_eval: evaluate an eval config (judge quality) using existing dataset
202       items. Scoped by the eval's golden filter, so its items are always TaskRuns.
203    2) task_run_eval: evaluate a range of task run configs, generating new run output.
204       Scoped by the `split` it is given, whose items may come from either store.
205       Multi-turn synthetic EvalInputs are re-driven as a full conversation per run
206       config using the drive config stamped on each item; stored multi-turn TaskRun
207       chains are skipped.
208    """
209
210    def __init__(
211        self,
212        eval_configs: List[EvalConfig],
213        run_configs: List[TaskRunConfig] | None,
214        eval_run_type: Literal["eval_config_eval", "task_run_eval"],
215        split: ResolvedSplit | None = None,
216        save_context: SaveContext | None = None,
217    ):
218        if len(eval_configs) == 0:
219            raise ValueError("Eval runner requires at least one eval config")
220        target_eval = eval_configs[0].parent_eval()
221        if target_eval is None:
222            raise ValueError("Eval config requires a parent eval")
223        for eval_config in eval_configs:
224            parent_eval = eval_config.parent_eval()
225            if parent_eval is None:
226                raise ValueError("Eval config requires a parent eval")
227            if parent_eval.id != target_eval.id:
228                raise ValueError("All eval configs must have the same parent eval")
229
230        target_task = target_eval.parent_task()
231        if target_task is None:
232            raise ValueError("Eval config requires a (grand)parent task")
233
234        # Both modes settle *what defines* their item scope here rather than at collect
235        # time, but to different depths, and the difference is deliberate. task_run_eval
236        # takes a ResolvedSplit: the items themselves, fixed now, so a split cannot be
237        # accepted and then quietly re-resolved to something else. eval_config_eval takes
238        # the golden filter id and still applies it to self.task.runs() at collect time,
239        # so items added after construction are picked up — golden is TaskRun-only by
240        # definition, so there is no source ambiguity for that to hide.
241        self.golden_filter_id: DatasetFilterId | None = None
242        if eval_run_type == "task_run_eval":
243            if run_configs is None or len(run_configs) == 0:
244                raise ValueError("Task run eval requires run configs")
245            for run_config in run_configs:
246                parent_task = run_config.parent_task()
247                if parent_task is None:
248                    raise ValueError("All run configs must have a parent task")
249                if parent_task.id != target_task.id:
250                    raise ValueError(
251                        "Run config is not for the same task as the eval configs"
252                    )
253            if split is None:
254                raise ValueError("Task run eval requires a resolved split")
255            if split.eval_id != target_eval.id:
256                raise ValueError(
257                    f"Split '{split.name}' was resolved from eval '{split.eval_id}', not from "
258                    f"eval '{target_eval.id}' whose configs are being run"
259                )
260        else:
261            if run_configs is not None:
262                raise ValueError("Mode 'eval_config_eval' does not support run configs")
263            if split is not None:
264                raise ValueError(
265                    "Mode 'eval_config_eval' does not support a split: it is scoped by the eval's golden filter"
266                )
267            if target_eval.eval_configs_filter_id is None:
268                raise ValueError(no_golden_set_message(target_eval))
269            self.golden_filter_id = target_eval.eval_configs_filter_id
270
271        self.eval_run_type = eval_run_type
272        self.eval_configs = eval_configs
273        self.run_configs = run_configs
274        self.split = split
275        self.task = target_task
276        self.eval = target_eval
277        self._skills: SkillsDict = self._preload_skills()
278        self._save_context: SaveContext = save_context or default_save_context
279        # Live, not precomputed like `already_run`: a trace persisted by one job has to be
280        # visible to the next, whether that next job is running concurrently under a
281        # different eval config or is this job's own retry (functional spec 4.2, 4.3).
282        self._trace_index = TraceIndex(self.task)
283
284    def collect_tasks(self) -> List[EvalJob]:
285        if self.eval_run_type == "eval_config_eval":
286            if self.golden_filter_id is None:
287                raise ValueError(no_golden_set_message(self.eval))
288            return self.collect_tasks_for_eval_config_eval(self.golden_filter_id)
289        return self.collect_tasks_for_task_run_eval()
290
291    def collect_tasks_for_eval_config_eval(
292        self, eval_configs_filter_id: DatasetFilterId
293    ) -> List[EvalJob]:
294        """
295        Collect all jobs for this run, excluding any that have already been run.
296
297        This variant is used for mode "eval_config_eval", using existing dataset run data (input/output).
298
299        The tasks:
300        - should be in the eval config set filter
301        - should not have already been run for this eval config + dataset item pair
302        """
303        filter = dataset_filter_from_id(eval_configs_filter_id)
304
305        # already_run[eval_config_id][dataset_id]
306        already_run: Dict[ID_TYPE, Set[ID_TYPE]] = {}
307        for eval_config in self.eval_configs:
308            already_run[eval_config.id] = set()
309            for run in eval_config.runs(readonly=True):
310                # Only calibration records mark a golden item done: the same eval
311                # config also accumulates task_run_eval records, and a golden
312                # TaskRun that was scored as a test item must still be calibrated.
313                if not run.eval_config_eval:
314                    continue
315                already_run[eval_config.id].add(run.dataset_id)
316
317        return [
318            EvalJob(
319                item=task_run,
320                eval_config=eval_config,
321                type="eval_config_eval",
322            )
323            for task_run in self.task.runs(readonly=True)
324            if filter(task_run)
325            for eval_config in self.eval_configs
326            if task_run.id not in already_run[eval_config.id]
327        ]
328
329    def collect_tasks_for_task_run_eval(self) -> List[EvalJob]:
330        """
331        Collect all jobs for this run, excluding any that have already been run.
332
333        This variant is used for mode "task_run_eval", generating new run output using existing dataset item input.
334
335        The tasks:
336        - are the items of the split this runner was given, from whichever store backs it
337        - should not have already been run for this eval config + run config + dataset item
338        """
339        if self.split is None:
340            raise ValueError("Task run eval requires a resolved split")
341
342        # already_run[eval_config_id][run_config_id][item_key]
343        already_run: Dict[ID_TYPE, Dict[ID_TYPE, Set[ItemKey]]] = {}
344        for eval_config in self.eval_configs:
345            already_run[eval_config.id] = {
346                run_config.id: set() for run_config in self.run_configs or []
347            }
348            for run in eval_config.runs(readonly=True):
349                # Scopes the dedupe to the run configs actually being evaluated: an eval
350                # config accumulates runs for every run config ever compared against it.
351                # The `is not None` is not redundant with the membership test — ID_TYPE is
352                # `str | None`, so a run config whose file carries a null id would make
353                # None a real key and fold every calibration record into its set.
354                if (
355                    run.task_run_config_id is not None
356                    and run.task_run_config_id in already_run[eval_config.id]
357                ):
358                    already_run[eval_config.id][run.task_run_config_id].add(
359                        eval_run_item_key(run)
360                    )
361
362        return [
363            EvalJob(
364                item=item,
365                task_run_config=run_config,
366                type="task_run_eval",
367                eval_config=eval_config,
368            )
369            for item in self.split.items
370            for eval_config in self.eval_configs
371            for run_config in self.run_configs or []
372            if (self.split.source, item.id)
373            not in already_run[eval_config.id][run_config.id]
374        ]
375
376    def validate_multi_turn_drive_readiness(
377        self, check_run_configs: bool = True
378    ) -> None:
379        """Fail fast on config problems every multi-turn re-drive job would
380        hit: no item carrying a synthetic-user drive config, a stamped
381        config with an unknown model provider, run configs that aren't
382        Kiln agent configs, or run configs that answer in two messages per
383        turn. Callers can invoke this before starting a batch
384        so the user gets one clear error up front instead of one opaque
385        error per job. The per-job checks stay as the backstop for
386        standalone runner use.
387
388        A partially stamped split is NOT an error here: the stamped items
389        can still run, and each unstamped item records a per-item skip
390        naming the fix. Only an entirely unstamped split fails up front,
391        because nothing could run and one clear error beats a page of
392        identical skips.
393
394        `check_run_configs=False` limits validation to the drive configs —
395        for callers running a fleet the user didn't hand-pick, where one
396        incompatible run config shouldn't block every other config's jobs.
397
398        Raises ValueError listing every problem; no-op unless re-drive jobs
399        can actually occur (task_run_eval over a split containing multi-turn
400        synthetic items — stored-TaskRun sources never re-drive, and a
401        single-turn split has nothing to drive).
402        """
403        if self.eval_run_type != "task_run_eval" or self.split is None:
404            return
405        multi_turn_items = [
406            item.data
407            for item in self.split.items
408            if isinstance(item, EvalInput)
409            and isinstance(item.data, MultiTurnSyntheticEvalInputData)
410        ]
411        if not multi_turn_items:
412            return
413        problems: list[str] = []
414        stamped_configs = [
415            data.drive_config
416            for data in multi_turn_items
417            if data.drive_config is not None
418        ]
419        if not stamped_configs:
420            problems.append(
421                "none of this eval's multi-turn items has a synthetic user "
422                "configuration (create a new batch to mint items that carry one)"
423            )
424        for provider in sorted({cfg.model_provider for cfg in stamped_configs}):
425            try:
426                ModelProviderName(provider)
427            except ValueError:
428                problems.append(
429                    "a multi-turn item's synthetic-user drive config has "
430                    f"unknown model provider '{provider}'"
431                )
432        if check_run_configs:
433            for run_config in self.run_configs or []:
434                try:
435                    agent_properties = as_kiln_agent_run_config(
436                        run_config.run_config_properties
437                    )
438                except ValueError:
439                    problems.append(
440                        f"run config '{run_config.name}' is not a Kiln agent "
441                        "config, so it can't hold a multi-turn conversation"
442                    )
443                    continue
444                if _splits_a_turn_into_two_messages(agent_properties, self.task):
445                    problems.append(
446                        f"run config '{run_config.name}' uses a chain of thought "
447                        "prompt with a model that has no reasoning step of its own, "
448                        "so it answers in two messages per turn and can't hold a "
449                        "multi-turn conversation (pick a reasoning-capable model, or "
450                        "a prompt without thinking instructions)"
451                    )
452        if problems:
453            raise ValueError(
454                "Cannot re-drive this eval's multi-turn conversations: "
455                + "; ".join(problems)
456                + "."
457            )
458
459    def _preload_skills(self) -> SkillsDict:
460        """Collect all skill IDs from run configs and bulk-load them once."""
461        if self.run_configs is None:
462            return {}
463        merged: SkillsDict = {}
464        for rc in self.run_configs:
465            skills = load_skills_for_task(self.task, rc.run_config_properties)
466            merged.update(skills)
467        return merged
468
469    async def run(self, concurrency: int = 25) -> AsyncGenerator[Progress, None]:
470        """
471        Runs the configured eval run with parallel workers and yields progress updates.
472        """
473        jobs = self.collect_tasks()
474
475        runner = AsyncJobRunner(
476            concurrency=concurrency,
477            jobs=jobs,
478            run_job_fn=self.run_job,
479            max_retries=2,
480        )
481        async for progress in runner.run():
482            yield progress
483
484    async def run_job(self, job: EvalJob) -> bool:
485        try:
486            if job.eval_config.config_type == EvalConfigType.v2:
487                return await self._run_v2_job(job)
488            else:
489                return await self._run_legacy_job(job)
490        except RetryableError as e:
491            # Already classified by whoever raised it: this is a deliberate, expected
492            # ask for another attempt (an unusable generation the retry replaces), not a
493            # failure. Re-raised untouched, and logged without a stacktrace so a
494            # self-healing event doesn't read as a crash in the logs.
495            logger.warning(f"Retrying eval job for dataset item {job.item.id}: {e}")
496            raise
497        except Exception as e:
498            if is_retryable_error(e):
499                logger.error(
500                    f"Transient error running eval job for dataset item {job.item.id}: {e}",
501                    exc_info=True,
502                )
503                # KilnRunError's own message is genericized user-facing text; keep
504                # the underlying provider detail for the developer-facing error log.
505                raise RetryableError(str(unwrap_kiln_run_error(e))) from e
506            logger.error(
507                f"Error running eval job for dataset item {job.item.id}: {e}",
508                exc_info=True,
509            )
510            raise
511
512    async def _run_legacy_job(self, job: EvalJob) -> bool:
513        if not isinstance(job.item, TaskRun):
514            raise ValueError("Legacy eval jobs require a TaskRun item")
515
516        evaluator = legacy_eval_adapter_from_type(job.eval_config)(
517            job.eval_config,
518            job.task_run_config.run_config_properties if job.task_run_config else None,
519            skills=self._skills,
520        )
521        if not isinstance(evaluator, BaseEval):
522            raise ValueError("Not able to create evaluator from eval config")
523
524        task_output: str | None = None
525        reference_answer: str | None = None
526        trace: str | None = None
527        scores: EvalScores | None = None
528        intermediate_outputs: Dict[str, str] | None = None
529        task_run_usage: Usage | None = None
530        if job.type == "eval_config_eval":
531            scores, intermediate_outputs = await evaluator.run_eval(job.item)
532            task_output = job.item.output.output
533            task_run_usage = job.item.usage
534        else:
535            (
536                result_task_run,
537                scores,
538                intermediate_outputs,
539            ) = await evaluator.run_task_and_eval(job.item)
540            task_output = result_task_run.output.output
541            task_run_usage = result_task_run.usage
542
543            parent_eval = job.eval_config.parent_eval()
544            if (
545                parent_eval
546                and parent_eval.evaluation_data_type == EvalDataType.full_trace
547                and result_task_run.trace
548            ):
549                trace = serialize_trace(result_task_run.trace)
550
551            if (
552                parent_eval
553                and parent_eval.evaluation_data_type == EvalDataType.reference_answer
554            ):
555                reference_answer = job.item.output.output
556
557        async with self._save_context():
558            eval_run = EvalRun(
559                parent=job.eval_config,
560                task_run_config_id=job.task_run_config.id
561                if job.task_run_config
562                else None,
563                dataset_id=job.item.id,
564                eval_config_eval=job.type == "eval_config_eval",
565                scores=scores,
566                input=job.item.input,
567                output=task_output,
568                reference_answer=reference_answer,
569                intermediate_outputs=intermediate_outputs,
570                task_run_trace=trace,
571                task_run_usage=task_run_usage,
572            )
573            eval_run.save_to_file()
574
575        return True
576
577    async def _run_v2_job(self, job: EvalJob) -> bool:
578        from kiln_ai.adapters.eval.registry import v2_eval_adapter_from_config
579
580        try:
581            rc_props = (
582                job.task_run_config.run_config_properties
583                if job.task_run_config
584                else None
585            )
586            evaluator = v2_eval_adapter_from_config(
587                job.eval_config, rc_props, self._skills
588            )
589        except NotImplementedError:
590            return await self._persist_skip(
591                job,
592                SkippedReason.type_not_available,
593                "V2 eval type not yet implemented",
594            )
595
596        if (
597            isinstance(job.item, EvalInput)
598            and isinstance(job.item.data, MultiTurnSyntheticEvalInputData)
599            and job.type == "task_run_eval"
600        ):
601            # Multi-turn synthetic input: re-drive the conversation fresh
602            # for this run config, then judge the new trace. The job.type
603            # guard is defensive — collect_tasks never pairs eval_config_eval
604            # with EvalInput items, because judge calibration is scoped by the
605            # golden filter and golden filters only address TaskRuns. There is
606            # no runtime handler for a hand-built job of that shape.
607            seed = (
608                job.item.data.first_message.text if job.item.data.first_message else ""
609            )
610            return await self._run_v2_multi_turn_synthetic_job(
611                job, evaluator, job.item, job.item.data, seed
612            )
613
614        # Both skips come before `_resolve_trace`, so a job that can never be scored
615        # never pays for a generation.
616        if (
617            isinstance(job.item, TaskRun)
618            and job.item.parent_task_run_id is not None
619            and job.type == "task_run_eval"
620        ):
621            return await self._persist_skip(
622                job,
623                SkippedReason.incompatible_input_shape,
624                "Stored multi-turn conversations can't be re-run for a run config",
625            )
626
627        trace = await self._resolve_trace(job, evaluator)
628        eval_task_input = EvalTaskInput.from_trace(trace, job.item)
629        result = await evaluator.evaluate(eval_task_input)
630        return await self._persist_judgment(job, trace, result)
631
632    async def _resolve_trace(
633        self, job: EvalJob, evaluator: BaseV2EvalBridge
634    ) -> TaskRun:
635        """The TaskRun this job scores, generating it only if the task has none."""
636        golden = _calibration_item(job)
637        if golden is not None:
638            return golden
639
640        if job.task_run_config is None:
641            raise ValueError("A task_run_eval job requires a run config")
642
643        key = trace_key(item_key(job.item), job.task_run_config.id)
644        # `TraceIndex` logs both outcomes itself — debug on reuse, info on generation
645        # (architecture 8) — so the was_generated bool has no reader here.
646        trace, _ = await self._trace_index.get_or_create(
647            key, lambda: self._generate_and_persist(job, evaluator, key)
648        )
649        return trace
650
651    async def _generate_and_persist(
652        self, job: EvalJob, evaluator: BaseV2EvalBridge, key: TraceKey
653    ) -> TaskRun:
654        """Run the task for this job's item, and make the result durable before scoring.
655
656        Stamped from `key` rather than re-derived from the job, so the run files itself
657        under exactly the key the index filed it under. A run that disagrees is never
658        found again, and the eval regenerates it on every future run.
659        """
660        source_type, source_id, run_config_id = key
661        trace = await evaluator.run_task(job.item, run_config_id=run_config_id)
662        if trace.id is None:
663            # `run_task` builds its adapter with allow_saving=False, and every adapter
664            # clears the id of a run it did not persist (base_adapter.py:346). The runner
665            # is the one persisting here, so it mints the id — the same thing
666            # data_gen_api.py:474 does with an unsaved adapter run.
667            #
668            # Deliberately not allow_saving=True instead: the adapter would persist the
669            # run before `eval_source` is stamped on it, so a crash in that window would
670            # leave an eval trace permanently indistinguishable from a curated dataset
671            # row — the contamination Task.runs()' default-exclude exists to prevent.
672            trace.id = generate_model_id()
673        trace.eval_source = EvalItemSource(source_type=source_type, source_id=source_id)
674        async with self._save_context():
675            trace.save_to_file()
676        return trace
677
678    async def _persist_score(
679        self,
680        job: EvalJob,
681        *,
682        scored_run_id: ID_TYPE = None,
683        scores: EvalScores | None = None,
684        skipped_reason: str | None = None,
685        skipped_detail: str | None = None,
686        intermediate_outputs: Dict[str, str] | None = None,
687        eval_usage: Usage | None = None,
688    ) -> bool:
689        """Write one V2 score record.
690
691        The single place the item-identity fields are filled in, because `collect_tasks`
692        dedupes on exactly those: a skip record and a score record that disagreed would
693        be two identities for one job. No inline trace field is ever set — they are
694        deprecated, and the trace lives on the TaskRun (functional spec 3.2).
695        """
696        async with self._save_context():
697            EvalRun(
698                parent=job.eval_config,
699                task_run_config_id=job.task_run_config.id
700                if job.task_run_config
701                else None,
702                dataset_id=job.item.id if isinstance(job.item, TaskRun) else None,
703                eval_input_id=job.item.id if isinstance(job.item, EvalInput) else None,
704                eval_config_eval=job.type == "eval_config_eval",
705                scored_run_id=scored_run_id,
706                scores=scores or {},
707                skipped_reason=skipped_reason,
708                skipped_detail=skipped_detail,
709                intermediate_outputs=intermediate_outputs,
710                eval_usage=eval_usage,
711            ).save_to_file()
712        return True
713
714    async def _persist_skip(
715        self, job: EvalJob, reason: SkippedReason, detail: str
716    ) -> bool:
717        """A job skipped before it reached the judge.
718
719        Calibration still names its trace: the golden item *is* what would have been
720        scored, it is already on disk, and a skip that happens one step later — inside
721        `evaluate()` — records it (functional spec 4.6, row 2). Where the failure
722        happened shouldn't change the record's shape, and Phase 5 migrates old
723        calibration skips to exactly this.
724
725        A scoring job genuinely has nothing to point at. Generating a trace for a job
726        that can never be scored is the spend these early skips exist to avoid.
727        """
728        golden = _calibration_item(job)
729        return await self._persist_score(
730            job,
731            scored_run_id=golden.id if golden is not None else None,
732            skipped_reason=reason.value,
733            skipped_detail=detail,
734        )
735
736    async def _persist_judgment(
737        self,
738        job: EvalJob,
739        trace: TaskRun,
740        result: V2EvalResult,
741    ) -> bool:
742        """The score for one item, pointing at the trace it was computed over.
743
744        Also the home of a scoring-time skip: the trace exists and this judge could not
745        score it, so the record carries both `scored_run_id` and `skipped_reason`
746        (functional spec 4.6, row 2).
747        """
748        return await self._persist_score(
749            job,
750            scored_run_id=trace.id,
751            scores=result.scores,
752            skipped_reason=result.skipped_reason.value
753            if result.skipped_reason
754            else None,
755            skipped_detail=result.skipped_detail,
756            intermediate_outputs=result.intermediate_outputs,
757            eval_usage=result.usage,
758        )
759
760    async def _run_v2_multi_turn_synthetic_job(
761        self,
762        job: EvalJob,
763        evaluator: BaseV2EvalBridge,
764        eval_input: EvalInput,
765        data: MultiTurnSyntheticEvalInputData,
766        seed: str,
767    ) -> bool:
768        """task_run_eval over a multi-turn synthetic input.
769
770        The run config under evaluation drives the agent while the drive
771        config stamped on the item plays the synthetic user, so each run
772        config gets its own fresh conversation — the property that makes
773        run-config comparison meaningful for multi-turn. The conversation
774        persists as one standalone TaskRun through the trace index, exactly
775        like single-turn generations: durable before scoring, and reusable
776        by every other judge over the same item and run config.
777        """
778        drive_config = data.drive_config
779        if drive_config is None:
780            return await self._persist_skip(
781                job,
782                SkippedReason.missing_drive_config,
783                "This item has no synthetic user configuration. "
784                "Create a new batch to replace it.",
785            )
786
787        if not seed:
788            return await self._persist_skip(
789                job,
790                SkippedReason.incompatible_input_shape,
791                "Multi-turn synthetic input has no first_message to open the conversation",
792            )
793
794        if job.task_run_config is None:
795            raise ValueError("Task run eval requires a run config")
796        try:
797            agent_run_config = as_kiln_agent_run_config(
798                job.task_run_config.run_config_properties
799            )
800        except ValueError as e:
801            raise ValueError(
802                "Multi-turn re-drive requires a Kiln agent run config; "
803                f"run config '{job.task_run_config.name}' is a different type"
804            ) from e
805        try:
806            su_provider = ModelProviderName(drive_config.model_provider)
807        except ValueError as e:
808            raise ValueError(
809                "Invalid synthetic-user model provider on this item's "
810                f"drive config: {drive_config.model_provider}"
811            ) from e
812
813        key = trace_key(item_key(eval_input), job.task_run_config.id)
814
815        async def drive_and_persist() -> TaskRun:
816            # No app-level timeout on the re-drive: it terminates
817            # structurally (the turn ceiling, the adapter's tool-call cap,
818            # the model client's per-request timeout). This path runs as a
819            # background job, so the watchdog log is how a pathologically
820            # slow drive gets noticed.
821            async with log_if_slow(f"eval re-drive for eval item {eval_input.id}"):
822                drive_result = await drive_case_for_eval(
823                    seed_prompt=seed,
824                    synthetic_user_info=data.synthetic_user_info,
825                    target_task=self.task,
826                    target_run_config=agent_run_config,
827                    su_driver_config=SyntheticUserDriverConfig(
828                        model_name=drive_config.model_name,
829                        model_provider_name=su_provider,
830                    ),
831                    turns=drive_config.turns,
832                    skills=self._skills,
833                )
834            turns_run = len(drive_result.chain)
835            ended_by_su = drive_result.ended_early(drive_config.turns)
836            if turns_run == 1 and drive_config.turns > 1:
837                # Not enforced: discarding a paid drive over a rule the synthetic
838                # user's prompt states but cannot guarantee would cost more than it
839                # saves. Logged because a one-turn conversation is a single-turn
840                # trace wearing a multi-turn label.
841                logger.warning(
842                    "The driven conversation for eval item %s ended on its first "
843                    "turn — the synthetic user ended it before the agent had "
844                    "answered a second message",
845                    eval_input.id,
846                )
847            leaf_trace = drive_result.chain[-1].trace if drive_result.chain else None
848            # Gated on the turns that actually ran, not the item's ceiling: the drive
849            # is still in hand, so its own chain is the exact count the leaf's trace
850            # has to match. A conversation the synthetic user ended is shorter than
851            # the ceiling and still complete, while one whose trace lost turns the
852            # chain says it ran is a broken record either way.
853            problem = conversation_health_problem(leaf_trace, turns_run)
854            if problem is not None:
855                # Persisting it would index an incomplete conversation that every judge
856                # of this item reuses from then on, so it is a failed generation rather
857                # than a cheap result. Retryable, so the job re-drives.
858                raise RetryableError(
859                    f"The driven conversation for eval item {eval_input.id} is "
860                    f"incomplete ({problem}), so it was not saved."
861                )
862            return await self._persist_driven_conversation(
863                key, drive_result, seed=seed, ended_by_su=ended_by_su
864            )
865
866        # A raising drive persists nothing, so the job's retry re-drives; a
867        # successful one is on disk before the judge sees it, so a scoring
868        # failure re-scores without paying for the conversation again.
869        trace, _ = await self._trace_index.get_or_create(key, drive_and_persist)
870
871        eval_task_input = EvalTaskInput.from_trace(trace, eval_input)
872        result = await evaluator.evaluate(eval_task_input)
873        return await self._persist_judgment(job, trace, result)
874
875    async def _persist_driven_conversation(
876        self,
877        key: TraceKey,
878        drive_result: DriveCaseResult,
879        *,
880        seed: str,
881        ended_by_su: bool,
882    ) -> TaskRun:
883        """One standalone TaskRun holding a freshly driven multi-turn conversation.
884
885        The drive itself touches no disk, and its leaf's trace already holds the
886        whole conversation — so a single childless run is the entire durable
887        record. Persisting the chain's ancestors would store partial copies of
888        the same conversation, and a child run would be invisible to the trace
889        index's childless-only seed.
890
891        Stamped from `key` rather than re-derived, for the same reason as
892        `_generate_and_persist`: the run must file itself under exactly the key
893        the index filed it under, or it is never found again.
894        """
895        source_type, source_id, run_config_id = key
896        leaf = drive_result.chain[-1]
897        if leaf.output.source is None:
898            # Fail before anything is saved: a run without an output source can't
899            # carry the run_config_id half of its reuse key, so persisting it would
900            # strand a paid conversation on disk that the index can never serve.
901            raise ValueError(
902                "Driven conversation has no output source; cannot stamp its reuse key"
903            )
904        # The drive's adapter runs detached from any TaskRunConfig, so the leaf's
905        # output source has no run_config_id; the index requires it as half the key.
906        output_source = leaf.output.source.model_copy(
907            update={"run_config_id": run_config_id}
908        )
909        run = TaskRun(
910            parent=self.task,
911            # The seed, not the leaf's own input (the synthetic user's LAST message):
912            # this run stands alone, so its input is what opened the conversation.
913            input=seed,
914            input_source=leaf.input_source,
915            output=leaf.output.model_copy(update={"source": output_source}),
916            trace=leaf.trace,
917            usage=_conversation_usage(drive_result.chain),
918            # The synthetic-user driver's spend rides its own field so `usage`
919            # stays honestly assistant-only. The whole Usage, not a cost-only
920            # stub: the SU is usually a different model on a different provider
921            # from the agent, so its tokens are the half that makes the figure
922            # reconcilable. `drive_case` already returns None for a drive whose
923            # provider reported nothing.
924            synthetic_user_usage=drive_result.su_usage,
925            cumulative_usage=leaf.cumulative_usage
926            or MessageUsage.from_trace(leaf.trace),
927            eval_source=EvalItemSource(source_type=source_type, source_id=source_id),
928            tags=[TAG_SU_ENDED_CONVERSATION] if ended_by_su else [],
929        )
930        # The drive runs with allow_saving=False, so nothing touched disk before
931        # this fully-stamped run — no crash window in which a driven conversation
932        # could persist without eval_source and pass for a curated dataset row.
933        async with self._save_context():
934            run.save_to_file()
935        return run
936
937
938def _conversation_usage(chain: list[TaskRun]) -> Usage | None:
939    """Conversation-total assistant spend for one driven multi-turn conversation.
940
941    The leaf's own `usage` covers only its final turn, so tokens and cost come
942    from the cumulative recompute over the full trace, and latency sums each
943    turn's in-flight accumulator. None when no turn reported anything — an
944    all-None Usage would read as a report where there was none.
945    """
946    leaf = chain[-1]
947    totals = leaf.cumulative_usage or MessageUsage.from_trace(leaf.trace)
948    latencies = [
949        run.usage.total_llm_latency_ms
950        for run in chain
951        if run.usage is not None and run.usage.total_llm_latency_ms is not None
952    ]
953    usage = Usage(
954        input_tokens=totals.input_tokens,
955        output_tokens=totals.output_tokens,
956        total_tokens=totals.total_tokens,
957        cost=totals.cost,
958        cached_tokens=totals.cached_tokens,
959        total_llm_latency_ms=sum(latencies) if latencies else None,
960    )
961    if all(v is None for v in usage.model_dump().values()):
962        return None
963    return usage