reach.artifact

Data structures representing evaluation outcomes, confusion matrices, and serialized run artifacts.

Assemble, validate, and serialize schema-versioned evaluation artifacts.

Abstention

Bases: BaseModel

Report overall, false, and out-of-scope abstention rates with bounds.

Source code in src/reach/artifact.py

class Abstention(BaseModel):
    """Report overall, false, and out-of-scope abstention rates with bounds."""

    model_config = ConfigDict(frozen=True)

    rate: float
    false_rate: float
    out_of_scope_detection: float | None = None
    scored: int = Field(ge=0)
    abstentions: int = Field(ge=0)
    in_scope: int = Field(ge=0)
    false_abstentions: int = Field(ge=0)
    out_of_scope: int = Field(default=0, ge=0)
    out_of_scope_detected: int = Field(default=0, ge=0)

    @computed_field
    @property
    def 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_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_interval(self) -> Interval | None:
        """Calculate the Wilson confidence interval for out-of-scope detection rate."""
        return wilson_interval(self.out_of_scope_detected, self.out_of_scope)

false_interval property

false_interval: Interval | None

Calculate the Wilson confidence interval for false abstention rate.

interval property

interval: Interval | None

Calculate the Wilson confidence interval for overall abstention rate.

out_of_scope_interval property

out_of_scope_interval: Interval | None

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

Artifact

Bases: BaseModel

Represent complete evaluation results, summary scores, and provenance.

Source code in src/reach/artifact.py

class Artifact(BaseModel):
    """Represent complete evaluation results, summary scores, and provenance."""

    model_config = ConfigDict(frozen=True)

    schema_version: str = SCHEMA_VERSION
    digests: Provenance = Field(
        description="Configuration fingerprint, corpus digest, and query set digest.",
    )
    verified_digests: tuple[str, ...] = Field(
        default=(),
        description="Provenance digests corroborated by individual probe results.",
    )
    provenance: RunProvenance
    catalog_id: str
    catalog_mode: CatalogMode
    catalog_size: int = Field(ge=0)
    catalog_target: str | None = None
    resolved_roots: tuple[ResolvedRoot, ...] = ()
    contested_skills: tuple[ContestedSkill, ...] = Field(
        default=(),
        description="Skills provided by multiple roots and resolved by precedence.",
    )
    skills: tuple[SkillScore, ...] = Field(
        default=(),
        description="Per-skill evaluation scores in catalog order.",
    )
    scores: RunScores
    spread: Spread
    confusion: tuple[ConfusionPair, ...] = ()
    queries: tuple[QueryRecord, ...] = ()
    probes: int = Field(default=0, ge=0)
    errors: int = Field(default=0, ge=0)
    spend_usd: float = Field(default=0.0, ge=0.0)
    reused: int = Field(default=0, ge=0)

    @field_validator("schema_version")
    @classmethod
    def _readable_version(cls, value: str) -> str:
        """Refuse a document written to a contract this code does not implement."""
        if value.split(".", maxsplit=1)[0] != SCHEMA_VERSION.split(".", maxsplit=1)[0]:
            msg = (
                f"artifact schema_version {value!r} is not readable by "
                f"{SCHEMA_VERSION!r}: the major version differs"
            )
            raise ValueError(
                msg,
            )
        return value

    @model_validator(mode="after")
    def _every_digest_is_stated(self) -> Self:
        """Validate that all required provenance digests are populated."""
        missing = [
            field
            for field in ("config_fingerprint", "corpus_digest", "queries_digest")
            if not getattr(self.digests, field)
        ]
        if missing:
            msg = f"an artifact must state every digest; missing: {missing}"
            raise ValueError(msg)
        return self

    @property
    def unreached(self) -> tuple[SkillScore, ...]:
        """Return the skills that were asked for and never arrived at across any turn."""
        return tuple(
            s
            for s in self.skills
            if s.recall is not None
            and (s.trajectory_recall if s.trajectory_recall is not None else s.recall) == 0.0
        )

    @property
    def attractors(self) -> tuple[SkillScore, ...]:
        """Return skills selected more often than requested in ground truth."""
        taking = (s for s in self.skills if s.reached + s.absorbed > s.probes)
        return tuple(sorted(taking, key=lambda s: (-s.absorbed, s.skill)))

    @property
    def unclean(self) -> tuple[QueryRecord, ...]:
        """Return the queries that did not agree with ground truth on every attempt."""
        return tuple(q for q in self.queries if q.probes > 0 and not q.clean)

    @classmethod
    def assemble(
        cls,
        composition: Composition,
        results: Sequence[ProbeResult],
        *,
        roots: Sequence[Path] | None = None,
        contested: Sequence[ContestedSkill] = (),
        sample_queries: int = DEFAULT_SAMPLE_QUERIES,
        fit: CatalogFit | None = None,
        spend_usd: float = 0.0,
        reused: int = 0,
        difficulty: Mapping[str, LexicalRank] | None = None,
        digests: Provenance | None = None,
        cross_check: bool = True,
    ) -> Artifact:
        """Assemble an Artifact from a validated Composition and probe results.

        Derive query set, catalog, skills, and configuration directly from the
        Composition instance, reducing the orchestration seam from fifteen
        parameters to two essential inputs.
        """
        return _ArtifactAssembler(
            results=results,
            query_set=composition.query_set,
            catalog=composition.catalog,
            skills=composition.skills,
            config=composition.config,
            roots=roots,
            contested=contested,
            sample_queries=sample_queries,
            fit=fit,
            spend_usd=spend_usd,
            reused=reused,
            difficulty=difficulty,
            digests=digests,
            cross_check=cross_check,
        ).assemble()

