reach.metrics

Compute standard evaluation, classification, routing, and multi-step trajectory metrics for skill selection probes.

---

Multi-Step Trajectory Routing

Modern agent runtimes execute multi-turn conversational trajectories where a task may require entering via one skill, performing a precursor handoff, and terminating at the target capability:

flowchart LR
    Query["User Query"] --> E["Step 1: Entrypoint Skill<br/>(Entrypoint Accuracy)"]
    E -->|Handoff| P["Step 2: Precursor Skill"]
    P -->|Handoff| T["Step 3: Target Capability<br/>(Trajectory Reachability)"]

---

Trajectory Metrics

Reach evaluates multi-step trajectories against the ground-truth capability target \(T\):

Metric	Symbol	Definition & Meaning
Entrypoint Accuracy	\(A_{\text{entry}}\)	Fraction of queries where the very first skill invoked matches the target skill \(T\) (\(\vec{s}_1 = T\)). Penalizes misrouted initial dispatch.
Trajectory Reachability	\(R_{\text{traj}}\)	Fraction of queries where the target skill \(T\) is reached anywhere in the trajectory (\(T \in \vec{s}\)).
Step Efficiency (MRR)	\(\text{MRR}\)	Reciprocal rank \(\frac{1}{\text{step}}\), measuring how directly and promptly the agent invoked the target skill.
Skill Selection F1	\(F_1\)	Harmonic mean of precision (target reached / total unique skills invoked) and recall (target reached).
Skill Redundancy	\(\text{Redundancy}\)	Excess invocations beyond the target: \(\max(0, \text{len}(\vec{s}) - 1)\). Zero indicates optimal, direct execution.

---

API Reference

Compute standard evaluation and routing metrics for skill selection probes.

ClassificationReport

Bases: BaseModel

Hold aggregated classification and routing metrics for an evaluation run.

Source code in src/reach/metrics.py

class ClassificationReport(BaseModel):
    """Hold aggregated classification and routing metrics for an evaluation run."""

    model_config = ConfigDict(frozen=True)

    errors: int
    macro_f1: float
    macro_precision: float
    macro_recall: float
    per_class: tuple[ClassMetrics, ...]
    probes: int
    scored: int
    abstentions: int = 0
    false_abstentions: int = 0
    in_scope: int = 0
    out_of_scope: int = 0
    out_of_scope_detected: int = 0
    top1_hits: int = 0
    entrypoint_hits: int = 0
    trajectory_hits: int = 0
    step_efficiency: float = 0.0
    skill_f1: float = 0.0
    redundancy: float = 0.0

    @computed_field
    @property
    def entrypoint_accuracy(self) -> float:
        """Calculate entrypoint accuracy across scored probes."""
        return self.entrypoint_hits / self.scored if self.scored else 0.0

    @computed_field
    @property
    def entrypoint_interval(self) -> Interval | None:
        """Calculate the Wilson confidence interval for entrypoint accuracy."""
        return wilson_interval(self.entrypoint_hits, self.scored)

    @computed_field
    @property
    def trajectory_reachability(self) -> float:
        """Calculate trajectory reachability across scored probes."""
        return self.trajectory_hits / self.scored if self.scored else 0.0

    @computed_field
    @property
    def trajectory_interval(self) -> Interval | None:
        """Calculate the Wilson confidence interval for trajectory reachability."""
        return wilson_interval(self.trajectory_hits, self.scored)

    @computed_field
    @property
    def top1_accuracy(self) -> float:
        """Calculate top-1 accuracy across scored probes."""
        return self.top1_hits / self.scored if self.scored else 0.0

    @computed_field
    @property
    def top1_interval(self) -> Interval | None:
        """Calculate the Wilson confidence interval for top-1 accuracy."""
        return wilson_interval(self.top1_hits, self.scored)

    @computed_field
    @property
    def abstention_rate(self) -> float:
        """Calculate the overall abstention rate across scored probes."""
        return self.abstentions / self.scored if self.scored else 0.0

    @computed_field
    @property
    def abstention_interval(self) -> Interval | None:
        """Calculate the Wilson confidence interval for overall abstention rate."""
        return wilson_interval(self.abstentions, self.scored)

    @computed_field
    @property
    def false_abstention_rate(self) -> float:
        """Calculate the false abstention rate over in-scope queries."""
        return self.false_abstentions / self.in_scope if self.in_scope else 0.0

    @computed_field
    @property
    def false_abstention_interval(self) -> Interval | None:
        """Calculate the Wilson confidence interval for false abstention rate."""
        return wilson_interval(self.false_abstentions, self.in_scope)

    @computed_field
    @property
    def out_of_scope_detection(self) -> float | None:
        """Calculate the out-of-scope detection accuracy."""
        if not self.out_of_scope:
            return None
        return self.out_of_scope_detected / self.out_of_scope

    @computed_field
    @property
    def out_of_scope_interval(self) -> Interval | None:
        """Calculate the Wilson confidence interval for out-of-scope detection."""
        return wilson_interval(self.out_of_scope_detected, self.out_of_scope)

    def by_label(self, label: str) -> ClassMetrics:
        """Return per-class metrics for a specified label."""
        for entry in self.per_class:
            if entry.label == label:
                return entry
        msg = f"no metrics for label {label!r}"
        raise KeyError(msg)

    def worst_recall(self, limit: int = 5) -> tuple[ClassMetrics, ...]:
        """Return classes with the lowest recall scores."""
        real = [c for c in self.per_class if c.label != NO_SKILL and c.support]
        return tuple(sorted(real, key=lambda c: (c.recall, c.label))[:limit])

    def top_attractors(self, limit: int = 5) -> tuple[ClassMetrics, ...]:
        """Return classes receiving the highest false-positive traffic."""
        real = [c for c in self.per_class if c.label != NO_SKILL and c.false_positives]
        return tuple(sorted(real, key=lambda c: (-c.false_positives, c.label))[:limit])

