HPC ATP Benchmark

Mix.install([
  {:atp_benchmark_runner, github: "penthooose/atp_benchmark_runner", force: true}
])

1. Configure the environment (setup overlay)

# Fields are pre-filled from `.env` / `.env.example` and from the
# last session (the file only fills blank fields; path fields default to the
# Livebook session temp dir).

setup =
	AtpBenchmarkRunner.prepare_livebook_setup(
		env_file: Path.expand("../.env", __DIR__),
		fallback_env_file: Path.expand("../.env.example", __DIR__),
		submit_label: "Setup"
	)

env_map = setup.env_map
env_file = setup.env_file
ssh_key_path = setup.ssh_key_path

2. Bootstrap the HPC session

# Use the preconfigured values from the §1 setup overlay directly, so this works
# on any machine / server without relying on a local `.env`. `env_file` (the
# overlay-written temp .env) supplies the remaining session settings (work/vault
# dirs, steady connection, retry, ...).
boot =
	HpcConnect.bootstrap(
		mode: :local,
		env_file: env_file,
		cluster: env_map["HPC_CONNECT_CLUSTER"] || "helma",
		username: env_map["HPC_CONNECT_USERNAME"],
		key_path: env_map["HPC_CONNECT_IDENTITY_FILE"]
	)

session = boot.session

IO.inspect(
	%{

		cluster: session.cluster.name,
		ssh_alias: session.ssh_alias,
		username: session.username,
		work_dir: session.work_dir,
		vault_dir: session.vault_dir
	},
	label: "Resolved HPC session"
)

HpcConnect.bootstrap/1 installs the remote helper scripts and uploads the generic bundled hpc_connect def-file set.

2b. Download problems

selected_problems = [
  "ALG001+0.p",
  "GRP001-0.p",
  "GRP002+0.p",
  "GRP003+0.p",
  "GRP004+0.p",
  "GRP720+1.p",
  "GRP752-1.p",
  "LAT001+0.p",
  "PUZ001+0.p",
  "REL001+0.p",
  "SET001^0.p",
  "SMT001+0.p",
  "SYN000+0.p",
  "THF001+0.p"
]

{downloaded, download_warnings} =
	AtpBenchmarkRunner.download_tptp_problems!(selected_problems, force: false)

IO.puts("Downloaded #{length(downloaded)} problem(s):")
Enum.each(downloaded, fn problem ->
	IO.puts("  ✅ #{problem.name}  status=#{problem.expected_status}  rating=#{problem.rating}")
	IO.puts("     -> #{problem.path}")
end)

if download_warnings == [] do
	IO.puts("No warnings.")
else
	IO.puts("Warnings (#{length(download_warnings)}):")
	Enum.each(download_warnings, fn w ->
		IO.puts("  ⚠️  #{w.name}: #{inspect(w.reason)}")
	end)
end

3. Pick provers and select problems from TPTP_ROOT

selected_provers = [
   :tableaux,
   :shot_tx,
   :eprover,
   :vampire,
   :cvc5,
   :zipperposition,
   :leo3,
   :leo2,
   :lash
]

# Use the configured TPTP_ROOT (C:\tmp\tptp from .env).
# Problems can be selected by name; they resolve from the TPTP archive,
# bundled examples, or a livebook cache automatically.
problems =
	AtpBenchmarkRunner.select_problems(
		names: selected_problems,
		limit: 10
	)

IO.inspect(Enum.map(problems, &%{name: &1.name, path: &1.path}), label: "Problems")

4. Inspect the remote image prep plan

prover_structs = Enum.map(selected_provers, &AtpBenchmarkRunner.Prover.builtin!/1)

image_plan =
	AtpBenchmarkRunner.image_build_plan(
		session,
		prover_structs
	)

IO.inspect(image_plan, label: "Remote image build plan", limit: :infinity)

5. Optionally prepare ATP prover images now

prepare_images_now? = false

