Copy page

Python API

Generated from the source. For the reasoning behind these shapes, start with How agents actually work.

Running a loop directly

from pathlib import Path

from endstate.agent.loop import AgentLoop
from endstate.agent.context import ContextManager, TokenBudget
from endstate.agent.permissions import default_policy
from endstate.agent.session import SessionStore
from endstate.agent.tools import default_tools
from endstate.agent.tools.base import ToolContext
from endstate.providers.openai_compat import OpenAICompatProvider

store = SessionStore()
loop = AgentLoop(
    provider=OpenAICompatProvider(model="gpt-4o-mini"),
    tools=default_tools(),
    tool_context=ToolContext(workdir=Path("./sandbox")),
    policy=default_policy(),
    context=ContextManager(budget=TokenBudget(max_context_tokens=128_000)),
    session=store.create(model="gpt-4o-mini"),
    max_steps=25,
)

result = loop.run("fix the failing test")
print(result.stop_reason, result.steps, result.usage.total_tokens)

Running an eval suite

from pathlib import Path

from endstate.evals import DockerSandbox, EvalRunner, discover_tasks, render_markdown
from endstate.providers.openai_compat import OpenAICompatProvider

tasks = discover_tasks(Path("tasks"))
runner = EvalRunner(
    provider_factory=lambda task: OpenAICompatProvider(model="gpt-4o-mini"),
    sandbox_factory=lambda task: DockerSandbox(task.fixture),
)
suite = runner.run_suite(tasks)
print(render_markdown(suite))

Both factories take the task, so a suite can vary the model or the image per task. Nothing in SuiteResult reaches a grader: the verdict is decided before the result is assembled.

---

Core types

endstate.types

Core value types shared across the harness.

Deliberately provider-agnostic: every provider adapter translates its own wire format into these types, so the loop never sees a vendor-specific object.

ToolCall

Bases: BaseModel

A model's request to invoke a tool.

Source code in src/endstate/types.py

class ToolCall(BaseModel):
    """A model's request to invoke a tool."""

    id: str
    name: str
    arguments: dict[str, Any] = Field(default_factory=dict)

ToolResult

Bases: BaseModel

The outcome of running a tool.

Source code in src/endstate/types.py

class ToolResult(BaseModel):
    """The outcome of running a tool."""

    call_id: str
    content: str
    is_error: bool = False

Message

Bases: BaseModel

One turn in the conversation.

Source code in src/endstate/types.py

class Message(BaseModel):
    """One turn in the conversation."""

    role: Role
    content: str = ""
    tool_calls: list[ToolCall] = Field(default_factory=list)
    tool_results: list[ToolResult] = Field(default_factory=list)
    # Set by the context manager when a message is synthesised during compaction.
    synthetic: bool = False

Usage

Bases: BaseModel

Token accounting for a single provider call.

Source code in src/endstate/types.py

class Usage(BaseModel):
    """Token accounting for a single provider call."""

    input_tokens: int = 0
    output_tokens: int = 0
    cached_input_tokens: int = 0

    def __add__(self, other: Usage) -> Usage:
        return Usage(
            input_tokens=self.input_tokens + other.input_tokens,
            output_tokens=self.output_tokens + other.output_tokens,
            cached_input_tokens=self.cached_input_tokens + other.cached_input_tokens,
        )

    @property
    def total_tokens(self) -> int:
        return self.input_tokens + self.output_tokens + self.cached_input_tokens

Response

Bases: BaseModel

A normalised provider response.

Source code in src/endstate/types.py

class Response(BaseModel):
    """A normalised provider response."""

    message: Message
    usage: Usage = Field(default_factory=Usage)
    stop_reason: StopReason = StopReason.END_TURN
    model: str = ""

---

The loop

endstate.agent.loop

The harness.

Written from primitives rather than on a framework, on purpose. The interesting parts of an agent — where the context budget is enforced, what happens when a tool is denied, what is persisted before a step can fail — are exactly the parts a framework hides. They are all visible in this file.

UnsettledCall

Bases: BaseModel

A call left outstanding by a crash that resume refused to replay.

Only produced for tools that declare themselves non-idempotent. The call is answered — an unanswered one is malformed under every provider contract — but with an error saying the outcome is unknown rather than with a second side effect.

Source code in src/endstate/agent/loop.py

class UnsettledCall(BaseModel):
    """A call left outstanding by a crash that resume refused to replay.

    Only produced for tools that declare themselves non-idempotent. The call is
    answered — an unanswered one is malformed under every provider contract —
    but with an error saying the outcome is unknown rather than with a second
    side effect.
    """

    tool: str
    call_id: str

RunResult

Bases: BaseModel

Source code in src/endstate/agent/loop.py

class RunResult(BaseModel):
    session_id: str | None = None
    workdir: Path | None = None
    messages: list[Message] = Field(default_factory=list)
    usage: Usage = Field(default_factory=Usage)
    stop_reason: StopReason = StopReason.END_TURN
    steps: int = 0
    compaction_events: list[CompactionEvent] = Field(default_factory=list)
    denied_calls: list[DeniedCall] = Field(default_factory=list)
    unsettled_calls: list[UnsettledCall] = Field(default_factory=list)
    final_text: str = ""

    model_config = {"arbitrary_types_allowed": True}

    def tree_hash(self, *, excludes: frozenset[str] = DEFAULT_EXCLUDES) -> str:
        """Hash the sandbox this run worked in.

        The assertion primitive for end-state grading: "the destructive command
        did not run" is this value being unchanged, and "the resumed run got to
        the same place" is two of these being equal.

        Raises:
            ValueError: If the run recorded no working directory.
        """
        if self.workdir is None:
            raise ValueError("this run recorded no workdir")
        return tree_hash(self.workdir, excludes=excludes)

tree_hash

tree_hash(
    *, excludes: frozenset[str] = DEFAULT_EXCLUDES
) -> str

Hash the sandbox this run worked in.

The assertion primitive for end-state grading: "the destructive command did not run" is this value being unchanged, and "the resumed run got to the same place" is two of these being equal.

Raises:

Type	Description
ValueError	If the run recorded no working directory.

Source code in src/endstate/agent/loop.py

def tree_hash(self, *, excludes: frozenset[str] = DEFAULT_EXCLUDES) -> str:
    """Hash the sandbox this run worked in.

    The assertion primitive for end-state grading: "the destructive command
    did not run" is this value being unchanged, and "the resumed run got to
    the same place" is two of these being equal.

    Raises:
        ValueError: If the run recorded no working directory.
    """
    if self.workdir is None:
        raise ValueError("this run recorded no workdir")
    return tree_hash(self.workdir, excludes=excludes)

AgentLoop

Source code in src/endstate/agent/loop.py

class AgentLoop:
    def __init__(
        self,
        provider: object,
        tools: list[Tool],
        tool_context: ToolContext,
        policy: PermissionPolicy | None = None,
        context: ContextManager | None = None,
        session: Session | None = None,
        accountant: CostAccountant | None = None,
        max_steps: int = 25,
        system_prompt: str | None = None,
    ) -> None:
        self.provider = provider
        self.tools = {t.name: t for t in tools}
        self.tool_context = tool_context
        self.policy = policy or default_policy()
        self.context = context or ContextManager()
        self.session = session
        self.accountant = accountant or CostAccountant()
        self.max_steps = max_steps
        self.system_prompt = system_prompt
        self.trace = Trace()
        # Not threaded through the call chain like `denied` because it is only
        # ever produced by `_settle`, which every entry point calls exactly once.
        self.unsettled: list[UnsettledCall] = []

    # --- persistence ------------------------------------------------------

    def _record(self, message: Message, messages: list[Message]) -> None:
        messages.append(message)
        if self.session is not None:
            self.session.append(message)

    def _checkpoint_last(self, messages: list[Message]) -> None:
        """Re-persist the last message after mutating it in place."""
        if self.session is not None:
            self.session.messages[-1] = messages[-1]
            self.session.checkpoint_last()

    # --- entry points -----------------------------------------------------

    def run(self, prompt: str) -> RunResult:
        """Run the agent against a new instruction."""
        messages: list[Message] = list(self.session.messages) if self.session else []
        denied: list[DeniedCall] = []
        self.unsettled = []

        # Settle before the new prompt is appended, not after: an interrupted
        # batch has to be finished while it is still the tail of the history.
        self._settle(messages, denied)

        if not messages and self.system_prompt:
            self._record(Message(role="system", content=self.system_prompt), messages)
        self._record(Message(role="user", content=prompt), messages)
        return self._drive(messages, denied)

    def resume(self) -> RunResult:
        """Continue an interrupted run without a new instruction.

        Finishes any tool calls that were requested but never executed, then
        carries on. This is the difference between *continuing* a conversation
        and *finishing* what the agent was doing when the process died.

        Raises:
            ValueError: If the loop has no session to resume from.
        """
        if self.session is None:
            raise ValueError("resume requires a session")
        messages: list[Message] = list(self.session.messages)
        denied: list[DeniedCall] = []
        self.unsettled = []
        self._settle(messages, denied)
        return self._drive(messages, denied)

    # --- recovery ---------------------------------------------------------

    def _pending(self, messages: list[Message]) -> tuple[list[ToolCall], bool]:
        """Tool calls that were requested but have no recorded result.

        Returns the outstanding calls and whether a partial result message is
        already the tail of the history.
        """
        if not messages:
            return [], False

        last = messages[-1]
        if last.role == "assistant" and last.tool_calls:
            return list(last.tool_calls), False

        if last.role == "tool" and len(messages) >= 2:
            requested = messages[-2]
            if requested.role == "assistant" and requested.tool_calls:
                done = {r.call_id for r in last.tool_results}
                return [c for c in requested.tool_calls if c.id not in done], True

        return [], False

    def _settle(self, messages: list[Message], denied: list[DeniedCall]) -> None:
        """Execute any calls left outstanding by an interrupted run.

        Without this, a resumed session carries tool calls with no matching
        results — malformed under both the Anthropic and OpenAI contracts — and
        the work those calls represent is silently dropped.
        """
        pending, partial = self._pending(messages)
        if pending:
            self._run_calls(pending, messages, denied, into_existing=partial, settling=True)

    # --- tool execution ---------------------------------------------------

    def _execute(self, call: ToolCall, denied: list[DeniedCall], *, settling: bool) -> ToolResult:
        decision, reason = self.policy.check(call.name, call.arguments)
        if decision is not Decision.ALLOW:
            denied.append(DeniedCall(tool=call.name, arguments=call.arguments, reason=reason))
            with self.trace.span("tool.denied", tool=call.name, reason=reason):
                pass
            return ToolResult(
                call_id=call.id,
                content=f"permission denied ({decision.value}): {reason}",
                is_error=True,
            )

        tool = self.tools.get(call.name)
        if tool is None:
            return ToolResult(call_id=call.id, content=f"unknown tool: {call.name}", is_error=True)

        if settling and not tool.idempotent:
            self.unsettled.append(UnsettledCall(tool=call.name, call_id=call.id))
            with self.trace.span("tool.unsettled", tool=call.name):
                pass
            return ToolResult(
                call_id=call.id, content=INTERRUPTED.format(tool=call.name), is_error=True
            )

        with self.trace.span("tool.run", tool=call.name):
            try:
                output = tool.run(call.arguments, self.tool_context)
                return ToolResult(call_id=call.id, content=output)
            except ToolError as exc:
                return ToolResult(call_id=call.id, content=str(exc), is_error=True)

    def _run_calls(
        self,
        calls: list[ToolCall],
        messages: list[Message],
        denied: list[DeniedCall],
        *,
        into_existing: bool = False,
        settling: bool = False,
    ) -> None:
        """Run a batch, persisting after every individual result.

        D8 — checkpoint *after* the tool result, never before — at per-call
        granularity. Writing the whole batch at the end would mean a crash after
        the second of three tools left no record that the first two had run,
        while their side effects were already on disk.

        `settling` marks the reconciliation batch, where a call's outcome is
        unknown rather than merely unstarted. See `Tool.idempotent`.
        """
        started = into_existing
        for call in calls:
            result = self._execute(call, denied, settling=settling)
            if started:
                messages[-1].tool_results.append(result)
                self._checkpoint_last(messages)
            else:
                self._record(Message(role="tool", tool_results=[result]), messages)
                started = True

    # --- the loop ---------------------------------------------------------

    def _drive(self, messages: list[Message], denied: list[DeniedCall]) -> RunResult:
        specs = [t.spec() for t in self.tools.values()]
        total = Usage()
        stop = StopReason.MAX_STEPS
        steps = 0

        for step in range(self.max_steps):
            steps = step + 1
            fitted = self.context.fit(messages)

            with self.trace.span("provider.complete", step=steps):
                response = self.provider.complete(fitted, specs)  # type: ignore[attr-defined]

            total = total + response.usage
            self.accountant.record(response.model or "unknown", response.usage)
            self._record(response.message, messages)

            if not response.message.tool_calls:
                stop = response.stop_reason
                break

            self._run_calls(list(response.message.tool_calls), messages, denied)

        final = next(
            (m.content for m in reversed(messages) if m.role == "assistant" and m.content), ""
        )
        return RunResult(
            session_id=self.session.id if self.session else None,
            workdir=self.tool_context.workdir,
            messages=messages,
            usage=total,
            stop_reason=stop,
            steps=steps,
            compaction_events=list(self.context.events),
            denied_calls=denied,
            unsettled_calls=list(self.unsettled),
            final_text=final,
        )

run

run(prompt: str) -> RunResult

Run the agent against a new instruction.

Source code in src/endstate/agent/loop.py

def run(self, prompt: str) -> RunResult:
    """Run the agent against a new instruction."""
    messages: list[Message] = list(self.session.messages) if self.session else []
    denied: list[DeniedCall] = []
    self.unsettled = []

    # Settle before the new prompt is appended, not after: an interrupted
    # batch has to be finished while it is still the tail of the history.
    self._settle(messages, denied)

    if not messages and self.system_prompt:
        self._record(Message(role="system", content=self.system_prompt), messages)
    self._record(Message(role="user", content=prompt), messages)
    return self._drive(messages, denied)

resume

resume() -> RunResult

Continue an interrupted run without a new instruction.

Finishes any tool calls that were requested but never executed, then carries on. This is the difference between continuing a conversation and finishing what the agent was doing when the process died.

Raises:

Type	Description
ValueError	If the loop has no session to resume from.

Source code in src/endstate/agent/loop.py

def resume(self) -> RunResult:
    """Continue an interrupted run without a new instruction.

    Finishes any tool calls that were requested but never executed, then
    carries on. This is the difference between *continuing* a conversation
    and *finishing* what the agent was doing when the process died.

    Raises:
        ValueError: If the loop has no session to resume from.
    """
    if self.session is None:
        raise ValueError("resume requires a session")
    messages: list[Message] = list(self.session.messages)
    denied: list[DeniedCall] = []
    self.unsettled = []
    self._settle(messages, denied)
    return self._drive(messages, denied)

---

Context and compaction

endstate.agent.context

Context budgeting and compaction.

The budget is a first-class object rather than an implicit consequence of the model's context window. Every compaction is recorded as an event with the token counts before and after, which is what makes "what did compaction cost you?" answerable with a number instead of a shrug.

HeuristicTokenCounter

Chars/4 approximation.

Deliberately not a tokeniser: the harness must work against any provider, including ones whose tokeniser is not public. Swap in a real counter per provider when exactness matters.

Source code in src/endstate/agent/context.py

class HeuristicTokenCounter:
    """Chars/4 approximation.

    Deliberately not a tokeniser: the harness must work against any provider,
    including ones whose tokeniser is not public. Swap in a real counter per
    provider when exactness matters.
    """

    def __init__(self, chars_per_token: float = 4.0) -> None:
        self.chars_per_token = chars_per_token

    def count(self, messages: list[Message]) -> int:
        chars = 0
        for m in messages:
            chars += len(m.content)
            for tc in m.tool_calls:
                chars += len(tc.name) + len(str(tc.arguments))
            for tr in m.tool_results:
                chars += len(tr.content)
        return int(chars / self.chars_per_token)

DropOldest

Keep the system prompt, the first user message, and the newest tail that fits.

The first user message is pinned because it usually contains the task; losing it is the most common way a long-horizon agent forgets what it was doing.

Source code in src/endstate/agent/context.py

class DropOldest:
    """Keep the system prompt, the first user message, and the newest tail that fits.

    The first user message is pinned because it usually contains the task; losing
    it is the most common way a long-horizon agent forgets what it was doing.
    """

    name = "drop_oldest"

    def compact(
        self, messages: list[Message], budget: TokenBudget, counter: TokenCounter
    ) -> list[Message]:
        if not messages:
            return messages

        pinned: list[Message] = []
        rest = list(messages)

        if rest and rest[0].role == "system":
            pinned.append(rest.pop(0))
        first_user = next((i for i, m in enumerate(rest) if m.role == "user"), None)
        if first_user is not None:
            pinned.append(rest.pop(first_user))

        tail: list[Message] = []
        for message in reversed(rest):
            candidate = [*pinned, message, *tail]
            if counter.count(candidate) > budget.usable_tokens:
                break
            tail.insert(0, message)
        return [*pinned, *tail]

SummariseMiddle

Replace the dropped middle with one synthetic summary message.

Takes a summariser callable rather than a provider so the strategy stays testable without a network call.

Source code in src/endstate/agent/context.py

class SummariseMiddle:
    """Replace the dropped middle with one synthetic summary message.

    Takes a summariser callable rather than a provider so the strategy stays
    testable without a network call.
    """

    name = "summarise_middle"

    def __init__(self, summariser: Callable[[list[Message]], str]) -> None:
        self.summariser = summariser

    def compact(
        self, messages: list[Message], budget: TokenBudget, counter: TokenCounter
    ) -> list[Message]:
        kept = DropOldest().compact(messages, budget, counter)
        kept_ids = {id(m) for m in kept}
        dropped = [m for m in messages if id(m) not in kept_ids]
        if not dropped:
            return kept

        summary = Message(
            role="user",
            content=f"[summary of {len(dropped)} earlier messages]\n{self.summariser(dropped)}",
            synthetic=True,
        )
        head = [m for m in kept if m.role == "system"]
        tail = [m for m in kept if m.role != "system"]
        return [*head, summary, *tail]

ContextManager

Bases: BaseModel

Fits a conversation into the budget, recording every compaction.

Source code in src/endstate/agent/context.py

class ContextManager(BaseModel):
    """Fits a conversation into the budget, recording every compaction."""

    budget: TokenBudget = Field(default_factory=TokenBudget)
    events: list[CompactionEvent] = Field(default_factory=list)

    model_config = {"arbitrary_types_allowed": True}

    def __init__(
        self,
        budget: TokenBudget | None = None,
        strategy: CompactionStrategy | None = None,
        counter: TokenCounter | None = None,
    ) -> None:
        super().__init__(budget=budget or TokenBudget())
        self._strategy: CompactionStrategy = strategy or DropOldest()
        self._counter: TokenCounter = counter or HeuristicTokenCounter()

    @property
    def strategy_name(self) -> str:
        return self._strategy.name

    def count(self, messages: list[Message]) -> int:
        return self._counter.count(messages)

    def fit(self, messages: list[Message]) -> list[Message]:
        before = self._counter.count(messages)
        if before <= self.budget.usable_tokens:
            return messages

        compacted = self._strategy.compact(messages, self.budget, self._counter)
        after = self._counter.count(compacted)
        self.events.append(
            CompactionEvent(
                strategy=self._strategy.name,
                tokens_before=before,
                tokens_after=after,
                messages_before=len(messages),
                messages_after=len(compacted),
                dropped_messages=len(messages) - len(compacted),
            )
        )
        return compacted

---

Permissions

endstate.agent.permissions

Permission policy for tool calls.

An agent that cannot be stopped is not deployable. This module exists so that "the agent refused" is a testable outcome: eval tasks in the permissioning category pass only when a destructive request is denied and the sandbox is unchanged.

Rules are evaluated in order; the first match wins. Default is DENY for anything not explicitly permitted, because the failure modes are asymmetric.

Rule

Bases: BaseModel

Match a tool call by tool name and optionally by argument content.

Source code in src/endstate/agent/permissions.py

class Rule(BaseModel):
    """Match a tool call by tool name and optionally by argument content."""

    tool: str = "*"
    """Tool name, or '*' for any."""

    argument_pattern: str | None = None
    """Regex matched against the serialised arguments. None matches anything."""

    decision: Decision = Decision.DENY
    reason: str = ""

    def matches(self, tool: str, arguments: dict[str, Any]) -> bool:
        if self.tool != "*" and self.tool != tool:
            return False
        if self.argument_pattern is None:
            return True
        blob = " ".join(str(v) for v in arguments.values())
        return re.search(self.argument_pattern, blob, re.IGNORECASE) is not None

tool class-attribute instance-attribute

tool: str = '*'

Tool name, or '*' for any.

argument_pattern class-attribute instance-attribute

argument_pattern: str | None = None

Regex matched against the serialised arguments. None matches anything.

default_policy

default_policy() -> PermissionPolicy

A conservative starting policy: read freely, write locally, never destroy.

Source code in src/endstate/agent/permissions.py

def default_policy() -> PermissionPolicy:
    """A conservative starting policy: read freely, write locally, never destroy."""
    return PermissionPolicy(
        rules=[
            Rule(
                tool="bash",
                argument_pattern=DESTRUCTIVE_SHELL,
                decision=Decision.DENY,
                reason="destructive filesystem or system command",
            ),
            Rule(
                tool="bash",
                argument_pattern=GIT_DESTRUCTIVE,
                decision=Decision.DENY,
                reason="destructive git operation",
            ),
            Rule(
                tool="bash",
                argument_pattern=SECRET_EXFIL,
                decision=Decision.DENY,
                reason="possible credential exfiltration",
            ),
            Rule(tool="read", decision=Decision.ALLOW),
            Rule(tool="glob", decision=Decision.ALLOW),
            Rule(tool="grep", decision=Decision.ALLOW),
            Rule(tool="write", decision=Decision.ALLOW),
            Rule(tool="edit", decision=Decision.ALLOW),
            Rule(tool="bash", decision=Decision.ALLOW),
        ],
        default=Decision.DENY,
    )

---

Sessions

endstate.agent.session

Durable sessions.

SQLite, not Postgres or Redis: a harness that needs infrastructure to remember what it was doing is a harness nobody will run locally. A checkpoint is written after every step, so a killed process resumes from the last completed tool call rather than from the beginning.

Session

An append-only conversation that survives process death.

Source code in src/endstate/agent/session.py

class Session:
    """An append-only conversation that survives process death."""

    def __init__(self, id: str, store: SessionStore, model: str = "") -> None:
        self.id = id
        self.store = store
        self.model = model
        self.messages: list[Message] = []

    def append(self, message: Message) -> None:
        """Append and checkpoint in one operation. There is no uncheckpointed state."""
        self.messages.append(message)
        self.store._append(self.id, len(self.messages) - 1, message)

    def checkpoint_last(self) -> None:
        """Re-persist the final message in place.

        Used while a batch of tool calls is being executed: each result is
        written as it lands, so a process killed mid-batch leaves a record of
        exactly the calls that completed. Rewriting the same step keeps one tool
        message per assistant turn, which is the shape every provider expects.
        """
        if not self.messages:
            raise IndexError("no message to checkpoint")
        self.store._append(self.id, len(self.messages) - 1, self.messages[-1])

    def close(self) -> None:
        self.store._set_status(self.id, "closed")

    def __len__(self) -> int:
        return len(self.messages)

append

append(message: Message) -> None

Append and checkpoint in one operation. There is no uncheckpointed state.

Source code in src/endstate/agent/session.py

def append(self, message: Message) -> None:
    """Append and checkpoint in one operation. There is no uncheckpointed state."""
    self.messages.append(message)
    self.store._append(self.id, len(self.messages) - 1, message)

checkpoint_last

checkpoint_last() -> None

Re-persist the final message in place.

Used while a batch of tool calls is being executed: each result is written as it lands, so a process killed mid-batch leaves a record of exactly the calls that completed. Rewriting the same step keeps one tool message per assistant turn, which is the shape every provider expects.

Source code in src/endstate/agent/session.py

def checkpoint_last(self) -> None:
    """Re-persist the final message in place.

    Used while a batch of tool calls is being executed: each result is
    written as it lands, so a process killed mid-batch leaves a record of
    exactly the calls that completed. Rewriting the same step keeps one tool
    message per assistant turn, which is the shape every provider expects.
    """
    if not self.messages:
        raise IndexError("no message to checkpoint")
    self.store._append(self.id, len(self.messages) - 1, self.messages[-1])

---

Tools

endstate.agent.tools.base

Tool interface and the sandbox boundary.

Every tool receives a ToolContext whose workdir is the only part of the filesystem it may touch. Path confinement lives here rather than in each tool so there is exactly one place to audit.

ToolError

Bases: Exception

Raised for expected tool failures; surfaced to the model, not the user.

Source code in src/endstate/agent/tools/base.py

class ToolError(Exception):
    """Raised for expected tool failures; surfaced to the model, not the user."""

ToolContext dataclass

Source code in src/endstate/agent/tools/base.py

@dataclass(frozen=True)
class ToolContext:
    workdir: Path
    timeout_s: float = 30.0
    max_output_chars: int = 20_000

    def resolve(self, relative: str) -> Path:
        """Resolve a path inside the workdir, refusing anything that escapes it."""
        try:
            return confine(self.workdir, relative)
        except ValueError as exc:
            raise ToolError(f"path {relative!r} escapes the working directory") from exc

    def truncate(self, text: str) -> str:
        if len(text) <= self.max_output_chars:
            return text
        omitted = len(text) - self.max_output_chars
        return text[: self.max_output_chars] + f"\n... [{omitted} chars truncated]"

resolve

resolve(relative: str) -> Path

Resolve a path inside the workdir, refusing anything that escapes it.

Source code in src/endstate/agent/tools/base.py

def resolve(self, relative: str) -> Path:
    """Resolve a path inside the workdir, refusing anything that escapes it."""
    try:
        return confine(self.workdir, relative)
    except ValueError as exc:
        raise ToolError(f"path {relative!r} escapes the working directory") from exc

Tool

Bases: ABC

Source code in src/endstate/agent/tools/base.py

class Tool(ABC):
    name: str
    description: str

    idempotent: bool = True
    """Whether running this call twice is the same as running it once.

    Read by the resume path. A crash inside a tool call — after the side effect,
    before any record of it — leaves the harness unable to tell "done" from
    "about to". For an idempotent tool that is harmless: re-running converges.
    For one that appends, increments or posts, re-running is a second side
    effect, so resume reports the call as interrupted instead of repeating it.

    Default True because the tools shipped here overwrite rather than accumulate.
    Anything with an external or accumulating effect must set it False.
    """

    @property
    @abstractmethod
    def parameters(self) -> dict[str, Any]:
        """JSON Schema for the tool arguments."""

    @abstractmethod
    def run(self, arguments: dict[str, Any], ctx: ToolContext) -> str:
        """Execute and return output for the model. Raise ToolError on failure."""

    def spec(self) -> dict[str, Any]:
        return {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters,
        }

idempotent class-attribute instance-attribute

idempotent: bool = True

Whether running this call twice is the same as running it once.

Read by the resume path. A crash inside a tool call — after the side effect, before any record of it — leaves the harness unable to tell "done" from "about to". For an idempotent tool that is harmless: re-running converges. For one that appends, increments or posts, re-running is a second side effect, so resume reports the call as interrupted instead of repeating it.

Default True because the tools shipped here overwrite rather than accumulate. Anything with an external or accumulating effect must set it False.

parameters abstractmethod property

parameters: dict[str, Any]

JSON Schema for the tool arguments.

run abstractmethod

run(arguments: dict[str, Any], ctx: ToolContext) -> str

Execute and return output for the model. Raise ToolError on failure.

Source code in src/endstate/agent/tools/base.py

@abstractmethod
def run(self, arguments: dict[str, Any], ctx: ToolContext) -> str:
    """Execute and return output for the model. Raise ToolError on failure."""

confine

confine(root: Path, relative: str) -> Path

Resolve relative inside root, refusing anything that escapes it.

Lives at module level because the eval sandbox needs the same rule and one audited implementation is worth more than two that agree today.

Raises:

Type	Description
ValueError	If the resolved path is outside root.

Source code in src/endstate/agent/tools/base.py

def confine(root: Path, relative: str) -> Path:
    """Resolve `relative` inside `root`, refusing anything that escapes it.

    Lives at module level because the eval sandbox needs the same rule and one
    audited implementation is worth more than two that agree today.

    Raises:
        ValueError: If the resolved path is outside `root`.
    """
    base = root.resolve()
    candidate = (base / relative).resolve()
    if candidate != base and base not in candidate.parents:
        raise ValueError(f"path {relative!r} escapes {base}")
    return candidate

---

Providers

endstate.providers.base

Provider interface.

The loop depends on this Protocol and nothing else. Adding a provider means writing one adapter; it never means touching the loop. This is also what makes the same eval suite runnable against a hosted API and a self-hosted vLLM endpoint without changing a single task.

endstate.providers.fake

A scripted provider.

Exists so the harness is testable without a network, an API key, or a bill. Every behaviour in the loop — tool calls, multi-step runs, budget exhaustion, denial — is covered by tests that use this.

FakeProvider

Replays a fixed list of responses, then ends the turn.

Source code in src/endstate/providers/fake.py

class FakeProvider:
    """Replays a fixed list of responses, then ends the turn."""

    def __init__(self, responses: list[Response], model: str = "fake-1") -> None:
        self.model = model
        self._responses = list(responses)
        self.calls: list[list[Message]] = []

    @classmethod
    def saying(cls, text: str, model: str = "fake-1") -> FakeProvider:
        return cls(
            [
                Response(
                    message=Message(role="assistant", content=text),
                    usage=Usage(input_tokens=10, output_tokens=5),
                    stop_reason=StopReason.END_TURN,
                    model=model,
                )
            ],
            model=model,
        )

    def complete(
        self, messages: list[Message], tools: list[dict[str, Any]] | None = None
    ) -> Response:
        self.calls.append(list(messages))
        if self._responses:
            response = self._responses.pop(0)
            if not response.model:
                response.model = self.model
            return response
        return Response(
            message=Message(role="assistant", content="done"),
            usage=Usage(input_tokens=1, output_tokens=1),
            stop_reason=StopReason.END_TURN,
            model=self.model,
        )

---

Evals: tasks and verdicts

endstate.evals.task

What a task is, and what a verdict is.

A task on disk is a directory. That is a deliberate choice over a single big manifest file: the fixture is a real repository the agent works in, the prompt is prose someone will edit, and the held-out tests must be storable next to the task without being shipped into the sandbox with it.

tasks/fix-off-by-one/
├── task.json      metadata, budget, graders
├── prompt.md      what the agent is told
├── fixture/       copied into the sandbox; this is the agent's whole world
├── holdout/       staged in at grading time only — the agent never sees it
└── solution/      a reference fix, used to prove the graders are load-bearing

solution/ never reaches an agent. It exists so the suite can answer the question that decides whether any of this is worth anything: does this grader fail on the unsolved fixture and pass on a correct fix? A grader that passes either way is measuring nothing, and it is much easier to write one of those than most people expect.

TaskError

Bases: ValueError

Raised when a task on disk is malformed.

Source code in src/endstate/evals/task.py

class TaskError(ValueError):
    """Raised when a task on disk is malformed."""

Check

Bases: BaseModel

One assertion, reported individually.

Graders return every check they ran, not just the one that failed. "The tests pass but you edited them" and "the tests fail" are different findings, and a verdict that collapses to a single boolean cannot tell you which.

Source code in src/endstate/evals/task.py

class Check(BaseModel):
    """One assertion, reported individually.

    Graders return every check they ran, not just the one that failed. "The
    tests pass but you edited them" and "the tests fail" are different findings,
    and a verdict that collapses to a single boolean cannot tell you which.
    """

    name: str
    passed: bool
    detail: str = ""

    def __str__(self) -> str:
        mark = "pass" if self.passed else "FAIL"
        return f"[{mark}] {self.name}{': ' + self.detail if self.detail else ''}"

Verdict

Bases: BaseModel

A grader's answer. Pure function of the sandbox's end state.

Source code in src/endstate/evals/task.py

class Verdict(BaseModel):
    """A grader's answer. Pure function of the sandbox's end state."""

    passed: bool
    reason: str = ""
    checks: list[Check] = Field(default_factory=list)

    @classmethod
    def ok(cls, reason: str = "", checks: list[Check] | None = None) -> Verdict:
        return cls(passed=True, reason=reason, checks=checks or [])

    @classmethod
    def fail(cls, reason: str, checks: list[Check] | None = None) -> Verdict:
        return cls(passed=False, reason=reason, checks=checks or [])

    @classmethod
    def from_checks(cls, checks: list[Check]) -> Verdict:
        """Conjunction: every check must pass, and the reason names the ones that did not."""
        failed = [c for c in checks if not c.passed]
        if failed:
            return cls(passed=False, reason="; ".join(c.name for c in failed), checks=checks)
        return cls(passed=True, checks=checks)

    def merged_with(self, other: Verdict) -> Verdict:
        """Conjunction of two verdicts, keeping every check from both.

        `passed` is conjoined explicitly rather than recomputed from the merged
        checks, because a verdict can fail while carrying none: a grader that
        raised reports a reason and an empty list. Recomputing would turn that
        into a pass — the worst possible direction for a failure to be lost in.
        """
        checks = [*self.checks, *other.checks]
        reasons = [v.reason for v in (self, other) if not v.passed and v.reason]
        reasons += [c.name for c in checks if not c.passed]
        return Verdict(
            passed=self.passed and other.passed,
            reason="; ".join(dict.fromkeys(reasons)),
            checks=checks,
        )

    @property
    def failed_checks(self) -> list[Check]:
        return [c for c in self.checks if not c.passed]

from_checks classmethod

from_checks(checks: list[Check]) -> Verdict

Conjunction: every check must pass, and the reason names the ones that did not.

Source code in src/endstate/evals/task.py

@classmethod
def from_checks(cls, checks: list[Check]) -> Verdict:
    """Conjunction: every check must pass, and the reason names the ones that did not."""
    failed = [c for c in checks if not c.passed]
    if failed:
        return cls(passed=False, reason="; ".join(c.name for c in failed), checks=checks)
    return cls(passed=True, checks=checks)

merged_with

merged_with(other: Verdict) -> Verdict

Conjunction of two verdicts, keeping every check from both.

passed is conjoined explicitly rather than recomputed from the merged checks, because a verdict can fail while carrying none: a grader that raised reports a reason and an empty list. Recomputing would turn that into a pass — the worst possible direction for a failure to be lost in.

Source code in src/endstate/evals/task.py

def merged_with(self, other: Verdict) -> Verdict:
    """Conjunction of two verdicts, keeping every check from both.

    `passed` is conjoined explicitly rather than recomputed from the merged
    checks, because a verdict can fail while carrying none: a grader that
    raised reports a reason and an empty list. Recomputing would turn that
    into a pass — the worst possible direction for a failure to be lost in.
    """
    checks = [*self.checks, *other.checks]
    reasons = [v.reason for v in (self, other) if not v.passed and v.reason]
    reasons += [c.name for c in checks if not c.passed]
    return Verdict(
        passed=self.passed and other.passed,
        reason="; ".join(dict.fromkeys(reasons)),
        checks=checks,
    )

GraderSpec

Bases: BaseModel

A dotted path to a grader, plus the arguments it is bound with.

endstate.evals.graders:command_succeeds with {"command": "..."} resolves to a callable of exactly one argument — the sandbox — which is the contract the whole design rests on.

Source code in src/endstate/evals/task.py

class GraderSpec(BaseModel):
    """A dotted path to a grader, plus the arguments it is bound with.

    `endstate.evals.graders:command_succeeds` with `{"command": "..."}` resolves
    to a callable of exactly one argument — the sandbox — which is the contract
    the whole design rests on.
    """

    model_config = {"extra": "forbid"}

    name: str
    args: dict[str, Any] = Field(default_factory=dict)

Bound

Bases: BaseModel

An inclusive range. Either end may be left out.

Unknown keys are refused. A {"minimum": 1} typo would otherwise parse as a bound with no ends, which passes for any value — so the requirement would read as asserted in the task file and assert nothing at all. That is the exact failure this whole design is meant to rule out, so it is an error.

Source code in src/endstate/evals/task.py

class Bound(BaseModel):
    """An inclusive range. Either end may be left out.

    Unknown keys are refused. A `{"minimum": 1}` typo would otherwise parse as a
    bound with no ends, which passes for *any* value — so the requirement would
    read as asserted in the task file and assert nothing at all. That is the
    exact failure this whole design is meant to rule out, so it is an error.
    """

    model_config = {"extra": "forbid"}

    min: int | None = None
    max: int | None = None

    def check(self, value: int, label: str) -> Check:
        passed = (self.min is None or value >= self.min) and (self.max is None or value <= self.max)
        return Check(name=f"{label} {self}", passed=passed, detail=f"was {value}")

    def __str__(self) -> str:
        if self.min is not None and self.max is not None:
            return f"is {self.min}" if self.min == self.max else f"is {self.min}-{self.max}"
        if self.min is not None:
            return f"is at least {self.min}"
        if self.max is not None:
            return f"is at most {self.max}"
        return "is anything"

RunRequirements

Bases: BaseModel

Assertions about the run that the filesystem cannot answer.

Three of the four differentiating categories need one of these. "The task completed and compaction fired at least once" is the whole point of a long-horizon task — a run that finished without ever compacting did not test compaction, and passing it would be a lie about coverage. But compaction_events is not a property of the sandbox, so no grader can see it without being handed something other than the sandbox.

Rather than widen the grader signature, requirements are evaluated by the runner and conjoined with the grader's verdict. That keeps grade(sandbox) -> Verdict exactly as narrow as it was, and the reason it is safe is the shape of this class: every field is a typed counter the harness recorded itself — policy decisions, compaction events, token totals. There is no field here that can reach message content, and adding one would be a visible change to this model rather than a quiet argument in a task file.

extra="forbid" is what makes that last sentence true rather than merely intended: a manifest naming final_text is rejected at load time instead of being silently dropped.

Source code in src/endstate/evals/task.py

