`harness`¶

The open_atp.harness package is the agent concern composed by the AgentProver: for one agent CLI it owns the launch script, credential forwarding, asset staging, and token/cost parsing. The compute concern (where the command runs, with Lean+Mathlib) lives in the injected ComputeBackend.

See the per-harness prover pages under Provers for credential setup.

Base¶

class open_atp.harness.base.Harness(*, model: str = 'claude-opus-4-8', effort: str = 'high')[source]¶

Base class for an agent CLI harness.

Parameters:

modelstr, default “claude-opus-4-8”: Model id the agent runs. Subclasses override the default with the model their CLI drives.
effortstr, default “high”: Reasoning-effort level passed to harnesses that support it.

property command: str¶

Bash command the backend runs to launch the agent.

The backend has already cd’d into the workdir and symlinked .lake; we export $PROMPT from the written prompt file (the launch scripts reference it) and run the rendered script.

Returns:

str: The bash one-liner that exports $PROMPT and runs agent.sh.

agent_auth() → AgentAuth[source]¶

Resolve this harness’s credentials into a ready-to-forward auth bundle.

Raises MissingCredentials if a key this harness requires is absent from both its explicit arguments and the host environment.

Returns:

AgentAuth: Resolved env (name -> value) and (host_dir, dest_basename) mounts the prover forwards into the sandbox.

Examples

An explicit oauth_token is resolved into the forwarded env:

>>> from open_atp.harness import ClaudeCodeHarness
>>> harness = ClaudeCodeHarness(oauth_token="sk-ant-oat-fake")
>>> harness.agent_auth().env["CLAUDE_CODE_OAUTH_TOKEN"]
'sk-ant-oat-fake'

abstractmethod auth_status() → AuthStatus[source]¶

Report this harness’s credential status without resolving or staging it.

Returns:

AuthStatus: Where the credential lives, whether it is there, and when it expires.

stage_wd(wd: Path) → None[source]¶

Populate wd with the harness’s launch script.

Everything the harness itself owns – not the skills list (the prover stages it via stage_skills()) and not the prompt (the prover and task own it, written via write_prompt()). Subclasses that need more (Vibe’s VIBE_HOME, ax-prover’s per-target setup) override and call super().stage_wd.

Parameters:

wdpathlib.Path: The agent working directory to populate; must already exist.

Raises:

RuntimeError: If wd does not exist.

write_prompt(wd: Path, prompt: str) → None[source]¶

Write the composed prompt where this harness’s launch script reads it.

The prompt’s content is owned by the prover (its prover prompt) and the task (the optional user prompt); the harness owns only the file location and the cat $PROMPT launch contract, so it provides the write mechanism.

Parameters:

wdpathlib.Path: The agent working directory the launch script reads the prompt from.
promptstr: The composed prompt text to write to PROMPT_FILE.

parse_result(lines: list[str], wd: Path) → HarnessRunResult[source]¶

Parse the agent’s run into a HarnessRunResult.

wd is passed explicitly (not stashed on the instance) so a single harness shared across concurrently-running tasks reads this task’s usage: harnesses whose cost/tokens live in workdir files (Vibe’s session log, ax-prover’s -o JSON) read them from wd here. The CLI harnesses parse from lines alone and ignore wd.

Parameters:

lineslist[str]: The agent’s streamed stdout, one JSON object per line.
wdpathlib.Path: The agent working directory this run used; where a harness reads any workdir-local usage files.

Returns:

HarnessRunResult: Token totals, cost, and stop metadata parsed from the run.

collect_logs(wd: Path, logs_dir: Path) → None[source]¶

Move this harness’s rich log files out of wd into logs_dir.

The streamed event JSONL the prover captures from stdout is the agent’s transcript for every CLI harness, so the default does nothing. Harnesses that also drop a richer record inside the workdir override this to relocate those files, so the downloaded workdir stays the proof project and the downloaded logs carry the full record. Runs after parse_result(), which may read those files for cost, so moving them is safe.

Parameters:

wdpathlib.Path: The agent working directory rich log files are moved out of.
logs_dirpathlib.Path: The run’s log directory the files are relocated into.

stage_skills(wd: Path, skill_dirs: list[Path]) → None[source]¶

Copy resolved skill source dirs into this harness’s skill location.

