use std::collections::BTreeMap;
use std::fmt;
use serde::{Deserialize, Serialize};
use crate::binding::{BindingRegistry, ImplStatus};
use crate::discharge::summary::{self, Discharged, L4Source};
use crate::schema::Contract;
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub enum ProofLevel {
L1,
L2,
L3,
L4,
L5,
}
impl fmt::Display for ProofLevel {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let s = match self {
Self::L1 => "L1",
Self::L2 => "L2",
Self::L3 => "L3",
Self::L4 => "L4",
Self::L5 => "L5",
};
write!(f, "{s}")
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ContractProofStatus {
pub stem: String,
pub proof_level: ProofLevel,
pub obligations: u32,
#[serde(default)]
pub not_applicable: u32,
pub falsification_tests: u32,
pub kani_harnesses: u32,
pub lean_proved: u32,
pub lean_grounded: u32,
pub l4_self_declared: bool,
pub bindings_implemented: u32,
pub bindings_total: u32,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KernelClassSummary {
pub label: String,
pub description: String,
pub contract_stems: Vec<String>,
pub min_proof_level: ProofLevel,
pub all_bound: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProofStatusReport {
pub schema_version: String,
pub timestamp: String,
pub contracts: Vec<ContractProofStatus>,
pub kernel_classes: Vec<KernelClassSummary>,
pub l4_self_declared_excluded: bool,
#[serde(default)]
pub l4_source: L4Source,
#[serde(default)]
pub l4_withheld: Option<String>,
pub totals: ProofStatusTotals,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProofStatusTotals {
pub contracts: u32,
pub obligations: u32,
#[serde(default)]
pub not_applicable: u32,
pub falsification_tests: u32,
pub kani_harnesses: u32,
pub lean_proved: u32,
pub lean_grounded: u32,
pub l4_self_declared: u32,
pub bindings_implemented: u32,
pub bindings_total: u32,
}
fn kernel_class_map() -> Vec<(&'static str, &'static str, &'static [&'static str])> {
vec![
(
"A",
"GQA+RMSNorm+SiLU+SwiGLU+RoPE",
&[
"rmsnorm-kernel-v1",
"silu-kernel-v1",
"swiglu-kernel-v1",
"rope-kernel-v1",
"gqa-kernel-v1",
"softmax-kernel-v1",
"matmul-kernel-v1",
],
),
(
"B",
"MHA+LayerNorm+GELU+AbsPos",
&[
"layernorm-kernel-v1",
"gelu-kernel-v1",
"attention-kernel-v1",
"softmax-kernel-v1",
"matmul-kernel-v1",
"absolute-position-v1",
],
),
(
"C",
"MHA+LayerNorm+GELU+ALiBi",
&[
"layernorm-kernel-v1",
"gelu-kernel-v1",
"attention-kernel-v1",
"softmax-kernel-v1",
"alibi-kernel-v1",
"matmul-kernel-v1",
],
),
(
"D",
"LayerNorm+GELU+SiLU+GQA",
&[
"layernorm-kernel-v1",
"gelu-kernel-v1",
"silu-kernel-v1",
"gqa-kernel-v1",
"softmax-kernel-v1",
"matmul-kernel-v1",
],
),
(
"E",
"RMSNorm+SwiGLU+GQA",
&[
"rmsnorm-kernel-v1",
"swiglu-kernel-v1",
"gqa-kernel-v1",
"softmax-kernel-v1",
"matmul-kernel-v1",
],
),
]
}
#[must_use]
pub fn is_lean_proved_with_grounding(contract: &Contract, grounded: u32) -> bool {
let total = contract.proof_obligations.len() as u32;
if total == 0 {
return false;
}
let not_applicable = contract
.verification_summary
.as_ref()
.map_or(0, |vs| vs.l4_not_applicable);
grounded > 0 && grounded + not_applicable >= total
}
#[must_use]
pub fn is_l4_self_declared(contract: &Contract) -> bool {
is_l4_self_declared_with_grounding(contract, count_lean_theorems_for_contract(contract))
}
#[must_use]
pub fn is_l4_self_declared_with_grounding(contract: &Contract, grounded: u32) -> bool {
let total = contract.proof_obligations.len() as u32;
if total == 0 {
return false;
}
let Some(vs) = contract.verification_summary.as_ref() else {
return false;
};
let claim_covers = vs.l4_lean_proved > 0 && vs.l4_lean_proved + vs.l4_not_applicable >= total;
claim_covers && !is_lean_proved_with_grounding(contract, grounded)
}
#[must_use]
#[allow(clippy::cast_possible_truncation)]
pub fn count_not_applicable(contract: &Contract) -> u32 {
contract
.proof_obligations
.iter()
.filter(|ob| ob.is_not_applicable())
.count() as u32
}
fn is_fully_bound(binding_status: Option<(u32, u32)>) -> bool {
binding_status.is_some_and(|(implemented, total)| total > 0 && implemented == total)
}
#[allow(clippy::cast_possible_truncation)]
pub fn compute_proof_level(contract: &Contract, binding_status: Option<(u32, u32)>) -> ProofLevel {
compute_proof_level_with_grounding(
contract,
binding_status,
count_lean_theorems_for_contract(contract),
)
}
#[allow(clippy::cast_possible_truncation)]
#[must_use]
pub fn compute_proof_level_with_grounding(
contract: &Contract,
binding_status: Option<(u32, u32)>,
grounded: u32,
) -> ProofLevel {
let total_obligations = contract.proof_obligations.len() as u32;
let ft_count = contract.falsification_tests.len() as u32;
let kani_count = contract.kani_harnesses.len() as u32;
if is_lean_proved_with_grounding(contract, grounded) {
return if is_fully_bound(binding_status) {
ProofLevel::L5
} else {
ProofLevel::L4
};
}
let has_tests = ft_count >= total_obligations && total_obligations > 0;
if kani_count > 0 && has_tests {
return ProofLevel::L3;
}
if has_tests {
return ProofLevel::L2;
}
ProofLevel::L1
}
pub(crate) const LEAN_THEOREM_BASES: &[&str] = &[
"crates/aprender-contracts-staging/lean",
"lean",
"../provable-contracts/lean",
];
fn insert_name_forms(names: &mut std::collections::HashSet<String>, label: &str) {
names.insert(format!("Theorems.{label}"));
names.insert(label.to_string());
names.insert(label.to_lowercase());
}
pub(crate) fn camel_case(snake: &str) -> String {
snake
.split('_')
.map(|s| {
let mut c = s.chars();
match c.next() {
None => String::new(),
Some(f) => f.to_uppercase().chain(c).collect(),
}
})
.collect()
}
pub(crate) fn first_camel_word(camel: &str) -> String {
camel
.chars()
.enumerate()
.take_while(|(i, c)| *i == 0 || !c.is_uppercase())
.map(|(_, c)| c)
.collect()
}
fn insert_theorem_names_from_content(names: &mut std::collections::HashSet<String>, content: &str) {
for line in content.lines() {
let Some(pos) = line.find("theorem ") else {
continue;
};
let tname: String = line[pos + 8..]
.chars()
.take_while(|c| c.is_alphanumeric() || *c == '_')
.collect();
if tname.is_empty() {
continue;
}
let camel = camel_case(&tname);
names.insert(format!("Theorems.{camel}"));
names.insert(camel.clone());
let first_word = first_camel_word(&camel);
if first_word.len() >= 3 {
names.insert(format!("Theorems.{first_word}"));
names.insert(first_word);
}
}
}
fn is_lean_ident_byte(c: u8) -> bool {
c.is_ascii_alphanumeric() || c == b'_' || c == b'\''
}
pub(crate) fn lean_has_sorry(content: &str) -> bool {
let b = content.as_bytes();
let mut i = 0;
let mut depth = 0usize;
while i < b.len() {
if depth > 0 {
(i, depth) = step_block_comment(b, i, depth);
continue;
}
match scan_code_at(b, i) {
LeanScan::Next(j) => i = j,
LeanScan::OpenBlock(j) => {
depth = 1;
i = j;
}
LeanScan::Sorry => return true,
LeanScan::EndsInLineComment => return false,
}
}
depth > 0
}
enum LeanScan {
Next(usize),
OpenBlock(usize),
Sorry,
EndsInLineComment,
}
fn sorry_token_at(b: &[u8], i: usize) -> bool {
b[i..].starts_with(b"sorry")
&& (i == 0 || !is_lean_ident_byte(b[i - 1]))
&& b.get(i + 5).is_none_or(|&c| !is_lean_ident_byte(c))
}
fn step_block_comment(b: &[u8], i: usize, depth: usize) -> (usize, usize) {
if b[i..].starts_with(b"/-") {
(i + 2, depth + 1)
} else if b[i..].starts_with(b"-/") {
(i + 2, depth - 1)
} else {
(i + 1, depth)
}
}
fn scan_code_at(b: &[u8], i: usize) -> LeanScan {
let prev_ident = i > 0 && is_lean_ident_byte(b[i - 1]);
if b[i..].starts_with(b"--") {
return match b[i..].iter().position(|&c| c == b'\n') {
Some(p) => LeanScan::Next(i + p + 1),
None => LeanScan::EndsInLineComment,
};
}
if b[i..].starts_with(b"/-") {
return LeanScan::OpenBlock(i + 2);
}
if b[i] == b'r' && !prev_ident && matches!(b.get(i + 1), Some(b'"' | b'#')) {
return scan_raw_string(b, i);
}
if b[i] == b'"' {
return scan_string(b, i);
}
if b[i] == b'\'' && !prev_ident {
return LeanScan::Next(skip_char_literal(b, i));
}
if sorry_token_at(b, i) {
LeanScan::Sorry
} else {
LeanScan::Next(i + 1)
}
}
fn scan_literal_body(b: &[u8], from: usize, to: usize, resume: usize) -> LeanScan {
if (from..to).any(|k| sorry_token_at(b, k)) {
LeanScan::Sorry
} else {
LeanScan::Next(resume)
}
}
fn scan_raw_string(b: &[u8], i: usize) -> LeanScan {
let hashes = b[i + 1..].iter().take_while(|&&c| c == b'#').count();
let open = i + 1 + hashes;
if b.get(open) != Some(&b'"') {
return LeanScan::Next(i + 1);
}
let mut close = vec![b'"'];
close.extend(std::iter::repeat_n(b'#', hashes));
let Some(p) = b[open + 1..]
.windows(close.len())
.position(|w| w == close.as_slice())
else {
return LeanScan::Sorry;
};
let end = open + 1 + p;
scan_literal_body(b, open + 1, end, end + close.len())
}
fn scan_string(b: &[u8], i: usize) -> LeanScan {
let mut j = i + 1;
loop {
match b.get(j) {
None => return LeanScan::Sorry,
Some(b'\\') => j += 2,
Some(b'"') => break,
Some(_) => j += 1,
}
}
scan_literal_body(b, i + 1, j, j + 1)
}
fn skip_char_literal(b: &[u8], i: usize) -> usize {
let from = if b.get(i + 1) == Some(&b'\\') {
i + 3
} else {
i + 2
};
match b
.get(from..b.len().min(i + 7))
.and_then(|s| s.iter().position(|&c| c == b'\''))
{
Some(p) => from + p + 1,
None => i + 1,
}
}
fn insert_domain_theorems(names: &mut std::collections::HashSet<String>, domain: &std::path::Path) {
let domain_name = domain
.file_name()
.unwrap_or_default()
.to_string_lossy()
.to_string();
let Ok(files) = std::fs::read_dir(domain) else {
return;
};
for file in files.flatten() {
let path = file.path();
if path.extension().is_none_or(|e| e != "lean") {
continue;
}
let Ok(content) = std::fs::read_to_string(&path) else {
continue;
};
if lean_has_sorry(&content) {
continue;
}
let stem = path
.file_stem()
.unwrap_or_default()
.to_string_lossy()
.to_string();
names.insert(format!("Theorems.{domain_name}.{stem}"));
insert_name_forms(names, &domain_name);
insert_name_forms(names, &stem);
insert_theorem_names_from_content(names, &content);
}
}
fn scan_theorem_base(base: &str) -> std::collections::HashSet<String> {
let mut names = std::collections::HashSet::new();
let search_dir = std::path::Path::new(base).join("ProvableContracts/Theorems");
if !search_dir.exists() {
return names;
}
let Ok(domains) = std::fs::read_dir(&search_dir) else {
return names;
};
for domain_entry in domains.flatten() {
let path = domain_entry.path();
if path.is_dir() {
insert_domain_theorems(&mut names, &path);
}
}
names
}
fn lean_theorem_names() -> &'static std::collections::HashSet<String> {
use std::sync::OnceLock;
static CACHE: OnceLock<std::collections::HashSet<String>> = OnceLock::new();
CACHE.get_or_init(|| {
for base in LEAN_THEOREM_BASES {
let names = scan_theorem_base(base);
if !names.is_empty() {
return names;
}
}
std::collections::HashSet::new()
})
}
struct Grounding {
source: L4Source,
discharged: Discharged,
}
fn grounding() -> &'static Grounding {
use std::sync::OnceLock;
static CACHE: OnceLock<Grounding> = OnceLock::new();
CACHE.get_or_init(|| {
let bases: Vec<&std::path::Path> = LEAN_THEOREM_BASES
.iter()
.map(std::path::Path::new)
.collect();
grounding_from(&bases, summary::current_tree_sha)
})
}
fn grounding_from(
bases: &[&std::path::Path],
tree_sha: impl Fn(&std::path::Path) -> Option<String>,
) -> Grounding {
for &dir in bases {
let path = summary::summary_path(dir);
if !dir.is_dir() || !path.is_file() {
continue;
}
let mut discharged = match summary::load(&path) {
Ok(s) => summary::discharged(&s, tree_sha(dir).as_deref()),
Err(why) => Discharged {
withheld: Some(why),
..Discharged::default()
},
};
refine(dir, &mut discharged);
return Grounding {
source: L4Source::Discharge,
discharged,
};
}
Grounding {
source: L4Source::SelfDeclared,
discharged: Discharged::default(),
}
}
fn refine(lean_dir: &std::path::Path, discharged: &mut Discharged) {
use crate::discharge::refinement;
if discharged.theorems.is_empty() {
return;
}
let l4 = match refinement::load(lean_dir) {
Ok(rec) => {
let modules = discharged.module_of.values().cloned().collect();
let root = workspace_root(lean_dir);
refinement::l4_modules(&refinement::judge(
&rec.models,
&modules,
refinement::workspace_resolver(&root),
))
}
Err(_) => std::collections::BTreeSet::new(),
};
discharged.require_refinement(&l4);
}
fn workspace_root(dir: &std::path::Path) -> std::path::PathBuf {
let abs = std::fs::canonicalize(dir).unwrap_or_else(|_| dir.to_path_buf());
abs.ancestors()
.find(|a| {
std::fs::read_to_string(a.join("Cargo.toml"))
.is_ok_and(|t| t.lines().any(|l| l.trim() == "[workspace]"))
})
.map_or_else(
|| std::path::PathBuf::from("."),
std::path::Path::to_path_buf,
)
}
#[must_use]
pub fn discharge_grounding() -> &'static Discharged {
&grounding().discharged
}
#[must_use]
pub fn count_discharged_for_contract(contract: &Contract, discharged: &Discharged) -> u32 {
let n = contract
.equations
.values()
.filter_map(|eq| eq.lean_theorem.as_deref())
.filter(|t| discharged.grants(t))
.count();
u32::try_from(n).unwrap_or(u32::MAX)
}
fn count_lean_theorems_for_contract(contract: &Contract) -> u32 {
let g = grounding();
match g.source {
L4Source::Discharge => count_discharged_for_contract(contract, &g.discharged),
L4Source::SelfDeclared => count_scanned_theorems_for_contract(contract),
}
}
fn count_scanned_theorems_for_contract(contract: &Contract) -> u32 {
let theorems = lean_theorem_names();
let mut count = 0u32;
for eq in contract.equations.values() {
if let Some(ref theorem_ref) = eq.lean_theorem {
let name = theorem_ref.trim().trim_matches('"');
if theorems.contains(name)
|| theorems.contains(name.strip_prefix("Theorems.").unwrap_or(name))
|| theorems.contains(&name.to_lowercase())
{
count += 1;
}
}
}
count
}
#[allow(clippy::cast_possible_truncation)]
pub fn proof_status_report(
contracts: &[(String, &Contract)],
binding: Option<&BindingRegistry>,
include_classes: bool,
) -> ProofStatusReport {
let mut statuses = Vec::new();
let mut totals = ProofStatusTotals {
contracts: contracts.len() as u32,
obligations: 0,
not_applicable: 0,
falsification_tests: 0,
kani_harnesses: 0,
lean_proved: 0,
lean_grounded: 0,
l4_self_declared: 0,
bindings_implemented: 0,
bindings_total: 0,
};
for (stem, contract) in contracts {
let contract_file = format!("{stem}.yaml");
let obligations = contract.proof_obligations.len() as u32;
let not_applicable = count_not_applicable(contract);
let ft_count = contract.falsification_tests.len() as u32;
let kani_count = contract.kani_harnesses.len() as u32;
let lean_proved = contract
.verification_summary
.as_ref()
.map_or(0, |vs| vs.l4_lean_proved);
let lean_grounded = count_lean_theorems_for_contract(contract);
let lean_proved = if lean_proved == 0 {
lean_grounded
} else {
lean_proved
};
let l4_self_declared = is_l4_self_declared_with_grounding(contract, lean_grounded);
let (b_impl, b_total) = if let Some(reg) = binding {
count_bindings(&contract_file, contract, reg)
} else {
(0, contract.equations.len() as u32)
};
let binding_status = if binding.is_some() {
Some((b_impl, b_total))
} else {
None
};
let proof_level =
compute_proof_level_with_grounding(contract, binding_status, lean_grounded);
totals.obligations += obligations;
totals.not_applicable += not_applicable;
totals.falsification_tests += ft_count;
totals.kani_harnesses += kani_count;
totals.lean_proved += lean_proved;
totals.lean_grounded += lean_grounded;
totals.l4_self_declared += u32::from(l4_self_declared);
totals.bindings_implemented += b_impl;
totals.bindings_total += b_total;
statuses.push(ContractProofStatus {
stem: stem.clone(),
proof_level,
obligations,
not_applicable,
falsification_tests: ft_count,
kani_harnesses: kani_count,
lean_proved,
lean_grounded,
l4_self_declared,
bindings_implemented: b_impl,
bindings_total: b_total,
});
}
let kernel_classes = if include_classes {
build_kernel_classes(&statuses)
} else {
Vec::new()
};
let timestamp = current_timestamp();
ProofStatusReport {
schema_version: "1.0.0".to_string(),
l4_self_declared_excluded: true,
l4_source: grounding().source,
l4_withheld: grounding().discharged.withheld.clone(),
timestamp,
contracts: statuses,
kernel_classes,
totals,
}
}
pub fn format_text(report: &ProofStatusReport) -> String {
let mut out = String::new();
out.push_str(&format!(
"Proof Status ({} contracts)\n\n",
report.totals.contracts
));
out.push_str(&format!(
" {:<35} {:>5} {:>6} {:>5} {:>4} {:>4} {:>9} {:>13}\n",
"Contract", "Level", "Obligs", "Tests", "Kani", "Lean", "Bindings", "L4 evidence"
));
out.push_str(&format!(" {}\n", "─".repeat(86)));
for c in &report.contracts {
let l4_evidence = if c.l4_self_declared {
"self-declared"
} else if c.lean_grounded > 0 {
"grounded"
} else {
"-"
};
out.push_str(&format!(
" {:<35} {:>5} {:>6} {:>5} {:>4} {:>4} {:>4}/{:<4} {:>13}\n",
truncate(&c.stem, 35),
c.proof_level,
c.obligations,
c.falsification_tests,
c.kani_harnesses,
c.lean_proved,
c.bindings_implemented,
c.bindings_total,
l4_evidence,
));
}
if !report.kernel_classes.is_empty() {
out.push_str("\nKernel Classes:\n");
for kc in &report.kernel_classes {
let bound_str = if kc.all_bound { "all bound" } else { "gaps" };
out.push_str(&format!(
" {} ({}): min={}, {} contracts, {}\n",
kc.label,
kc.description,
kc.min_proof_level,
kc.contract_stems.len(),
bound_str,
));
}
}
out.push_str(&format!(
"\nTotals: {} obligations ({} N/A, never counted as proved), {} tests, {} kani, {} lean claimed ({} grounded), {}/{} bound\n\
L4 evidence: {} contract(s) self-declared and excluded from L4 (ONT-2a andon); grounded means a \
theorem the equation names that a green, fresh discharge-summary.json lists (EV-8b), or with no summary \
a sorry-free in-tree Lean theorem\n",
report.totals.obligations,
report.totals.not_applicable,
report.totals.falsification_tests,
report.totals.kani_harnesses,
report.totals.lean_proved,
report.totals.lean_grounded,
report.totals.bindings_implemented,
report.totals.bindings_total,
report.totals.l4_self_declared,
));
let source = match report.l4_source {
L4Source::Discharge => "discharge (discharge-summary.json)",
L4Source::SelfDeclared => "self-declared (no discharge-summary.json; in-tree .lean scan)",
};
out.push_str(&format!("L4 source: {source}\n"));
if let Some(why) = &report.l4_withheld {
out.push_str(&format!("L4 withheld: {why}\n"));
}
out
}
#[allow(clippy::cast_possible_truncation)]
pub(crate) fn count_bindings(
contract_file: &str,
contract: &Contract,
binding: &BindingRegistry,
) -> (u32, u32) {
let total = contract.equations.len() as u32;
let implemented = binding
.bindings_for(contract_file)
.iter()
.filter(|b| b.status == ImplStatus::Implemented)
.count() as u32;
(implemented, total)
}
fn build_kernel_classes(statuses: &[ContractProofStatus]) -> Vec<KernelClassSummary> {
let status_map: BTreeMap<&str, &ContractProofStatus> =
statuses.iter().map(|s| (s.stem.as_str(), s)).collect();
kernel_class_map()
.into_iter()
.map(|(label, desc, stems)| {
let found_stems: Vec<String> = stems
.iter()
.filter(|s| status_map.contains_key(**s))
.map(|s| (*s).to_string())
.collect();
let min_level = found_stems
.iter()
.filter_map(|s| status_map.get(s.as_str()))
.map(|c| c.proof_level)
.min()
.unwrap_or(ProofLevel::L1);
let all_bound = !found_stems.is_empty()
&& found_stems.iter().all(|s| {
status_map.get(s.as_str()).is_some_and(|c| {
c.bindings_total > 0 && c.bindings_implemented == c.bindings_total
})
});
KernelClassSummary {
label: label.to_string(),
description: desc.to_string(),
contract_stems: found_stems,
min_proof_level: min_level,
all_bound,
}
})
.collect()
}
fn truncate(s: &str, max: usize) -> &str {
if s.len() > max {
&s[..max]
} else {
s
}
}
fn current_timestamp() -> String {
let duration = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default();
format!("{}Z", duration.as_secs())
}
#[cfg(test)]
#[path = "proof_status_tests.rs"]
mod tests;