API reference

Generated from the package docstrings.

Artifacts

recension.artifact

Versioned text artifacts with provenance.

A :class:TextArtifact holds the text being optimized (a prompt, a context template, a skill file) together with an append-only, linear version history. Every version after the root carries a :class:Provenance: the diagnosis that motivated the change, the scores that justified it, the sibling candidates that were rejected, and a unified diff against the parent. A reviewer can reconstruct every accepted edit from the artifact alone.

RejectedCandidate dataclass

A sibling candidate that lost to the accepted version.

Kept in full (text included) so the comparison that justified the accepted edit can be reproduced later.

Source code in recension/artifact.py

@dataclass(frozen=True)
class RejectedCandidate:
    """A sibling candidate that lost to the accepted version.

    Kept in full (text included) so the comparison that justified the accepted
    edit can be reproduced later.
    """

    candidate_id: str
    text: str
    score: float | None
    leakage_flags: tuple[str, ...] = ()

Provenance dataclass

Why a version exists.

Attributes:

Name	Type	Description
diagnosis	str	Free-text hypothesis about what in the parent text caused the observed failures (or a note such as a rollback reason).
failure_example_ids	tuple[str, ...]	Ids of the train examples whose failures motivated the change.
incumbent_score	float | None	Held-out score of the parent version, if measured.
candidate_score	float | None	Held-out score of this version, if measured.
rejected_candidates	tuple[RejectedCandidate, ...]	Sibling candidates considered in the same round, with their scores and any leakage flags.
diff	str	Unified diff against the parent text. Always computed by :meth:TextArtifact.commit, never supplied by the caller, so it cannot drift from the actual texts.

Source code in recension/artifact.py

@dataclass(frozen=True)
class Provenance:
    """Why a version exists.

    Attributes:
        diagnosis: Free-text hypothesis about what in the parent text caused
            the observed failures (or a note such as a rollback reason).
        failure_example_ids: Ids of the train examples whose failures
            motivated the change.
        incumbent_score: Held-out score of the parent version, if measured.
        candidate_score: Held-out score of this version, if measured.
        rejected_candidates: Sibling candidates considered in the same round,
            with their scores and any leakage flags.
        diff: Unified diff against the parent text. Always computed by
            :meth:`TextArtifact.commit`, never supplied by the caller, so it
            cannot drift from the actual texts.
    """

    diagnosis: str
    failure_example_ids: tuple[str, ...] = ()
    incumbent_score: float | None = None
    candidate_score: float | None = None
    rejected_candidates: tuple[RejectedCandidate, ...] = ()
    diff: str = ""

Version dataclass

One immutable entry in an artifact's history.

Source code in recension/artifact.py

@dataclass(frozen=True)
class Version:
    """One immutable entry in an artifact's history."""

    version_id: str
    parent_id: str | None
    text: str
    created_at: str
    provenance: Provenance | None = None

TextArtifact

A text under optimization, with its full version history.

The history is linear and append-only: each version has exactly one parent, and nothing is ever rewritten or deleted. rollback therefore appends a new version whose text restores an earlier one, rather than moving a pointer backwards. The record of having tried and reverted is itself part of the audit trail.

Source code in recension/artifact.py