Each <name>/SKILL.md tree lands at wd/<skills_dest>/<dir-name>/ (an upstream tests/ fixture dir is dropped). A no-op for a harness that does not consume skills (skills_dest is None, e.g. ax-prover). The prover owns the list (AgentProver.skills, resolved by resolve_skill); the harness owns where it goes.

Parameters:

wdpathlib.Path: The agent working directory the skills are copied into (under skills_dest).
skill_dirslist[pathlib.Path]: Resolved skill source dirs (each a <name>/SKILL.md tree) to copy.

class open_atp.harness.base.HarnessRunResult(input_tokens: int = 0, cached_input_tokens: int = 0, output_tokens: int = 0, stop_reason: str | None = None, cost_usd: float | None = None, subtype: str | None = None, result_text: str | None = None)[source]¶

Token totals and cost parsed from an agent’s streamed output.

Parameters:

input_tokensint, default 0: Total input (prompt) tokens the run consumed.
cached_input_tokensint, default 0: The cache-hit subset of input_tokens, when the agent reports one.
output_tokensint, default 0: Total output (completion) tokens the run produced.
stop_reasonstr, optional: Why the agent stopped, when the stream reports it; None otherwise.
cost_usdfloat, optional: USD cost if the harness self-reports it (Claude Code, OpenCode); None when it must be estimated from token counts (Codex, ax-prover).
subtypestr, optional: Final type:"result" subtype (Claude Code: success / error_max_turns / error_during_execution); None if not reported.
result_textstr, optional: The agent’s final result text (Claude Code’s result field); None otherwise.

class open_atp.harness.base.AgentAuth(env: dict[str, str] = <factory>, mounts: list[tuple[~pathlib.Path, str]] = <factory>)[source]¶

Resolved credentials a harness hands the prover to wire into the sandbox.

Unlike a declarative spec, env here holds resolved name->**value** pairs – the harness has already read the host environment (and any explicit overrides) and validated that required credentials are present. The prover only forwards them; it never touches os.environ.

Parameters:

envdict[str, str], optional: Environment variables (name -> value) to forward into the sandbox. Defaults to an empty mapping.
mountslist[tuple[pathlib.Path, str]], optional: Host directories to expose under the sandbox’s $HOME, as (host_dir, dest_basename) pairs (e.g. (~/.codex, ".codex")). Defaults to an empty list.

exception open_atp.harness.base.MissingCredentials[source]¶: A credential a prover needs to run is absent.

Harnesses¶

Each harness adapts one agent CLI and is a Harness subclass (set as AgentProver’s harness).

class open_atp.harness.claude_code.ClaudeCodeHarness(*, model: str = 'claude-opus-4-8', effort: str = 'high', plugins: list[str] | None = None, oauth_token: str | None = None)[source]¶

Bases: Harness

Claude Code CLI, authenticated by a long-lived CLAUDE_CODE_OAUTH_TOKEN.

Claude Code is the only harness that loads plugins, so they live here rather than on the prover’s shared skills list.

Parameters:

modelstr, default “claude-opus-4-8”: Model id the agent runs.
effortstr, default “high”: Reasoning-effort level.
pluginslist[str], default [“lean4”]: Claude Code plugins to load, each a name (resolved from the vendored lean4-skills catalog) or a full path to a .claude-plugin/plugin.json tree. An empty list loads none.
oauth_tokenstr, optional: The CLAUDE_CODE_OAUTH_TOKEN (from claude setup-token) to forward into the sandbox. None (default) reads it from the host CLAUDE_CODE_OAUTH_TOKEN env var; resolution fails if neither is set.

Examples

Constructing the harness resolves its defaults:

>>> from open_atp.harness import ClaudeCodeHarness
>>> harness = ClaudeCodeHarness()
>>> harness.name
'claude_code'
>>> harness.plugins
['lean4']

With the credential supplied explicitly, agent_auth() resolves the full forwarded env without touching the host environment:

>>> harness = ClaudeCodeHarness(plugins=[], oauth_token="sk-ant-oat-fake")
>>> harness.agent_auth().env
{'IS_SANDBOX': '1', 'CLAUDE_CODE_OAUTH_TOKEN': 'sk-ant-oat-fake'}

auth_status() → AuthStatus[source]¶

Report this harness’s credential status without resolving or staging it.