abstention_interval property

abstention_interval: Interval | None

Calculate the Wilson confidence interval for overall abstention rate.

abstention_rate property

abstention_rate: float

Calculate the overall abstention rate across scored probes.

entrypoint_accuracy property

entrypoint_accuracy: float

Calculate entrypoint accuracy across scored probes.

entrypoint_interval property

entrypoint_interval: Interval | None

Calculate the Wilson confidence interval for entrypoint accuracy.

false_abstention_interval property

false_abstention_interval: Interval | None

Calculate the Wilson confidence interval for false abstention rate.

false_abstention_rate property

false_abstention_rate: float

Calculate the false abstention rate over in-scope queries.

out_of_scope_detection property

out_of_scope_detection: float | None

Calculate the out-of-scope detection accuracy.

out_of_scope_interval property

out_of_scope_interval: Interval | None

Calculate the Wilson confidence interval for out-of-scope detection.

top1_accuracy property

top1_accuracy: float

Calculate top-1 accuracy across scored probes.

top1_interval property

top1_interval: Interval | None

Calculate the Wilson confidence interval for top-1 accuracy.

trajectory_interval property

trajectory_interval: Interval | None

Calculate the Wilson confidence interval for trajectory reachability.

trajectory_reachability property

trajectory_reachability: float

Calculate trajectory reachability across scored probes.

by_label

by_label(label: str) -> ClassMetrics

Return per-class metrics for a specified label.

Source code in src/reach/metrics.py

def by_label(self, label: str) -> ClassMetrics:
    """Return per-class metrics for a specified label."""
    for entry in self.per_class:
        if entry.label == label:
            return entry
    msg = f"no metrics for label {label!r}"
    raise KeyError(msg)

top_attractors

top_attractors(limit: int = 5) -> tuple[ClassMetrics, ...]

Return classes receiving the highest false-positive traffic.

Source code in src/reach/metrics.py

def top_attractors(self, limit: int = 5) -> tuple[ClassMetrics, ...]:
    """Return classes receiving the highest false-positive traffic."""
    real = [c for c in self.per_class if c.label != NO_SKILL and c.false_positives]
    return tuple(sorted(real, key=lambda c: (-c.false_positives, c.label))[:limit])