class TextArtifact:
    """A text under optimization, with its full version history.

    The history is linear and append-only: each version has exactly one
    parent, and nothing is ever rewritten or deleted. ``rollback`` therefore
    *appends* a new version whose text restores an earlier one, rather than
    moving a pointer backwards. The record of having tried and reverted is
    itself part of the audit trail.
    """

    def __init__(self, versions: list[Version], name: str = "artifact") -> None:
        """Build an artifact from an existing linear history.

        Most callers should use :meth:`from_text` or :meth:`from_file`.

        Raises:
            ArtifactError: If ``versions`` is empty or not a linear chain.
        """
        if not versions:
            raise ArtifactError("an artifact needs at least one version")
        for i, version in enumerate(versions):
            expected_parent = None if i == 0 else versions[i - 1].version_id
            if version.parent_id != expected_parent:
                raise ArtifactError(
                    f"versions do not form a linear chain at {version.version_id!r}"
                )
        self.name = name
        self._versions: list[Version] = list(versions)

    # -- construction -------------------------------------------------------

    @classmethod
    def from_text(cls, text: str, name: str = "artifact") -> TextArtifact:
        """Create a new artifact whose root version holds ``text``."""
        root = Version(
            version_id=_version_id(None, text),
            parent_id=None,
            text=text,
            created_at=datetime.now(UTC).isoformat(),
        )
        return cls([root], name=name)

    @classmethod
    def from_file(cls, path: str | Path, name: str | None = None) -> TextArtifact:
        """Create a new artifact from the contents of a text file.

        The artifact name defaults to the file's stem.
        """
        p = Path(path)
        return cls.from_text(p.read_text(encoding="utf-8"), name=name or p.stem)

    # -- reading ------------------------------------------------------------

    def current(self) -> Version:
        """The latest version (the incumbent)."""
        return self._versions[-1]

    @property
    def text(self) -> str:
        """The current version's text."""
        return self.current().text

    def history(self) -> list[Version]:
        """All versions, root first, current last."""
        return list(self._versions)

    def get(self, version_id: str) -> Version:
        """Look up a version by id.

        Raises:
            ArtifactError: If no version has that id.
        """
        for version in self._versions:
            if version.version_id == version_id:
                return version
        raise ArtifactError(f"unknown version id {version_id!r} in artifact {self.name!r}")

    def diff(self, version_a: str, version_b: str) -> str:
        """Unified diff between two versions' texts, by version id."""
        a, b = self.get(version_a), self.get(version_b)
        return _diff_texts(a.text, b.text, version_a, version_b)

    def verify(self) -> list[str]:
        """Check content-addressing integrity of the version history.

        Version ids are a hash of ``(parent_id, text)``, so editing a version's
        text or id after the fact, or breaking the parent chain, is detectable.
        Returns a list of human-readable problems, empty when the history is
        intact. This is the self-contained tamper-evidence behind
        :meth:`recension.record.RunRecord.verify`.
        """
        problems: list[str] = []
        for i, version in enumerate(self._versions):
            expected_parent = None if i == 0 else self._versions[i - 1].version_id
            if version.parent_id != expected_parent:
                problems.append(
                    f"version {version.version_id!r} has parent {version.parent_id!r}, "
                    f"expected {expected_parent!r}"
                )
            expected_id = _version_id(version.parent_id, version.text)
            if version.version_id != expected_id:
                problems.append(
                    f"version {version.version_id!r} does not match its content hash "
                    f"({expected_id!r}); the text or id may have been altered"
                )
        return problems

    # -- writing ------------------------------------------------------------

    def commit(self, text: str, provenance: Provenance) -> Version:
        """Append a new version with ``text`` and ``provenance``.

        The diff against the parent is computed here and written into the
        stored provenance; any caller-supplied ``provenance.diff`` is ignored.

        Raises:
            ArtifactError: If ``text`` is identical to the current text
                (a no-op commit would corrupt the history's meaning).
        """
        parent = self.current()
        if text == parent.text:
            raise ArtifactError("refusing no-op commit: text is identical to the current version")
        new_id = _version_id(parent.version_id, text)
        stored = replace(provenance, diff=_diff_texts(parent.text, text, parent.version_id, new_id))
        version = Version(
            version_id=new_id,
            parent_id=parent.version_id,
            text=text,
            created_at=datetime.now(UTC).isoformat(),
            provenance=stored,
        )
        self._versions.append(version)
        return version

    def rollback(self, version_id: str) -> Version:
        """Restore an earlier version's text by appending a new version.

        Raises:
            ArtifactError: If ``version_id`` is unknown or already current.
        """
        target = self.get(version_id)
        if target.version_id == self.current().version_id:
            raise ArtifactError(f"version {version_id!r} is already current")
        return self.commit(
            target.text,
            Provenance(diagnosis=f"rollback to version {version_id}"),
        )

    # -- serialization ------------------------------------------------------

    def to_dict(self) -> dict[str, Any]:
        """Plain-dict form, suitable for embedding in a run record.

        JSON-pure (lists, not tuples), so a serialize/deserialize round trip
        is the identity.
        """
        return {
            "name": self.name,
            "versions": [_version_to_dict(v) for v in self._versions],
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> TextArtifact:
        """Inverse of :meth:`to_dict`."""
        versions = [
            Version(
                version_id=v["version_id"],
                parent_id=v["parent_id"],
                text=v["text"],
                created_at=v["created_at"],
                provenance=_provenance_from_dict(v["provenance"]) if v["provenance"] else None,
            )
            for v in data["versions"]
        ]
        return cls(versions, name=data["name"])

    def to_json(self, *, indent: int | None = 2) -> str:
        """Serialize the full artifact (history included) to JSON."""
        return json.dumps(self.to_dict(), indent=indent, ensure_ascii=False)

    @classmethod
    def from_json(cls, payload: str) -> TextArtifact:
        """Inverse of :meth:`to_json`."""
        return cls.from_dict(json.loads(payload))

text property

text

The current version's text.

__init__

__init__(versions, name='artifact')

Build an artifact from an existing linear history.

Most callers should use :meth:from_text or :meth:from_file.

Raises:

Type	Description
ArtifactError	If versions is empty or not a linear chain.

Source code in recension/artifact.py

def __init__(self, versions: list[Version], name: str = "artifact") -> None:
    """Build an artifact from an existing linear history.

    Most callers should use :meth:`from_text` or :meth:`from_file`.

    Raises:
        ArtifactError: If ``versions`` is empty or not a linear chain.
    """
    if not versions:
        raise ArtifactError("an artifact needs at least one version")
    for i, version in enumerate(versions):
        expected_parent = None if i == 0 else versions[i - 1].version_id
        if version.parent_id != expected_parent:
            raise ArtifactError(
                f"versions do not form a linear chain at {version.version_id!r}"
            )
    self.name = name
    self._versions: list[Version] = list(versions)

from_text classmethod

from_text(text, name='artifact')

Create a new artifact whose root version holds text.

Source code in recension/artifact.py

@classmethod
def from_text(cls, text: str, name: str = "artifact") -> TextArtifact:
    """Create a new artifact whose root version holds ``text``."""
    root = Version(
        version_id=_version_id(None, text),
        parent_id=None,
        text=text,
        created_at=datetime.now(UTC).isoformat(),
    )
    return cls([root], name=name)

from_file classmethod

from_file(path, name=None)

Create a new artifact from the contents of a text file.

The artifact name defaults to the file's stem.

Source code in recension/artifact.py

@classmethod
def from_file(cls, path: str | Path, name: str | None = None) -> TextArtifact:
    """Create a new artifact from the contents of a text file.

    The artifact name defaults to the file's stem.
    """
    p = Path(path)
    return cls.from_text(p.read_text(encoding="utf-8"), name=name or p.stem)

current

current()

The latest version (the incumbent).

Source code in recension/artifact.py

def current(self) -> Version:
    """The latest version (the incumbent)."""
    return self._versions[-1]

history

history()

All versions, root first, current last.

Source code in recension/artifact.py

def history(self) -> list[Version]:
    """All versions, root first, current last."""
    return list(self._versions)

get

get(version_id)

Look up a version by id.

Raises:

Type	Description
ArtifactError	If no version has that id.

Source code in recension/artifact.py

def get(self, version_id: str) -> Version:
    """Look up a version by id.

    Raises:
        ArtifactError: If no version has that id.
    """
    for version in self._versions:
        if version.version_id == version_id:
            return version
    raise ArtifactError(f"unknown version id {version_id!r} in artifact {self.name!r}")

diff

diff(version_a, version_b)

Unified diff between two versions' texts, by version id.

Source code in recension/artifact.py

def diff(self, version_a: str, version_b: str) -> str:
    """Unified diff between two versions' texts, by version id."""
    a, b = self.get(version_a), self.get(version_b)
    return _diff_texts(a.text, b.text, version_a, version_b)

verify

verify()

Check content-addressing integrity of the version history.

Version ids are a hash of (parent_id, text), so editing a version's text or id after the fact, or breaking the parent chain, is detectable. Returns a list of human-readable problems, empty when the history is intact. This is the self-contained tamper-evidence behind :meth:recension.record.RunRecord.verify.

Source code in recension/artifact.py

def verify(self) -> list[str]:
    """Check content-addressing integrity of the version history.

    Version ids are a hash of ``(parent_id, text)``, so editing a version's
    text or id after the fact, or breaking the parent chain, is detectable.
    Returns a list of human-readable problems, empty when the history is
    intact. This is the self-contained tamper-evidence behind
    :meth:`recension.record.RunRecord.verify`.
    """
    problems: list[str] = []
    for i, version in enumerate(self._versions):
        expected_parent = None if i == 0 else self._versions[i - 1].version_id
        if version.parent_id != expected_parent:
            problems.append(
                f"version {version.version_id!r} has parent {version.parent_id!r}, "
                f"expected {expected_parent!r}"
            )
        expected_id = _version_id(version.parent_id, version.text)
        if version.version_id != expected_id:
            problems.append(
                f"version {version.version_id!r} does not match its content hash "
                f"({expected_id!r}); the text or id may have been altered"
            )
    return problems

commit

commit(text, provenance)

Append a new version with text and provenance.

The diff against the parent is computed here and written into the stored provenance; any caller-supplied provenance.diff is ignored.

Raises:

Type	Description
ArtifactError	If text is identical to the current text (a no-op commit would corrupt the history's meaning).

Source code in recension/artifact.py

def commit(self, text: str, provenance: Provenance) -> Version:
    """Append a new version with ``text`` and ``provenance``.

    The diff against the parent is computed here and written into the
    stored provenance; any caller-supplied ``provenance.diff`` is ignored.

    Raises:
        ArtifactError: If ``text`` is identical to the current text
            (a no-op commit would corrupt the history's meaning).
    """
    parent = self.current()
    if text == parent.text:
        raise ArtifactError("refusing no-op commit: text is identical to the current version")
    new_id = _version_id(parent.version_id, text)
    stored = replace(provenance, diff=_diff_texts(parent.text, text, parent.version_id, new_id))
    version = Version(
        version_id=new_id,
        parent_id=parent.version_id,
        text=text,
        created_at=datetime.now(UTC).isoformat(),
        provenance=stored,
    )
    self._versions.append(version)
    return version

rollback

rollback(version_id)

Restore an earlier version's text by appending a new version.

Raises:

Type	Description
ArtifactError	If version_id is unknown or already current.

Source code in recension/artifact.py

def rollback(self, version_id: str) -> Version:
    """Restore an earlier version's text by appending a new version.

    Raises:
        ArtifactError: If ``version_id`` is unknown or already current.
    """
    target = self.get(version_id)
    if target.version_id == self.current().version_id:
        raise ArtifactError(f"version {version_id!r} is already current")
    return self.commit(
        target.text,
        Provenance(diagnosis=f"rollback to version {version_id}"),
    )

to_dict

to_dict()

Plain-dict form, suitable for embedding in a run record.

JSON-pure (lists, not tuples), so a serialize/deserialize round trip is the identity.

Source code in recension/artifact.py

def to_dict(self) -> dict[str, Any]:
    """Plain-dict form, suitable for embedding in a run record.

    JSON-pure (lists, not tuples), so a serialize/deserialize round trip
    is the identity.
    """
    return {
        "name": self.name,
        "versions": [_version_to_dict(v) for v in self._versions],
    }

from_dict classmethod

from_dict(data)

Inverse of :meth:to_dict.

Source code in recension/artifact.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> TextArtifact:
    """Inverse of :meth:`to_dict`."""
    versions = [
        Version(
            version_id=v["version_id"],
            parent_id=v["parent_id"],
            text=v["text"],
            created_at=v["created_at"],
            provenance=_provenance_from_dict(v["provenance"]) if v["provenance"] else None,
        )
        for v in data["versions"]
    ]
    return cls(versions, name=data["name"])

to_json

to_json(*, indent=2)

Serialize the full artifact (history included) to JSON.

Source code in recension/artifact.py

def to_json(self, *, indent: int | None = 2) -> str:
    """Serialize the full artifact (history included) to JSON."""
    return json.dumps(self.to_dict(), indent=indent, ensure_ascii=False)

from_json classmethod

from_json(payload)

Inverse of :meth:to_json.

Source code in recension/artifact.py

@classmethod
def from_json(cls, payload: str) -> TextArtifact:
    """Inverse of :meth:`to_json`."""
    return cls.from_dict(json.loads(payload))

Evaluation data

recension.evalset

Held-out evaluation data: examples with an explicit train/validation split, and an optional locked test split.

The split is the integrity backbone of the whole library: the optimizer diagnoses failures on train and accepts candidates only on validation. An optional test split is never touched during optimization and is scored exactly once at the end, giving an unbiased estimate that is not subject to the multiple-comparisons bias of selecting on validation across many rounds. :class:EvalSet enforces the separation at construction time and fails loud on anything that would corrupt the acceptance signal.

Example dataclass

One evaluation example.

Attributes:

Name	Type	Description
id	str	Stable identifier; referenced by diagnoses and provenance.
input	str	The text given to the model (alongside the artifact).
expected	str | None	Gold output for reference-based objectives, if any.
rubric	str | None	Per-example grading rubric for model-graded objectives, if any.
metadata	dict[str, Any]	Any extra fields the objective may need.

Source code in recension/evalset.py

@dataclass(frozen=True)
class Example:
    """One evaluation example.

    Attributes:
        id: Stable identifier; referenced by diagnoses and provenance.
        input: The text given to the model (alongside the artifact).
        expected: Gold output for reference-based objectives, if any.
        rubric: Per-example grading rubric for model-graded objectives, if any.
        metadata: Any extra fields the objective may need.
    """

    id: str
    input: str
    expected: str | None = None
    rubric: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

EvalSet

Examples partitioned into train, validation, and optional test.

Raises:

Type	Description
DegenerateEvalError	If train or validation is empty, an id is duplicated within a split, or an id appears in more than one split.

Source code in recension/evalset.py

class EvalSet:
    """Examples partitioned into ``train``, ``validation``, and optional ``test``.

    Raises:
        DegenerateEvalError: If ``train`` or ``validation`` is empty, an id is
            duplicated within a split, or an id appears in more than one split.
    """

    def __init__(
        self,
        train: Sequence[Example],
        validation: Sequence[Example],
        test: Sequence[Example] | None = None,
    ) -> None:
        if not train:
            raise DegenerateEvalError("train split is empty")
        if not validation:
            raise DegenerateEvalError("validation split is empty")
        test = test or ()
        train_ids = [e.id for e in train]
        val_ids = [e.id for e in validation]
        test_ids = [e.id for e in test]
        for label, ids in (("train", train_ids), ("validation", val_ids), ("test", test_ids)):
            if len(set(ids)) != len(ids):
                raise DegenerateEvalError(f"duplicate example ids in {label} split")
        # Every pair of splits must be disjoint: a shared id would let an edit
        # "improve" on data it was also diagnosed or accepted on.
        for a_label, a_ids, b_label, b_ids in (
            ("train", train_ids, "validation", val_ids),
            ("train", train_ids, "test", test_ids),
            ("validation", val_ids, "test", test_ids),
        ):
            overlap = set(a_ids) & set(b_ids)
            if overlap:
                raise DegenerateEvalError(
                    f"example ids appear in both {a_label} and {b_label} splits "
                    f"(would contaminate the held-out signal): {sorted(overlap)}"
                )
        self._train = tuple(train)
        self._validation = tuple(validation)
        self._test = tuple(test)

    @property
    def train(self) -> tuple[Example, ...]:
        """Examples used to diagnose failures."""
        return self._train

    @property
    def validation(self) -> tuple[Example, ...]:
        """Held-out examples used to accept or reject candidates."""
        return self._validation

    @property
    def test(self) -> tuple[Example, ...]:
        """Locked examples scored once at the end; empty if none were supplied."""
        return self._test

    @classmethod
    def from_records(cls, records: Iterable[Mapping[str, Any]]) -> EvalSet:
        """Build an eval set from dict records.

        Each record needs ``id``, ``input``, and ``split`` (``"train"``,
        ``"validation"``, or the optional ``"test"``), plus optional
        ``expected`` and ``rubric``. Unknown keys land in
        :attr:`Example.metadata`.

        Raises:
            DegenerateEvalError: On a missing key, an unknown split value, or
                any split-integrity violation.
        """
        buckets: dict[str, list[Example]] = {"train": [], "validation": [], "test": []}
        for i, record in enumerate(records):
            try:
                example_id = str(record["id"])
                example_input = str(record["input"])
                split = record["split"]
            except KeyError as exc:
                raise DegenerateEvalError(f"record {i} is missing required key {exc}") from exc
            if split not in buckets:
                raise DegenerateEvalError(
                    f"record {example_id!r} has unknown split {split!r} "
                    "(expected 'train', 'validation', or 'test')"
                )
            buckets[split].append(
                Example(
                    id=example_id,
                    input=example_input,
                    expected=None if record.get("expected") is None else str(record["expected"]),
                    rubric=None if record.get("rubric") is None else str(record["rubric"]),
                    metadata={k: v for k, v in record.items() if k not in _KNOWN_KEYS},
                )
            )
        return cls(buckets["train"], buckets["validation"], buckets["test"])

    @classmethod
    def from_jsonl(cls, path: str | Path) -> EvalSet:
        """Build an eval set from a JSONL file of records (see ``from_records``).

        Raises:
            DegenerateEvalError: On a line that is not valid JSON (with the
                file and line number), or any split-integrity violation.
        """
        records = []
        p = Path(path)
        with p.open(encoding="utf-8") as fh:
            for lineno, raw in enumerate(fh, 1):
                line = raw.strip()
                if not line:
                    continue
                try:
                    records.append(json.loads(line))
                except json.JSONDecodeError as exc:
                    raise DegenerateEvalError(f"{p} line {lineno}: invalid JSON: {exc}") from exc
        return cls.from_records(records)

    def __repr__(self) -> str:
        return (
            f"EvalSet(train={len(self._train)}, validation={len(self._validation)}, "
            f"test={len(self._test)})"
        )

train property

train

Examples used to diagnose failures.

validation property

validation

Held-out examples used to accept or reject candidates.

test property

test

Locked examples scored once at the end; empty if none were supplied.

from_records classmethod

from_records(records)

Build an eval set from dict records.

Each record needs id, input, and split ("train", "validation", or the optional "test"), plus optional expected and rubric. Unknown keys land in :attr:Example.metadata.

Raises:

Type	Description
DegenerateEvalError	On a missing key, an unknown split value, or any split-integrity violation.

Source code in recension/evalset.py

@classmethod
def from_records(cls, records: Iterable[Mapping[str, Any]]) -> EvalSet:
    """Build an eval set from dict records.

    Each record needs ``id``, ``input``, and ``split`` (``"train"``,
    ``"validation"``, or the optional ``"test"``), plus optional
    ``expected`` and ``rubric``. Unknown keys land in
    :attr:`Example.metadata`.

    Raises:
        DegenerateEvalError: On a missing key, an unknown split value, or
            any split-integrity violation.
    """
    buckets: dict[str, list[Example]] = {"train": [], "validation": [], "test": []}
    for i, record in enumerate(records):
        try:
            example_id = str(record["id"])
            example_input = str(record["input"])
            split = record["split"]
        except KeyError as exc:
            raise DegenerateEvalError(f"record {i} is missing required key {exc}") from exc
        if split not in buckets:
            raise DegenerateEvalError(
                f"record {example_id!r} has unknown split {split!r} "
                "(expected 'train', 'validation', or 'test')"
            )
        buckets[split].append(
            Example(
                id=example_id,
                input=example_input,
                expected=None if record.get("expected") is None else str(record["expected"]),
                rubric=None if record.get("rubric") is None else str(record["rubric"]),
                metadata={k: v for k, v in record.items() if k not in _KNOWN_KEYS},
            )
        )
    return cls(buckets["train"], buckets["validation"], buckets["test"])

from_jsonl classmethod

from_jsonl(path)

Build an eval set from a JSONL file of records (see from_records).

Raises:

Type	Description
DegenerateEvalError	On a line that is not valid JSON (with the file and line number), or any split-integrity violation.

Source code in recension/evalset.py

@classmethod
def from_jsonl(cls, path: str | Path) -> EvalSet:
    """Build an eval set from a JSONL file of records (see ``from_records``).

    Raises:
        DegenerateEvalError: On a line that is not valid JSON (with the
            file and line number), or any split-integrity violation.
    """
    records = []
    p = Path(path)
    with p.open(encoding="utf-8") as fh:
        for lineno, raw in enumerate(fh, 1):
            line = raw.strip()
            if not line:
                continue
            try:
                records.append(json.loads(line))
            except json.JSONDecodeError as exc:
                raise DegenerateEvalError(f"{p} line {lineno}: invalid JSON: {exc}") from exc
    return cls.from_records(records)

Objectives

recension.objective

Objectives: how a model output is scored against an example.

An :class:Objective maps (model_output, example) to a float, higher is better, and aggregates per-example scores into one number (mean by default). Ships :class:ExactMatch, token-level :class:F1, and the model-graded :class:LLMJudge.

Objective

Bases: Protocol

Protocol every objective implements.

Attributes:

Name	Type	Description
name	str	Short identifier recorded in run records.
model_graded	bool	True when scoring itself calls a model (e.g. a judge). Model-graded acceptance is flagged in the audit record so a reviewer knows the metric is not reference-based.

Source code in recension/objective.py

@runtime_checkable
class Objective(Protocol):
    """Protocol every objective implements.

    Attributes:
        name: Short identifier recorded in run records.
        model_graded: True when scoring itself calls a model (e.g. a judge).
            Model-graded acceptance is flagged in the audit record so a
            reviewer knows the metric is not reference-based.
    """

    name: str
    model_graded: bool

    def score(self, model_output: str, example: Example) -> float:
        """Score one output against one example; higher is better."""
        ...

    def aggregate(self, scores: Sequence[float]) -> float:
        """Combine per-example scores into a single number."""
        ...

score

score(model_output, example)

Score one output against one example; higher is better.

Source code in recension/objective.py

def score(self, model_output: str, example: Example) -> float:
    """Score one output against one example; higher is better."""
    ...

aggregate

aggregate(scores)

Combine per-example scores into a single number.

Source code in recension/objective.py

def aggregate(self, scores: Sequence[float]) -> float:
    """Combine per-example scores into a single number."""
    ...

ExactMatch

1.0 if the output equals the expected value, else 0.0.

Comparison strips surrounding whitespace and, unless case_sensitive, casefolds both sides.

Source code in recension/objective.py

class ExactMatch:
    """1.0 if the output equals the expected value, else 0.0.

    Comparison strips surrounding whitespace and, unless ``case_sensitive``,
    casefolds both sides.
    """

    model_graded = False

    def __init__(self, *, case_sensitive: bool = False) -> None:
        self.name = "exact_match"
        self.case_sensitive = case_sensitive

    def _norm(self, text: str) -> str:
        text = text.strip()
        return text if self.case_sensitive else text.casefold()

    def score(self, model_output: str, example: Example) -> float:
        """1.0 on a normalized exact match with ``example.expected``, else 0.0."""
        expected = _require_expected(example, self.name)
        return 1.0 if self._norm(model_output) == self._norm(expected) else 0.0

    def aggregate(self, scores: Sequence[float]) -> float:
        """Mean of the per-example scores."""
        return _mean(scores)

score

score(model_output, example)

1.0 on a normalized exact match with example.expected, else 0.0.

Source code in recension/objective.py

def score(self, model_output: str, example: Example) -> float:
    """1.0 on a normalized exact match with ``example.expected``, else 0.0."""
    expected = _require_expected(example, self.name)
    return 1.0 if self._norm(model_output) == self._norm(expected) else 0.0

aggregate

aggregate(scores)

Mean of the per-example scores.

Source code in recension/objective.py

def aggregate(self, scores: Sequence[float]) -> float:
    """Mean of the per-example scores."""
    return _mean(scores)

F1

Token-level F1 between the output and the expected value.

Tokens are whitespace-separated, casefolded words, the conventional SQuAD-style metric. Returns 1.0 when both sides are empty.

Source code in recension/objective.py

class F1:
    """Token-level F1 between the output and the expected value.

    Tokens are whitespace-separated, casefolded words, the conventional
    SQuAD-style metric. Returns 1.0 when both sides are empty.
    """

    model_graded = False

    def __init__(self) -> None:
        self.name = "f1"

    def score(self, model_output: str, example: Example) -> float:
        """Token-level F1 against ``example.expected``."""
        expected = _require_expected(example, self.name)
        out_tokens = model_output.casefold().split()
        gold_tokens = expected.casefold().split()
        if not out_tokens and not gold_tokens:
            return 1.0
        if not out_tokens or not gold_tokens:
            return 0.0
        common = Counter(out_tokens) & Counter(gold_tokens)
        overlap = sum(common.values())
        if overlap == 0:
            return 0.0
        precision = overlap / len(out_tokens)
        recall = overlap / len(gold_tokens)
        return 2 * precision * recall / (precision + recall)

    def aggregate(self, scores: Sequence[float]) -> float:
        """Mean of the per-example scores."""
        return _mean(scores)

score

score(model_output, example)

Token-level F1 against example.expected.

Source code in recension/objective.py

def score(self, model_output: str, example: Example) -> float:
    """Token-level F1 against ``example.expected``."""
    expected = _require_expected(example, self.name)
    out_tokens = model_output.casefold().split()
    gold_tokens = expected.casefold().split()
    if not out_tokens and not gold_tokens:
        return 1.0
    if not out_tokens or not gold_tokens:
        return 0.0
    common = Counter(out_tokens) & Counter(gold_tokens)
    overlap = sum(common.values())
    if overlap == 0:
        return 0.0
    precision = overlap / len(out_tokens)
    recall = overlap / len(gold_tokens)
    return 2 * precision * recall / (precision + recall)

aggregate

aggregate(scores)

Mean of the per-example scores.

Source code in recension/objective.py

def aggregate(self, scores: Sequence[float]) -> float:
    """Mean of the per-example scores."""
    return _mean(scores)

MaxLength

Guard objective: 1.0 if the output is within max_chars, else 0.0.

Intended as a non-regression guard (ReflectiveOptimizer(guards=[...])): a candidate that starts producing over-long outputs lowers this score and is rejected even if it improves the primary metric. Reference-free, so no expected is needed.

Source code in recension/objective.py

class MaxLength:
    """Guard objective: 1.0 if the output is within ``max_chars``, else 0.0.

    Intended as a non-regression *guard* (``ReflectiveOptimizer(guards=[...])``):
    a candidate that starts producing over-long outputs lowers this score and is
    rejected even if it improves the primary metric. Reference-free, so no
    ``expected`` is needed.
    """

    model_graded = False

    def __init__(self, max_chars: int) -> None:
        self.name = f"max_length({max_chars})"
        self.max_chars = max_chars

    def score(self, model_output: str, example: Example) -> float:
        """1.0 if ``model_output`` is at most ``max_chars`` long, else 0.0."""
        return 1.0 if len(model_output) <= self.max_chars else 0.0

    def aggregate(self, scores: Sequence[float]) -> float:
        """Mean (the fraction of outputs within the limit)."""
        return _mean(scores)

score

score(model_output, example)

1.0 if model_output is at most max_chars long, else 0.0.

Source code in recension/objective.py

def score(self, model_output: str, example: Example) -> float:
    """1.0 if ``model_output`` is at most ``max_chars`` long, else 0.0."""
    return 1.0 if len(model_output) <= self.max_chars else 0.0

aggregate

aggregate(scores)

Mean (the fraction of outputs within the limit).

Source code in recension/objective.py

def aggregate(self, scores: Sequence[float]) -> float:
    """Mean (the fraction of outputs within the limit)."""
    return _mean(scores)

LLMJudge

Model-graded objective: scores outputs against a rubric, 0..1.

Intended as the held-out (validation) judge. Runs flagged as model_graded in the audit record, and every judge call counts toward the model-call budget. A per-example rubric overrides the judge-level one.

Raises:

Type	Description
DegenerateEvalError	If no rubric is available for an example, or the judge reply contains no parseable number, since a silent default would corrupt the measurement.

Source code in recension/objective.py

class LLMJudge:
    """Model-graded objective: scores outputs against a rubric, 0..1.

    Intended as the held-out (validation) judge. Runs flagged as
    ``model_graded`` in the audit record, and every judge call counts toward
    the model-call budget. A per-example ``rubric`` overrides the judge-level
    one.

    Raises:
        DegenerateEvalError: If no rubric is available for an example, or the
            judge reply contains no parseable number, since a silent default would
            corrupt the measurement.
    """

    model_graded = True

    def __init__(self, model: Model, rubric: str | None = None, *, max_tokens: int = 16) -> None:
        self.name = "llm_judge"
        self.model = model
        self.rubric = rubric
        self.max_tokens = max_tokens

    def score(self, model_output: str, example: Example) -> float:
        """Ask the judge model for a 0 to 10 grade and normalize it to 0..1."""
        rubric = example.rubric or self.rubric
        if rubric is None:
            raise DegenerateEvalError(
                f"no rubric for example {example.id!r}: pass one to LLMJudge or the example"
            )
        messages: list[Message] = [
            {
                "role": "user",
                "content": _JUDGE_PROMPT.format(
                    input=example.input, output=model_output, rubric=rubric
                ),
            }
        ]
        reply = self.model.complete(messages, max_tokens=self.max_tokens, temperature=0.0)
        match = _NUMBER.search(reply)
        if match is None:
            raise DegenerateEvalError(
                f"judge reply for example {example.id!r} contains no numeric score: {reply!r}"
            )
        grade = float(match.group())
        return max(0.0, min(grade, 10.0)) / 10.0

    def aggregate(self, scores: Sequence[float]) -> float:
        """Mean of the per-example scores."""
        return _mean(scores)

score

score(model_output, example)

Ask the judge model for a 0 to 10 grade and normalize it to 0..1.

Source code in recension/objective.py

def score(self, model_output: str, example: Example) -> float:
    """Ask the judge model for a 0 to 10 grade and normalize it to 0..1."""
    rubric = example.rubric or self.rubric
    if rubric is None:
        raise DegenerateEvalError(
            f"no rubric for example {example.id!r}: pass one to LLMJudge or the example"
        )
    messages: list[Message] = [
        {
            "role": "user",
            "content": _JUDGE_PROMPT.format(
                input=example.input, output=model_output, rubric=rubric
            ),
        }
    ]
    reply = self.model.complete(messages, max_tokens=self.max_tokens, temperature=0.0)
    match = _NUMBER.search(reply)
    if match is None:
        raise DegenerateEvalError(
            f"judge reply for example {example.id!r} contains no numeric score: {reply!r}"
        )
    grade = float(match.group())
    return max(0.0, min(grade, 10.0)) / 10.0

aggregate

aggregate(scores)

Mean of the per-example scores.

Source code in recension/objective.py

def aggregate(self, scores: Sequence[float]) -> float:
    """Mean of the per-example scores."""
    return _mean(scores)

Budget

recension.budget

The update-time compute dial.

Every knob the optimizer spends model calls on is here, caller-controlled. Nothing about update-time compute is hardcoded in the loop.

Budget dataclass

Caller-controlled limits on update-time compute.

Attributes:

Name	Type	Description
candidates_per_round	int	Distinct candidate edits generated per round.
rounds	int	Maximum optimization rounds.
diagnosis_depth	int	How many failed train examples are analyzed per round.
max_model_calls	int | None	Hard ceiling on total model calls for the run (task scoring, diagnosis, proposals, and judge calls all count). None means unlimited. The optimizer raises :class:~recension.exceptions.BudgetExceeded when the ceiling would be crossed.

Source code in recension/budget.py

@dataclass(frozen=True)
class Budget:
    """Caller-controlled limits on update-time compute.

    Attributes:
        candidates_per_round: Distinct candidate edits generated per round.
        rounds: Maximum optimization rounds.
        diagnosis_depth: How many failed train examples are analyzed per round.
        max_model_calls: Hard ceiling on total model calls for the run
            (task scoring, diagnosis, proposals, and judge calls all count).
            ``None`` means unlimited. The optimizer raises
            :class:`~recension.exceptions.BudgetExceeded` when the ceiling
            would be crossed.
    """

    candidates_per_round: int = 4
    rounds: int = 3
    diagnosis_depth: int = 1
    max_model_calls: int | None = None

    def __post_init__(self) -> None:
        if self.candidates_per_round < 1:
            raise ValueError("candidates_per_round must be >= 1")
        if self.rounds < 1:
            raise ValueError("rounds must be >= 1")
        if self.diagnosis_depth < 1:
            raise ValueError("diagnosis_depth must be >= 1")
        if self.max_model_calls is not None and self.max_model_calls < 1:
            raise ValueError("max_model_calls must be >= 1 or None")

    def to_dict(self) -> dict[str, Any]:
        """Plain-dict form for embedding in a run record."""
        return asdict(self)

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> Budget:
        """Inverse of :meth:`to_dict`."""
        return cls(**data)

to_dict

to_dict()

Plain-dict form for embedding in a run record.

Source code in recension/budget.py

def to_dict(self) -> dict[str, Any]:
    """Plain-dict form for embedding in a run record."""
    return asdict(self)

from_dict classmethod

from_dict(data)

Inverse of :meth:to_dict.

Source code in recension/budget.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Budget:
    """Inverse of :meth:`to_dict`."""
    return cls(**data)

Optimizer

recension.optimizer

The propose/test/accept loop.

:class:ReflectiveOptimizer holds the model fixed and optimizes the text artifact against held-out evidence: diagnose failures on the train split, propose distinct candidate edits, score them on the validation split, and accept only a candidate that beats the incumbent by min_improvement and survives the leakage checks. Every decision lands in the returned :class:~recension.record.RunRecord.

ReflectiveOptimizer

Optimizes a :class:TextArtifact against a frozen model.

Parameters:

Name	Type	Description	Default
artifact	TextArtifact	The text under optimization. Mutated in place: accepted candidates are committed to it with full provenance.	required
evalset	EvalSet	Held-out examples. Failures are diagnosed on train; acceptance is decided only on validation.	required
objective	Objective	Scoring function; higher is better.	required
model	Model	The frozen model, used for task execution, diagnosis, and proposals. The model is never changed by the optimizer; only the text is.	required
budget	Budget | None	Update-time compute limits. Defaults to Budget().	None
seed	int | None	Optional seed forwarded (derived per call) to the model for reproducible runs against :class:~recension.models.MockModel.	None
min_improvement	float	A candidate must beat the incumbent's validation score by more than this to be accepted.	1e-06
strict_leakage	bool	When True, a winning candidate that trips a leakage heuristic raises :class:LeakageDetected (with the partial record attached) instead of being accepted with flags.	False
stop_on_no_improvement	bool	Stop after the first round whose best candidate fails to beat the incumbent, instead of spending the remaining rounds.	True
overfit_gap	float	When the eval set has a test split, the run flags validation_overfit if the final validation score exceeds the test score by more than this gap.	0.1
accept_significant	bool	When True, a candidate is accepted only if its validation gain is statistically significant (a paired bootstrap CI on the per-example gain that excludes 0), not merely larger than min_improvement. Off by default, preserving 0.1.0 behavior.	False
alpha	float	Significance level for the bootstrap CI (default 0.05 = 95%).	0.05
bootstrap_resamples	int	Resamples for the significance bootstrap.	2000
slice_by	str | None	An Example.metadata key. When set, the record reports per-slice baseline-vs-final scores so a run that improves overall but regresses a subgroup is visible. None disables slicing.	None
slice_tolerance	float	A slice is announced as regressed when its score drops by more than this (default 0.0).	0.0
guards	Sequence[Objective]	Secondary objectives that must not regress. A candidate that improves the primary objective but lowers any guard's validation score (beyond guard_tolerance) is rejected. Guards are scored from the same model outputs, so cheap reference-free guards (e.g. :class:~recension.objective.MaxLength) add no model calls.	()
guard_tolerance	float	Allowed drop on a guard before it counts as a regression (default 0.0).	0.0
proposer	Proposer | None	The candidate generator. Defaults to the built-in :class:~recension.proposer.DefaultProposer; inject a custom :class:~recension.proposer.Proposer (e.g. one wrapping an external optimizer) to supply edits while recension keeps owning measurement, leakage, and the audit record.	None
task_max_tokens	int	max_tokens for task (example-scoring) calls.	1024
render	Renderer | None	Maps (artifact_text, example) to the messages sent to the model. See _default_render for the default convention.	None
on_progress	ProgressCallback | None	Optional callback receiving human-readable progress lines.	None

Raises:

Type	Description
BudgetExceeded	When budget.max_model_calls is hit (counting task, diagnosis, proposal, and judge calls). The partial run record, including the round in progress, is attached as .record.
LeakageDetected	In strict mode, when the winning candidate trips a leakage heuristic. Partial record attached as .record.

Source code in recension/optimizer.py

class ReflectiveOptimizer:
    """Optimizes a :class:`TextArtifact` against a frozen model.

    Args:
        artifact: The text under optimization. Mutated in place: accepted
            candidates are committed to it with full provenance.
        evalset: Held-out examples. Failures are diagnosed on ``train``;
            acceptance is decided only on ``validation``.
        objective: Scoring function; higher is better.
        model: The frozen model, used for task execution, diagnosis, and
            proposals. The model is never changed by the optimizer; only
            the text is.
        budget: Update-time compute limits. Defaults to ``Budget()``.
        seed: Optional seed forwarded (derived per call) to the model for
            reproducible runs against :class:`~recension.models.MockModel`.
        min_improvement: A candidate must beat the incumbent's validation
            score by more than this to be accepted.
        strict_leakage: When True, a winning candidate that trips a leakage
            heuristic raises :class:`LeakageDetected` (with the partial record
            attached) instead of being accepted with flags.
        stop_on_no_improvement: Stop after the first round whose best
            candidate fails to beat the incumbent, instead of spending the
            remaining rounds.
        overfit_gap: When the eval set has a ``test`` split, the run flags
            ``validation_overfit`` if the final validation score exceeds the
            test score by more than this gap.
        accept_significant: When True, a candidate is accepted only if its
            validation gain is *statistically significant* (a paired bootstrap
            CI on the per-example gain that excludes 0), not merely larger than
            ``min_improvement``. Off by default, preserving 0.1.0 behavior.
        alpha: Significance level for the bootstrap CI (default 0.05 = 95%).
        bootstrap_resamples: Resamples for the significance bootstrap.
        slice_by: An ``Example.metadata`` key. When set, the record reports
            per-slice baseline-vs-final scores so a run that improves overall
            but regresses a subgroup is visible. ``None`` disables slicing.
        slice_tolerance: A slice is announced as regressed when its score drops
            by more than this (default 0.0).
        guards: Secondary objectives that must not regress. A candidate that
            improves the primary objective but lowers any guard's validation
            score (beyond ``guard_tolerance``) is rejected. Guards are scored
            from the same model outputs, so cheap reference-free guards (e.g.
            :class:`~recension.objective.MaxLength`) add no model calls.
        guard_tolerance: Allowed drop on a guard before it counts as a
            regression (default 0.0).
        proposer: The candidate generator. Defaults to the built-in
            :class:`~recension.proposer.DefaultProposer`; inject a custom
            :class:`~recension.proposer.Proposer` (e.g. one wrapping an external
            optimizer) to supply edits while recension keeps owning measurement,
            leakage, and the audit record.
        task_max_tokens: ``max_tokens`` for task (example-scoring) calls.
        render: Maps ``(artifact_text, example)`` to the messages sent to the
            model. See ``_default_render`` for the default convention.
        on_progress: Optional callback receiving human-readable progress lines.

    Raises:
        BudgetExceeded: When ``budget.max_model_calls`` is hit (counting task,
            diagnosis, proposal, and judge calls). The partial run record,
            including the round in progress, is attached as ``.record``.
        LeakageDetected: In strict mode, when the winning candidate trips a
            leakage heuristic. Partial record attached as ``.record``.
    """

    def __init__(
        self,
        artifact: TextArtifact,
        evalset: EvalSet,
        objective: Objective,
        model: Model,
        budget: Budget | None = None,
        *,
        seed: int | None = None,
        min_improvement: float = 1e-6,
        strict_leakage: bool = False,
        stop_on_no_improvement: bool = True,
        overfit_gap: float = 0.1,
        accept_significant: bool = False,
        alpha: float = 0.05,
        bootstrap_resamples: int = 2000,
        slice_by: str | None = None,
        slice_tolerance: float = 0.0,
        guards: Sequence[Objective] = (),
        guard_tolerance: float = 0.0,
        proposer: Proposer | None = None,
        task_max_tokens: int = 1024,
        render: Renderer | None = None,
        on_progress: ProgressCallback | None = None,
    ) -> None:
        self.artifact = artifact
        self.evalset = evalset
        self.objective = objective
        self.budget = budget or Budget()
        self.proposer: Proposer = proposer or DefaultProposer()
        self.seed = seed
        self.min_improvement = min_improvement
        self.strict_leakage = strict_leakage
        self.stop_on_no_improvement = stop_on_no_improvement
        self.overfit_gap = overfit_gap
        self.accept_significant = accept_significant
        self.alpha = alpha
        self.bootstrap_resamples = bootstrap_resamples
        self.slice_by = slice_by
        self.slice_tolerance = slice_tolerance
        self.guards = tuple(guards)
        self.guard_tolerance = guard_tolerance
        self.task_max_tokens = task_max_tokens
        self._render: Renderer = render or _default_render
        self._on_progress = on_progress
        self._counter = _CallCounter()
        self._model = _GuardedModel(model, self.budget.max_model_calls, self._counter)

    # -- public -------------------------------------------------------------

    def run(self) -> RunRecord:
        """Execute the optimization loop and return the complete audit record."""
        started_at = datetime.now(UTC).isoformat()
        rounds: list[RoundRecord] = []
        baseline_version_id = self.artifact.current().version_id
        baseline_score = float("nan")
        incumbent_score = float("nan")
        restore = self._gate_objective_model()
        try:
            baseline_score, baseline_val_scores, baseline_val_outputs = self._score_set(
                self.artifact.text, self.evalset.validation
            )
            self._progress(f"baseline validation score: {baseline_score:.4f}")
            incumbent_score = baseline_score
            stopped_reason = "completed"
            train_cache: _TrainEval | None = None
            # Incumbent validation results threaded for the significance + guard gates.
            val_cache = _ValEval(baseline_val_scores, self._guard_scores(baseline_val_outputs))
            for round_index in range(1, self.budget.rounds + 1):
                round_record, incumbent_score, train_cache, val_cache = self._run_round(
                    round_index, incumbent_score, train_cache, val_cache
                )
                rounds.append(round_record)
                if round_record.accepted_version_id is None and self.stop_on_no_improvement:
                    stopped_reason = "no_improvement"
                    break
            slice_scores = self._slice_scores(baseline_val_scores, val_cache.primary_scores)
            # The locked test split is a final estimate, not optimization spend.
            # If the budget is exhausted before it can be scored, skip it (and say
            # so) rather than discarding the completed run as a budget failure.
            try:
                final_test_score, gap, overfit = self._score_test(incumbent_score)
            except BudgetExceeded:
                self._progress(
                    "budget exhausted before the locked test split could be scored; skipping it"
                )
                final_test_score, gap, overfit = None, None, False
            return self._build_record(
                started_at, baseline_version_id, baseline_score, rounds,
                incumbent_score, stopped_reason,
                final_test_score=final_test_score,
                test_validation_gap=gap,
                validation_overfit=overfit,
                slice_scores=slice_scores,
            )
        except _BudgetInRound as exc:
            rounds.append(exc.round_record)
            partial = self._build_record(
                started_at, baseline_version_id, baseline_score, rounds,
                incumbent_score, "budget_exceeded",
            )
            raise BudgetExceeded(str(exc.original), record=partial) from exc.original
        except BudgetExceeded as exc:
            # Hit during baseline scoring, before any round started.
            partial = self._build_record(
                started_at, baseline_version_id, baseline_score, rounds,
                incumbent_score, "budget_exceeded",
            )
            raise BudgetExceeded(str(exc), record=partial) from exc
        except _StrictLeakage as exc:
            rounds.append(exc.round_record)
            partial = self._build_record(
                started_at, baseline_version_id, baseline_score, rounds,
                incumbent_score, "leakage_detected",
            )
            raise LeakageDetected(
                f"candidate {exc.candidate_id} tripped leakage heuristics in strict mode: "
                + "; ".join(exc.flags),
                record=partial,
            ) from None
        finally:
            if restore is not None:
                restore()

    # -- internals ----------------------------------------------------------

    def _gate_objective_model(self) -> Callable[[], None] | None:
        """Route a model-graded objective's judge calls through the budget gate.

        # DESIGN NOTE: a model-graded objective (LLMJudge) holds its own model,
        # so its calls would otherwise bypass the optimizer's ceiling. We swap
        # that model for a guard sharing this run's counter for the duration of
        # run(), restoring the caller's original object afterwards (no permanent
        # mutation, and no assumption that task and judge share one instance).
        """
        objective: Any = self.objective
        if not getattr(objective, "model_graded", False) or not hasattr(objective, "model"):
            return None
        original = objective.model
        objective.model = _GuardedModel(original, self.budget.max_model_calls, self._counter)

        def restore() -> None:
            objective.model = original

        return restore

    def _run_round(
        self,
        round_index: int,
        incumbent_score: float,
        train_cache: _TrainEval | None,
        val_cache: _ValEval,
    ) -> tuple[RoundRecord, float, _TrainEval, _ValEval]:
        t0 = time.perf_counter()
        calls_before = self._model.call_count
        usage_before = self._counter.usage
        incumbent_text = self.artifact.text
        incumbent_version_id = self.artifact.current().version_id

        # Progressive state, captured into a partial record if the budget runs
        # out partway through the round (so the audit trail keeps what was done).
        train_score = float("nan")
        train_scores: list[float] = []
        train_outputs: list[str] = []
        failures: list[FailureCase] = []
        diagnosis = ""
        scored: list[tuple[str, str, float, list[str]]] = []  # (id, text, score, flags)
        val_scores_by_id: dict[str, list[float]] = {}  # per-example validation scores
        guard_scores_by_id: dict[str, list[float]] = {}  # per-candidate guard aggregates
        accepted_train_score: float | None = None
        significance: SignificanceRecord | None = None
        significance_id: str | None = None
        guard_records: tuple[GuardScore, ...] = ()
        guard_id: str | None = None
        next_val_cache = val_cache

        try:
            if train_cache is not None and train_cache.text == incumbent_text:
                self._progress(f"round {round_index}: reusing cached incumbent train score")
                train_score = train_cache.score
                train_scores = train_cache.scores
                train_outputs = train_cache.outputs
            else:
                self._progress(f"round {round_index}: scoring incumbent on train")
                train_score, train_scores, train_outputs = self._score_set(
                    incumbent_text, self.evalset.train
                )

            failures = self._collect_failures(train_scores, train_outputs)
            diagnosis = self.proposer.diagnose(
                self._model,
                incumbent_text,
                failures,
                seed=self._seed_for(f"diagnose:{round_index}"),
            )
            self._progress(f"round {round_index}: diagnosis: {diagnosis}")

            candidate_texts = self.proposer.propose(
                self._model,
                incumbent_text,
                diagnosis,
                self.budget.candidates_per_round,
                seed=self._seed_for(f"propose:{round_index}"),
            )
            if len(candidate_texts) < self.budget.candidates_per_round:
                self._progress(
                    f"round {round_index}: only {len(candidate_texts)} distinct candidate(s) "
                    f"of {self.budget.candidates_per_round} requested"
                )

            for i, text in enumerate(candidate_texts, 1):
                candidate_id = f"r{round_index}-c{i}"
                score, val_scores, val_outputs = self._score_set(text, self.evalset.validation)
                val_scores_by_id[candidate_id] = val_scores
                guard_scores_by_id[candidate_id] = self._guard_scores(val_outputs)
                flags = [str(f) for f in check_candidate(text, incumbent_text, self.evalset)]
                scored.append((candidate_id, text, score, flags))
                self._progress(
                    f"round {round_index}: candidate {candidate_id} scored {score:.4f}"
                )

            accepted_id: str | None = None
            accepted_version_id: str | None = None
            new_incumbent_score = incumbent_score
            next_cache = _TrainEval(incumbent_text, train_score, train_scores, train_outputs)

            if scored:
                best_id, best_text, best_score, _ = max(scored, key=lambda s: s[2])
                beats_threshold = best_score > incumbent_score + self.min_improvement
                accept = beats_threshold

                # Significance gate: a candidate that clears the margin must also
                # show a statistically significant per-example validation gain.
                if beats_threshold and self.accept_significant:
                    boot = paired_bootstrap(
                        val_cache.primary_scores,
                        val_scores_by_id[best_id],
                        alpha=self.alpha,
                        n_resamples=self.bootstrap_resamples,
                        seed=self._seed_for(f"significance:{round_index}:{best_id}"),
                    )
                    significance = SignificanceRecord(
                        mean_difference=boot.mean_difference,
                        ci_low=boot.ci_low,
                        ci_high=boot.ci_high,
                        alpha=boot.alpha,
                        significant=boot.significant,
                    )
                    significance_id = best_id
                    if not boot.significant:
                        accept = False
                        self._progress(
                            f"round {round_index}: {best_id} beat the incumbent by "
                            f"{best_score - incumbent_score:+.4f} but the gain is not "
                            f"significant ({int((1 - boot.alpha) * 100)}% CI "
                            f"[{boot.ci_low:+.4f}, {boot.ci_high:+.4f}]); rejecting"
                        )

                # Guard gate: the best candidate must not regress any guard objective.
                if beats_threshold and self.guards:
                    incumbent_guards = val_cache.guard_scores
                    candidate_guards = guard_scores_by_id[best_id]
                    guard_records = tuple(
                        GuardScore(guard.name, incumbent_guards[gi], candidate_guards[gi])
                        for gi, guard in enumerate(self.guards)
                    )
                    guard_id = best_id
                    regressed = [
                        gr.name
                        for gr in guard_records
                        if gr.candidate_score < gr.incumbent_score - self.guard_tolerance
                    ]
                    if regressed and accept:
                        accept = False
                        self._progress(
                            f"round {round_index}: {best_id} regressed guard(s) "
                            f"{', '.join(regressed)}; rejecting"
                        )

                if accept:
                    cand_train_score, cand_train_scores, cand_train_outputs = self._score_set(
                        best_text, self.evalset.train
                    )
                    accepted_train_score = cand_train_score
                    full_flags = [
                        str(f)
                        for f in check_candidate(
                            best_text,
                            incumbent_text,
                            self.evalset,
                            train_gain=cand_train_score - train_score,
                            validation_gain=best_score - incumbent_score,
                        )
                    ]
                    scored = [
                        (cid, text, score, full_flags if cid == best_id else flags)
                        for cid, text, score, flags in scored
                    ]
                    if full_flags and self.strict_leakage:
                        round_record = self._round_record(
                            round_index, incumbent_version_id, incumbent_score, train_score,
                            failures, diagnosis, scored, None, None, None,
                            significance_id, significance, guard_id, guard_records,
                            calls_before, usage_before, t0,
                        )
                        raise _StrictLeakage(round_record, best_id, full_flags)
                    if full_flags:
                        self._progress(
                            f"round {round_index}: accepting {best_id} WITH leakage flags: "
                            + "; ".join(full_flags)
                        )
                    accepted_id = best_id
                    version = self.artifact.commit(
                        best_text,
                        Provenance(
                            diagnosis=diagnosis,
                            failure_example_ids=tuple(f.example.id for f in failures),
                            incumbent_score=incumbent_score,
                            candidate_score=best_score,
                            rejected_candidates=tuple(
                                RejectedCandidate(cid, text, score, tuple(flags))
                                for cid, text, score, flags in scored
                                if cid != best_id
                            ),
                        ),
                    )
                    accepted_version_id = version.version_id
                    new_incumbent_score = best_score
                    next_cache = _TrainEval(
                        best_text, cand_train_score, cand_train_scores, cand_train_outputs
                    )
                    next_val_cache = _ValEval(
                        val_scores_by_id[best_id], guard_scores_by_id[best_id]
                    )
                    self._progress(
                        f"round {round_index}: accepted {best_id} as version "
                        f"{accepted_version_id}"
                    )
                elif not beats_threshold:
                    self._progress(f"round {round_index}: no candidate beat the incumbent")
        except BudgetExceeded as exc:
            partial = self._round_record(
                round_index, incumbent_version_id, incumbent_score, train_score,
                failures, diagnosis, scored, None, None, None,
                significance_id, significance, guard_id, guard_records,
                calls_before, usage_before, t0,
            )
            raise _BudgetInRound(partial, exc) from exc

        round_record = self._round_record(
            round_index, incumbent_version_id, incumbent_score, train_score,
            failures, diagnosis, scored, accepted_id, accepted_version_id,
            accepted_train_score, significance_id, significance, guard_id, guard_records,
            calls_before, usage_before, t0,
        )
        return round_record, new_incumbent_score, next_cache, next_val_cache

    def _round_record(
        self,
        round_index: int,
        incumbent_version_id: str,
        incumbent_score: float,
        train_score: float,
        failures: list[FailureCase],
        diagnosis: str,
        scored: list[tuple[str, str, float, list[str]]],
        accepted_id: str | None,
        accepted_version_id: str | None,
        accepted_train_score: float | None,
        significance_id: str | None,
        significance: SignificanceRecord | None,
        guard_id: str | None,
        guard_records: tuple[GuardScore, ...],
        calls_before: int,
        usage_before: TokenUsage,
        t0: float,
    ) -> RoundRecord:
        incumbent_text_at_start = self.artifact.get(incumbent_version_id).text
        usage = self._counter.usage
        return RoundRecord(
            round_index=round_index,
            incumbent_version_id=incumbent_version_id,
            incumbent_validation_score=incumbent_score,
            train_score=train_score,
            failure_example_ids=tuple(f.example.id for f in failures),
            diagnosis=diagnosis,
            candidates=tuple(
                CandidateRecord(
                    candidate_id=cid,
                    text=text,
                    validation_score=score,
                    diff=_diff_texts(incumbent_text_at_start, text, incumbent_version_id, cid),
                    leakage_flags=tuple(flags),
                    accepted=cid == accepted_id,
                    train_score=accepted_train_score if cid == accepted_id else None,
                    significance=significance if cid == significance_id else None,
                    guard_scores=guard_records if cid == guard_id else (),
                )
                for cid, text, score, flags in scored
            ),
            accepted_version_id=accepted_version_id,
            model_calls_used=self._model.call_count - calls_before,
            elapsed_seconds=time.perf_counter() - t0,
            input_tokens=usage.input_tokens - usage_before.input_tokens,
            output_tokens=usage.output_tokens - usage_before.output_tokens,
        )

    def _build_record(
        self,
        started_at: str,
        baseline_version_id: str,
        baseline_score: float,
        rounds: list[RoundRecord],
        final_score: float,
        stopped_reason: str,
        *,
        final_test_score: float | None = None,
        test_validation_gap: float | None = None,
        validation_overfit: bool = False,
        slice_scores: list[SliceScore] | None = None,
    ) -> RunRecord:
        return RunRecord(
            artifact=self.artifact.to_dict(),
            objective_name=self.objective.name,
            model_graded=self.objective.model_graded,
            seed=self.seed,
            budget=self.budget.to_dict(),
            baseline_version_id=baseline_version_id,
            baseline_score=baseline_score,
            rounds=list(rounds),
            final_version_id=self.artifact.current().version_id,
            final_score=final_score,
            final_test_score=final_test_score,
            test_validation_gap=test_validation_gap,
            validation_overfit=validation_overfit,
            slice_scores=list(slice_scores) if slice_scores else [],
            total_input_tokens=self._counter.usage.input_tokens,
            total_output_tokens=self._counter.usage.output_tokens,
            total_model_calls=self._model.call_count,
            stopped_reason=stopped_reason,
            started_at=started_at,
            finished_at=datetime.now(UTC).isoformat(),
        )

    def _score_test(self, final_validation_score: float) -> tuple[float | None, float | None, bool]:
        """Score the final incumbent on the locked test split, exactly once.

        Returns (test_score, validation/test gap, overfit_flag). All-None/False
        when no test split was provided.
        """
        if not self.evalset.test:
            return None, None, False
        self._progress("scoring final incumbent on the locked test split")
        test_score, _, _ = self._score_set(self.artifact.text, self.evalset.test)
        gap = final_validation_score - test_score
        overfit = gap > self.overfit_gap
        if overfit:
            self._progress(
                f"validation/test gap {gap:.4f} exceeds {self.overfit_gap:.4f}: "
                "the validation score may be optimistic (overfit to validation)"
            )
        return test_score, gap, overfit

    def _guard_scores(self, outputs: list[str]) -> list[float]:
        """Aggregate score of each guard over the validation outputs (guard order)."""
        validation = self.evalset.validation
        return [
            guard.aggregate(
                [guard.score(out, ex) for out, ex in zip(outputs, validation, strict=True)]
            )
            for guard in self.guards
        ]

    def _slice_scores(
        self, baseline_val_scores: list[float], final_val_scores: list[float]
    ) -> list[SliceScore]:
        """Per-slice baseline-vs-final scores, grouped by the ``slice_by`` key."""
        slice_by = self.slice_by
        if slice_by is None:
            return []

        def group(scores: list[float]) -> dict[str, list[float]]:
            groups: dict[str, list[float]] = {}
            for example, score in zip(self.evalset.validation, scores, strict=True):
                key = str(example.metadata.get(slice_by, "(unset)"))
                groups.setdefault(key, []).append(score)
            return groups

        base_groups = group(baseline_val_scores)
        final_groups = group(final_val_scores)
        result: list[SliceScore] = []
        for key in sorted(base_groups):
            base = self.objective.aggregate(base_groups[key])
            final = self.objective.aggregate(final_groups[key])
            result.append(SliceScore(slice=key, n=len(base_groups[key]),
                                     baseline_score=base, final_score=final))
            if final < base - self.slice_tolerance:
                self._progress(f"slice {key!r} regressed: {base:.4f} -> {final:.4f}")
        return result

    def _score_set(
        self, artifact_text: str, examples: Sequence[Example]
    ) -> tuple[float, list[float], list[str]]:
        """Score ``artifact_text`` on ``examples``; returns (aggregate, scores, outputs)."""
        scores: list[float] = []
        outputs: list[str] = []
        for example in examples:
            output = self._model.complete(
                self._render(artifact_text, example),
                max_tokens=self.task_max_tokens,
                temperature=0.0,
                seed=self._seed_for(f"task:{example.id}"),
            )
            outputs.append(output)
            scores.append(self.objective.score(output, example))
        return self.objective.aggregate(scores), scores, outputs

    def _collect_failures(
        self, train_scores: list[float], train_outputs: list[str]
    ) -> list[FailureCase]:
        indexed = sorted(enumerate(train_scores), key=lambda pair: (pair[1], pair[0]))
        worst = indexed[: self.budget.diagnosis_depth]
        return [
            FailureCase(
                example=self.evalset.train[i], output=train_outputs[i], score=score
            )
            for i, score in worst
        ]

    def _seed_for(self, tag: str) -> int | None:
        if self.seed is None:
            return None
        return zlib.crc32(f"{self.seed}:{tag}".encode())

    def _progress(self, message: str) -> None:
        if self._on_progress is not None:
            self._on_progress(message)

run

run()

Execute the optimization loop and return the complete audit record.

Source code in recension/optimizer.py

def run(self) -> RunRecord:
    """Execute the optimization loop and return the complete audit record."""
    started_at = datetime.now(UTC).isoformat()
    rounds: list[RoundRecord] = []
    baseline_version_id = self.artifact.current().version_id
    baseline_score = float("nan")
    incumbent_score = float("nan")
    restore = self._gate_objective_model()
    try:
        baseline_score, baseline_val_scores, baseline_val_outputs = self._score_set(
            self.artifact.text, self.evalset.validation
        )
        self._progress(f"baseline validation score: {baseline_score:.4f}")
        incumbent_score = baseline_score
        stopped_reason = "completed"
        train_cache: _TrainEval | None = None
        # Incumbent validation results threaded for the significance + guard gates.
        val_cache = _ValEval(baseline_val_scores, self._guard_scores(baseline_val_outputs))
        for round_index in range(1, self.budget.rounds + 1):
            round_record, incumbent_score, train_cache, val_cache = self._run_round(
                round_index, incumbent_score, train_cache, val_cache
            )
            rounds.append(round_record)
            if round_record.accepted_version_id is None and self.stop_on_no_improvement:
                stopped_reason = "no_improvement"
                break
        slice_scores = self._slice_scores(baseline_val_scores, val_cache.primary_scores)
        # The locked test split is a final estimate, not optimization spend.
        # If the budget is exhausted before it can be scored, skip it (and say
        # so) rather than discarding the completed run as a budget failure.
        try:
            final_test_score, gap, overfit = self._score_test(incumbent_score)
        except BudgetExceeded:
            self._progress(
                "budget exhausted before the locked test split could be scored; skipping it"
            )
            final_test_score, gap, overfit = None, None, False
        return self._build_record(
            started_at, baseline_version_id, baseline_score, rounds,
            incumbent_score, stopped_reason,
            final_test_score=final_test_score,
            test_validation_gap=gap,
            validation_overfit=overfit,
            slice_scores=slice_scores,
        )
    except _BudgetInRound as exc:
        rounds.append(exc.round_record)
        partial = self._build_record(
            started_at, baseline_version_id, baseline_score, rounds,
            incumbent_score, "budget_exceeded",
        )
        raise BudgetExceeded(str(exc.original), record=partial) from exc.original
    except BudgetExceeded as exc:
        # Hit during baseline scoring, before any round started.
        partial = self._build_record(
            started_at, baseline_version_id, baseline_score, rounds,
            incumbent_score, "budget_exceeded",
        )
        raise BudgetExceeded(str(exc), record=partial) from exc
    except _StrictLeakage as exc:
        rounds.append(exc.round_record)
        partial = self._build_record(
            started_at, baseline_version_id, baseline_score, rounds,
            incumbent_score, "leakage_detected",
        )
        raise LeakageDetected(
            f"candidate {exc.candidate_id} tripped leakage heuristics in strict mode: "
            + "; ".join(exc.flags),
            record=partial,
        ) from None
    finally:
        if restore is not None:
            restore()

score_artifact

score_artifact(artifact_text, examples, objective, model, *, render=None, task_max_tokens=1024)

Score one artifact on a set of examples, without running an optimization.

The aggregate objective score of artifact_text over examples against the frozen model. Used by recension check to compare the current artifact to a recorded baseline (a prompt regression test).

Source code in recension/optimizer.py

def score_artifact(
    artifact_text: str,
    examples: Sequence[Example],
    objective: Objective,
    model: Model,
    *,
    render: Renderer | None = None,
    task_max_tokens: int = 1024,
) -> float:
    """Score one artifact on a set of examples, without running an optimization.

    The aggregate objective score of ``artifact_text`` over ``examples`` against
    the frozen ``model``. Used by ``recension check`` to compare the current
    artifact to a recorded baseline (a prompt regression test).
    """
    render = render or _default_render
    scores = [
        objective.score(
            model.complete(render(artifact_text, example), max_tokens=task_max_tokens),
            example,
        )
        for example in examples
    ]
    return objective.aggregate(scores)

Proposer

recension.proposer

Candidate generation: diagnose failures, propose distinct revisions.

The proposer turns observed failures into a structured hypothesis (:func:diagnose) and then into genuinely different candidate edits (:func:propose). Distinctness matters: comparing four rewordings of one idea tests nothing, so near-duplicate candidates are rejected and regenerated.

FailureCase dataclass

One failed train example: what went in, what came out, how it scored.

Source code in recension/proposer.py

@dataclass(frozen=True)
class FailureCase:
    """One failed train example: what went in, what came out, how it scored."""

    example: Example
    output: str
    score: float

Proposer

Bases: Protocol

Pluggable candidate generator: diagnose failures, then propose edits.

The built-in heuristic proposer (:class:DefaultProposer) is one implementation. A custom proposer, for example one wrapping an external optimizer such as DSPy or GEPA, can be injected with ReflectiveOptimizer(proposer=...) without changing the artifact, evalset, or record abstractions: recension keeps owning versioning, held-out measurement, leakage detection, and the audit record; the proposer only supplies the candidate edits. This is the seam that lets recension act as the measurement-and-governance layer on top of any optimizer.

Source code in recension/proposer.py

@runtime_checkable
class Proposer(Protocol):
    """Pluggable candidate generator: diagnose failures, then propose edits.

    The built-in heuristic proposer (:class:`DefaultProposer`) is one
    implementation. A custom proposer, for example one wrapping an external
    optimizer such as DSPy or GEPA, can be injected with
    ``ReflectiveOptimizer(proposer=...)`` **without changing the artifact,
    evalset, or record abstractions**: recension keeps owning versioning,
    held-out measurement, leakage detection, and the audit record; the proposer
    only supplies the candidate edits. This is the seam that lets recension act
    as the measurement-and-governance layer on top of any optimizer.
    """

    def diagnose(
        self,
        model: Model,
        artifact_text: str,
        failures: list[FailureCase],
        *,
        seed: int | None = None,
    ) -> str:
        """Return a short hypothesis about why ``artifact_text`` failed."""
        ...

    def propose(
        self,
        model: Model,
        artifact_text: str,
        diagnosis: str,
        n: int,
        *,
        seed: int | None = None,
    ) -> list[str]:
        """Return up to ``n`` distinct candidate revisions of ``artifact_text``."""
        ...

diagnose

diagnose(model, artifact_text, failures, *, seed=None)

Return a short hypothesis about why artifact_text failed.

Source code in recension/proposer.py

def diagnose(
    self,
    model: Model,
    artifact_text: str,
    failures: list[FailureCase],
    *,
    seed: int | None = None,
) -> str:
    """Return a short hypothesis about why ``artifact_text`` failed."""
    ...

propose

propose(model, artifact_text, diagnosis, n, *, seed=None)

Return up to n distinct candidate revisions of artifact_text.

Source code in recension/proposer.py

def propose(
    self,
    model: Model,
    artifact_text: str,
    diagnosis: str,
    n: int,
    *,
    seed: int | None = None,
) -> list[str]:
    """Return up to ``n`` distinct candidate revisions of ``artifact_text``."""
    ...

DefaultProposer

The built-in proposer: the module-level :func:diagnose/:func:propose.

Source code in recension/proposer.py

class DefaultProposer:
    """The built-in proposer: the module-level :func:`diagnose`/:func:`propose`."""

    def diagnose(
        self,
        model: Model,
        artifact_text: str,
        failures: list[FailureCase],
        *,
        seed: int | None = None,
    ) -> str:
        return diagnose(model, artifact_text, failures, seed=seed)

    def propose(
        self,
        model: Model,
        artifact_text: str,
        diagnosis: str,
        n: int,
        *,
        seed: int | None = None,
    ) -> list[str]:
        return propose(model, artifact_text, diagnosis, n, seed=seed)

CallableProposer

Adapt plain functions into a :class:Proposer.

Wrap an external optimizer's propose function (and optionally a diagnose function) without writing a class. Either may be None for diagnose, in which case the built-in diagnosis is used. The seam for "bring your own optimizer": recension governs the run; your function proposes the edits.

Source code in recension/proposer.py

class CallableProposer:
    """Adapt plain functions into a :class:`Proposer`.

    Wrap an external optimizer's propose function (and optionally a diagnose
    function) without writing a class. Either may be ``None`` for diagnose, in
    which case the built-in diagnosis is used. The seam for "bring your own
    optimizer": recension governs the run; your function proposes the edits.
    """

    def __init__(self, propose_fn: ProposeFn, diagnose_fn: DiagnoseFn | None = None) -> None:
        self._propose_fn = propose_fn
        self._diagnose_fn = diagnose_fn

    def diagnose(
        self,
        model: Model,
        artifact_text: str,
        failures: list[FailureCase],
        *,
        seed: int | None = None,
    ) -> str:
        if self._diagnose_fn is not None:
            return self._diagnose_fn(model, artifact_text, failures, seed)
        return diagnose(model, artifact_text, failures, seed=seed)

    def propose(
        self,
        model: Model,
        artifact_text: str,
        diagnosis: str,
        n: int,
        *,
        seed: int | None = None,
    ) -> list[str]:
        return self._propose_fn(model, artifact_text, diagnosis, n, seed)

diagnose

diagnose(model, artifact_text, failures, *, max_tokens=1024, seed=None)

Ask the model why the artifact produced these failures.

Returns the model's hypothesis as free text (recorded verbatim in the round record).

Source code in recension/proposer.py

def diagnose(
    model: Model,
    artifact_text: str,
    failures: list[FailureCase],
    *,
    max_tokens: int = 1024,
    seed: int | None = None,
) -> str:
    """Ask the model why the artifact produced these failures.

    Returns the model's hypothesis as free text (recorded verbatim in the
    round record).
    """
    cases = []
    for i, case in enumerate(failures, 1):
        expected = (
            f"\nexpected: {case.example.expected}" if case.example.expected is not None else ""
        )
        cases.append(
            f"<case index={i} example_id={case.example.id!r} score={case.score:.4f}>\n"
            f"input: {case.example.input}\n"
            f"output: {case.output}{expected}\n"
            f"</case>"
        )
    messages: list[Message] = [
        {"role": "system", "content": _DIAGNOSE_SYSTEM},
        {
            "role": "user",
            "content": (
                f"<artifact>\n{artifact_text}\n</artifact>\n\n"
                f"Failed cases:\n\n" + "\n\n".join(cases)
            ),
        },
    ]
    return model.complete(messages, max_tokens=max_tokens, temperature=0.0, seed=seed).strip()

propose

propose(model, artifact_text, diagnosis, n, *, max_tokens=4096, seed=None)

Generate up to n distinct candidate revisions of the artifact.

Near-duplicates (of the incumbent or of each other, by difflib.SequenceMatcher ratio) are rejected and regenerated, up to ATTEMPTS_PER_CANDIDATE attempts per requested candidate. If the model cannot produce n distinct candidates within that allowance, the distinct subset found so far is returned; the optimizer records how many candidates each round actually compared, so a shortfall is visible in the audit record rather than silently padded.

Source code in recension/proposer.py

def propose(
    model: Model,
    artifact_text: str,
    diagnosis: str,
    n: int,
    *,
    max_tokens: int = 4096,
    seed: int | None = None,
) -> list[str]:
    """Generate up to ``n`` distinct candidate revisions of the artifact.

    Near-duplicates (of the incumbent or of each other, by
    ``difflib.SequenceMatcher`` ratio) are rejected and regenerated, up to
    ``ATTEMPTS_PER_CANDIDATE`` attempts per requested candidate. If the model
    cannot produce ``n`` distinct candidates within that allowance, the
    distinct subset found so far is returned; the optimizer records how many
    candidates each round actually compared, so a shortfall is visible in the
    audit record rather than silently padded.
    """
    candidates: list[str] = []
    attempts = 0
    max_attempts = n * ATTEMPTS_PER_CANDIDATE
    while len(candidates) < n and attempts < max_attempts:
        attempts += 1
        call_seed = None if seed is None else seed + attempts
        reply = model.complete(
            _proposal_messages(artifact_text, diagnosis, len(candidates) + 1, n, candidates),
            max_tokens=max_tokens,
            temperature=0.0,
            seed=call_seed,
        )
        text = extract_candidate(reply)
        if not text.strip():
            continue
        if _is_near_duplicate(text, artifact_text) or any(
            _is_near_duplicate(text, existing) for existing in candidates
        ):
            continue
        candidates.append(text)
    return candidates

extract_candidate

extract_candidate(reply)

Pull the revised artifact out of a proposal reply.

Prefers the <revised_artifact> tags the prompt asks for; falls back to the whole reply (stripped) when a model ignores the tagging instruction.

Source code in recension/proposer.py

def extract_candidate(reply: str) -> str:
    """Pull the revised artifact out of a proposal reply.

    Prefers the ``<revised_artifact>`` tags the prompt asks for; falls back to
    the whole reply (stripped) when a model ignores the tagging instruction.
    """
    match = _CANDIDATE_RE.search(reply)
    if match:
        return match.group(1)
    return reply.strip()

Leakage heuristics

recension.leakage

Leakage and overfitting heuristics for candidate artifacts.

These checks are heuristics, not proofs. They catch the two cheapest ways an edit can game a held-out score:

1. Verbatim validation spans: the candidate text embeds a long span copied from a validation example (its input, expected output, or rubric). That is memorization of the held-out set, not generalization.
2. Implausible gain: the candidate's validation gain is large while its train gain is flat or negative. Honest improvements usually move both; a validation-only jump suggests the edit exploits validation-specific cues.

A tripped heuristic produces a :class:LeakageFlag. By default flags are surfaced in the run record, not silently enforced; the optimizer's strict mode turns them into :class:~recension.exceptions.LeakageDetected.

LeakageFlag dataclass

One tripped heuristic, with enough detail to review it.

Source code in recension/leakage.py

@dataclass(frozen=True)
class LeakageFlag:
    """One tripped heuristic, with enough detail to review it."""

    kind: str
    detail: str

    def __str__(self) -> str:
        return f"{self.kind}: {self.detail}"

check_candidate

check_candidate(candidate_text, incumbent_text, evalset, *, train_gain=None, validation_gain=None, min_span_length=DEFAULT_MIN_SPAN_LENGTH, min_validation_gain=DEFAULT_MIN_VALIDATION_GAIN, gain_ratio=DEFAULT_GAIN_RATIO)

Run all leakage heuristics against one candidate.

Parameters:

Name	Type	Description	Default
candidate_text	str	The proposed artifact text.	required
incumbent_text	str	The current artifact text. Spans already present in the incumbent are not re-flagged, only newly introduced validation content counts.	required
evalset	EvalSet	Source of the validation examples to scan for.	required
train_gain	float | None	Candidate train score minus incumbent train score, if measured. Both gains are required for the implausible-gain check.	None
validation_gain	float | None	Candidate validation score minus incumbent validation score, if measured.	None
min_span_length	int	Shortest copied span (characters) considered leakage.	DEFAULT_MIN_SPAN_LENGTH
min_validation_gain	float	Validation gain below which the implausible-gain heuristic never fires (small gains are noise, not leakage).	DEFAULT_MIN_VALIDATION_GAIN
gain_ratio	float	Fire when validation_gain > gain_ratio * train_gain (for positive train gain).	DEFAULT_GAIN_RATIO

Returns:

Type	Description
list[LeakageFlag]	All tripped flags, empty if the candidate looks clean.

Source code in recension/leakage.py

def check_candidate(
    candidate_text: str,
    incumbent_text: str,
    evalset: EvalSet,
    *,
    train_gain: float | None = None,
    validation_gain: float | None = None,
    min_span_length: int = DEFAULT_MIN_SPAN_LENGTH,
    min_validation_gain: float = DEFAULT_MIN_VALIDATION_GAIN,
    gain_ratio: float = DEFAULT_GAIN_RATIO,
) -> list[LeakageFlag]:
    """Run all leakage heuristics against one candidate.

    Args:
        candidate_text: The proposed artifact text.
        incumbent_text: The current artifact text. Spans already present in
            the incumbent are not re-flagged, only *newly introduced*
            validation content counts.
        evalset: Source of the validation examples to scan for.
        train_gain: Candidate train score minus incumbent train score, if
            measured. Both gains are required for the implausible-gain check.
        validation_gain: Candidate validation score minus incumbent
            validation score, if measured.
        min_span_length: Shortest copied span (characters) considered leakage.
        min_validation_gain: Validation gain below which the implausible-gain
            heuristic never fires (small gains are noise, not leakage).
        gain_ratio: Fire when ``validation_gain > gain_ratio * train_gain``
            (for positive train gain).

    Returns:
        All tripped flags, empty if the candidate looks clean.
    """
    flags = _verbatim_span_flags(candidate_text, incumbent_text, evalset, min_span_length)
    if train_gain is not None and validation_gain is not None:
        flag = _implausible_gain_flag(train_gain, validation_gain, min_validation_gain, gain_ratio)
        if flag is not None:
            flags.append(flag)
    return flags

Statistics

recension.stats

Seeded bootstrap statistics for honest acceptance decisions.

The optimizer can require an accepted candidate's validation gain to be statistically significant, not merely larger than an epsilon, so a candidate that wins by noise is rejected. This module provides the paired-difference bootstrap that backs that gate. It is pure stdlib (random.Random) and fully deterministic given a seed, so a seeded run against MockModel stays reproducible.

BootstrapResult dataclass

A paired-difference bootstrap of two aligned per-example score vectors.

Attributes:

Name	Type	Description
mean_difference	float	Mean of candidate_i - incumbent_i over the validation examples (the observed per-example gain).
ci_low	float	Lower bound of the 1 - alpha confidence interval on the mean difference.
ci_high	float	Upper bound of that interval.
alpha	float	The significance level used (e.g. 0.05 for a 95% interval).
n_resamples	int	Number of bootstrap resamples drawn.

Source code in recension/stats.py

@dataclass(frozen=True)
class BootstrapResult:
    """A paired-difference bootstrap of two aligned per-example score vectors.

    Attributes:
        mean_difference: Mean of ``candidate_i - incumbent_i`` over the
            validation examples (the observed per-example gain).
        ci_low: Lower bound of the ``1 - alpha`` confidence interval on the
            mean difference.
        ci_high: Upper bound of that interval.
        alpha: The significance level used (e.g. ``0.05`` for a 95% interval).
        n_resamples: Number of bootstrap resamples drawn.
    """

    mean_difference: float
    ci_low: float
    ci_high: float
    alpha: float
    n_resamples: int

    @property
    def significant(self) -> bool:
        """True when the interval excludes 0, i.e. the gain is significantly positive."""
        return self.ci_low > 0.0

significant property

significant

True when the interval excludes 0, i.e. the gain is significantly positive.

paired_bootstrap

paired_bootstrap(incumbent, candidate, *, alpha=0.05, n_resamples=2000, seed=None)

Bootstrap a confidence interval on the mean paired score difference.

Resamples the per-example differences candidate_i - incumbent_i with replacement to estimate a percentile confidence interval on their mean.

Parameters:

Name	Type	Description	Default
incumbent	Sequence[float]	Per-example validation scores of the incumbent.	required
candidate	Sequence[float]	Per-example validation scores of the candidate, aligned with incumbent (same examples, same order).	required
alpha	float	Significance level; the interval is 1 - alpha.	0.05
n_resamples	int	Number of resamples.	2000
seed	int | None	Seed for the resampling RNG; pass one for reproducibility.	None

Raises:

Type	Description
ValueError	If the vectors differ in length or are empty.

Source code in recension/stats.py

def paired_bootstrap(
    incumbent: Sequence[float],
    candidate: Sequence[float],
    *,
    alpha: float = 0.05,
    n_resamples: int = 2000,
    seed: int | None = None,
) -> BootstrapResult:
    """Bootstrap a confidence interval on the mean paired score difference.

    Resamples the per-example differences ``candidate_i - incumbent_i`` with
    replacement to estimate a percentile confidence interval on their mean.

    Args:
        incumbent: Per-example validation scores of the incumbent.
        candidate: Per-example validation scores of the candidate, aligned with
            ``incumbent`` (same examples, same order).
        alpha: Significance level; the interval is ``1 - alpha``.
        n_resamples: Number of resamples.
        seed: Seed for the resampling RNG; pass one for reproducibility.

    Raises:
        ValueError: If the vectors differ in length or are empty.
    """
    if len(incumbent) != len(candidate):
        raise ValueError("paired bootstrap needs equal-length score vectors")
    n = len(incumbent)
    if n == 0:
        raise ValueError("cannot bootstrap an empty score vector")
    diffs = [c - i for i, c in zip(incumbent, candidate, strict=True)]
    mean_difference = sum(diffs) / n

    rng = random.Random(seed)
    resample_means: list[float] = []
    for _ in range(n_resamples):
        total = 0.0
        for _ in range(n):
            total += diffs[rng.randrange(n)]
        resample_means.append(total / n)
    resample_means.sort()

    lo_idx = int((alpha / 2) * n_resamples)
    hi_idx = int((1 - alpha / 2) * n_resamples) - 1
    hi_idx = max(lo_idx, min(hi_idx, n_resamples - 1))
    return BootstrapResult(
        mean_difference=mean_difference,
        ci_low=resample_means[lo_idx],
        ci_high=resample_means[hi_idx],
        alpha=alpha,
        n_resamples=n_resamples,
    )

Run records

recension.record

The audit record: a complete, serializable history of an optimization run.

A :class:RunRecord must be complete enough that a reviewer who did not run the optimization can reconstruct every decision: the baseline, every round's diagnosis, every candidate (accepted and rejected) with its scores and leakage flags, the diffs, the model-call counts, and why the run stopped. The full artifact (with version history) is embedded so the record stands alone.

GuardScore dataclass

A guard objective's incumbent-vs-candidate score for the best candidate.

Recorded when the optimizer runs with guards=[...], so a reviewer sees why a candidate was held back (or that it cleared the guards).

Source code in recension/record.py

@dataclass(frozen=True)
class GuardScore:
    """A guard objective's incumbent-vs-candidate score for the best candidate.

    Recorded when the optimizer runs with ``guards=[...]``, so a reviewer sees
    why a candidate was held back (or that it cleared the guards).
    """

    name: str
    incumbent_score: float
    candidate_score: float

    @property
    def regressed(self) -> bool:
        """True when the candidate scores worse than the incumbent on this guard."""
        return self.candidate_score < self.incumbent_score

regressed property

regressed

True when the candidate scores worse than the incumbent on this guard.

SliceScore dataclass

Baseline vs final score for one subgroup of the validation set.

Recorded per distinct value of the optimizer's slice_by metadata key, so a run that improves overall but regresses a segment is visible rather than averaged away.

Source code in recension/record.py

@dataclass(frozen=True)
class SliceScore:
    """Baseline vs final score for one subgroup of the validation set.

    Recorded per distinct value of the optimizer's ``slice_by`` metadata key, so
    a run that improves overall but regresses a segment is visible rather than
    averaged away.
    """

    slice: str
    n: int
    baseline_score: float
    final_score: float

    @property
    def regressed(self) -> bool:
        """True when this slice scored worse at the end than at the start."""
        return self.final_score < self.baseline_score

regressed property

regressed

True when this slice scored worse at the end than at the start.

SignificanceRecord dataclass

The significance test applied to a candidate's validation gain.

Recorded for the best candidate of a round when the optimizer runs with accept_significant=True, so a reviewer can see not just the score delta but whether it cleared the confidence bar.

Attributes:

Name	Type	Description
mean_difference	float	Mean per-example validation gain over the incumbent.
ci_low	float	Lower bound of the bootstrap confidence interval on the gain.
ci_high	float	Upper bound of that interval.
alpha	float	Significance level (the interval is 1 - alpha).
significant	bool	True when the interval excludes 0 (gain significantly > 0).

Source code in recension/record.py

@dataclass(frozen=True)
class SignificanceRecord:
    """The significance test applied to a candidate's validation gain.

    Recorded for the best candidate of a round when the optimizer runs with
    ``accept_significant=True``, so a reviewer can see not just the score delta
    but whether it cleared the confidence bar.

    Attributes:
        mean_difference: Mean per-example validation gain over the incumbent.
        ci_low: Lower bound of the bootstrap confidence interval on the gain.
        ci_high: Upper bound of that interval.
        alpha: Significance level (the interval is ``1 - alpha``).
        significant: True when the interval excludes 0 (gain significantly > 0).
    """

    mean_difference: float
    ci_low: float
    ci_high: float
    alpha: float
    significant: bool

CandidateRecord dataclass

One candidate edit considered in a round.

Attributes:

Name	Type	Description
candidate_id	str	Stable id within the run (e.g. "r1-c2").
text	str	The full candidate artifact text.
validation_score	float | None	Aggregate held-out score, or None if scoring was cut short (e.g. budget exhausted).
diff	str	Unified diff against the incumbent at proposal time.
leakage_flags	tuple[str, ...]	Human-readable descriptions of tripped heuristics.
accepted	bool	Whether this candidate became the new incumbent.
train_score	float | None	Aggregate train score of this candidate, populated only for the accepted candidate (the only one re-scored on train, to check the implausible-gain heuristic). None otherwise. With RoundRecord.train_score (the incumbent's train score) this lets a reviewer reconstruct the train-vs-validation gain that drove the leakage decision.
significance	SignificanceRecord | None	The significance test on this candidate's validation gain, populated for the round's best candidate when the run used accept_significant. None otherwise.

Source code in recension/record.py

@dataclass(frozen=True)
class CandidateRecord:
    """One candidate edit considered in a round.

    Attributes:
        candidate_id: Stable id within the run (e.g. ``"r1-c2"``).
        text: The full candidate artifact text.
        validation_score: Aggregate held-out score, or ``None`` if scoring
            was cut short (e.g. budget exhausted).
        diff: Unified diff against the incumbent at proposal time.
        leakage_flags: Human-readable descriptions of tripped heuristics.
        accepted: Whether this candidate became the new incumbent.
        train_score: Aggregate train score of this candidate, populated only
            for the accepted candidate (the only one re-scored on train, to
            check the implausible-gain heuristic). ``None`` otherwise. With
            ``RoundRecord.train_score`` (the incumbent's train score) this
            lets a reviewer reconstruct the train-vs-validation gain that
            drove the leakage decision.
        significance: The significance test on this candidate's validation
            gain, populated for the round's best candidate when the run used
            ``accept_significant``. ``None`` otherwise.
    """

    candidate_id: str
    text: str
    validation_score: float | None
    diff: str
    leakage_flags: tuple[str, ...] = ()
    accepted: bool = False
    train_score: float | None = None
    significance: SignificanceRecord | None = None
    guard_scores: tuple[GuardScore, ...] = ()

RoundRecord dataclass

Everything that happened in one optimization round.

Source code in recension/record.py

@dataclass(frozen=True)
class RoundRecord:
    """Everything that happened in one optimization round."""

    round_index: int
    incumbent_version_id: str
    incumbent_validation_score: float
    train_score: float
    failure_example_ids: tuple[str, ...]
    diagnosis: str
    candidates: tuple[CandidateRecord, ...]
    accepted_version_id: str | None
    model_calls_used: int
    elapsed_seconds: float
    input_tokens: int = 0
    output_tokens: int = 0

RunRecord dataclass

Complete, serializable history of one ReflectiveOptimizer.run().

Attributes:

Name	Type	Description
artifact	dict[str, Any]	Full snapshot (TextArtifact.to_dict()) of the artifact after the run, version history and provenance included.
objective_name	str	The objective used for all scoring.
model_graded	bool	True if the objective itself calls a model (:class:~recension.objective.LLMJudge); flagged so reviewers know the acceptance metric is not reference-based.
seed	int | None	The seed the optimizer was constructed with, if any.
budget	dict[str, Any]	Budget.to_dict() snapshot.
baseline_version_id	str	Incumbent version at the start of the run.
baseline_score	float	Held-out validation score of the baseline.
rounds	list[RoundRecord]	One :class:RoundRecord per executed round.
final_version_id	str	Incumbent version at the end of the run.
final_score	float	Held-out validation score of the final incumbent.
final_test_score	float | None	Score of the final incumbent on the locked test split, computed exactly once; None if no test split was given.
test_validation_gap	float | None	final_score - final_test_score when a test split exists; a large positive gap suggests the validation score is optimistic (overfit to validation across rounds). None without a test split.
validation_overfit	bool	True when test_validation_gap exceeds the optimizer's overfit_gap threshold; surfaced, not hidden.
total_input_tokens	int	Total input tokens reported by the model across the run (0 if the model does not report usage).
total_output_tokens	int	Total output tokens reported across the run.
total_model_calls	int	All model calls spent, including judge calls.
stopped_reason	str	Why the run ended ("completed", "no_improvement", "budget_exceeded").
started_at	str	ISO 8601 timestamp.
finished_at	str	ISO 8601 timestamp.

Source code in recension/record.py

@dataclass
class RunRecord:
    """Complete, serializable history of one ``ReflectiveOptimizer.run()``.

    Attributes:
        artifact: Full snapshot (``TextArtifact.to_dict()``) of the artifact
            after the run, version history and provenance included.
        objective_name: The objective used for all scoring.
        model_graded: True if the objective itself calls a model
            (:class:`~recension.objective.LLMJudge`); flagged so reviewers
            know the acceptance metric is not reference-based.
        seed: The seed the optimizer was constructed with, if any.
        budget: ``Budget.to_dict()`` snapshot.
        baseline_version_id: Incumbent version at the start of the run.
        baseline_score: Held-out validation score of the baseline.
        rounds: One :class:`RoundRecord` per executed round.
        final_version_id: Incumbent version at the end of the run.
        final_score: Held-out validation score of the final incumbent.
        final_test_score: Score of the final incumbent on the locked test
            split, computed exactly once; ``None`` if no test split was given.
        test_validation_gap: ``final_score - final_test_score`` when a test
            split exists; a large positive gap suggests the validation score
            is optimistic (overfit to validation across rounds). ``None``
            without a test split.
        validation_overfit: True when ``test_validation_gap`` exceeds the
            optimizer's ``overfit_gap`` threshold; surfaced, not hidden.
        total_input_tokens: Total input tokens reported by the model across the
            run (0 if the model does not report usage).
        total_output_tokens: Total output tokens reported across the run.
        total_model_calls: All model calls spent, including judge calls.
        stopped_reason: Why the run ended (``"completed"``,
            ``"no_improvement"``, ``"budget_exceeded"``).
        started_at: ISO 8601 timestamp.
        finished_at: ISO 8601 timestamp.
    """

    artifact: dict[str, Any]
    objective_name: str
    model_graded: bool
    seed: int | None
    budget: dict[str, Any]
    baseline_version_id: str
    baseline_score: float
    rounds: list[RoundRecord] = field(default_factory=list)
    final_version_id: str = ""
    final_score: float = 0.0
    final_test_score: float | None = None
    test_validation_gap: float | None = None
    validation_overfit: bool = False
    slice_scores: list[SliceScore] = field(default_factory=list)
    total_input_tokens: int = 0
    total_output_tokens: int = 0
    total_model_calls: int = 0
    stopped_reason: str = ""
    started_at: str = ""
    finished_at: str = ""

    # -- serialization ------------------------------------------------------

    def to_dict(self) -> dict[str, Any]:
        """Plain-dict form of the whole record."""
        data = asdict(self)
        return data

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> RunRecord:
        """Inverse of :meth:`to_dict`."""
        rounds = [
            RoundRecord(
                round_index=r["round_index"],
                incumbent_version_id=r["incumbent_version_id"],
                incumbent_validation_score=_restore_score(r["incumbent_validation_score"]),
                train_score=_restore_score(r["train_score"]),
                failure_example_ids=tuple(r["failure_example_ids"]),
                diagnosis=r["diagnosis"],
                candidates=tuple(
                    CandidateRecord(
                        candidate_id=c["candidate_id"],
                        text=c["text"],
                        validation_score=c["validation_score"],
                        diff=c["diff"],
                        leakage_flags=tuple(c["leakage_flags"]),
                        accepted=c["accepted"],
                        train_score=c.get("train_score"),
                        significance=_significance_from_dict(c.get("significance")),
                        guard_scores=tuple(
                            GuardScore(
                                name=g["name"],
                                incumbent_score=g["incumbent_score"],
                                candidate_score=g["candidate_score"],
                            )
                            for g in c.get("guard_scores", [])
                        ),
                    )
                    for c in r["candidates"]
                ),
                accepted_version_id=r["accepted_version_id"],
                model_calls_used=r["model_calls_used"],
                elapsed_seconds=r["elapsed_seconds"],
                input_tokens=r.get("input_tokens", 0),
                output_tokens=r.get("output_tokens", 0),
            )
            for r in data["rounds"]
        ]
        return cls(
            artifact=data["artifact"],
            objective_name=data["objective_name"],
            model_graded=data["model_graded"],
            seed=data["seed"],
            budget=data["budget"],
            baseline_version_id=data["baseline_version_id"],
            baseline_score=_restore_score(data["baseline_score"]),
            rounds=rounds,
            final_version_id=data["final_version_id"],
            final_score=_restore_score(data["final_score"]),
            final_test_score=data.get("final_test_score"),
            test_validation_gap=data.get("test_validation_gap"),
            validation_overfit=data.get("validation_overfit", False),
            slice_scores=[
                SliceScore(
                    slice=s["slice"],
                    n=s["n"],
                    baseline_score=s["baseline_score"],
                    final_score=s["final_score"],
                )
                for s in data.get("slice_scores", [])
            ],
            total_input_tokens=data.get("total_input_tokens", 0),
            total_output_tokens=data.get("total_output_tokens", 0),
            total_model_calls=data["total_model_calls"],
            stopped_reason=data["stopped_reason"],
            started_at=data["started_at"],
            finished_at=data["finished_at"],
        )

    def to_json(self, *, indent: int | None = 2) -> str:
        """Serialize to JSON (valid even for partial records with uncomputed scores)."""
        return json.dumps(_json_safe(self.to_dict()), indent=indent, ensure_ascii=False)

    @classmethod
    def from_json(cls, payload: str) -> RunRecord:
        """Inverse of :meth:`to_json`."""
        return cls.from_dict(json.loads(payload))

    def save(self, path: str | Path) -> None:
        """Write the record to a JSON file."""
        Path(path).write_text(self.to_json(), encoding="utf-8")

    @classmethod
    def load(cls, path: str | Path) -> RunRecord:
        """Read a record from a JSON file."""
        return cls.from_json(Path(path).read_text(encoding="utf-8"))

    # -- reporting ----------------------------------------------------------

    def restored_artifact(self) -> TextArtifact:
        """Rehydrate the embedded artifact (for diffs and inspection)."""
        return TextArtifact.from_dict(self.artifact)

    # -- integrity ----------------------------------------------------------

    def fingerprint(self) -> str:
        """Deterministic SHA-256 over the canonical record JSON.

        Two records with identical content produce the same fingerprint. Store
        it (or a signature of it) somewhere trusted to detect later tampering
        with any field; the artifact lineage is additionally self-verifiable
        via :meth:`verify`.
        """
        canonical = json.dumps(_json_safe(self.to_dict()), sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    def verify(self) -> list[str]:
        """Integrity problems with the embedded artifact's version chain.

        Empty when intact. Because version ids are content-addressed, this
        catches tampering with a version's text or id without needing any
        external reference. (Tampering with non-versioned fields such as a
        recorded score is caught instead by comparing :meth:`fingerprint` or a
        signature against a trusted copy.)
        """
        return self.restored_artifact().verify()

    def sign(self, key: str) -> str:
        """HMAC-SHA256 of the fingerprint with ``key`` (hex), for signed records."""
        digest = hmac.new(
            key.encode("utf-8"), self.fingerprint().encode("utf-8"), hashlib.sha256
        )
        return digest.hexdigest()

    def verify_signature(self, key: str, signature: str) -> bool:
        """Constant-time check that ``signature`` matches :meth:`sign` for ``key``."""
        return hmac.compare_digest(self.sign(key), signature)

    def summary(self) -> str:
        """Human-readable account of the run: baseline, rounds, diffs, scores."""
        lines = [
            f"artifact: {self.artifact.get('name', 'artifact')}",
            f"objective: {self.objective_name}"
            + (" (model-graded)" if self.model_graded else ""),
            f"baseline: version {self.baseline_version_id}  "
            f"validation score {self.baseline_score:.4f}",
        ]
        for r in self.rounds:
            lines.append("")
            lines.append(
                f"round {r.round_index}: train score {r.train_score:.4f}, "
                f"failures analyzed: {', '.join(r.failure_example_ids) or 'none'}"
            )
            lines.append(f"  diagnosis: {r.diagnosis}")
            for c in r.candidates:
                score = "n/a" if c.validation_score is None else f"{c.validation_score:.4f}"
                status = "ACCEPTED" if c.accepted else "rejected"
                flags = f"  flags: {', '.join(c.leakage_flags)}" if c.leakage_flags else ""
                sig = ""
                if c.significance is not None:
                    s = c.significance
                    verdict = "significant" if s.significant else "NOT significant"
                    sig = (
                        f"  [{verdict}: gain {s.mean_difference:+.4f}, "
                        f"{int((1 - s.alpha) * 100)}% CI [{s.ci_low:+.4f}, {s.ci_high:+.4f}]]"
                    )
                guards = ""
                if c.guard_scores:
                    parts = [
                        f"{g.name} {g.incumbent_score:.4f}->{g.candidate_score:.4f}"
                        + ("!" if g.regressed else "")
                        for g in c.guard_scores
                    ]
                    guards = "  guards: " + ", ".join(parts)
                lines.append(
                    f"  candidate {c.candidate_id}: {score}  [{status}]{flags}{sig}{guards}"
                )
            if r.accepted_version_id:
                lines.append(f"  new incumbent: version {r.accepted_version_id}")
                accepted = next((c for c in r.candidates if c.accepted), None)
                if accepted is None:
                    # Defensive: a complete record always has the accepted
                    # candidate, but a hand-edited or truncated one might not.
                    lines.append("  (accepted candidate not found in record)")
                elif accepted.diff:
                    lines.append("  diff:")
                    lines.extend(f"    {line}" for line in accepted.diff.splitlines())
        lines.append("")
        lines.append(
            f"final: version {self.final_version_id}  "
            f"validation score {self.final_score:.4f}  "
            f"({self.baseline_score:.4f} -> {self.final_score:.4f})"
        )
        if self.final_test_score is not None:
            gap = self.test_validation_gap if self.test_validation_gap is not None else 0.0
            warn = (
                "  [WARNING: possible overfitting to validation]"
                if self.validation_overfit
                else ""
            )
            lines.append(
                f"test (locked, scored once): {self.final_test_score:.4f}  "
                f"validation/test gap {gap:.4f}{warn}"
            )
        if self.slice_scores:
            lines.append("slices:")
            for sl in self.slice_scores:
                mark = "  [REGRESSED]" if sl.regressed else ""
                lines.append(
                    f"  {sl.slice} (n={sl.n}): "
                    f"{sl.baseline_score:.4f} -> {sl.final_score:.4f}{mark}"
                )
        lines.append(f"model calls: {self.total_model_calls}")
        if self.total_input_tokens or self.total_output_tokens:
            lines.append(
                f"tokens: {self.total_input_tokens} in / {self.total_output_tokens} out"
            )
        lines.append(f"stopped: {self.stopped_reason}")
        return "\n".join(lines)

to_dict

to_dict()

Plain-dict form of the whole record.

Source code in recension/record.py

def to_dict(self) -> dict[str, Any]:
    """Plain-dict form of the whole record."""
    data = asdict(self)
    return data

from_dict classmethod

from_dict(data)

Inverse of :meth:to_dict.

Source code in recension/record.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> RunRecord:
    """Inverse of :meth:`to_dict`."""
    rounds = [
        RoundRecord(
            round_index=r["round_index"],
            incumbent_version_id=r["incumbent_version_id"],
            incumbent_validation_score=_restore_score(r["incumbent_validation_score"]),
            train_score=_restore_score(r["train_score"]),
            failure_example_ids=tuple(r["failure_example_ids"]),
            diagnosis=r["diagnosis"],
            candidates=tuple(
                CandidateRecord(
                    candidate_id=c["candidate_id"],
                    text=c["text"],
                    validation_score=c["validation_score"],
                    diff=c["diff"],
                    leakage_flags=tuple(c["leakage_flags"]),
                    accepted=c["accepted"],
                    train_score=c.get("train_score"),
                    significance=_significance_from_dict(c.get("significance")),
                    guard_scores=tuple(
                        GuardScore(
                            name=g["name"],
                            incumbent_score=g["incumbent_score"],
                            candidate_score=g["candidate_score"],
                        )
                        for g in c.get("guard_scores", [])
                    ),
                )
                for c in r["candidates"]
            ),
            accepted_version_id=r["accepted_version_id"],
            model_calls_used=r["model_calls_used"],
            elapsed_seconds=r["elapsed_seconds"],
            input_tokens=r.get("input_tokens", 0),
            output_tokens=r.get("output_tokens", 0),
        )
        for r in data["rounds"]
    ]
    return cls(
        artifact=data["artifact"],
        objective_name=data["objective_name"],
        model_graded=data["model_graded"],
        seed=data["seed"],
        budget=data["budget"],
        baseline_version_id=data["baseline_version_id"],
        baseline_score=_restore_score(data["baseline_score"]),
        rounds=rounds,
        final_version_id=data["final_version_id"],
        final_score=_restore_score(data["final_score"]),
        final_test_score=data.get("final_test_score"),
        test_validation_gap=data.get("test_validation_gap"),
        validation_overfit=data.get("validation_overfit", False),
        slice_scores=[
            SliceScore(
                slice=s["slice"],
                n=s["n"],
                baseline_score=s["baseline_score"],
                final_score=s["final_score"],
            )
            for s in data.get("slice_scores", [])
        ],
        total_input_tokens=data.get("total_input_tokens", 0),
        total_output_tokens=data.get("total_output_tokens", 0),
        total_model_calls=data["total_model_calls"],
        stopped_reason=data["stopped_reason"],
        started_at=data["started_at"],
        finished_at=data["finished_at"],
    )

to_json

to_json(*, indent=2)

Serialize to JSON (valid even for partial records with uncomputed scores).

Source code in recension/record.py

def to_json(self, *, indent: int | None = 2) -> str:
    """Serialize to JSON (valid even for partial records with uncomputed scores)."""
    return json.dumps(_json_safe(self.to_dict()), indent=indent, ensure_ascii=False)

from_json classmethod

from_json(payload)

Inverse of :meth:to_json.

Source code in recension/record.py

@classmethod
def from_json(cls, payload: str) -> RunRecord:
    """Inverse of :meth:`to_json`."""
    return cls.from_dict(json.loads(payload))

save

save(path)

Write the record to a JSON file.

Source code in recension/record.py

def save(self, path: str | Path) -> None:
    """Write the record to a JSON file."""
    Path(path).write_text(self.to_json(), encoding="utf-8")

load classmethod

load(path)

Read a record from a JSON file.

Source code in recension/record.py

@classmethod
def load(cls, path: str | Path) -> RunRecord:
    """Read a record from a JSON file."""
    return cls.from_json(Path(path).read_text(encoding="utf-8"))

restored_artifact

restored_artifact()

Rehydrate the embedded artifact (for diffs and inspection).

Source code in recension/record.py

def restored_artifact(self) -> TextArtifact:
    """Rehydrate the embedded artifact (for diffs and inspection)."""
    return TextArtifact.from_dict(self.artifact)

fingerprint

fingerprint()

Deterministic SHA-256 over the canonical record JSON.

Two records with identical content produce the same fingerprint. Store it (or a signature of it) somewhere trusted to detect later tampering with any field; the artifact lineage is additionally self-verifiable via :meth:verify.

Source code in recension/record.py

def fingerprint(self) -> str:
    """Deterministic SHA-256 over the canonical record JSON.

    Two records with identical content produce the same fingerprint. Store
    it (or a signature of it) somewhere trusted to detect later tampering
    with any field; the artifact lineage is additionally self-verifiable
    via :meth:`verify`.
    """
    canonical = json.dumps(_json_safe(self.to_dict()), sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

verify

verify()

Integrity problems with the embedded artifact's version chain.

Empty when intact. Because version ids are content-addressed, this catches tampering with a version's text or id without needing any external reference. (Tampering with non-versioned fields such as a recorded score is caught instead by comparing :meth:fingerprint or a signature against a trusted copy.)

Source code in recension/record.py

def verify(self) -> list[str]:
    """Integrity problems with the embedded artifact's version chain.

    Empty when intact. Because version ids are content-addressed, this
    catches tampering with a version's text or id without needing any
    external reference. (Tampering with non-versioned fields such as a
    recorded score is caught instead by comparing :meth:`fingerprint` or a
    signature against a trusted copy.)
    """
    return self.restored_artifact().verify()

sign

sign(key)

HMAC-SHA256 of the fingerprint with key (hex), for signed records.

Source code in recension/record.py

def sign(self, key: str) -> str:
    """HMAC-SHA256 of the fingerprint with ``key`` (hex), for signed records."""
    digest = hmac.new(
        key.encode("utf-8"), self.fingerprint().encode("utf-8"), hashlib.sha256
    )
    return digest.hexdigest()

verify_signature

verify_signature(key, signature)

Constant-time check that signature matches :meth:sign for key.

Source code in recension/record.py

def verify_signature(self, key: str, signature: str) -> bool:
    """Constant-time check that ``signature`` matches :meth:`sign` for ``key``."""
    return hmac.compare_digest(self.sign(key), signature)

summary

summary()

Human-readable account of the run: baseline, rounds, diffs, scores.

Source code in recension/record.py

def summary(self) -> str:
    """Human-readable account of the run: baseline, rounds, diffs, scores."""
    lines = [
        f"artifact: {self.artifact.get('name', 'artifact')}",
        f"objective: {self.objective_name}"
        + (" (model-graded)" if self.model_graded else ""),
        f"baseline: version {self.baseline_version_id}  "
        f"validation score {self.baseline_score:.4f}",
    ]
    for r in self.rounds:
        lines.append("")
        lines.append(
            f"round {r.round_index}: train score {r.train_score:.4f}, "
            f"failures analyzed: {', '.join(r.failure_example_ids) or 'none'}"
        )
        lines.append(f"  diagnosis: {r.diagnosis}")
        for c in r.candidates:
            score = "n/a" if c.validation_score is None else f"{c.validation_score:.4f}"
            status = "ACCEPTED" if c.accepted else "rejected"
            flags = f"  flags: {', '.join(c.leakage_flags)}" if c.leakage_flags else ""
            sig = ""
            if c.significance is not None:
                s = c.significance
                verdict = "significant" if s.significant else "NOT significant"
                sig = (
                    f"  [{verdict}: gain {s.mean_difference:+.4f}, "
                    f"{int((1 - s.alpha) * 100)}% CI [{s.ci_low:+.4f}, {s.ci_high:+.4f}]]"
                )
            guards = ""
            if c.guard_scores:
                parts = [
                    f"{g.name} {g.incumbent_score:.4f}->{g.candidate_score:.4f}"
                    + ("!" if g.regressed else "")
                    for g in c.guard_scores
                ]
                guards = "  guards: " + ", ".join(parts)
            lines.append(
                f"  candidate {c.candidate_id}: {score}  [{status}]{flags}{sig}{guards}"
            )
        if r.accepted_version_id:
            lines.append(f"  new incumbent: version {r.accepted_version_id}")
            accepted = next((c for c in r.candidates if c.accepted), None)
            if accepted is None:
                # Defensive: a complete record always has the accepted
                # candidate, but a hand-edited or truncated one might not.
                lines.append("  (accepted candidate not found in record)")
            elif accepted.diff:
                lines.append("  diff:")
                lines.extend(f"    {line}" for line in accepted.diff.splitlines())
    lines.append("")
    lines.append(
        f"final: version {self.final_version_id}  "
        f"validation score {self.final_score:.4f}  "
        f"({self.baseline_score:.4f} -> {self.final_score:.4f})"
    )
    if self.final_test_score is not None:
        gap = self.test_validation_gap if self.test_validation_gap is not None else 0.0
        warn = (
            "  [WARNING: possible overfitting to validation]"
            if self.validation_overfit
            else ""
        )
        lines.append(
            f"test (locked, scored once): {self.final_test_score:.4f}  "
            f"validation/test gap {gap:.4f}{warn}"
        )
    if self.slice_scores:
        lines.append("slices:")
        for sl in self.slice_scores:
            mark = "  [REGRESSED]" if sl.regressed else ""
            lines.append(
                f"  {sl.slice} (n={sl.n}): "
                f"{sl.baseline_score:.4f} -> {sl.final_score:.4f}{mark}"
            )
    lines.append(f"model calls: {self.total_model_calls}")
    if self.total_input_tokens or self.total_output_tokens:
        lines.append(
            f"tokens: {self.total_input_tokens} in / {self.total_output_tokens} out"
        )
    lines.append(f"stopped: {self.stopped_reason}")
    return "\n".join(lines)

HTML report

recension.report

A self-contained HTML audit report rendered from a :class:RunRecord.

render_report turns the full audit record into a single standalone HTML page (inline CSS, no assets, no network) that a reviewer can open, share, or attach to a change request. It surfaces everything the record carries: the baseline and final scores, the locked test estimate and overfit flag, every round's diagnosis and candidates (with significance, guard, and leakage detail), the accepted diff, the per-slice breakdown, the token ledger, and the record's integrity status.

render_report

render_report(record)

Render record as a complete, standalone HTML document (a string).

Source code in recension/report.py

def render_report(record: RunRecord) -> str:
    """Render ``record`` as a complete, standalone HTML document (a string)."""
    name = html.escape(str(record.artifact.get("name", "artifact")))
    objective = html.escape(record.objective_name)
    graded = " (model-graded)" if record.model_graded else ""
    parts: list[str] = [
        "<!doctype html>",
        '<html lang="en"><head><meta charset="utf-8">',
        '<meta name="viewport" content="width=device-width, initial-scale=1">',
        f"<title>recension audit: {name}</title>",
        f"<style>{_STYLE}</style></head><body>",
        f"<h1>recension audit: {name}</h1>",
        f'<p class="sub">objective: {objective}{graded} &middot; '
        f"stopped: {html.escape(record.stopped_reason)}</p>",
        _stats_block(record),
        _slices_block(record),
    ]
    parts.append("<h2>Rounds</h2>")
    if not record.rounds:
        parts.append("<p>No rounds were executed.</p>")
    for round_record in record.rounds:
        parts.append(_round_block(round_record))
    parts.append("</body></html>")
    return "\n".join(parts)

Models

recension.models.base

The minimal model interface every backend implements.

The optimizer core is provider-agnostic: it talks to anything satisfying the :class:Model protocol. Backends ship for Anthropic (optional extra) and a deterministic mock for offline tests.

Role module-attribute

Role = Literal['system', 'user', 'assistant']

Message roles understood by every backend.

Message

Bases: TypedDict

One chat message: a role and its text content.

Source code in recension/models/base.py

class Message(TypedDict):
    """One chat message: a role and its text content."""

    role: Role
    content: str

TokenUsage dataclass

Input/output token counts for one completion (or a sum of them).

Source code in recension/models/base.py

@dataclass(frozen=True)
class TokenUsage:
    """Input/output token counts for one completion (or a sum of them)."""

    input_tokens: int = 0
    output_tokens: int = 0

    def __add__(self, other: TokenUsage) -> TokenUsage:
        return TokenUsage(
            self.input_tokens + other.input_tokens,
            self.output_tokens + other.output_tokens,
        )

SupportsUsage

Bases: Protocol

Optional capability: a model that reports the token usage of its last call.

Models that implement it feed the optimizer's cost ledger; models that do not simply contribute zeros, so usage reporting is fully backward compatible.

Source code in recension/models/base.py

@runtime_checkable
class SupportsUsage(Protocol):
    """Optional capability: a model that reports the token usage of its last call.

    Models that implement it feed the optimizer's cost ledger; models that do
    not simply contribute zeros, so usage reporting is fully backward
    compatible.
    """

    @property
    def last_usage(self) -> TokenUsage:
        """Token usage of the most recent ``complete`` call."""
        ...

last_usage property

last_usage

Token usage of the most recent complete call.

Model

Bases: Protocol

Narrow protocol for a chat-completion model.

Implementations must count every completion in :attr:call_count; the optimizer uses it to enforce Budget.max_model_calls.

Source code in recension/models/base.py

@runtime_checkable
class Model(Protocol):
    """Narrow protocol for a chat-completion model.

    Implementations must count every completion in :attr:`call_count`; the
    optimizer uses it to enforce ``Budget.max_model_calls``.
    """

    @property
    def call_count(self) -> int:
        """Number of ``complete`` calls made so far on this instance."""
        ...

    def complete(
        self,
        messages: list[Message],
        *,
        max_tokens: int = 1024,
        temperature: float = 0.0,
        seed: int | None = None,
    ) -> str:
        """Return the model's text completion for ``messages``.

        Args:
            messages: Conversation so far; at most one ``system`` message.
            max_tokens: Upper bound on generated tokens.
            temperature: Sampling temperature; 0 for greedy.
            seed: Optional determinism hint. Backends that cannot honor it
                (e.g. hosted APIs) document that they ignore it.
        """
        ...

call_count property

call_count

Number of complete calls made so far on this instance.

complete

complete(messages, *, max_tokens=1024, temperature=0.0, seed=None)

Return the model's text completion for messages.

Parameters:

Name	Type	Description	Default
messages	list[Message]	Conversation so far; at most one system message.	required
max_tokens	int	Upper bound on generated tokens.	1024
temperature	float	Sampling temperature; 0 for greedy.	0.0
seed	int | None	Optional determinism hint. Backends that cannot honor it (e.g. hosted APIs) document that they ignore it.	None

Source code in recension/models/base.py

def complete(
    self,
    messages: list[Message],
    *,
    max_tokens: int = 1024,
    temperature: float = 0.0,
    seed: int | None = None,
) -> str:
    """Return the model's text completion for ``messages``.

    Args:
        messages: Conversation so far; at most one ``system`` message.
        max_tokens: Upper bound on generated tokens.
        temperature: Sampling temperature; 0 for greedy.
        seed: Optional determinism hint. Backends that cannot honor it
            (e.g. hosted APIs) document that they ignore it.
    """
    ...

recension.models.mock

Deterministic mock model for offline tests and reproducible examples.

The entire test suite runs against :class:MockModel: no network, no API key. Given the same messages, seed, and script, it always returns the same output, which is what makes seeded optimizer runs reproducible.

MockModel

A deterministic, scriptable stand-in for a real model.

Parameters:

Name	Type	Description	Default
script	Callable[[list[Message]], str] | None	Optional callable mapping the message list to a reply. Use it to simulate task answers, diagnoses, judges, or candidate proposals in tests and examples. When omitted, replies are deterministic pseudo-text derived from a hash of the messages and seed.	None
seed	int	Folded into the unscripted reply hash, so different seeds give different (but stable) outputs.	0

Source code in recension/models/mock.py

class MockModel:
    """A deterministic, scriptable stand-in for a real model.

    Args:
        script: Optional callable mapping the message list to a reply. Use it
            to simulate task answers, diagnoses, judges, or candidate
            proposals in tests and examples. When omitted, replies are
            deterministic pseudo-text derived from a hash of the messages
            and seed.
        seed: Folded into the unscripted reply hash, so different seeds give
            different (but stable) outputs.
    """

    def __init__(
        self,
        script: Callable[[list[Message]], str] | None = None,
        *,
        seed: int = 0,
    ) -> None:
        self.script = script
        self.seed = seed
        self._calls = 0
        self._last_usage = TokenUsage()

    @property
    def call_count(self) -> int:
        """Number of ``complete`` calls made on this instance."""
        return self._calls

    @property
    def last_usage(self) -> TokenUsage:
        """Synthetic, deterministic token usage of the last call (roughly chars/4)."""
        return self._last_usage

    def complete(
        self,
        messages: list[Message],
        *,
        max_tokens: int = 1024,
        temperature: float = 0.0,
        seed: int | None = None,
    ) -> str:
        """Return a deterministic reply for ``messages``.

        The ``seed`` argument, when given, overrides the instance seed for
        this call. ``max_tokens`` and ``temperature`` are accepted for
        protocol compatibility; they do not change the output.
        """
        self._calls += 1
        if self.script is not None:
            reply = self.script(list(messages))
        else:
            effective_seed = self.seed if seed is None else seed
            digest = hashlib.sha256()
            digest.update(str(effective_seed).encode())
            for message in messages:
                digest.update(message["role"].encode())
                digest.update(b"\x00")
                digest.update(message["content"].encode())
                digest.update(b"\x01")
            reply = f"mock-output-{digest.hexdigest()[:16]}"
        prompt_chars = sum(len(m["content"]) for m in messages)
        self._last_usage = TokenUsage(prompt_chars // 4, len(reply) // 4)
        return reply

call_count property

call_count

Number of complete calls made on this instance.

last_usage property

last_usage

Synthetic, deterministic token usage of the last call (roughly chars/4).

complete

complete(messages, *, max_tokens=1024, temperature=0.0, seed=None)

Return a deterministic reply for messages.

The seed argument, when given, overrides the instance seed for this call. max_tokens and temperature are accepted for protocol compatibility; they do not change the output.

Source code in recension/models/mock.py

def complete(
    self,
    messages: list[Message],
    *,
    max_tokens: int = 1024,
    temperature: float = 0.0,
    seed: int | None = None,
) -> str:
    """Return a deterministic reply for ``messages``.

    The ``seed`` argument, when given, overrides the instance seed for
    this call. ``max_tokens`` and ``temperature`` are accepted for
    protocol compatibility; they do not change the output.
    """
    self._calls += 1
    if self.script is not None:
        reply = self.script(list(messages))
    else:
        effective_seed = self.seed if seed is None else seed
        digest = hashlib.sha256()
        digest.update(str(effective_seed).encode())
        for message in messages:
            digest.update(message["role"].encode())
            digest.update(b"\x00")
            digest.update(message["content"].encode())
            digest.update(b"\x01")
        reply = f"mock-output-{digest.hexdigest()[:16]}"
    prompt_chars = sum(len(m["content"]) for m in messages)
    self._last_usage = TokenUsage(prompt_chars // 4, len(reply) // 4)
    return reply

recension.models.anthropic

Anthropic backend for the :class:~recension.models.base.Model protocol.

Requires the optional extra: pip install "recension[anthropic]". The API key is read from the environment (ANTHROPIC_API_KEY) by the Anthropic SDK itself; this class never accepts a key argument, so a key cannot end up in code or config.

AnthropicModel

Model backend that calls the Anthropic Messages API.

Parameters:

Name	Type	Description	Default
model	str	Anthropic model id. Defaults to claude-opus-4-8.	DEFAULT_MODEL
max_retries	int	Passed through to the SDK client.	2
send_temperature	bool | None	Whether to send the temperature parameter. None (default) infers it: recent model families (Opus 4.7+, Fable 5) reject sampling parameters with HTTP 400, so it is dropped for them and sent otherwise. Pass True/False to override the inference, an escape hatch for a future model id the built-in heuristic does not yet know about.	None

Raises:

Type	Description
ImportError	If the anthropic package is not installed.

Note

The seed parameter of :meth:complete is ignored, since the Anthropic API does not support sampling seeds. Determinism in tests comes from :class:~recension.models.mock.MockModel, never from this backend.

Source code in recension/models/anthropic.py

class AnthropicModel:
    """Model backend that calls the Anthropic Messages API.

    Args:
        model: Anthropic model id. Defaults to ``claude-opus-4-8``.
        max_retries: Passed through to the SDK client.
        send_temperature: Whether to send the ``temperature`` parameter.
            ``None`` (default) infers it: recent model families (Opus 4.7+,
            Fable 5) reject sampling parameters with HTTP 400, so it is dropped
            for them and sent otherwise. Pass ``True``/``False`` to override the
            inference, an escape hatch for a future model id the built-in
            heuristic does not yet know about.

    Raises:
        ImportError: If the ``anthropic`` package is not installed.

    Note:
        The ``seed`` parameter of :meth:`complete` is ignored, since the Anthropic
        API does not support sampling seeds. Determinism in tests comes from
        :class:`~recension.models.mock.MockModel`, never from this backend.
    """

    def __init__(
        self,
        model: str = DEFAULT_MODEL,
        *,
        max_retries: int = 2,
        send_temperature: bool | None = None,
    ) -> None:
        try:
            import anthropic
        except ImportError as exc:  # pragma: no cover - exercised only without the extra
            raise ImportError(
                "the Anthropic backend needs the 'anthropic' package; "
                'install it with: pip install "recension[anthropic]"'
            ) from exc
        self.model = model
        self.send_temperature = send_temperature
        self._client = anthropic.Anthropic(max_retries=max_retries)
        self._calls = 0
        self._last_usage = TokenUsage()

    @property
    def call_count(self) -> int:
        """Number of ``complete`` calls made on this instance."""
        return self._calls

    @property
    def last_usage(self) -> TokenUsage:
        """Token usage of the last call, read from the API response."""
        return self._last_usage

    def complete(
        self,
        messages: list[Message],
        *,
        max_tokens: int = 1024,
        temperature: float = 0.0,
        seed: int | None = None,
    ) -> str:
        """Send ``messages`` to the Anthropic API and return the reply text.

        ``system`` messages are lifted into the API's ``system`` parameter;
        ``user``/``assistant`` messages pass through in order.
        """
        self._calls += 1
        system, chat = split_system(messages)
        kwargs: dict[str, Any] = {
            "model": self.model,
            "max_tokens": max_tokens,
            "messages": chat,
        }
        if system:
            kwargs["system"] = system
        if should_send_temperature(self.model, self.send_temperature):
            kwargs["temperature"] = temperature
        response = self._client.messages.create(**kwargs)
        usage = getattr(response, "usage", None)
        self._last_usage = TokenUsage(
            getattr(usage, "input_tokens", 0) or 0,
            getattr(usage, "output_tokens", 0) or 0,
        )
        return "".join(
            block.text for block in response.content if getattr(block, "type", "") == "text"
        )

call_count property

call_count

Number of complete calls made on this instance.

last_usage property

last_usage

Token usage of the last call, read from the API response.

complete

complete(messages, *, max_tokens=1024, temperature=0.0, seed=None)

Send messages to the Anthropic API and return the reply text.

system messages are lifted into the API's system parameter; user/assistant messages pass through in order.

Source code in recension/models/anthropic.py

def complete(
    self,
    messages: list[Message],
    *,
    max_tokens: int = 1024,
    temperature: float = 0.0,
    seed: int | None = None,
) -> str:
    """Send ``messages`` to the Anthropic API and return the reply text.

    ``system`` messages are lifted into the API's ``system`` parameter;
    ``user``/``assistant`` messages pass through in order.
    """
    self._calls += 1
    system, chat = split_system(messages)
    kwargs: dict[str, Any] = {
        "model": self.model,
        "max_tokens": max_tokens,
        "messages": chat,
    }
    if system:
        kwargs["system"] = system
    if should_send_temperature(self.model, self.send_temperature):
        kwargs["temperature"] = temperature
    response = self._client.messages.create(**kwargs)
    usage = getattr(response, "usage", None)
    self._last_usage = TokenUsage(
        getattr(usage, "input_tokens", 0) or 0,
        getattr(usage, "output_tokens", 0) or 0,
    )
    return "".join(
        block.text for block in response.content if getattr(block, "type", "") == "text"
    )

should_send_temperature

should_send_temperature(model, override)

Decide whether to send temperature for model.

override wins when set; otherwise infer from the known no-sampling model prefixes. Pure and SDK-free so the decision is testable offline.

Source code in recension/models/anthropic.py

def should_send_temperature(model: str, override: bool | None) -> bool:
    """Decide whether to send ``temperature`` for ``model``.

    ``override`` wins when set; otherwise infer from the known no-sampling model
    prefixes. Pure and SDK-free so the decision is testable offline.
    """
    if override is not None:
        return override
    return not model.startswith(_NO_SAMPLING_PREFIXES)

split_system

split_system(messages)

Split a message list into (system text, chat messages).

Multiple system messages are joined with blank lines. Exposed as a module function so the conversion is testable without the SDK installed.

Source code in recension/models/anthropic.py

def split_system(messages: list[Message]) -> tuple[str, list[dict[str, str]]]:
    """Split a message list into (system text, chat messages).

    Multiple ``system`` messages are joined with blank lines. Exposed as a
    module function so the conversion is testable without the SDK installed.
    """
    system_parts: list[str] = []
    chat: list[dict[str, str]] = []
    for message in messages:
        if message["role"] == "system":
            system_parts.append(message["content"])
        else:
            chat.append({"role": message["role"], "content": message["content"]})
    return "\n\n".join(system_parts), chat

recension.models.openai

OpenAI backend for the :class:~recension.models.base.Model protocol.

Requires the optional extra: pip install "recension[openai]". The API key is read from the environment (OPENAI_API_KEY) by the OpenAI SDK itself; this class never accepts a key argument, so a key cannot end up in code or config.

Because it speaks the standard Chat Completions API, this backend also drives OpenAI-compatible servers (vLLM, LM Studio, OpenRouter, Together, and others) by passing their base_url.

OpenAIModel

Model backend that calls the OpenAI Chat Completions API.

Parameters:

Name	Type	Description	Default
model	str	OpenAI model id. Defaults to gpt-4o-mini.	DEFAULT_MODEL
base_url	str | None	Optional base URL for an OpenAI-compatible server. None uses the OpenAI API.	None
max_retries	int	Passed through to the SDK client.	2
send_temperature	bool | None	Whether to send temperature. None (default) infers it: reasoning models (o-series, gpt-5) reject a custom value, so it is dropped for them and sent otherwise. Pass True/False to override.	None
client	Any	A pre-built OpenAI client to use instead of constructing one (for advanced configs such as Azure, or for testing). When given, the openai package is not imported.	None

Raises:

Type	Description
ImportError	If the openai package is not installed and no client is supplied.

Note

seed is forwarded to the API as a best-effort determinism hint; the OpenAI API does not guarantee reproducibility. Deterministic tests use :class:~recension.models.mock.MockModel, never this backend.

Source code in recension/models/openai.py

class OpenAIModel:
    """Model backend that calls the OpenAI Chat Completions API.

    Args:
        model: OpenAI model id. Defaults to ``gpt-4o-mini``.
        base_url: Optional base URL for an OpenAI-compatible server. ``None``
            uses the OpenAI API.
        max_retries: Passed through to the SDK client.
        send_temperature: Whether to send ``temperature``. ``None`` (default)
            infers it: reasoning models (o-series, gpt-5) reject a custom value,
            so it is dropped for them and sent otherwise. Pass ``True``/``False``
            to override.
        client: A pre-built OpenAI client to use instead of constructing one
            (for advanced configs such as Azure, or for testing). When given,
            the ``openai`` package is not imported.

    Raises:
        ImportError: If the ``openai`` package is not installed and no ``client``
            is supplied.

    Note:
        ``seed`` is forwarded to the API as a best-effort determinism hint; the
        OpenAI API does not guarantee reproducibility. Deterministic tests use
        :class:`~recension.models.mock.MockModel`, never this backend.
    """

    def __init__(
        self,
        model: str = DEFAULT_MODEL,
        *,
        base_url: str | None = None,
        max_retries: int = 2,
        send_temperature: bool | None = None,
        client: Any = None,
    ) -> None:
        self.model = model
        self.send_temperature = send_temperature
        if client is not None:
            self._client = client
        else:
            try:
                import openai
            except ImportError as exc:  # pragma: no cover - exercised only without the extra
                raise ImportError(
                    "the OpenAI backend needs the 'openai' package; "
                    'install it with: pip install "recension[openai]"'
                ) from exc
            kwargs: dict[str, Any] = {"max_retries": max_retries}
            if base_url is not None:
                kwargs["base_url"] = base_url
            self._client = openai.OpenAI(**kwargs)
        self._calls = 0
        self._last_usage = TokenUsage()

    @property
    def call_count(self) -> int:
        """Number of ``complete`` calls made on this instance."""
        return self._calls

    @property
    def last_usage(self) -> TokenUsage:
        """Token usage of the last call, read from the API response."""
        return self._last_usage

    def complete(
        self,
        messages: list[Message],
        *,
        max_tokens: int = 1024,
        temperature: float = 0.0,
        seed: int | None = None,
    ) -> str:
        """Send ``messages`` to the Chat Completions API and return the reply text."""
        self._calls += 1
        kwargs: dict[str, Any] = {
            "model": self.model,
            "messages": [dict(m) for m in messages],
        }
        # Reasoning models require max_completion_tokens; standard models use max_tokens.
        token_param = "max_completion_tokens" if is_reasoning_model(self.model) else "max_tokens"
        kwargs[token_param] = max_tokens
        if seed is not None:
            kwargs["seed"] = seed
        if should_send_temperature(self.model, self.send_temperature):
            kwargs["temperature"] = temperature
        response = self._client.chat.completions.create(**kwargs)
        self._last_usage = usage_from_response(response)
        return _first_content(response)

call_count property

call_count

Number of complete calls made on this instance.

last_usage property

last_usage

Token usage of the last call, read from the API response.

complete

complete(messages, *, max_tokens=1024, temperature=0.0, seed=None)

Send messages to the Chat Completions API and return the reply text.

Source code in recension/models/openai.py

def complete(
    self,
    messages: list[Message],
    *,
    max_tokens: int = 1024,
    temperature: float = 0.0,
    seed: int | None = None,
) -> str:
    """Send ``messages`` to the Chat Completions API and return the reply text."""
    self._calls += 1
    kwargs: dict[str, Any] = {
        "model": self.model,
        "messages": [dict(m) for m in messages],
    }
    # Reasoning models require max_completion_tokens; standard models use max_tokens.
    token_param = "max_completion_tokens" if is_reasoning_model(self.model) else "max_tokens"
    kwargs[token_param] = max_tokens
    if seed is not None:
        kwargs["seed"] = seed
    if should_send_temperature(self.model, self.send_temperature):
        kwargs["temperature"] = temperature
    response = self._client.chat.completions.create(**kwargs)
    self._last_usage = usage_from_response(response)
    return _first_content(response)

is_reasoning_model

is_reasoning_model(model)

True for OpenAI reasoning models (o-series, gpt-5). Pure; testable offline.

Source code in recension/models/openai.py

def is_reasoning_model(model: str) -> bool:
    """True for OpenAI reasoning models (o-series, gpt-5). Pure; testable offline."""
    return model.startswith(_REASONING_PREFIXES)

should_send_temperature

should_send_temperature(model, override)

Decide whether to send temperature for model.

override wins when set; otherwise infer from the reasoning-model prefixes. Pure and SDK-free so the decision is testable offline.

Source code in recension/models/openai.py

def should_send_temperature(model: str, override: bool | None) -> bool:
    """Decide whether to send ``temperature`` for ``model``.

    ``override`` wins when set; otherwise infer from the reasoning-model
    prefixes. Pure and SDK-free so the decision is testable offline.
    """
    if override is not None:
        return override
    return not is_reasoning_model(model)

usage_from_response

usage_from_response(response)

Read token usage from a Chat Completions response (SDK-free; testable).

Source code in recension/models/openai.py

def usage_from_response(response: Any) -> TokenUsage:
    """Read token usage from a Chat Completions response (SDK-free; testable)."""
    usage = getattr(response, "usage", None)
    return TokenUsage(
        getattr(usage, "prompt_tokens", 0) or 0,
        getattr(usage, "completion_tokens", 0) or 0,
    )

recension.models.gemini

Google Gemini backend for the :class:~recension.models.base.Model protocol.

Requires the optional extra: pip install "recension[gemini]". The API key is read from the environment (GEMINI_API_KEY or GOOGLE_API_KEY) by the google-genai SDK itself; this class never accepts a key argument, so a key cannot end up in code or config.

GeminiModel

Model backend that calls the Google Gemini API (google-genai SDK).

Parameters:

Name	Type	Description	Default
model	str	Gemini model id. Defaults to gemini-2.0-flash.	DEFAULT_MODEL
client	Any	A pre-built google.genai.Client to use instead of constructing one (for advanced configs or testing). When given, the google-genai package is not imported.	None

Raises:

Type	Description
ImportError	If the google-genai package is not installed and no client is supplied.

Note

seed is forwarded as a best-effort determinism hint; the API does not guarantee reproducibility. Deterministic tests use :class:~recension.models.mock.MockModel, never this backend.

Source code in recension/models/gemini.py

class GeminiModel:
    """Model backend that calls the Google Gemini API (google-genai SDK).

    Args:
        model: Gemini model id. Defaults to ``gemini-2.0-flash``.
        client: A pre-built ``google.genai.Client`` to use instead of
            constructing one (for advanced configs or testing). When given, the
            ``google-genai`` package is not imported.

    Raises:
        ImportError: If the ``google-genai`` package is not installed and no
            ``client`` is supplied.

    Note:
        ``seed`` is forwarded as a best-effort determinism hint; the API does not
        guarantee reproducibility. Deterministic tests use
        :class:`~recension.models.mock.MockModel`, never this backend.
    """

    def __init__(self, model: str = DEFAULT_MODEL, *, client: Any = None) -> None:
        self.model = model
        if client is not None:
            self._client = client
        else:
            try:
                from google import genai
            except ImportError as exc:  # pragma: no cover - exercised only without the extra
                raise ImportError(
                    "the Gemini backend needs the 'google-genai' package; "
                    'install it with: pip install "recension[gemini]"'
                ) from exc
            self._client = genai.Client()
        self._calls = 0
        self._last_usage = TokenUsage()

    @property
    def call_count(self) -> int:
        """Number of ``complete`` calls made on this instance."""
        return self._calls

    @property
    def last_usage(self) -> TokenUsage:
        """Token usage of the last call, read from the API response."""
        return self._last_usage

    def complete(
        self,
        messages: list[Message],
        *,
        max_tokens: int = 1024,
        temperature: float = 0.0,
        seed: int | None = None,
    ) -> str:
        """Send ``messages`` to the Gemini API and return the reply text.

        ``system`` messages become the ``system_instruction``; ``assistant``
        messages are mapped to Gemini's ``model`` role.
        """
        self._calls += 1
        system, contents = to_gemini_contents(messages)
        config: dict[str, Any] = {"max_output_tokens": max_tokens, "temperature": temperature}
        if system:
            config["system_instruction"] = system
        if seed is not None:
            config["seed"] = seed
        response = self._client.models.generate_content(
            model=self.model, contents=contents, config=config
        )
        self._last_usage = usage_from_response(response)
        # `.text` is a property that raises (not returns None) when the candidate
        # has no text part, e.g. a safety-blocked response; treat that as empty.
        try:
            text = response.text
        except Exception:
            text = None
        return text or ""

call_count property

call_count

Number of complete calls made on this instance.

last_usage property

last_usage

Token usage of the last call, read from the API response.

complete

complete(messages, *, max_tokens=1024, temperature=0.0, seed=None)

Send messages to the Gemini API and return the reply text.

system messages become the system_instruction; assistant messages are mapped to Gemini's model role.

Source code in recension/models/gemini.py

def complete(
    self,
    messages: list[Message],
    *,
    max_tokens: int = 1024,
    temperature: float = 0.0,
    seed: int | None = None,
) -> str:
    """Send ``messages`` to the Gemini API and return the reply text.

    ``system`` messages become the ``system_instruction``; ``assistant``
    messages are mapped to Gemini's ``model`` role.
    """
    self._calls += 1
    system, contents = to_gemini_contents(messages)
    config: dict[str, Any] = {"max_output_tokens": max_tokens, "temperature": temperature}
    if system:
        config["system_instruction"] = system
    if seed is not None:
        config["seed"] = seed
    response = self._client.models.generate_content(
        model=self.model, contents=contents, config=config
    )
    self._last_usage = usage_from_response(response)
    # `.text` is a property that raises (not returns None) when the candidate
    # has no text part, e.g. a safety-blocked response; treat that as empty.
    try:
        text = response.text
    except Exception:
        text = None
    return text or ""

to_gemini_contents

to_gemini_contents(messages)

Split messages into (system instruction, Gemini contents).

system messages are joined into the instruction; user stays user and assistant becomes model. Pure and SDK-free so the mapping is testable offline.

Source code in recension/models/gemini.py

def to_gemini_contents(messages: list[Message]) -> tuple[str, list[dict[str, Any]]]:
    """Split messages into (system instruction, Gemini ``contents``).

    ``system`` messages are joined into the instruction; ``user`` stays ``user``
    and ``assistant`` becomes ``model``. Pure and SDK-free so the mapping is
    testable offline.
    """
    system_parts: list[str] = []
    contents: list[dict[str, Any]] = []
    for message in messages:
        if message["role"] == "system":
            system_parts.append(message["content"])
            continue
        role = "model" if message["role"] == "assistant" else "user"
        contents.append({"role": role, "parts": [{"text": message["content"]}]})
    return "\n\n".join(system_parts), contents

usage_from_response

usage_from_response(response)

Read token usage from a Gemini response (SDK-free; testable).

Source code in recension/models/gemini.py

def usage_from_response(response: Any) -> TokenUsage:
    """Read token usage from a Gemini response (SDK-free; testable)."""
    usage = getattr(response, "usage_metadata", None)
    return TokenUsage(
        getattr(usage, "prompt_token_count", 0) or 0,
        getattr(usage, "candidates_token_count", 0) or 0,
    )

recension.models.ollama

Ollama backend for the :class:~recension.models.base.Model protocol.

Runs local models served by Ollama <https://ollama.com>_. Needs no extra dependency: it talks to the local HTTP API with the standard library. Point it at a running Ollama server (default http://localhost:11434) and pull the model first (ollama pull llama3.2).

OllamaModel

Model backend that calls a local Ollama server's chat API.

Parameters:

Name	Type	Description	Default
model	str	Ollama model name (must be pulled locally). Defaults to llama3.2.	DEFAULT_MODEL
host	str	Base URL of the Ollama server.	'http://localhost:11434'
timeout	float	Per-request timeout in seconds.	120.0
transport	Transport | None	Override the HTTP transport (url, payload, timeout) -> response_dict; used for testing. Defaults to a stdlib urllib POST.	None

Note

seed is forwarded to Ollama; local models honor it fairly well, but deterministic tests still use :class:~recension.models.mock.MockModel, never a live server.

Source code in recension/models/ollama.py

class OllamaModel:
    """Model backend that calls a local Ollama server's chat API.

    Args:
        model: Ollama model name (must be pulled locally). Defaults to
            ``llama3.2``.
        host: Base URL of the Ollama server.
        timeout: Per-request timeout in seconds.
        transport: Override the HTTP transport ``(url, payload, timeout) ->
            response_dict``; used for testing. Defaults to a stdlib ``urllib``
            POST.

    Note:
        ``seed`` is forwarded to Ollama; local models honor it fairly well, but
        deterministic tests still use :class:`~recension.models.mock.MockModel`,
        never a live server.
    """

    def __init__(
        self,
        model: str = DEFAULT_MODEL,
        *,
        host: str = "http://localhost:11434",
        timeout: float = 120.0,
        transport: Transport | None = None,
    ) -> None:
        self.model = model
        self._url = host.rstrip("/") + "/api/chat"
        self._timeout = timeout
        self._transport: Transport = transport or _http_post
        self._calls = 0
        self._last_usage = TokenUsage()

    @property
    def call_count(self) -> int:
        """Number of ``complete`` calls made on this instance."""
        return self._calls

    @property
    def last_usage(self) -> TokenUsage:
        """Token usage of the last call, read from the Ollama response."""
        return self._last_usage

    def complete(
        self,
        messages: list[Message],
        *,
        max_tokens: int = 1024,
        temperature: float = 0.0,
        seed: int | None = None,
    ) -> str:
        """Send ``messages`` to the Ollama chat API and return the reply text."""
        self._calls += 1
        payload = build_payload(self.model, messages, max_tokens, temperature, seed)
        data = self._transport(self._url, payload, self._timeout)
        self._last_usage = usage_from_response(data)
        return data.get("message", {}).get("content", "") or ""

call_count property

call_count

Number of complete calls made on this instance.

last_usage property

last_usage

Token usage of the last call, read from the Ollama response.

complete

complete(messages, *, max_tokens=1024, temperature=0.0, seed=None)

Send messages to the Ollama chat API and return the reply text.

Source code in recension/models/ollama.py

def complete(
    self,
    messages: list[Message],
    *,
    max_tokens: int = 1024,
    temperature: float = 0.0,
    seed: int | None = None,
) -> str:
    """Send ``messages`` to the Ollama chat API and return the reply text."""
    self._calls += 1
    payload = build_payload(self.model, messages, max_tokens, temperature, seed)
    data = self._transport(self._url, payload, self._timeout)
    self._last_usage = usage_from_response(data)
    return data.get("message", {}).get("content", "") or ""

build_payload

build_payload(model, messages, max_tokens, temperature, seed)

Build the Ollama /api/chat request body (pure; testable).

Source code in recension/models/ollama.py

def build_payload(
    model: str,
    messages: list[Message],
    max_tokens: int,
    temperature: float,
    seed: int | None,
) -> dict[str, Any]:
    """Build the Ollama ``/api/chat`` request body (pure; testable)."""
    options: dict[str, Any] = {"num_predict": max_tokens, "temperature": temperature}
    if seed is not None:
        options["seed"] = seed
    return {
        "model": model,
        "messages": [dict(m) for m in messages],
        "stream": False,
        "options": options,
    }

usage_from_response

usage_from_response(data)

Read token usage from an Ollama response (pure; testable).

Source code in recension/models/ollama.py

def usage_from_response(data: dict[str, Any]) -> TokenUsage:
    """Read token usage from an Ollama response (pure; testable)."""
    return TokenUsage(
        int(data.get("prompt_eval_count", 0) or 0),
        int(data.get("eval_count", 0) or 0),
    )

Exceptions

recension.exceptions

Exception types for recension.

All library-specific errors derive from :class:RecensionError so callers can catch the whole family with one clause. Measurement-integrity problems get their own loud types; they are never swallowed.

RecensionError

Bases: Exception

Base class for all recension errors.

Source code in recension/exceptions.py

class RecensionError(Exception):
    """Base class for all recension errors."""

ArtifactError

Bases: RecensionError

Raised for invalid artifact operations (unknown version, no-op commit).

Source code in recension/exceptions.py

class ArtifactError(RecensionError):
    """Raised for invalid artifact operations (unknown version, no-op commit)."""

BudgetExceeded

Bases: RecensionError

Raised when an optimization run hits Budget.max_model_calls.

When raised by ReflectiveOptimizer.run(), :attr:record carries the partial run record up to the point of the overrun, so the audit trail is not lost with the failure.

Source code in recension/exceptions.py

class BudgetExceeded(RecensionError):
    """Raised when an optimization run hits ``Budget.max_model_calls``.

    When raised by ``ReflectiveOptimizer.run()``, :attr:`record` carries the
    partial run record up to the point of the overrun, so the audit trail is
    not lost with the failure.
    """

    def __init__(self, message: str, *, record: RunRecord | None = None) -> None:
        super().__init__(message)
        self.record = record

DegenerateEvalError

Bases: RecensionError

Raised when an eval set cannot produce an honest signal.

Examples: an empty train or validation split, or an example id that appears in both splits (which would contaminate the acceptance signal).

Source code in recension/exceptions.py

class DegenerateEvalError(RecensionError):
    """Raised when an eval set cannot produce an honest signal.

    Examples: an empty train or validation split, or an example id that
    appears in both splits (which would contaminate the acceptance signal).
    """

LeakageDetected

Bases: RecensionError

Raised in strict mode when a winning candidate trips a leakage heuristic.

Outside strict mode the same condition is recorded as a flag on the candidate instead of raising. See :mod:recension.leakage. When raised by ReflectiveOptimizer.run(), :attr:record carries the partial run record so the audit trail survives the failure.

Source code in recension/exceptions.py

class LeakageDetected(RecensionError):
    """Raised in strict mode when a winning candidate trips a leakage heuristic.

    Outside strict mode the same condition is recorded as a flag on the
    candidate instead of raising. See :mod:`recension.leakage`. When raised by
    ``ReflectiveOptimizer.run()``, :attr:`record` carries the partial run
    record so the audit trail survives the failure.
    """

    def __init__(self, message: str, *, record: RunRecord | None = None) -> None:
        super().__init__(message)
        self.record = record

ConfigError

Bases: RecensionError

Raised for invalid or incomplete CLI/run configuration.

Source code in recension/exceptions.py

class ConfigError(RecensionError):
    """Raised for invalid or incomplete CLI/run configuration."""

CLI

recension.cli

The recension command-line interface.

A thin wrapper over the library; no logic lives only here. Three commands:

• recension run --config run.yaml executes an optimization and writes the run record (and optionally the optimized artifact text) to disk.
• recension show record.json prints a human-readable summary of a record.
• recension diff record.json vA vB prints the diff between two artifact versions stored in a record.

Exit codes: 0 success; 1 configuration or usage error; 2 measurement-integrity failure (budget exceeded or leakage in strict mode, where the partial record is still written so the audit trail survives).

main

main(argv=None)

Entry point for the recension console script.

Source code in recension/cli.py

def main(argv: list[str] | None = None) -> int:
    """Entry point for the ``recension`` console script."""
    parser = _build_parser()
    args = parser.parse_args(argv)
    try:
        result: int = args.func(args)
        return result
    except ConfigError as exc:
        print(f"config error: {exc}", file=sys.stderr)
        return 1
    except RecensionError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1