"""Main orchestrator for coordinating task evaluation."""

import asyncio

import logging

import re

import time

import uuid

from collections.abc import Callable

from contextlib import suppress

from dataclasses import dataclass

from datetime import datetime

from inspect import isawaitable

from pathlib import Path

from typing import Any

from urllib.parse import urlparse

from .agent import Agent

from .agents.watchdog import ThreadedWatchdog

from .analysis import calculate_command_statistics

from .config import settings

from .criteria.commands_efficiency import compute_commands_efficiency

from .errors import (

AgentCrashError,

BudgetExceededError,

TaskTimeoutError,

TurnTimeoutError,

)

from .errors.executor import execute_with_retry

from .errors.retry import create_error_context

from .evaluation.checker import SuccessChecker, _short_failure_reason

from .litellm_cost import apply_actual_cost, load_cost_records

from .models import (

DEFAULT_STOP_EARLY_GATE_THRESHOLD,

ROUTE_NAMES,

AgentKind,

ApiRoute,

BedrockRoute,

ConfigLineageEntry,

CriterionResult,

DirectRoute,

EvaluationResult,

FinalStatus,

JudgeCriterionResult,

LiteLLMRoute,

PostRunCommand,

PostRunResult,

PreRunCommand,

PreservationMode,

SimulationConfig,

SimulationTelemetry,

TaskConfigRecord,

TaskDefinition,

TokenUsage,

TurnRecord,

UserMessage,

resolve_evaluation_route,

resolve_route,

)

from .orchestration.early_stop import EarlyStopWatcher, early_stop_active, validate_early_stop

from .orchestration.evaluation import load_reference

from .path_utils import format_task_log_id, task_log_path

from .sandbox import Sandbox

from .simulation import DialogStopReason, SimulatorResult, UserSimulator, evaluate_stop

from .streaming.callbacks import CompositeStreamCallback, StreamCallback, TaskScopedCallback, safe_emit

from .streaming.events import CriteriaCheckEvent, CriterionSummary

from .telemetry import Scalar, hash_identifier

from .utils import get_version_info, looks_like_version, runtime_uip_versions

# Get module logger

logger = logging.getLogger(__name__)

# Grace on outer wait_for so the agent's in-band watchdog (which preserves a partial)

# wins the race against the asyncio cancel path (which doesn't).

_WAIT_FOR_GRACE_SECONDS = 2.0

async def _pump_stream(

stream: asyncio.StreamReader | None,

log_fn: Callable[..., None],

label: str,

chunks: list[str],

) -> None:

"""Read ``stream`` line-by-line, log each non-empty line via ``log_fn``,

and accumulate the raw text into ``chunks`` for later capture.

Used to forward post_run subprocess output to the orchestrator log in

real time while still preserving it for ``PostRunResult``. If a single

line exceeds the StreamReader buffer (rare — only for binary-ish or

malformed output), it is drained as a chunk and logged as a partial.

"""

if stream is None:

return

while True:

try:

raw = await stream.readline()

except asyncio.LimitOverrunError as e:

# Single line larger than the buffer; drain the buffered bytes so

# readline() can make progress on the next iteration.

raw = await stream.readexactly(e.consumed)

text = raw.decode(errors="replace")

chunks.append(text)

log_fn("[%s] (partial line, %d bytes)", label, len(raw))

continue

if not raw:

break

text = raw.decode(errors="replace")

chunks.append(text)

line = text.rstrip()

if line:

log_fn("[%s] %s", label, line)

# Structural tags emitted by ClaudeCodeAgent._format_messages. Other

# bracketed words (markdown footnotes, pylint codes, unknown SDK message types

# like [TaskStartedMessage]) are intentionally NOT matched — they pass through

# as content. Source of truth for the tag vocabulary is

# ``ClaudeCodeAgent._format_messages``; this regex is telemetry-only (utterance

# extraction for the per-task log), not a correctness-critical parser.

_UTTERANCE_TAG_RE = re.compile(r"^\[(ASSISTANT|RESULT - SUCCESS|RESULT - ERROR|TOOL USE)\](?: (.*))?$")

def _format_routing(route: ApiRoute, effective_model: str | None = None) -> str:

"""Format the route name for the ``API routing:`` log line.

For ``DirectRoute`` the resolved judge transport is appended so the choice

(anthropic / none) is visible on every run, not only in the persisted

``environment_info`` record. For ``LiteLLMRoute`` the model is shown — the

``effective_model`` (the resolved ``agent.model``, e.g. from ``--model``)

when supplied, else the route's own default — so the line reflects what the

agent will actually send rather than the route-level fallback.

"""

name = ROUTE_NAMES[type(route)]

if isinstance(route, DirectRoute):

return f"{name} (judge transport: {route.judge_transport or 'none'})"