# One call: uploads each prover's apptainer.def and builds the remote .sif.
# force: true rebuilds even existing .sif images
# (apptainer build --force --ignore-fakeroot-command).
# build_on_login_node: true (default) builds directly on the login node;
# set false to chain all builds into a single compute-node sbatch job.
if prepare_images_now? do
	built_images =
		AtpBenchmarkRunner.build_prover_images!(
			session,
			prover_structs,
			install_scripts: false,
			force: false,
			build_on_login_node: true
		)

	IO.inspect(built_images, label: "Built ATP prover images")
else
	:ok
end

6. Bootstrap an HPC benchmark plan

Select the execution mode and resource strategy:

node_size controls how many CPUs of a node are requested. Auto-detected per cluster (Helma CPU: 384/192, Fritz: 72/36, spr*: 104/52).

# Single-node sequential, full node (default)
plan =
	AtpBenchmarkRunner.bootstrap(
		session,
		selected_provers,
		problems,
		mode: :hpc,
		hpc_mode: :single_node,
		single_node_mode: :sequential,
		timeout_seconds: 30,
		wait_for_completion: false,
		prepare_images: false,
    extended_debug: false
	)

IO.inspect(plan, label: "Single-node sequential, full node (default)", limit: :infinity)

The other execution modes (:half node, single-node parallel, multi-node) are documented with ready-to-paste examples in the cheat sheet and the manual.

If you want the runner itself to prepare missing prover images immediately before launch, set prepare_images: true in the benchmark plan above.

7. Submit the run and collect results

IO.puts("Starting HPC benchmark run...")
IO.puts("  Run ID: #{plan.id}")
IO.puts("  Provers: #{inspect(Enum.map(selected_provers, & &1))}")
IO.puts("  Problems: #{length(problems)}")
IO.puts("  Plan mode: #{plan.metadata[:hpc][:hpc_mode]}/#{plan.metadata[:hpc][:single_node_mode]}")
IO.puts("  Partition: #{plan.metadata[:hpc][:partition]}")
IO.puts("")

results = AtpBenchmarkRunner.run_benchmark(plan)

7b. Collect only last run's results

The run's ID is printed when section 7 starts (Run ID: ...), and the run manifest (<run_id>.run.json) is persisted in the store the moment the run starts. So the last run is always identifiable, even if the notebook cell crashes or the SSH connection dies while fetching.

# results =
# 	try do
# 		AtpBenchmarkRunner.collect_last_hpc_results!(session)
# 	rescue
# 		e ->
# 			IO.puts("Resume failed: #{Exception.message(e)}")
# 			[]
# 	end

# IO.puts("Fetched #{length(results)} results for the last submitted run")

To resume a specific run instead of the newest one:

# results = AtpBenchmarkRunner.collect_hpc_results!(session, "run_20260827_155553_1028")

8. Results table

results = binding()[:results] || []

AtpBenchmarkRunner.results_table(results, memory: true)

9. Aggregated report

report = AtpBenchmarkRunner.report(results, plan)

IO.puts("Run ID: #{report.run_id}")
IO.puts("Generated: #{report.generated_at}")
IO.puts("")
AtpBenchmarkRunner.print_report_markdown(report)

Per-prover breakdown

IO.puts("| Prover | Total | Solved | Failed | Solve Rate |")
IO.puts("|--------|------:|-------:|-------:|-----------:|")

Enum.each(report.by_prover, fn p ->
	IO.puts("| #{p.prover} | #{p.total} | #{p.solved} | #{p.failed} | #{Float.round(p.solve_rate * 100, 1)}% |")
end)

Per-problem comparison

AtpBenchmarkRunner.print_per_problem(report)

Full explained results

IO.puts("Full explained results with proofs:\n")
AtpBenchmarkRunner.print_explain_full(results)

Filtered results by prover

target_prover = :vampire

AtpBenchmarkRunner.print_verbose_report(results, prover: target_prover, solved_only: true)

View a specific proof

target_prover = :vampire
target_problem = "GRP001-0"

AtpBenchmarkRunner.show_proof(results, target_prover, target_problem)