worst_recall

worst_recall(limit: int = 5) -> tuple[ClassMetrics, ...]

Return classes with the lowest recall scores.

Source code in src/reach/metrics.py

def worst_recall(self, limit: int = 5) -> tuple[ClassMetrics, ...]:
    """Return classes with the lowest recall scores."""
    real = [c for c in self.per_class if c.label != NO_SKILL and c.support]
    return tuple(sorted(real, key=lambda c: (c.recall, c.label))[:limit])

ClassMetrics

Bases: BaseModel

Report precision, recall, support, and F1 metrics for a single skill class.

Source code in src/reach/metrics.py

class ClassMetrics(BaseModel):
    """Report precision, recall, support, and F1 metrics for a single skill class."""

    model_config = ConfigDict(frozen=True)

    false_negatives: int
    false_positives: int
    label: str
    predicted: int
    support: int
    true_positives: int
    trajectory_true_positives: int = 0

    @model_validator(mode="after")
    def _ensure_trajectory_at_least_top1(self) -> Self:
        """Ensure trajectory true positives are at least top-1 true positives."""
        if self.trajectory_true_positives < self.true_positives:
            object.__setattr__(self, "trajectory_true_positives", self.true_positives)
        return self

    @property
    def precision(self) -> float:
        """Calculate precision (true positives / predicted positives)."""
        denominator = self.true_positives + self.false_positives
        return self.true_positives / denominator if denominator else 0.0

    @property
    def recall(self) -> float:
        """Calculate recall (true positives / ground truth support)."""
        denominator = self.true_positives + self.false_negatives
        return self.true_positives / denominator if denominator else 0.0

    @property
    def trajectory_recall(self) -> float:
        """Calculate trajectory recall (trajectory true positives / ground truth support)."""
        return self.trajectory_true_positives / self.support if self.support else 0.0

    @property
    def f1(self) -> float:
        """Calculate F1 score (harmonic mean of precision and recall)."""
        return compute_f1(self.precision, self.recall)

    @property
    def recall_interval(self) -> Interval | None:
        """Calculate the Wilson confidence interval for recall."""
        return wilson_interval(
            self.true_positives,
            self.true_positives + self.false_negatives,
        )

    @property
    def precision_interval(self) -> Interval | None:
        """Calculate the Wilson confidence interval for precision."""
        return wilson_interval(
            self.true_positives,
            self.true_positives + self.false_positives,
        )

f1 property

f1: float

Calculate F1 score (harmonic mean of precision and recall).

precision property

precision: float

Calculate precision (true positives / predicted positives).

precision_interval property

precision_interval: Interval | None

Calculate the Wilson confidence interval for precision.

recall property

recall: float

Calculate recall (true positives / ground truth support).

recall_interval property

recall_interval: Interval | None

Calculate the Wilson confidence interval for recall.

trajectory_recall property

trajectory_recall: float

Calculate trajectory recall (trajectory true positives / ground truth support).

DecompositionResult

Bases: BaseModel

Represent decomposition of pass-rate drop between baseline and scaled catalogs.

Source code in src/reach/metrics.py

class DecompositionResult(BaseModel):
    """Represent decomposition of pass-rate drop between baseline and scaled catalogs."""

    model_config = ConfigDict(frozen=True)

    baseline_pass_rate: float
    scaled_pass_rate: float
    delta_total: float
    delta_context: float
    delta_shadowing: float
    delta_total_ci: tuple[float, float] = (0.0, 0.0)
    delta_context_ci: tuple[float, float] = (0.0, 0.0)
    delta_shadowing_ci: tuple[float, float] = (0.0, 0.0)
    sample_size: int = 0
    baseline_ci: tuple[float, float] = (0.0, 0.0)
    scaled_ci: tuple[float, float] = (0.0, 0.0)

PrecursorEdge

Bases: BaseModel

Represent an empirical directed transition between two skills in a trajectory.

Source code in src/reach/metrics.py