attractors property

attractors: tuple[SkillScore, ...]

Return skills selected more often than requested in ground truth.

unclean property

unclean: tuple[QueryRecord, ...]

Return the queries that did not agree with ground truth on every attempt.

unreached property

unreached: tuple[SkillScore, ...]

Return the skills that were asked for and never arrived at across any turn.

assemble classmethod

assemble(
    composition: Composition,
    results: Sequence[ProbeResult],
    *,
    roots: Sequence[Path] | None = None,
    contested: Sequence[ContestedSkill] = (),
    sample_queries: int = DEFAULT_SAMPLE_QUERIES,
    fit: CatalogFit | None = None,
    spend_usd: float = 0.0,
    reused: int = 0,
    difficulty: Mapping[str, LexicalRank] | None = None,
    digests: Provenance | None = None,
    cross_check: bool = True,
) -> Artifact

Assemble an Artifact from a validated Composition and probe results.

Derive query set, catalog, skills, and configuration directly from the Composition instance, reducing the orchestration seam from fifteen parameters to two essential inputs.

Source code in src/reach/artifact.py

@classmethod
def assemble(
    cls,
    composition: Composition,
    results: Sequence[ProbeResult],
    *,
    roots: Sequence[Path] | None = None,
    contested: Sequence[ContestedSkill] = (),
    sample_queries: int = DEFAULT_SAMPLE_QUERIES,
    fit: CatalogFit | None = None,
    spend_usd: float = 0.0,
    reused: int = 0,
    difficulty: Mapping[str, LexicalRank] | None = None,
    digests: Provenance | None = None,
    cross_check: bool = True,
) -> Artifact:
    """Assemble an Artifact from a validated Composition and probe results.

    Derive query set, catalog, skills, and configuration directly from the
    Composition instance, reducing the orchestration seam from fifteen
    parameters to two essential inputs.
    """
    return _ArtifactAssembler(
        results=results,
        query_set=composition.query_set,
        catalog=composition.catalog,
        skills=composition.skills,
        config=composition.config,
        roots=roots,
        contested=contested,
        sample_queries=sample_queries,
        fit=fit,
        spend_usd=spend_usd,
        reused=reused,
        difficulty=difficulty,
        digests=digests,
        cross_check=cross_check,
    ).assemble()

ConfusionPair

Bases: BaseModel

Record misroute pairings, probe counts, and sample queries.

Source code in src/reach/artifact.py

class ConfusionPair(BaseModel):
    """Record misroute pairings, probe counts, and sample queries."""

    model_config = ConfigDict(frozen=True)

    expected: str
    invoked: str
    probes: int = Field(ge=1)
    collisions: int = Field(default=0, ge=0)
    queries: tuple[SampleQuery, ...] = ()

    @model_validator(mode="after")
    def _collisions_are_a_subset(self) -> Self:
        """Validate that collisions do not exceed total probe count."""
        if self.collisions > self.probes:
            msg = (
                f"{self.expected} -> {self.invoked}: {self.collisions} collisions "
                f"out of {self.probes} probes"
            )
            raise ValueError(
                msg,
            )
        return self

ContestedSkill

Bases: BaseModel

Record a skill name provided by multiple roots and the resolved path used.

Source code in src/reach/artifact.py

class ContestedSkill(BaseModel):
    """Record a skill name provided by multiple roots and the resolved path used."""

    model_config = ConfigDict(frozen=True)

    name: str
    kept: Path
    dropped: tuple[Path, ...]
    ranked: bool = True