if isinstance(route, LiteLLMRoute):

return f"{name} (model: {effective_model or route.model or 'default'})"

return name

def _extract_utterance(raw: str) -> str:

"""Collapse a ClaudeCodeAgent-formatted transcript to a clean utterance.

Input looks like:

[ASSISTANT] Sure, I'll do X.

[TOOL USE] Read

[ASSISTANT] Here is the answer...

[RESULT - SUCCESS] Here is the answer...

The SDK's ``ResultMessage`` duplicates the final assistant text, which

makes conversation.log read as if every message is repeated. Prefer the

``[RESULT - ...]`` payload when it is non-empty (it is the canonical

final utterance); otherwise fall back to concatenated ``[ASSISTANT]``

blocks. ``[TOOL USE]`` lines are dropped. Input that does not look

tagged at all (plain user text like a pinned initial_prompt) is

returned unchanged.

Pre-tag content handling: any content appearing BEFORE the first

tagged line is collected into an implicit ``[ASSISTANT]`` block. It

survives in the output only on the ASSISTANT-fallback path (no

``[RESULT - ...]`` in the transcript). When a ``[RESULT - SUCCESS]``

is present, it supersedes all ``[ASSISTANT]`` content — including

any pre-tag content — because the ResultMessage is the SDK's

canonical final utterance and ASSISTANT lines are chain-of-thought

that the RESULT already incorporates. ClaudeCodeAgent always begins

its output with a tag in practice, so this mostly matters for

defensive handling of upstream format drift.

Asymmetry note: ``[RESULT - SUCCESS]`` strips its label (it is the

canonical answer); ``[RESULT - ERROR]`` keeps a ``[RESULT - ERROR]``

prefix in the output so the error state remains visible in the log.

"""

if not raw:

return ""

lines = raw.splitlines()

if not any(_UTTERANCE_TAG_RE.match(ln) for ln in lines):

return raw

assistant_parts: list[str] = []

result_parts: list[str] = []

# Pre-tag content becomes an implicit ASSISTANT block (not dropped).

current_tag: str = "ASSISTANT"

current_buf: list[str] = []

def _flush() -> None:

text = "\n".join(current_buf).strip()

if not text:

return

if current_tag == "ASSISTANT":

assistant_parts.append(text)

elif current_tag == "RESULT - SUCCESS":

result_parts.append(text)

elif current_tag == "RESULT - ERROR":

result_parts.append(f"[RESULT - ERROR] {text}")

# TOOL USE is dropped.

for ln in lines:

match = _UTTERANCE_TAG_RE.match(ln)

if match:

_flush()

current_tag = match.group(1)

current_buf = [match.group(2) or ""]

else:

current_buf.append(ln)

_flush()

if result_parts:

return "\n\n".join(result_parts)

if assistant_parts:

return "\n\n".join(assistant_parts)

return raw

def _extract_failure_reason(result: CriterionResult) -> str | None:

"""Streaming-event wrapper around ``_short_failure_reason``.

Preserves the historical ``None``-for-no-content contract so

``CriterionSummary.failure_reason`` stays ``None`` when there's nothing

to show. The actual reason text is produced by the shared helper so the

console FAILED log and the streamed event render identical strings.

"""

if not result.error and not result.details:

return None

reason = _short_failure_reason(result)

return reason if reason != "no details" else None

def build_task_event(result: EvaluationResult, *, driver: str, variant_id: str) -> tuple[str, dict[str, Scalar]]:

"""Build the (event_name, properties) for a finalized task's telemetry event.

Shared by the in-process path (``Orchestrator._finalize_result``) and the

docker path (``orchestration/batch.py``) so both drivers emit an identical

``CoderEval.Task.End`` event. Carries only enums/counts/durations/config-derived

ids — no user content. None-safe.

Every task emits the SAME event name (``CoderEval.Task.End``); the outcome

lives in dimensions, never the name. ``Status`` carries the exact

``FinalStatus`` and ``Category`` carries the canonical ``FinalStatus.category``

bucket (``succeeded`` / ``failed`` / ``error``) — the single source of truth

shared with reports. Slicing belongs in dimensions (the App Insights idiom),

so dashboards group by ``Status``/``Category`` rather than matching event

names, and the telemetry bucketing can never drift from ``category``.

The return type is the scalar event contract (``dict[str, Scalar]``) so a

non-scalar property is caught here by pyright, not just str()-coerced at

runtime by the telemetry layer. Token counts are intentionally NOT emitted —

this is usage telemetry, not eval analytics. Task/variant ids are emitted as

stable one-way hashes (``hash_identifier``) so an author-defined free-text id

that could encode sensitive data never reaches the telemetry store verbatim.

"""