class PrecursorEdge(BaseModel):
    """Represent an empirical directed transition between two skills in a trajectory."""

    model_config = ConfigDict(frozen=True)

    precursor: str
    target: str
    attempts: int
    handoffs: int
    handoff_rate: float
    avg_step_latency: float
    is_declared_dependency: bool = False

TrajectoryScore

Bases: BaseModel

Evaluation outcomes for a single query across its invocation trajectory.

Source code in src/reach/metrics.py

class TrajectoryScore(BaseModel):
    """Evaluation outcomes for a single query across its invocation trajectory."""

    model_config = ConfigDict(frozen=True)

    entrypoint_hit: bool
    trajectory_hit: bool
    step_efficiency: float
    skill_f1: float
    redundancy: int

classification_report

classification_report(
    results: Sequence[ProbeResult],
    queries: Sequence[Query],
    labels: Sequence[str] | None = None,
) -> ClassificationReport

Generate a comprehensive classification report across all probe results.

Source code in src/reach/metrics.py

def classification_report(
    results: Sequence[ProbeResult],
    queries: Sequence[Query],
    labels: Sequence[str] | None = None,
) -> ClassificationReport:
    """Generate a comprehensive classification report across all probe results."""
    pairs = _paired(results, queries)
    y_true = [q.truth_label for q, _ in pairs]
    y_pred = [q.effective_predicted_label(r) for q, r in pairs]
    universe = _label_universe(y_true, y_pred, labels)
    traj_scores = [score_trajectory(q, r.invoked_skills) for q, r in pairs]
    traj_hits_by_label = Counter(
        t for t, s in zip(y_true, traj_scores, strict=True) if s.trajectory_hit
    )
    per_class = _build_per_class(
        y_true,
        y_pred,
        universe,
        trajectory_hits_by_label=traj_hits_by_label,
    )
    precision, recall, f1 = _macro_averages(per_class)
    in_count, false_abs, out_count, out_detected = _scope_metrics(y_true, y_pred)

    entrypoint_hits = sum(1 for s in traj_scores if s.entrypoint_hit)
    trajectory_hits = sum(1 for s in traj_scores if s.trajectory_hit)
    step_eff = _mean([s.step_efficiency for s in traj_scores])
    s_f1 = _mean([s.skill_f1 for s in traj_scores])
    redundancy = _mean([float(s.redundancy) for s in traj_scores])

    return ClassificationReport(
        probes=len(results),
        errors=sum(1 for r in results if r.error),
        scored=len(y_true),
        per_class=per_class,
        top1_hits=sum(t == p for t, p in zip(y_true, y_pred, strict=True)),
        entrypoint_hits=entrypoint_hits,
        trajectory_hits=trajectory_hits,
        step_efficiency=round(step_eff, 4),
        skill_f1=round(s_f1, 4),
        redundancy=round(redundancy, 4),
        macro_precision=precision,
        macro_recall=recall,
        macro_f1=f1,
        abstentions=sum(p == NO_SKILL for p in y_pred),
        in_scope=in_count,
        false_abstentions=false_abs,
        out_of_scope=out_count,
        out_of_scope_detected=out_detected,
    )

classify_invocation_pattern

classify_invocation_pattern(
    query: Query, invoked_skills: Sequence[str]
) -> InvocationPattern

Classify trajectory invocation behavior relative to query ground truth.

Source code in src/reach/metrics.py

def classify_invocation_pattern(
    query: Query,
    invoked_skills: Sequence[str],
) -> InvocationPattern:
    """Classify trajectory invocation behavior relative to query ground truth."""
    invoked = query.scored_invocations(invoked_skills)
    if query.is_out_of_scope:
        return (
            InvocationPattern.CORRECT_ABSTENTION
            if not invoked
            else InvocationPattern.UNWANTED_TRIGGER
        )
    if not invoked:
        return InvocationPattern.ABANDONED
    valid_targets = query.valid_skills
    if set(invoked) <= valid_targets:
        return InvocationPattern.ORACLE_ONLY
    if set(invoked) & valid_targets:
        return InvocationPattern.MIXED_ORACLE
    return InvocationPattern.DISTRACTOR_HIJACK