NotHeadline

Bases: BaseModel

Hold secondary evaluation metrics (macro precision, recall, F1).

Source code in src/reach/artifact.py

class NotHeadline(BaseModel):
    """Hold secondary evaluation metrics (macro precision, recall, F1)."""

    model_config = ConfigDict(frozen=True)

    macro_f1: float
    macro_precision: float = 0.0
    macro_recall: float = 0.0
    labels: tuple[str, ...] = ()

QueryRecord

Bases: BaseModel

Stage 3 telemetry: summarize probe outcomes, confidence interval, and leak status.

Source code in src/reach/artifact.py

class QueryRecord(BaseModel):
    """Stage 3 telemetry: summarize probe outcomes, confidence interval, and leak status."""

    model_config = ConfigDict(frozen=True)

    query_id: str
    text: str
    kind: QueryKind | None = None
    expected: str
    probes: int = Field(default=0, ge=0)
    hits: int = Field(default=0, ge=0)
    selections: tuple[str, ...] = ()
    difficulty_rank: int | None = None
    leak: Leak | None = None

    @property
    def clean(self) -> bool:
        """Return True if all probe attempts matched ground truth."""
        return self.probes > 0 and self.hits == self.probes

    @computed_field
    @property
    def interval(self) -> Interval | None:
        """Calculate the Wilson score confidence interval for query hit rate."""
        return wilson_interval(self.hits, self.probes)

clean property

clean: bool

Return True if all probe attempts matched ground truth.

interval property

interval: Interval | None

Calculate the Wilson score confidence interval for query hit rate.

ResolvedRoot

Bases: BaseModel

Record a source tree root and the count of skills loaded from it.

Source code in src/reach/artifact.py

class ResolvedRoot(BaseModel):
    """Record a source tree root and the count of skills loaded from it."""

    model_config = ConfigDict(frozen=True)

    path: Path
    skills: int = Field(ge=0)

RunProvenance

Bases: BaseModel

Record runtime, model, attempt count, and experimental arm for a run.

Source code in src/reach/artifact.py

class RunProvenance(BaseModel):
    """Record runtime, model, attempt count, and experimental arm for a run."""

    model_config = ConfigDict(frozen=True)

    runtime: str
    model: str
    resolved_model: str = ""
    attempts: int = Field(ge=1)
    arm: str
    condition: str = ""
    catalog_fit: CatalogFit | None = None

RunScores

Bases: BaseModel

Hold primary run-level evaluation scores and confidence intervals.

Source code in src/reach/artifact.py

class RunScores(BaseModel):
    """Hold primary run-level evaluation scores and confidence intervals."""

    model_config = ConfigDict(frozen=True)

    consistency: float
    top1_accuracy: float
    abstention: Abstention
    not_headline: NotHeadline
    unanimous_queries: int = Field(ge=0)
    observed_queries: int = Field(ge=0)
    top1_hits: int = Field(ge=0)
    entrypoint_hits: int = Field(default=0, ge=0)
    entrypoint_accuracy: float = 0.0
    trajectory_hits: int = Field(default=0, ge=0)
    trajectory_reachability: float = 0.0
    step_efficiency: float = 0.0
    skill_f1: float = 0.0
    redundancy: float = 0.0
    scored: int = Field(ge=0)

    @computed_field
    @property
    def consistency_interval(self) -> Interval | None:
        """Calculate the Wilson confidence interval for query consistency."""
        return wilson_interval(self.unanimous_queries, self.observed_queries)

    @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 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_interval(self) -> Interval | None:
        """Calculate the Wilson confidence interval for trajectory reachability."""
        return wilson_interval(self.trajectory_hits, self.scored)

    @model_validator(mode="after")
    def _rates_match_the_counts_they_came_from(self) -> Self:
        """Validate that stored rates match underlying hit and total counts."""
        for name, rate, hits, total in (
            (
                "consistency",
                self.consistency,
                self.unanimous_queries,
                self.observed_queries,
            ),
            ("top1_accuracy", self.top1_accuracy, self.top1_hits, self.scored),
            (
                "entrypoint_accuracy",
                self.entrypoint_accuracy,
                self.entrypoint_hits,
                self.scored,
            ),
            (
                "trajectory_reachability",
                self.trajectory_reachability,
                self.trajectory_hits,
                self.scored,
            ),
        ):
            if not total:
                continue
            if abs(rate - hits / total) > RATE_TOLERANCE:
                msg = f"{name} is {rate} but its counts give {hits}/{total}"
                raise ValueError(msg)
        return self

consistency_interval property

consistency_interval: Interval | None