props: dict[str, Scalar] = {

"TaskId": hash_identifier(result.task_id),

"VariantId": hash_identifier(variant_id),

"Status": result.final_status.value,

"Category": result.final_status.category,

"DurationMs": int((result.duration_seconds or 0.0) * 1000),

"Score": float(result.weighted_score or 0.0),

"Iterations": result.iteration_count,

"AgentType": result.agent_type or "",

"Model": result.model_used or "",

"Driver": driver,

"EarlyStopped": result.early_stop is not None,

"EarlyStopReason": (result.early_stop.reason.value if result.early_stop is not None else ""),

}

return "CoderEval.Task.End", props

@dataclass

class _SolicitedMessage:

"""Result of asking the user simulator for one utterance (opener or in-loop).

``message is None`` signals a simulator failure (the caller bumps

``simulator_failures`` and reacts); otherwise the caller inspects

``message.stop_requested`` and folds ``sim_in``/``sim_out`` into its running

token totals. Module-private — NOT a public model.

"""

message: UserMessage | None

sim_in: int

sim_out: int

@dataclass

class _OpenerOutcome:

"""Result of the pure-simulation opener acquisition (``initial_prompt is None``).

When ``short_circuit`` is True the opener already wrote simulation telemetry and the

dialog loop must ``return return_value`` immediately (simulator failure → ERROR, or an

opener carrying the stop token → STOP_TOKEN, both before any agent turn). Otherwise the

caller folds ``sim_in``/``sim_out``/``failures`` into its counters and binds

``current_prompt``/``pending_user_turn``. Module-private — NOT a public model.

"""

short_circuit: bool

return_value: bool = False

current_prompt: str | None = None

pending_user_turn: UserMessage | None = None

sim_in: int = 0

sim_out: int = 0

failures: int = 0

class Orchestrator:

"""Coordinates the full evaluation loop for a task.

Manages the sandbox, agent, and evaluators to run a complete

task evaluation with multiple iterations.

"""

def __init__(

self,

task: TaskDefinition,

run_dir: Path,

preservation_mode: PreservationMode = PreservationMode.MOVE_ON_WRITE,

task_file: Path | None = None,

stream_callback: StreamCallback | None = None,

sandbox: Sandbox | None = None,

*,

variant_id: str,

source_yaml: str = "",

config_lineage: dict[str, ConfigLineageEntry] | None = None,

replicate_index: int = 0,

workspace_dir: Path | None = None,

):

"""Initialize the orchestrator.

Args:

task: Task definition to evaluate

run_dir: Per-task directory within a run (e.g., runs/2025-10-09_15-30-45/default/hello_date/00/)

preservation_mode: How to persist the sandbox after completion

(NONE / MOVE_ON_WRITE / DIRECT_WRITE). The driver-derived

default is resolved upstream at the batch dispatch seam.

task_file: Path to task YAML file (for resolving reference file paths)

stream_callback: Optional callback for real-time event streaming

sandbox: Pre-built Sandbox to use directly; if None, creates one from task config and runs the agent

variant_id: Experiment variant identifier for this task

source_yaml: Raw YAML text from the task file

config_lineage: Config lineage dict (dotted-path -> ConfigLineageEntry)

replicate_index: Zero-indexed trial number (for simulation tasks with n_trials > 1).

Defaults to 0, which covers single-shot tasks and single-trial simulations.

workspace_dir: Docker WORKDIR alignment. When set, the agent runs

in-place at this absolute container path (the task image's WORKDIR) instead of

run_dir/artifacts/<task>, and the workspace is copied out to run_dir/artifacts/<task>

at cleanup. Resolved host-side by DockerRunner; None keeps standard behavior.

Takes precedence over preservation_mode when set.

"""

self.task = task

self.run_dir = run_dir

# Per-attempt nonce for the LiteLLM cost-log join. The proxy log is

# append-only and the run_id is a deterministic hash of run_dir, so a

# re-run into the same --run-dir would otherwise re-match (and double-count)

# a prior attempt's rows. A fresh nonce per Orchestrator (one per process

# invocation) scopes the join to THIS attempt's records.

self._cost_attempt_nonce = uuid.uuid4().hex

self.preservation_mode = preservation_mode

self.workspace_dir = workspace_dir

self.task_file = task_file

self.stream_callback = stream_callback

self.sandbox = sandbox

self.variant_id = variant_id

self.source_yaml = source_yaml

self.config_lineage = config_lineage or {}

self.replicate_index = replicate_index

# Derived paths

self.report_path = self.run_dir / "task.json"

self.html_report_path = self.run_dir / "task.html"

# Clean user<->agent transcript for simulation runs. Written alongside

# task.log so a human can follow the conversation without the

# orchestrator noise in between.

self.conversation_log_path = self.run_dir / "conversation.log"