collisions

collisions(
    results: Sequence[ProbeResult], queries: Sequence[Query]
) -> Counter[tuple[str, str]]

Count misroutes between skill pairs.

Source code in src/reach/metrics.py

def collisions(
    results: Sequence[ProbeResult],
    queries: Sequence[Query],
) -> Counter[tuple[str, str]]:
    """Count misroutes between skill pairs."""
    truth = {q.id: q for q in queries}
    pairs: Counter[tuple[str, str]] = Counter()
    for result in results:
        if result.error or not result.selected:
            continue
        query = truth.get(result.query_id)
        if query is None:
            continue
        effective_invoked = query.effective_invoked_skill(result)
        if effective_invoked is None or query.matches_skill(effective_invoked):
            continue
        pairs[(query.truth_label, effective_invoked)] += 1
    return pairs

compute_f1

compute_f1(precision: float, recall: float) -> float

Calculate the harmonic mean of precision and recall.

Source code in src/reach/metrics.py

def compute_f1(precision: float, recall: float) -> float:
    """Calculate the harmonic mean of precision and recall."""
    total = precision + recall
    return 2 * precision * recall / total if total else 0.0

compute_precursor_graph

compute_precursor_graph(
    results: Sequence[ProbeResult],
    queries: Sequence[Query],
    skills: Sequence[Skill] = (),
    min_observations: int = 1,
) -> tuple[PrecursorEdge, ...]

Compute empirical precursor transition matrix T_i,j from observed trajectories.

Source code in src/reach/metrics.py

def compute_precursor_graph(
    results: Sequence[ProbeResult],
    queries: Sequence[Query],
    skills: Sequence[Skill] = (),
    min_observations: int = 1,
) -> tuple[PrecursorEdge, ...]:
    """Compute empirical precursor transition matrix T_i,j from observed trajectories."""
    truth = {q.id: q.truth_label for q in queries if not q.is_out_of_scope}
    declared_map = {s.name: set(s.declared_dependencies) for s in skills}
    transitions: dict[tuple[str, str], list[int]] = defaultdict(list)
    attempts: Counter[tuple[str, str]] = Counter()

    for r in results:
        if r.error:
            continue
        target = truth.get(r.query_id)
        if not target or len(r.invoked_skills) < MIN_TRANSITION_STEPS:
            continue

        for idx, skill in enumerate(r.invoked_skills):
            if skill == target:
                break
            attempts[(skill, target)] += 1
            remaining = r.invoked_skills[idx + 1 :]
            if target in remaining:
                step_gap = remaining.index(target) + 1
                transitions[(skill, target)].append(step_gap)

    edges = []
    for (src, dst), total_attempts in attempts.items():
        if total_attempts >= min_observations:
            gaps = transitions.get((src, dst), [])
            is_declared = src in declared_map.get(dst, set()) or dst in declared_map.get(src, set())
            edges.append(
                PrecursorEdge(
                    precursor=src,
                    target=dst,
                    attempts=total_attempts,
                    handoffs=len(gaps),
                    handoff_rate=round(len(gaps) / total_attempts, 3),
                    avg_step_latency=round(statistics.fmean(gaps), 2) if gaps else 0.0,
                    is_declared_dependency=is_declared,
                )
            )
    return tuple(sorted(edges, key=lambda e: (-e.handoff_rate, -e.attempts, e.precursor, e.target)))

confusion

confusion(
    results: Sequence[ProbeResult], queries: Sequence[Query]
) -> Counter[tuple[str, str | None]]

Count occurrences of expected-to-invoked skill selection pairs.

Source code in src/reach/metrics.py

def confusion(
    results: Sequence[ProbeResult],
    queries: Sequence[Query],
) -> Counter[tuple[str, str | None]]:
    """Count occurrences of expected-to-invoked skill selection pairs."""
    truth = {q.id: q for q in queries}
    pairs: Counter[tuple[str, str | None]] = Counter()
    for result in results:
        if result.error:
            continue
        query = truth.get(result.query_id)
        if query is None:
            continue
        effective_invoked = query.effective_invoked_skill(result)
        pairs[(query.truth_label, effective_invoked)] += 1
    return pairs