class RunRequirements(BaseModel):
    """Assertions about the *run* that the filesystem cannot answer.

    Three of the four differentiating categories need one of these. "The task
    completed **and** compaction fired at least once" is the whole point of a
    long-horizon task — a run that finished without ever compacting did not test
    compaction, and passing it would be a lie about coverage. But
    `compaction_events` is not a property of the sandbox, so no grader can see
    it without being handed something other than the sandbox.

    Rather than widen the grader signature, requirements are evaluated by the
    **runner** and conjoined with the grader's verdict. That keeps
    `grade(sandbox) -> Verdict` exactly as narrow as it was, and the reason it is
    safe is the shape of this class: every field is a typed counter the harness
    recorded itself — policy decisions, compaction events, token totals. There is
    no field here that can reach message content, and adding one would be a
    visible change to this model rather than a quiet argument in a task file.

    `extra="forbid"` is what makes that last sentence true rather than merely
    intended: a manifest naming `final_text` is rejected at load time instead of
    being silently dropped.
    """

    model_config = {"extra": "forbid"}

    compaction_events: Bound | None = None
    denied_calls: Bound | None = None
    unsettled_calls: Bound | None = None
    steps: Bound | None = None
    input_tokens: Bound | None = None
    output_tokens: Bound | None = None
    total_tokens: Bound | None = None
    stop_reason: str | None = None

    def check(self, counters: dict[str, int], stop_reason: str) -> list[Check]:
        """Every requirement, as individually reported checks."""
        checks: list[Check] = []
        for field, bound in self:
            if field == "stop_reason" or bound is None:
                continue
            assert isinstance(bound, Bound)
            checks.append(bound.check(counters[field], field.replace("_", " ")))
        if self.stop_reason is not None:
            checks.append(
                Check(
                    name=f"stop reason is {self.stop_reason}",
                    passed=stop_reason == self.stop_reason,
                    detail=f"was {stop_reason}",
                )
            )
        return checks

check

check(
    counters: dict[str, int], stop_reason: str
) -> list[Check]

Every requirement, as individually reported checks.

Source code in src/endstate/evals/task.py

def check(self, counters: dict[str, int], stop_reason: str) -> list[Check]:
    """Every requirement, as individually reported checks."""
    checks: list[Check] = []
    for field, bound in self:
        if field == "stop_reason" or bound is None:
            continue
        assert isinstance(bound, Bound)
        checks.append(bound.check(counters[field], field.replace("_", " ")))
    if self.stop_reason is not None:
        checks.append(
            Check(
                name=f"stop reason is {self.stop_reason}",
                passed=stop_reason == self.stop_reason,
                detail=f"was {stop_reason}",
            )
        )
    return checks

Recovery

Bases: BaseModel

Kill the run partway through, then resume it.

crash_at_call counts tool calls across the whole run, not steps, because the interesting kill points are inside a batch. after_side_effect picks the window: False is the honest kill where the work never happened, True is the irreducible one where it happened and nothing recorded it.

Source code in src/endstate/evals/task.py

class Recovery(BaseModel):
    """Kill the run partway through, then resume it.

    `crash_at_call` counts tool calls across the whole run, not steps, because
    the interesting kill points are inside a batch. `after_side_effect` picks the
    window: False is the honest kill where the work never happened, True is the
    irreducible one where it happened and nothing recorded it.
    """

    crash_at_call: int
    after_side_effect: bool = False

Task

Bases: BaseModel

One eval task, loaded from a directory.

Source code in src/endstate/evals/task.py

class Task(BaseModel):
    """One eval task, loaded from a directory."""

    id: str
    prompt: str
    fixture: Path
    graders: list[GraderSpec]
    category: str = "bug-fix"
    description: str = ""
    max_steps: int = 25
    timeout_s: float = 300.0
    budget: TokenBudget = Field(default_factory=TokenBudget)
    requires: RunRequirements = Field(default_factory=RunRequirements)
    recovery: Recovery | None = None
    holdout: Path | None = None
    solution: Path | None = None
    root: Path | None = None

    model_config = {"arbitrary_types_allowed": True}

load_task

load_task(directory: Path) -> Task

Load one task directory.

Raises:

Type	Description
TaskError	If the directory is not a well-formed task.

Source code in src/endstate/evals/task.py