# Note: artifacts directory (run_dir/artifacts) is created on-demand during sandbox preservation

# Components (initialized in run())

self.agent: Agent[Any] | None = None

self.success_checker: SuccessChecker | None = None

# API routing (initialized in _setup)

self.route: ApiRoute | None = None

# Route for the evaluation side (llm_judge / agent_judge / simulated user):

# pinned to a constant Claude backend so grading stays comparable when the

# agent runs on an open-weight (LiteLLM) model. Equals self.route for the

# Direct/Bedrock backends.

self.eval_route: ApiRoute | None = None

# Result tracking

self.result: EvaluationResult | None = None

# Reference solution cache (loaded on-demand)

self._reference_code: str | None = None

# Early-stop watcher (created in _setup only when a criterion carries a

# stop_early: block and the kill switch is not thrown; None otherwise,

# so the default path is entirely unaffected).

self._early_stop_watcher: EarlyStopWatcher | None = None

# One-shot flag: emit the "cost budget configured but no cost data" warning

# exactly once per task even if _check_run_limits fires every turn.

self._cost_budget_skipped_logged: bool = False

# One-shot flag: emit the expected_turns rollup warning exactly once per

# task run even though _check_expected_turns is called after every turn.

self._expected_turns_warning_emitted: bool = False

# Canonical id shared with run_dir layout, tqdm label, and streaming events.

self._log_task_id = format_task_log_id(variant_id, task.task_id, replicate_index)

@property

def _agent_name(self) -> str:

"""Agent name for error-context telemetry.

The agent kind string for any resolved task (``"none"`` for no-op tasks);

falls back to ``"none"`` only on the evaluate-only path, where no agent

is attached. ``str(...)`` handles both built-in ``AgentKind`` members and

plugin-registered raw-string kinds.

"""

if self.task.agent is not None and self.task.agent.type is not None:

return str(self.task.agent.type)

return AgentKind.NONE.value

async def run(self) -> EvaluationResult:

"""Run the complete evaluation.

Returns:

Evaluation result with all details

Raises:

RuntimeError: If evaluation fails catastrophically

"""

from .logging_config import task_log_handler

# Agent must be resolved before reaching the orchestrator. No-op (type: none)

# tasks are resolved to a NoneAgentConfig like any other agent, so there is

# no separate "no agent" branch here.

assert self.task.agent is not None, (

f"Task '{self.task.task_id}' has no agent config. Ensure experiment resolution ran before orchestration."

)

assert self.task.agent.type is not None, (

f"Task '{self.task.task_id}' has no agent.type. Ensure experiment resolution + CLI overrides "

"ran before orchestration."

)

agent_type = self.task.agent.type

start_time = time.time()

started_at = datetime.now()

# Initialize result

self.result = EvaluationResult(

task_id=self.task.task_id,

task_description=self.task.description,

variant_id=self.variant_id,

agent_type=agent_type,

started_at=started_at,

final_status=FinalStatus.FAILURE, # Will be updated

iteration_count=0,

environment_info=get_version_info(),

)

# Calculate task log path

task_log_file = task_log_path(self.run_dir)

task_log_file.parent.mkdir(parents=True, exist_ok=True) # noqa: CE002 — mkdir on local FS is nanoseconds

# Use context manager for automatic log handler management

with task_log_handler(task_log_file, task_id=self._log_task_id) as log_tail:

try:

# Setup components

await self._setup()

# Run pre-run commands inside the sandbox before the agent starts.

# A failing command with fail_on_error=True raises RuntimeError,

# which propagates to the outer except Exception below and lands

# the run as FinalStatus.ERROR; _run_post_run_commands and

# _cleanup still execute via the finally block.

await self._run_pre_run_commands()

# Enforce task-level timeout via an OS-thread watchdog that

# SIGKILLs the in-flight CLI subprocess AND cancels this

# task. The threaded approach is immune to anyio cancel

# scopes that were silently swallowing asyncio.wait_for

# cancellations during long rate-limited API calls.

task_timeout = self.task.run_limits.task_timeout if self.task.run_limits else None

def _kill_agent_subprocess_sync() -> None:

if self.agent is not None:

# kill_sync is a synchronous SIGKILL-by-PID, safe to

# call from a non-asyncio thread.

with suppress(Exception):

self.agent.kill_sync()

with ThreadedWatchdog(

timeout_seconds=task_timeout,

on_timeout=_kill_agent_subprocess_sync,

asyncio_task_to_cancel=asyncio.current_task(),

label=f"task_timeout ({self.task.task_id})",

) as wd:

try:

success = await self._evaluation_loop()

except asyncio.CancelledError:

if wd.fired:

raise TaskTimeoutError(

task_timeout or 0,

task_id=self.task.task_id,

elapsed_seconds=time.time() - start_time,

) from None

raise

# Belt-and-suspenders: if the loop returned normally but the

# watchdog fired during post-loop work or the inner coro

# swallowed the cancel, still classify as TIMEOUT.

if wd.fired and task_timeout is not None:

raise TaskTimeoutError(

task_timeout,

task_id=self.task.task_id,

elapsed_seconds=time.time() - start_time,

)

# Update final status

if success:

self.result.final_status = FinalStatus.SUCCESS

elif self.result.max_turns_exhausted:

self.result.final_status = FinalStatus.MAX_TURNS_EXHAUSTED

else:

self.result.final_status = FinalStatus.FAILURE

except asyncio.CancelledError:

# Re-raise cancellation to allow proper task cancellation

raise

except TaskTimeoutError as e:

# Task-level timeout gets a dedicated status (not generic ERROR)

self.result.final_status = FinalStatus.TIMEOUT

self.result.error_message = str(e)

self.result.error_details = create_error_context(

error=e,

task_id=self.task.task_id,

attempt=max(self.result.iteration_count, 1),

component="orchestrator.task_timeout",

agent_name=self._agent_name,

)

logger.error(f"Task timed out: {e}")

# Recover the turn in flight when the watchdog killed the agent.

# Nothing else on this path does: the cancel arrives as a

# BaseException, so it never reaches the retry executor's

# per-attempt hook that drains the slot on a turn-level timeout.

await self._drain_killed_turn()

except BudgetExceededError as e:

# Map token-budget breaches and cost-budget breaches to distinct

# statuses so per-task records preserve the failure mode.

if e.budget_name == "usd":

self.result.final_status = FinalStatus.COST_BUDGET_EXCEEDED

component = "orchestrator.run_limits.cost"

else:

self.result.final_status = FinalStatus.TOKEN_BUDGET_EXCEEDED

component = "orchestrator.run_limits.tokens"

self.result.error_message = str(e)

self.result.error_details = create_error_context(

error=e,

task_id=self.task.task_id,

attempt=max(self.result.iteration_count, 1),

component=component,

agent_name=self._agent_name,

)

logger.warning(f"Run limit exceeded: {e}")

except Exception as e:

# Handle catastrophic errors

self.result.final_status = FinalStatus.ERROR

self.result.error_message = str(e)

# Determine which component failed (setup vs. iteration N)

if self.result.iteration_count == 0:

failed_component = "orchestrator.setup"

else:

failed_component = f"orchestrator.iteration_{self.result.iteration_count}"

# Capture detailed error context

self.result.error_details = create_error_context(

error=e,

task_id=self.task.task_id,

attempt=max(self.result.iteration_count, 1), # Actual iteration attempt (1-indexed)

component=failed_component,

agent_name=self._agent_name,

)

logger.error(f"Evaluation failed: {e}", exc_info=True)

finally:

# Teardown must be interrupt-proof: the task-timeout watchdog can

# fire while post-run commands are awaiting and deliver its

# CancelledError right here in the finally block, which used to

# abort it wholesale — skipping _cleanup() (tempdir leaked) AND

# _finalize_result() (task.json lost, so the task silently drops

# out of the run). Catch the interrupt, finish the full teardown,

# then re-raise it at the end so callers observe the same exception

# as before. The watchdog cancels exactly once, so the teardown

# awaits below run normally after the CancelledError is caught.

teardown_interrupt: BaseException | None = None

try:

# BEFORE post-run/cleanup: needs the live sandbox to resolve

# the agent-aligned `uip`, and post-task tool state on disk.

self._refresh_runtime_tool_versions()

await self._run_post_run_commands()

except (Exception, asyncio.CancelledError) as e:

teardown_interrupt = e

logger.warning(

"Teardown interrupted during post-run (%s: %s); completing cleanup before re-raising",

type(e).__name__,

e,

)

await self._cleanup()

# Capture the sanitised log tail AFTER teardown so any errors

# logged during post-run / cleanup also land in the report,

# but BEFORE _finalize_result so task.json includes the field.

# Allowlist non-success terminal statuses; SUCCESS and

# MAX_TURNS_EXHAUSTED skip the tail to keep task.json compact.

if self.result.final_status in {

FinalStatus.ERROR,

FinalStatus.TIMEOUT,

FinalStatus.FAILURE,

FinalStatus.TOKEN_BUDGET_EXCEEDED,

FinalStatus.COST_BUDGET_EXCEEDED,

}:

self.result.error_log_tail = log_tail.get_text() or None

self._finalize_result(start_time)

if teardown_interrupt is not None:

raise teardown_interrupt

return self.result

async def _drain_killed_turn(self) -> None:

"""Move a hard-killed turn's partial record from the agent onto the result.

The only reader of ``pending_turn`` on the task-timeout path. Ordering

matters both ways: it must run before ``_cleanup`` (whose ``agent.stop()``

clears the slot) and before ``_finalize_result``, so the recovered turn

feeds token aggregation and command stats like any other.

Best-effort: a task killed before its first turn has nothing parked, and

this runs on the way to a saved row, so it must not raise.

"""

if self.agent is None or self.result is None:

return

try:

partial = self.agent.pending_turn

# `pending_turn` is a slot any agent implementation fills, so a non-record

# here would fail validation during teardown and take the row down with it.

if not isinstance(partial, TurnRecord):

logger.debug("[%s] Hard-killed task preserved no partial turn", self.task.task_id)

return

self.result.iterations.append(partial)

await self.agent.discard_pending_turn()

usage = partial.token_usage

logger.info(

"[%s] Recovered the hard-killed turn: %d tokens, %s",

self.task.task_id,

usage.total_tokens if usage is not None else 0,

f"${usage.total_cost_usd:.4f}"

if usage is not None and usage.total_cost_usd is not None

else "unpriced",

)

except Exception:

logger.warning("[%s] Could not recover the hard-killed turn", self.task.task_id, exc_info=True)

def _finalize_result(self, start_time: float) -> None:

"""Finalize the evaluation result: scores, telemetry, and persistence."""

if not self.result:

return

# Every resolved task carries an agent config (no-op tasks resolve to a

# NoneAgentConfig); a missing one is a resolution bug. The evaluate-only

# path doesn't reach here with task.agent unset.

if self.task.agent is None:

logger.error("Cannot finalize result: task.agent is None")

return

self.result.completed_at = datetime.now()

self.result.duration_seconds = time.time() - start_time

# Weighted score. This call site is wrapped because _finalize_result runs

# inside run()'s finally — an unguarded raise here would skip persistence and

# lose task.json. The other calculate_weighted_score calls (the simulation

# path) run inside run()'s try, whose broad `except Exception` already converts

# a raise into a populated ERROR result, so they intentionally stay unwrapped.

try:

self.result.calculate_weighted_score(self.task.success_criteria)

except ValueError as e:

logger.error("Weighted-score computation failed; marking row ERROR: %s", e, exc_info=True)

self.result.weighted_score = None

self.result.final_status = FinalStatus.ERROR

self.result.error_message = str(e)

self.result.error_details = create_error_context(

error=e,

task_id=self.task.task_id,

attempt=max(self.result.iteration_count, 1),

component="orchestrator.finalize.weighted_score",

agent_name=self._agent_name,

)

# Command statistics

if self.result.iterations:

self.result.command_stats = calculate_command_statistics(self.result.iterations)

# Resolve model_used (last turn with model wins, then agent config)

if self.result.iterations:

for turn in reversed(self.result.iterations):

if turn.model_used:

self.result.model_used = turn.model_used

break

if not self.result.model_used and self.task.agent is not None and self.task.agent.model:

self.result.model_used = self.task.agent.model

# Open-weight (LiteLLM) backend: replace per-turn cost with the ACTUAL

# per-call OpenRouter cost captured proxy-side. Runs BEFORE aggregation so

# the run total re-derives from the corrected per-turn costs.

self._join_litellm_actual_cost()

# Aggregate token usage

self._aggregate_token_usage()

# Record whether per-turn cost data was available when a cost budget was set.

# Lets users audit whether a configured max_usd budget was actually enforceable.

if self.task.run_limits is not None and self.task.run_limits.max_usd is not None:

any_cost_reported = any(

t.token_usage is not None and t.token_usage.total_cost_usd is not None for t in self.result.iterations

)

self.result.environment_info["cost_data_available"] = any_cost_reported

if self.result.iterations:

self.result.total_assistant_turns = sum(t.assistant_turn_count for t in self.result.iterations)

# command_stats counts crashed partials too — they're real executed work.

if self.result.iterations and self.result.command_stats:

actual_cmds = self.result.command_stats.total_commands

self.result.actual_commands = actual_cmds

if self.task.expected_commands is not None:

self.result.expected_commands = self.task.expected_commands

self.result.commands_efficiency = compute_commands_efficiency(actual_cmds, self.task.expected_commands)

# SDK options snapshot

if self.agent:

self.result.sdk_options = self.agent.get_sdk_options()

# Task config record (warnings=False: discriminated unions produce benign warnings)

self.result.task_config = TaskConfigRecord(

resolved=self.task.model_dump(warnings=False),

source_yaml=self.source_yaml,

source_file=str(self.task_file) if self.task_file else None,

lineage=self.config_lineage,

)

# Terminal per-task summary line. Emitted before report writes so a