Returns:

AuthStatus: Where the credential lives, whether it is there, and when it expires.

class open_atp.harness.codex.CodexHarness(*, model: str = 'gpt-5.5', effort: str = 'high', auth_file: Path | None = None)[source]¶

Bases: Harness

Codex CLI, authenticated by a mounted auth.json credential.

Codex authenticates via ChatGPT/OpenAI, so it must run an OpenAI model; model defaults to gpt-5.5 rather than the Anthropic base default.

Parameters:

modelstr, default “gpt-5.5”: Model id the agent runs; must be an OpenAI model.
effortstr, default “high”: Reasoning-effort level.
auth_filepathlib.Path, optional: The Codex auth.json to mount. None (default) uses ~/.codex/auth.json (from codex login); resolution fails if the file is absent.

Examples

Constructing the harness resolves its defaults:

>>> from open_atp.harness import CodexHarness
>>> harness = CodexHarness()
>>> harness.name
'codex'
>>> harness.model
'gpt-5.5'

auth_status() → AuthStatus[source]¶

Report this harness’s credential status without resolving or staging it.

Returns:

AuthStatus: Where the credential lives, whether it is there, and when it expires.

class open_atp.harness.opencode.OpenCodeHarness(*, provider: str = 'deepseek', model: str = 'deepseek-v4-pro', effort: str = 'high', auth: str = 'api_key', api_key: str | None = None)[source]¶

Bases: Harness

OpenCode CLI driving any supported OpenCode model provider.

Parameters:

providerstr, default “deepseek”: opencode provider name (e.g. "deepseek"). Any OpenCode provider is accepted.
modelstr, default “deepseek-v4-pro”: Model id the agent runs. Must be supported by the chosen provider.
effortstr, default “high”: Reasoning-effort level.
authstr, default “api_key”: Authentication strategy, "api_key" or "login". Any other value raises ValueError. See Authentication for details.
api_keystr, optional: For auth="api_key", the provider’s API key. None (default) reads the host environment, and resolution fails if the key is set in neither. Ignored when auth="login".

Examples

By default, the harness authenticates with the provider’s API key and reads the value from the host environment.

>>> from open_atp.harness import OpenCodeHarness
>>> harness = OpenCodeHarness(provider="openai", model="gpt-5.5")
>>> harness.name
'opencode'
>>> harness.provider
'openai'

auth_status() → AuthStatus[source]¶

Report this harness’s credential status without resolving or staging it.

Returns:

AuthStatus: Where the credential lives, whether it is there, and when it expires.

class open_atp.harness.vibe.VibeHarness(*, model: str = 'labs-leanstral-1-5', effort: str = 'high', agent: str = 'lean', max_turns: int | None = None, max_price: float | None = None, mistral_api_key: str | None = None)[source]¶

Bases: Harness

Mistral Vibe CLI driving its builtin lean agent (Leanstral) in a sandbox.

Vibe’s builtin lean agent is Leanstral: vibe -p ... --agent lean pins the model via the builtin agent profile (there is no --model flag), which as of vibe 2.19 is Leanstral 1.5 (labs-leanstral-1-5). Reaching it requires Lab Model access enabled by a Mistral org admin. The lean agent is install_required – vibe gates it behind an opt-in (the interactive /leanstall command), which does nothing but add "lean" to installed_agents; stage_wd() writes that config key directly, so no interactive install step is needed.

Two things differ from the other harnesses:

VIBE_HOME is workdir-local. vibe_agent.sh exports VIBE_HOME=$PWD/.vibe so vibe’s config (which un-gates the builtin lean agent) and the per-session log all live under the workdir and sync back out.
Cost comes from the session log, not stdout. --output streaming carries only conversation messages – no token/cost totals. Those live in vibe’s per-session meta.json; parse_result() reads it from the synced-back log dir.

Parameters:

modelstr, default “labs-leanstral-1-5”: Model id recorded in the run metadata. Vibe has no --model flag and the builtin lean agent pins its own model, so this is reported only, not passed to vibe. The default is the model that agent pins.
effortstr, default “high”: Reasoning-effort level.
agentstr, default “lean”: Which builtin vibe agent to drive. The default is Leanstral.
max_turnsint, optional: vibe -p turn guard; None (default) leaves it unset.
max_pricefloat, optional: vibe -p price guard; None (default) leaves it unset.
mistral_api_keystr, optional: Mistral La Plateforme key forwarded as MISTRAL_API_KEY. None (default) reads it from the host env var; resolution fails if neither is set.

