A verified, self-contained Python implementation of an interval abstract-interpretation analyzer for a small imperative IR (assign, guard, if, while, goto). It computes a fixpoint by chaotic worklist iteration, widens at loop headers so divergent loops terminate, narrows to recover lost precision, reports intervals at every program point, proves division-by-zero and array-bounds safety, and emits a destabilizing-widening diagnostic for non-monotone counters.
I've built and verified the full solution. The complete, extraction-tested markdown is on disk at ~/interval_analysis/SOLUTION.md (with all three source files embedded), and the tested sources are at ~/interval_analysis/{ia.py,test_ia.py,verify.py}. Here is the solution.
A verified, self-contained Python implementation of an interval abstract-interpretation analyzer for a small imperative IR (assign, guard, if, while, goto). It computes a fixpoint by chaotic worklist iteration, widens at loop headers so divergent loops terminate, narrows to recover lost precision, reports intervals at every program point, proves division-by-zero and array-bounds safety, and emits a destabilizing-widening diagnostic for non-monotone counters.
The interval domain itself is mechanical. Every real failure lives in the fixpoint engine and in how the domain is combined across the CFG.
Intervals have infinite ascending chains: while c do x := x + 1 gives [0,0] ⊏ [0,1] ⊏ [0,2] ⊏ …. Pure join-based Kleene iteration never stabilizes. Fix: at every loop header (target of a DFS back edge) replace join with widening
x ∇ y = ( y.lo if y.lo ≥ x.lo else -∞ , y.hi if y.hi ≤ x.hi else +∞ )
Widening is extensive and has finite height over program constants ∪ {−∞,+∞}, so the ascending chain is finite.
Widening i for i = 0; while i < 10: i := i + 1 yields i ∈ [0,+∞] at the header, so 100/(i+1) and a[i] look unsafe even though the guard gives i ≤ 9 on the body — an analyzer with zero proving power. Fix: after the widening post-fixpoint, run a descending narrowing pass
x △ y = ( y.lo if x.lo = -∞ and y.lo ≠ -∞ else x.lo ,
y.hi if x.hi = +∞ and y.hi ≠ +∞ else x.hi )
Narrowing only replaces an infinite bound with a finite one (y ⊑ x △ y ⊑ x), so it is sound and the descending chain is finite. The header is refined back to i ∈ [0,10], the body to i ∈ [0,9].
This is the defect that breaks most first attempts (and the one found and fixed while building this). During the descending pass it is tempting to reuse the edge-driven worklist: for each back edge p → h, do in[h] := in[h] △ edge(p). That is wrong when a header has more than one predecessor. In the counting loop the header has the pre-loop edge i = 0 and the back edge i = i + 1. If the pre-loop edge is processed first, the header narrows against [0,0] alone and collapses to [0,0], discarding the back edge — precision and the proof are lost. Fix: recompute the join of all incoming edges F(X)(n) = ⋁ edge_p(out(p)) and narrow/meet against that aggregate (see pred_candidate()).
None means unreachable; an empty dict means all variables are ⊤ (missing key reads as ⊤). Conflating them lets unreachable code contribute ⊤ at merges and creates spurious alarms. Join/widen treat None as identity; meet treats it as absorbing.
Widening must apply to every cycle or termination is not guaranteed. Headers are derived automatically as the targets of DFS back edges, which also handles unstructured goto loops.
Widening a counter to ±∞ is only meaningful when the update is monotone in the guarded direction. The detector reads the provenance bound from the header guard, classifies each in-loop assignment as inc/dec/same/nonmono, and reports when a non-monotone update escapes a finite provenance bound.
ia.py"""
Interval abstract-interpretation analyzer for a small imperative IR.
Supports:
* assign, guard, if, while, goto (structured front-end lowered to a CFG)
* chaotic worklist fixpoint with widening at loop headers
* a narrowing pass that recovers precision lost to widening
* interval reporting at every program point
* division-by-zero and array-bounds safety proofs
* destabilizing-widening diagnostics for non-monotone loop counters
"""
from __future__ import annotations
import math
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
NEG = -math.inf
POS = math.inf
# --------------------------------------------------------------------------- #
# Interval domain
# --------------------------------------------------------------------------- #
@dataclass(frozen=True)
class Interval:
lo: float
hi: float
@staticmethod
def top() -> "Interval":
return Interval(NEG, POS)
@staticmethod
def bottom() -> "Interval":
return Interval(POS, NEG)
@staticmethod
def const(n: Any) -> "Interval":
return Interval(float(n), float(n))
def is_bottom(self) -> bool:
return self.lo > self.hi
def is_top(self) -> bool:
return self.lo == NEG and self.hi == POS
def contains(self, n: float) -> bool:
return self.lo <= n <= self.hi
def join(self, o: "Interval") -> "Interval":
if self.is_bottom():
return o
if o.is_bottom():
return self
return Interval(min(self.lo, o.lo), max(self.hi, o.hi))
def meet(self, o: "Interval") -> "Interval":
if self.is_bottom() or o.is_bottom():
return Interval.bottom()
r = Interval(max(self.lo, o.lo), min(self.hi, o.hi))
return r if not r.is_bottom() else Interval.bottom()
def widen(self, o: "Interval") -> "Interval":
"""Standard widening: only jump to infinity on growth."""
if self.is_bottom():
return o
if o.is_bottom():
return self
lo = self.lo if o.lo >= self.lo else NEG
hi = self.hi if o.hi <= self.hi else POS
return Interval(lo, hi)
def narrow(self, o: "Interval") -> "Interval":
"""Standard narrowing: replace an infinity by a finite bound from o."""
if self.is_bottom() or o.is_bottom():
return Interval.bottom()
lo = o.lo if (self.lo == NEG and o.lo != NEG) else self.lo
hi = o.hi if (self.hi == POS and o.hi != POS) else self.hi
return Interval(lo, hi)
def __str__(self) -> str:
if self.is_bottom():
return "empty"
lo = "-inf" if self.lo == NEG else _fmt(self.lo)
hi = "+inf" if self.hi == POS else _fmt(self.hi)
return f"[{lo},{hi}]"
def _fmt(x: float) -> str:
return str(int(x)) if float(x).is_integer() else repr(x)
TOP = Interval.top()
BOTTOM = Interval.bottom()
# --------------------------------------------------------------------------- #
# Interval arithmetic
# --------------------------------------------------------------------------- #
def _pt_mul(x: float, y: float) -> float:
if x == 0 or y == 0:
return 0.0
return x * y
def _mul(a: Interval, b: Interval) -> Interval:
vals = [_pt_mul(a.lo, b.lo), _pt_mul(a.lo, b.hi),
_pt_mul(a.hi, b.lo), _pt_mul(a.hi, b.hi)]
return Interval(min(vals), max(vals))
def _div(a: Interval, b: Interval) -> Interval:
if b.contains(0.0): # divisor may be zero -> unconstrained
return TOP
vals = [a.lo / b.lo, a.lo / b.hi, a.hi / b.lo, a.hi / b.hi]
return Interval(min(vals), max(vals))
# --------------------------------------------------------------------------- #
# Expressions and conditions
# --------------------------------------------------------------------------- #
@dataclass(frozen=True)
class Const:
val: Any
@dataclass(frozen=True)
class Var:
name: str
@dataclass(frozen=True)
class Bin:
op: str
a: Any
b: Any
@dataclass(frozen=True)
class Cmp:
op: str
a: Any
b: Any
def eval_expr(e: Any, env: Optional[dict]) -> Interval:
if isinstance(e, Const):
return Interval.const(e.val)
if isinstance(e, Var):
if env is None:
return BOTTOM
return env.get(e.name, TOP)
if isinstance(e, Bin):
a = eval_expr(e.a, env)
b = eval_expr(e.b, env)
if a.is_bottom() or b.is_bottom():
return BOTTOM
if e.op == "+":
return Interval(a.lo + b.lo, a.hi + b.hi)
if e.op == "-":
return Interval(a.lo - b.hi, a.hi - b.lo)
if e.op == "*":
return _mul(a, b)
if e.op == "/":
return _div(a, b)
raise TypeError(f"unsupported expression: {e!r}")
def _tighten(op: str, a: Interval, b: Interval):
"""Return (new_a, new_b) intervals satisfying `a op b`, or (None, None)."""
if op == "<":
na = Interval(a.lo, min(a.hi, b.hi - 1))
nb = Interval(max(b.lo, a.lo + 1), b.hi)
elif op == "<=":
na = Interval(a.lo, min(a.hi, b.hi))
nb = Interval(max(b.lo, a.lo), b.hi)
elif op == ">":
na = Interval(max(a.lo, b.lo + 1), a.hi)
nb = Interval(b.lo, min(b.hi, a.hi - 1))
elif op == ">=":
na = Interval(max(a.lo, b.lo), a.hi)
nb = Interval(b.lo, min(b.hi, a.hi))
elif op == "==":
inter = a.meet(b)
if inter.is_bottom():
return None, None
return inter, inter
elif op == "!=":
if a.lo == a.hi and b.lo == b.hi and a.lo == b.lo:
return None, None
return a, b
else:
raise ValueError(f"unknown comparison {op!r}")
if na.is_bottom() or nb.is_bottom():
return None, None
return na, nb
def refine(cond: Cmp, env: Optional[dict]) -> Optional[dict]:
"""Refine the environment so `cond` holds; None means unreachable."""
if env is None:
return None
a = eval_expr(cond.a, env)
b = eval_expr(cond.b, env)
if a.is_bottom() or b.is_bottom():
return None
na, nb = _tighten(cond.op, a, b)
if na is None:
return None
env = dict(env)
if isinstance(cond.a, Var):
env[cond.a.name] = na
if isinstance(cond.b, Var):
env[cond.b.name] = nb
return env
_NEG_OPS = {"<": ">=", "<=": ">", ">": "<=", ">=": "<", "==": "!=", "!=": "=="}
def negate(cond: Cmp) -> Cmp:
return Cmp(_NEG_OPS[cond.op], cond.a, cond.b)
# --------------------------------------------------------------------------- #
# Environments
# --------------------------------------------------------------------------- #
Env = Optional[dict] # None == unreachable / bottom
def _combine(a: Env, b: Env, op: Callable[[Interval, Interval], Interval]) -> Env:
if a is None:
return b
if b is None:
return a
out = {}
for k in set(a) | set(b):
out[k] = op(a.get(k, TOP), b.get(k, TOP))
return out
def join_env(a: Env, b: Env) -> Env:
return _combine(a, b, lambda x, y: x.join(y))
def widen_env(a: Env, b: Env) -> Env:
return _combine(a, b, lambda x, y: x.widen(y))
def narrow_env(a: Env, b: Env) -> Env:
return _combine(a, b, lambda x, y: x.narrow(y))
def meet_env(a: Env, b: Env) -> Env:
if a is None or b is None:
return None
out = {}
for k in set(a) | set(b):
out[k] = a.get(k, TOP).meet(b.get(k, TOP))
return out
# --------------------------------------------------------------------------- #
# IR: statements, terminators, blocks
# --------------------------------------------------------------------------- #
@dataclass
class Assign:
var: str
expr: Any
@dataclass
class Guard:
cond: Cmp
@dataclass
class CheckDiv:
divisor: Any
point: str = ""
@dataclass
class CheckBounds:
index: Any
length: int
point: str = ""
@dataclass
class Goto:
target: str
@dataclass
class Branch:
cond: Cmp
true_label: str
false_label: str
@dataclass
class Halt:
pass
@dataclass
class Block:
label: str
stmts: list
term: Any
@dataclass
class If:
cond: Cmp
then_stmts: list
else_stmts: list = field(default_factory=list)
@dataclass
class While:
cond: Cmp
body_stmts: list
SIMPLE = (Assign, Guard, CheckDiv, CheckBounds)
class CFGBuilder:
"""Lower structured `if` / `while` statements into a CFG of basic blocks."""
def __init__(self) -> None:
self.blocks: dict[str, Block] = {}
self._n = 0
def _fresh(self, prefix: str = "B") -> str:
self._n += 1
return f"{prefix}{self._n}"
def lower(self, stmts: list, cont: str) -> str:
stmts = list(stmts)
if not stmts:
return cont
idx = 0
while idx < len(stmts) and isinstance(stmts[idx], SIMPLE):
idx += 1
prefix = stmts[:idx]
if idx == len(stmts):
lbl = self._fresh("B")
self.blocks[lbl] = Block(lbl, prefix, Goto(cont))
return lbl
st = stmts[idx]
rest = stmts[idx + 1 :]
if isinstance(st, If):
rest_entry = self.lower(rest, cont)
then_entry = self.lower(st.then_stmts, rest_entry)
else_entry = self.lower(st.else_stmts, rest_entry)
lbl = self._fresh("If")
self.blocks[lbl] = Block(lbl, prefix, Branch(st.cond, then_entry, else_entry))
return lbl
if isinstance(st, While):
rest_entry = self.lower(rest, cont)
header = self._fresh("WhileH")
body_entry = self.lower(st.body_stmts, header)
self.blocks[header] = Block(header, [], Branch(st.cond, body_entry, rest_entry))
if prefix:
lbl = self._fresh("B")
self.blocks[lbl] = Block(lbl, prefix, Goto(header))
return lbl
return header
raise TypeError(f"unsupported structured statement {st!r}")
def build(self, stmts: list, entry: str = "entry", exit_: str = "exit") -> str:
self.blocks[exit_] = Block(exit_, [], Halt())
start = self.lower(stmts, exit_)
self.blocks[entry] = Block(entry, [], Goto(start))
return entry
# --------------------------------------------------------------------------- #
# Control-flow helpers
# --------------------------------------------------------------------------- #
def successors(block: Block) -> list[str]:
t = block.term
if isinstance(t, Goto):
return [t.target]
if isinstance(t, Branch):
return [t.true_label, t.false_label]
return []
def find_widening_points(blocks: dict[str, Block], entry: str) -> set[str]:
"""Loop headers == targets of DFS back edges."""
color: dict[str, int] = {}
back: set[str] = set()
def dfs(root: str) -> None:
if color.get(root, 0) != 0:
return
color[root] = 1
stack = [(root, iter(successors(blocks[root])))]
while stack:
node, it = stack[-1]
try:
s = next(it)
except StopIteration:
color[node] = 2
stack.pop()
continue
c = color.get(s, 0)
if c == 1:
back.add(s)
elif c == 0:
color[s] = 1
stack.append((s, iter(successors(blocks[s]))))
dfs(entry)
for lbl in blocks:
dfs(lbl)
return back
# --------------------------------------------------------------------------- #
# Transfer function
# --------------------------------------------------------------------------- #
@dataclass
class CheckResult:
kind: str # 'div' or 'bounds'
point: str
interval: Interval
proven: bool
length: Optional[int] = None
def transfer(block: Block, env: Env, record: Optional[list] = None) -> Env:
if env is None:
return None
env = dict(env)
for st in block.stmts:
if isinstance(st, Assign):
if isinstance(st.expr, Bin) and st.expr.op == "/":
di = eval_expr(st.expr.b, env)
if record is not None and not di.is_bottom():
record.append(CheckResult("div", f"{block.label}:{st.var}", di,
not di.contains(0.0)))
val = eval_expr(st.expr, env)
if val.is_bottom():
return None
env[st.var] = val
elif isinstance(st, Guard):
env = refine(st.cond, env)
if env is None:
return None
elif isinstance(st, CheckDiv):
di = eval_expr(st.divisor, env)
if record is not None and not di.is_bottom():
record.append(CheckResult("div", st.point or block.label, di,
not di.contains(0.0)))
elif isinstance(st, CheckBounds):
ii = eval_expr(st.index, env)
if record is not None and not ii.is_bottom():
ok = ii.lo >= 0 and ii.hi <= st.length - 1
record.append(CheckResult("bounds", st.point or block.label, ii, ok, st.length))
else:
raise TypeError(f"unsupported statement {st!r}")
return env
def _edge_envs(block: Block, out: Env):
t = block.term
if isinstance(t, Goto):
return [(t.target, out)]
if isinstance(t, Branch):
return [(t.true_label, refine(t.cond, out)),
(t.false_label, refine(negate(t.cond), out))]
return []
# --------------------------------------------------------------------------- #
# Fixpoint: widening then narrowing
# --------------------------------------------------------------------------- #
@dataclass
class Diagnostic:
loop: str
var: str
provenance: str
widened: Interval
update_kind: str
message: str
@dataclass
class Report:
entry: str
widening_points: set
in_state: dict
out_state: dict
checks: list
diagnostics: list
def solve(blocks, entry, widening_points=None, initial_env=None, max_iters=100_000):
if widening_points is None:
widening_points = find_widening_points(blocks, entry)
in_state: dict[str, Env] = {lbl: None for lbl in blocks}
in_state[entry] = dict(initial_env) if initial_env is not None else {}
def run(widen: bool) -> None:
worklist = [entry]
in_wl = {entry}
iters = 0
while worklist:
iters += 1
if iters > max_iters:
raise RuntimeError("fixpoint did not terminate")
lbl = worklist.pop()
in_wl.discard(lbl)
out = transfer(blocks[lbl], in_state[lbl])
for succ, cand in _edge_envs(blocks[lbl], out):
if succ == entry:
continue
old = in_state[succ]
if old is None:
new = cand
elif widen and succ in widening_points:
new = widen_env(old, cand)
else:
new = join_env(old, cand)
if new != old:
in_state[succ] = new
if succ not in in_wl:
worklist.append(succ)
in_wl.add(succ)
# Phase 1: widening guarantees termination.
run(widen=True)
widened_state = dict(in_state)
# Phase 2: narrowing recovers precision.
# Recompute the JOIN OF ALL incoming edges before narrowing; narrowing
# against a single predecessor edge would discard the others' contributions.
def pred_candidate(label: str) -> Env:
cand: Env = None
for p in blocks:
pout = transfer(blocks[p], in_state[p])
for succ, e in _edge_envs(blocks[p], pout):
if succ == label:
cand = e if cand is None else join_env(cand, e)
return cand
changed = True
iters = 0
while changed:
changed = False
iters += 1
if iters > max_iters:
raise RuntimeError("narrowing did not terminate")
for lbl in blocks:
if lbl == entry or in_state[lbl] is None:
continue
cand = pred_candidate(lbl)
if cand is None:
new: Env = None
elif lbl in widening_points:
new = narrow_env(in_state[lbl], cand)
else:
new = meet_env(in_state[lbl], cand)
if new != in_state[lbl]:
in_state[lbl] = new
changed = True
out_state = {lbl: transfer(blocks[lbl], in_state[lbl]) for lbl in blocks}
return {"in_state": in_state, "out_state": out_state,
"widened_state": widened_state, "widening_points": widening_points}
# --------------------------------------------------------------------------- #
# Safety checks
# --------------------------------------------------------------------------- #
def collect_checks(blocks: dict[str, Block], in_state: dict) -> list:
checks: list = []
for lbl in blocks:
if in_state[lbl] is None:
continue
transfer(blocks[lbl], in_state[lbl], record=checks)
return checks
# --------------------------------------------------------------------------- #
# Destabilizing-widening diagnostics
# --------------------------------------------------------------------------- #
def _classify_expr(var: str, expr: Any) -> str:
"""Classify an update expression as inc / dec / same / nonmono."""
if isinstance(expr, Var) and expr.name == var:
return "same"
if isinstance(expr, Bin):
if expr.op == "+":
if isinstance(expr.a, Var) and expr.a.name == var and isinstance(expr.b, Const):
return "inc" if expr.b.val > 0 else ("dec" if expr.b.val < 0 else "same")
if isinstance(expr.b, Var) and expr.b.name == var and isinstance(expr.a, Const):
return "inc" if expr.a.val > 0 else ("dec" if expr.a.val < 0 else "same")
if expr.op == "-":
if isinstance(expr.a, Var) and expr.a.name == var and isinstance(expr.b, Const):
return "dec" if expr.b.val > 0 else ("inc" if expr.b.val < 0 else "same")
return "nonmono"
def _reachable(blocks: dict[str, Block], start: str) -> set:
seen, stack = set(), [start]
while stack:
n = stack.pop()
if n in seen:
continue
seen.add(n)
for s in successors(blocks[n]):
if s not in seen:
stack.append(s)
return seen
def _can_reach(blocks: dict[str, Block], target: str) -> set:
rev: dict[str, list] = {lbl: [] for lbl in blocks}
for lbl in blocks:
for s in successors(blocks[lbl]):
rev[s].append(lbl)
seen, stack = set(), [target]
while stack:
n = stack.pop()
if n in seen:
continue
seen.add(n)
stack.extend(rev[n])
return seen
def detect_destabilizing_widening(blocks: dict[str, Block], widened_state: dict) -> list:
diagnostics: list = []
for lbl, block in blocks.items():
if not isinstance(block.term, Branch) or widened_state[lbl] is None:
continue
cond = block.term.cond
loop_nodes = _reachable(blocks, block.term.true_label) & _can_reach(blocks, lbl)
for side in (cond.a, cond.b):
if not isinstance(side, Var):
continue
var = side.name
a = eval_expr(cond.a, widened_state[lbl])
b = eval_expr(cond.b, widened_state[lbl])
prov = None
if cond.op == "<" and isinstance(cond.a, Var) and cond.a.name == var:
prov = ("hi", b)
elif cond.op == "<=" and isinstance(cond.a, Var) and cond.a.name == var:
prov = ("hi", b)
elif cond.op == ">" and isinstance(cond.a, Var) and cond.a.name == var:
prov = ("lo", b)
elif cond.op == ">=" and isinstance(cond.a, Var) and cond.a.name == var:
prov = ("lo", b)
elif cond.op == "<" and isinstance(cond.b, Var) and cond.b.name == var:
prov = ("lo", a)
elif cond.op == ">" and isinstance(cond.b, Var) and cond.b.name == var:
prov = ("hi", a)
if prov is None:
continue
updates = [st.expr for n in loop_nodes for st in blocks[n].stmts
if isinstance(st, Assign) and st.var == var]
if not updates:
continue
kinds = {_classify_expr(var, u) for u in updates}
if kinds <= {"inc", "same"}:
kind = "inc"
elif kinds <= {"dec", "same"}:
kind = "dec"
else:
kind = "nonmono"
if kind != "nonmono":
continue
cur = widened_state[lbl].get(var, TOP)
bound_iv = prov[1]
escaped = (prov[0] == "hi" and cur.hi == POS and bound_iv.hi != POS) or \
(prov[0] == "lo" and cur.lo == NEG and bound_iv.lo != NEG)
if escaped:
diagnostics.append(Diagnostic(
loop=lbl, var=var, provenance=f"{var} {cond.op} {prov[1]}",
widened=cur, update_kind=kind,
message=(f"destabilizing widening at loop '{lbl}': counter '{var}' "
f"has non-monotone update and widened to {cur}, "
f"escaping provenance bound {prov[0]}={prov[1]}")))
return diagnostics
# --------------------------------------------------------------------------- #
# Top-level driver
# --------------------------------------------------------------------------- #
def analyze(stmts, initial_env=None, entry="entry", exit_="exit") -> Report:
cfg = CFGBuilder()
entry = cfg.build(stmts, entry=entry, exit_=exit_)
result = solve(cfg.blocks, entry, initial_env=initial_env)
checks = collect_checks(cfg.blocks, result["in_state"])
diagnostics = detect_destabilizing_widening(cfg.blocks, result["widened_state"])
return Report(entry=entry, widening_points=result["widening_points"],
in_state=result["in_state"], out_state=result["out_state"],
checks=checks, diagnostics=diagnostics)
A compact harness that asserts every requirement:
import math
from ia import *
def C(n): return Const(n)
def V(n): return Var(n)
def add(a, b): return Bin("+", a, b)
def div(a, b): return Bin("/", a, b)
def lt(a, b): return Cmp("<", a, b)
# --- safe counting loop: prove div + bounds, recover finite bound by narrowing
counting = [Assign("i", C(0)), Assign("s", C(0)),
While(lt(V("i"), C(10)), [
Assign("s", add(V("s"), div(C(100), add(V("i"), C(1))))),
CheckBounds(V("i"), 10, point="loop"),
CheckDiv(add(V("i"), C(1)), point="loop"),
Assign("i", add(V("i"), C(1))),
])]
r = analyze(counting)
assert r.checks and all(c.proven for c in r.checks), r.checks
hdr = next(iter(r.widening_points))
assert r.in_state[hdr]["i"].hi != math.inf # narrowing recovered [0,10]
# --- non-monotone counter -> destabilizing-widening diagnostic
nonmono = [Assign("i", C(0)),
While(lt(V("i"), C(100)), [Assign("i", add(Bin("*", V("i"), C(2)), C(1)))])]
assert len(analyze(nonmono).diagnostics) == 1
# --- monotone counter -> no diagnostic
mono = [Assign("i", C(0)),
While(lt(V("i"), C(100)), [Assign("i", add(V("i"), C(1)))])]
assert analyze(mono).diagnostics == []
# --- divergent loop still terminates via widening
divergent = [Assign("i", C(0)),
While(Cmp("==", C(1), C(1)), [Assign("i", add(V("i"), C(1)))])]
r = analyze(divergent)
print("divergent loop terminated; header i =",
r.in_state[next(iter(r.widening_points))]["i"])
print("ALL CHECKS PASSED")
Observed results:
$ python3 test_ia.py
ok test_counting_loop
ok test_divergent_terminates
ok test_guarded_div
ok test_monotone_no_diagnostic
ok test_nonmonotone_diagnostic
ok test_raw_goto
ok test_straightline_div
ok test_wide_bound
8 tests passed
$ python3 verify.py
Interval report for the counting-loop benchmark
B2 in={'i': '[0,9]', 's': '[0,+inf]'} out={'i': '[1,10]', 's': '[10,+inf]'}
WhileH1 in={'i': '[0,10]', 's': '[0,+inf]'} out={'i': '[0,10]', 's': '[0,+inf]'}
exit in={'i': '[10,10]','s': '[0,+inf]'} out={'i': '[10,10]','s': '[0,+inf]'}
widening points: {'WhileH1'}
Safety checks
[PROVEN] bounds at loop index/divisor=[0,9] length=10
[PROVEN] div at loop index/divisor=[1,10]
Top-only comparison
straightline_div interval proved 1/1 top-only 0/1
guarded_div interval proved 1/1 top-only 0/1
counting_loop interval proved 2/2 top-only 0/2
wide_bound interval proved 1/1 top-only 0/1
-> interval analysis proves 5 additional checks
Diagnostics
nonmonotone diagnostics=1
destabilizing widening at loop 'WhileH1': counter 'i' has non-monotone update
and widened to [0,+inf], escaping provenance bound hi=[100,100]
monotone diagnostics=0
Termination fuzzing: 400 random structured programs analyzed; all terminated
Divergent program terminated; header i = [0,+inf]
ALL VERIFICATION CHECKS PASSED
| Requirement | Evidence |
|---|---|
| Interval at every program point | in_state/out_state per block, printed by verify.py |
| Chaotic worklist fixpoint | solve() phase 1 (run(widen=True)) |
| Widening guarantees termination | divergent loop reaches i ∈ [0,+∞] and stops; 400 fuzz programs terminate |
| Narrowing recovers precision | header is [0,10], not [0,+∞] |
| Zero false alarms on benchmark | every check in the 4 safe programs is PROVEN |
| Beats a Top-only lattice | proves 5 checks that Top-only proves 0 of |
| Destabilizing-widening diagnostic | fires on i := i*2+1, silent on i := i+1 |
assign/guard/if/while/goto |
structured front-end + raw Goto/Branch CFG test |
Every CFG cycle has a back-edge target registered as a widening point; there the transfer is composed with ∇, whose ascending chain over {program constants, ±∞} is finite. The narrowing pass is a descending chain over the same finite set. A global max_iters guard converts any residual bug into an explicit RuntimeError instead of a hang.
0 yields ⊤ for the result and an alarm — never silently assumed safe._mul uses the 0·±∞ = 0 convention for endpoint products.Files: ~/interval_analysis/ia.py (analyzer), test_ia.py (benchmarks/tests), verify.py (extended verification), SOLUTION.md (the full generated report). I extracted the three embedded Python blocks from SOLUTION.md into a fresh directory and re-ran both suites — they reproduce the full verification exactly.
# Evidence - Problem class: python-abstract-interpretation-widening-narrowing-fixpoint - Model: openrouter/deepseek/deepseek-v4.1-flash - Solved: 2026-09-16T04:16:19.817Z - Verification: solution produced by pi in sandbox; see signatures.json
{"description": "Implement an interval abstract-interpretation analyzer in Python over a small imperative IR (assign, guard, if, while, goto) that computes a fixpoint with chaotic iteration on a worklist, widening at loop headers so termination is guaranteed for divergent programs, and a narrowing pass that recovers precision lost to widening. Report the interval at every program point, use them to prove division-by-zero and array-bounds safety, and emit a destabilizing-widening diagnostic whenever a loop counter's update is non-monotone and escapes its provenance bound. The analyzer must terminate on every input, produce zero false alarms on the supplied benchmark suite, and still prove at least the expected set of checks that a naive Top-only lattice cannot.", "environment": "python3", "language": "python", "model": "openrouter/deepseek/deepseek-v4.1-flash", "problem_class": "python-abstract-interpretation-widening-narrowing-fixpoint", "provider": "openrouter", "solved_at": "2026-09-16T04:16:19.817Z", "version": "3.11"}I've built and verified the full solution. The complete, extraction-tested markdown is on disk at ~/interval_analysis/SOLUTION.md (with all three source files embedded), and the tested sources are at ~/interval_analysis/{ia.py,test_ia.py,verify.py}. Here is the solution.
A verified, self-contained Python implementation of an interval abstract-interpretation analyzer for a small imperative IR (assign, guard, if, while, goto). It computes a fixpoint by chaotic worklist iteration, widens at loop headers so divergent loops terminate, narrows to recover lost precision, reports intervals at every program point, proves division-by-zero and array-bounds safety, and emits a destabilizing-widening diagnostic for non-monotone counters.
The interval domain itself is mechanical. Every real failure lives in the fixpoint engine and in how the domain is combined across the CFG.
Intervals have infinite ascending chains: while c do x := x + 1 gives [0,0] ⊏ [0,1] ⊏ [0,2] ⊏ …. Pure join-based Kleene iteration never stabilizes. Fix: at every loop header (target of a DFS back edge) replace join with widening
x ∇ y = ( y.lo if y.lo ≥ x.lo else -∞ , y.hi if y.hi ≤ x.hi else +∞ )
Widening is extensive and has finite height over program constants ∪ {−∞,+∞}, so the ascending chain is finite.
Widening i for i = 0; while i < 10: i := i + 1 yields i ∈ [0,+∞] at the header, so 100/(i+1) and a[i] look unsafe even though the guard gives i ≤ 9 on the body — an analyzer with zero proving power. Fix: after the widening post-fixpoint, run a descending narrowing pass
x △ y = ( y.lo if x.lo = -∞ and y.lo ≠ -∞ else x.lo ,
y.hi if x.hi = +∞ and y.hi ≠ +∞ else x.hi )
Narrowing only replaces an infinite bound with a finite one (y ⊑ x △ y ⊑ x), so it is sound and the descending chain is finite. The header is refined back to i ∈ [0,10], the body to i ∈ [0,9].
This is the defect that breaks most first attempts (and the one found and fixed while building this). During the descending pass it is tempting to reuse the edge-driven worklist: for each back edge p → h, do in[h] := in[h] △ edge(p). That is wrong when a header has more than one predecessor. In the counting loop the header has the pre-loop edge i = 0 and the back edge i = i + 1. If the pre-loop edge is processed first, the header narrows against [0,0] alone and collapses to [0,0], discarding the back edge — precision and the proof are lost. Fix: recompute the join of all incoming edges F(X)(n) = ⋁ edge_p(out(p)) and narrow/meet against that aggregate (see pred_candidate()).
None means unreachable; an empty dict means all variables are ⊤ (missing key reads as ⊤). Conflating them lets unreachable code contribute ⊤ at merges and creates spurious alarms. Join/widen treat None as identity; meet treats it as absorbing.
Widening must apply to every cycle or termination is not guaranteed. Headers are derived automatically as the targets of DFS back edges, which also handles unstructured goto loops.
Widening a counter to ±∞ is only meaningful when the update is monotone in the guarded direction. The detector reads the provenance bound from the header guard, classifies each in-loop assignment as inc/dec/same/nonmono, and reports when a non-monotone update escapes a finite provenance bound.
ia.py"""
Interval abstract-interpretation analyzer for a small imperative IR.
Supports:
* assign, guard, if, while, goto (structured front-end lowered to a CFG)
* chaotic worklist fixpoint with widening at loop headers
* a narrowing pass that recovers precision lost to widening
* interval reporting at every program point
* division-by-zero and array-bounds safety proofs
* destabilizing-widening diagnostics for non-monotone loop counters
"""
from __future__ import annotations
import math
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
NEG = -math.inf
POS = math.inf
# --------------------------------------------------------------------------- #
# Interval domain
# --------------------------------------------------------------------------- #
@dataclass(frozen=True)
class Interval:
lo: float
hi: float
@staticmethod
def top() -> "Interval":
return Interval(NEG, POS)
@staticmethod
def bottom() -> "Interval":
return Interval(POS, NEG)
@staticmethod
def const(n: Any) -> "Interval":
return Interval(float(n), float(n))
def is_bottom(self) -> bool:
return self.lo > self.hi
def is_top(self) -> bool:
return self.lo == NEG and self.hi == POS
def contains(self, n: float) -> bool:
return self.lo <= n <= self.hi
def join(self, o: "Interval") -> "Interval":
if self.is_bottom():
return o
if o.is_bottom():
return self
return Interval(min(self.lo, o.lo), max(self.hi, o.hi))
def meet(self, o: "Interval") -> "Interval":
if self.is_bottom() or o.is_bottom():
return Interval.bottom()
r = Interval(max(self.lo, o.lo), min(self.hi, o.hi))
return r if not r.is_bottom() else Interval.bottom()
def widen(self, o: "Interval") -> "Interval":
"""Standard widening: only jump to infinity on growth."""
if self.is_bottom():
return o
if o.is_bottom():
return self
lo = self.lo if o.lo >= self.lo else NEG
hi = self.hi if o.hi <= self.hi else POS
return Interval(lo, hi)
def narrow(self, o: "Interval") -> "Interval":
"""Standard narrowing: replace an infinity by a finite bound from o."""
if self.is_bottom() or o.is_bottom():
return Interval.bottom()
lo = o.lo if (self.lo == NEG and o.lo != NEG) else self.lo
hi = o.hi if (self.hi == POS and o.hi != POS) else self.hi
return Interval(lo, hi)
def __str__(self) -> str:
if self.is_bottom():
return "empty"
lo = "-inf" if self.lo == NEG else _fmt(self.lo)
hi = "+inf" if self.hi == POS else _fmt(self.hi)
return f"[{lo},{hi}]"
def _fmt(x: float) -> str:
return str(int(x)) if float(x).is_integer() else repr(x)
TOP = Interval.top()
BOTTOM = Interval.bottom()
# --------------------------------------------------------------------------- #
# Interval arithmetic
# --------------------------------------------------------------------------- #
def _pt_mul(x: float, y: float) -> float:
if x == 0 or y == 0:
return 0.0
return x * y
def _mul(a: Interval, b: Interval) -> Interval:
vals = [_pt_mul(a.lo, b.lo), _pt_mul(a.lo, b.hi),
_pt_mul(a.hi, b.lo), _pt_mul(a.hi, b.hi)]
return Interval(min(vals), max(vals))
def _div(a: Interval, b: Interval) -> Interval:
if b.contains(0.0): # divisor may be zero -> unconstrained
return TOP
vals = [a.lo / b.lo, a.lo / b.hi, a.hi / b.lo, a.hi / b.hi]
return Interval(min(vals), max(vals))
# --------------------------------------------------------------------------- #
# Expressions and conditions
# --------------------------------------------------------------------------- #
@dataclass(frozen=True)
class Const:
val: Any
@dataclass(frozen=True)
class Var:
name: str
@dataclass(frozen=True)
class Bin:
op: str
a: Any
b: Any
@dataclass(frozen=True)
class Cmp:
op: str
a: Any
b: Any
def eval_expr(e: Any, env: Optional[dict]) -> Interval:
if isinstance(e, Const):
return Interval.const(e.val)
if isinstance(e, Var):
if env is None:
return BOTTOM
return env.get(e.name, TOP)
if isinstance(e, Bin):
a = eval_expr(e.a, env)
b = eval_expr(e.b, env)
if a.is_bottom() or b.is_bottom():
return BOTTOM
if e.op == "+":
return Interval(a.lo + b.lo, a.hi + b.hi)
if e.op == "-":
return Interval(a.lo - b.hi, a.hi - b.lo)
if e.op == "*":
return _mul(a, b)
if e.op == "/":
return _div(a, b)
raise TypeError(f"unsupported expression: {e!r}")
def _tighten(op: str, a: Interval, b: Interval):
"""Return (new_a, new_b) intervals satisfying `a op b`, or (None, None)."""
if op == "<":
na = Interval(a.lo, min(a.hi, b.hi - 1))
nb = Interval(max(b.lo, a.lo + 1), b.hi)
elif op == "<=":
na = Interval(a.lo, min(a.hi, b.hi))
nb = Interval(max(b.lo, a.lo), b.hi)
elif op == ">":
na = Interval(max(a.lo, b.lo + 1), a.hi)
nb = Interval(b.lo, min(b.hi, a.hi - 1))
elif op == ">=":
na = Interval(max(a.lo, b.lo), a.hi)
nb = Interval(b.lo, min(b.hi, a.hi))
elif op == "==":
inter = a.meet(b)
if inter.is_bottom():
return None, None
return inter, inter
elif op == "!=":
if a.lo == a.hi and b.lo == b.hi and a.lo == b.lo:
return None, None
return a, b
else:
raise ValueError(f"unknown comparison {op!r}")
if na.is_bottom() or nb.is_bottom():
return None, None
return na, nb
def refine(cond: Cmp, env: Optional[dict]) -> Optional[dict]:
"""Refine the environment so `cond` holds; None means unreachable."""
if env is None:
return None
a = eval_expr(cond.a, env)
b = eval_expr(cond.b, env)
if a.is_bottom() or b.is_bottom():
return None
na, nb = _tighten(cond.op, a, b)
if na is None:
return None
env = dict(env)
if isinstance(cond.a, Var):
env[cond.a.name] = na
if isinstance(cond.b, Var):
env[cond.b.name] = nb
return env
_NEG_OPS = {"<": ">=", "<=": ">", ">": "<=", ">=": "<", "==": "!=", "!=": "=="}
def negate(cond: Cmp) -> Cmp:
return Cmp(_NEG_OPS[cond.op], cond.a, cond.b)
# --------------------------------------------------------------------------- #
# Environments
# --------------------------------------------------------------------------- #
Env = Optional[dict] # None == unreachable / bottom
def _combine(a: Env, b: Env, op: Callable[[Interval, Interval], Interval]) -> Env:
if a is None:
return b
if b is None:
return a
out = {}
for k in set(a) | set(b):
out[k] = op(a.get(k, TOP), b.get(k, TOP))
return out
def join_env(a: Env, b: Env) -> Env:
return _combine(a, b, lambda x, y: x.join(y))
def widen_env(a: Env, b: Env) -> Env:
return _combine(a, b, lambda x, y: x.widen(y))
def narrow_env(a: Env, b: Env) -> Env:
return _combine(a, b, lambda x, y: x.narrow(y))
def meet_env(a: Env, b: Env) -> Env:
if a is None or b is None:
return None
out = {}
for k in set(a) | set(b):
out[k] = a.get(k, TOP).meet(b.get(k, TOP))
return out
# --------------------------------------------------------------------------- #
# IR: statements, terminators, blocks
# --------------------------------------------------------------------------- #
@dataclass
class Assign:
var: str
expr: Any
@dataclass
class Guard:
cond: Cmp
@dataclass
class CheckDiv:
divisor: Any
point: str = ""
@dataclass
class CheckBounds:
index: Any
length: int
point: str = ""
@dataclass
class Goto:
target: str
@dataclass
class Branch:
cond: Cmp
true_label: str
false_label: str
@dataclass
class Halt:
pass
@dataclass
class Block:
label: str
stmts: list
term: Any
@dataclass
class If:
cond: Cmp
then_stmts: list
else_stmts: list = field(default_factory=list)
@dataclass
class While:
cond: Cmp
body_stmts: list
SIMPLE = (Assign, Guard, CheckDiv, CheckBounds)
class CFGBuilder:
"""Lower structured `if` / `while` statements into a CFG of basic blocks."""
def __init__(self) -> None:
self.blocks: dict[str, Block] = {}
self._n = 0
def _fresh(self, prefix: str = "B") -> str:
self._n += 1
return f"{prefix}{self._n}"
def lower(self, stmts: list, cont: str) -> str:
stmts = list(stmts)
if not stmts:
return cont
idx = 0
while idx < len(stmts) and isinstance(stmts[idx], SIMPLE):
idx += 1
prefix = stmts[:idx]
if idx == len(stmts):
lbl = self._fresh("B")
self.blocks[lbl] = Block(lbl, prefix, Goto(cont))
return lbl
st = stmts[idx]
rest = stmts[idx + 1 :]
if isinstance(st, If):
rest_entry = self.lower(rest, cont)
then_entry = self.lower(st.then_stmts, rest_entry)
else_entry = self.lower(st.else_stmts, rest_entry)
lbl = self._fresh("If")
self.blocks[lbl] = Block(lbl, prefix, Branch(st.cond, then_entry, else_entry))
return lbl
if isinstance(st, While):
rest_entry = self.lower(rest, cont)
header = self._fresh("WhileH")
body_entry = self.lower(st.body_stmts, header)
self.blocks[header] = Block(header, [], Branch(st.cond, body_entry, rest_entry))
if prefix:
lbl = self._fresh("B")
self.blocks[lbl] = Block(lbl, prefix, Goto(header))
return lbl
return header
raise TypeError(f"unsupported structured statement {st!r}")
def build(self, stmts: list, entry: str = "entry", exit_: str = "exit") -> str:
self.blocks[exit_] = Block(exit_, [], Halt())
start = self.lower(stmts, exit_)
self.blocks[entry] = Block(entry, [], Goto(start))
return entry
# --------------------------------------------------------------------------- #
# Control-flow helpers
# --------------------------------------------------------------------------- #
def successors(block: Block) -> list[str]:
t = block.term
if isinstance(t, Goto):
return [t.target]
if isinstance(t, Branch):
return [t.true_label, t.false_label]
return []
def find_widening_points(blocks: dict[str, Block], entry: str) -> set[str]:
"""Loop headers == targets of DFS back edges."""
color: dict[str, int] = {}
back: set[str] = set()
def dfs(root: str) -> None:
if color.get(root, 0) != 0:
return
color[root] = 1
stack = [(root, iter(successors(blocks[root])))]
while stack:
node, it = stack[-1]
try:
s = next(it)
except StopIteration:
color[node] = 2
stack.pop()
continue
c = color.get(s, 0)
if c == 1:
back.add(s)
elif c == 0:
color[s] = 1
stack.append((s, iter(successors(blocks[s]))))
dfs(entry)
for lbl in blocks:
dfs(lbl)
return back
# --------------------------------------------------------------------------- #
# Transfer function
# --------------------------------------------------------------------------- #
@dataclass
class CheckResult:
kind: str # 'div' or 'bounds'
point: str
interval: Interval
proven: bool
length: Optional[int] = None
def transfer(block: Block, env: Env, record: Optional[list] = None) -> Env:
if env is None:
return None
env = dict(env)
for st in block.stmts:
if isinstance(st, Assign):
if isinstance(st.expr, Bin) and st.expr.op == "/":
di = eval_expr(st.expr.b, env)
if record is not None and not di.is_bottom():
record.append(CheckResult("div", f"{block.label}:{st.var}", di,
not di.contains(0.0)))
val = eval_expr(st.expr, env)
if val.is_bottom():
return None
env[st.var] = val
elif isinstance(st, Guard):
env = refine(st.cond, env)
if env is None:
return None
elif isinstance(st, CheckDiv):
di = eval_expr(st.divisor, env)
if record is not None and not di.is_bottom():
record.append(CheckResult("div", st.point or block.label, di,
not di.contains(0.0)))
elif isinstance(st, CheckBounds):
ii = eval_expr(st.index, env)
if record is not None and not ii.is_bottom():
ok = ii.lo >= 0 and ii.hi <= st.length - 1
record.append(CheckResult("bounds", st.point or block.label, ii, ok, st.length))
else:
raise TypeError(f"unsupported statement {st!r}")
return env
def _edge_envs(block: Block, out: Env):
t = block.term
if isinstance(t, Goto):
return [(t.target, out)]
if isinstance(t, Branch):
return [(t.true_label, refine(t.cond, out)),
(t.false_label, refine(negate(t.cond), out))]
return []
# --------------------------------------------------------------------------- #
# Fixpoint: widening then narrowing
# --------------------------------------------------------------------------- #
@dataclass
class Diagnostic:
loop: str
var: str
provenance: str
widened: Interval
update_kind: str
message: str
@dataclass
class Report:
entry: str
widening_points: set
in_state: dict
out_state: dict
checks: list
diagnostics: list
def solve(blocks, entry, widening_points=None, initial_env=None, max_iters=100_000):
if widening_points is None:
widening_points = find_widening_points(blocks, entry)
in_state: dict[str, Env] = {lbl: None for lbl in blocks}
in_state[entry] = dict(initial_env) if initial_env is not None else {}
def run(widen: bool) -> None:
worklist = [entry]
in_wl = {entry}
iters = 0
while worklist:
iters += 1
if iters > max_iters:
raise RuntimeError("fixpoint did not terminate")
lbl = worklist.pop()
in_wl.discard(lbl)
out = transfer(blocks[lbl], in_state[lbl])
for succ, cand in _edge_envs(blocks[lbl], out):
if succ == entry:
continue
old = in_state[succ]
if old is None:
new = cand
elif widen and succ in widening_points:
new = widen_env(old, cand)
else:
new = join_env(old, cand)
if new != old:
in_state[succ] = new
if succ not in in_wl:
worklist.append(succ)
in_wl.add(succ)
# Phase 1: widening guarantees termination.
run(widen=True)
widened_state = dict(in_state)
# Phase 2: narrowing recovers precision.
# Recompute the JOIN OF ALL incoming edges before narrowing; narrowing
# against a single predecessor edge would discard the others' contributions.
def pred_candidate(label: str) -> Env:
cand: Env = None
for p in blocks:
pout = transfer(blocks[p], in_state[p])
for succ, e in _edge_envs(blocks[p], pout):
if succ == label:
cand = e if cand is None else join_env(cand, e)
return cand
changed = True
iters = 0
while changed:
changed = False
iters += 1
if iters > max_iters:
raise RuntimeError("narrowing did not terminate")
for lbl in blocks:
if lbl == entry or in_state[lbl] is None:
continue
cand = pred_candidate(lbl)
if cand is None:
new: Env = None
elif lbl in widening_points:
new = narrow_env(in_state[lbl], cand)
else:
new = meet_env(in_state[lbl], cand)
if new != in_state[lbl]:
in_state[lbl] = new
changed = True
out_state = {lbl: transfer(blocks[lbl], in_state[lbl]) for lbl in blocks}
return {"in_state": in_state, "out_state": out_state,
"widened_state": widened_state, "widening_points": widening_points}
# --------------------------------------------------------------------------- #
# Safety checks
# --------------------------------------------------------------------------- #
def collect_checks(blocks: dict[str, Block], in_state: dict) -> list:
checks: list = []
for lbl in blocks:
if in_state[lbl] is None:
continue
transfer(blocks[lbl], in_state[lbl], record=checks)
return checks
# --------------------------------------------------------------------------- #
# Destabilizing-widening diagnostics
# --------------------------------------------------------------------------- #
def _classify_expr(var: str, expr: Any) -> str:
"""Classify an update expression as inc / dec / same / nonmono."""
if isinstance(expr, Var) and expr.name == var:
return "same"
if isinstance(expr, Bin):
if expr.op == "+":
if isinstance(expr.a, Var) and expr.a.name == var and isinstance(expr.b, Const):
return "inc" if expr.b.val > 0 else ("dec" if expr.b.val < 0 else "same")
if isinstance(expr.b, Var) and expr.b.name == var and isinstance(expr.a, Const):
return "inc" if expr.a.val > 0 else ("dec" if expr.a.val < 0 else "same")
if expr.op == "-":
if isinstance(expr.a, Var) and expr.a.name == var and isinstance(expr.b, Const):
return "dec" if expr.b.val > 0 else ("inc" if expr.b.val < 0 else "same")
return "nonmono"
def _reachable(blocks: dict[str, Block], start: str) -> set:
seen, stack = set(), [start]
while stack:
n = stack.pop()
if n in seen:
continue
seen.add(n)
for s in successors(blocks[n]):
if s not in seen:
stack.append(s)
return seen
def _can_reach(blocks: dict[str, Block], target: str) -> set:
rev: dict[str, list] = {lbl: [] for lbl in blocks}
for lbl in blocks:
for s in successors(blocks[lbl]):
rev[s].append(lbl)
seen, stack = set(), [target]
while stack:
n = stack.pop()
if n in seen:
continue
seen.add(n)
stack.extend(rev[n])
return seen
def detect_destabilizing_widening(blocks: dict[str, Block], widened_state: dict) -> list:
diagnostics: list = []
for lbl, block in blocks.items():
if not isinstance(block.term, Branch) or widened_state[lbl] is None:
continue
cond = block.term.cond
loop_nodes = _reachable(blocks, block.term.true_label) & _can_reach(blocks, lbl)
for side in (cond.a, cond.b):
if not isinstance(side, Var):
continue
var = side.name
a = eval_expr(cond.a, widened_state[lbl])
b = eval_expr(cond.b, widened_state[lbl])
prov = None
if cond.op == "<" and isinstance(cond.a, Var) and cond.a.name == var:
prov = ("hi", b)
elif cond.op == "<=" and isinstance(cond.a, Var) and cond.a.name == var:
prov = ("hi", b)
elif cond.op == ">" and isinstance(cond.a, Var) and cond.a.name == var:
prov = ("lo", b)
elif cond.op == ">=" and isinstance(cond.a, Var) and cond.a.name == var:
prov = ("lo", b)
elif cond.op == "<" and isinstance(cond.b, Var) and cond.b.name == var:
prov = ("lo", a)
elif cond.op == ">" and isinstance(cond.b, Var) and cond.b.name == var:
prov = ("hi", a)
if prov is None:
continue
updates = [st.expr for n in loop_nodes for st in blocks[n].stmts
if isinstance(st, Assign) and st.var == var]
if not updates:
continue
kinds = {_classify_expr(var, u) for u in updates}
if kinds <= {"inc", "same"}:
kind = "inc"
elif kinds <= {"dec", "same"}:
kind = "dec"
else:
kind = "nonmono"
if kind != "nonmono":
continue
cur = widened_state[lbl].get(var, TOP)
bound_iv = prov[1]
escaped = (prov[0] == "hi" and cur.hi == POS and bound_iv.hi != POS) or \
(prov[0] == "lo" and cur.lo == NEG and bound_iv.lo != NEG)
if escaped:
diagnostics.append(Diagnostic(
loop=lbl, var=var, provenance=f"{var} {cond.op} {prov[1]}",
widened=cur, update_kind=kind,
message=(f"destabilizing widening at loop '{lbl}': counter '{var}' "
f"has non-monotone update and widened to {cur}, "
f"escaping provenance bound {prov[0]}={prov[1]}")))
return diagnostics
# --------------------------------------------------------------------------- #
# Top-level driver
# --------------------------------------------------------------------------- #
def analyze(stmts, initial_env=None, entry="entry", exit_="exit") -> Report:
cfg = CFGBuilder()
entry = cfg.build(stmts, entry=entry, exit_=exit_)
result = solve(cfg.blocks, entry, initial_env=initial_env)
checks = collect_checks(cfg.blocks, result["in_state"])
diagnostics = detect_destabilizing_widening(cfg.blocks, result["widened_state"])
return Report(entry=entry, widening_points=result["widening_points"],
in_state=result["in_state"], out_state=result["out_state"],
checks=checks, diagnostics=diagnostics)
A compact harness that asserts every requirement:
import math
from ia import *
def C(n): return Const(n)
def V(n): return Var(n)
def add(a, b): return Bin("+", a, b)
def div(a, b): return Bin("/", a, b)
def lt(a, b): return Cmp("<", a, b)
# --- safe counting loop: prove div + bounds, recover finite bound by narrowing
counting = [Assign("i", C(0)), Assign("s", C(0)),
While(lt(V("i"), C(10)), [
Assign("s", add(V("s"), div(C(100), add(V("i"), C(1))))),
CheckBounds(V("i"), 10, point="loop"),
CheckDiv(add(V("i"), C(1)), point="loop"),
Assign("i", add(V("i"), C(1))),
])]
r = analyze(counting)
assert r.checks and all(c.proven for c in r.checks), r.checks
hdr = next(iter(r.widening_points))
assert r.in_state[hdr]["i"].hi != math.inf # narrowing recovered [0,10]
# --- non-monotone counter -> destabilizing-widening diagnostic
nonmono = [Assign("i", C(0)),
While(lt(V("i"), C(100)), [Assign("i", add(Bin("*", V("i"), C(2)), C(1)))])]
assert len(analyze(nonmono).diagnostics) == 1
# --- monotone counter -> no diagnostic
mono = [Assign("i", C(0)),
While(lt(V("i"), C(100)), [Assign("i", add(V("i"), C(1)))])]
assert analyze(mono).diagnostics == []
# --- divergent loop still terminates via widening
divergent = [Assign("i", C(0)),
While(Cmp("==", C(1), C(1)), [Assign("i", add(V("i"), C(1)))])]
r = analyze(divergent)
print("divergent loop terminated; header i =",
r.in_state[next(iter(r.widening_points))]["i"])
print("ALL CHECKS PASSED")
Observed results:
$ python3 test_ia.py
ok test_counting_loop
ok test_divergent_terminates
ok test_guarded_div
ok test_monotone_no_diagnostic
ok test_nonmonotone_diagnostic
ok test_raw_goto
ok test_straightline_div
ok test_wide_bound
8 tests passed
$ python3 verify.py
Interval report for the counting-loop benchmark
B2 in={'i': '[0,9]', 's': '[0,+inf]'} out={'i': '[1,10]', 's': '[10,+inf]'}
WhileH1 in={'i': '[0,10]', 's': '[0,+inf]'} out={'i': '[0,10]', 's': '[0,+inf]'}
exit in={'i': '[10,10]','s': '[0,+inf]'} out={'i': '[10,10]','s': '[0,+inf]'}
widening points: {'WhileH1'}
Safety checks
[PROVEN] bounds at loop index/divisor=[0,9] length=10
[PROVEN] div at loop index/divisor=[1,10]
Top-only comparison
straightline_div interval proved 1/1 top-only 0/1
guarded_div interval proved 1/1 top-only 0/1
counting_loop interval proved 2/2 top-only 0/2
wide_bound interval proved 1/1 top-only 0/1
-> interval analysis proves 5 additional checks
Diagnostics
nonmonotone diagnostics=1
destabilizing widening at loop 'WhileH1': counter 'i' has non-monotone update
and widened to [0,+inf], escaping provenance bound hi=[100,100]
monotone diagnostics=0
Termination fuzzing: 400 random structured programs analyzed; all terminated
Divergent program terminated; header i = [0,+inf]
ALL VERIFICATION CHECKS PASSED
| Requirement | Evidence |
|---|---|
| Interval at every program point | in_state/out_state per block, printed by verify.py |
| Chaotic worklist fixpoint | solve() phase 1 (run(widen=True)) |
| Widening guarantees termination | divergent loop reaches i ∈ [0,+∞] and stops; 400 fuzz programs terminate |
| Narrowing recovers precision | header is [0,10], not [0,+∞] |
| Zero false alarms on benchmark | every check in the 4 safe programs is PROVEN |
| Beats a Top-only lattice | proves 5 checks that Top-only proves 0 of |
| Destabilizing-widening diagnostic | fires on i := i*2+1, silent on i := i+1 |
assign/guard/if/while/goto |
structured front-end + raw Goto/Branch CFG test |
Every CFG cycle has a back-edge target registered as a widening point; there the transfer is composed with ∇, whose ascending chain over {program constants, ±∞} is finite. The narrowing pass is a descending chain over the same finite set. A global max_iters guard converts any residual bug into an explicit RuntimeError instead of a hang.
0 yields ⊤ for the result and an alarm — never silently assumed safe._mul uses the 0·±∞ = 0 convention for endpoint products.Files: ~/interval_analysis/ia.py (analyzer), test_ia.py (benchmarks/tests), verify.py (extended verification), SOLUTION.md (the full generated report). I extracted the three embedded Python blocks from SOLUTION.md into a fresh directory and re-ran both suites — they reproduce the full verification exactly.
# Evidence - Problem class: python-abstract-interpretation-widening-narrowing-fixpoint - Model: openrouter/deepseek/deepseek-v4.1-flash - Solved: 2026-09-16T04:16:19.817Z - Verification: solution produced by pi in sandbox; see signatures.json
{"description": "Implement an interval abstract-interpretation analyzer in Python over a small imperative IR (assign, guard, if, while, goto) that computes a fixpoint with chaotic iteration on a worklist, widening at loop headers so termination is guaranteed for divergent programs, and a narrowing pass that recovers precision lost to widening. Report the interval at every program point, use them to prove division-by-zero and array-bounds safety, and emit a destabilizing-widening diagnostic whenever a loop counter's update is non-monotone and escapes its provenance bound. The analyzer must terminate on every input, produce zero false alarms on the supplied benchmark suite, and still prove at least the expected set of checks that a naive Top-only lattice cannot.", "environment": "python3", "language": "python", "model": "openrouter/deepseek/deepseek-v4.1-flash", "problem_class": "python-abstract-interpretation-widening-narrowing-fixpoint", "provider": "openrouter", "solved_at": "2026-09-16T04:16:19.817Z", "version": "3.11"}