consistency

consistency(
    results: Sequence[ProbeResult], queries: Sequence[Query]
) -> float

Calculate the fraction of queries with unanimous selection outcomes.

Source code in src/reach/metrics.py

def consistency(results: Sequence[ProbeResult], queries: Sequence[Query]) -> float:
    """Calculate the fraction of queries with unanimous selection outcomes."""
    unanimous, observed = consistency_counts(results, queries)
    return unanimous / observed if observed else 0.0

consistency_counts

consistency_counts(
    results: Sequence[ProbeResult], queries: Sequence[Query]
) -> tuple[int, int]

Count unanimous and total observed queries across attempt replicates.

Source code in src/reach/metrics.py

def consistency_counts(
    results: Sequence[ProbeResult],
    queries: Sequence[Query],
) -> tuple[int, int]:
    """Count unanimous and total observed queries across attempt replicates."""
    _paired(results, queries)
    grouped: dict[str, set[str]] = {}
    for result in results:
        if not result.error:
            grouped.setdefault(result.query_id, set()).add(result.predicted_label)
    return sum(len(picks) == 1 for picks in grouped.values()), len(grouped)

decompose_pass_rate_drop

decompose_pass_rate_drop(
    baseline_results: Sequence[ProbeResult],
    scaled_results: Sequence[ProbeResult],
    queries: Sequence[Query] | None = None,
    iterations: int = 2000,
    seed: int = 42,
) -> DecompositionResult

Decompose overall pass-rate drop into context dilution versus skill shadowing components.

Source code in src/reach/metrics.py

def decompose_pass_rate_drop(
    baseline_results: Sequence[ProbeResult],
    scaled_results: Sequence[ProbeResult],
    queries: Sequence[Query] | None = None,
    iterations: int = 2000,
    seed: int = 42,
) -> DecompositionResult:
    """Decompose overall pass-rate drop into context dilution versus skill shadowing components."""
    truth_map: dict[str, Query] = {q.id: q for q in queries} if queries else {}
    base_by_query = _group_valid_results_by_query(baseline_results)
    scaled_by_query = _group_valid_results_by_query(scaled_results)

    common_qids = _resolve_evaluation_query_ids(base_by_query, scaled_by_query)
    if not common_qids:
        return DecompositionResult(
            baseline_pass_rate=0.0,
            scaled_pass_rate=0.0,
            delta_total=0.0,
            delta_context=0.0,
            delta_shadowing=0.0,
            delta_total_ci=(0.0, 0.0),
            delta_context_ci=(0.0, 0.0),
            delta_shadowing_ci=(0.0, 0.0),
            sample_size=0,
        )

    drops = [
        _decompose_query_drop(
            base_by_query.get(qid, []),
            scaled_by_query.get(qid, []),
            truth_map.get(qid),
        )
        for qid in common_qids
    ]

    q_deltas = [d.delta for d in drops]
    q_ctx_deltas = [d.delta_context for d in drops]
    q_shd_deltas = [d.delta_shadowing for d in drops]

    base_pass_rate = statistics.fmean(d.base_pass for d in drops)
    scaled_pass_rate = statistics.fmean(d.scaled_pass for d in drops)
    delta_total = statistics.fmean(q_deltas)
    delta_ctx = statistics.fmean(q_ctx_deltas)
    delta_shd = statistics.fmean(q_shd_deltas)

    delta_ci, ctx_ci, shd_ci = _bootstrap_decomposition_ci(
        q_deltas, q_ctx_deltas, q_shd_deltas, iterations=iterations, seed=seed
    )

    m = len(common_qids)
    return DecompositionResult(
        baseline_pass_rate=base_pass_rate,
        scaled_pass_rate=scaled_pass_rate,
        delta_total=delta_total,
        delta_context=delta_ctx,
        delta_shadowing=delta_shd,
        delta_total_ci=delta_ci,
        delta_context_ci=ctx_ci,
        delta_shadowing_ci=shd_ci,
        sample_size=m,
        baseline_ci=_compute_pass_rate_interval(base_pass_rate, m),
        scaled_ci=_compute_pass_rate_interval(scaled_pass_rate, m),
    )