Examples

Constructing the harness resolves its defaults, including the agent it drives:

>>> from open_atp.harness import VibeHarness
>>> harness = VibeHarness()
>>> harness.name
'vibe'
>>> harness.agent
'lean'

With the key supplied explicitly, agent_auth() forwards it as MISTRAL_API_KEY without reading the host environment:

>>> harness = VibeHarness(mistral_api_key="msk-fake")
>>> harness.agent_auth().env
{'MISTRAL_API_KEY': 'msk-fake'}

auth_status() → AuthStatus[source]¶

Report this harness’s credential status without resolving or staging it.

Returns:

AuthStatus: Where the credential lives, whether it is there, and when it expires.

class open_atp.harness.axproverbase.AxProverBaseHarness(*, model: str = 'claude-opus-4-8', effort: str = 'high', max_iterations: int | None = None, provider_api_key: str | None = None)[source]¶

Bases: Harness

ax-prover-base (LangGraph Lean agent), driven by ax-prover prove in-sandbox.

ax-prover is a self-contained proving agent (its own proposer->builder->reviewer->memory loop) that edits the target .lean file in place. It slots in as a harness rather than a standalone prover because AgentProver.prove() already supplies everything around the edges (staging, snapshot/diff, sandbox run, key forwarding) and the shared Verifier – not ax-prover’s own reviewer – remains the source of truth for compile/sorry/axiom.

Two things differ from the CLI harnesses (it mirrors VibeHarness here):

Config lives in a workdir YAML, not flags. stage_wd() writes axprover.yaml selecting the model/effort/iterations; it layers on top of ax-prover’s bundled default.yaml (auto-prepended by the CLI), so it only needs to override the deltas.
Cost is not on stdout. ax-prover streams human-readable logs, so token usage is read from the JSON it writes per target instead. cost_usd is left None and the prover converts tokens->USD via the fallback table, exactly like CodexHarness.

Parameters:

modelstr, default “claude-opus-4-8”: Model id the agent runs, mapped to ax-prover’s provider:model string.
effortstr, default “high”: Reasoning-effort level, mapped to each provider’s knob.
max_iterationsint, optional: Cap on ax-prover’s proposer->builder->reviewer loop. None (default) keeps ax-prover’s own default (50).
provider_api_keystr, optional: The selected provider’s API key, forwarded under its canonical env var (ANTHROPIC_API_KEY / OPENAI_API_KEY / …). None (default) reads that env var from the host; resolution fails if neither is set.

Examples

Constructing the harness resolves its defaults:

>>> from open_atp.harness import AxProverBaseHarness
>>> harness = AxProverBaseHarness()
>>> harness.name
'axproverbase'
>>> harness.model
'claude-opus-4-8'

With the provider key supplied explicitly, agent_auth() forwards it under the provider’s canonical env var without reading the host environment:

>>> harness = AxProverBaseHarness(provider_api_key="sk-fake")
>>> harness.agent_auth().env
{'ANTHROPIC_API_KEY': 'sk-fake'}

auth_status() → AuthStatus[source]¶

Report this harness’s credential status without resolving or staging it.

Returns:

AuthStatus: Where the credential lives, whether it is there, and when it expires.

class open_atp.harness.kimi.KimiHarness(*, model: str = 'kimi-code/k3', effort: str = 'high', home_dir: Path | None = None)[source]¶

Bases: Harness

Moonshot Kimi Code CLI, authenticated by its file-stored OAuth credential.

Parameters:

modelstr, default “kimi-code/k3”: Model alias the agent runs (a config.toml alias). The default is K3.
effortstr, default “high”: Reasoning-effort level, one of "low", "high", or "max".
home_dirpathlib.Path, optional: The Kimi Code data directory to stage credentials from. None (default) uses $KIMI_CODE_HOME or ~/.kimi-code (from kimi login); resolution fails if absent.

Examples

Constructing the harness resolves its defaults:

>>> from open_atp.harness import KimiHarness
>>> harness = KimiHarness()
>>> harness.name
'kimi'
>>> harness.model
'kimi-code/k3'