# write failure cannot swallow the one-line outcome.

logger.info(

"Task finished: status=%s duration=%.1fs score=%.3f iterations=%d",

self.result.final_status.value,

self.result.duration_seconds or 0.0,

self.result.weighted_score or 0.0,

self.result.iteration_count,

)

# Usage telemetry (non-fatal; placed before persistence since track_event

# cannot raise). For the docker driver this in-process emit runs INSIDE the

# container where telemetry is off (the connection-string env vars aren't

# forwarded), so the host emits the event from batch.py instead — see

# build_task_event. Non-docker tasks finalize on the host and emit here.

from .telemetry import track_event

driver = self.task.sandbox.driver if self.task.sandbox else ""

name, props = build_task_event(self.result, driver=driver, variant_id=self.variant_id or "")

track_event(name, props)

# Persist

self.report_path.parent.mkdir(parents=True, exist_ok=True) # noqa: CE002 — mkdir on local FS is nanoseconds

# Spill any judge transcripts to sibling judge-<idx>.yaml files BEFORE

# we dump task.json, so transcript_path is set on each judge result.

# The inline `transcript` field stays in memory — HTML rendering below

# uses it directly. We strip it from the JSON dump via `exclude=...`.

from .evaluation.judge_persistence import spill_judge_transcripts

spill_judge_transcripts(self.result, self.report_path.parent)

# Atomic write: tmp file + os.replace. A SIGKILL mid-write (e.g. the

# docker-driver host-heartbeat watchdog firing) would otherwise leave

# a truncated task.json that the host parses as malformed-JSON rather

# than as "no result", conflating two distinct failure modes.

import os as _os

report_tmp = self.report_path.with_suffix(self.report_path.suffix + ".tmp")

report_tmp.write_text( # noqa: CE002 — small JSON write at end of run

self.result.model_dump_json(

indent=2,

# Strip inline transcripts: they live in sibling judge-<idx>.yaml

# next to task.json, referenced by transcript_path. Excluding

# `transcript` here avoids ~20-100 KB of bloat per judge result

# in the row record without losing any data.

exclude={"success_criteria_results": {"__all__": {"transcript"}}},

),

encoding="utf-8",

)

_os.replace(report_tmp, self.report_path)

# Also emit an HTML trace/report alongside task.json. HTML failure must

# never mask the underlying run outcome — write_task_html logs and

# returns None on failure.

from .reports_html import write_task_html

write_task_html(self.result, self.html_report_path)

def _check_run_limits(self, *, iteration: int) -> None:

"""Raise BudgetExceededError if any RunLimits budget is exceeded.

Called after each completed turn. Aggregates across self.result.iterations.

No-op when self.task.run_limits is None.

"""

assert self.result is not None

limits = self.task.run_limits

if limits is None:

return

usages = [t.token_usage for t in self.result.iterations if t.token_usage is not None]

if not usages:

return

input_tokens = sum(u.uncached_input_tokens for u in usages)

if limits.count_cache_creation:

input_tokens += sum(u.cache_creation_input_tokens for u in usages)

if limits.count_cached_input:

input_tokens += sum(u.cache_read_input_tokens for u in usages)

output_tokens = sum(u.output_tokens for u in usages)

total_tokens = input_tokens + output_tokens

if limits.max_input_tokens is not None and input_tokens > limits.max_input_tokens:

raise BudgetExceededError(

"input_tokens",

actual=input_tokens,

limit=limits.max_input_tokens,

task_id=self.task.task_id,

iteration=iteration,

)

if limits.max_output_tokens is not None and output_tokens > limits.max_output_tokens:

raise BudgetExceededError(

"output_tokens",

actual=output_tokens,

limit=limits.max_output_tokens,

task_id=self.task.task_id,

iteration=iteration,

)

if limits.max_total_tokens is not None and total_tokens > limits.max_total_tokens:

raise BudgetExceededError(

"total_tokens",

actual=total_tokens,

limit=limits.max_total_tokens,

task_id=self.task.task_id,

iteration=iteration,

)

if limits.max_usd is not None:

costs = [u.total_cost_usd for u in usages if u.total_cost_usd is not None]

if not costs:

if not self._cost_budget_skipped_logged:

logger.warning(

"[%s] max_usd budget configured but no turn reported cost; skipping cost check",

self.task.task_id,

)

self._cost_budget_skipped_logged = True

return

total_cost = sum(costs)

if total_cost > limits.max_usd:

raise BudgetExceededError(

"usd",

actual=total_cost,

limit=limits.max_usd,

task_id=self.task.task_id,

iteration=iteration,

)

def _check_expected_turns(self, *, iteration: int) -> None:

"""Emit a one-shot warning if visible turns exceed expected_turns.

Soft sibling of ``_check_run_limits.max_turns``: never aborts the run.

``max_turns`` remains the hard cap (enforced inside the SDK). A

"turn" here is one timeline entry: each tool call plus the final

reply when present — the same metric evalboard renders. Cumulative

across iterations so simulation/dialog tasks compare against the

budget the user set.

"""

if self.result is None:

return

limits = self.task.run_limits

if limits is None or limits.expected_turns is None:

return

if self._expected_turns_warning_emitted:

return

from .reports_stats import visible_turn_count

total = visible_turn_count(self.result)

if total > limits.expected_turns:

logger.warning(

"Visible turns (%d) exceeded expected_turns (%d) at iteration %d "

+ "for task %s. Run continues — max_turns remains the hard cap.",

total,

limits.expected_turns,

iteration,

self.task.task_id,

)

self._expected_turns_warning_emitted = True

@property

def _cost_correlation_run_id(self) -> str:

"""The LiteLLM cost-log correlation run id — a stable hash of the run dir.

Single accessor used by BOTH the stamp site (``_create_agent``, into

``x-ce-run-id``) and the join site (``_join_litellm_actual_cost``); keeping

the derivation in one place means the two can't drift and silently revert

every turn to static pricing.

"""

return hash_identifier(self.run_dir.as_posix())

def _join_litellm_actual_cost(self) -> None:

"""Override per-turn cost with the proxy-captured ACTUAL per-call OpenRouter

cost (and attach the per-call cache breakdown) for the open-weight backend.

No-op unless the agent ran on a ``LiteLLMRoute`` AND ``LITELLM_COST_LOG`` is

configured. Never fatal: a failure, or an empty/mismatched log, leaves each

turn's static rate-card estimate in place (the whole-turn fallback).

"""

if not (isinstance(self.route, LiteLLMRoute) and settings.litellm_cost_log and self.result is not None):

return

try:

applied = apply_actual_cost(

self.result,

run_id=self._cost_correlation_run_id,

task_id=self._log_task_id,

attempt=self._cost_attempt_nonce,

records=load_cost_records(settings.litellm_cost_log),

)

if applied:

logger.info("LiteLLM actual-cost join: real per-call cost applied to %d turn(s)", applied)

else:

# Tags were stamped but nothing matched (file absent, proxy never

# wrote, wrong path, or a run/task/attempt mismatch). The run stays

# on the static rate card — warn so it isn't mistaken for the real bill.

logger.warning(

"LiteLLM actual-cost join found no matching records in %s (run=%s task=%s); cost stays static",

settings.litellm_cost_log,

self._cost_correlation_run_id,

self._log_task_id,

)

except Exception:

logger.warning("LiteLLM actual-cost join failed; keeping static pricing", exc_info=True)

def _aggregate_token_usage(self) -> None:

"""Aggregate token usage from turns, storing on self.result.

Per-turn ``TurnRecord.token_usage`` is filled by the agent SDK's

ResultMessage on DirectRoute / BedrockRoute (the bundled CLI parses

the response's ``usage`` field). Every iteration carries the right

per-turn value here, and we just sum them. Judge / sub-agent token

usage is captured on

the corresponding ``JudgeCriterionResult.token_usage`` and is

intentionally NOT included in this aggregate — it represents the

main agent's bill, not the eval-machinery overhead.

"""

assert self.result is not None

# Include crashed=True partials: each API call is billed independently.

if self.result.iterations:

usages = [t.token_usage for t in self.result.iterations if t.token_usage is not None]

if usages:

costs = [u.total_cost_usd for u in usages if u.total_cost_usd is not None]

self.result.total_token_usage = TokenUsage(

uncached_input_tokens=sum(u.uncached_input_tokens for u in usages),

output_tokens=sum(u.output_tokens for u in usages),

cache_creation_input_tokens=sum(u.cache_creation_input_tokens for u in usages),

cache_read_input_tokens=sum(u.cache_read_input_tokens for u in usages),

total_cost_usd=sum(costs) if costs else None,

)

async def _setup(self) -> None:

"""Set up all components for evaluation.

Raises:

RuntimeError: If setup fails

"""

# Defensive early-stop guardrails for the library-use and in-container

# paths (the CLI already validated during resolution). No-op unless

# some criterion carries a stop_early: block.

validate_early_stop(self.task)

# Build the early-stop watcher once, up front, when armed (>= 1 criterion

# with a stop_early: block and the run_limits.stop_early kill switch not

# thrown). This sits BEFORE the evaluate-only early return below, so an

# armed evaluate-only re-grade builds an inert (never-fed) watcher —

# harmless, and keeps a single creation point.

if early_stop_active(self.task):

self._early_stop_watcher = EarlyStopWatcher.for_task(self.task)

if self.sandbox is not None: