Skip to content

blockether.vis.evals

Evaluation tools: write and run Harbor tasks, read the results and compare them.

Write a Vis task with write_task, and run tasks with the Vis agent with run_tasks. Give write_task a list of user messages for a task with several turns. Read one Vis run with load_trajectory or parse_trajectory, and join the turns of one session with merge_trajectories. Read the cases of a Harbor dataset with read_tasks, and the trials of a job or round with read_trials. Count the trials with case_results or summarize. Seal a finished round with seal_round. Compare two results with compare_cases, or test paired changes with paired_change and mcnemar_exact. Score repeated attempts with task_attempts and repeat_stats. Summarize a round from its Harbor job folders with round_summary, then score and diagnose it with bench_report and diagnose. Keep human cause labels with save_label. Place a round on a Harbor Hub board with read_leaderboard. Check automatic causes against human labels with agreement, and judge stages with stage_agreement. Estimate pass@k and pass^k with pass_at_k and pass_all_k, and the agreement of several raters with krippendorff_alpha. Build all of it for one round with round_report, or run python -m blockether.vis.evals.report. For installation and sessions, use the Python SDK guide.

ALPHA_RELIABLE = 0.8
ALPHA_TENTATIVE = 0.667
CAUSES = ('dead_tools', 'agent_errors', 'agent_timeouts', 'early_stops', 'unscored', 'other')
EARLY_STOP_SHARE = 0.25
KAPPA_REVIEW = 0.6
KAPPA_TRUST = 0.8
LEADERBOARD_URL = 'https://api.harborframework.com/functions/v1/leaderboard-read'
MIN_LABELS = 30
NEAR_MISS_RATIO = 0.8
NOTE_LIMIT = 500
REPORT_FILE = 'vis-report.json'
STAGES = ('understand', 'write', 'verify', 'other')
@dataclass(frozen=True)
class Agreement:

How well automatic results agree with human labels of the same items.

labelled counts the paired items. agreement_percent is the share of equal pairs, or None without pairs. kappa is Cohen's kappa and alpha is Krippendorff's alpha of the same pairs. Each is rounded to 3 places, or None when it is undefined. disagreements names each item where the two results differ. advice says if you can trust the automatic results.

Agreement( labelled: int, agreement_percent: float | None, kappa: float | None, alpha: float | None, disagreements: tuple[str, ...], advice: str)
labelled: int
agreement_percent: float | None
kappa: float | None
alpha: float | None
disagreements: tuple[str, ...]
advice: str
@dataclass(frozen=True)
class BenchReport:

The benchmark score of one round, with its usage.

pass_rate_percent divides solved tasks by scored tasks, by the latest scored attempt of each task. strict_pass_rate_percent divides them by all dataset tasks, so unscored, unattempted and GPU tasks count as failures. The attempt pass rate counts every scored attempt, with the attempts of one task as one cluster. Each _ci95 value is a 95% Wilson interval in percent. A second round of the same size cannot detect a change below detectable_change_points. attempts_per_task is the most common scored attempt count of a task. When it is more than one, pass_at_k_percent and pass_all_k_percent give the unbiased pass@k and pass^k with that count as k: the chance that at least one or all of k attempts solve a task. Each task uses all its scored attempts. Otherwise they are None. unstable_tasks lists tasks with mixed results. estimated_metered_cost_usd and cost_per_solved_usd use metered API prices. They are not subscription charges.

BenchReport( dataset: str, round_name: str | None, model: str, total_tasks: int, scored_tasks: int, solved_tasks: int, pass_rate_percent: float, pass_rate_ci95: tuple[float, float] | None, strict_pass_rate_percent: float, strict_pass_rate_ci95: tuple[float, float] | None, attempts_per_task: int, scored_attempts: int, solved_attempts: int, attempt_pass_rate_percent: float, attempt_pass_rate_ci95: tuple[float, float] | None, detectable_change_points: float | None, pass_at_k_percent: float | None, pass_all_k_percent: float | None, unstable_tasks: tuple[str, ...], outcomes: tuple[OutcomeCount, ...], unscored_tasks: tuple[str, ...], gpu_tasks: tuple[str, ...], unattempted_tasks: int, attempts: int, model_attempts: int, agent_hours: float, input_tokens: int, cached_input_percent: float, output_tokens: int, estimated_metered_cost_usd: float, cost_per_solved_usd: float | None, repeated_final_errors: tuple[str, ...])
dataset: str
round_name: str | None
model: str
total_tasks: int
scored_tasks: int
solved_tasks: int
pass_rate_percent: float
pass_rate_ci95: tuple[float, float] | None
strict_pass_rate_percent: float
strict_pass_rate_ci95: tuple[float, float] | None
attempts_per_task: int
scored_attempts: int
solved_attempts: int
attempt_pass_rate_percent: float
attempt_pass_rate_ci95: tuple[float, float] | None
detectable_change_points: float | None
pass_at_k_percent: float | None
pass_all_k_percent: float | None
unstable_tasks: tuple[str, ...]
outcomes: tuple[OutcomeCount, ...]
unscored_tasks: tuple[str, ...]
gpu_tasks: tuple[str, ...]
unattempted_tasks: int
attempts: int
model_attempts: int
agent_hours: float
input_tokens: int
cached_input_percent: float
output_tokens: int
estimated_metered_cost_usd: float
cost_per_solved_usd: float | None
repeated_final_errors: tuple[str, ...]
@dataclass(frozen=True)
class CaseChange:

One case whose pass rate changed between the two sides, as passed/runs.

CaseChange(case: str, baseline: str, candidate: str)
case: str
baseline: str
candidate: str
@dataclass(frozen=True)
class CaseComparison:

A paired comparison of the cases that both sides ran.

The rates are the mean pass rates of the paired cases, in percent. difference_ci95 is the Agresti-Min 95% interval of the paired difference, in points. verdict is better or worse only when that interval excludes zero. Without margin_points, every other result is no measurable difference. With it, the position of the interval against the margin gives the verdict:

  • equivalent: the interval is inside the margin on both sides.
  • not worse: the interval is above the negative margin.
  • not better: the interval is below the margin, but reaches the negative margin.
  • inconclusive: the interval reaches both sides of the margin.

The verdict uses the 95% interval before rounding. With a margin, a better or worse verdict gets a caveat when the change is, or can be, smaller than the margin.

At this case count, a true change of detectable_change_points or more shows with about 80% power, by a normal approximation. p_value is the two-sided exact McNemar p-value of the changed cases. With repeats, it is the exact sign test of the case changes. routes holds this test for each route. worse and better list the changed cases, largest change first. unpaired counts cases on one side only; the comparison leaves them out. Caveats name unpaired cases, different routes, a change smaller than the margin and a verdict that the exact test does not confirm.

CaseComparison( paired_cases: int, baseline_rate_percent: float, candidate_rate_percent: float, difference_points: float, difference_ci95: tuple[float, float], verdict: str, detectable_change_points: float, p_value: float, routes: tuple[RouteTest, ...], worse: tuple[CaseChange, ...], better: tuple[CaseChange, ...], unpaired: int, caveats: tuple[str, ...], margin_points: float | None = None)
paired_cases: int
baseline_rate_percent: float
candidate_rate_percent: float
difference_points: float
difference_ci95: tuple[float, float]
verdict: str
detectable_change_points: float
p_value: float
routes: tuple[RouteTest, ...]
worse: tuple[CaseChange, ...]
better: tuple[CaseChange, ...]
unpaired: int
caveats: tuple[str, ...]
margin_points: float | None = None
@dataclass(frozen=True)
class Diagnosis:

Where the benchmark loses tasks, largest lever first.

estimate projects the effect of full attempts on agent-error tasks. It is an estimate, not a measurement. label_check compares the causes with human labels.

Diagnosis( scored_tasks: int, solved_tasks: int, findings: tuple[Finding, ...], estimate: str, label_check: Agreement)
scored_tasks: int
solved_tasks: int
findings: tuple[Finding, ...]
estimate: str
label_check: Agreement
@dataclass(frozen=True)
class Finding:

One problem class. share is the fraction of scored tasks it affects.

Finding( kind: str, title: str, task_count: int, share: float, detail: str, tasks: tuple[TaskNote, ...])
kind: str
title: str
task_count: int
share: float
detail: str
tasks: tuple[TaskNote, ...]
@dataclass(frozen=True, slots=True)
class FunctionCall:

One host function call in a code block: one Activity row and its place in the run.

operation names the function. arguments, result and error are the text evidence that the engine observed, or None. state is the outcome. A linked call, such as a shell poll, has the id of its parent row in parent_id. step_metrics has the tokens and the cost of the whole step, not of this call.

FunctionCall( row: blockether.vis.activity.ActivityRow, step_id: int, iteration: int | None, tool_call_id: str, parent_id: str | None, depth: int, step_metrics: StepMetrics | None)
step_id: int
iteration: int | None
tool_call_id: str
parent_id: str | None
depth: int
step_metrics: StepMetrics | None
id: str

The Activity row id. It is unique in its code block.

sequence: int

The order of the call in its code block.

operation: str

The name of the function, for example grep or shell.

state: str

The outcome: idle, running, succeeded, failed or cancelled.

arguments: str | None

The arguments as the engine showed them, or None.

result: str | None

The result as the engine showed it, or None.

error: str | None

The error text of a failed call, or None.

summary: str

The display summary. Use state for the outcome.

result_summary: str | None

A short result text, or None.

error_summary: str | None

A short error text, or None.

duration_ms: int | None

The run time of the call in milliseconds, or None.

presenter: str

The Activity presenter of the call.

signal: str

The kind of effect: observation, mutation, verification, external or generic.

argument_key: str | None

A key that is equal for calls with identical arguments, or None.

def to_dict(self) -> dict[str, typing.Any]:

Return the call as plain JSON data.

row has the other Activity fields, without children: each child is a separate FunctionCall.

@dataclass(frozen=True, slots=True)
class HarborTask:

One Harbor task, read from its task.toml and instruction.md.

name is the task folder name, as in case_results. instruction is the prompt that the agent gets. The timeouts are in seconds, or None when task.toml gives none. gpus is the GPU count that the environment needs. metadata holds the [metadata] table, for example the lang of an e2e scenario.

HarborTask( name: str, instruction: str, agent_timeout_sec: float | None, verifier_timeout_sec: float | None, gpus: int, metadata: Mapping[str, typing.Any], path: pathlib._local.Path)
name: str
instruction: str
agent_timeout_sec: float | None
verifier_timeout_sec: float | None
gpus: int
metadata: Mapping[str, typing.Any]
path: pathlib._local.Path
turns: tuple[str, ...]

The user messages of the task, one for each turn; see split_turns.

@dataclass(frozen=True, slots=True)
class HarborTrial:

One Harbor trial, read from its result.json.

task is the Harbor task name. agent is the agent name that the trial reports. model is the model name of the trial configuration, for example openai/gpt-6.1-sol. reward is the reward value of the verifier. It is None when the verifier gave no reward, for example after an infrastructure error. rewards holds all named rewards. error is the exception type of a failed trial. The token counts and cost_usd are None when the agent did not report them. path is the result.json file.

HarborTrial( job: str, trial: str, task: str, agent: str, agent_version: str | None, model: str, reward: float | None, rewards: Mapping[str, float], error: str | None, input_tokens: int | None, cache_tokens: int | None, output_tokens: int | None, cost_usd: float | None, task_checksum: str | None, started_at: str | None, finished_at: str | None, path: pathlib._local.Path)
job: str
trial: str
task: str
agent: str
agent_version: str | None
model: str
reward: float | None
rewards: Mapping[str, float]
error: str | None
input_tokens: int | None
cache_tokens: int | None
output_tokens: int | None
cost_usd: float | None
task_checksum: str | None
started_at: str | None
finished_at: str | None
path: pathlib._local.Path
is_scored: bool

True when the verifier gave a reward.

is_passed: bool

True when the reward is 1 or more.

@dataclass(frozen=True)
class Leaderboard:

A Harbor Hub leaderboard. vis is None without a local result for its dataset.

Leaderboard( package: str, name: str, title: str, rows: tuple[LeaderboardRow, ...], vis: VisStanding | None)
package: str
name: str
title: str
rows: tuple[LeaderboardRow, ...]
vis: VisStanding | None
@dataclass(frozen=True)
class LeaderboardRow:

One published agent and model result.

accuracy_percent is over all trials of the row. Per-trial values divide the row totals by trials, and cost_per_success_usd divides the total cost by successes. None means that the board does not publish the value.

LeaderboardRow( rank: int, agent: str, model: str, reasoning_effort: str, accuracy_percent: float, ci95_half_width: float | None, trials: int, successes: int, total_cost_usd: float | None, cost_per_trial_usd: float | None, cost_per_success_usd: float | None, tokens_per_trial: int | None, avg_trial_minutes: float | None, date: str)
rank: int
agent: str
model: str
reasoning_effort: str
accuracy_percent: float
ci95_half_width: float | None
trials: int
successes: int
total_cost_usd: float | None
cost_per_trial_usd: float | None
cost_per_success_usd: float | None
tokens_per_trial: int | None
avg_trial_minutes: float | None
date: str
@dataclass(frozen=True)
class OutcomeCount:

Tasks whose latest scored attempt ended the same way.

OutcomeCount(outcome: str, count: int, tasks: tuple[str, ...])
outcome: str
count: int
tasks: tuple[str, ...]
@dataclass(frozen=True)
class RepeatRow:

pass@k and pass^k in percent for one k, over the tasks with k attempts.

RepeatRow( k: int, tasks: int, pass_at_k_percent: float | None, pass_all_k_percent: float | None)
k: int
tasks: int
pass_at_k_percent: float | None
pass_all_k_percent: float | None
@dataclass(frozen=True)
class RepeatStats:

The statistics of the repeated attempts of a set of tasks.

attempts_per_task is the most common attempt count, the larger one on a tie. attempt_pass_rate_ci95 is the 95% cluster interval of the attempt pass rate over the tasks. A true change of detectable_change or more shows with 80% power. pass_at_k is the mean chance over the tasks that at least one of k attempts solves the task, and pass_all_k that all k attempts solve it. Each task gives the unbiased estimate from all its attempts, so a task with more than k attempts uses all of them. Only tasks with at least k attempts count. Both are None when no task has k attempts, or when each task has one attempt and you give no k. unstable_tasks names the tasks with both solved and failed attempts.

RepeatStats( attempts_per_task: int, k: int, scored_attempts: int, solved_attempts: int, attempt_pass_rate: float | None, attempt_pass_rate_ci95: tuple[float, float] | None, detectable_change: float | None, pass_at_k: float | None, pass_all_k: float | None, unstable_tasks: tuple[str, ...])
attempts_per_task: int
k: int
scored_attempts: int
solved_attempts: int
attempt_pass_rate: float | None
attempt_pass_rate_ci95: tuple[float, float] | None
detectable_change: float | None
pass_at_k: float | None
pass_all_k: float | None
unstable_tasks: tuple[str, ...]
@dataclass(frozen=True)
class RoundReport:

The full report of one round.

score and diagnosis are the bench_report and diagnose results. repeats has one row for each k from 1 to the attempts per task. judge compares the stages of a judge with human stage labels, or is None without a judge folder.

RoundReport( jobs: str, score: BenchReport, repeats: tuple[RepeatRow, ...], diagnosis: Diagnosis, judge: StageAgreement | None)
jobs: str
score: BenchReport
repeats: tuple[RepeatRow, ...]
diagnosis: Diagnosis
judge: StageAgreement | None
@dataclass(frozen=True)
class RouteTest:

The exact McNemar test of the paired cases of one route.

worse and better count the cases whose pass rate fell or rose. p_value is the two-sided exact p-value. q_value is its Benjamini-Hochberg value over all routes of the comparison.

RouteTest( route: str, paired_cases: int, worse: int, better: int, p_value: float, q_value: float)
route: str
paired_cases: int
worse: int
better: int
p_value: float
q_value: float
@dataclass(frozen=True, slots=True)
class RunSummary:

The counts, the pass rate and the cost of a set of trials.

A case is one task with one agent and one model. pass_rate is passed / scored, and pass_rate_ci95 is its 95% cluster interval over the cases. Both are fractions, or None without a scored trial. errors counts the trials with an exception. cost_usd is the sum of the reported costs, or None when no trial reports a cost. cost_per_pass_usd divides that sum by the passed trials.

RunSummary( trials: int, scored: int, passed: int, errors: int, cases: int, pass_rate: float | None, pass_rate_ci95: tuple[float, float] | None, input_tokens: int, cache_tokens: int, output_tokens: int, cost_usd: float | None, cost_per_pass_usd: float | None, tasks: tuple[str, ...], agents: tuple[str, ...], models: tuple[str, ...], started_at: str | None, finished_at: str | None)
trials: int
scored: int
passed: int
errors: int
cases: int
pass_rate: float | None
pass_rate_ci95: tuple[float, float] | None
input_tokens: int
cache_tokens: int
output_tokens: int
cost_usd: float | None
cost_per_pass_usd: float | None
tasks: tuple[str, ...]
agents: tuple[str, ...]
models: tuple[str, ...]
started_at: str | None
finished_at: str | None
@dataclass(frozen=True)
class StageAgreement:

How well the failure stages of one judge agree with human stage labels.

stages names the judged stage of each trial: the most common stage of its samples. sample_agreement_percent is the share of trials whose samples all agree, or None with one sample. sample_alpha is Krippendorff's alpha of the samples, corrected for agreement by chance, and sample_advice applies its limits. kappa, alpha and advice compare the judged stages with the labels, with the limits of labels.

StageAgreement( judged: int, errors: int, stages: tuple[str, ...], sample_agreement_percent: float | None, sample_alpha: float | None, sample_advice: str, labelled: int, agreement_percent: float | None, kappa: float | None, alpha: float | None, disagreements: tuple[str, ...], advice: str)
judged: int
errors: int
stages: tuple[str, ...]
sample_agreement_percent: float | None
sample_alpha: float | None
sample_advice: str
labelled: int
agreement_percent: float | None
kappa: float | None
alpha: float | None
disagreements: tuple[str, ...]
advice: str
@dataclass(frozen=True)
class StageLabel:

A human failure stage of one scenario trial, kept in a stage labels file.

judged_stage is the stage that a judge gave the same trial, or None.

StageLabel( trial: str, stage: str, judged_stage: str | None, note: str, labelled_at: str)
trial: str
stage: str
judged_stage: str | None
note: str
labelled_at: str
@dataclass(frozen=True, slots=True)
class Step:

One ATIF step: the user request, or one agent iteration and its code blocks.

message is text for Vis runs. On the final agent step it is the final answer. extra keeps every Vis field of the step, also unknown ones.

Step( step_id: int, source: str, message: Any, timestamp: str | None, model_name: str | None, reasoning_effort: str | None, reasoning_content: str | None, metrics: StepMetrics | None, tool_calls: tuple[ToolCall, ...], extra: Mapping[str, typing.Any])
step_id: int
source: str
message: Any
timestamp: str | None
model_name: str | None
reasoning_effort: str | None
reasoning_content: str | None
metrics: StepMetrics | None
tool_calls: tuple[ToolCall, ...]
extra: Mapping[str, typing.Any]
iteration: int | None

The Vis iteration number, or None for the user step.

is_final: bool

True when Vis marked this step as the final answer.

provider: str | None

The provider of this step's model call, or None.

error: str | None

The error text of this step, or None.

finished_at_ms: int | None

The end time of the step in epoch milliseconds, or None.

text: str | None

The message as text, or None.

function_calls: tuple[FunctionCall, ...]

Each host function call of this step in run order.

def to_dict(self) -> dict[str, typing.Any]:

Return the step, its blocks and its function calls as plain JSON data.

@dataclass(frozen=True, slots=True)
class StepMetrics:

The tokens and the cost of one provider call.

prompt_tokens includes cached_tokens. A value is None when the run did not record it. These metrics belong to the whole step, so do not add them up over the function calls of the step.

StepMetrics( prompt_tokens: int | None = None, completion_tokens: int | None = None, cached_tokens: int | None = None, cost_usd: float | None = None, cache_creation_tokens: int | None = None, reasoning_tokens: int | None = None)
prompt_tokens: int | None
completion_tokens: int | None
cached_tokens: int | None
cost_usd: float | None
cache_creation_tokens: int | None
reasoning_tokens: int | None
def to_dict(self) -> dict[str, typing.Any]:

Return the metrics as plain JSON data. An unknown value is None.

@dataclass(frozen=True)
class TaskNote:

One task in a finding.

checks is passed/total verifier tests, or empty. budget_used is the share of the task's agent time limit that the agent used.

TaskNote( task: str, outcome: str, checks: str, pass_ratio: float | None, budget_used: float | None, minutes: float | None, iterations: int | None)
task: str
outcome: str
checks: str
pass_ratio: float | None
budget_used: float | None
minutes: float | None
iterations: int | None
@dataclass(frozen=True, slots=True)
class ToolCall:

One code block that the model ran, and its outcome.

output is the observation that the model saw: the standard output and the error text. activity is the Activity receipt of the block, or None when the run did not record one. Each host function call in the block is in function_calls. extra keeps every Vis field of the call, also unknown ones.

ToolCall( step_id: int, iteration: int | None, tool_call_id: str, function_name: str, arguments: Mapping[str, typing.Any], output: str | None, status: str | None, ref: str | None, started_at_ms: int | None, finished_at_ms: int | None, is_repaired: bool, is_timeout: bool, stdout_chars: int | None, error: Mapping[str, typing.Any] | None, repaired_source: str | None, activity: blockether.vis.activity.ActivityProjection | None, activity_problem: str | None, step_metrics: StepMetrics | None, extra: Mapping[str, typing.Any])
step_id: int
iteration: int | None
tool_call_id: str
function_name: str
arguments: Mapping[str, typing.Any]
output: str | None
status: str | None
ref: str | None
started_at_ms: int | None
finished_at_ms: int | None
is_repaired: bool
is_timeout: bool
stdout_chars: int | None
error: Mapping[str, typing.Any] | None
repaired_source: str | None
activity_problem: str | None
step_metrics: StepMetrics | None
extra: Mapping[str, typing.Any]
code: str | None

The Python source of the block, or None.

stdout: str

The standard output of the block: the start of output.

The first stdout_chars Unicode code points of output are the standard output. A newline separates them from the error text. Vis leaves a blank standard output out of output, so a blank or missing standard output gives "".

activity_state: str | None

The state of the whole block, or None without Activity.

activity_counts: blockether.vis.activity.ActivityCounts | None

The function calls of the block by outcome, or None without Activity.

function_calls: tuple[FunctionCall, ...]

Each host function call of the block in run order, child rows included.

def to_dict(self) -> dict[str, typing.Any]:

Return the block and its function calls as plain JSON data.

@dataclass(frozen=True, slots=True)
class Trajectory:

One Vis run read from an ATIF trajectory.

The typed fields hold what evaluations check. document keeps the whole input, also the keys that this reader does not know. Child runs are in subagent_trajectories.

Trajectory( schema_version: str, session_id: str | None, trajectory_id: str | None, agent: Mapping[str, typing.Any], steps: tuple[Step, ...], final_metrics: TrajectoryMetrics | None, extra: Mapping[str, typing.Any], subagent_trajectories: tuple[Trajectory, ...], document: Mapping[str, typing.Any])
schema_version: str
session_id: str | None
trajectory_id: str | None
agent: Mapping[str, typing.Any]
steps: tuple[Step, ...]
final_metrics: TrajectoryMetrics | None
extra: Mapping[str, typing.Any]
subagent_trajectories: tuple[Trajectory, ...]
document: Mapping[str, typing.Any]
agent_name: str | None

The agent name, for example vis.

agent_version: str | None

The agent version, or None.

model_name: str | None

The model of the run, or None.

provider: str | None

The provider of the run, or None.

profile: Mapping[str, typing.Any] | None

The run options that shaped the agent, or None.

status: str | None

The run status that Vis recorded, or None.

error: str | None

The error text of the run, or None.

goal: Mapping[str, typing.Any] | None

The saved session goal of a persisted run, or None.

session_usage: Mapping[str, typing.Any] | None

The usage rollup of a persisted session, or None.

evaluation: Mapping[str, typing.Any] | None

The structured answer check (extra.eval), or None.

user_message: str | None

The text of the first user step, or None.

agent_steps: tuple[Step, ...]

The agent steps in run order.

final_step: Step | None

The step with the final answer, or None when the run has no final answer.

Vis marks that step with is_final. A file without marks, such as one from an earlier Vis version, ends with the last agent step, if that step has no code blocks and no error.

final_answer: str | None

The final answer text, or None.

tool_calls: tuple[ToolCall, ...]

Each code block of the run in order.

def function_calls( self, *, operation: str | None = None, state: str | None = None) -> tuple[FunctionCall, ...]:

Each host function call of this run in order, optionally filtered.

Arguments:
  • operation: Keep only calls of this function, for example grep.
  • state: Keep only calls with this outcome, for example failed.

The calls of child runs are in subagent_trajectories, not here.

def to_dict(self) -> dict[str, typing.Any]:

Return the typed view of the run as plain JSON data.

For the original ATIF document, use document.

@dataclass(frozen=True, slots=True)
class TrajectoryMetrics:

The totals of a run: tokens, cost, steps and time.

total_prompt_tokens includes total_cached_tokens. A value is None when the run did not record it.

TrajectoryMetrics( total_prompt_tokens: int | None = None, total_completion_tokens: int | None = None, total_cached_tokens: int | None = None, total_cost_usd: float | None = None, total_steps: int | None = None, cache_creation_tokens: int | None = None, reasoning_tokens: int | None = None, total_tokens: int | None = None, iteration_count: int | None = None, duration_ms: int | None = None, subagent_cost_usd: float | None = None)
total_prompt_tokens: int | None
total_completion_tokens: int | None
total_cached_tokens: int | None
total_cost_usd: float | None
total_steps: int | None
cache_creation_tokens: int | None
reasoning_tokens: int | None
total_tokens: int | None
iteration_count: int | None
duration_ms: int | None
subagent_cost_usd: float | None
def to_dict(self) -> dict[str, typing.Any]:

Return the totals as plain JSON data. An unknown value is None.

@dataclass(frozen=True)
class TrialLabel:

A human cause for one failed attempt, kept in the labels file of its round.

diagnosed_cause is the automatic attempt_cause of the same attempt.

TrialLabel( task: str, trial: str, round_name: str | None, cause: str, diagnosed_cause: str, note: str, labelled_at: str)
task: str
trial: str
round_name: str | None
cause: str
diagnosed_cause: str
note: str
labelled_at: str
@dataclass(frozen=True)
class VisStanding:

Where the local Vis benchmark result would stand on the board.

Like the board accuracy, pass_rate_percent counts every scored attempt. strict_pass_rate_percent also counts each unscored, unattempted and GPU task as attempts_per_task failed attempts. Each _ci95 value is a 95% interval with the attempts of one task as one cluster. The ranks insert the rates among the board rows, and each rank range inserts the ends of its interval. Per-attempt values use model attempts. caveats lists the limits of the comparison.

VisStanding( model: str, round_name: str | None, scored_tasks: int, total_tasks: int, attempts_per_task: int, pass_rate_percent: float, pass_rate_ci95: tuple[float, float] | None, strict_pass_rate_percent: float, strict_pass_rate_ci95: tuple[float, float] | None, rank_by_pass_rate: int, rank_range: tuple[int, int] | None, rank_by_strict_rate: int, strict_rank_range: tuple[int, int] | None, cost_per_attempt_usd: float | None, cost_per_solved_usd: float | None, tokens_per_attempt: int | None, avg_attempt_minutes: float | None, caveats: tuple[str, ...])
model: str
round_name: str | None
scored_tasks: int
total_tasks: int
attempts_per_task: int
pass_rate_percent: float
pass_rate_ci95: tuple[float, float] | None
strict_pass_rate_percent: float
strict_pass_rate_ci95: tuple[float, float] | None
rank_by_pass_rate: int
rank_range: tuple[int, int] | None
rank_by_strict_rate: int
strict_rank_range: tuple[int, int] | None
cost_per_attempt_usd: float | None
cost_per_solved_usd: float | None
tokens_per_attempt: int | None
avg_attempt_minutes: float | None
caveats: tuple[str, ...]
def agent_limits(dataset: pathlib._local.Path | str) -> dict[str, float]:

Read the agent time limit in seconds of each task of a dataset directory.

Each task directory holds a task.toml with agent.timeout_sec. A directory that does not exist gives an empty map.

def agreement( items: Iterable[tuple[str, str, str]], *, source: str, noun: str, labeller: str, fix: str) -> Agreement:

Compare automatic results with human labels.

Each item is (name, automatic, human). source names the automatic side in each disagreement, for example "judge". noun, labeller and fix go to kappa_advice.

def alpha_advice(alpha: float | None, *, noun: str = 'samples') -> str:

Say if Krippendorff's alpha of repeated noun makes them reliable.

def attempt_cause(trial: Mapping, summary: Mapping, limits: Mapping[str, float]) -> str:

Give the diagnose cause of one attempt, or solved.

The causes are dead_tools, agent_errors, agent_timeouts, unscored, early_stops and other. Empty limits use the agent_limits of the summary.

def attempt_outcome(trial: Mapping) -> str:

Name how one attempt ended, with the task outcome rules of the round summary.

A Harbor exception or a failed trajectory gives agent_error:<type>.

def bench_report( summary: Mapping, round_name: str | None = None) -> BenchReport:

Score one round from its round_summary data.

def benjamini_hochberg(p_values: Sequence[float]) -> list[float]:

Return the Benjamini-Hochberg q-value of each p-value, in the input order.

A q-value is the smallest false discovery rate at which its test is a discovery. A larger p-value never gets a smaller q-value, and each q-value is at most 1.

def case_results( trials: Iterable[HarborTrial]) -> dict[tuple[str, str, str], tuple[int, int]]:

Count the passed and the scored trials of each case: a task, an agent and a model.

The keys are (task, agent, model). A trial without a reward does not count, so a case without a scored trial is absent.

def cause_check( summary: Mapping, labels: Mapping[str, Mapping], limits: Mapping[str, float]) -> Agreement:

Compare the diagnose causes with the human labels of the same attempts.

labels maps a trial name to a label with a cause value. Labels of trials that are not in the summary do not count.

def cluster_interval(units: Sequence[tuple[int, int]]) -> tuple[float, float] | None:

Return the 95% interval of the pooled pass rate of clustered units, or None.

def cohen_kappa(pairs: Sequence[tuple[Hashable, Hashable]]) -> float | None:

Return Cohen's kappa of two raters over the same items, or None when undefined.

def compare_cases( baseline: Mapping[tuple[str, ...], tuple[int, int]], candidate: Mapping[tuple[str, ...], tuple[int, int]], *, baseline_name: str = 'baseline', candidate_name: str = 'candidate', margin_points: float | None = None) -> CaseComparison:

Compare the (passed, runs) cases of a baseline and a candidate.

Each key is (task, agent, model). When the sides share no key and each side has one route, the comparison pairs the cases by task and adds a caveat. The names appear in error messages. margin_points is the smallest change in points that matters, above 0 and at most 100. Raise ValueError when no case pairs, or for a margin outside that range.

def detectable_change(units: Sequence[tuple[int, int]]) -> float | None:

Return the smallest change of the pass rate that a second run of this size detects.

The test compares two independent runs of the same size, two-sided at the 5% level, with 80% power. The change holds for a rise and for a fall, so the larger of the two applies. A direction without a detectable change before 0 or 1 does not count. The change has steps of 0.001. None means that there are no runs, or that no change from 0 to 1 is detectable.

def diagnose( summary: Mapping, labels: Mapping[str, Mapping] | None = None, limits: Mapping[str, float] | None = None) -> Diagnosis:

Find where a round loses tasks, from its round_summary data.

limits gives the agent time limit of each task, from agent_limits. Without limits, it uses the agent_limits of the summary, and without those no attempt counts as an early stop. labels maps a trial name to a human label with a cause value.

def effective_size(units: Sequence[tuple[int, int]]) -> float:

Return the run count of clustered results, made smaller by their design effect.

Each unit is (passed, runs) of one case. The size stays between the unit count and the run count. Without any variation, it is the unit count.

def find_trial(summary: Mapping, task: str) -> dict:

Find an attempt by trial name, or the latest scored attempt of a task.

A name with __ is a trial name, such as task__id. Raises LookupError when the summary has no such attempt.

def judge_errors(job_dir: pathlib._local.Path | str) -> int:

Count the judge results with an error in judge_report.json, or 0 without it.

def judged_samples(job_dir: pathlib._local.Path | str) -> dict[str, list[str]]:

Read the stage of each judge sample, grouped by trial, from judge_stages.json.

A file that is missing or is not a JSON object gives an empty map.

def kappa_advice( count: int, kappa: float | None, *, noun: str, labeller: str, fix: str) -> str:

Say if count labels with this kappa make the automatic noun usable.

labeller names the way to add a label, and fix is the action for a low kappa.

def krippendorff_alpha(units: Iterable[Sequence[Hashable | None]]) -> float | None:

Return Krippendorff's alpha of nominal values, or None when it is undefined.

Each unit holds the values that the raters gave one item. None is a missing value. A unit with fewer than two values does not count. Alpha is 1 for full agreement, 0 for agreement by chance, and below 0 for less agreement than chance. It is undefined without two values in one unit, or when all values are the same.

def latest_trials(summary: Mapping) -> dict[str, dict]:

Give the latest scored trial of each task, by its finish time.

def leaderboard_row(raw: Mapping) -> LeaderboardRow:

Read one row of a Harbor Hub leaderboard page.

def load_labels(path: pathlib._local.Path | str) -> dict[str, dict]:

Read the labels file of a round: a map from trial name to its label.

A file that is missing or is not a JSON object gives an empty map.

def load_stage_labels(path: pathlib._local.Path | str) -> dict[str, dict]:

Read a stage labels file: a map from trial name to its label.

A file that is missing or is not a JSON object gives an empty map.

def load_trajectory( path: str | os.PathLike[str], *, strict: bool = True) -> Trajectory:

Read an ATIF trajectory file (UTF-8 JSON) and parse it.

Raises:
  • OSError: The file cannot be read.
  • ValueError: The file is not JSON or not a valid ATIF trajectory.
def majority( found: Sequence[str], order: Sequence[str] = ('understand', 'write', 'verify', 'other')) -> str:

Return the most common value of found. A tie gives the value first in order.

def mcnemar_exact(worse: int, better: int) -> float:

Return the two-sided exact McNemar p-value of paired pass and fail results.

worse counts the cases that pass only in the baseline. better counts the cases that pass only in the candidate. Cases with the same result on both sides do not change the value. Without a changed case, the value is 1.

def merge_trajectories(documents: list[Mapping[str, typing.Any]]) -> dict[str, typing.Any]:

Join the ATIF documents of the turns of one session into one document.

Give the documents in turn order. The steps follow in that order with new step ids from 1. final_metrics adds the numbers of all turns, and subagent_trajectories keeps every child. The other keys come from the last document. extra.turns is the count of turns. The first turn error stays in extra.error, so a failed turn also fails the merged run. An empty list raises ValueError.

def paired_change(changes: Sequence[float]) -> tuple[tuple[float, float], float]:

Return the Agresti-Min 95% interval of a mean paired change, and its floor.

Each change is the candidate pass rate of one case minus its baseline pass rate. The floor is the smallest true change that this many cases detect with 80% power. The interval stays inside the possible changes, from -1 to 1.

def parse_trajectory( document: Mapping[str, typing.Any], *, strict: bool = True) -> Trajectory:

Build a Trajectory from an ATIF document, a decoded JSON object.

The result is immutable and independent of document. With strict=True, an Activity receipt that breaks the Activity contract raises ValueError. With strict=False, that block gets activity=None and the reason in activity_problem, so older files stay readable.

Raises:
  • ValueError: The document is not an ATIF trajectory, or an Activity receipt is invalid. The message never contains run content.
def pass_all_k(runs: int, passed: int, k: int) -> float:

Return the unbiased estimate of pass^k from passed solved of runs attempts.

pass^k is the chance that all of k attempts solve the task. The limits of pass_at_k apply.

def pass_at_k(runs: int, passed: int, k: int) -> float:

Return the unbiased estimate of pass@k from passed solved of runs attempts.

pass@k is the chance that at least one of k attempts solves the task. The estimate uses all runs attempts, not only the first k. Raises ValueError unless 0 <= passed <= runs and 1 <= k <= runs.

def read_leaderboard( package: str, name: str, summary: Mapping | None = None, round_name: str | None = None, post: Callable[[str, dict], dict] | None = None) -> Leaderboard:

Read every page of one board, and place the round of summary on it.

package and name select the Harbor Hub board, for example terminal-bench/terminal-bench and 4-0-0.

Sends one unauthenticated request for each page, at most 50. post sends one request and gives the parsed reply; the default uses urllib. vis is None without a summary or when the summary covers another dataset version.

def read_tasks( dataset: pathlib._local.Path | str) -> tuple[HarborTask, ...]:

Read the tasks of a Harbor dataset directory, sorted by name.

Each task is a folder with a task.toml. Other folders do not count. A directory that does not exist raises FileNotFoundError. The e2e directory of the Vis repository is such a dataset.

def read_trial( path: pathlib._local.Path | str) -> HarborTrial | None:

Read one finished trial from its result.json, or return None for other files.

A job result.json, an unfinished trial and a file that is not JSON give None.

def read_trials( directory: pathlib._local.Path | str) -> tuple[HarborTrial, ...]:

Read the finished trials of a Harbor job directory or of a round of jobs.

A trial result.json is one or two levels below directory. The trials come in start order. A directory that does not exist raises FileNotFoundError.

def render_report(report: RoundReport) -> str:

Give the text that the command prints for report.

def repeat_stats( attempts: Mapping[str, Sequence[bool]], k: int | None = None) -> RepeatStats:

Give the repeat statistics of the attempts of each task, True when solved.

k is the attempt count of pass@k and pass^k. Without it, k is attempts_per_task. Raises ValueError for a k below 1.

def round_report( jobs_dir: pathlib._local.Path | str, dataset_dir: pathlib._local.Path | str | None = None, *, model: str | None = None, round_name: str | None = None, labels: pathlib._local.Path | str | None = None, judge_dir: pathlib._local.Path | str | None = None, stage_labels: pathlib._local.Path | str | None = None) -> RoundReport:

Build the full report of the round in jobs_dir.

dataset_dir and model go to round_summary. labels is the cause labels file, by default LABELS_FILE in jobs_dir. judge_dir is the job folder of a judge. Without it, the report uses jobs_dir when it holds judge stages. stage_labels is the stage labels file, by default STAGE_LABELS_FILE in the judge folder. Raises FileNotFoundError when jobs_dir holds no trial.

def round_summary( jobs_dir: pathlib._local.Path | str, dataset_dir: pathlib._local.Path | str | None = None, *, model: str | None = None, dataset: str | None = None) -> dict:

Summarize every trial under jobs_dir, as bench_report and diagnose read it.

jobs_dir is one Harbor job folder or a folder of jobs. dataset_dir holds one folder with a task.toml for each task. It gives the total, the GPU tasks, the tasks without an attempt and the agent_limits that diagnose uses. Without it, the total is the attempted tasks. dataset names the dataset. Without it, the name is the source that Harbor wrote in the trials. model is the model name or the provider/model route of the agent. An attempt of another model does not count. With no model, any model counts. Each task counts once, by its latest scored attempt.

def run_tasks( path: pathlib._local.Path | str, model: str, *, jobs_dir: pathlib._local.Path | str = 'jobs', job_name: str | None = None, attempts: int = 1, concurrency: int = 1, agent_options: Mapping[str, str] | None = None, report: bool = True) -> pathlib._local.Path:

Run the tasks in path with the Vis agent on this host and return the job folder.

path is one task folder or a dataset folder of tasks. model is the provider/model route of the agent. Harbor writes the job to jobs_dir/job_name. Without job_name, Harbor names the job with the time. attempts is the number of trials for each task. concurrency is the number of trials that run at the same time. agent_options are the agent options of harbor run --ak, for example {"source": "/path/to/vis"}. The call sets HARBOR_TELEMETRY=off, so Harbor sends no usage data. Each trial makes paid model calls. Call it outside a running event loop. With report, the call then writes the round report vis-report.json in the job folder, as python -m blockether.vis.evals.report does. A report error gives a warning and keeps the job folder.

def sample_agreement(samples: Mapping[str, Sequence[str]]) -> float | None:

Return the share of items whose samples all agree, as a percent.

Only items with more than one sample count. Return None without such items.

def sample_alpha(samples: Mapping[str, Sequence[str]]) -> float | None:

Return Krippendorff's alpha of the samples of each item, rounded to 3 places.

Each item is one unit and each sample one value. Items can have different sample counts. Return None when alpha is undefined, for example with one sample.

def save_label( path: pathlib._local.Path | str, summary: Mapping, task: str, cause: str, note: str = '', *, limits: Mapping[str, float] | None = None, round_name: str | None = None) -> TrialLabel:

Save the human cause of one failed attempt in the labels file at path.

task is a task name or a trial name, as in find_trial. cause is one of CAUSES. The write replaces the file at once. Raises ValueError for an unknown cause, a note longer than NOTE_LIMIT or a solved attempt.

def save_stage_label( path: pathlib._local.Path | str, trial: str, stage: str, note: str = '', *, judged_stage: str | None = None) -> StageLabel:

Save your failure stage of one trial in the stage labels file at path.

stage is one of STAGES. A new label of the same trial replaces the old one, and the write replaces the file at once. Raises ValueError for an unknown stage or a note longer than NOTE_LIMIT.

def seal_digest(body: Mapping[str, typing.Any]) -> str:

Return the sha256 of a seal body with sorted keys, without its digest key.

def seal_round( round_name: str, runs: Sequence[Mapping[str, typing.Any]], trials: Iterable[HarborTrial], root: pathlib._local.Path | str, sealed_at: str | None = None) -> dict[str, typing.Any]:

Return the seal of a round: its run records, one row for each trial and a digest.

Each run record holds the commit, the pins, the profiles and the artifact sha256 of one run. The rows hold the task checksums and the result digests. sealed_at defaults to the current UTC time.

def seal_rows( trials: Iterable[HarborTrial], root: pathlib._local.Path | str) -> list[dict[str, typing.Any]]:

Return one seal row for each trial, with the task checksum and the result digest.

result is the path of the trial result.json, relative to root, and result_sha256 is the sha256 of that file.

def split_turns(instruction: str) -> list[str]:

Return the user messages of instruction, one for each turn.

Each line with only TURN_MARKER ends one message and starts the next. An instruction without that line is one turn and stays unchanged. An empty message raises ValueError.

def stage_agreement( samples: Mapping[str, Sequence[str]], labels: Mapping[str, Mapping], errors: int = 0) -> StageAgreement:

Compare the judged stages with human stage labels of the same trials.

samples maps a trial name to the stage of each judge sample, as from judged_samples. labels maps a trial name to a label with a stage value, as from load_stage_labels. Labels of trials without a judged stage do not count.

def standing( rows: tuple[LeaderboardRow, ...], data: Mapping, package: str, name: str, round_name: str | None = None) -> VisStanding | None:

Place the round of data, its summary.json data, among the board rows.

Gives None when the summary covers another dataset or version.

def summarize( trials: Iterable[HarborTrial]) -> RunSummary:

Summarize a set of trials: counts, the pass rate with its interval, and the cost.

def task_problems(task: pathlib._local.Path | str) -> tuple[str, ...]:

Return the problems of one task folder that give a wrong score; empty is valid.

The check reads only files and makes no model calls. It needs an executable tests/test.sh and the folder environment/workspace. It checks tests/checks.json with check_problems and needs each Python file that the checks name. The harness writes .vis/config.yml in the trial workspace, so a seed file there is a problem. Each problem starts with the task name. run_tasks and python -m blockether.vis.evals.check refuse tasks with problems.

def summary_attempts(summary: Mapping) -> dict[str, list[bool]]:

List the scored attempts of each task in finish order, True when solved.

def task_attempts( trials: Iterable[HarborTrial]) -> dict[str, list[bool]]:

List the scored attempts of each task in finish order, True when passed.

def task_note( task: str, outcome: str, trial: Mapping, limit: float | None) -> TaskNote:

Describe one attempt of a task for a finding. limit is its agent time limit.

def wilson_interval( passed: int, runs: int, size: float | None = None) -> tuple[float, float] | None:

Return the 95% Wilson interval of a pass rate, or None without runs.

size replaces the run count when the runs are not independent.

def write_report( report: RoundReport, path: pathlib._local.Path | str) -> pathlib._local.Path:

Write report as JSON to path and replace the old file at once.

def write_task( dataset: pathlib._local.Path | str, name: str, instruction: str | Sequence[str], checks: Mapping[str, typing.Any], *, files: Mapping[str, str] | None = None, agent_timeout_sec: float = 300.0, verifier_timeout_sec: float = 300.0, metadata: Mapping[str, str | int | float | bool] | None = None, artifacts: Sequence[str] = (), filesystem_roots: Mapping[str, str] | None = None, test_files: Mapping[str, str] | None = None) -> HarborTask:

Write the Vis task name in the folder dataset and return it.

name is lowercase words joined by hyphens. instruction is the prompt of the agent. A list of strings gives one user message for each turn of one session; write_task joins them with TURN_MARKER lines. checks is the tests/checks.json object that the Vis grader reads, for example {"want": {"m.py": ["def c("]}}. files maps each seed file path to its text. The paths are relative to environment/workspace/. artifacts lists the workspace files that Harbor copies to artifacts/ in the trial folder after the check, for example the call record of a mock extension. filesystem_roots maps each filesystem root id to a workspace folder; the trial registers them. test_files maps each file name in tests/ to its text, for example the verifier that the verify check names. The agent never sees these files. The task gets tests/test.sh, which runs the grader. An existing task folder, an invalid name, invalid checks or a path outside its folder raise an error.