labeled_pairs

labeled_pairs(
    results: Sequence[ProbeResult], queries: Sequence[Query]
) -> tuple[list[str], list[str]]

Extract aligned ground truth and predicted label sequences.

Source code in src/reach/metrics.py

def labeled_pairs(
    results: Sequence[ProbeResult],
    queries: Sequence[Query],
) -> tuple[list[str], list[str]]:
    """Extract aligned ground truth and predicted label sequences."""
    pairs = _paired(results, queries)
    return (
        [query.truth_label for query, _ in pairs],
        [query.effective_predicted_label(result) for query, result in pairs],
    )

score_trajectory

score_trajectory(
    query: Query, invoked_skills: Sequence[str]
) -> TrajectoryScore

Evaluate an observed skill trajectory against query target skill and acceptable skills.

Source code in src/reach/metrics.py

def score_trajectory(
    query: Query,
    invoked_skills: Sequence[str],
) -> TrajectoryScore:
    """Evaluate an observed skill trajectory against query target skill and acceptable skills."""
    invoked_seq = query.scored_invocations(invoked_skills)

    if query.is_out_of_scope:
        abstained = len(invoked_seq) == 0
        return TrajectoryScore(
            entrypoint_hit=abstained,
            trajectory_hit=abstained,
            step_efficiency=1.0 if abstained else 0.0,
            skill_f1=1.0 if abstained else 0.0,
            redundancy=len(invoked_seq),
        )

    valid_targets = query.valid_skills
    invoked_set = set(invoked_seq)
    entry_hit = bool(invoked_seq and invoked_seq[0] in valid_targets)
    traj_hit = bool(invoked_set & valid_targets)

    first_rank = next(
        (i + 1 for i, s in enumerate(invoked_seq) if s in valid_targets),
        None,
    )
    mrr = (1.0 / first_rank) if first_rank else 0.0

    prec = (
        (len(invoked_set & valid_targets) / len(invoked_set)) if (traj_hit and invoked_set) else 0.0
    )
    rec = 1.0 if traj_hit else 0.0
    f1 = (2 * prec * rec) / (prec + rec) if (prec + rec) else 0.0

    excess = max(0, len(invoked_seq) - 1)

    return TrajectoryScore(
        entrypoint_hit=entry_hit,
        trajectory_hit=traj_hit,
        step_efficiency=round(mrr, 4),
        skill_f1=round(f1, 4),
        redundancy=excess,
    )

trajectory_scores

trajectory_scores(
    results: Sequence[ProbeResult], queries: Sequence[Query]
) -> dict[str, TrajectoryScore]

Return mapping of query_id to its aggregated TrajectoryScore across replicates.

Source code in src/reach/metrics.py

def trajectory_scores(
    results: Sequence[ProbeResult],
    queries: Sequence[Query],
) -> dict[str, TrajectoryScore]:
    """Return mapping of query_id to its aggregated TrajectoryScore across replicates."""
    pairs = _paired(results, queries)
    by_query: dict[str, list[TrajectoryScore]] = defaultdict(list)
    for q, r in pairs:
        by_query[q.id].append(score_trajectory(q, r.invoked_skills))

    aggregated: dict[str, TrajectoryScore] = {}
    for q_id, scores in by_query.items():
        if len(scores) == 1:
            aggregated[q_id] = scores[0]
        else:
            aggregated[q_id] = TrajectoryScore(
                entrypoint_hit=sum(1 for s in scores if s.entrypoint_hit) * 2 >= len(scores),
                trajectory_hit=sum(1 for s in scores if s.trajectory_hit) * 2 >= len(scores),
                step_efficiency=round(_mean([s.step_efficiency for s in scores]), 4),
                skill_f1=round(_mean([s.skill_f1 for s in scores]), 4),
                redundancy=round(_mean([float(s.redundancy) for s in scores])),
            )
    return aggregated