Skip to content

Fix #1203: [Rule] ConjunctiveQueryFoldability to Satisfiability - #1241

Open
isPANN wants to merge 1 commit into
mainfrom
ar/issue-1203-cqf-sat
Open

isPANN wants to merge 1 commit into
mainfrom
ar/issue-1203-cqf-sat

Conversation

@isPANN

@isPANN isPANN commented Nov 4, 2026

Copy link
Copy Markdown
Collaborator

Fixes #1203

Adds the reduction ConjunctiveQueryFoldability → Satisfiability.

  • Rule: src/rules/conjunctivequeryfoldability_satisfiability.rs. Candidate substitution values are compressed to the terms that can occur in the image; one-hot selectors per undistinguished variable, pairwise atom-match variables, and row/column cover clauses encode the exact image-set equality. Overhead is registered as upper bounds for num_vars, num_clauses and num_literals; arithmetic is checked and overflow returns a typed error instead of panicking.
  • Result mapping: identity aggregate mapping (YES ↔ SAT, NO ↔ UNSAT); witness extraction reads the unique true selector of each variable and rejects unsatisfying or wrong-length witnesses.
  • Tests: 45 unit tests (closed loop incl. a non-idempotent swap, YES/NO worked examples, exhaustive small families, sampled grid, boundary/overflow cases, extraction rejection).
  • Example and paper: canonical rule example registered in src/rules/mod.rs; docs/paper/reductions.typ has the rule entry with proof and worked YES/NO examples.

make check and make paper pass locally.

🤖 Generated with Claude Code

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
@isPANN

isPANN commented Nov 4, 2026

Copy link
Copy Markdown
Collaborator Author

Verification certificate

Verdict: VERIFIED. The construction was checked by two independently written scripts and a Typst proof, and a third comparison ran the real pred reduce against both. All checks pass; there are no disagreements.

Type chain. ConjunctiveQueryFoldability (no variant parameters; witness = a vector of u term indices; value Or) → Satisfiability (CNF; value Or). The reduction is registered as an identity aggregate mapping (YES ↔ SAT, NO ↔ UNSAT). Overhead (upper bounds, with R = d+x+u, p/q the atom counts of the two queries): num_vars = uR + pq, num_clauses = u(1+R(R−1)/2) + pq(u+1) + p + q, num_literals = uR² + pq(3u+3).

Constructor script

TOTAL: 112,399 checks, 0 failures, 11,674 instances (2,392 target-satisfiable models extracted back, 9,282 NO-side checks, 532 raw-Boolean instances with full 2^n enumeration).

Section Checks
1 symbolic overhead 9
2 forward/backward (source feasible ⇔ target satisfiable) 14,612
3 extraction from every satisfying target witness 47,527
4 measured sizes vs exact formulas and bounds 23,350
5 well-formedness 26,894
6 worked YES example 4
7 worked NO example 3

Instance ranges: exhaustive unary list pairs (length ≤ 2) over two terms for the splits (d,x,u) = (0,0,2), (0,1,1), (1,0,1), (1,1,0) (196 instances); nullary lists with R = 0 (49); 12 special cases; full-arity atoms (k = u) for u = 1..5, plain, repeated and conflicting (14); seeded random samples of 300 instances in each of the 36 cells u = 0..5 × p+q = 0..5 (10,800 instances, random.Random(2026101305)); Hypothesis mixed and repeated-variable strategies, 300 examples each (@seed(2026101319) and @seed(2026101320), max_examples=300, database=None, deadline=None); both worked examples.

Independent check script

Written from the proof alone, with its own construction, source and target evaluators, a generic all-models DPLL and raw 2^n enumeration. TOTAL: 1,362,078 checks, 0 failures, 44,493 instances (7,883 YES, 36,610 NO).

Section Checks
1 symbolic overhead (sympy) anchored to measured output 36
2 forward/backward 662,041
3 extraction from every satisfying witness 80,135
4 measured sizes vs formulas and registered bounds 177,972
5 structure: numbering, one-hot, match semantics, witness rejection 424,610
6 worked YES example 14
7 worked NO example 10
8 special cases and prose claims 17,260

Instance ranges: every ordered pair of atom lists of length ≤ 2 for a binary relation under all splits with R ≤ 2 and under (0,0,3), (0,1,2), (1,1,1); a ternary relation with R = 2, splits (0,0,2) and (0,1,1); the signature (nullary, nullary, unary) with R ≤ 2; a unary relation under (0,0,4), (0,1,3), (0,0,5), (1,0,4) (reaching u = 5); nullary lists of length ≤ 3 with R = 0 (42,695 exhaustive instances); 871 seeded random instances (seed 2026110401); 25 special cases (swap, empty queries, R = 0, one-sided containment, duplicates, d up to 2^62, overflow as a typed error); both worked examples; three Hypothesis strategies (general: arities 0–3 with duplicates and d up to 10^12; planted: an arbitrary, possibly non-idempotent substitution then keep/drop/add/reverse a target atom; repeated: arity up to 6) with 300 examples each, seeds 2026110411/12/13, database=None, deadline=None. Source truth is full-domain brute force whenever R^u ≤ 4096, otherwise a restricted search validated against brute force wherever both apply. Mutation testing: 24 broken constructions and 17 solver/oracle mutants; each check of both scripts is killed by at least one.

Cross-comparison

  • Constructor vs independent script (no pred), 10,906 instances: targets identical as multisets on 10,904 and in exact clause order on 10,904 (161,207 clauses compared); model sets identical on 10,898; feasibility agrees on 10,898; 5,084 cross-extractions; both reject the 2 overflow instances; 0 disagreements.
  • Real pred reduce vs both scripts, 634 instances: pred == constructor == independent script on all 634, in exact clause order (9,026 clauses); 1,059 evaluator comparisons; the 2 overflow instances are rejected by all three; 0 problems.

Scripts

Both scripts need python3 with sympy and hypothesis; the independent script is split into a core module and a harness that imports it as core.

Typst proof
#set page(paper: "a4", margin: 22mm)
#set text(size: 10pt)
#set heading(numbering: "1.")

= Exact-image conjunctive query foldability to CNF satisfiability

*Theorem.* For the concrete, parameter-free variants
`ConjunctiveQueryFoldability` and `Satisfiability`, the construction below
preserves existence of a witness in both directions. Its completed feasibility
value mapping is the identity on `Or`. It makes no claim to preserve numbers
of witnesses. The size is polynomial in the explicit queries and the explicit
source witness dimension; the numeric representation limits are stated below.

== Resolved endpoint types and provenance

The type-resolution gate is *PASS*. In
`src/models/misc/conjunctive_query_foldability.rs:316–332`, the concrete
`impl Problem for ConjunctiveQueryFoldability` declares
`Solution = Vec<usize>` and `Value = crate::types::Or`, with empty variant
parameters. In `src/models/formula/sat.rs:208–228`, the concrete
`impl Problem for Satisfiability` declares `Solution = Vec<bool>` and the
same `Value = crate::types::Or`, also with empty variant parameters.
There are no generic substitutions or aliases to resolve.
`src/types.rs:533–563` defines `Or(pub bool)` and its existential disjunction
aggregate. Thus the full-domain value compatibility is `Or → Or`.
`docs/src/design.md`, “Result mappings”, prescribes an identity aggregate
mapping in addition to witness extraction for this decision equivalence.

Search before derivation: the source
#link("https://github.com/CodingThrust/problem-reductions/issues/448")[model proposal],
the #link("https://github.com/CodingThrust/problem-reductions/pull/687")[model implementation],
and the #link("https://github.com/CodingThrust/problem-reductions/issues/463")[incoming coloring proposal]
were consulted. They supply model history, not a theorem for this SAT encoding.
The incoming proposal uses containment language and cannot justify dropping
either exact-image cover. The existing sibling rules
`src/rules/naesatisfiability_satisfiability.rs` and
`src/rules/ksatisfiability_oneinthreesatisfiability.rs` were read for CNF
conventions, compaction, validation before decoding, and identity completed-value
mapping. No source rule module currently mentions this source model.

#link("https://doi.org/10.1145/800105.803397")[Chandra and Merlin, STOC 1977, pp. 77–90],
#link("https://www.cs.ubc.ca/~laks/cpsc504/chandra-merlin-1977.pdf")[primary paper],
defines folding on printed p. 81 using a homomorphism fixing its image.
Theorem 11 on p. 82 concerns that definition; Lemma 13 on p. 83 relates
query equivalence and homomorphisms. Those statements do not certify the
library's one-step exact-image predicate. The complete compressed SAT
construction and its proof are *derived here*, following the proposed
one-hot/match encoding, rather than attributed to that paper.

On baseline `e7bacb37`, a locally built `pred` was run with
`show ConjunctiveQueryFoldability`, `show Satisfiability`, and
`path ConjunctiveQueryFoldability Satisfiability --limit all`.
Both names fully qualify the respective unique, parameter-free variants.
The source has zero outgoing edges, and path search reports no reduction path.
Redundancy result: *PASS (no existing witness path; therefore no dominating path)*.
This is a graph observation at the recorded baseline, not a worldwide novelty claim.

== Admitted instances and notation

Let `d`, `x`, and `u` be the nonnegative counts of constants, distinguished
variables, and undistinguished variables. Let $R = d+x+u$.
Their disjoint term sets are encoded by consecutive integers:
constants in $[0,d)$, distinguished variables in $[d,d+x)$, and variable
$y$ in $[0,u)$ by $d+x+y$. An atom consists of a relation identifier and
an ordered argument tuple whose length is that relation's declared arity.
Arity zero is permitted. Every identifier and term must be in range.
Write $Q_1=(A_0, dots, A_(p-1))$ and $Q_2=(B_0, dots, B_(q-1))$.
Lists may be empty and may contain duplicate atoms. Relations have no
additional database interpretation in the evaluator.

A source witness is a vector $w$ of exactly $u$ integers in $[0,R)$.
Its substitution $sigma$ fixes terms below $d+x$ and replaces the term
$d+x+y$ with $w_y$, once. It does not substitute into the resulting term.
It is feasible exactly when the set of atoms $sigma(Q_1)$ equals the set
of atoms $Q_2$. In particular, coverage of only one set by the other is insufficient.

A target is unrestricted CNF with signed, one-based literal indices.
Its witness has exactly one Boolean per target variable. A clause is a
disjunction; the empty clause is false. The conjunction of zero clauses is
true. `Satisfiability::try_new` retains clause/literal multiplicities and
requires the variable count and nonzero literal magnitudes to fit `i64::MAX`.
`Problem::evaluate` validates the witness length, unlike the lower-level
clause helper that supplies false for a missing variable. The checker follows
the full `Problem` interface on both endpoints.

== Construction

1. Let $C$ be the sorted distinct terms occurring in $Q_2$, with term zero
   adjoined when $R>0$. Write $c=|C|$, and denote its zero-based entry $t$ by
   $C_t$. If $R=0$, then $u=0$ and $C$ is empty. No domain enumeration occurs.
2. Allocate selectors $e_(y,t)$ with one-based index $y c+t+1$, for
   $0<=y<u$ and $0<=t<c$. For every variable, emit the disjunction of its
   selectors and every pairwise clause $not e_(y,t) or not e_(y,s)$ for
   $t<s$. These clauses require exactly one chosen candidate.
3. Allocate match variable $z_(i,j)$ with index $u c+i q+j+1$, for
   $0<=i<p$ and $0<=j<q$. A pair is incompatible if its relation identifiers
   differ, a fixed source argument differs from the corresponding target
   term, or two occurrences of the same source variable require different
   target terms. For each incompatible pair emit the negative unit
   $not z_(i,j)$.
4. For a compatible pair, let $E_(i,j)$ be the distinct selectors required
   by argument equality, ordered by source variable. Each appearing source
   variable contributes one selector for its corresponding target term;
   the term belongs to $C$ by definition. Put $k_(i,j)=|E_(i,j)|$.
   Emit $not z_(i,j) or e$ for every $e in E_(i,j)$, and emit
   $z_(i,j) or or.big_(e in E_(i,j)) not e$. Thus the match is equivalent
   to the conjunction of its requirements. When there are no requirements,
   the last clause is the positive unit $z_(i,j)$.
5. For every source atom emit its row cover $or.big_j z_(i,j)$, and for
   every target atom emit its column cover $or.big_i z_(i,j)$.
   Preserve empty covers as empty clauses; never omit them.
6. The output variable count is $u c+p q$. Clause order is: one-hot blocks
   in source-variable order (positive clause then lexicographic pairs),
   match blocks in source-atom-major order (forward clauses in variable
   order then reverse clause), then rows and columns in list order.
   Literal order is the displayed order. Cross-implementation comparison
   may permute clauses and literals, but must preserve their multiplicities
   and the stated variable numbering. No other isomorphism is assumed.

== Correctness

*Compression lemma.* Suppose a source witness is feasible. For a variable
appearing in $Q_1$, its substituted term appears in an image atom. Exact
image equality implies that term occurs in $Q_2$, hence belongs to $C$.
A variable absent from $Q_1$ can instead map to zero without changing any
image atom. Its existence implies $u>0$, hence $R>0$, so this replacement
is legal and belongs to $C$. Consequently there is a candidate-valued
witness for every feasible source instance. Every candidate is a legal
source term, establishing the converse inclusion needed for extraction.

*(⇒)* Start from a feasible source witness and compress its unused
coordinates as in the lemma. Set exactly the selected candidate bit for
each variable. Set a match bit true exactly when the actual substituted
source atom equals the target atom. An incompatible pair cannot be equal:
the relation, a fixed argument, or consistency of a repeated variable
prevents equality. For a compatible pair, equality of every argument is
equivalent to the conjunction of its distinct required selectors, so all
its equivalence clauses hold. Exact equality of the image and target atom
sets means every source image is a target atom and every target atom has
a preimage. Therefore every row and column cover holds, and the constructed
CNF has a satisfying witness.

*(⇐)* Start independently from any satisfying assignment of the CNF.
The one-hot clauses select exactly one legal candidate for every source
variable. Interpret these choices as the vector $w$. Incompatible pairs
have false match bits. For every other pair, the forward implication
clauses and reverse conjunction clause force the match bit to equal
argumentwise equality under this $w$, including all fixed arguments and
repeated occurrences. Row covers then show that every image atom belongs
to the target set. Column covers show that every target atom occurs in
the image set. These two inclusions establish exact set equality, so
$w$ is feasible. This argument applies to every satisfying target witness,
not just a witness produced by the forward map.

*Completed values.* Existence is equivalent in both directions. The
existential aggregate is therefore `Or(true)` on both sides or `Or(false)`
on both sides. Its total mapping is identity. Candidate compression may
change the number of satisfying witnesses and is not a counting reduction.

== Solution extraction

Validate the complete target witness once through the target evaluator,
including its exact length, and reject invalid or unsatisfied vectors.
For each $y$, read its unique true selector and return the encoded term
$C_t$ in coordinate $y$. All match bits and cover clauses have already
been validated. The backward proof establishes feasibility of this exact
vector. Never recursively substitute, normalize its values, default a
missing choice, repair a non-solution, or silently truncate a witness.
When $u=0$, return the empty vector after validation; no choice is read.

== Overhead and arithmetic

Let $H$ be the set of compatible atom pairs and let
$K=sum_((i,j) in H) k_(i,j)$. We have $0<=K<=p q u$ and $c<=R$.
One-hot blocks contribute $u(1+c(c-1)/2)$ clauses and $u c^2$ literals.
Each incompatible match contributes one clause and one literal.
Each compatible match contributes $k_(i,j)+1$ clauses and
$3k_(i,j)+1$ literal occurrences. Covers contribute $p+q$ clauses and
$2p q$ literals, including zero-length clauses. Nothing is deduplicated or
normalized out. Thus the exact measured sizes are as follows.

#table(columns: (1fr, 1.4fr, 2.2fr), inset: 5pt,
  [Target parameter], [Exact constructed size], [Registered upper bound],
  [`num_vars`], [$u c+p q$], [$u R+p q$],
  [`num_clauses`], [$u(1+c(c-1)/2)+p q+K+p+q$],
    [$u(1+R(R-1)/2)+p q(u+1)+p+q$],
  [`num_literals`], [$u c^2+3K+3p q$], [$u R^2+p q(3u+3)$],
)

For variables the gap is $u(R-c)$. For clauses it is
$u(R-c)(R+c-1)/2 + (p q u-K)$; for literals it is
$u(R-c)(R+c)+3(p q u-K)$. These are nonnegative when $R>=1$.
If $R=0$, we have $u=c=K=0$, and all three formulas hold directly;
the polynomial product $R(R-1)$ equals zero. This handles the boundary
without performing an unsigned subtraction at zero.

The registered source parameters are `domain_size`, `num_distinguished`,
`num_undistinguished`, `num_conjuncts_q1`, `num_conjuncts_q2`, and
`num_relations`. Substitute their first three counts for $R$, and the fourth
and fifth for $p,q$. The three target parameter names above are taken from
the target's own registration. The candidate count and $K$ are not registered
source parameters, so *all three registry fields are upper bounds*.
Agreement with a particular example cannot make them exact.

If $a_2$ is the number of target argument occurrences, then
$c<=min(R,1+a_2)$; when $R=0$, $c=0$. Computing the candidate list scans
explicit arguments. Matching scans explicit atom arguments and distinct
variable requirements. Emitting selectors costs $O(u c^2)$, match clauses
cost $O(p q(u+1))$ apart from the explicit argument scans, and covers cost
$O(p q+p+q)$. Comparison of encoded terms includes their bit lengths.
There is no enumeration proportional to a large declared constant domain.
This is polynomial in the explicit queries and explicit coordinate dimension
$u$, not in a binary encoding of $u$ alone: an empty query pair can declare
arbitrarily many unused variables, and the source still requires a vector
of length $u$. A stronger succinct-input polynomiality claim is not established.

Machine arithmetic is separate from feasibility. On the audited 64-bit host,
declared source counts are `usize`; target literals are `i64`. Check the
total term range, products $u c,p q$, their sum and emitted collection sizes
before machine arithmetic or signed conversion. Out-of-range sizes must
return typed construction/reduction errors, never wrap, panic or report NO.
Use the target's fallible constructor and retain its construction error.
The metadata formulas use exact integer/rational arithmetic and may exceed
`u64`; evaluation must then return `OutputOutOfRange`, even if the compressed
constructed target is tiny. This is not a mathematical failure of the bound.

*Baseline discrepancy.* The current source `validate()` checks atom and
term indices, but does not check the sum of its declared counts. Its
`evaluate()` and `dimensions()` add them unchecked. The proposal's statement
that source construction/loading already rejects this overflow is false on
the inspected baseline. The Python constructor explicitly rejects an
unrepresentable total instead of inventing wrapped semantics. A subsequent
Rust implementation must guard it before using the source's unchecked
helpers; no claim about existing source overflow rejection is assumed here.
Arithmetic boundary behavior of that future implementation remains to be tested.

== Special cases

- *Repeated atoms:* retain their separate match rows and columns and count
  their emitted clauses. Equal duplicates impose repeated coverage of the
  same atom and do not require an injective matching or multiset equality.
- *Repeated variable inside an atom:* a target tuple demanding two different
  images is incompatible, even if the two selectors could both be mentioned
  syntactically. Consistent repetitions contribute one requirement, not one
  per occurrence. A source loop cannot become a tuple with unequal arguments.
- *Nullary relations:* equal relation identifiers give an empty conjunction
  and a positive match unit. Different identifiers give a negative match
  unit. Distinct symbols of equal arity remain distinct.
- *No substitutable variables:* with $u=0$, there are no selectors. Matches
  are positive or negative units according to fixed atom equality. The target
  is feasible exactly when the two fixed atom sets agree.
- *Empty term universe:* $R=0$ implies $u=0$ and $C$ empty. Only nullary
  atoms are legal, and the same match and cover rules decide equality.
- *Empty queries:* both empty give only one-hot clauses, satisfiable by
  choosing zero for every unused variable if one exists. With no variables,
  the empty target has the unique empty witness. If exactly one query is
  empty and the other nonempty, its corresponding covers are empty clauses,
  so both source and target are NO.
- *Fixed terms:* unequal constants or distinguished variables cannot be
  changed by substitution and force incompatible matches. Candidate zero
  does not allow changing a fixed argument.
- *One-step maps:* with $d=x=0,u=2$ and a binary relation, the source atom
  `(0,1)` can map to `(1,0)` using witness `(1,0)`. This is a legal swap
  that is not idempotent; recursive substitution or a retraction condition
  would change the answer or extracted witness.
- *Compression:* unused variables may have source images outside $C$;
  setting them to zero preserves their atom image. Large legal constant
  domains are handled by collecting just the target terms and zero.

== Worked YES example

Use $d=0,x=1,u=3,R=4$ and a binary relation named `R`, distinct from the
integer $R$ above. The term codes are `x=0,u=1,v=2,a=3`.

Q1 = [R(x,u), R(u,v), R(v,x), R(u,u)]
Q2 = [R(x,a), R(a,a), R(a,x)]


The source vector `(3,3,3)` gives image list
`[R(x,a),R(a,a),R(a,x),R(a,a)]`, whose set equals the target set.
Here $p=4,q=3,c=2$, with candidates `(0,3)`.
The numbers of requirements for the compatible matches, with a dash marking
an incompatible pair, are

[1, -, -]
[2, 2, 2]
[-, -, 1]
[-, 1, -]


Thus $|H|=6$ and $K=9$. The complete 18-bit satisfying witness is

[0,1,0,1,0,1, 1,0,0, 0,1,0, 0,0,1, 0,1,0]


The selectors use six variables. The twelve match bits form rows
`(1,0,0),(0,1,0),(0,0,1),(0,1,0)`. Every row and column is covered.
The one-hot blocks have 6 clauses and 12 literals; matches have
$12+9=21$ clauses and $12+27=39$ literals; covers have
$4+3=7$ clauses and $2 dot 4 dot 3=24$ literals.
Total: *18 variables, 34 clauses, 75 literals*. Extraction returns `(3,3,3)`.
The registry upper bounds are *24 variables, 76 clauses, 192 literals*.
The unused last coordinate gives four source witnesses but only two target
witnesses, because the target restricts it to candidates zero and three.

== Worked NO example

Keep the same four term codes, three substitutable variables, relation and
source query, and replace the target query with

Q2 = [R(x,v), R(v,a), R(a,x)]


Every target atom has unequal arguments, whereas the last source atom
`R(u,u)` has equal arguments under every substitution. It has no target
image, so no source witness exists. Candidates are `(0,2,3)`, hence $c=3$.
Compatible match requirement counts are

[1, -, -]
[2, 2, 2]
[-, -, 1]
[-, -, -]


Thus $|H|=5,K=8$. Selector indices are 1 through 9 and match indices
10 through 21. Negative units force bits 19, 20 and 21 false, contradicting
the final row cover `(19,20,21)`. No target witness exists.
One-hot blocks contribute 12 clauses and 27 literals; match blocks contribute
20 clauses and 36 literals; covers contribute 7 clauses and 24 literals.
Total: *21 variables, 39 clauses, 87 literals*. The same registered upper
bounds *24, 76, 192* apply. This checks an infeasible source and target,
rather than just failure of one attempted witness.

== Executable claim coverage and limits

`constructor.py` implements seven nonempty sections, numbered in its audit:

#table(columns: (1fr, 3fr), inset: 5pt,
  [Section], [Claims tested],
  [1 — symbolic], [Sympy derives one-hot pair counts, exact clause/literal sums,
    and all three upper-bound gaps; measured target counts anchor each identity.
    Empty-universe normalization is tested.],
  [2 — forward/backward], [Full-domain source enumeration versus complete generic
    DPLL target enumeration; every distinct compressed feasible source extends.
    Small raw Boolean enumeration also checks DPLL completeness.],
  [3 — extraction], [Every satisfying target witness is decoded and checked by
    direct atom-set evaluation; complete solution sets agree modulo unused
    coordinates. Unsatisfied witnesses are rejected; a non-idempotent swap is recovered.],
  [4 — measured sizes], [All exact counts and registered bounds against actual
    output, using independent local atom-equality searches to count compatible
    requirements. Large compressed domains keep their target size.],
  [5 — structure], [Candidate and literal ranges, numbering validity, exact one-hot
    choices and match equality in every target model, malformed witness lengths
    and non-Boolean entries, and overflow as an error rather than infeasibility.],
  [6 — YES], [The exact candidates, sizes, complete bit vector, and extracted
    substitution above.],
  [7 — NO], [The exact candidates and sizes above; exhaustive infeasibility on
    both sides, not a single unsuccessful assignment.],
)

The finite exhaustive family is every ordered unary query-list pair of length
at most two over two terms, for splits `(d,x,u)` equal to `(0,0,2)`, `(0,1,1)`,
`(1,0,1)`, `(1,1,0)`: 196 instances. There are another 49 exhaustive pairs of
nullary lists of length at most two over two distinct relation symbols with
empty term universe. For each of the 36 cells $u=0,dots,5$ and
$p+q=0,dots,5$, 300 samples use seed `2026101305`, constant and distinguished
counts independently zero or one, and arities zero through two. Two distinct
Hypothesis strategies use 300 examples each, seeds `2026101319` and
`2026101320`, `database=None`, `deadline=None`: mixed legal queries and
queries stressing repeated atoms/variables with nullary atoms.
Special instances cover every case above and large compressed universes.
Fourteen additional instances use every $u=1,dots,5$ variable in one atom,
then repeat those arguments consistently and inconsistently (arity up to ten).
These exercise the $k=u$ match-size boundary beyond binary relations.
Exhaustive enumeration of all queries through five variables is infeasible:
arities and list lengths are unbounded. Sampling does not claim that coverage.
Repeated source witnesses having the same compressed vector are enumerated
but do not inflate the forward check count.

The theorem and nonnegative gap arguments cover all finite mathematical
instances. Finite checks search for counterexamples to the concrete encoding;
they do not prove unbounded claims, complexity, or future Rust integer safety.
Those claims rest on the argument above, and future arithmetic error handling
needs separate Rust boundary tests. This document completes the constructor
stage only; the independent check script and cross-comparison must still pass
before the full verification certificate may say VERIFIED.
Constructor script
#!/usr/bin/env python3
"""Independent executable construction; run with the user's Python + sympy/Hypothesis.

Encoded atoms are (relation, tuple of zero-based concatenated D,X,Y terms).
No library code or saved expected targets are imported.
"""

from dataclasses import dataclass
from itertools import combinations, product
from collections import Counter
import random

import sympy as sp
from hypothesis import given, seed, settings, strategies as st


@dataclass(frozen=True)
class Source:
    d: int
    x: int
    u: int
    arities: tuple
    q1: tuple
    q2: tuple

    @property
    def r(self):
        return self.d + self.x + self.u

    def __post_init__(self):
        if any(type(v) is not int or not 0 <= v < 2**64
               for v in (self.d, self.x, self.u, *self.arities)):
            raise ValueError("invalid source dimensions")
        for rel, args in self.q1 + self.q2:
            if not 0 <= rel < len(self.arities) or len(args) != self.arities[rel]:
                raise ValueError("invalid relation or arity")
            if any(type(t) is not int or not 0 <= t < self.r for t in args):
                raise ValueError("invalid encoded term")


@dataclass(frozen=True)
class Target:
    candidates: tuple
    num_vars: int
    clauses: tuple

    @property
    def sizes(self):
        return self.num_vars, len(self.clauses), sum(map(len, self.clauses))


def image_atom(s, atom, witness):
    rel, args = atom
    fixed = s.d + s.x
    return rel, tuple(t if t < fixed else witness[t - fixed] for t in args)


def source_ok(s, witness):
    if len(witness) != s.u or any(type(t) is not int or not 0 <= t < s.r
                                 for t in witness):
        raise ValueError("invalid source witness")
    return {image_atom(s, a, witness) for a in s.q1} == set(s.q2)


def target_ok(t, bits):
    if len(bits) != t.num_vars or any(type(b) is not bool for b in bits):
        raise ValueError("invalid target witness")
    return all(any(bits[abs(lit) - 1] == (lit > 0) for lit in clause)
               for clause in t.clauses)


def construct(s):
    if s.r >= 2**64:
        raise OverflowError("term universe does not fit usize on the audited 64-bit host")
    candidates = tuple(sorted({t for _, args in s.q2 for t in args}
                              | ({0} if s.r else set())))
    c, p, q = len(candidates), len(s.q1), len(s.q2)
    n = s.u*c + p*q
    if n > 2**63 - 1:
        raise OverflowError("SAT variable count exceeds signed literal range")
    clauses = []
    for y in range(s.u):
        selectors = tuple(y*c + j + 1 for j in range(c))
        clauses.append(selectors)
        clauses.extend((-a, -b) for a, b in combinations(selectors, 2))
    candidate_index = {t: j for j, t in enumerate(candidates)}
    for i, (ra, aa) in enumerate(s.q1):
        for j, (rb, ab) in enumerate(s.q2):
            z = s.u*c + i*q + j + 1
            required = {}
            compatible = ra == rb
            if compatible:
                for a, b in zip(aa, ab):
                    if a < s.d + s.x:
                        compatible &= a == b
                    else:
                        y = a - s.d - s.x
                        if y in required and required[y] != b:
                            compatible = False
                        required[y] = b
            if not compatible:
                clauses.append((-z,))
            else:
                selectors = [y*c + candidate_index[b] + 1
                             for y, b in sorted(required.items())]
                clauses.extend((-z, e) for e in selectors)
                clauses.append((z, *(-e for e in selectors)))
    clauses.extend(tuple(s.u*c + i*q + j + 1 for j in range(q)) for i in range(p))
    clauses.extend(tuple(s.u*c + i*q + j + 1 for i in range(p)) for j in range(q))
    return Target(candidates, n, tuple(clauses))


def extract(s, t, bits):
    if not target_ok(t, bits):
        raise ValueError("unsatisfied target witness")
    c = len(t.candidates)
    return tuple(t.candidates[next(j for j in range(c) if bits[y*c+j])]
                 for y in range(s.u))


def models(t):
    """Generic complete DPLL, including all free-variable completions.

    No selector, atom, or construction knowledge is used by this solver.
    Pure-literal elimination is deliberately absent: it loses satisfying models.
    """
    def visit(clauses, assigned):
        while True:
            if any(not clause for clause in clauses):
                return
            unit = next((clause[0] for clause in clauses if len(clause) == 1), None)
            if unit is None:
                break
            v, value = abs(unit), unit > 0
            if v in assigned and assigned[v] != value:
                return
            assigned[v] = value
            clauses = tuple(tuple(l for l in clause if l != -unit)
                            for clause in clauses if unit not in clause)
        if not clauses:
            free = [v for v in range(1, t.num_vars + 1) if v not in assigned]
            for values in product((False, True), repeat=len(free)):
                complete = assigned | dict(zip(free, values))
                yield tuple(complete[v] for v in range(1, t.num_vars + 1))
            return
        v = abs(min(clauses, key=len)[0])
        for value in (False, True):
            yield from visit(clauses + ((v if value else -v,),), assigned.copy())

    yield from visit(t.clauses, {})


def encode(s, t, witness):
    """Forward map uses atom equality, independently of match-clause generation."""
    selectors = tuple(witness[y] == a for y in range(s.u) for a in t.candidates)
    matches = tuple(image_atom(s, a, witness) == b for a in s.q1 for b in s.q2)
    return selectors + matches


COUNTS = Counter()
INSTANCES = Counter()
STATS = Counter()


def check(section, target, condition, description):
    # Every audit has a concrete constructed target in its data flow.
    if not isinstance(target, Target):
        raise TypeError("audit requires a constructed target")
    COUNTS[section] += 1
    if not condition:
        raise AssertionError(f"section {section}: {description}; target={target}")


def rejected(s, t, bits):
    try:
        extract(s, t, bits)
    except ValueError:
        return True
    return False


