use std::collections::{HashMap, HashSet};
use crate::context::Context;
use crate::error::{KernelError, KernelResult};
use crate::term::Term;
struct GuardContext {
fix_positions: HashMap<String, usize>,
struct_param: String,
struct_type: String,
smaller_than: HashSet<String>,
struct_param_live: bool,
shadowed_fix: HashSet<String>,
}
pub fn check_termination(ctx: &Context, fix_name: &str, body: &Term) -> KernelResult<()> {
let (struct_param, struct_type, struct_pos, inner) = extract_structural_param(ctx, body)?;
let mut fix_positions = HashMap::new();
fix_positions.insert(fix_name.to_string(), struct_pos);
let guard_ctx = GuardContext {
fix_positions,
struct_param,
struct_type,
smaller_than: HashSet::new(),
struct_param_live: true,
shadowed_fix: HashSet::new(),
};
check_guarded(ctx, &guard_ctx, inner)
}
pub fn check_termination_mutual(ctx: &Context, defs: &[(String, Term)]) -> KernelResult<()> {
if defs.is_empty() {
return Err(KernelError::TerminationViolation {
fix_name: String::new(),
reason: "empty mutual fixpoint block".to_string(),
});
}
let mut fix_positions = HashMap::new();
let mut params: Vec<(String, String)> = Vec::with_capacity(defs.len());
for (name, body) in defs {
let (struct_param, struct_type, struct_pos, _) = extract_structural_param(ctx, body)?;
fix_positions.insert(name.clone(), struct_pos);
params.push((struct_param, struct_type));
}
for (i, (_, body)) in defs.iter().enumerate() {
let (_, _, _, inner) = extract_structural_param(ctx, body)?;
let (struct_param, struct_type) = params[i].clone();
let guard_ctx = GuardContext {
fix_positions: fix_positions.clone(),
struct_param,
struct_type,
smaller_than: HashSet::new(),
struct_param_live: true,
shadowed_fix: HashSet::new(),
};
check_guarded(ctx, &guard_ctx, inner)?;
}
Ok(())
}
fn extract_structural_param<'a>(
ctx: &Context,
body: &'a Term,
) -> KernelResult<(String, String, usize, &'a Term)> {
let mut chain: Vec<(&'a str, &'a Term)> = Vec::new();
let mut cur = body;
while let Term::Lambda { param, param_type, body: inner } = cur {
chain.push((param.as_str(), param_type.as_ref()));
cur = inner;
}
let scrutinee = match cur {
Term::Match { discriminant, .. } => match discriminant.as_ref() {
Term::Var(d) => chain.iter().position(|(n, _)| n == d),
_ => None,
},
_ => None,
};
let pos = scrutinee
.filter(|&p| extract_inductive_name(ctx, chain[p].1).is_some())
.or_else(|| chain.iter().position(|(_, t)| extract_inductive_name(ctx, t).is_some()))
.ok_or_else(|| KernelError::TerminationViolation {
fix_name: String::new(),
reason: "No inductive parameter found for structural recursion".to_string(),
})?;
let name = chain[pos].0.to_string();
let ind = extract_inductive_name(ctx, chain[pos].1).unwrap();
let mut inner = body;
for _ in 0..=pos {
if let Term::Lambda { body: b, .. } = inner {
inner = b;
}
}
Ok((name, ind, pos, inner))
}
fn extract_inductive_name(ctx: &Context, ty: &Term) -> Option<String> {
match ty {
Term::Global(name) if ctx.is_inductive(name) => Some(name.clone()),
Term::App(func, _) => extract_inductive_name(ctx, func),
_ => None,
}
}
fn enter_binder(guard_ctx: &GuardContext, param: &str) -> GuardContext {
let mut child = GuardContext {
fix_positions: guard_ctx.fix_positions.clone(),
struct_param: guard_ctx.struct_param.clone(),
struct_type: guard_ctx.struct_type.clone(),
smaller_than: guard_ctx.smaller_than.clone(),
struct_param_live: guard_ctx.struct_param_live,
shadowed_fix: guard_ctx.shadowed_fix.clone(),
};
child.smaller_than.remove(param);
if param == guard_ctx.struct_param {
child.struct_param_live = false;
}
if guard_ctx.fix_positions.contains_key(param) {
child.shadowed_fix.insert(param.to_string());
}
child
}
fn check_guarded(ctx: &Context, guard_ctx: &GuardContext, term: &Term) -> KernelResult<()> {
match term {
Term::App(func, arg) => {
let mut args: Vec<&Term> = vec![arg.as_ref()];
let mut head = func.as_ref();
while let Term::App(inner_func, inner_arg) = head {
args.push(inner_arg.as_ref());
head = inner_func.as_ref();
}
args.reverse();
if let Term::Var(name) = head {
if !guard_ctx.shadowed_fix.contains(name) {
if let Some(&pos) = guard_ctx.fix_positions.get(name) {
match args.get(pos) {
Some(sarg) => verify_structural_arg_smaller(guard_ctx, sarg)?,
None => {
return Err(KernelError::TerminationViolation {
fix_name: name.clone(),
reason: format!(
"recursive call to '{}' is missing its structural argument (position {})",
name, pos
),
})
}
}
for a in &args {
check_guarded(ctx, guard_ctx, a)?;
}
return Ok(());
}
}
}
check_guarded(ctx, guard_ctx, head)?;
for a in &args {
check_guarded(ctx, guard_ctx, a)?;
}
Ok(())
}
Term::Match {
discriminant,
motive,
cases,
} => {
check_guarded(ctx, guard_ctx, motive)?;
if let Term::Var(disc_name) = discriminant.as_ref() {
if guard_ctx.struct_param_live && disc_name == &guard_ctx.struct_param {
return check_match_cases_guarded(ctx, guard_ctx, cases);
}
}
check_guarded(ctx, guard_ctx, discriminant)?;
for case in cases {
check_guarded(ctx, guard_ctx, case)?;
}
Ok(())
}
Term::Lambda { param, param_type, body } => {
check_guarded(ctx, guard_ctx, param_type)?;
let child = enter_binder(guard_ctx, param);
check_guarded(ctx, &child, body)
}
Term::Pi { param, param_type, body_type } => {
check_guarded(ctx, guard_ctx, param_type)?;
let child = enter_binder(guard_ctx, param);
check_guarded(ctx, &child, body_type)
}
Term::Fix { name, body } => {
let child = enter_binder(guard_ctx, name);
check_guarded(ctx, &child, body)
}
Term::MutualFix { defs, .. } => {
let mut child = enter_binder(guard_ctx, &defs[0].0);
for (n, _) in &defs[1..] {
child = enter_binder(&child, n);
}
for (_, b) in defs {
check_guarded(ctx, &child, b)?;
}
Ok(())
}
Term::Let { name, ty, value, body, .. } => {
check_guarded(ctx, guard_ctx, ty)?;
check_guarded(ctx, guard_ctx, value)?;
let child = enter_binder(guard_ctx, name);
check_guarded(ctx, &child, body)
}
Term::Var(name)
if guard_ctx.fix_positions.contains_key(name)
&& !guard_ctx.shadowed_fix.contains(name) =>
{
Err(KernelError::TerminationViolation {
fix_name: name.clone(),
reason: format!(
"recursive name '{}' occurs as a first-class value, not applied to a structurally-smaller argument",
name
),
})
}
Term::Sort(_) | Term::Var(_) | Term::Global(_) | Term::Lit(_) | Term::Hole
| Term::Const { .. } => Ok(()),
}
}
impl GuardContext {
fn block_name(&self) -> String {
let mut names: Vec<&String> = self.fix_positions.keys().collect();
names.sort();
names.iter().map(|s| s.as_str()).collect::<Vec<_>>().join("/")
}
}
fn verify_structural_arg_smaller(guard_ctx: &GuardContext, first_arg: &Term) -> KernelResult<()> {
let mut head = first_arg;
while let Term::App(f, _) = head {
head = f;
}
match head {
Term::Var(arg_name) if guard_ctx.smaller_than.contains(arg_name) => Ok(()),
Term::Var(arg_name) => Err(KernelError::TerminationViolation {
fix_name: guard_ctx.block_name(),
reason: format!(
"Recursive call with '{}' which is not structurally smaller than '{}'",
arg_name, guard_ctx.struct_param
),
}),
_ => Err(KernelError::TerminationViolation {
fix_name: guard_ctx.block_name(),
reason: "Recursive call whose structural argument is not headed by a \
structurally-smaller variable"
.to_string(),
}),
}
}
fn check_match_cases_guarded(
ctx: &Context,
guard_ctx: &GuardContext,
cases: &[Term],
) -> KernelResult<()> {
let constructors = ctx.get_constructors(&guard_ctx.struct_type);
for (case, (ctor_name, ctor_type)) in cases.iter().zip(constructors.iter()) {
let param_count = count_pi_params(ctor_type);
let (smaller_vars, case_body) = extract_lambda_params(case, param_count);
let mut extended_ctx = GuardContext {
fix_positions: guard_ctx.fix_positions.clone(),
struct_param: guard_ctx.struct_param.clone(),
struct_type: guard_ctx.struct_type.clone(),
smaller_than: guard_ctx.smaller_than.clone(),
struct_param_live: guard_ctx.struct_param_live,
shadowed_fix: guard_ctx.shadowed_fix.clone(),
};
let _ = ctor_name;
for var in &smaller_vars {
extended_ctx.smaller_than.insert(var.clone());
if var == &guard_ctx.struct_param {
extended_ctx.struct_param_live = false;
}
if guard_ctx.fix_positions.contains_key(var) {
extended_ctx.shadowed_fix.insert(var.clone());
}
}
check_guarded(ctx, &extended_ctx, case_body)?;
}
Ok(())
}
fn count_pi_params(ty: &Term) -> usize {
match ty {
Term::Pi { body_type, .. } => 1 + count_pi_params(body_type),
_ => 0,
}
}
fn extract_lambda_params(term: &Term, count: usize) -> (Vec<String>, &Term) {
if count == 0 {
return (Vec::new(), term);
}
match term {
Term::Lambda { param, body, .. } => {
let (mut params, final_body) = extract_lambda_params(body, count - 1);
params.insert(0, param.clone());
(params, final_body)
}
_ => (Vec::new(), term),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::term::Universe;
fn nat_context() -> Context {
let mut ctx = Context::new();
let nat = Term::Global("Nat".to_string());
ctx.add_inductive("Nat", Term::Sort(Universe::Type(0)));
ctx.add_constructor("Zero", "Nat", nat.clone());
ctx.add_constructor(
"Succ",
"Nat",
Term::Pi { param: "_".to_string(), param_type: Box::new(nat.clone()), body_type: Box::new(nat) },
);
ctx.add_inductive("False", Term::Sort(Universe::Prop));
ctx
}
fn nat() -> Term {
Term::Global("Nat".to_string())
}
fn app(f: Term, x: Term) -> Term {
Term::App(Box::new(f), Box::new(x))
}
fn var(n: &str) -> Term {
Term::Var(n.to_string())
}
#[test]
fn recursive_name_smuggled_as_a_first_class_argument_is_rejected() {
let ctx = nat_context();
let nat_to_false = Term::Pi {
param: "_".to_string(),
param_type: Box::new(nat()),
body_type: Box::new(Term::Global("False".to_string())),
};
let zero_case = app(
Term::Lambda {
param: "g".to_string(),
param_type: Box::new(nat_to_false),
body: Box::new(app(var("g"), Term::Global("Zero".to_string()))),
},
var("f"),
);
let succ_case = Term::Lambda {
param: "k".to_string(),
param_type: Box::new(nat()),
body: Box::new(app(var("f"), var("k"))),
};
let body = Term::Lambda {
param: "n".to_string(),
param_type: Box::new(nat()),
body: Box::new(Term::Match {
discriminant: Box::new(var("n")),
motive: Box::new(Term::Lambda {
param: "_".to_string(),
param_type: Box::new(nat()),
body: Box::new(Term::Global("False".to_string())),
}),
cases: vec![zero_case, succ_case],
}),
};
let result = check_termination(&ctx, "f", &body);
assert!(
result.is_err(),
"kernel soundness: a fixpoint that passes its recursive name as a first-class value inhabits \
False and MUST be rejected by the termination guard, but it was accepted"
);
}
#[test]
fn genuine_structural_recursion_still_passes() {
let ctx = nat_context();
let succ_case = Term::Lambda {
param: "k".to_string(),
param_type: Box::new(nat()),
body: Box::new(app(var("f"), var("k"))), };
let body = Term::Lambda {
param: "n".to_string(),
param_type: Box::new(nat()),
body: Box::new(Term::Match {
discriminant: Box::new(var("n")),
motive: Box::new(Term::Lambda {
param: "_".to_string(),
param_type: Box::new(nat()),
body: Box::new(nat()),
}),
cases: vec![Term::Global("Zero".to_string()), succ_case],
}),
};
assert!(check_termination(&ctx, "f", &body).is_ok(), "honest structural recursion must still pass");
}
#[test]
fn recursive_name_returned_bare_from_a_branch_is_rejected() {
let ctx = nat_context();
let succ_case = Term::Lambda {
param: "k".to_string(),
param_type: Box::new(nat()),
body: Box::new(app(var("f"), var("k"))),
};
let body = Term::Lambda {
param: "n".to_string(),
param_type: Box::new(nat()),
body: Box::new(Term::Match {
discriminant: Box::new(var("n")),
motive: Box::new(Term::Lambda {
param: "_".to_string(),
param_type: Box::new(nat()),
body: Box::new(nat()),
}),
cases: vec![var("f"), succ_case], }),
};
assert!(check_termination(&ctx, "f", &body).is_err(), "a branch returning the bare fixpoint must be rejected");
}
#[test]
fn recursive_call_on_a_constructor_applied_to_a_smaller_var_is_rejected() {
let ctx = nat_context();
let succ_case = Term::Lambda {
param: "k".to_string(),
param_type: Box::new(nat()),
body: Box::new(app(var("f"), app(Term::Global("Succ".to_string()), var("k")))),
};
let body = Term::Lambda {
param: "n".to_string(),
param_type: Box::new(nat()),
body: Box::new(Term::Match {
discriminant: Box::new(var("n")),
motive: Box::new(Term::Lambda {
param: "_".to_string(),
param_type: Box::new(nat()),
body: Box::new(nat()),
}),
cases: vec![Term::Global("Zero".to_string()), succ_case],
}),
};
assert!(
check_termination(&ctx, "f", &body).is_err(),
"kernel soundness: a recursive call on `Succ k` (larger than the matched `k`) must be rejected"
);
}
#[test]
fn recursive_call_on_a_non_smaller_variable_applied_is_rejected() {
let ctx = nat_context();
let nat_to_nat = Term::Pi {
param: "_".to_string(),
param_type: Box::new(nat()),
body_type: Box::new(nat()),
};
let succ_case = Term::Lambda {
param: "k".to_string(),
param_type: Box::new(nat()),
body: Box::new(app(app(var("f"), var("h")), app(var("h"), var("k")))),
};
let body = Term::Lambda {
param: "h".to_string(),
param_type: Box::new(nat_to_nat),
body: Box::new(Term::Lambda {
param: "n".to_string(),
param_type: Box::new(nat()),
body: Box::new(Term::Match {
discriminant: Box::new(var("n")),
motive: Box::new(Term::Lambda {
param: "_".to_string(),
param_type: Box::new(nat()),
body: Box::new(nat()),
}),
cases: vec![Term::Global("Zero".to_string()), succ_case],
}),
}),
};
assert!(
check_termination(&ctx, "f", &body).is_err(),
"kernel soundness: recursing on `h k` where `h` is not structurally smaller must be rejected"
);
}
#[test]
fn recursive_call_hidden_in_the_match_motive_is_rejected() {
let ctx = nat_context();
let succ_case = Term::Lambda {
param: "k".to_string(),
param_type: Box::new(nat()),
body: Box::new(app(var("f"), var("k"))),
};
let body = Term::Lambda {
param: "n".to_string(),
param_type: Box::new(nat()),
body: Box::new(Term::Match {
discriminant: Box::new(var("n")),
motive: Box::new(Term::Lambda {
param: "_".to_string(),
param_type: Box::new(nat()),
body: Box::new(app(var("f"), var("n"))),
}),
cases: vec![Term::Global("Zero".to_string()), succ_case],
}),
};
assert!(
check_termination(&ctx, "f", &body).is_err(),
"a recursive call in the match motive must be rejected"
);
}
#[test]
fn recursive_call_hidden_in_a_binder_domain_is_rejected() {
let ctx = nat_context();
let succ_case = Term::Lambda {
param: "k".to_string(),
param_type: Box::new(nat()),
body: Box::new(Term::Lambda {
param: "_".to_string(),
param_type: Box::new(app(var("f"), var("n"))),
body: Box::new(app(var("f"), var("k"))),
}),
};
let body = Term::Lambda {
param: "n".to_string(),
param_type: Box::new(nat()),
body: Box::new(Term::Match {
discriminant: Box::new(var("n")),
motive: Box::new(Term::Lambda {
param: "_".to_string(),
param_type: Box::new(nat()),
body: Box::new(nat()),
}),
cases: vec![Term::Global("Zero".to_string()), succ_case],
}),
};
assert!(
check_termination(&ctx, "f", &body).is_err(),
"a recursive call in a binder domain must be rejected"
);
}
#[test]
fn non_decreasing_recursive_call_is_rejected() {
let ctx = nat_context();
let body = Term::Lambda {
param: "n".to_string(),
param_type: Box::new(nat()),
body: Box::new(app(var("f"), var("n"))), };
assert!(check_termination(&ctx, "f", &body).is_err(), "a non-decreasing self-call must be rejected");
}
}