Calculate the Wilson confidence interval for query consistency.

entrypoint_interval property

entrypoint_interval: Interval | None

Calculate the Wilson confidence interval for entrypoint accuracy.

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.

SampleQuery

Bases: BaseModel

Quote one query verbatim with probe count and model reasoning traces.

Source code in src/reach/artifact.py

class SampleQuery(BaseModel):
    """Quote one query verbatim with probe count and model reasoning traces."""

    model_config = ConfigDict(frozen=True)

    query_id: str
    text: str
    probes: int = Field(
        ge=1,
        description="Number of attempts that resulted in this confusion pairing.",
    )
    reasoning: tuple[str, ...] = ()

SkillScore

Bases: BaseModel

Represent precision, recall, and uncertainty for an individual skill.

Source code in src/reach/artifact.py

class SkillScore(BaseModel):
    """Represent precision, recall, and uncertainty for an individual skill."""

    model_config = ConfigDict(frozen=True)

    skill: str
    root: Path | None = None
    probes: int = Field(ge=0)
    reached: int = Field(ge=0)
    recall: float | None = None
    trajectory_reached: int = Field(default=0, ge=0)
    trajectory_recall: float | None = None
    absorbed: int = Field(default=0, ge=0)
    precision: float | None = None

    @model_validator(mode="after")
    def _rates_agree_with_their_counts(self) -> Self:
        """Tie each rate to the count it was taken over, in both directions."""
        if (self.probes > 0) != (self.recall is not None):
            msg = (
                f"{self.skill}: recall is defined exactly when a query named the "
                f"skill; got recall={self.recall} over {self.probes} probes"
            )
            raise ValueError(
                msg,
            )
        if self.trajectory_reached < self.reached:
            object.__setattr__(self, "trajectory_reached", self.reached)
        if self.probes > 0 and self.trajectory_recall is None:
            object.__setattr__(
                self,
                "trajectory_recall",
                self.trajectory_reached / self.probes,
            )
        if (self.probes > 0) != (self.trajectory_recall is not None):
            msg = (
                f"{self.skill}: trajectory_recall is defined exactly when a query "
                f"named the skill; got trajectory_recall={self.trajectory_recall} "
                f"over {self.probes} probes"
            )
            raise ValueError(
                msg,
            )
        if (self.reached + self.absorbed > 0) != (self.precision is not None):
            msg = (
                f"{self.skill}: precision is defined exactly when the skill was "
                f"selected; got precision={self.precision} over "
                f"{self.reached + self.absorbed} selections"
            )
            raise ValueError(
                msg,
            )
        return self

    @computed_field
    @property
    def f1(self) -> float | None:
        """Return harmonic mean of precision and recall, or None if undefined."""
        if self.recall is None:
            return None
        return compute_f1(self.precision or 0.0, self.recall)

    @computed_field
    @property
    def recall_interval(self) -> Interval | None:
        """Calculate the Wilson confidence interval for this skill's recall."""
        return wilson_interval(self.reached, self.probes)

    @computed_field
    @property
    def precision_interval(self) -> Interval | None:
        """Calculate the Wilson confidence interval for this skill's precision."""
        return wilson_interval(self.reached, self.reached + self.absorbed)

    @classmethod
    def from_class_metrics(
        cls,
        metrics: ClassMetrics,
        root: Path | None = None,
    ) -> SkillScore:
        """Construct a SkillScore model from ClassMetrics and optional root path."""
        selected = metrics.true_positives + metrics.false_positives
        return cls(
            skill=metrics.label,
            root=root,
            probes=metrics.support,
            reached=metrics.true_positives,
            recall=metrics.recall if metrics.support else None,
            trajectory_reached=metrics.trajectory_true_positives,
            trajectory_recall=metrics.trajectory_recall if metrics.support else None,
            absorbed=metrics.false_positives,
            precision=metrics.precision if selected else None,
        )

f1 property

f1: float | None

Return harmonic mean of precision and recall, or None if undefined.

precision_interval property

precision_interval: Interval | None

Calculate the Wilson confidence interval for this skill's precision.

recall_interval property

recall_interval: Interval | None

Calculate the Wilson confidence interval for this skill's recall.

from_class_metrics classmethod

from_class_metrics(
    metrics: ClassMetrics, root: Path | None = None
) -> SkillScore

Construct a SkillScore model from ClassMetrics and optional root path.

Source code in src/reach/artifact.py

@classmethod
def from_class_metrics(
    cls,
    metrics: ClassMetrics,
    root: Path | None = None,
) -> SkillScore:
    """Construct a SkillScore model from ClassMetrics and optional root path."""
    selected = metrics.true_positives + metrics.false_positives
    return cls(
        skill=metrics.label,
        root=root,
        probes=metrics.support,
        reached=metrics.true_positives,
        recall=metrics.recall if metrics.support else None,
        trajectory_reached=metrics.trajectory_true_positives,
        trajectory_recall=metrics.trajectory_recall if metrics.support else None,
        absorbed=metrics.false_positives,
        precision=metrics.precision if selected else None,
    )

Spread

Bases: BaseModel

Summarize variance and standard error across probe replicates.

Source code in src/reach/artifact.py

class Spread(BaseModel):
    """Summarize variance and standard error across probe replicates."""

    model_config = ConfigDict(frozen=True)

    replicates: int = Field(default=0, ge=0)
    top1_by_attempt: tuple[float, ...] = ()
    mean: float | None = None
    repeated_queries: int = Field(default=0, ge=0)
    standard_error: float | None = None

    @model_validator(mode="after")
    def _figures_match_what_was_observed(self) -> Self:
        """Validate replicate counts against observed statistics."""
        if len(self.top1_by_attempt) != self.replicates:
            msg = f"{self.replicates} replicates but {len(self.top1_by_attempt)} accuracies"
            raise ValueError(
                msg,
            )
        if (self.replicates > 0) != (self.mean is not None):
            msg = f"{self.replicates} replicates cannot mean to {self.mean}"
            raise ValueError(msg)
        if (self.replicates > 0) != (self.standard_error is not None):
            msg = (
                f"a probed run has an error on its estimate; got {self.replicates} "
                f"replicates and standard_error={self.standard_error}"
            )
            raise ValueError(
                msg,
            )
        return self

artifact_path

artifact_path(results_path: Path) -> Path

Return the default artifact path corresponding to a results file.

Source code in src/reach/artifact.py

def artifact_path(results_path: Path) -> Path:
    """Return the default artifact path corresponding to a results file."""
    return Path(f"{results_path}{ARTIFACT_SUFFIX}")

filter_query_set

filter_query_set(
    query_set: QuerySet,
    *,
    subset: QuerySet | None = None,
    filter_skill: Sequence[str] = (),
    filter_id: Sequence[str] = (),
) -> QuerySet

Filter a QuerySet to a subset QuerySet and/or skill and query ID globs.

Source code in src/reach/artifact.py

def filter_query_set(
    query_set: QuerySet,
    *,
    subset: QuerySet | None = None,
    filter_skill: Sequence[str] = (),
    filter_id: Sequence[str] = (),
) -> QuerySet:
    """Filter a QuerySet to a subset QuerySet and/or skill and query ID globs."""
    if subset is None and not filter_skill and not filter_id:
        return query_set

    queries = _resolve_subset_queries(query_set, subset)

    for pat in filter_skill:
        if not any(_query_matches_skill(q, pat) for q in queries):
            msg = f"query slice matched 0 queries for filter_skill pattern {pat!r}"
            raise ValueError(msg)
    if filter_skill:
        queries = [q for q in queries if any(_query_matches_skill(q, pat) for pat in filter_skill)]

    for pat in filter_id:
        if not any(fnmatchcase(q.id, pat) for q in queries):
            msg = f"query slice matched 0 queries for filter_id pattern {pat!r}"
            raise ValueError(msg)
    if filter_id:
        queries = [q for q in queries if any(fnmatchcase(q.id, pat) for pat in filter_id)]
    if not queries:
        msg = "query slice matched 0 queries"
        raise ValueError(msg)

    return query_set.model_copy(update={"queries": tuple(queries)})

read_artifact

read_artifact(path: Path) -> Artifact

Read and validate a Artifact from a JSON file.

Source code in src/reach/artifact.py

def read_artifact(path: Path) -> Artifact:
    """Read and validate a Artifact from a JSON file."""
    return Artifact.model_validate_json(Path(path).read_text(encoding="utf-8"))

write_artifact

write_artifact(artifact: Artifact, path: Path) -> Path

Serialize and write a Artifact model to disk as formatted JSON.

Source code in src/reach/artifact.py

def write_artifact(artifact: Artifact, path: Path) -> Path:
    """Serialize and write a Artifact model to disk as formatted JSON."""
    destination = Path(path)
    if destination.exists() and not _is_artifact(destination):
        msg = (
            f"refusing to overwrite {destination}: it is not an artifact, and "
            "writing one here would destroy it. Artifacts belong at the path "
            "artifact_path() names."
        )
        raise ValueError(
            msg,
        )
    return write_model(artifact, destination)