auth_status() → AuthStatus[source]¶

Report this harness’s credential status without resolving or staging it.

Returns:

AuthStatus: Where the credential lives, whether it is there, and when it expires.

Pricing¶

class open_atp.harness.cost.ModelPrice(input: float, output: float, cached_input: float | None = None)[source]¶

A model’s USD price per million tokens.

Parameters:

inputfloat: Uncached (cache-miss) input tokens.
outputfloat: Output tokens.
cached_inputfloat, optional: Cache-hit input tokens. None (default) when the provider publishes no cached rate, in which case cached tokens are billed at input.

open_atp.harness.cost.compute_cost_usd(model: str, input_tokens: int, output_tokens: int, cached_input_tokens: int = 0) → float | None[source]¶

Estimate the USD cost of a run from token counts.

Parameters:

modelstr: Model id to look up in COST_PER_MTOK.
input_tokensint: Total input (prompt) tokens.
output_tokensint: Total output (completion) tokens.
cached_input_tokensint, default 0: The cache-hit subset of input_tokens, if the harness reports it. If unreported, the whole input is billed as uncached, an upper bound.

Returns:

float or None: Estimated USD cost, or None when model is absent from COST_PER_MTOK.

open_atp.harness.cost.COST_PER_MTOK: dict[str, ModelPrice] = {'claude-fable-5': ModelPrice(input=10.0, output=50.0, cached_input=1.0), 'claude-haiku-4-5': ModelPrice(input=1.0, output=5.0, cached_input=0.1), 'claude-opus-4-6': ModelPrice(input=5.0, output=25.0, cached_input=0.5), 'claude-opus-4-7': ModelPrice(input=5.0, output=25.0, cached_input=0.5), 'claude-opus-4-8': ModelPrice(input=5.0, output=25.0, cached_input=0.5), 'claude-sonnet-4-5': ModelPrice(input=3.0, output=15.0, cached_input=0.3), 'claude-sonnet-4-6': ModelPrice(input=3.0, output=15.0, cached_input=0.3), 'deepseek-v4-flash': ModelPrice(input=0.14, output=0.28, cached_input=0.0028), 'deepseek-v4-pro': ModelPrice(input=0.435, output=0.87, cached_input=0.003625), 'gemini-3.1-pro-preview': ModelPrice(input=2.0, output=12.0, cached_input=0.2), 'gpt-4.1': ModelPrice(input=2.0, output=8.0, cached_input=None), 'gpt-4o': ModelPrice(input=2.5, output=10.0, cached_input=None), 'gpt-4o-mini': ModelPrice(input=0.15, output=0.6, cached_input=None), 'gpt-5.4': ModelPrice(input=2.5, output=15.0, cached_input=0.25), 'gpt-5.4-mini': ModelPrice(input=0.75, output=4.5, cached_input=0.075), 'gpt-5.4-nano': ModelPrice(input=0.2, output=1.25, cached_input=0.02), 'gpt-5.4-pro': ModelPrice(input=30.0, output=180.0, cached_input=None), 'gpt-5.5': ModelPrice(input=5.0, output=30.0, cached_input=0.5), 'gpt-5.5-pro': ModelPrice(input=30.0, output=180.0, cached_input=None), 'gpt-5.6-luna': ModelPrice(input=1.0, output=6.0, cached_input=0.1), 'gpt-5.6-sol': ModelPrice(input=5.0, output=30.0, cached_input=0.5), 'gpt-5.6-terra': ModelPrice(input=2.5, output=15.0, cached_input=0.25), 'grok-4.3': ModelPrice(input=1.25, output=2.5, cached_input=0.2), 'grok-4.5': ModelPrice(input=2.0, output=6.0, cached_input=0.3), 'grok-build-0.1': ModelPrice(input=1.0, output=2.0, cached_input=0.2), 'kimi-code/k3': ModelPrice(input=3.0, output=15.0, cached_input=0.3), 'kimi-code/kimi-for-coding': ModelPrice(input=0.95, output=4.0, cached_input=0.19), 'kimi-code/kimi-for-coding-highspeed': ModelPrice(input=1.9, output=8.0, cached_input=0.38)}¶: Price per million tokens, keyed by the model id a harness reports.

harness¶

Base¶

Harnesses¶

Pricing¶

`harness`¶