def audit(s, family, raw=False):
    t = construct(s)
    INSTANCES[family] += 1
    # All full-domain source witnesses, not just candidates: tests compression.
    source_solutions = set()
    used = {a - s.d - s.x for _, args in s.q1 for a in args if a >= s.d+s.x}
    for w in product(range(s.r), repeat=s.u):
        if source_ok(s, w):
            compressed = tuple(w[y] if y in used else 0 for y in range(s.u))
            if compressed not in source_solutions:
                bits = encode(s, t, compressed)
                check(2, t, target_ok(t, bits), "every distinct compressed feasible source witness extends")
            source_solutions.add(compressed)
    target_solutions = set(models(t))
    decoded = set()
    for bits in target_solutions:
        w = extract(s, t, bits)
        check(3, t, target_ok(t, bits) and source_ok(s, w), "every target model extracts a source witness")
        c, q = len(t.candidates), len(s.q2)
        check(5, t, all(sum(bits[y*c:(y+1)*c]) == 1 for y in range(s.u))
              and all(bits[s.u*c+i*q+j] == (image_atom(s,a,w) == b)
                      for i,a in enumerate(s.q1) for j,b in enumerate(s.q2)),
              "one-hot and match equivalence invariants in every target model")
        decoded.add(tuple(w[y] if y in used else 0 for y in range(s.u)))
    check(2, t, bool(source_solutions) == bool(target_solutions), "feasibility equivalence")
    check(3, t, decoded == source_solutions, "complete extracted solution sets modulo unused coordinates")
    STATS['target_models'] += len(target_solutions)
    STATS['yes' if target_solutions else 'no'] += 1
    if raw and t.num_vars <= 10:
        brute = {b for b in product((False, True), repeat=t.num_vars) if target_ok(t, b)}
        check(2, t, brute == target_solutions, "DPLL agrees with complete raw Boolean search")
        STATS['raw_boolean_instances'] += 1

    c, p, q = len(t.candidates), len(s.q1), len(s.q2)
    # Independent size oracle: try local assignments and compare actual atoms.
    k_total = 0
    for a in s.q1:
        variables = sorted({v - s.d - s.x for v in a[1] if v >= s.d+s.x})
        for b in s.q2:
            possible = False
            for values in product(t.candidates, repeat=len(variables)):
                w = [0] * s.u
                for y, value in zip(variables, values):
                    w[y] = value
                if image_atom(s, a, w) == b:
                    possible = True
                    break
            if possible:
                k_total += len(variables)
    exact = (s.u*c+p*q,
             s.u*(1+c*(c-1)//2)+p*q+k_total+p+q,
             s.u*c*c+3*k_total+3*p*q)
    upper = (s.u*s.r+p*q,
             s.u*(1+s.r*(s.r-1)//2)+p*q*(s.u+1)+p+q,
             s.u*s.r*s.r+p*q*(3*s.u+3))
    check(4, t, t.sizes == exact, "exact counts derived from independent pair feasibility")
    check(4, t, all(a <= b for a, b in zip(t.sizes, upper)), "all registered upper bounds")
    expected_candidates = tuple(sorted({a for _,args in s.q2 for a in args}
                                       | ({0} if s.r else set())))
    check(5, t, c <= s.r and t.candidates == expected_candidates
          and all(0 <= a < s.r for a in t.candidates)
          and all(type(l) is int and 1 <= abs(l) <= t.num_vars for clause in t.clauses for l in clause),
          "legal compressed terms, distinct ordering, signed one-based literals")
    check(5, t, rejected(s, t, (False,)*(t.num_vars+1))
          and (t.num_vars == 0 or (rejected(s, t, (False,)*(t.num_vars-1))
                                  and rejected(s, t, (0,)*t.num_vars))),
          "wrong witness lengths and non-Boolean entries rejected")
    for bits in (tuple(False for _ in range(t.num_vars)),
                 tuple(True for _ in range(t.num_vars)),
                 tuple(bool(i % 2) for i in range(t.num_vars))):
        if not target_ok(t, bits):
            check(3, t, rejected(s, t, bits), "non-solution rejected before decoding")
    return t


def symbolic_checks():
    u, c, r, p, q, k, h = sp.symbols('u c r p q k h', integer=True, nonnegative=True)
    a, b = sp.symbols('a b', integer=True)
    pairs = sp.summation(sp.summation(1, (b, a+1, c-1)), (a, 0, c-1))
    onehot_literals = u*c + 2*u*pairs
    # h is the number of compatible pairs; k is their sum of distinct requirements.
    exact_c = u*(1+pairs)+(p*q-h)+(k+h)+p+q
    exact_l = onehot_literals+(p*q-h)+(3*k+h)+2*p*q
    bound_c = u*(1+r*(r-1)/2)+p*q*(u+1)+p+q
    bound_l = u*r**2+p*q*(3*u+3)
    t = audit(examples()[0], 'symbolic-anchor')
    values = {u:3, c:2, r:4, p:4, q:3, k:9, h:6}
    identities = (
        (u*c+p*q, u*c+p*q, t.num_vars),
        (exact_c, u*(1+c*(c-1)/2)+p*q+k+p+q, t.sizes[1]),
        (exact_l, u*c**2+3*k+3*p*q, t.sizes[2]),
        (u*r+p*q-(u*c+p*q), u*(r-c), 24-t.num_vars),
        (bound_c-exact_c, u*(r-c)*(r+c-1)/2+(p*q*u-k), 76-t.sizes[1]),
        (bound_l-exact_l, u*(r-c)*(r+c)+3*(p*q*u-k), 192-t.sizes[2]),
    )
    for left, right, measured in identities:
        check(1, t, sp.simplify(left-right) == 0 and left.subs(values) == measured,
              "symbolic block identity agrees with constructed target count or bound gap")
    for s in (Source(0,0,0,(),(),()),
              Source(0,0,1,(1,),((0,(0,)),),((0,(0,)),)), examples()[0]):
        target = construct(s)
        vals = {r:s.r, u:s.u, c:len(target.candidates), p:len(s.q1), q:len(s.q2)}
        bounds = tuple(expr.subs(vals) for expr in (u*r+p*q, bound_c, bound_l))
        check(1, target, all(sp.denom(v) == 1 for v in bounds)
              and all(a <= b for a, b in zip(target.sizes, bounds)),
              "integer normalized overhead including empty universe, measured against target")


def exhaustive_checks():
    atoms = ((0,(0,)), (0,(1,)))
    lists = [a for length in range(3) for a in product(atoms, repeat=length)]
    for d, x, u in ((0,0,2), (0,1,1), (1,0,1), (1,1,0)):
        for a, b in product(lists, repeat=2):
            audit(Source(d,x,u,(1,),a,b), 'exhaustive-unary', raw=True)
    nullary = [a for length in range(3)
               for a in product(((0,()), (1,())), repeat=length)]
    for a, b in product(nullary, repeat=2):
        audit(Source(0,0,0,(0,0),a,b), 'exhaustive-nullary', raw=True)


def sampled_checks():
    rng = random.Random(2026101305)
    for u in range(6):
        for m in range(6):
            for _ in range(300):
                d, x = rng.randrange(2), rng.randrange(2)
                r = d+x+u
                arities = (0,0,1,2) if r else (0,0)
                atoms = []
                for _ in range(m):
                    rel = rng.randrange(len(arities))
                    atoms.append((rel, tuple(rng.randrange(r) for _ in range(arities[rel]))))
                p = rng.randrange(m+1)
                audit(Source(d,x,u,arities,tuple(atoms[:p]),tuple(atoms[p:])),
                      f'sampled u={u} m={m}')
            print(f"SAMPLES u={u} p+q={m}: 300 instances", flush=True)


@st.composite
def mixed_queries(draw):
    u = draw(st.integers(0,5))
    d = draw(st.integers(0,5-u))
    x = draw(st.integers(0,5-u-d))
    r = d+x+u
    arities = (0,1,2) if r else (0,)
    @st.composite
    def atom(draw):
        rel = draw(st.integers(0,len(arities)-1))
        args = tuple(draw(st.lists(st.integers(0,r-1), min_size=arities[rel], max_size=arities[rel]))) if arities[rel] else ()
        return rel, args
    a = tuple(draw(st.lists(atom(), max_size=2)))
    b = tuple(draw(st.lists(atom(), max_size=2)))
    return Source(d,x,u,arities,a,b)


@st.composite
def repeated_queries(draw):
    u = draw(st.integers(1,5))
    a, b = draw(st.tuples(st.integers(0,u-1), st.integers(0,u-1)))
    copies = draw(st.integers(1,3))
    return Source(0,0,u,(0,2), ((1,(a,a)),)*copies + ((0,()),),
                  ((1,(a,b)),)*draw(st.integers(1,3)) + ((0,()),))


@seed(2026101319)
@settings(max_examples=300, database=None, deadline=None)
@given(mixed_queries())
def hypothesis_mixed(s):
    audit(s, 'hypothesis-mixed', raw=True)


@seed(2026101320)
@settings(max_examples=300, database=None, deadline=None)
@given(repeated_queries())
def hypothesis_repeated(s):
    audit(s, 'hypothesis-repeated', raw=True)


def special_checks():
    cases = (
        Source(0,0,0,(),(),()),
        Source(0,0,3,(),(),()),
        Source(0,0,0,(0,),((0,()),),()),
        Source(0,0,0,(0,),(),((0,()),)),
        Source(0,0,0,(0,0),((0,()),),((1,()),)),
        Source(0,0,0,(0,),((0,()),)*2,((0,()),)),
        Source(2,0,0,(1,),((0,(0,)),),((0,(1,)),)),
        Source(0,2,0,(1,),((0,(0,)),),((0,(1,)),)),
        Source(0,0,2,(2,),((0,(0,0)),),((0,(0,1)),)),
    )
    for s in cases:
        audit(s, 'special', raw=True)
    for s in (
        Source(0,0,2,(1,),((0,(0,)),),((0,(0,)),(0,(1,)))),
        Source(0,0,2,(1,),((0,(0,)),(0,(1,))),((0,(0,)),)),
        Source(0,0,2,(2,),((0,(0,1)),),((0,(1,0)),)),
    ):
        audit(s, 'special', raw=True)
    # Exercise the k=u match-size boundary; binary samples alone cannot reach it.
    for u in range(1,6):
        args = tuple(range(u))
        images = tuple(reversed(args))
        audit(Source(0,0,u,(u,),((0,args),),((0,images),)), 'full-arity')
        audit(Source(0,0,u,(2*u,),((0,args+args),),((0,images+images),)),
              'full-arity-repeated')
        if u > 1:
            conflict = images[:-1]+((images[-1]+1) % u,)
            audit(Source(0,0,u,(2*u,),((0,args+args),),((0,images+conflict),)),
                  'full-arity-conflict')
    swap = Source(0,0,2,(2,),((0,(0,1)),),((0,(1,0)),))
    t = construct(swap)
    check(3, t, extract(swap,t,encode(swap,t,(1,0))) == (1,0), "one-step non-idempotent swap")
    # Huge domains are checked with specified witnesses, never enumerated.
    for d in (2**32, 2**64-2):
        s = Source(d,0,1,(1,),((0,(d,)),),((0,(d-1,)),))
        t = construct(s)
        check(2, t, t.candidates == (0,d-1) and target_ok(t,encode(s,t,(d-1,)))
              and source_ok(s,extract(s,t,encode(s,t,(d-1,)))), "large compressed universe")
        check(4, t, t.sizes == (3,6,10), "large universe target stays constant size")
    t = construct(Source(0,0,0,(),(),()))
    overflow = Source(2**64-1,1,0,(),(),())
    try:
        construct(overflow)
    except OverflowError:
        check(5,t,t.sizes == (0,0,0), "overflow is an error, not infeasibility")
    else:
        check(5,t,False,"overflow must be rejected")


def examples():
    q1 = ((0, (0, 1)), (0, (1, 2)), (0, (2, 0)), (0, (1, 1)))
    yes = Source(0, 1, 3, (2,), q1,
                 ((0, (0, 3)), (0, (3, 3)), (0, (3, 0))))
    no = Source(0, 1, 3, (2,), q1,
                ((0, (0, 2)), (0, (2, 3)), (0, (3, 0))))
    return yes, no


def example_checks():
    yes, no = examples()
    ty, tn = audit(yes,'worked-YES'), audit(no,'worked-NO')
    witness = tuple(map(bool, (0,1,0,1,0,1, 1,0,0, 0,1,0, 0,0,1, 0,1,0)))
    check(6, ty, ty.sizes == (18, 34, 75) and ty.candidates == (0, 3),
          "YES worked sizes and candidates")
    check(6, ty, target_ok(ty, witness) and extract(yes, ty, witness) == (3,3,3)
          and source_ok(yes, (3,3,3)), "YES complete witness and extraction")
    check(6, ty, {extract(yes,ty,b) for b in models(ty)} == {(3,3,0),(3,3,3)}
          and sum(source_ok(yes,w) for w in product(range(yes.r),repeat=yes.u)) == 4,
          "compression preserves existence, not witness counts")
    check(7, tn, tn.sizes == (21, 39, 87) and tn.candidates == (0,2,3),
          "NO worked sizes and candidates")
    check(7, tn, not any(source_ok(no, w) for w in product(range(no.r), repeat=no.u))
          and not list(models(tn)), "NO infeasible on both sides")
    for section, s, t, expected in (
        (6,yes,ty,((6,12),(21,39),(7,24))),
        (7,no,tn,((12,27),(20,36),(7,24))),
    ):
        c = len(t.candidates)
        onehot_end = s.u*(1+c*(c-1)//2)
        covers_start = len(t.clauses)-len(s.q1)-len(s.q2)
        blocks = (t.clauses[:onehot_end], t.clauses[onehot_end:covers_start],
                  t.clauses[covers_start:])
        check(section,t,tuple((len(b),sum(map(len,b))) for b in blocks) == expected,
              "worked example clause and literal subtotals")


if __name__ == "__main__":
    try:
        symbolic_checks()
        exhaustive_checks()
        special_checks()
        sampled_checks()
        hypothesis_mixed()
        hypothesis_repeated()
        example_checks()
        for section, title in enumerate(("symbolic overhead", "forward/backward", "extraction",
                                         "measured sizes", "well-formedness", "YES example", "NO example"), 1):
            if not COUNTS[section]:
                raise AssertionError(f"empty section {section}")
            print(f"SECTION {section} ({title}): {COUNTS[section]} checks, 0 failures")
        print(f"INSTANCES: {sum(INSTANCES.values())}; FAMILIES: {dict(INSTANCES)}")
        print(f"SOLUTIONS: {dict(STATS)}")
        print(f"TOTAL: {sum(COUNTS.values())} checks, 0 failures")
    except Exception:
        print(f"TOTAL: {sum(COUNTS.values())} checks, 1 failures", flush=True)
        raise

@isPANN

isPANN commented Nov 4, 2026

Copy link
Copy Markdown
Collaborator Author

Verification certificate (continued)

Source of the independent check script (run python3 <harness> with the core module saved as core.py next to it; --quick runs a reduced version).

Independent check script: core module
"""Independent implementation of the exact-image ConjunctiveQueryFoldability -> CNF-SAT
construction, written from the proof alone.

Instance representation: (d, x, u, arities, q1, q2) where arities is a tuple of ints and each
query is a tuple of atoms (rel, args) with args a tuple of encoded terms in [0, d+x+u).
Encoding: constants [0,d), distinguished [d,d+x), undistinguished y -> d+x+y.
A CNF is (num_vars, clauses) with clauses a list of lists of signed one-based ints.
"""
from itertools import product


class InvalidWitness(ValueError):
    pass


# ---------------------------------------------------------------- source side
def universe(inst):
    d, x, u = inst[0], inst[1], inst[2]
    return d + x + u


def validate_instance(inst):
    d, x, u, arities, q1, q2 = inst
    for v in (d, x, u):
        if not isinstance(v, int) or v < 0:
            raise ValueError("bad count")
    R = d + x + u
    for q in (q1, q2):
        for rel, args in q:
            if not (0 <= rel < len(arities)) or len(args) != arities[rel]:
                raise ValueError("bad atom")
            for t in args:
                if not (0 <= t < R):
                    raise ValueError("bad term")


def substitute(inst, w, atom):
    d, x = inst[0], inst[1]
    fixed = d + x
    rel, args = atom
    # one step only: never look at w again for the produced term
    return (rel, tuple(t if t < fixed else w[t - fixed] for t in args))


def source_eval(inst, w):
    """Mirror of the model evaluator: exact length, range, exact atom-set equality."""
    u = inst[2]
    if len(w) != u:
        raise InvalidWitness("length")
    R = universe(inst)
    for v in w:
        if not isinstance(v, int) or isinstance(v, bool) or not (0 <= v < R):
            raise InvalidWitness("range")
    image = {substitute(inst, w, a) for a in inst[4]}
    return image == set(inst[5])


def used_vars(inst):
    fixed = inst[0] + inst[1]
    return sorted({t - fixed for _, args in inst[4] for t in args if t >= fixed})


def source_witnesses_full(inst):
    """Every source witness by brute force over the whole universe (small R only)."""
    R, u = universe(inst), inst[2]
    return [w for w in product(range(R), repeat=u) if source_eval(inst, w)]


def source_feasible(inst):
    """Exact source decision without domain enumeration: a variable that occurs in Q1 must take a
    term that occurs in Q2 (its image atom lies in Q2); other coordinates never affect the image,
    so they are fixed to 0 (legal whenever u>0). Validated against source_witnesses_full on
    small instances by the harness."""
    u = inst[2]
    uv = used_vars(inst)
    q2_terms = sorted({t for _, args in inst[5] for t in args})
    for vals in product(q2_terms, repeat=len(uv)):
        w = [0] * u
        for y, v in zip(uv, vals):
            w[y] = v
        if source_eval(inst, tuple(w)):
            return True, tuple(w)
    if not uv and u == 0:
        pass
    return False, None


# ---------------------------------------------------------------- construction
def candidates(inst):
    R = universe(inst)
    terms = {t for _, args in inst[5] for t in args}
    if R > 0:
        terms.add(0)
    return sorted(terms)


def sel(inst, C, y, t):
    return y * len(C) + t + 1


def mvar(inst, C, i, j):
    return inst[2] * len(C) + i * len(inst[5]) + j + 1


def requirements(inst, a, b):
    """None if incompatible; else dict y -> target term required (distinct vars)."""
    fixed = inst[0] + inst[1]
    if a[0] != b[0] or len(a[1]) != len(b[1]):
        return None
    req = {}
    for s, t in zip(a[1], b[1]):
        if s < fixed:
            if s != t:
                return None
        else:
            y = s - fixed
            if y in req and req[y] != t:
                return None
            req[y] = t
    return req


def reduce(inst):
    validate_instance(inst)
    d, x, u, arities, q1, q2 = inst
    if d + x + u >= 2 ** 64:  # proof: total term range must fit usize on the 64-bit host
        raise OverflowError("term universe exceeds usize")
    C = candidates(inst)
    if u * len(C) + len(q1) * len(q2) > 2 ** 63 - 1:  # signed literal range
        raise OverflowError("variable count exceeds i64")
    pos = {t: k for k, t in enumerate(C)}
    c, p, q = len(C), len(q1), len(q2)
    clauses = []
    for y in range(u):
        clauses.append([sel(inst, C, y, t) for t in range(c)])
        for t in range(c):
            for s in range(t + 1, c):
                clauses.append([-sel(inst, C, y, t), -sel(inst, C, y, s)])
    for i in range(p):
        for j in range(q):
            z = mvar(inst, C, i, j)
            req = requirements(inst, q1[i], q2[j])
            if req is None:
                clauses.append([-z])
                continue
            E = [sel(inst, C, y, pos[req[y]]) for y in sorted(req)]
            for e in E:
                clauses.append([-z, e])
            clauses.append([z] + [-e for e in E])
    for i in range(p):
        clauses.append([mvar(inst, C, i, j) for j in range(q)])
    for j in range(q):
        clauses.append([mvar(inst, C, i, j) for i in range(p)])
    return u * c + p * q, clauses


def target_eval(cnf, a):
    n, clauses = cnf
    if not isinstance(a, (list, tuple)) or len(a) != n:
        raise InvalidWitness("length")
    if any(not isinstance(v, bool) for v in a):
        raise InvalidWitness("non-boolean")
    return all(any((l > 0) == a[abs(l) - 1] for l in cl) for cl in clauses)


def extract(inst, cnf, a):
    if not target_eval(cnf, a):
        raise InvalidWitness("unsatisfied")
    C = candidates(inst)
    out = []
    for y in range(inst[2]):
        chosen = [t for t in range(len(C)) if a[sel(inst, C, y, t) - 1]]
        if len(chosen) != 1:
            raise InvalidWitness("not one-hot")
        out.append(C[chosen[0]])
    return tuple(out)


def forward_assignment(inst, w):
    """Forward map of the proof: compress unused coordinates to 0, one-hot, match bits = equality."""
    C = candidates(inst)
    uv = set(used_vars(inst))
    w = tuple(v if y in uv else 0 for y, v in enumerate(w))
    n = inst[2] * len(C) + len(inst[4]) * len(inst[5])
    a = [False] * n
    for y, v in enumerate(w):
        a[sel(inst, C, y, C.index(v)) - 1] = True
    for i, A in enumerate(inst[4]):
        for j, B in enumerate(inst[5]):
            a[mvar(inst, C, i, j) - 1] = substitute(inst, w, A) == B
    return a


# ---------------------------------------------------------------- generic SAT
def all_models(cnf, limit=None):
    """Complete DPLL enumeration of every satisfying total assignment (generic CNF)."""
    n, clauses = cnf
    if any(len(cl) == 0 for cl in clauses):
        return []
    out = []

    def rec(assign):
        assign = dict(assign)
        while True:
            unit = None
            for cl in clauses:
                sat = False
                free = []
                for l in cl:
                    v = assign.get(abs(l))
                    if v is None:
                        free.append(l)
                    elif v == (l > 0):
                        sat = True
                        break
                if sat:
                    continue
                if not free:
                    return
                if len(free) == 1:
                    unit = free[0]
                    break
            if unit is None:
                break
            assign[abs(unit)] = unit > 0
        free_vars = [v for v in range(1, n + 1) if v not in assign]
        if not free_vars:
            out.append([assign[v] for v in range(1, n + 1)])
            return
        # branch on a variable occurring in an unsatisfied clause, else first free
        pick = None
        for cl in clauses:
            if not any(assign.get(abs(l)) == (l > 0) for l in cl):
                for l in cl:
                    if abs(l) not in assign:
                        pick = abs(l)
                        break
            if pick:
                break
        if pick is None:
            pick = free_vars[0]
        for val in (True, False):
            if limit is not None and len(out) >= limit:
                return
            assign[pick] = val
            rec(assign)
            del assign[pick]

    rec({})
    return out


def brute_models(cnf):
    n, _ = cnf
    return [list(bits) for bits in product((False, True), repeat=n) if target_eval(cnf, list(bits))]
Independent check script: harness
"""Independent verification harness for ConjunctiveQueryFoldability -> Satisfiability.

Written from the proof only. The implementation under test is IMPL (default: core.reduce /
core.extract); every expected quantity below is recomputed by oracle code in this file, so that a
mutated implementation (see mutants.py) is caught. Prints a per-section audit and ends with
"TOTAL: <checks> checks, <failures> failures".
"""
import argparse
import random
import sys
from collections import Counter
from itertools import combinations, product
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
import core  # noqa: E402

SECTIONS = {
    1: "symbolic overhead identities (sympy) anchored to measured output",
    2: "forward/backward: source feasible <=> target satisfiable",
    3: "extraction from every satisfying target witness",
    4: "measured sizes vs exact formulas and registered upper bounds",
    5: "structure: numbering, one-hot, match semantics, witness rejection",
    6: "worked YES example, number for number",
    7: "worked NO example, both sides infeasible",
    8: "special cases and prose claims",
}

IMPL = {"reduce": core.reduce, "extract": core.extract}


class Audit:
    def __init__(self, max_fail_print=8):
        self.checks = Counter()
        self.fails = Counter()
        self.msgs = []
        self.max = max_fail_print
        self.instances = Counter()
        self.sites = Counter()       # (section, line of the check) -> times evaluated
        self.site_fails = Counter()  # (section, line of the check) -> times failed
        self.answers = Counter()

    def ok(self, sec, cond, msg="", _line=None):
        self.checks[sec] += 1
        site = (sec, _line or sys._getframe(1).f_lineno)
        self.sites[site] += 1
        if not cond:
            self.fails[sec] += 1
            self.site_fails[site] += 1
            if len(self.msgs) < self.max:
                self.msgs.append(f"[S{sec}] {msg}")
        return cond


A = Audit()


def chk(sec, f, msg=""):
    """A.ok with a lazily evaluated condition: an exception counts as a failure of this check."""
    line = sys._getframe(1).f_lineno
    try:
        cond = bool(f())
    except Exception as e:
        cond, msg = False, f"{msg} (raised {e!r})"
    return A.ok(sec, cond, msg, _line=line)

# ------------------------------------------------------------------ oracle (independent of IMPL)
def o_C(inst):
    R = core.universe(inst)
    s = {t for _, args in inst[5] for t in args}
    if R:
        s.add(0)
    return sorted(s)


def o_pairs(inst):
    """compatibility and k for every pair, by brute force over all images: a pair (i,j) is
    compatible iff some substitution maps A_i to B_j; k = distinct undistinguished vars in A_i."""
    d, x, u, arities, q1, q2 = inst
    fixed = d + x
    out = {}
    for i, a in enumerate(q1):
        vars_a = sorted({t - fixed for t in a[1] if t >= fixed})
        for j, b in enumerate(q2):
            comp = a[0] == b[0]
            if comp:
                # build the forced map position by position, independent formulation
                forced = {}
                for pos in range(len(a[1])):
                    s, t = a[1][pos], b[1][pos]
                    if s < fixed:
                        comp &= s == t
                    else:
                        forced.setdefault(s, set()).add(t)
                comp &= all(len(v) == 1 for v in forced.values())
            out[(i, j)] = (comp, len(vars_a) if comp else None)
    return out


def o_sizes(inst):
    d, x, u, arities, q1, q2 = inst
    c, p, q = len(o_C(inst)), len(q1), len(q2)
    pr = o_pairs(inst)
    K = sum(k for comp, k in pr.values() if comp)
    H = sum(1 for comp, _ in pr.values() if comp)
    R = d + x + u
    return dict(c=c, K=K, H=H, R=R, p=p, q=q, u=u,
                vars=u * c + p * q,
                clauses=u * (1 + c * (c - 1) // 2) + p * q + K + p + q,
                literals=u * c * c + 3 * K + 3 * p * q,
                ub_vars=u * R + p * q,
                ub_clauses=u * (1 + R * (R - 1) // 2) + p * q * (u + 1) + p + q,
                ub_literals=u * R * R + p * q * (3 * u + 3))


def o_sel(inst, y, t_index):
    return y * len(o_C(inst)) + t_index + 1


def o_z(inst, i, j):
    return inst[2] * len(o_C(inst)) + i * len(inst[5]) + j + 1


def o_source_witnesses(inst):
    return core.source_witnesses_full(inst)


def o_feasible_candidate_vectors(inst):
    C = o_C(inst)
    return {w for w in product(C, repeat=inst[2]) if core.source_eval(inst, w)}


def o_forward(inst, w, compress_all=False):
    """proof's forward map; an unused coordinate is moved to term 0 when its value is not a
    candidate (compress_all: always moved, the literal reading of the lemma)."""
    C = o_C(inst)
    used = set(core.used_vars(inst))
    w = tuple(v if y in used or (v in C and not compress_all) else 0 for y, v in enumerate(w))
    n = inst[2] * len(C) + len(inst[4]) * len(inst[5])
    a = [False] * n
    for y, v in enumerate(w):
        a[o_sel(inst, y, C.index(v)) - 1] = True
    for i, Aat in enumerate(inst[4]):
        for j, B in enumerate(inst[5]):
            a[o_z(inst, i, j) - 1] = core.substitute(inst, w, Aat) == B
    return a


def sat_eval(cnf, a):
    n, cls = cnf
    return len(a) == n and all(any((l > 0) == a[abs(l) - 1] for l in cl) for cl in cls)


# ------------------------------------------------------------------ per-instance audit
FULL_LIMIT = 4096  # brute-force the full source space when R^u <= this


def audit(inst, tag="", full=None, raw_limit=14):
    """Audit one instance; an exception raised by the implementation under test (or by an
    oracle confronted with malformed output) counts as a failed check, never aborts the run."""
    try:
        return _audit(inst, tag, full, raw_limit)
    except Exception as e:
        A.ok(5, False, f"{tag} crashed {e!r} on {inst}")
        return None


def _audit(inst, tag, full, raw_limit):
    A.instances[tag] += 1
    d, x, u, arities, q1, q2 = inst
    R = d + x + u
    try:
        cnf = IMPL["reduce"](inst)
    except Exception as e:  # construction must succeed on every legal instance
        A.ok(5, False, f"{tag} reduce raised {e!r} on {inst}")
        return
    n, cls = cnf
    sz = o_sizes(inst)
    # --- S5 well-formedness
    A.ok(5, all(isinstance(l, int) and l != 0 and abs(l) <= n for cl in cls for l in cl),
         f"{tag} literal out of range {inst}")
    # --- S4 sizes
    nl = sum(len(cl) for cl in cls)
    A.ok(4, n == sz["vars"], f"{tag} vars {n} != {sz['vars']} {inst}")
    A.ok(4, len(cls) == sz["clauses"], f"{tag} clauses {len(cls)} != {sz['clauses']} {inst}")
    A.ok(4, nl == sz["literals"], f"{tag} literals {nl} != {sz['literals']} {inst}")
    A.ok(4, n <= sz["ub_vars"] and len(cls) <= sz["ub_clauses"] and nl <= sz["ub_literals"],
         f"{tag} exceeds registered bound {inst}")
    # --- S5 layout: covers are the last p+q clauses with the documented variable sets
    p, q = len(q1), len(q2)
    if len(cls) >= p + q:
        covers = cls[len(cls) - p - q:]
        exp_rows = [sorted(o_z(inst, i, j) for j in range(q)) for i in range(p)]
        exp_cols = [sorted(o_z(inst, i, j) for i in range(p)) for j in range(q)]
        A.ok(5, [sorted(c) for c in covers] == exp_rows + exp_cols, f"{tag} cover layout {inst}")
    # one-hot block layout: per variable y, one positive clause then c(c-1)/2 negative pairs
    c = sz["c"]
    expected_onehot = []
    for y in range(u):
        expected_onehot.append(sorted(o_sel(inst, y, t) for t in range(c)))
        for t, s in combinations(range(c), 2):
            expected_onehot.append(sorted([-o_sel(inst, y, t), -o_sel(inst, y, s)]))
    A.ok(5, [sorted(cl) for cl in cls[:len(expected_onehot)]] == expected_onehot,
         f"{tag} one-hot layout {inst}")
    # --- target models
    models = core.all_models(cnf)
    if n <= raw_limit:
        raw = core.brute_models(cnf)
        A.ok(5, sorted(map(tuple, raw)) == sorted(map(tuple, models)),
             f"{tag} DPLL incomplete {inst}")
    tgt = bool(models)
    A.answers["YES" if tgt else "NO"] += 1
    # --- source truth
    src_fast, wit = core.source_feasible(inst)
    if full is None:
        full = R ** u <= FULL_LIMIT
    if full:
        sw = o_source_witnesses(inst)
        A.ok(2, src_fast == bool(sw), f"{tag} oracle disagreement {inst}")
        # forward: every source witness maps (after compression) to a satisfying assignment
        for w in sw:
            A.ok(2, sat_eval(cnf, o_forward(inst, w)), f"{tag} forward map unsat for {w} {inst}")
            A.ok(2, sat_eval(cnf, o_forward(inst, w, True)), f"{tag} forward(compress all) unsat {w} {inst}")
    elif wit is not None:
        A.ok(2, sat_eval(cnf, o_forward(inst, wit)), f"{tag} forward map unsat for {wit} {inst}")
    A.ok(2, src_fast == tgt, f"{tag} source {src_fast} target {tgt} {inst}")
    # --- S3 extraction from every model
    got = set()
    for a in models:
        try:
            w = IMPL["extract"](inst, cnf, a)
        except Exception as e:
            A.ok(3, False, f"{tag} extract raised {e!r} {inst}")
            continue
        try:
            fe = core.source_eval(inst, tuple(w))
        except core.InvalidWitness as e:
            fe = False
        A.ok(3, fe, f"{tag} extracted {w} infeasible {inst}")
        got.add(tuple(w))
    if c ** u <= FULL_LIMIT:
        A.ok(3, got == o_feasible_candidate_vectors(inst),
             f"{tag} extracted set != feasible candidate vectors {inst}")
    # --- S5 semantics in every model: one-hot and z == equality
    C = o_C(inst)
    for a in models:
        if not A.ok(5, len(a) == sz["vars"], f"{tag} model length {len(a)} {inst}"):
            break
        good = True
        for y in range(u):
            good &= sum(a[o_sel(inst, y, t) - 1] for t in range(c)) == 1
        if good:
            w = tuple(C[[t for t in range(c) if a[o_sel(inst, y, t) - 1]][0]] for y in range(u))
            for i in range(p):
                for j in range(q):
                    good &= a[o_z(inst, i, j) - 1] == (core.substitute(inst, w, q1[i]) == q2[j])
        A.ok(5, good, f"{tag} model violates documented variable meaning {inst}")
    # --- S5 rejection of non-witnesses
    if n:
        bad = [not v for v in models[0]] if models else [False] * n
        if not sat_eval(cnf, bad):
            A.ok(5, raises(lambda: IMPL["extract"](inst, cnf, bad)), f"{tag} unsatisfied accepted {inst}")
    base = models[0] if models else [False] * n
    A.ok(5, raises(lambda: IMPL["extract"](inst, cnf, list(base) + [False])),
         f"{tag} long witness accepted {inst}")
    if n:
        A.ok(5, raises(lambda: IMPL["extract"](inst, cnf, list(base)[:-1])),
             f"{tag} short witness accepted {inst}")
        A.ok(5, raises(lambda: IMPL["extract"](inst, cnf, [int(v) for v in base])),
             f"{tag} non-boolean witness accepted {inst}")
    return cnf, models


def raises(f):
    try:
        f()
    except Exception:
        return True
    return False


# ------------------------------------------------------------------ section 1: symbolic
def claimed(sp, u, c, R, p, q, K):
    """Every closed form the proof states (overhead section and table)."""
    return dict(
        pairs=c * (c - 1) / 2, onehot_lits=u * c ** 2,
        ex_vars=u * c + p * q, ex_cl=u * (1 + c * (c - 1) / 2) + p * q + K + p + q,
        ex_li=u * c ** 2 + 3 * K + 3 * p * q,
        ub_vars=u * R + p * q, ub_cl=u * (1 + R * (R - 1) / 2) + p * q * (u + 1) + p + q,
        ub_li=u * R ** 2 + p * q * (3 * u + 3),
        gap_vars=u * (R - c), gap_cl=u * (R - c) * (R + c - 1) / 2 + (p * q * u - K),
        gap_li=u * (R - c) * (R + c) + 3 * (p * q * u - K))


CLAIMS = claimed  # mutants.py may replace this with a perturbed statement of the proof


def section1(anchors):
    import sympy as sp
    u, c, R, p, q, K, t, s = sp.symbols("u c R p q K t s", integer=True, nonnegative=True)
    F = CLAIMS(sp, u, c, R, p, q, K)
    # derive the block counts from the construction's loops, then compare with the claims
    pairs = sp.summation(sp.summation(1, (s, t + 1, c - 1)), (t, 0, c - 1))
    A.ok(1, sp.simplify(pairs - F["pairs"]) == 0, "pair count")
    A.ok(1, sp.simplify(u * (c + 2 * pairs) - F["onehot_lits"]) == 0, "one-hot literal count")
    H = sp.Symbol("H", integer=True, nonnegative=True)  # compatible pairs; K = sum of their k
    derived_cl = u * (1 + pairs) + (p * q - H) + (K + H) + p + q
    derived_li = u * (c + 2 * pairs) + (p * q - H) + (3 * K + H) + 2 * p * q
    A.ok(1, sp.expand(derived_cl - F["ex_cl"]) == 0, "exact clauses from blocks")
    A.ok(1, sp.expand(derived_li - F["ex_li"]) == 0, "exact literals from blocks")
    A.ok(1, sp.expand(F["ub_vars"] - F["ex_vars"] - F["gap_vars"]) == 0, "var gap")
    A.ok(1, sp.expand(F["ub_cl"] - F["ex_cl"] - F["gap_cl"]) == 0, "clause gap")
    A.ok(1, sp.expand(F["ub_li"] - F["ex_li"] - F["gap_li"]) == 0, "literal gap")
    # the registered bounds are the exact counts at c=R, K=pqu (the worst case)
    A.ok(1, sp.expand(F["ex_cl"].subs({c: R, K: p * q * u}) - F["ub_cl"]) == 0, "clause bound is worst case")
    A.ok(1, sp.expand(F["ex_li"].subs({c: R, K: p * q * u}) - F["ub_li"]) == 0, "literal bound is worst case")
    # R = 0 boundary: u=c=K=0 gives the zero product, bound equals exact
    A.ok(1, sp.expand(F["ub_cl"].subs({R: 0, u: 0, K: 0}) - F["ex_cl"].subs({u: 0, c: 0, K: 0})) == 0, "R=0 equal")
    # gaps nonnegative for 0<=c<=R, R>=1, 0<=K<=pqu: write R=c+a, K=pqu-b; with R>=1 either a=0
    # (term vanishes) or a>=1 and a+2c-1>=0
    a_, b_ = sp.symbols("a b", integer=True, nonnegative=True)
    g = sp.expand(F["gap_cl"].subs({R: c + a_, K: p * q * u - b_}))
    A.ok(1, sp.expand(g - (u * a_ * (a_ + 2 * c - 1) / 2 + b_)) == 0, "clause gap form")
    gl = sp.expand(F["gap_li"].subs({R: c + a_, K: p * q * u - b_}))
    A.ok(1, sp.expand(gl - (u * a_ * (a_ + 2 * c) + 3 * b_)) == 0, "literal gap form")
    ex_cl, ex_li, ub_cl, ub_li = F["ex_cl"], F["ex_li"], F["ub_cl"], F["ub_li"]
    # anchors: formulas evaluated symbolically equal measured counts
    for inst in anchors:
        cnf = IMPL["reduce"](inst)
        sz = o_sizes(inst)
        sub = {u: sz["u"], c: sz["c"], R: sz["R"], p: sz["p"], q: sz["q"], K: sz["K"]}
        A.ok(1, ex_cl.subs(sub) == len(cnf[1]), f"anchor clauses {inst}")
        A.ok(1, ex_li.subs(sub) == sum(map(len, cnf[1])), f"anchor literals {inst}")
        A.ok(1, F["ex_vars"].subs(sub) == cnf[0], f"anchor vars {inst}")
        A.ok(1, ub_cl.subs(sub) >= len(cnf[1]) and ub_li.subs(sub) >= sum(map(len, cnf[1])), "anchor ub")


# ------------------------------------------------------------------ examples
X, U, V, AA = 0, 1, 2, 3
YES = (0, 1, 3, (2,), ((0, (X, U)), (0, (U, V)), (0, (V, X)), (0, (U, U))),
       ((0, (X, AA)), (0, (AA, AA)), (0, (AA, X))))
NO = YES[:5] + (((0, (X, V)), (0, (V, AA)), (0, (AA, X))),)


def kmatrix(inst):
    pr = o_pairs(inst)
    return [[pr[(i, j)][1] if pr[(i, j)][0] else None for j in range(len(inst[5]))]
            for i in range(len(inst[4]))]


def blocks(inst, cnf):
    """split clauses into one-hot / match / cover blocks by the documented layout and count."""
    sz = o_sizes(inst)
    n1 = sz["u"] * (1 + sz["c"] * (sz["c"] - 1) // 2)
    n3 = sz["p"] + sz["q"]
    cls = cnf[1]
    parts = [cls[:n1], cls[n1:len(cls) - n3], cls[len(cls) - n3:]]
    return [(len(b), sum(map(len, b))) for b in parts]


def section6():
    cnf, models = audit(YES, "yes-example") or ((0, []), [])
    sz = o_sizes(YES)
    chk(6, lambda: o_C(YES) == [0, 3], "candidates")
    chk(6, lambda: core.source_eval(YES, (3, 3, 3)), "(3,3,3) feasible")
    img = [core.substitute(YES, (3, 3, 3), a) for a in YES[4]]
    chk(6, lambda: img == [(0, (X, AA)), (0, (AA, AA)), (0, (AA, X)), (0, (AA, AA))], "image list")
    chk(6, lambda: kmatrix(YES) == [[1, None, None], [2, 2, 2], [None, None, 1], [None, 1, None]], "k matrix")
    chk(6, lambda: sz["H"] == 6 and sz["K"] == 9, "H,K")
    chk(6, lambda: (cnf[0], len(cnf[1]), sum(map(len, cnf[1]))) == (18, 34, 75), "18/34/75")
    chk(6, lambda: blocks(YES, cnf) == [(6, 12), (21, 39), (7, 24)], f"blocks {blocks(YES, cnf)}")
    wit = [bool(b) for b in [0, 1, 0, 1, 0, 1, 1, 0, 0, 0, 1, 0, 0, 0, 1, 0, 1, 0]]
    chk(6, lambda: core.target_eval(cnf, wit), "18-bit witness satisfies")
    chk(6, lambda: wit in models, "witness among enumerated models")
    chk(6, lambda: tuple(IMPL["extract"](YES, cnf, wit)) == (3, 3, 3), "extraction (3,3,3)")
    chk(6, lambda: wit == o_forward(YES, (3, 3, 3)), "forward map gives the 18-bit witness")
    chk(6, lambda: (sz["ub_vars"], sz["ub_clauses"], sz["ub_literals"]) == (24, 76, 192), "registry bounds")
    chk(6, lambda: len(o_source_witnesses(YES)) == 4 and len(models) == 2, "4 source vs 2 target witnesses")
    chk(6, lambda: all(w[:2] == (3, 3) for w in o_source_witnesses(YES)), "unused last coordinate")


def section7():
    cnf, models = audit(NO, "no-example") or ((0, []), [])
    sz = o_sizes(NO)
    chk(7, lambda: o_C(NO) == [0, 2, 3], "candidates")
    chk(7, lambda: kmatrix(NO) == [[1, None, None], [2, 2, 2], [None, None, 1], [None, None, None]], "k matrix")
    chk(7, lambda: sz["H"] == 5 and sz["K"] == 8, "H,K")
    chk(7, lambda: (cnf[0], len(cnf[1]), sum(map(len, cnf[1]))) == (21, 39, 87), "21/39/87")
    chk(7, lambda: blocks(NO, cnf) == [(12, 27), (20, 36), (7, 24)], f"blocks {blocks(NO, cnf)}")
    chk(7, lambda: all([-v] in cnf[1] for v in (19, 20, 21)) and [19, 20, 21] in cnf[1], "units 19-21 and row cover")
    chk(7, lambda: len(o_source_witnesses(NO)) == 0, "source infeasible (all 64 vectors)")
    chk(7, lambda: len(models) == 0 and len(core.brute_models(cnf)) == 0, "target infeasible (all 2^21)")
    chk(7, lambda: (sz["ub_vars"], sz["ub_clauses"], sz["ub_literals"]) == (24, 76, 192), "registry bounds")
    chk(7, lambda: all(core.substitute(NO, w, NO[4][3])[1][0] == core.substitute(NO, w, NO[4][3])[1][1]
                for w in product(range(4), repeat=3)), "R(u,u) always loops")


# ------------------------------------------------------------------ exhaustive families
def atoms_over(arities, R):
    return [(r, args) for r, a in enumerate(arities) for args in product(range(R), repeat=a)]


def lists_upto(atoms, L):
    for n in range(L + 1):
        yield from product(atoms, repeat=n)


def exhaustive():
    fams = []
    # binary relation, all splits with R<=2, lists <= 2
    for R in (1, 2):
        for d in range(R + 1):
            for x in range(R - d + 1):
                fams.append(((d, x, R - d - x), (2,), 2))
    # binary, R=3, lists <= 2, three splits
    for split in ((0, 0, 3), (0, 1, 2), (1, 1, 1)):
        fams.append((split, (2,), 2))
    # ternary R=2 lists<=2
    for split in ((0, 0, 2), (0, 1, 1)):
        fams.append((split, (3,), 2))
    # mixed nullary/unary signature, R<=2, lists <= 2
    for split in ((0, 0, 2), (1, 0, 1), (0, 1, 1), (0, 0, 1)):
        fams.append((split, (0, 0, 1), 2))
    # empty universe: nullary only, lists <= 3
    fams.append(((0, 0, 0), (0, 0), 3))
    # unary relation, u up to 5, lists <= 2
    for split in ((0, 0, 4), (0, 1, 3), (0, 0, 5), (1, 0, 4)):
        fams.append((split, (1,), 2))
    total = 0
    for (d, x, u), arities, L in fams:
        R = d + x + u
        atoms = atoms_over(arities, R)
        ls = list(lists_upto(atoms, L))
        for q1 in ls:
            for q2 in ls:
                audit((d, x, u, arities, tuple(q1), tuple(q2)), "exhaustive")
                total += 1
    return total


# ------------------------------------------------------------------ hypothesis strategies
def hypothesis_runs(n_examples, seeds):
    from hypothesis import given, settings, seed, HealthCheck, strategies as st

    @st.composite
    def general(draw):
        """Mixed legal instances: nullary..ternary relations, duplicates, occasionally huge d."""
        big = draw(st.booleans())
        d = draw(st.integers(0, 2)) if not big else draw(st.integers(10 ** 6, 10 ** 12))
        x = draw(st.integers(0, 2))
        u = draw(st.integers(0, 3))
        R = d + x + u
        nrel = draw(st.integers(1, 3))
        arities = tuple(draw(st.integers(0, 3)) if R else 0 for _ in range(nrel))

        def term():
            if big and d:
                # sample from a handful of terms spread across the huge universe
                return draw(st.sampled_from(sorted({0, 1, d - 1, d // 2, d, d + x, R - 1} & set(range(R)))
                                            if R < 10 else [0, 1, d - 1, d // 2] + list(range(d, R))))
            return draw(st.integers(0, R - 1))

        def atom():
            r = draw(st.integers(0, nrel - 1))
            return (r, tuple(term() for _ in range(arities[r])))
        q1 = tuple(atom() for _ in range(draw(st.integers(0, 4))))
        q2 = tuple(atom() for _ in range(draw(st.integers(0, 4))))
        return (d, x, u, arities, q1, q2)

    @st.composite
    def planted(draw):
        """Planted YES via an arbitrary (possibly non-idempotent) substitution, then optionally
        broken by dropping/adding one target atom: stresses exact-image vs containment."""
        d = draw(st.integers(0, 1)); x = draw(st.integers(0, 2)); u = draw(st.integers(1, 4))
        R = d + x + u
        arities = (draw(st.integers(1, 4)), 0, 2)
        fixed = d + x

        def atom():
            r = draw(st.integers(0, 2))
            # bias to few variables so repetitions inside an atom are frequent
            pool = list(range(fixed, R))[:draw(st.integers(1, u))] + list(range(fixed))
            return (r, tuple(draw(st.sampled_from(pool)) for _ in range(arities[r])))
        q1 = tuple(atom() for _ in range(draw(st.integers(1, 4))))
        w = tuple(draw(st.integers(0, R - 1)) for _ in range(u))
        img = [core.substitute((d, x, u, arities, q1, ()), w, a) for a in q1]
        q2 = list(img) + [draw(st.sampled_from(img)) for _ in range(draw(st.integers(0, 2)))]
        mode = draw(st.sampled_from(["keep", "keep", "drop", "add", "swap"]))
        if mode == "drop" and q2:
            victim = draw(st.sampled_from(q2))
            q2 = [b for b in q2 if b != victim]
        elif mode == "add":
            q2.append(atom())
        elif mode == "swap" and q2:
            k = draw(st.integers(0, len(q2) - 1))
            rel, args = q2[k]
            q2[k] = (rel, tuple(reversed(args)))
        q2 = draw(st.permutations(q2))
        return (d, x, u, arities, q1, tuple(q2)), w, mode

    @st.composite
    def repeated(draw):
        """High-arity atoms with repeated variables; targets consistent or conflicting."""
        u = draw(st.integers(1, 5)); d = draw(st.integers(0, 1)); x = draw(st.integers(0, 1))
        R = d + x + u
        k = draw(st.integers(1, 6))
        arities = (k,)
        fixed = d + x
        q1 = []
        for _ in range(draw(st.integers(1, 2))):
            q1.append((0, tuple(draw(st.integers(0, R - 1)) for _ in range(k))))
        q2 = []
        for a in q1:
            w = tuple(draw(st.integers(0, R - 1)) for _ in range(u))
            b = list(core.substitute((d, x, u, arities, (), ()), w, a)[1])
            if draw(st.booleans()):
                pos = draw(st.integers(0, k - 1)); b[pos] = draw(st.integers(0, R - 1))
            q2.append((0, tuple(b)))
        return (d, x, u, arities, tuple(q1), tuple(q2))

    ran = Counter()
    cfg = dict(max_examples=n_examples, database=None, deadline=None,
               suppress_health_check=list(HealthCheck))

    @seed(seeds[0])
    @settings(**cfg)
    @given(general())
    def t_general(inst):
        ran["general"] += 1
        audit(inst, "hyp-general")

    @seed(seeds[1])
    @settings(**cfg)
    @given(planted())
    def t_planted(arg):
        inst, w, mode = arg
        ran["planted"] += 1
        res = audit(inst, "hyp-planted")
        if mode == "keep":
            A.ok(2, core.source_eval(inst, w), f"planted witness not feasible {inst}")
            A.ok(2, res is not None and bool(res[1]), f"planted YES but target UNSAT {inst}")

    @seed(seeds[2])
    @settings(**cfg)
    @given(repeated())
    def t_repeated(inst):
        ran["repeated"] += 1
        audit(inst, "hyp-repeated")

    t_general(); t_planted(); t_repeated()
    return ran


# ------------------------------------------------------------------ special cases / prose
def section8():
    # swap: d=x=0,u=2, (0,1) -> (1,0) via witness (1,0)
    sw = (0, 0, 2, (2,), ((0, (0, 1)),), ((0, (1, 0)),))
    cnf, models = audit(sw, "special") or ((0, []), [])
    chk(8, lambda: {tuple(IMPL["extract"](sw, cnf, a)) for a in models} == {(1, 0)}, "swap witnesses")
    # recursive substitution would map y0 -> y1 -> y0: the image differs, so one-step matters
    chk(8, lambda: core.substitute(sw, (1, 0), sw[4][0]) == (0, (1, 0)), "one-step image")
    # both queries empty, u>0: feasible, free one-hot choices
    e = (2, 0, 2, (1,), (), ())
    cnf, models = audit(e, "special") or ((0, []), [])
    chk(8, lambda: len(models) == len(o_C(e)) ** 2 and len(cnf[1]) == 2 * (1 + 0), "empty queries free choices")
    chk(8, lambda: all(core.source_eval(e, tuple(IMPL["extract"](e, cnf, a))) for a in models), "empty queries extraction")
    # no variables at all: empty target, empty witness
    z = (0, 0, 0, (0,), (), ())
    cnf, models = audit(z, "special") or ((0, []), [])
    chk(8, lambda: cnf == (0, []) and models == [[]] and tuple(IMPL["extract"](z, cnf, [])) == (), "empty target")
    # exactly one query empty -> empty clause, both NO
    for inst in [(0, 0, 1, (0,), ((0, ()),), ()), (0, 0, 1, (0,), (), ((0, ()),)),
                 (0, 1, 1, (1,), ((0, (1,)),), ())]:
        cnf, models = audit(inst, "special") or ((0, []), [])
        chk(8, lambda: [] in cnf[1] and not models and not core.source_feasible(inst)[0], f"one empty {inst}")
    # nullary equal -> positive unit, different -> negative unit, R=0
    nz = (0, 0, 0, (0, 0), ((0, ()), (1, ())), ((0, ()),))
    cnf, models = audit(nz, "special") or ((0, []), [])
    chk(8, lambda: [1] in cnf[1] and [-2] in cnf[1] and not models, "nullary units")
    # u=0: fixed atom sets decide
    for q2, exp in [(((0, (0, 1)),), True), (((0, (1, 0)),), False)]:
        inst = (1, 1, 0, (2,), ((0, (0, 1)), (0, (0, 1))), q2)
        cnf, models = audit(inst, "special") or ((0, []), [])
        chk(8, lambda: bool(models) == exp and cnf[0] == 2, f"u=0 fixed {inst}")
    # fixed terms cannot change; candidate zero does not help
    inst = (2, 0, 1, (2,), ((0, (1, 2)),), ((0, (0, 0)),))
    cnf, models = audit(inst, "special") or ((0, []), [])
    chk(8, lambda: not models and not core.source_feasible(inst)[0], "fixed constant")
    # repeated var conflict: R(y,y) -> R(0,1) incompatible; consistent -> one requirement
    inst = (2, 0, 1, (2,), ((0, (2, 2)),), ((0, (0, 1)),))
    cnf, _ = audit(inst, "special") or ((0, []), [])
    chk(8, lambda: [-o_z(inst, 0, 0)] in cnf[1], "conflicting repeat negative unit")
    inst = (2, 0, 1, (3,), ((0, (2, 2, 2)),), ((0, (1, 1, 1)),))
    cnf, _ = audit(inst, "special") or ((0, []), [])
    chk(8, lambda: o_sizes(inst)["K"] == 1 and len(cnf[1]) == o_sizes(inst)["clauses"], "consistent repeat one requirement")
    # containment only (one direction) must be NO: Q1 image strictly inside Q2 and vice versa
    for inst in [(0, 1, 1, (2,), ((0, (0, 1)),), ((0, (0, 0)), (0, (0, 1)))),
                 (0, 1, 1, (2,), ((0, (0, 1)), (0, (1, 1))), ((0, (0, 0)),))]:
        cnf, models = audit(inst, "special") or ((0, []), [])
        src = core.source_feasible(inst)[0]
        chk(8, lambda: src == bool(models), f"containment {inst}")
    # repeated atoms: duplicates in either list do not require multiset equality
    inst = (0, 1, 2, (2,), ((0, (1, 2)), (0, (1, 2)), (0, (2, 1))), ((0, (0, 0)),))
    cnf, models = audit(inst, "special") or ((0, []), [])
    chk(8, lambda: bool(models) and core.source_feasible(inst)[0], "duplicates collapse")
    # large compressed universes: target size independent of d
    # y2 is unused, so term 0 (a constant once d>=1) is a candidate it may take; in the second
    # base Q2 has no terms at all, so term 0 is the only candidate of every variable
    bases = [(0, 1, 3, (2,), ((0, (0, 1)), (0, (1, 2))), ((0, (0, 0)),)),
             (0, 0, 2, (0,), ((0, ()),), ((0, ()),))]
    for bi, base in enumerate(bases):
        sizes = set()
        for d in (1, 5, 10 ** 6, 10 ** 15, 2 ** 62):  # d=0 differs: term 0 is then x, already in Q2
            inst = (d,) + base[1:3] + (base[3],) + (tuple((r, tuple(t + d for t in a)) for r, a in base[4]),
                                                    tuple((r, tuple(t + d for t in a)) for r, a in base[5]))
            res = audit(inst, "special", full=False)
            if not chk(8, lambda: res is not None, f"large d={d} reduce failed"):
                continue
            cnf, models = res
            chk(8, lambda: bool(models), f"large d={d} base {bi} should be YES")
            sizes.add((cnf[0], len(cnf[1])))
            chk(8, lambda: bool(models) == core.source_feasible(inst)[0], f"large d={d}")
        chk(8, lambda: len(sizes) == 1, f"compressed size depends on d: {sizes}")
    # numeric contract: unrepresentable totals are typed errors, never a NO answer
    for inst in [(2 ** 64 - 1, 1, 0, (), (), ()), (2 ** 63, 2 ** 63, 0, (0,), ((0, ()),), ((0, ()),))]:
        try:
            IMPL["reduce"](inst)
            got = "accepted"
        except OverflowError:
            got = "OverflowError"
        except Exception as e:
            got = repr(e)
        chk(8, lambda: got == "OverflowError", f"overflow {inst[:3]}: {got}")
    edge = (2 ** 64 - 2, 1, 0, (1,), ((0, (2 ** 64 - 2,)),), ((0, (2 ** 64 - 2,)),))
    try:
        cnf = IMPL["reduce"](edge)
        chk(8, lambda: cnf[0] == 1 and core.target_eval(cnf, [True]), "largest legal universe")
    except Exception as e:
        chk(8, lambda: False, f"largest legal universe raised {e!r}")
    # c <= min(R, 1 + a2) and c=0 iff R=0; K <= pqu; gaps nonnegative (over many random instances)
    rng = random.Random(2026110401)
    for _ in range(3000):
        d, x, u = rng.randint(0, 3), rng.randint(0, 2), rng.randint(0, 3)
        R = d + x + u
        arities = tuple(rng.randint(0, 3) if R else 0 for _ in range(2))
        mk = lambda: tuple((r, tuple(rng.randrange(R) for _ in range(arities[r])))
                           for r in (rng.randrange(2) for _ in range(rng.randint(0, 4))))
        inst = (d, x, u, arities, mk(), mk())
        sz = o_sizes(inst)
        a2 = sum(len(a) for _, a in inst[5])
        chk(8, lambda: sz["c"] <= (min(R, 1 + a2) if R else 0), f"c bound {inst}")
        chk(8, lambda: sz["K"] <= sz["p"] * sz["q"] * u, "K bound")
        chk(8, lambda: sz["vars"] <= sz["ub_vars"] and sz["clauses"] <= sz["ub_clauses"]
             and sz["literals"] <= sz["ub_literals"], "bounds")
        # compression lemma directly: any full witness compressed stays feasible and lies in C^u
        if R ** u <= 256:
            for w in o_source_witnesses(inst):
                used = set(core.used_vars(inst))
                cw = tuple(v if y in used else 0 for y, v in enumerate(w))
                chk(8, lambda: core.source_eval(inst, cw) and set(cw) <= set(o_C(inst)), f"compression {inst} {w}")
        if rng.random() < 0.3:
            audit(inst, "random")


ANCHORS = [YES, NO, (0, 0, 2, (2,), ((0, (0, 1)),), ((0, (1, 0)),)), (0, 0, 0, (0,), (), ()),
           (2, 0, 2, (1,), (), ()), (0, 0, 1, (0,), ((0, ()),), ())]


def run(quick=False, seeds=(2026110411, 2026110412, 2026110413), n_hyp=300, out=print):
    global A
    A = Audit()
    section1(ANCHORS)
    section6()
    section7()
    section8()
    fam = {}
    fam["exhaustive"] = exhaustive() if not quick else 0
    fam.update(hypothesis_runs(n_hyp if not quick else 40, seeds))
    out("Independent audit (ConjunctiveQueryFoldability -> Satisfiability)")
    for k, v in fam.items():
        out(f"  family {k}: {v} instances")
    for k, v in sorted(A.instances.items()):
        out(f"  audited [{k}]: {v}")
    for s in sorted(SECTIONS):
        out(f"  Section {s} ({SECTIONS[s]}): {A.checks[s]} checks, {A.fails[s]} failures")
    out(f"  target answers: {dict(A.answers)}")
    for m in A.msgs:
        out("  FAIL " + m)
    tot, bad = sum(A.checks.values()), sum(A.fails.values())
    out(f"TOTAL: {tot} checks, {bad} failures")
    return A


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--quick", action="store_true")
    args = ap.parse_args()
    res = run(quick=args.quick)
    sys.exit(1 if sum(res.fails.values()) else 0)

@isPANN

isPANN commented Nov 4, 2026

Copy link
Copy Markdown
Collaborator Author

Agentic Review Report

Subject: rule ConjunctiveQueryFoldability → Satisfiability (closes #1203)

Verdict: no open findings.

The change touches four files:

  • src/rules/conjunctivequeryfoldability_satisfiability.rs (new)
  • src/unit_tests/rules/conjunctivequeryfoldability_satisfiability.rs (new)
  • src/rules/mod.rs (registration and canonical example)
  • docs/paper/reductions.typ (rule entry)

Structural and semantic

  • Registration. The impl names the source type literally (impl ReduceTo<Satisfiability> for ConjunctiveQueryFoldability). It is registered with #[reduction(transform = upper_bound { … })], which covers num_vars, num_clauses and num_literals exactly once each.

  • Transform. All three fields are upper bounds. This is correct: the candidate count and the number of compatible requirements are not registered source parameters. The formulas match the issue: uR+pq, u(1+R(R−1)/2)+pq(u+1)+p+q and uR²+pq(3u+3), with R = d+x+u.

  • Result mapping. #[aggregate_reduction(identity)] is registered, as docs/src/design.md ("Result mappings") prescribes for a YES/NO → YES/NO rule. Or maps to Or; both endpoint Values resolve to crate::types::Or.

  • Extraction. extract_solution calls validate_target_witness once, then reads the unique true selector of each variable. Wrong lengths and unsatisfied assignments return an ExtractionError. There is no clamp, default or recursive substitution.

  • Arithmetic. d+x, d+x+u, u·c, p·q and their sum use checked arithmetic; overflow returns ReductionError::IntegerOverflow. The variable count is converted with exact_i64 before any literal is formed. The target is built through Satisfiability::try_new, with errors mapped by target_construction.

  • Exports. No generated exports are in the diff, and no shared test or file is weakened. mod.rs only gains two lines.

  • Hand trace. I traced reduce_to() on d=1, x=0, u=2, Q1 = [R0(y0,y1), R1()], Q2 = [R0(y1,y0), R1()]. My prediction:

    • candidates (0,1,2) and 10 variables;
    • 18 clauses: 8 one-hot, then [−7,3], [−7,5], [7,−3,−5], [−8], [−9], [10], then the covers [7,8], [9,10], [7,9], [8,10];
    • 36 literals;
    • unique folding (2,1), a non-idempotent swap.

    pred reduce produced exactly these clauses in this order. pred path … instance.json measures 10 / 18 / 36, and pred solve returns [2, 1].

  • Paper. reduction-rule loads the example with load-example. Every YES and NO number is computed from the fixture and guarded by assert. The proof covers construction, both directions, extraction and overhead. Its contrast with retraction-based folding matches Chandra–Merlin (STOC 1977, printed p. 81: the folding homomorphism fixes a set V that contains its image). No new bibliography entry was added; chandra1977 already exists, with a DOI.

  • Issue compliance. The issue's YES instance is the canonical example. Its NO instance, the swap, and the validation-plan families are tested; see the quality section.

Quality

The rule has 45 tests; the slowest takes 0.14 s.

  • Closed loop. test_conjunctivequeryfoldability_to_satisfiability_closed_loop covers the non-idempotent swap.
  • Worked examples. YES: 18/34/75 sizes, the full 18-bit witness and extraction (3,3,3). NO: 21/39/87 sizes, the source shown infeasible by brute force, units −19..−21 against the row cover (19,20,21).
  • Rejections. Extraction rejects lengths 17 and 19, all-false and all-true assignments.
  • Exhaustive families. All unary list pairs of length ≤ 2 over two terms, for splits (0,0,2), (0,1,1), (1,0,1), (1,1,0). Nullary lists with an empty universe. Both queries empty with u = 0 and u = 3.
  • High-arity cases. k = u full-arity atoms, repeated consistently and with conflicts, for u = 1..5.
  • Sampled. 300 seeded samples in each of the 36 cells u = 0..5, p+q = 0..5.
  • Large domains and boundaries. Large compressed domains (2^32 and usize::MAX−2, constant target size, OutputOutOfRange metadata). Numeric boundaries on both sides for every checked operation and for the i64 literal limit.
  • What each audit checks:
    • source feasibility against target satisfiability, using complete model enumeration;
    • identity value mapping;
    • every model extracts to a feasible folding;
    • one-hot selectors in every model;
    • assert_parameter_predictions;
    • for small targets, raw enumeration of all 2^n assignments against BruteForce.

Feature test

Run with the PR's pred:

  • pred list --rules and pred show list the new edge with its bounds.
  • pred create --example ConjunctiveQueryFoldability --to Satisfiability → pred reduce --via → pred solve returns [3, 3, 3]. pred evaluate --config [3,3,3] returns Or(true).
  • Aggregate route. pred reduce --aggregate gives Or(true) for the YES example and infeasible for a NO instance.
  • pred extract.
    • --value true and --value false map identically.
    • --config with the canonical witness recovers [3, 3, 3].
    • An unsatisfying witness fails with "target witness is infeasible".
    • A wrong-length witness fails with a length error.
  • Overflow. An instance with d+x+u = 2^64 fails with "integer overflow while counting fixed terms", exit 1, no panic.
  • Comparison and round trips. On 634 instances, pred reduce matched two independent reference encodings of the construction clause for clause. Round trips on 100 instances with independently known answers were all correct, 50 of them infeasible.
  • Builds. make check, clippy with -D warnings, fmt and make paper pass. Line coverage of the new module is 150/150.

Findings

None.

Out of scope, pre-existing on main, and worth separate issues:

  1. Parameter contracts render as unavailable. docs/paper/reductions.typ:643-645 tests o.contract == "bound-only", but the exported graph uses "upper_bound". Every bounded contract is therefore rendered as "unavailable" (137 of 297 rule contracts, including this rule's).
  2. Model doc says "containment". The ConjunctiveQueryFoldability model documentation (and paper p. 5890) describes the problem as equivalent to query containment. The evaluator actually tests exact equality of the substituted atom set with Q2.
  3. Unchecked sum in the model. ConjunctiveQueryFoldability::evaluate/dimensions add d+x+u unchecked, so an overflowing instance can panic in pred inspect/solve. This rule guards the sum itself.

@codecov

codecov Bot commented Nov 4, 2026

Copy link
Copy Markdown

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 97.20%. Comparing base (e7bacb3) to head (9b80a63).

Additional details and impacted files
@@           Coverage Diff            @@
##             main    #1241    +/-   ##
========================================
  Coverage   97.19%   97.20%            
========================================
  Files        1146     1148     +2     
  Lines      151558   152057   +499     
========================================
+ Hits       147308   147807   +499     
  Misses       4250     4250            

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
  • 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[Rule] ConjunctiveQueryFoldability to Satisfiability

1 participant