Raw prover debug output

AtpBenchmarkRunner.print_raw_output(results, :vampire, "GRP001-0")

10. Visualizations

results = binding()[:results] || []
results = if results == [], do: VisualizeSample.results(), else: results

target =
  Enum.find(results, fn r ->
    r.prover == :eprover and AtpBenchmarkRunner.Result.solved?(r)
  end)

if target, do: AtpBenchmarkRunner.visualize(target), else: "No solved eprover result found."

10b. Proof dependency graph

For provers that emit machine-readable TPTP clause refutations (E, Vampire) this renders the proof as a dependency DAG. Other provers fall back to a pipeline diagram.

results = binding()[:results] || []
results = if results == [], do: VisualizeSample.results(), else: results

target =
  Enum.find(results, fn r ->
    match?({:ok, _}, AtpBenchmarkRunner.Visualize.Proof.parse(r))
  end)

if target, do: AtpBenchmarkRunner.visualize_proof(target), else: "No parseable proof found."

10c. Run summary (status + wall-time chart)

results = binding()[:results] || []
results = if results == [], do: VisualizeSample.results(), else: results

case AtpBenchmarkRunner.visualize(results) do
  :ok -> :ok
  markdowns -> Enum.each(markdowns, &IO.puts/1)
end

10d. Report scoreboard

results = binding()[:results] || []
results = if results == [], do: VisualizeSample.results(), else: results

report = AtpBenchmarkRunner.report(results)
AtpBenchmarkRunner.visualize(report)

10e. Raw Mermaid source (copy into a markdown cell)

results = binding()[:results] || []
results = if results == [], do: VisualizeSample.results(), else: results

target =
  Enum.find(results, fn r ->
    match?({:ok, _}, AtpBenchmarkRunner.Visualize.Proof.parse(r))
  end)

if target do
  IO.puts(AtpBenchmarkRunner.Visualize.markdown(AtpBenchmarkRunner.Visualize.proof(target)))
else
  IO.puts("No parseable proof; nothing to show.")
end

11. Persist results to local store

run = AtpBenchmarkRunner.new_run(
	title: "HPC smoke test",
	problems: problems,
	provers: Enum.map(selected_provers, & &1),
	walltime: "00:30:00",
	problem_timeout_seconds: 30
)

paths = AtpBenchmarkRunner.persist_run!(run, results, report)
IO.puts("Run saved: #{paths.run_path}")
IO.puts("Results saved: #{paths.results_path}")
IO.puts("Report saved: #{paths.report_path}")

12. Cross-run comparison & statistics

Compares all stored runs at once. Results stay in JSON files under the store dir (no database yet, as columns may still change during development).

Ranking counts Timeout/GaveUp as real attempts (never excluded). By default InputError/TypeError/UnsupportedLogic/Error are treated as non-attempts (prover not applicable to that problem); override with :exclude_statuses (pass [] to include them).

# One call: load all runs, aggregate, and print the comparison tables.
stats = AtpBenchmarkRunner.compare_all_runs()

# Livebook-friendly panel (Kino.Markdown when available).
AtpBenchmarkRunner.MultiRun.panel(stats)

# Save the comparison as a Markdown file.
path =
  AtpBenchmarkRunner.save_multi_run_stats!(
    Path.join(AtpBenchmarkRunner.store_dir(), "comparison.md")
  )

IO.puts("Comparison saved: #{path}")

# Same-problem-set ranking + error-status control.
# only_common_problems: rank only on problems every prover ran.
# exclude_statuses: [] keeps input/type errors as real attempts.
stats_common =
  AtpBenchmarkRunner.multi_run_stats(nil,
    only_common_problems: true,
    exclude_statuses: []
  )

AtpBenchmarkRunner.MultiRun.print(stats_common)

13. Clean up the temporary SSH key (Optional)

Deletes the SSH key that was uploaded in §1 (stored temporarily), if any. A persistent ~/.ssh key is never touched.

AtpBenchmarkRunner.cleanup_livebook_setup()