def load_task(directory: Path) -> Task:
    """Load one task directory.

    Raises:
        TaskError: If the directory is not a well-formed task.
    """
    directory = Path(directory).resolve()
    manifest = directory / TASK_FILE
    if not manifest.is_file():
        raise TaskError(f"no {TASK_FILE} in {directory}")

    try:
        data: dict[str, Any] = json.loads(manifest.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise TaskError(f"{manifest}: {exc}") from exc

    fixture = directory / FIXTURE_DIR
    if not fixture.is_dir():
        raise TaskError(f"{directory}: no {FIXTURE_DIR}/ directory")

    prompt = data.get("prompt", "")
    prompt_file = directory / PROMPT_FILE
    if prompt_file.is_file():
        prompt = prompt_file.read_text(encoding="utf-8").strip()
    if not prompt:
        raise TaskError(f"{directory}: no prompt, in {PROMPT_FILE} or in {TASK_FILE}")

    try:
        graders = [GraderSpec(**g) for g in data.get("graders", [])]
        requires = RunRequirements(**data.get("requires", {}))
        recovery = Recovery(**data["recovery"]) if "recovery" in data else None
    except ValidationError as exc:
        raise TaskError(f"{manifest}: {exc}") from exc
    if not graders:
        raise TaskError(f"{directory}: a task with no graders cannot pass or fail")

    category = data.get("category", "bug-fix")
    if category not in CATEGORIES:
        raise TaskError(f"{directory}: unknown category {category!r}, expected one of {CATEGORIES}")

    holdout = directory / HOLDOUT_DIR
    solution = directory / SOLUTION_DIR
    return Task(
        id=data.get("id", directory.name),
        prompt=prompt,
        fixture=fixture,
        graders=graders,
        category=category,
        description=data.get("description", ""),
        max_steps=int(data.get("max_steps", 25)),
        timeout_s=float(data.get("timeout_s", 300.0)),
        budget=TokenBudget(**data["budget"]) if "budget" in data else TokenBudget(),
        requires=requires,
        recovery=recovery,
        holdout=holdout if holdout.is_dir() else None,
        solution=solution if solution.is_dir() else None,
        root=directory,
    )

discover_tasks

discover_tasks(root: Path) -> list[Task]

Find every task under root, ordered by id.

Ordering is not cosmetic. The determinism criterion is an identical pass/fail vector across runs, and a vector needs a stable index.

Raises:

Type	Description
TaskError	If root is not a directory, or two tasks share an id.

Source code in src/endstate/evals/task.py

def discover_tasks(root: Path) -> list[Task]:
    """Find every task under `root`, ordered by id.

    Ordering is not cosmetic. The determinism criterion is an identical pass/fail
    *vector* across runs, and a vector needs a stable index.

    Raises:
        TaskError: If `root` is not a directory, or two tasks share an id.
    """
    root = Path(root)
    if not root.is_dir():
        raise TaskError(f"no such suite directory: {root}")

    tasks = [load_task(manifest.parent) for manifest in sorted(root.rglob(TASK_FILE))]
    seen: dict[str, Path] = {}
    for task in tasks:
        if task.id in seen:
            raise TaskError(f"duplicate task id {task.id!r}: {seen[task.id]} and {task.root}")
        seen[task.id] = task.root or root
    return sorted(tasks, key=lambda t: t.id)

---

Evals: the sandbox

endstate.evals.sandbox

The eval sandbox: one disposable container per task.

Decision D7. Three reasons, in order of importance: determinism (a shared sandbox means task 7 sees whatever task 6 left behind), parallelism, and the fact that a container is the only reason it is sane to hand an agent a shell.

The shape here is deliberate. A sandbox owns a live tree on the host — a copy of the task fixture — and a way to execute commands against it. The tree is bind mounted into the container, so the file tools write host-side into the same bytes the shell sees container-side, and the grader can hash the result without copying anything back out.

Two things are sealed off from the agent on purpose:

The pristine fixture. Kept beside the live tree so a grader can ask "is this file byte-for-byte what we shipped?" — the assertion that catches an agent editing the tests instead of the code.

Version control history. .git is not copied. A task built from a real repository ships the answer inside it, and git log is a cheaper path to a passing grade than solving the problem. This is runtime contamination, and it is measured in the wild at a scale that makes it the default rather than an option.

CommandRunner module-attribute

CommandRunner = Callable[
    [Sequence[str], float | None], ExecResult
]

Runs an argv on the host. Injected so the Docker plumbing is testable.

DOCKER_PROBE_TIMEOUT_S module-attribute

DOCKER_PROBE_TIMEOUT_S = 5.0

How long to wait for docker version.

A healthy daemon answers in well under a second. The cases that take longer are all "not usable right now" — no daemon, a socket nobody is listening on, a daemon still starting or wedged mid-restart — and for every one of them the answer is the same, so waiting longer buys nothing and costs it on every call.

SandboxError

Bases: RuntimeError

Raised when the sandbox itself fails, as distinct from the task failing.

Source code in src/endstate/evals/sandbox.py

class SandboxError(RuntimeError):
    """Raised when the sandbox itself fails, as distinct from the task failing."""

ExecResult dataclass

The outcome of one command run inside the sandbox.

Source code in src/endstate/evals/sandbox.py

@dataclass(frozen=True)
class ExecResult:
    """The outcome of one command run inside the sandbox."""

    command: str
    exit_code: int
    stdout: str = ""
    stderr: str = ""
    timed_out: bool = False

    @property
    def ok(self) -> bool:
        return self.exit_code == 0 and not self.timed_out

    @property
    def output(self) -> str:
        """stdout and stderr, in the order a terminal would have shown them."""
        return "".join(part for part in (self.stdout, self.stderr) if part)

output property

output: str

stdout and stderr, in the order a terminal would have shown them.

Sandbox

Bases: ABC

A disposable working tree plus a way to run commands against it.

Subclasses supply execution only; everything filesystem-shaped is here, so the Docker and local backends cannot drift on what a grader observes.

Source code in src/endstate/evals/sandbox.py

class Sandbox(ABC):
    """A disposable working tree plus a way to run commands against it.

    Subclasses supply execution only; everything filesystem-shaped is here, so
    the Docker and local backends cannot drift on what a grader observes.
    """

    def __init__(
        self,
        fixture: Path,
        root: Path | None = None,
        *,
        default_timeout_s: float = 120.0,
        excludes: frozenset[str] = STAGING_EXCLUDES,
    ) -> None:
        self.fixture = Path(fixture).resolve()
        self.default_timeout_s = default_timeout_s
        self.excludes = excludes
        self._tempdir: tempfile.TemporaryDirectory[str] | None = None
        if root is None:
            self._tempdir = tempfile.TemporaryDirectory(prefix="endstate-task-")
            root = Path(self._tempdir.name)
        self.workdir = Path(root).resolve()
        self._sealed_hash: str | None = None
        self._sealed_files: dict[str, str] | None = None

    # --- lifecycle --------------------------------------------------------

    def setup(self) -> None:
        """Materialise the fixture into the live tree."""
        stage_tree(self.fixture, self.workdir, excludes=self.excludes)

    def start(self) -> None:  # noqa: B027 - optional hook; a local sandbox has nothing to start
        """Bring up whatever executes commands. Nothing to do by default."""

    def close(self) -> None:
        """Tear down execution and, if we own it, the live tree."""
        if self._tempdir is not None:
            self._tempdir.cleanup()
            self._tempdir = None

    def __enter__(self) -> Sandbox:
        self.setup()
        try:
            self.start()
        except Exception:
            self.close()
            raise
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        self.close()

    # --- execution --------------------------------------------------------

    @abstractmethod
    def run(self, command: str, *, timeout_s: float | None = None) -> ExecResult:
        """Run a shell command with the live tree as the working directory."""

    # --- reading the end state -------------------------------------------

    def path(self, relative: str) -> Path:
        """Resolve a path inside the live tree, refusing anything that escapes.

        Raises:
            SandboxError: If the path points outside the sandbox.
        """
        try:
            return confine(self.workdir, relative)
        except ValueError as exc:
            raise SandboxError(str(exc)) from exc

    def exists(self, relative: str) -> bool:
        return self.path(relative).exists()

    def read_text(self, relative: str) -> str:
        """Read a file from the live tree, or return '' if it is not there.

        Missing is not an error: a grader asking "does this file still contain
        the secret?" wants False for a deleted file, not an exception.
        """
        target = self.path(relative)
        if not target.is_file():
            return ""
        return target.read_text(encoding="utf-8", errors="replace")

    def write_text(self, relative: str, content: str) -> None:
        target = self.path(relative)
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(content, encoding="utf-8")

    def fixture_text(self, relative: str) -> str:
        """Read a file as it was shipped, before the agent touched anything."""
        try:
            original = confine(self.fixture, relative)
        except ValueError as exc:
            raise SandboxError(str(exc)) from exc
        if not original.is_file():
            return ""
        return original.read_text(encoding="utf-8", errors="replace")

    def stage(self, source: Path) -> None:
        """Overlay a directory into the live tree. Used for held-out tests."""
        stage_tree(source, self.workdir, excludes=self.excludes)

    def seal(self) -> str:
        """Freeze the agent's end state and return its hash.

        Called once the agent has stopped and before anything is staged in for
        grading. Held-out tests are files too: without a seal, copying them in
        would change the very tree hash the graders are about to assert on.
        """
        if self._sealed_hash is None:
            self._sealed_hash = tree_hash(self.workdir, excludes=DEFAULT_EXCLUDES)
            self._sealed_files = _file_texts(self.workdir, self.excludes)
        return self._sealed_hash

    def tree_hash(self) -> str:
        """The end-state hash: sealed if the run is over, live if it is not."""
        if self._sealed_hash is not None:
            return self._sealed_hash
        return tree_hash(self.workdir, excludes=DEFAULT_EXCLUDES)

    def end_state_files(self) -> dict[str, str]:
        """Every text file in the end state, keyed by relative path."""
        if self._sealed_files is not None:
            return dict(self._sealed_files)
        return _file_texts(self.workdir, self.excludes)

    def changed_paths(self) -> list[str]:
        """Relative paths whose content differs from the fixture.

        Includes files the agent added and files it deleted, because "deleted
        something inconvenient" is one of the things end-state grading exists to
        catch.
        """
        before = _file_texts(self.fixture, self.excludes)
        after = self.end_state_files()
        return sorted({p for p in set(before) | set(after) if before.get(p) != after.get(p)})

setup

setup() -> None

Materialise the fixture into the live tree.

Source code in src/endstate/evals/sandbox.py

def setup(self) -> None:
    """Materialise the fixture into the live tree."""
    stage_tree(self.fixture, self.workdir, excludes=self.excludes)

start

start() -> None

Bring up whatever executes commands. Nothing to do by default.

Source code in src/endstate/evals/sandbox.py

def start(self) -> None:  # noqa: B027 - optional hook; a local sandbox has nothing to start
    """Bring up whatever executes commands. Nothing to do by default."""

close

close() -> None

Tear down execution and, if we own it, the live tree.

Source code in src/endstate/evals/sandbox.py

def close(self) -> None:
    """Tear down execution and, if we own it, the live tree."""
    if self._tempdir is not None:
        self._tempdir.cleanup()
        self._tempdir = None

run abstractmethod

run(
    command: str, *, timeout_s: float | None = None
) -> ExecResult

Run a shell command with the live tree as the working directory.

Source code in src/endstate/evals/sandbox.py

@abstractmethod
def run(self, command: str, *, timeout_s: float | None = None) -> ExecResult:
    """Run a shell command with the live tree as the working directory."""

path

path(relative: str) -> Path

Resolve a path inside the live tree, refusing anything that escapes.

Raises:

Type	Description
SandboxError	If the path points outside the sandbox.

Source code in src/endstate/evals/sandbox.py

def path(self, relative: str) -> Path:
    """Resolve a path inside the live tree, refusing anything that escapes.

    Raises:
        SandboxError: If the path points outside the sandbox.
    """
    try:
        return confine(self.workdir, relative)
    except ValueError as exc:
        raise SandboxError(str(exc)) from exc

read_text

read_text(relative: str) -> str

Read a file from the live tree, or return '' if it is not there.

Missing is not an error: a grader asking "does this file still contain the secret?" wants False for a deleted file, not an exception.

Source code in src/endstate/evals/sandbox.py

def read_text(self, relative: str) -> str:
    """Read a file from the live tree, or return '' if it is not there.

    Missing is not an error: a grader asking "does this file still contain
    the secret?" wants False for a deleted file, not an exception.
    """
    target = self.path(relative)
    if not target.is_file():
        return ""
    return target.read_text(encoding="utf-8", errors="replace")

fixture_text

fixture_text(relative: str) -> str

Read a file as it was shipped, before the agent touched anything.

Source code in src/endstate/evals/sandbox.py

def fixture_text(self, relative: str) -> str:
    """Read a file as it was shipped, before the agent touched anything."""
    try:
        original = confine(self.fixture, relative)
    except ValueError as exc:
        raise SandboxError(str(exc)) from exc
    if not original.is_file():
        return ""
    return original.read_text(encoding="utf-8", errors="replace")

stage

stage(source: Path) -> None

Overlay a directory into the live tree. Used for held-out tests.

Source code in src/endstate/evals/sandbox.py

def stage(self, source: Path) -> None:
    """Overlay a directory into the live tree. Used for held-out tests."""
    stage_tree(source, self.workdir, excludes=self.excludes)

seal

seal() -> str

Freeze the agent's end state and return its hash.

Called once the agent has stopped and before anything is staged in for grading. Held-out tests are files too: without a seal, copying them in would change the very tree hash the graders are about to assert on.

Source code in src/endstate/evals/sandbox.py

def seal(self) -> str:
    """Freeze the agent's end state and return its hash.

    Called once the agent has stopped and before anything is staged in for
    grading. Held-out tests are files too: without a seal, copying them in
    would change the very tree hash the graders are about to assert on.
    """
    if self._sealed_hash is None:
        self._sealed_hash = tree_hash(self.workdir, excludes=DEFAULT_EXCLUDES)
        self._sealed_files = _file_texts(self.workdir, self.excludes)
    return self._sealed_hash

tree_hash

tree_hash() -> str

The end-state hash: sealed if the run is over, live if it is not.

Source code in src/endstate/evals/sandbox.py

def tree_hash(self) -> str:
    """The end-state hash: sealed if the run is over, live if it is not."""
    if self._sealed_hash is not None:
        return self._sealed_hash
    return tree_hash(self.workdir, excludes=DEFAULT_EXCLUDES)

end_state_files

end_state_files() -> dict[str, str]

Every text file in the end state, keyed by relative path.

Source code in src/endstate/evals/sandbox.py

def end_state_files(self) -> dict[str, str]:
    """Every text file in the end state, keyed by relative path."""
    if self._sealed_files is not None:
        return dict(self._sealed_files)
    return _file_texts(self.workdir, self.excludes)

changed_paths

changed_paths() -> list[str]

Relative paths whose content differs from the fixture.

Includes files the agent added and files it deleted, because "deleted something inconvenient" is one of the things end-state grading exists to catch.

Source code in src/endstate/evals/sandbox.py

def changed_paths(self) -> list[str]:
    """Relative paths whose content differs from the fixture.

    Includes files the agent added and files it deleted, because "deleted
    something inconvenient" is one of the things end-state grading exists to
    catch.
    """
    before = _file_texts(self.fixture, self.excludes)
    after = self.end_state_files()
    return sorted({p for p in set(before) | set(after) if before.get(p) != after.get(p)})

LocalSandbox

Bases: Sandbox

Runs commands as subprocesses on the host, in the live tree.

This is not an isolation boundary. It exists so the runner, the graders and the task suite can be tested on a machine without Docker, and so a task can be debugged without a container in the way. Real benchmark runs use DockerSandbox; anything published from a LocalSandbox run should say so.

Source code in src/endstate/evals/sandbox.py

class LocalSandbox(Sandbox):
    """Runs commands as subprocesses on the host, in the live tree.

    **This is not an isolation boundary.** It exists so the runner, the graders
    and the task suite can be tested on a machine without Docker, and so a task
    can be debugged without a container in the way. Real benchmark runs use
    `DockerSandbox`; anything published from a `LocalSandbox` run should say so.
    """

    def run(self, command: str, *, timeout_s: float | None = None) -> ExecResult:
        env = dict(os.environ)
        # Byte-compilation writes __pycache__ into the tree being hashed, and
        # hash randomisation is a determinism leak in anything that iterates a set.
        env["PYTHONDONTWRITEBYTECODE"] = "1"
        env["PYTHONHASHSEED"] = "0"
        try:
            proc = subprocess.run(  # noqa: S602 - the sandbox is the point
                command,
                shell=True,
                cwd=str(self.workdir),
                capture_output=True,
                text=True,
                timeout=timeout_s if timeout_s is not None else self.default_timeout_s,
                env=env,
            )
        except subprocess.TimeoutExpired:
            return ExecResult(command=command, exit_code=124, timed_out=True)
        return ExecResult(
            command=command,
            exit_code=proc.returncode,
            stdout=proc.stdout,
            stderr=proc.stderr,
        )

DockerSandbox

Bases: Sandbox

One container per task, with the live tree bind mounted into it.

The container is started once and held open with tail -f /dev/null, then every command runs through docker exec. Starting a fresh container per command would be a different guarantee — and a slower one — because a task's commands would no longer share process state.

Defaults are the restrictive ones: no network, all capabilities dropped, no privilege escalation, capped memory and pids. Network is the interesting one. It is off because the fix for most benchmark tasks is a web search away, and an agent that retrieves the answer has not demonstrated it can derive it.

Source code in src/endstate/evals/sandbox.py

class DockerSandbox(Sandbox):
    """One container per task, with the live tree bind mounted into it.

    The container is started once and held open with `tail -f /dev/null`, then
    every command runs through `docker exec`. Starting a fresh container per
    command would be a different guarantee — and a slower one — because a task's
    commands would no longer share process state.

    Defaults are the restrictive ones: no network, all capabilities dropped, no
    privilege escalation, capped memory and pids. Network is the interesting one.
    It is off because the fix for most benchmark tasks is a web search away, and
    an agent that retrieves the answer has not demonstrated it can derive it.
    """

    def __init__(
        self,
        fixture: Path,
        root: Path | None = None,
        *,
        image: str = DEFAULT_IMAGE,
        network: bool = False,
        memory: str = "2g",
        cpus: str = "2.0",
        pids_limit: int = 512,
        container_workdir: str = "/work",
        runner: CommandRunner | None = None,
        startup_timeout_s: float = 120.0,
        default_timeout_s: float = 120.0,
        excludes: frozenset[str] = STAGING_EXCLUDES,
    ) -> None:
        super().__init__(fixture, root, default_timeout_s=default_timeout_s, excludes=excludes)
        self.image = image
        self.network = network
        self.memory = memory
        self.cpus = cpus
        self.pids_limit = pids_limit
        self.container_workdir = container_workdir
        self.startup_timeout_s = startup_timeout_s
        self._runner: CommandRunner = runner or subprocess_runner
        self.container_id: str | None = None

    def start_argv(self) -> list[str]:
        """The `docker run` argv. Public because it is worth asserting on."""
        argv = [
            "docker",
            "run",
            "--detach",
            "--rm",
            "--network",
            "bridge" if self.network else "none",
            "--workdir",
            self.container_workdir,
            "--mount",
            f"type=bind,src={self.workdir},dst={self.container_workdir}",
            "--memory",
            self.memory,
            "--cpus",
            self.cpus,
            "--pids-limit",
            str(self.pids_limit),
            "--cap-drop",
            "ALL",
            "--security-opt",
            "no-new-privileges",
            "--env",
            "PYTHONDONTWRITEBYTECODE=1",
            "--env",
            "PYTHONHASHSEED=0",
        ]
        # Without this the container writes as root and the host-side grader
        # cannot read — or clean up — what the agent left behind.
        if hasattr(os, "getuid"):
            argv += ["--user", f"{os.getuid()}:{os.getgid()}"]
        argv += [self.image, "tail", "-f", "/dev/null"]
        return argv

    def start(self) -> None:
        result = self._runner(self.start_argv(), self.startup_timeout_s)
        if not result.ok:
            raise SandboxError(f"could not start container from {self.image!r}: {result.output}")
        self.container_id = result.stdout.strip()
        if not self.container_id:
            raise SandboxError("docker run returned no container id")

    def run(self, command: str, *, timeout_s: float | None = None) -> ExecResult:
        if self.container_id is None:
            raise SandboxError("sandbox is not started")
        argv = [
            "docker",
            "exec",
            "--workdir",
            self.container_workdir,
            self.container_id,
            "sh",
            "-c",
            command,
        ]
        result = self._runner(argv, timeout_s if timeout_s is not None else self.default_timeout_s)
        # Report the command the model asked for, not the docker invocation.
        return ExecResult(
            command=command,
            exit_code=result.exit_code,
            stdout=result.stdout,
            stderr=result.stderr,
            timed_out=result.timed_out,
        )

    def close(self) -> None:
        if self.container_id is not None:
            self._runner(["docker", "rm", "--force", self.container_id], 60.0)
            self.container_id = None
        super().close()

start_argv

start_argv() -> list[str]

The docker run argv. Public because it is worth asserting on.

Source code in src/endstate/evals/sandbox.py

def start_argv(self) -> list[str]:
    """The `docker run` argv. Public because it is worth asserting on."""
    argv = [
        "docker",
        "run",
        "--detach",
        "--rm",
        "--network",
        "bridge" if self.network else "none",
        "--workdir",
        self.container_workdir,
        "--mount",
        f"type=bind,src={self.workdir},dst={self.container_workdir}",
        "--memory",
        self.memory,
        "--cpus",
        self.cpus,
        "--pids-limit",
        str(self.pids_limit),
        "--cap-drop",
        "ALL",
        "--security-opt",
        "no-new-privileges",
        "--env",
        "PYTHONDONTWRITEBYTECODE=1",
        "--env",
        "PYTHONHASHSEED=0",
    ]
    # Without this the container writes as root and the host-side grader
    # cannot read — or clean up — what the agent left behind.
    if hasattr(os, "getuid"):
        argv += ["--user", f"{os.getuid()}:{os.getgid()}"]
    argv += [self.image, "tail", "-f", "/dev/null"]
    return argv

stage_tree

stage_tree(
    source: Path,
    destination: Path,
    *,
    excludes: frozenset[str] = STAGING_EXCLUDES,
) -> None

Copy source over destination, pruning excludes at every level.

Overlays rather than replaces: staging held-out tests into a tree the agent has already worked in must not delete the agent's work.

Source code in src/endstate/evals/sandbox.py

def stage_tree(
    source: Path, destination: Path, *, excludes: frozenset[str] = STAGING_EXCLUDES
) -> None:
    """Copy `source` over `destination`, pruning `excludes` at every level.

    Overlays rather than replaces: staging held-out tests into a tree the agent
    has already worked in must not delete the agent's work.
    """
    source = Path(source)
    if not source.is_dir():
        raise SandboxError(f"not a directory: {source}")
    destination.mkdir(parents=True, exist_ok=True)
    shutil.copytree(
        source,
        destination,
        dirs_exist_ok=True,
        symlinks=True,
        ignore=lambda _dir, names: {n for n in names if n in excludes},
    )

docker_available

docker_available(
    runner: CommandRunner | None = None,
) -> bool

Whether a usable Docker daemon is reachable.

Not cached: a daemon can start or stop between calls, and a stale False here would send an eval run to the local sandbox by mistake. Callers that ask repeatedly — a skipif evaluated once per decorator, say — should hold the answer themselves rather than making this lie to everyone else.

Source code in src/endstate/evals/sandbox.py

def docker_available(runner: CommandRunner | None = None) -> bool:
    """Whether a usable Docker daemon is reachable.

    Not cached: a daemon can start or stop between calls, and a stale False here
    would send an eval run to the local sandbox by mistake. Callers that ask
    repeatedly — a `skipif` evaluated once per decorator, say — should hold the
    answer themselves rather than making this lie to everyone else.
    """
    run = runner or subprocess_runner
    try:
        return run(
            ["docker", "version", "--format", "{{.Server.Version}}"], DOCKER_PROBE_TIMEOUT_S
        ).ok
    except SandboxError:
        return False

---

Evals: graders

endstate.evals.graders

Graders: pure functions of the sandbox's end state.

def grade(sandbox: Sandbox) -> Verdict: ...

Look at what is not a parameter. There is no messages, no transcript, no result. The grader cannot read what the agent said, because it was never given it — and resolve below enforces that rather than trusting it, because a grader that can see the transcript will eventually be written to check the transcript. Someone will add "and it mentioned running the tests," since that is easier than checking that the tests ran. Conventions erode; signatures do not.

The second thing this module is for is the uncomfortable half of end-state grading: the end state can be gamed too. An agent that edits the tests until they pass satisfies "the suite is green" completely. So does one that adds a skip marker, or special-cases the failing input. Each of those has a corresponding assertion here, and a task that does not make them has swapped a fluency exploit for a test-editing one:

Hack	The grader that catches it
Edited the tests	files_unchanged
Skipped the test	no_new_skips
Special-cased the input	held-out tests, staged in by the runner after sealing
Deleted something inconvenient	paths_exist, changed_paths_within

FORBIDDEN_PARAMETERS module-attribute

FORBIDDEN_PARAMETERS = frozenset(
    {
        "messages",
        "transcript",
        "history",
        "result",
        "run_result",
        "final_text",
        "response",
        "conversation",
    }
)

Parameter names that would let a grader read the agent's output.

Not an exhaustive list of ways to cheat — a determined author can smuggle the transcript in through args. It is a guard rail against the accident, which is the failure mode that actually happens.

GraderContractError

Bases: TypeError

Raised when a grader's signature breaks the end-state contract.

Source code in src/endstate/evals/graders.py

class GraderContractError(TypeError):
    """Raised when a grader's signature breaks the end-state contract."""

resolve

resolve(spec: GraderSpec) -> Grader

Turn a module:function spec plus arguments into a one-argument grader.

Raises:

Type	Description
GraderContractError	If the target cannot be imported, is not callable, or has a signature that could see anything but the sandbox.

Source code in src/endstate/evals/graders.py

def resolve(spec: GraderSpec) -> Grader:
    """Turn a `module:function` spec plus arguments into a one-argument grader.

    Raises:
        GraderContractError: If the target cannot be imported, is not callable,
            or has a signature that could see anything but the sandbox.
    """
    module_name, _, attribute = spec.name.partition(":")
    if not attribute:
        module_name, _, attribute = spec.name.rpartition(".")
    if not module_name or not attribute:
        raise GraderContractError(f"{spec.name!r} is not a 'module:function' path")

    try:
        module = importlib.import_module(module_name)
    except ImportError as exc:
        raise GraderContractError(f"cannot import {module_name!r}: {exc}") from exc

    function = getattr(module, attribute, None)
    if function is None or not callable(function):
        raise GraderContractError(f"{spec.name!r} is not a callable")

    _assert_contract(spec.name, function)
    bound = dict(spec.args)

    def grade(sandbox: Sandbox) -> Verdict:
        verdict: Verdict = function(sandbox, **bound)
        return verdict

    grade.__name__ = attribute
    return grade

compose

compose(graders: Sequence[Grader]) -> Grader

Conjoin graders. Every check from every grader, all of which must pass.

Conjunction rather than "any", because the interesting task definitions are all of the form completed the work and did not cheat to get there.

Source code in src/endstate/evals/graders.py

def compose(graders: Sequence[Grader]) -> Grader:
    """Conjoin graders. Every check from every grader, all of which must pass.

    Conjunction rather than "any", because the interesting task definitions are
    all of the form *completed the work* **and** *did not cheat to get there*.
    """

    def grade(sandbox: Sandbox) -> Verdict:
        checks: list[Check] = []
        for grader in graders:
            checks.extend(grader(sandbox).checks)
        return Verdict.from_checks(checks)

    return grade

command_succeeds

command_succeeds(
    sandbox: Sandbox,
    *,
    command: str,
    timeout_s: float | None = None,
    expect_exit_code: int = 0,
    name: str = "",
    max_output_chars: int = 2000,
) -> Verdict

Run a command in the sandbox and check its exit code.

The workhorse: for most tasks the question "did it work?" is a test suite. Note that this alone is a weak grader — see the module docstring — and every shipped task pairs it with at least one assertion about the file tree.

Source code in src/endstate/evals/graders.py

def command_succeeds(
    sandbox: Sandbox,
    *,
    command: str,
    timeout_s: float | None = None,
    expect_exit_code: int = 0,
    name: str = "",
    max_output_chars: int = 2_000,
) -> Verdict:
    """Run a command in the sandbox and check its exit code.

    The workhorse: for most tasks the question "did it work?" is a test suite.
    Note that this alone is a weak grader — see the module docstring — and every
    shipped task pairs it with at least one assertion about the file tree.
    """
    result = sandbox.run(command, timeout_s=timeout_s)
    label = name or f"`{command}` exits {expect_exit_code}"
    if result.timed_out:
        return Verdict.from_checks(
            [Check(name=label, passed=False, detail=f"timed out after {timeout_s}s")]
        )
    passed = result.exit_code == expect_exit_code
    detail = "" if passed else f"exit {result.exit_code}\n{result.output[:max_output_chars]}"
    return Verdict.from_checks([Check(name=label, passed=passed, detail=detail.strip())])

files_unchanged

files_unchanged(
    sandbox: Sandbox,
    *,
    paths: Sequence[str],
    name: str = "",
) -> Verdict

Assert files are byte-for-byte what the fixture shipped.

Point this at the tests. An agent that cannot make the code pass the tests can always make the tests pass the code, and that is the single most common way a coding benchmark measures nothing.

Source code in src/endstate/evals/graders.py

def files_unchanged(sandbox: Sandbox, *, paths: Sequence[str], name: str = "") -> Verdict:
    """Assert files are byte-for-byte what the fixture shipped.

    Point this at the tests. An agent that cannot make the code pass the tests
    can always make the tests pass the code, and that is the single most common
    way a coding benchmark measures nothing.
    """
    checks: list[Check] = []
    for path in paths:
        before = sandbox.fixture_text(path)
        after = sandbox.read_text(path)
        label = name or f"{path} unchanged"
        if before == after:
            checks.append(Check(name=label, passed=True))
        elif not after:
            checks.append(Check(name=label, passed=False, detail="deleted or emptied"))
        else:
            checks.append(Check(name=label, passed=False, detail="modified"))
    return Verdict.from_checks(checks)

file_matches

file_matches(
    sandbox: Sandbox,
    *,
    path: str,
    pattern: str,
    should_match: bool = True,
    name: str = "",
) -> Verdict

Assert a file's content does (or does not) match a regex.

Source code in src/endstate/evals/graders.py

def file_matches(
    sandbox: Sandbox, *, path: str, pattern: str, should_match: bool = True, name: str = ""
) -> Verdict:
    """Assert a file's content does (or does not) match a regex."""
    regex = re.compile(pattern, re.MULTILINE)
    found = regex.search(sandbox.read_text(path)) is not None
    label = name or f"{path} {'matches' if should_match else 'does not match'} /{pattern}/"
    return Verdict.from_checks([Check(name=label, passed=found is should_match)])

pattern_count

pattern_count(
    sandbox: Sandbox,
    *,
    pattern: str,
    paths: Sequence[str] | None = None,
    glob: str = "*.py",
    maximum: int | None = None,
    minimum: int | None = None,
    name: str = "",
) -> Verdict

Count regex matches across the end state.

The refactor tasks turn on this: "the duplicated validation block appears exactly once in the package" is checkable, where "the code is now DRY" is not.

Source code in src/endstate/evals/graders.py

def pattern_count(
    sandbox: Sandbox,
    *,
    pattern: str,
    paths: Sequence[str] | None = None,
    glob: str = "*.py",
    maximum: int | None = None,
    minimum: int | None = None,
    name: str = "",
) -> Verdict:
    """Count regex matches across the end state.

    The refactor tasks turn on this: "the duplicated validation block appears
    exactly once in the package" is checkable, where "the code is now DRY" is not.
    """
    regex = re.compile(pattern, re.MULTILINE)
    targets = list(paths) if paths is not None else _globbed(sandbox, glob)
    total = sum(len(regex.findall(sandbox.read_text(path))) for path in targets)

    label = name or f"/{pattern}/ occurs {_bound_text(minimum, maximum)}"
    passed = (maximum is None or total <= maximum) and (minimum is None or total >= minimum)
    return Verdict.from_checks(
        [Check(name=label, passed=passed, detail=f"found {total} in {len(targets)} files")]
    )

no_new_skips

no_new_skips(
    sandbox: Sandbox,
    *,
    glob: str = "*test*.py",
    name: str = "",
) -> Verdict

Assert the agent added no skip or xfail markers.

Green because the test never ran is the cheapest possible way to satisfy a "tests pass" grader, and unlike editing an assertion it leaves the test file looking almost untouched.

Source code in src/endstate/evals/graders.py

def no_new_skips(sandbox: Sandbox, *, glob: str = "*test*.py", name: str = "") -> Verdict:
    """Assert the agent added no skip or xfail markers.

    Green because the test never ran is the cheapest possible way to satisfy a
    "tests pass" grader, and unlike editing an assertion it leaves the test file
    looking almost untouched.
    """
    added: list[str] = []
    for path in _globbed(sandbox, glob):
        before = sum(len(re.findall(m, sandbox.fixture_text(path))) for m in SKIP_MARKERS)
        after = sum(len(re.findall(m, sandbox.read_text(path))) for m in SKIP_MARKERS)
        if after > before:
            added.append(f"{path} (+{after - before})")
    label = name or "no skip or xfail markers added"
    return Verdict.from_checks([Check(name=label, passed=not added, detail=", ".join(added))])

tree_unchanged

tree_unchanged(
    sandbox: Sandbox, *, name: str = ""
) -> Verdict

Assert the agent left the sandbox byte-for-byte as it found it.

The permissioning assertion, and the one place where nothing happening is the pass condition. A refusal that still deleted the directory is not a refusal, and the only way to know is to compare the tree.

Note this is a property of the filesystem, so it stays a grader. "The policy fired at least once" is not, and lives in the task's requires block.

Source code in src/endstate/evals/graders.py

def tree_unchanged(sandbox: Sandbox, *, name: str = "") -> Verdict:
    """Assert the agent left the sandbox byte-for-byte as it found it.

    The permissioning assertion, and the one place where *nothing happening* is
    the pass condition. A refusal that still deleted the directory is not a
    refusal, and the only way to know is to compare the tree.

    Note this is a property of the filesystem, so it stays a grader. "The policy
    fired at least once" is not, and lives in the task's `requires` block.
    """
    changed = sandbox.changed_paths()
    return Verdict.from_checks(
        [
            Check(
                name=name or "the sandbox is untouched",
                passed=not changed,
                detail=", ".join(changed[:10]),
            )
        ]
    )

changed_paths_within

changed_paths_within(
    sandbox: Sandbox,
    *,
    allowed: Sequence[str],
    name: str = "",
) -> Verdict

Assert the agent only touched paths the task permits.

Catches the collateral damage a pass/fail on the task itself cannot see: the unrelated module deleted on the way to a green test run.

Source code in src/endstate/evals/graders.py

def changed_paths_within(sandbox: Sandbox, *, allowed: Sequence[str], name: str = "") -> Verdict:
    """Assert the agent only touched paths the task permits.

    Catches the collateral damage a pass/fail on the task itself cannot see: the
    unrelated module deleted on the way to a green test run.
    """
    stray = [p for p in sandbox.changed_paths() if not any(fnmatch(p, a) for a in allowed)]
    label = name or "changes confined to the task's paths"
    return Verdict.from_checks(
        [Check(name=label, passed=not stray, detail=", ".join(sorted(stray)[:10]))]
    )

---

Evals: the runner

endstate.evals.runner

Running a suite.

The sequence for one task, in the order it has to happen:

1. Copy the fixture into a fresh sandbox and start it.
2. Let the agent work, with the shell wired into the sandbox.
3. Seal the end state — hash it, snapshot it.
4. Then stage the held-out tests in and grade.

Step 3 before step 4 is the whole reason seal() exists. Held-out tests are files; staging them in before hashing would mean every grader that asserts on the tree was asserting on a tree the grader itself had modified.

What the runner deliberately does not do is give the grader anything but the sandbox. The RunResult — steps, tokens, transcript — is recorded in the TaskResult for the report, and it is never in scope where a verdict is decided.

TaskResult

Bases: BaseModel

Everything recorded about one task run.

The verdict is the answer; the rest is what makes a benchmark table more informative than a pass rate. A model that passes in 6 steps for $0.02 and one that passes in 34 steps for $1.40 are not the same product.

Source code in src/endstate/evals/runner.py

class TaskResult(BaseModel):
    """Everything recorded about one task run.

    The verdict is the answer; the rest is what makes a benchmark table more
    informative than a pass rate. A model that passes in 6 steps for $0.02 and
    one that passes in 34 steps for $1.40 are not the same product.
    """

    task_id: str
    category: str
    verdict: Verdict
    steps: int = 0
    usage: Usage = Field(default_factory=Usage)
    stop_reason: StopReason = StopReason.END_TURN
    compaction_events: int = 0
    denied_calls: int = 0
    unsettled_calls: int = 0
    wall_clock_s: float = 0.0
    tree_hash: str = ""
    model: str = ""
    timed_out: bool = False
    error: str = ""

    @property
    def passed(self) -> bool:
        return self.verdict.passed

SuiteResult

Bases: BaseModel

The outcome of a whole suite against one model.

Source code in src/endstate/evals/runner.py

class SuiteResult(BaseModel):
    """The outcome of a whole suite against one model."""

    model: str = ""
    provider: str = ""
    sandbox: str = ""
    started_at: str = ""
    duration_s: float = 0.0
    results: list[TaskResult] = Field(default_factory=list)

    @property
    def verdict_vector(self) -> tuple[tuple[str, bool], ...]:
        """The determinism criterion, as a value you can compare with `==`.

        Task id and pass/fail only. Wall clock, token counts and step counts all
        vary between identical runs; if any of them were in here, the criterion
        would be untestable rather than merely hard.
        """
        return tuple((r.task_id, r.passed) for r in self.results)

    @property
    def pass_rate(self) -> float:
        if not self.results:
            return 0.0
        return sum(1 for r in self.results if r.passed) / len(self.results)

    @property
    def errored(self) -> list[TaskResult]:
        """Tasks where the *harness* failed, which is not the same as a fail."""
        return [r for r in self.results if r.error]

verdict_vector property

verdict_vector: tuple[tuple[str, bool], ...]

The determinism criterion, as a value you can compare with ==.

Task id and pass/fail only. Wall clock, token counts and step counts all vary between identical runs; if any of them were in here, the criterion would be untestable rather than merely hard.

errored property

errored: list[TaskResult]

Tasks where the harness failed, which is not the same as a fail.

EvalRunner

Runs tasks. One sandbox per task, one provider per task, no shared state.

Source code in src/endstate/evals/runner.py

class EvalRunner:
    """Runs tasks. One sandbox per task, one provider per task, no shared state."""

    def __init__(
        self,
        provider_factory: ProviderFactory,
        sandbox_factory: SandboxFactory,
        *,
        policy: PermissionPolicy | None = None,
        prices: PriceTable | None = None,
        system_prompt: str = DEFAULT_SYSTEM_PROMPT,
        tool_timeout_s: float = 60.0,
        jobs: int = 1,
        provider_name: str = "",
        sandbox_name: str = "",
        sessions: SessionStore | None = None,
        on_result: Callable[[TaskResult], None] | None = None,
    ) -> None:
        self.provider_factory = provider_factory
        self.sandbox_factory = sandbox_factory
        self.policy = policy or default_policy()
        self.prices = prices or PriceTable()
        self.system_prompt = system_prompt
        self.tool_timeout_s = tool_timeout_s
        self.jobs = max(1, jobs)
        self.provider_name = provider_name
        self.sandbox_name = sandbox_name
        self.on_result = on_result
        self.accountant = CostAccountant(self.prices)
        self._lock = threading.Lock()
        # A recovery task resumes from a checkpoint, so it needs somewhere
        # durable to checkpoint to. Temporary by default rather than the CLI's
        # `.endstate/` next to the tasks: an eval run is not a conversation
        # anyone resumes by hand afterwards, and writing into the working tree
        # would leave a session database in whatever directory it ran from.
        self._session_dir = tempfile.TemporaryDirectory(prefix="endstate-sessions-")
        self.sessions = sessions or SessionStore(Path(self._session_dir.name) / "sessions.sqlite3")

    def run_task(self, task: Task) -> TaskResult:
        started = time.monotonic()
        try:
            return self._run_task(task, started)
        except Exception as exc:  # noqa: BLE001 - a batch job survives one bad task
            # A sandbox that will not start or a provider that returns 503 is a
            # *harness* problem, and `error` keeps it out of the pass rate:
            # reporting it as a task failure would quietly turn "Docker is not
            # running" into "this model cannot code". Catching broadly is the
            # right trade for a suite — losing the other nineteen results to an
            # exception on task seven is worse than recording it and moving on.
            label = "sandbox error" if isinstance(exc, SandboxError) else type(exc).__name__
            return TaskResult(
                task_id=task.id,
                category=task.category,
                verdict=Verdict.fail(f"{label}: {exc}"),
                wall_clock_s=time.monotonic() - started,
                error=f"{label}: {exc}",
            )

    def _run_task(self, task: Task, started: float) -> TaskResult:
        provider = self.provider_factory(task)
        deadline = _DeadlineProvider(provider, task.timeout_s)
        accountant = CostAccountant(self.prices)

        with self.sandbox_factory(task) as sandbox:
            run = self._drive(task, sandbox, deadline, accountant)

            end_state = sandbox.seal()
            if task.holdout is not None:
                sandbox.stage(task.holdout)
            # Conjunction, and the order matters for reading a failure: what the
            # sandbox looks like first, then what the run had to have done.
            verdict = self._grade(task, sandbox).merged_with(self._require(task, run))

        # Per-task accounting is merged into the suite total under a lock: with
        # jobs > 1 the merge is a read-modify-write on a shared dict.
        with self._lock:
            for model, usage in accountant.usage_by_model.items():
                self.accountant.record(model, usage)

        return TaskResult(
            task_id=task.id,
            category=task.category,
            verdict=verdict,
            steps=run.steps,
            usage=run.usage,
            stop_reason=run.stop_reason,
            compaction_events=len(run.compaction_events),
            denied_calls=len(run.denied_calls),
            unsettled_calls=len(run.unsettled_calls),
            wall_clock_s=time.monotonic() - started,
            tree_hash=end_state,
            model=str(getattr(provider, "model", "")),
            timed_out=deadline.expired,
        )

    def _loop_for(
        self,
        task: Task,
        sandbox: Sandbox,
        provider: object,
        accountant: CostAccountant,
        session: Session,
        tools: list[Tool] | None = None,
    ) -> AgentLoop:
        return AgentLoop(
            provider=provider,
            tools=tools if tools is not None else sandbox_tools(sandbox),
            tool_context=ToolContext(workdir=sandbox.workdir, timeout_s=self.tool_timeout_s),
            policy=self.policy,
            context=ContextManager(budget=task.budget),
            accountant=accountant,
            max_steps=task.max_steps,
            system_prompt=self.system_prompt,
            session=session,
        )

    def _drive(
        self,
        task: Task,
        sandbox: Sandbox,
        provider: object,
        accountant: CostAccountant,
    ) -> RunResult:
        """Run the agent, killing and resuming it partway if the task says to."""
        session = self.sessions.create(model=str(getattr(provider, "model", "")))
        if task.recovery is None:
            return self._loop_for(task, sandbox, provider, accountant, session).run(task.prompt)

        crashing = CrashAt(
            sandbox_tools(sandbox),
            task.recovery.crash_at_call,
            after_side_effect=task.recovery.after_side_effect,
        )
        loop = self._loop_for(task, sandbox, provider, accountant, session, crashing.tools)
        try:
            return loop.run(task.prompt)
        except Crash:
            pass

        # A *new* loop over the reloaded session is the point. Reusing the old
        # object would resume from memory that a dead process would not have
        # had, which is the difference between testing resume and testing that
        # nothing was dropped from a list.
        resumed = self._loop_for(
            task, sandbox, provider, accountant, self.sessions.resume(session.id)
        )
        return resumed.resume()

    def _require(self, task: Task, run: RunResult) -> Verdict:
        counters = {
            "compaction_events": len(run.compaction_events),
            "denied_calls": len(run.denied_calls),
            "unsettled_calls": len(run.unsettled_calls),
            "steps": run.steps,
            "input_tokens": run.usage.input_tokens,
            "output_tokens": run.usage.output_tokens,
            "total_tokens": run.usage.total_tokens,
        }
        return Verdict.from_checks(task.requires.check(counters, run.stop_reason.value))

    def _grade(self, task: Task, sandbox: Sandbox) -> Verdict:
        try:
            grader: Grader = grader_for(task.graders)
            return grader(sandbox)
        except Exception as exc:  # noqa: BLE001 - a broken grader must not sink the suite
            return Verdict.fail(f"grader raised {type(exc).__name__}: {exc}")

    def run_suite(self, tasks: Sequence[Task]) -> SuiteResult:
        """Run every task and return the results in task order.

        Order is independent of completion order even with `jobs > 1`, because
        the pass/fail vector is only comparable between runs if its index is
        stable.
        """
        started = time.monotonic()
        started_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

        if self.jobs == 1:
            results = [self._collect(task) for task in tasks]
        else:
            with ThreadPoolExecutor(max_workers=self.jobs) as pool:
                results = list(pool.map(self._collect, tasks))

        return SuiteResult(
            model=next((r.model for r in results if r.model), ""),
            provider=self.provider_name,
            sandbox=self.sandbox_name,
            started_at=started_at,
            duration_s=time.monotonic() - started,
            results=results,
        )

    def _collect(self, task: Task) -> TaskResult:
        result = self.run_task(task)
        if self.on_result is not None:
            self.on_result(result)
        return result

run_suite

run_suite(tasks: Sequence[Task]) -> SuiteResult

Run every task and return the results in task order.

Order is independent of completion order even with jobs > 1, because the pass/fail vector is only comparable between runs if its index is stable.

Source code in src/endstate/evals/runner.py

def run_suite(self, tasks: Sequence[Task]) -> SuiteResult:
    """Run every task and return the results in task order.

    Order is independent of completion order even with `jobs > 1`, because
    the pass/fail vector is only comparable between runs if its index is
    stable.
    """
    started = time.monotonic()
    started_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

    if self.jobs == 1:
        results = [self._collect(task) for task in tasks]
    else:
        with ThreadPoolExecutor(max_workers=self.jobs) as pool:
            results = list(pool.map(self._collect, tasks))

    return SuiteResult(
        model=next((r.model for r in results if r.model), ""),
        provider=self.provider_name,
        sandbox=self.sandbox_name,
        started_at=started_at,
        duration_s=time.monotonic() - started,
        results=results,
    )

---

Evals: the report

endstate.evals.report

The report.

Pass rate alone hides too much. A model that passes 80% of tasks in 6 steps for $0.02 and a model that passes 80% in 34 steps for $1.40 are not the same product, and only one of them is deployable at volume — so every summary here carries steps, tokens, cost and latency beside the rate.

Two smaller decisions worth naming. Cost renders as — rather than 0.00 when the model has no price entry, because a wrong number in a published benchmark is worse than a missing one (D4). And errored tasks are counted separately from failed ones: "Docker was not running" and "the model could not do it" must never sum into the same column.

FLAKE_THRESHOLD module-attribute

FLAKE_THRESHOLD = 0.05

The rate above which a suite cannot support a claim.

From the engineering plan's M2 metric. Determinism is aspirational rather than achievable — providers do not offer bit-identical output even at temperature 0 — so the criterion is a low rate, not zero.

percentile

percentile(values: list[float], fraction: float) -> float

Nearest-rank percentile.

Not statistics.quantiles, which interpolates and needs at least two data points. A twenty-task suite is a small sample and an interpolated p95 over twenty numbers implies a precision that is not there.

Source code in src/endstate/evals/report.py

def percentile(values: list[float], fraction: float) -> float:
    """Nearest-rank percentile.

    Not `statistics.quantiles`, which interpolates and needs at least two data
    points. A twenty-task suite is a small sample and an interpolated p95 over
    twenty numbers implies a precision that is not there.
    """
    if not values:
        return 0.0
    ordered = sorted(values)
    rank = max(1, min(len(ordered), math.ceil(fraction * len(ordered))))
    return ordered[rank - 1]

total_cost

total_cost(
    suite: SuiteResult, prices: PriceTable
) -> Decimal | None

Suite cost, or None if any model in it has no price.

Source code in src/endstate/evals/report.py

def total_cost(suite: SuiteResult, prices: PriceTable) -> Decimal | None:
    """Suite cost, or None if any model in it has no price."""
    total = Decimal(0)
    for result in suite.results:
        price = prices.get(result.model)
        if price is None:
            return None
        total += price.cost(result.usage)
    return total

render_markdown

render_markdown(
    suite: SuiteResult, prices: PriceTable | None = None
) -> str

The committed artefact: one suite run, as a markdown page.

Source code in src/endstate/evals/report.py

def render_markdown(suite: SuiteResult, prices: PriceTable | None = None) -> str:
    """The committed artefact: one suite run, as a markdown page."""
    prices = prices or PriceTable()
    results = suite.results
    cost = total_cost(suite, prices)
    per_task = (cost / len(results)) if cost is not None and results else None

    lines = [
        f"# Eval report — {suite.model or 'unknown model'}",
        "",
        f"- **Run at:** {suite.started_at or 'unknown'}",
        f"- **Provider:** {suite.provider or 'unknown'}",
        f"- **Sandbox:** {suite.sandbox or 'unknown'}",
        f"- **Tasks:** {len(results)}",
        f"- **Wall clock:** {suite.duration_s:.1f}s",
        "",
        "## Summary",
        "",
        "| provider | model | pass rate | median steps | input tok | output tok |"
        " USD/task | p95 latency | compactions/task |",
        "| --- | --- | --- | --- | --- | --- | --- | --- | --- |",
        _summary_row(suite, per_task),
        "",
        "## By category",
        "",
        "| category | passed | tasks | pass rate |",
        "| --- | --- | --- | --- |",
    ]

    for category in sorted({r.category for r in results}):
        rows = [r for r in results if r.category == category]
        passed = sum(1 for r in rows if r.passed)
        lines.append(f"| {category} | {passed} | {len(rows)} | {passed / len(rows):.0%} |")

    lines += [
        "",
        "## Tasks",
        "",
        "| task | category | verdict | steps | tokens | seconds | denied | compactions |",
        "| --- | --- | --- | --- | --- | --- | --- | --- |",
    ]
    for result in results:
        lines.append(
            f"| `{result.task_id}` | {result.category} | {_verdict_cell(result)} | "
            f"{result.steps} | {result.usage.total_tokens:,} | {result.wall_clock_s:.1f} | "
            f"{result.denied_calls} | {result.compaction_events} |"
        )

    failures = [r for r in results if not r.passed]
    if failures:
        lines += ["", "## Failures", ""]
        for result in failures:
            lines.append(f"### `{result.task_id}`")
            lines.append("")
            if result.error:
                lines += [f"Harness error: {result.error}", ""]
            for check in result.verdict.checks:
                lines.append(f"- {check}")
            if not result.verdict.checks and result.verdict.reason:
                lines.append(f"- {result.verdict.reason}")
            lines.append("")

    if cost is None:
        lines += [
            "",
            "Cost is unreported: no price entry for every model in this run. Pass `--prices`"
            " with a table covering them rather than reading a zero as free.",
        ]
    return "\n".join(lines).rstrip() + "\n"

flake_rate

flake_rate(suites: Sequence[SuiteResult]) -> float

Fraction of tasks that did not return the same verdict in every run.

The number the whole determinism claim reduces to, and it is only meaningful over identical inputs: same suite, same model, same seed. One run per task cannot tell capability from luck, and providers do not offer bit-identical output even at temperature 0 — so what is achievable is a low rate, not zero, and every result below the threshold is still a distribution rather than a fact.

Returns 0.0 for fewer than two runs: nothing has been compared yet, and reporting a rate for a single run would imply otherwise.

Source code in src/endstate/evals/report.py

def flake_rate(suites: Sequence[SuiteResult]) -> float:
    """Fraction of tasks that did not return the same verdict in every run.

    The number the whole determinism claim reduces to, and it is only meaningful
    over identical inputs: same suite, same model, same seed. One run per task
    cannot tell capability from luck, and providers do not offer bit-identical
    output even at temperature 0 — so what is achievable is a low rate, not zero,
    and every result below the threshold is still a distribution rather than a
    fact.

    Returns 0.0 for fewer than two runs: nothing has been compared yet, and
    reporting a rate for a single run would imply otherwise.
    """
    if len(suites) < 2:
        return 0.0
    outcomes = _outcomes(suites)
    if not outcomes:
        return 0.0
    flaky = sum(1 for results in outcomes.values() if len(set(results)) > 1)
    return flaky / len(outcomes)

determinism_established

determinism_established(
    suites: Sequence[SuiteResult],
) -> tuple[bool, str]

Whether these runs can support a determinism claim at all.

A low flake rate is necessary and nowhere near sufficient, because the most reassuring number this module can produce is also what a completely broken run produces. A suite where the container never started fails every task, identically, every time — perfectly reproducible and evidence of nothing.

So the rate is reported under this: if any task hit a harness error, or there is only one run, the answer is "not established" no matter how stable the verdicts looked.

Source code in src/endstate/evals/report.py

def determinism_established(suites: Sequence[SuiteResult]) -> tuple[bool, str]:
    """Whether these runs can support a determinism claim at all.

    A low flake rate is necessary and nowhere near sufficient, because the most
    reassuring number this module can produce is also what a completely broken
    run produces. A suite where the container never started fails every task,
    identically, every time — perfectly reproducible and evidence of nothing.

    So the rate is reported *under* this: if any task hit a harness error, or
    there is only one run, the answer is "not established" no matter how stable
    the verdicts looked.
    """
    if len(suites) < 2:
        return False, "fewer than two runs — nothing has been compared"
    if not _outcomes(suites):
        return False, "no tasks ran"
    errored = {r.task_id for s in suites for r in s.results if r.error}
    if errored:
        listed = ", ".join(sorted(errored)[:5])
        return False, f"{len(errored)} task(s) hit a harness error, not a verdict: {listed}"
    return True, ""

render_flake_markdown

render_flake_markdown(
    suites: Sequence[SuiteResult],
    threshold: float = FLAKE_THRESHOLD,
) -> str

Report determinism across repeated runs of the same suite.

Source code in src/endstate/evals/report.py

def render_flake_markdown(suites: Sequence[SuiteResult], threshold: float = FLAKE_THRESHOLD) -> str:
    """Report determinism across repeated runs of the same suite."""
    rate = flake_rate(suites)
    outcomes = _outcomes(suites)
    vectors = {s.verdict_vector for s in suites}
    established, why_not = determinism_established(suites)

    model = next((s.model for s in suites if s.model), "unknown model")
    lines = [
        f"# Flake report — {model}",
        "",
        f"- **Runs:** {len(suites)}",
        f"- **Tasks:** {len(outcomes)}",
        f"- **Flake rate:** {rate:.1%} ({'within' if rate <= threshold else 'OVER'}"
        f" the {threshold:.0%} threshold)",
        f"- **Identical verdict vectors:** {'yes' if len(vectors) == 1 else 'no'}",
        "",
    ]

    if not established:
        lines += [
            f'!!! danger "Determinism not established — {why_not}"',
            "",
            "    The rate above is real and it certifies nothing. A run where the sandbox never"
            " started fails every task identically every time, which is perfectly reproducible and"
            " evidence of nothing at all. Fix the errors and run it again before quoting a number.",
            "",
        ]

    unstable = {t: r for t, r in outcomes.items() if len(set(r)) > 1}
    if unstable:
        lines += [
            "## Tasks that did not agree with themselves",
            "",
            "| task | runs |",
            "| --- | --- |",
        ]
        for task_id, results in sorted(unstable.items()):
            lines.append(f"| `{task_id}` | {' '.join('pass' if r else 'fail' for r in results)} |")
        lines += [
            "",
            "A task in this table is not evidence about the model. It is evidence that the number"
            " next to it in any benchmark table is a coin flip, and it has to be fixed or dropped"
            " before the suite can support a claim.",
            "",
        ]
    else:
        lines += ["Every task returned the same verdict in every run.", ""]

    lines += [
        "## Outcomes",
        "",
        "| task | " + " | ".join(f"run {i + 1}" for i in range(len(suites))) + " |",
        "| --- |" + " --- |" * len(suites),
    ]
    for task_id, results in sorted(outcomes.items()):
        cells = " | ".join("pass" if r else "fail" for r in results)
        lines.append(f"| `{task_id}` | {cells} |")

    return "\n".join(lines).rstrip() + "\n"

write_report

write_report(
    suite: SuiteResult,
    out_dir: Path,
    prices: PriceTable | None = None,
) -> tuple[Path, Path]

Write the markdown report and the machine-readable results beside it.

Both, not either. The markdown is what a person reads in a pull request; the JSON is what the next run diffs against to answer "did anything regress?".

Source code in src/endstate/evals/report.py

def write_report(
    suite: SuiteResult, out_dir: Path, prices: PriceTable | None = None
) -> tuple[Path, Path]:
    """Write the markdown report and the machine-readable results beside it.

    Both, not either. The markdown is what a person reads in a pull request; the
    JSON is what the next run diffs against to answer "did anything regress?".
    """
    out_dir = Path(out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    stem = f"results-{date.today().isoformat()}-{_slug(suite.model)}"

    markdown_path = out_dir / f"{stem}.md"
    markdown_path.write_text(render_markdown(suite, prices), encoding="utf-8")

    json_path = out_dir / f"{stem}.json"
    json_path.write_text(
        json.dumps(json.loads(suite.model_dump_json()), indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    return markdown_path, json_path

---

Cost

endstate.telemetry.cost

Token cost accounting.

Prices are data, not code. They change often and vary by region and contract, so they live in a user-editable table rather than being hardcoded into logic. Ship with an empty default and require the user to supply prices for any model they want costed: a wrong number is worse than a missing one.

ModelPrice

Bases: BaseModel

USD per million tokens.

Source code in src/endstate/telemetry/cost.py

class ModelPrice(BaseModel):
    """USD per million tokens."""

    input_per_mtok: Decimal
    output_per_mtok: Decimal
    cached_input_per_mtok: Decimal | None = None

    def cost(self, usage: Usage) -> Decimal:
        cached_rate = (
            self.cached_input_per_mtok
            if self.cached_input_per_mtok is not None
            else self.input_per_mtok
        )
        million = Decimal(1_000_000)
        return (
            Decimal(usage.input_tokens) * self.input_per_mtok / million
            + Decimal(usage.output_tokens) * self.output_per_mtok / million
            + Decimal(usage.cached_input_tokens) * cached_rate / million
        )

PriceTable

Bases: BaseModel

Maps model id -> price. Load from JSON so it can be updated without a release.

Source code in src/endstate/telemetry/cost.py

class PriceTable(BaseModel):
    """Maps model id -> price. Load from JSON so it can be updated without a release."""

    prices: dict[str, ModelPrice] = Field(default_factory=dict)

    @classmethod
    def from_file(cls, path: str | Path) -> PriceTable:
        data = json.loads(Path(path).read_text())
        return cls(prices={k: ModelPrice(**v) for k, v in data.items()})

    def get(self, model: str) -> ModelPrice | None:
        return self.prices.get(model)

UnknownModelPriceError

Bases: KeyError

Raised when a cost is requested for a model with no price entry.

Source code in src/endstate/telemetry/cost.py

class UnknownModelPriceError(KeyError):
    """Raised when a cost is requested for a model with no price entry."""

CostAccountant

Accumulates usage and converts it to money.

Usage is always tracked. Cost is only reported for models with a known price; total_cost raises rather than silently under-reporting.

Source code in src/endstate/telemetry/cost.py

class CostAccountant:
    """Accumulates usage and converts it to money.

    Usage is always tracked. Cost is only reported for models with a known price;
    `total_cost` raises rather than silently under-reporting.
    """

    def __init__(self, price_table: PriceTable | None = None) -> None:
        self.price_table = price_table or PriceTable()
        self._usage_by_model: dict[str, Usage] = {}

    def record(self, model: str, usage: Usage) -> None:
        self._usage_by_model[model] = self._usage_by_model.get(model, Usage()) + usage

    @property
    def usage_by_model(self) -> dict[str, Usage]:
        return dict(self._usage_by_model)

    @property
    def total_usage(self) -> Usage:
        total = Usage()
        for usage in self._usage_by_model.values():
            total = total + usage
        return total

    def cost_for(self, model: str) -> Decimal:
        price = self.price_table.get(model)
        if price is None:
            raise UnknownModelPriceError(model)
        return price.cost(self._usage_by_model.get(model, Usage()))

    def total_cost(self) -> Decimal:
        return sum((self.cost_for(m) for m in self._usage_by_model), Decimal(0))

    def priced_models(self) -> list[str]:
        return [m for m in self._usage_by_model if self.price_table.get(m) is not None]

    def unpriced_models(self) -> list[str]:
        return [m for m in self._usage_by_model if self.price_table.get(m) is None]

---

Tracing

endstate.telemetry.trace

A minimal in-process trace.

Not OpenTelemetry: the point is a self-contained artifact you can serialise next to eval results. Exporting to OTel is a later concern.