/**
 * @file mobius_evaluate.cpp
 * @brief Möbius-inversion exact route for safe UCQs (the last missing exact
 *        route of the Dalvi-Suciu dichotomy).
 *
 * Some unions of conjunctive queries are safe (PTIME data complexity) only
 * because the \#P-hard terms of their inclusion-exclusion expansion carry a
 * zero Möbius value on the CNF lattice and cancel.  The canonical witness is
 * QW / q9 (Dalvi-Suciu 2012; Monet & Olteanu 2018).  No other ProvSQL route
 * handles it in PTIME: the safe-query rewriter is per-CQ and hierarchical, the
 * query is not inversion-free (that is the point), and on adversarial data the
 * joint treewidth is unbounded.
 *
 * This file packages the **extensional** lattice-walking algorithm (Dalvi,
 * Schnaitter & Suciu, "Computing query probability with incidence algebras",
 * PODS 2010) the way the joint-width route already is: a compile-at-execution
 * step producing a certified circuit over the gathered data, and a linear
 * evaluation -- the only genuinely new evaluation primitive being a signed
 * linear combination at @c gate_mobius nodes (see @c provsql_utils.h and the
 * @c gate_mobius handling in @c probability_evaluate.cpp).
 *
 * The probability of a UCQ Q given in CNF as @f$\bigwedge_i d_i@f$ is, by
 * inclusion-exclusion on the @f$\lnot d_i@f$,
 * @f[ P(Q) = \sum_{\emptyset\neq s\subseteq[M]} (-1)^{|s|+1}
 *           P\Big(\bigvee_{i\in s} d_i\Big), @f]
 * and grouping the @f$\bigvee_{i\in s} d_i@f$ up to logical equivalence
 * collapses the hard term (its coefficient sums to zero -- the whole game).
 * Each surviving @f$\bigvee_{i\in s} d_i@f$ is a safe disjunctive sentence,
 * compiled recursively by the standard IndepStep / MobiusStep lifted-inference
 * recursion (component split, disjoint-symbol product, separator
 * independent-project, inner Möbius step) into certified-independent Boolean
 * islands, combined at the root @c gate_mobius by the signed coefficients.
 *
 * Self-joins and constants are handled by the two normalizations the JACM
 * dichotomy runs before its lifted recursion -- **ranking** (Def. 4.1) and
 * **shattering** (Prop. 2.10) -- implemented here as one shard split
 * (@c normalizeShards): each relation whose atoms do not all carry the trivial
 * all-distinct-variables pattern is partitioned into shard symbols, one per
 * (equality pattern, pinned query constants) signature of its tuples, and every
 * atom is expanded into the disjunction of its compatible shards.  Because the
 * shards partition the relation, disjoint shard symbols certify disjoint
 * tuples, which is what the independence steps of the recursion rest on.  A
 * within-disjunct self-join that survives normalization (two components over
 * one shard symbol, the JACM's @f$q_J@f$) is not independent and is not
 * declined either: it falls through to the Möbius step, whose
 * @f$P(c_1\land c_2) = P(c_1)+P(c_2)-P(c_1\lor c_2)@f$ is exactly the
 * disjunctive detour the self-join forces (JACM Example 3.1).
 *
 * Restrictions: tuple-independent (TID) inputs, one probabilistic tuple per
 * (relation, element tuple), and no two fact slots sharing a base tuple.  A
 * query shape the recursion cannot certify (an inversion, a \#P-hard CQ, a cap
 * exceeded) declines (a C++ exception caught at the SQL boundary), so the query
 * falls back to the normal provenance and never fails.
 */
extern "C" {
#include "postgres.h"
#include "fmgr.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "access/htup_details.h"   /* heap_form_tuple (PG 10 declares it here) */
#include "access/xact.h"
#include "catalog/pg_type.h"
#include "executor/spi.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/resowner.h"
#include "utils/uuid.h"

#include "compatibility.h"   /* TYPALIGN_INT / TYPALIGN_CHAR fallback for PG < 13 */
#include "provsql_utils.h"
#include "provsql_mmap.h"

/* Store-backed gate-type lookup (defined in provsql_mmap.c); used to enforce
 * the TID restriction (G3) -- every fact token must be a bare gate_input. */
extern Datum get_gate_type(PG_FUNCTION_ARGS);

/* Data-cost cap (provsql.mobius_max_gates GUC, defined in provsql.c): the
 * Möbius recursion is O(|D|^k), so a high-level safe query on large data is
 * bounded by declining once it has built this many gates. */
extern int provsql_mobius_max_gates;

PG_FUNCTION_INFO_V1(ucq_mobius_materialize_tracked);
PG_FUNCTION_INFO_V1(ucq_mobius_compile_stats);
PG_FUNCTION_INFO_V1(ucq_mobius_provenance_answer);
}

#include "c_cpp_compatibility.h"
#include "CertifiedDDMaterialize.h"
#include "mobius_evaluate.h"
#include "provsql_utils_cpp.h"

#include <algorithm>
#include <cmath>
#include <cstdint>
#include <functional>
#include <limits>
#include <map>
#include <set>
#include <stdexcept>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <vector>

namespace {

/** @brief Thrown when the Möbius route declines (unsafe shape, cap hit, ...). */
class MobiusDecline : public std::runtime_error {
public:
  explicit MobiusDecline(const std::string &w) : std::runtime_error(w) {}
};

/** @brief One argument position of an atom: a variable or a bound constant. */
struct Term {
  bool isVar;
  long v;        ///< Variable id (isVar) or constant element-id (!isVar).
  bool operator==(const Term &o) const { return isVar==o.isVar && v==o.v; }
};

/** @brief A relational atom: a relation symbol applied to terms. */
struct MAtom {
  unsigned rel;
  std::vector<Term> args;
};

using Disjunct = std::vector<MAtom>;   ///< A conjunction of atoms.
using Sentence = std::vector<Disjunct>;///< A disjunction of conjunctions (UCQ).

// ---------------------------------------------------------------------------
// Statistics surfaced by the stats SRF (the demonstrability requirement).
// ---------------------------------------------------------------------------
struct MobiusStats {
  int n_components = 0;       ///< distinct connected-component literals at top
  int n_cnf_conjuncts = 0;    ///< M, the CNF conjunct count at the top level
  int lattice_size = 0;       ///< 2^M subsets enumerated at the top level
  int lattice_collapsed = 0;  ///< distinct elements after equivalence collapse
  int n_nonzero = 0;          ///< elements with coefficient != 0
  int n_cancelled = 0;        ///< distinct elements with coefficient == 0
  bool cancelled_hard = false;///< some cancelled element is #P-hard (no separator)
  long dd_size = 0;           ///< gates materialised
  long memo_hits = 0;         ///< sentence-memo hits
  double probability = 0.0;   ///< P(Q)
};

// ===========================================================================
// Homomorphisms (CQ containment) over single-component conjunctions.
// ===========================================================================

/// Backtracking homomorphism search: does a mapping of @p p's variables to
/// @p q's terms exist sending every @p p-atom onto a @p q-atom (constants
/// matched verbatim)?  @p p, @p q are tiny (reduced-form components), so the
/// naive search is cheap.  Returns true iff @p p ⊑ @p q (i.e. @p q implies
/// @p p as a query -- a hom p->q means every model of p is a model of q is the
/// usual direction; here hom(p,q) maps p's atoms into q, witnessing q ⊑ p).
bool homExists(const Disjunct &p, const Disjunct &q,
               std::size_t ai, std::map<long,Term> &asg)
{
  if(ai == p.size())
    return true;
  const MAtom &pa = p[ai];
  for(const MAtom &qa : q) {
    if(qa.rel != pa.rel || qa.args.size() != pa.args.size())
      continue;
    std::vector<long> undo;
    bool ok = true;
    for(std::size_t k=0; k<pa.args.size(); ++k) {
      const Term &pt = pa.args[k];
      const Term &qt = qa.args[k];
      if(!pt.isVar) {
        if(qt.isVar || qt.v != pt.v) { ok=false; break; }
      } else {
        auto it = asg.find(pt.v);
        if(it == asg.end()) { asg[pt.v]=qt; undo.push_back(pt.v); }
        else if(!(it->second == qt)) { ok=false; break; }
      }
    }
    if(ok && homExists(p, q, ai+1, asg))
      return true;
    for(long u : undo) asg.erase(u);
  }
  return false;
}

/// hom from @p p into @p q (maps p's atoms onto q's): witnesses q ⊑ p.
bool hom(const Disjunct &p, const Disjunct &q)
{
  std::map<long,Term> asg;
  return homExists(p, q, 0, asg);
}

/// Logical equivalence of two single-component conjunctions: mutual
/// homomorphism, which is logical equivalence for conjunctive queries whether
/// or not the conjunction is a core (on a self-join-free component, where the
/// conjunction is its own core, it degenerates to isomorphism).
bool equiv(const Disjunct &p, const Disjunct &q)
{
  return hom(p,q) && hom(q,p);
}

// ===========================================================================
// Connected components.
// ===========================================================================

struct UF {
  std::vector<int> p;
  explicit UF(int n) : p(n) { for(int i=0;i<n;++i) p[i]=i; }
  int find(int x){ while(p[x]!=x){ p[x]=p[p[x]]; x=p[x]; } return x; }
  void uni(int a,int b){ p[find(a)]=find(b); }
};

/// Split a conjunction into its variable-connected components.  Distinct
/// components are probabilistically independent exactly when their fact
/// footprints are disjoint (@c disjointFootprints); a shared relation symbol
/// with no separating constant is a self-join, and goes to the Möbius step
/// instead.
std::vector<Disjunct> componentsOf(const Disjunct &d)
{
  const int n = static_cast<int>(d.size());
  UF uf(n);
  std::map<long, int> firstAtomOfVar;
  for(int i=0;i<n;++i)
    for(const auto &t : d[i].args)
      if(t.isVar) {
        auto it = firstAtomOfVar.find(t.v);
        if(it==firstAtomOfVar.end()) firstAtomOfVar[t.v]=i;
        else uf.uni(it->second, i);
      }
  std::map<int, Disjunct> byroot;
  for(int i=0;i<n;++i)
    byroot[uf.find(i)].push_back(d[i]);
  std::vector<Disjunct> out;
  for(auto &kv : byroot) out.push_back(std::move(kv.second));
  return out;
}

/// Is an atom fully ground (no variables)?
bool ground(const MAtom &a)
{
  for(const auto &t : a.args) if(t.isVar) return false;
  return true;
}

// ===========================================================================
// Data index (TID facts).
// ===========================================================================

struct FactIndex {
  // (rel, element tuple) -> provenance token (UUID string).  A certain
  // (untracked) fact has an empty token.
  std::map<std::pair<unsigned, std::vector<long>>, std::string> tok;
  std::set<std::pair<unsigned, std::vector<long>>> present;
  // (rel, position) -> active domain (distinct values seen there).
  std::map<std::pair<unsigned,int>, std::set<long>> domain;

  bool isPresent(unsigned rel, const std::vector<long> &el) const {
    return present.count({rel, el}) > 0;
  }
  const std::string *token(unsigned rel, const std::vector<long> &el) const {
    auto it = tok.find({rel, el});
    return it==tok.end() ? nullptr : &it->second;
  }
};

// ===========================================================================
// Canonical structural keys (memoisation, and duplicate elimination).
// ===========================================================================

std::string canonDisjunct(const Disjunct &d)
{
  // Canonically rename variables by first occurrence in a sorted atom order.
  // Build per-atom signatures with placeholder variable slots, then pick the
  // lexicographically smallest atom ordering greedily (sufficient: keys only
  // need same-structure -> same-key, not minimality).
  std::vector<std::string> atomsigRaw;
  for(const auto &a : d) {
    std::string sg = "r" + std::to_string(a.rel) + "(";
    for(const auto &t : a.args) {
      if(t.isVar) sg += "v"; else sg += "c"+std::to_string(t.v);
      sg += ",";
    }
    sg += ")";
    atomsigRaw.push_back(sg);
  }
  // Sort atoms by raw signature, then assign canonical var ids in first
  // occurrence order over that sorted sequence.
  std::vector<int> order(d.size());
  for(std::size_t i=0;i<d.size();++i) order[i]=static_cast<int>(i);
  std::sort(order.begin(), order.end(),
            [&](int a,int b){ return atomsigRaw[a]<atomsigRaw[b]; });
  std::map<long,int> ren;
  std::string out;
  for(int idx : order) {
    const MAtom &a = d[idx];
    out += "r"+std::to_string(a.rel)+"(";
    for(const auto &t : a.args) {
      if(t.isVar) {
        auto it = ren.find(t.v);
        int id = (it==ren.end()) ? (ren[t.v]=static_cast<int>(ren.size())) : it->second;
        out += "v"+std::to_string(id);
      } else out += "c"+std::to_string(t.v);
      out += ",";
    }
    out += ")";
  }
  return out;
}

std::string canonSentence(const Sentence &s)
{
  std::vector<std::string> parts;
  for(const auto &d : s) parts.push_back(canonDisjunct(d));
  std::sort(parts.begin(), parts.end());
  parts.erase(std::unique(parts.begin(), parts.end()), parts.end());
  std::string out;
  for(const auto &p : parts){ out += p; out += "|"; }
  return out;
}

// ===========================================================================
// Independence certificates: fact footprints.
//
// Every independent gate the compiler builds (the disjoint-symbol product, the
// independent union, the separator's independent project) is licensed by ONE
// property: the sub-sentences it combines read disjoint sets of base tuples.
// A footprint records that symbolically -- per atom, its relation and the
// positions it pins to a constant -- and two footprints are certified disjoint
// when every pair of atom patterns over one relation pins a common position to
// two different constants (no tuple can satisfy both).  Sound, and deliberately
// not complete: an undetected disjointness costs a decline, never a wrong
// answer.
// ===========================================================================

/// One atom's reach: its relation and its position -> constant pins.
using FactPattern = std::pair<unsigned, std::map<int,long>>;
using Footprint   = std::set<FactPattern>;

void addToFootprint(const Disjunct &d, Footprint &fp)
{
  for(const auto &a : d) {
    std::map<int,long> c;
    for(std::size_t i=0;i<a.args.size();++i)
      if(!a.args[i].isVar) c[static_cast<int>(i)] = a.args[i].v;
    fp.insert({a.rel, std::move(c)});
  }
}

Footprint footprintOf(const Disjunct &d)
{
  Footprint fp; addToFootprint(d, fp); return fp;
}

/// Can no tuple satisfy both atom patterns?
bool patternsDisjoint(const FactPattern &x, const FactPattern &y)
{
  if(x.first != y.first)
    return true;                       // distinct relations: disjoint tuples
  for(const auto &kv : x.second) {
    auto it = y.second.find(kv.first);
    if(it != y.second.end() && it->second != kv.second)
      return true;                     // a position pinned to two constants
  }
  return false;
}

bool disjointFootprints(const Footprint &a, const Footprint &b)
{
  for(const auto &x : a)
    for(const auto &y : b)
      if(!patternsDisjoint(x,y)) return false;
  return true;
}

// ===========================================================================
// Shard normalization: ranking (Dalvi-Suciu Def. 4.1) and shattering
// (Prop. 2.10).
//
// The lifted recursion certifies independence by relation symbol, which holds
// only when every atom over a symbol reads the same, full relation.  Two
// rewrites restore that:
//
//  - *shattering* separates an atom that pins a query constant (S(a,y), left
//    behind by a head pin or by a separator substitution) from one that does
//    not (S(x,z)): they overlap, and the shards S_{0=a} / S_{0≠a} do not;
//  - *ranking* separates an atom with a repeated variable (S(x,x)) from one
//    with distinct variables: the shards are the equality patterns of the
//    tuples (at most Bell(k) of them for arity k).
//
// Both are the same construction: partition each relation's tuples by their
// signature -- the equality pattern of their positions, plus the query constant
// each block is pinned to -- allocate one shard symbol per signature, and
// expand every atom into the disjunction of the shards it can match, unifying
// the variables and substituting the constants each shard forces.  Since every
// tuple lands in exactly one shard, the shards are disjoint (which is what the
// independence certificates need) and each shard atom is back in reduced form:
// distinct variables, no constants (pinned and duplicated positions are
// projected away, so a shard has the arity of its free blocks).
//
// The pass is a no-op -- symbols, facts and cost untouched -- on a sentence
// whose atoms all carry the trivial all-distinct-variables pattern, which is
// every query the array decoder builds; it fires on head-pinned per-answer
// queries and on manually written descriptors with repeated variables.
// ===========================================================================

/// Sentinel in @c ShardSig::cst for a block pinned to no query constant.
const long SHARD_FREE = std::numeric_limits<long>::min();

/// Signature of a tuple: for each position the representative position of its
/// equality block, and per representative the query constant the block is
/// pinned to (@c SHARD_FREE if its value is not a query constant).
struct ShardSig {
  std::vector<int>  rep;
  std::vector<long> cst;
  bool operator<(const ShardSig &o) const {
    if(rep != o.rep) return rep < o.rep;
    return cst < o.cst;
  }
};

/// One shard of a relation: its allocated symbol, its signature, and the
/// representative positions that survive the projection (its arity).
struct Shard {
  unsigned sym = 0;
  ShardSig sig;
  std::vector<int> freepos;
};

/// Does this atom carry a non-trivial pattern (a constant or a repeated term),
/// i.e. does its relation need to be split into shards?
bool atomNeedsShards(const MAtom &a)
{
  for(std::size_t i=0;i<a.args.size();++i) {
    if(!a.args[i].isVar) return true;
    for(std::size_t j=0;j<i;++j)
      if(a.args[j] == a.args[i]) return true;
  }
  return false;
}

ShardSig shardSigOf(const std::vector<long> &el, const std::set<long> &consts)
{
  const int k = static_cast<int>(el.size());
  ShardSig sg;
  sg.rep.assign(k, 0);
  sg.cst.assign(k, SHARD_FREE);
  for(int i=0;i<k;++i) {
    int r = i;
    for(int j=0;j<i;++j) if(el[j]==el[i]) { r = sg.rep[j]; break; }
    sg.rep[i] = r;
  }
  for(int i=0;i<k;++i)
    if(sg.rep[i]==i && consts.count(el[i])) sg.cst[i] = el[i];
  return sg;
}

/// Can @p a match a tuple of shard @p sh?  (A pre-filter: the equalities the
/// shard forces are applied afterwards, and re-checked on the result.)
bool shardCompatible(const MAtom &a, const Shard &sh)
{
  const int k = static_cast<int>(a.args.size());
  if(static_cast<int>(sh.sig.rep.size()) != k)
    return false;                       // arity mismatch with the data
  for(int p=0;p<k;++p) {
    const int r = sh.sig.rep[p];
    const Term &tp = a.args[p], &tr = a.args[r];
    if(!tp.isVar && !tr.isVar && tp.v != tr.v)
      return false;                     // one block, two different constants
    if(sh.sig.cst[r] != SHARD_FREE) {
      if(!tp.isVar && tp.v != sh.sig.cst[r]) return false;
    } else if(!tp.isVar) {
      return false;                     // a free block holds no query constant
    }
  }
  for(int p=0;p<k;++p)
    for(int q=p+1;q<k;++q)
      if(sh.sig.rep[p]!=sh.sig.rep[q] && a.args[p]==a.args[q])
        return false;                   // distinct blocks, one term
  return true;
}

/// Union-find over a disjunct's variables with per-class constant bindings:
/// the substitution a choice of shards forces.
class ShardSubst {
public:
  /// Unify two terms; false if that is inconsistent.
  bool unify(const Term &x, const Term &y) {
    if(!x.isVar && !y.isVar) return x.v == y.v;
    if(!x.isVar) return bind(y.v, x.v);
    if(!y.isVar) return bind(x.v, y.v);
    const long rx = find(x.v), ry = find(y.v);
    if(rx == ry) return true;
    auto cx = bound.find(rx), cy = bound.find(ry);
    if(cx!=bound.end() && cy!=bound.end() && cx->second != cy->second)
      return false;
    parent[ry] = rx;
    if(cy != bound.end()) {
      const long v = cy->second;
      bound.erase(ry);
      bound[rx] = v;
    }
    return true;
  }
  /// The term a variable / constant becomes under the substitution.
  Term apply(const Term &t) {
    if(!t.isVar) return t;
    const long r = find(t.v);
    auto it = bound.find(r);
    Term o;
    if(it != bound.end()) { o.isVar=false; o.v=it->second; }
    else                  { o.isVar=true;  o.v=r; }
    return o;
  }
private:
  std::map<long,long> parent;   // variable -> parent (union-find)
  std::map<long,long> bound;    // class root -> the constant it is pinned to
  long find(long v) {
    auto it = parent.find(v);
    if(it == parent.end()) { parent[v]=v; return v; }
    if(it->second == v) return v;
    const long r = find(it->second);
    parent[v] = r;
    return r;
  }
  bool bind(long var, long c) {
    const long r = find(var);
    auto it = bound.find(r);
    if(it != bound.end()) return it->second == c;
    bound[r] = c;
    return true;
  }
};

/// Cap on the disjuncts the shard expansion may produce (it is a product over
/// the atoms of a disjunct); past it the route declines to joint-width.  Well
/// above what the rest of the compiler can consume anyway: the CNF conjunct cap
/// and the transversal bound both bite long before a union this wide compiles.
const std::size_t SHARD_MAX_DISJUNCTS = 256;

/**
 * @brief One rank + shatter pass of @p s over @p fi, splitting the relations
 *        in @p need.
 *
 * The expansion substitutes constants and unifies variables, so it can leave a
 * non-trivial pattern on an atom of a relation that was NOT in @p need (an
 * unshattered @c R(u) becomes @c R(a) in the branch where @c u is pinned).
 * Such relations are reported through @p missing, and @c normalizeShards
 * iterates to a fixpoint.
 */
bool normalizeShardsPass(const Sentence &s, const FactIndex &fi,
                         const std::set<unsigned> &need,
                         const std::set<long> &consts, unsigned maxRel,
                         Sentence &sOut, FactIndex &fiOut,
                         std::set<unsigned> &missing)
{
  missing.clear();
  if(need.empty())
    return false;

  // Shard the facts: one shard per (relation, tuple signature), the tuple
  // projected onto the shard's free blocks.  A relation that needs no splitting
  // is copied through unchanged.
  std::map<unsigned, std::map<ShardSig, Shard>> shards;
  unsigned next = maxRel + 1;
  fiOut = FactIndex();
  for(const auto &kv : fi.tok) {
    const unsigned rel = kv.first.first;
    const std::vector<long> &el = kv.first.second;
    unsigned sym = rel;
    std::vector<long> proj = el;
    if(need.count(rel)) {
      const ShardSig sg = shardSigOf(el, consts);
      auto &byrel = shards[rel];
      auto sit = byrel.find(sg);
      if(sit == byrel.end()) {
        Shard sh;
        sh.sym = next++;
        sh.sig = sg;
        for(std::size_t i=0;i<el.size();++i)
          if(sg.rep[i]==static_cast<int>(i) && sg.cst[i]==SHARD_FREE)
            sh.freepos.push_back(static_cast<int>(i));
        sit = byrel.emplace(sg, sh).first;
      }
      sym = sit->second.sym;
      proj.clear();
      for(int p : sit->second.freepos) proj.push_back(el[p]);
    }
    fiOut.tok[{sym, proj}] = kv.second;
    fiOut.present.insert({sym, proj});
    for(std::size_t i=0;i<proj.size();++i)
      fiOut.domain[{sym, static_cast<int>(i)}].insert(proj[i]);
  }

  // Expand the sentence: per disjunct, every combination of one compatible
  // shard per atom, under the substitution that combination forces.
  sOut.clear();
  std::set<std::string> emitted;
  for(const Disjunct &d : s) {
    std::vector<std::vector<const Shard *>> opts;
    unsigned long prod = 1;
    bool dead = false;
    for(const MAtom &a : d) {
      std::vector<const Shard *> o;
      if(!need.count(a.rel)) {
        o.push_back(nullptr);                    // kept as is
      } else {
        auto rit = shards.find(a.rel);
        if(rit != shards.end())
          for(const auto &kv : rit->second)
            if(shardCompatible(a, kv.second)) o.push_back(&kv.second);
      }
      if(o.empty()) { dead = true; break; }       // no tuple matches this atom
      prod *= o.size();
      if(prod > SHARD_MAX_DISJUNCTS)
        throw MobiusDecline("Möbius: ranking/shattering expansion cap exceeded");
      opts.push_back(std::move(o));
    }
    if(dead)
      continue;                                   // the conjunction is FALSE

    std::vector<const Shard *> choice(opts.size(), nullptr);
    std::function<void(std::size_t)> expand = [&](std::size_t ai) {
      if(ai < opts.size()) {
        for(const Shard *sh : opts[ai]) { choice[ai]=sh; expand(ai+1); }
        return;
      }
      // The equalities and constants this combination forces.
      ShardSubst sub;
      for(std::size_t i=0;i<d.size();++i) {
        const Shard *sh = choice[i];
        if(sh == nullptr) continue;
        for(std::size_t p=0;p<d[i].args.size();++p) {
          const int r = sh->sig.rep[p];
          if(r != static_cast<int>(p) && !sub.unify(d[i].args[p], d[i].args[r]))
            return;
          if(sh->sig.cst[r] != SHARD_FREE) {
            Term c; c.isVar=false; c.v=sh->sig.cst[r];
            if(!sub.unify(d[i].args[p], c)) return;
          }
        }
      }
      // Apply it, and re-check every atom against its shard: distinct blocks
      // must not have collapsed onto one term, and a free block must not have
      // been pinned to a query constant (its tuples carry none).
      Disjunct nd;
      for(std::size_t i=0;i<d.size();++i) {
        const Shard *sh = choice[i];
        std::vector<Term> ap;
        ap.reserve(d[i].args.size());
        for(const Term &t : d[i].args) ap.push_back(sub.apply(t));
        MAtom na;
        if(sh == nullptr) {
          na.rel = d[i].rel;
          na.args = std::move(ap);
          if(atomNeedsShards(na))
            missing.insert(na.rel);   // the substitution left a pattern on it
        } else {
          for(std::size_t p=0;p<ap.size();++p)
            for(std::size_t q=p+1;q<ap.size();++q)
              if(sh->sig.rep[p]!=sh->sig.rep[q] && ap[p]==ap[q]) return;
          for(int p : sh->freepos)
            if(!ap[p].isVar) return;
          na.rel = sh->sym;
          for(int p : sh->freepos) na.args.push_back(ap[p]);
        }
        bool dup = false;
        for(const MAtom &prev : nd)
          if(prev.rel==na.rel && prev.args==na.args) { dup=true; break; }
        if(!dup) nd.push_back(std::move(na));
      }
      if(nd.empty()) return;
      if(sOut.size() >= SHARD_MAX_DISJUNCTS)
        throw MobiusDecline("Möbius: ranking/shattering expansion cap exceeded");
      sOut.push_back(std::move(nd));
    };
    expand(0);
  }

  // Drop duplicate disjuncts (idempotent in the union, and a smaller sentence
  // keeps the CNF lattice small).
  {
    Sentence uniq;
    for(const Disjunct &d : sOut)
      if(emitted.insert(canonDisjunct(d)).second) uniq.push_back(d);
    sOut = std::move(uniq);
  }

  // Re-globalise the variables: several expanded disjuncts descend from one
  // original disjunct and would otherwise share its variable ids, while their
  // existential quantifiers are separate.  The recursion unifies variables
  // ACROSS disjuncts by (relation, position), so a shared id would conflate
  // two independent variables (and cost a separator).
  {
    long nextvar = 1;
    for(const auto &d : s)
      for(const auto &a : d)
        for(const auto &t : a.args)
          if(t.isVar) nextvar = std::max(nextvar, t.v + 1);
    for(Disjunct &d : sOut) {
      std::map<long,long> ren;
      for(MAtom &a : d)
        for(Term &t : a.args)
          if(t.isVar) {
            auto it = ren.find(t.v);
            if(it == ren.end()) it = ren.emplace(t.v, nextvar++).first;
            t.v = it->second;
          }
    }
  }
  return true;
}

/**
 * @brief Rank + shatter @p s over @p fi into an equivalent reduced-form
 *        sentence over shard symbols.
 *
 * @return @c false when no relation needs splitting (then @p sOut / @p fiOut
 *         are untouched and the caller keeps the originals); @c true when the
 *         normalized pair has been written to @p sOut / @p fiOut.
 */
bool normalizeShards(const Sentence &s, const FactIndex &fi,
                     Sentence &sOut, FactIndex &fiOut)
{
  // Which relations need splitting, and the constants to shatter on (all
  // constants of the sentence: a variable can be substituted into any of them
  // by another atom's shard, so the closure is the whole set).
  std::set<unsigned> need;
  std::set<long> consts;
  std::set<unsigned> rels;
  unsigned maxRel = 0;
  for(const auto &d : s)
    for(const auto &a : d) {
      maxRel = std::max(maxRel, a.rel);
      rels.insert(a.rel);
      if(atomNeedsShards(a)) need.insert(a.rel);
      for(const auto &t : a.args) if(!t.isVar) consts.insert(t.v);
    }
  if(need.empty())
    return false;
  // Shard symbols are allocated above every EXISTING symbol -- including the
  // relations the gather brought in but this sentence does not mention (a
  // per-answer head pin compiles one disjunct of a wider descriptor), whose
  // facts are copied through under their own symbol.
  for(const auto &kv : fi.tok)
    maxRel = std::max(maxRel, kv.first.first);

  // Re-run until no unsharded relation is left carrying a pattern; each round
  // adds at least one relation, so there are at most |relations| of them.
  for(std::size_t round=0; round<=rels.size(); ++round) {
    std::set<unsigned> missing;
    const bool done = normalizeShardsPass(s, fi, need, consts, maxRel,
                                          sOut, fiOut, missing);
    if(missing.empty())
      return done;
    need.insert(missing.begin(), missing.end());
  }
  throw MobiusDecline("Möbius: ranking/shattering did not converge");
}

// ===========================================================================
// The compiler.
// ===========================================================================

class MobiusCompiler {
public:
  MobiusCompiler(const FactIndex &fi, MobiusStats &st) : fi(fi), st(st) {
    const constants_t c = get_constants(true);
    times_oid = c.OID_FUNCTION_PROVENANCE_TIMES;
    plus_oid  = c.OID_FUNCTION_PROVENANCE_PLUS;
    if(!OidIsValid(times_oid) || !OidIsValid(plus_oid))
      provsql_error("ucq_mobius: provenance_times / provenance_plus unavailable");
  }

  /// Compile the top sentence, returning the root token.  @p lineage is the
  /// token of the literal Boolean provenance of the query (the normal lineage
  /// the route falls back to); it is carried on the root gate_mobius as a
  /// designated transparent child so every non-probability evaluator (semiring,
  /// Shapley, Banzhaf, PROV export) sees the literal lineage and works, while
  /// probability uses the signed Möbius combination.  Empty = no lineage
  /// (measure-only; the manual descriptor entry points without a fallback).
  pg_uuid_t compileTop(const Sentence &s, const std::string &lineage = "") {
    pending_lineage = lineage;
    lineage_consumed = false;
    pg_uuid_t root = compile(s, /*top=*/true);
    if(!lineage_consumed && !(lineage.empty() && isMobiusGate(root)))
      // The top did not go through a Möbius step (a safe query that reached
      // this route, or one whose top rule was an independent union): wrap it in
      // a thin gate_mobius selector carrying the lineage, if any, and the value
      // with coefficient 1.  The root must be a gate_mobius even so: it is what
      // routes the token to the Möbius evaluator (the only one that reads the
      // signed combinations of any nested gate_mobius) and what
      // @c mobius_or_null tests to tell a success from a decline.
      //
      // Skipped when there is no lineage to attach AND the compiled root is
      // already a gate_mobius: the wrapper would then be a pure identity (a
      // one-element combination with coefficient 1) over a gate that already
      // satisfies both requirements.  That is the measure-only entry points'
      // case (@c ucq_mobius_provenance and friends pass no lineage, so
      // @c lineage_consumed never becomes true), where the top-level Möbius
      // step's own combination is the root the caller should see.  A root that
      // is NOT a gate_mobius -- a gate_zero / gate_one from a combination that
      // collapsed to a constant -- still gets wrapped, since only the wrapper
      // makes such a token recognisable as a Möbius success.
      root = mkMobius({root}, {1}, lineage);
    return root;
  }

private:
  const FactIndex &fi;
  MobiusStats &st;
  Oid times_oid = InvalidOid;        // provsql.provenance_times
  Oid plus_oid  = InvalidOid;        // provsql.provenance_plus
  std::unordered_map<std::string, std::string> memo;  // key -> uuid string
  std::unordered_set<std::string> created;
  std::string pending_lineage;   // lineage to inline into the top Möbius step
  bool lineage_consumed = false;

  // -- materialisation -----------------------------------------------------

  /// Delegate plus / times construction to the system's provenance_times /
  /// provenance_plus (the single source of truth: they filter the semiring
  /// identities, short-circuit 0 / 1 surviving children, and content-address
  /// the gate in the store).  The Möbius path must NOT mint its own plus / times
  /// gates -- using these functions keeps one gate-construction code path and
  /// lets the Möbius islands share sub-gates with the rest of the system.
  pg_uuid_t callProvenance(bool isAnd, const std::vector<pg_uuid_t> &ch) {
    std::vector<Datum> datums;
    datums.reserve(ch.size());
    for(const auto &u : ch)
      datums.push_back(UUIDPGetDatum(const_cast<pg_uuid_t *>(&u)));
    ArrayType *arr = construct_array(datums.empty()?nullptr:datums.data(),
                                     static_cast<int>(datums.size()),
                                     UUIDOID, 16, false, TYPALIGN_CHAR);
    Datum res = OidFunctionCall1(isAnd ? times_oid : plus_oid,
                                 PointerGetDatum(arr));
    /* Data-cost cap: the lifted-inference recursion is O(|D|^k), so bound the
     * total work and decline (-> joint-width / the ladder) past the cap rather
     * than let a high-level safe query on large data out-cost the general
     * pipeline.  0 disables the cap. */
    ++st.dd_size;
    if(provsql_mobius_max_gates > 0 &&
       static_cast<long>(st.dd_size) > static_cast<long>(provsql_mobius_max_gates))
      throw MobiusDecline("Möbius: data-cost cap (provsql.mobius_max_gates) "
                          "exceeded -- the |D|^k recursion is too large");
    return *DatumGetUUIDP(res);
  }

  /// gate_one (empty times) / gate_zero (empty plus).
  pg_uuid_t mkConst(bool one) { return callProvenance(one, {}); }

  /// True iff @p u is a @c gate_mobius this compilation minted.  @c created
  /// holds exactly the combination gates @c mkMobius produced (including one it
  /// found already content-addressed in the store), so this needs no store
  /// round-trip.  Used by @c compileTop to skip an identity wrapper.
  bool isMobiusGate(const pg_uuid_t &u) const {
    return created.count(uuid2string(u)) != 0;
  }

  /// Independent OR / AND over child tokens.  Children are deduplicated first:
  /// the certified-independent evaluation is read-once, so a child must not be
  /// double-counted (provenance_plus / provenance_times keep duplicates).  The
  /// 0 / 1-child and identity cases are handled by the provenance functions.
  pg_uuid_t mkBool(bool isAnd, std::vector<pg_uuid_t> ch) {
    std::vector<std::string> texts;
    texts.reserve(ch.size());
    for(const auto &c : ch) texts.push_back(uuid2string(c));
    std::sort(texts.begin(), texts.end());
    texts.erase(std::unique(texts.begin(), texts.end()), texts.end());
    std::vector<pg_uuid_t> uniq;
    uniq.reserve(texts.size());
    for(const auto &t : texts) uniq.push_back(string2uuid(t));
    return callProvenance(isAnd, uniq);
  }

  /// Signed Möbius combination gate over @p children with integer @p coeffs.
  /// Coefficients are stored keyed by child UUID (@c "uuid:coeff" tokens in
  /// @c extra), so evaluation is robust to any child reordering / dedup the
  /// store may apply; duplicate children are merged (coefficients summed) and
  /// zero-coefficient children dropped.
  /// @p lineage (optional): the literal-lineage token, carried as a
  /// designated transparent child marked @c "L:<uuid>" in @c extra (child 0).
  pg_uuid_t mkMobius(const std::vector<pg_uuid_t> &children,
                     const std::vector<long> &coeffs,
                     const std::string &lineage = "") {
    std::map<std::string,long> bycoeff;
    std::vector<std::string> order;
    for(std::size_t i=0;i<children.size();++i) {
      const std::string u = uuid2string(children[i]);
      if(!bycoeff.count(u)) order.push_back(u);
      bycoeff[u] += coeffs[i];
    }
    std::vector<pg_uuid_t> ch;
    std::string extra, name = "mobius[";
    if(!lineage.empty()) {
      ch.push_back(string2uuid(lineage));     // the literal lineage, child 0
      extra += "L:" + lineage + " ";
      name  += "L:" + lineage + ",";
    }
    for(const std::string &u : order) {
      if(bycoeff[u]==0) continue;
      ch.push_back(string2uuid(u));
      extra += u + ":" + std::to_string(bycoeff[u]) + " ";
      name  += u + ":" + std::to_string(bycoeff[u]) + ",";
    }
    name += "]";
    if(ch.empty() || (!lineage.empty() && ch.size()==1))
      return mkConst(false);   // no surviving combination -> probability 0
    pg_uuid_t u = provsqlUuidV5(name);
    if(created.insert(uuid2string(u)).second) {
      provsql_internal_create_gate(&u, gate_mobius,
                                   static_cast<unsigned>(ch.size()),
                                   ch.data());
      provsql_internal_set_extra(&u, extra.c_str());
      ++st.dd_size;
    }
    return u;
  }

  // -- recursion -----------------------------------------------------------

  pg_uuid_t compile(const Sentence &sentence, bool top=false);

  /// Find a separator: a unification class of variable positions present in
  /// every atom of every disjunct.  Returns the (rel,pos) occurrences of the
  /// class via @p occ and per-disjunct the variable ids of the class via
  /// @p classVarsPerDisj; returns false if no separator exists.
  bool findSeparator(const Sentence &s,
                     std::set<std::pair<unsigned,int>> &occ,
                     std::vector<std::set<long>> &classVarsPerDisj);

  /// Möbius (inclusion-exclusion) step over the components of @p s.
  pg_uuid_t mobiusStep(const Sentence &s, bool top);
};

// --- separator -------------------------------------------------------------

bool MobiusCompiler::findSeparator(const Sentence &s,
    std::set<std::pair<unsigned,int>> &occOut,
    std::vector<std::set<long>> &classVarsPerDisj)
{
  // Collect all variable ids and union those sharing a (rel,pos).
  std::map<long,int> id;
  auto vid = [&](long v)->int{
    auto it=id.find(v); if(it!=id.end()) return it->second;
    int n=static_cast<int>(id.size()); id[v]=n; return n;
  };
  for(const auto &d : s) for(const auto &a : d) for(const auto &t : a.args)
    if(t.isVar) vid(t.v);
  if(id.empty()) return false;
  UF uf(static_cast<int>(id.size()));
  std::map<std::pair<unsigned,int>, long> firstAtPos;
  for(const auto &d : s) for(const auto &a : d)
    for(std::size_t k=0;k<a.args.size();++k) {
      if(!a.args[k].isVar) continue;
      auto key = std::make_pair(a.rel, static_cast<int>(k));
      auto it = firstAtPos.find(key);
      if(it==firstAtPos.end()) firstAtPos[key]=a.args[k].v;
      else uf.uni(vid(it->second), vid(a.args[k].v));
    }
  // Candidate classes = class roots.  A class is a separator iff in every
  // disjunct, every atom contains a variable of that class.
  std::vector<long> idToVar(id.size());
  for(auto &kv : id) idToVar[kv.second]=kv.first;

  std::set<int> roots;
  for(int i=0;i<static_cast<int>(id.size());++i) roots.insert(uf.find(i));

  for(int root : roots) {
    bool covers = true;
    std::vector<std::set<long>> cvars(s.size());
    for(std::size_t di=0; di<s.size() && covers; ++di) {
      const Disjunct &d = s[di];
      for(const auto &a : d) {
        bool atomHas=false;
        for(const auto &t : a.args)
          if(t.isVar && uf.find(vid(t.v))==root) {
            atomHas=true; cvars[di].insert(t.v);
          }
        if(!atomHas){ covers=false; break; }
      }
    }
    if(!covers) continue;
    // A separator binds exactly ONE variable per disjunct: the disjunct's whole
    // conjunction then sits under that single existential and Q ≡ ⋁_a Q[x:=a].
    // With two class variables in one disjunct, substituting the same constant
    // for both is a strictly stronger query (∃x φ ∧ ∃y ψ is not ⋁_a φ(a)∧ψ(a)),
    // so this class is not a separator -- try the next one.
    {
      bool single = true;
      for(const auto &cv : cvars) if(cv.size() != 1) { single = false; break; }
      if(!single) continue;
    }
    // Record the (rel,pos) occurrences of this class (the substitution sites,
    // and the active domain to project over).
    std::set<std::pair<unsigned,int>> occ;
    for(const auto &d : s) for(const auto &a : d)
      for(std::size_t k=0;k<a.args.size();++k)
        if(a.args[k].isVar && uf.find(vid(a.args[k].v))==root)
          occ.insert({a.rel, static_cast<int>(k)});
    // Each relation must carry the class at a single position: the piece for
    // constant a then reads only tuples with a at that position, so pieces for
    // distinct constants are over disjoint tuples -- the certificate the
    // independent project rests on.  (Two class positions of one relation would
    // let a tuple be read by one piece through one and by another through the
    // other.)
    {
      std::set<unsigned> rels;
      bool onePos = true;
      for(const auto &rp : occ)
        if(!rels.insert(rp.first).second) { onePos = false; break; }
      if(!onePos) continue;
    }
    occOut = std::move(occ);
    classVarsPerDisj = std::move(cvars);
    return true;
  }
  return false;
}

// --- Möbius step -----------------------------------------------------------

pg_uuid_t MobiusCompiler::mobiusStep(const Sentence &s, bool top)
{
  // 1. Components of each disjunct -> a global literal pool with equivalence
  //    collapse.  Each disjunct becomes the set of its component literal ids.
  std::vector<Disjunct> literals;          // representative component per id
  auto litId = [&](const Disjunct &c)->int{
    for(std::size_t i=0;i<literals.size();++i)
      if(equiv(literals[i], c)) return static_cast<int>(i);
    literals.push_back(c);
    return static_cast<int>(literals.size()-1);
  };
  std::vector<std::set<int>> terms;        // DNF terms (per disjunct)
  for(const auto &d : s) {
    std::set<int> t;
    for(const auto &c : componentsOf(d)) t.insert(litId(c));
    terms.push_back(std::move(t));
  }
  if(top) st.n_components = static_cast<int>(literals.size());

  // Implication matrix on literals: imp[i][j] == (lit_i implies lit_j),
  // i.e. lit_i ⊑ lit_j, witnessed by hom(lit_j -> lit_i).
  const int L = static_cast<int>(literals.size());
  std::vector<std::vector<char>> imp(L, std::vector<char>(L, 0));
  for(int i=0;i<L;++i) for(int j=0;j<L;++j)
    imp[i][j] = hom(literals[j], literals[i]) ? 1 : 0;

  // Minimise a disjunction (literal set): drop lit i if some other lit j in
  // the set is implied-BY i (i ⊑ j), since i ∨ j ≡ j.  Keep the weakest.
  auto minimiseDisj = [&](std::set<int> in)->std::vector<int>{
    std::vector<int> v(in.begin(), in.end());
    std::vector<char> keep(v.size(),1);
    for(std::size_t a=0;a<v.size();++a) for(std::size_t b=0;b<v.size();++b)
      if(a!=b && keep[b] && imp[v[a]][v[b]]) { keep[a]=0; break; }
    std::vector<int> out;
    for(std::size_t a=0;a<v.size();++a) if(keep[a]) out.push_back(v[a]);
    std::sort(out.begin(), out.end());
    return out;
  };

  // 2. DNF -> CNF: a clause picks one literal from each term and ORs them.
  //    Cap the conjunct count.
  std::set<std::vector<int>> clauseSet;
  std::vector<std::set<int>> clauses;  // working as literal sets
  // Enumerate transversals.
  std::vector<int> pick(terms.size(), 0);
  // Bound the product to avoid blow-up before the M cap is even computed.
  unsigned long prod = 1;
  for(const auto &t : terms) prod *= std::max<std::size_t>(1, t.size());
  if(prod > 100000UL)
    throw MobiusDecline("Möbius: DNF->CNF transversal count too large");
  std::function<void(std::size_t,std::set<int>&)> gen =
    [&](std::size_t ti, std::set<int> &acc){
      if(ti==terms.size()){
        std::vector<int> mn = minimiseDisj(acc);
        if(!mn.empty()) clauseSet.insert(mn);
        return;
      }
      for(int lit : terms[ti]) {
        bool fresh = acc.insert(lit).second;
        gen(ti+1, acc);
        if(fresh) acc.erase(lit);
      }
    };
  { std::set<int> acc; gen(0, acc); }

  // Clause subsumption: drop clause B if some clause A ⊑ B (every literal of
  // A is implied by some literal of B... no: ⋁A implies ⋁B iff every a∈A has
  // a b∈B with a ⊑ b).  Then B is redundant in the conjunction.
  std::vector<std::vector<int>> cl(clauseSet.begin(), clauseSet.end());
  auto disjImplies = [&](const std::vector<int>&A, const std::vector<int>&B){
    for(int a : A){ bool ok=false; for(int b : B) if(imp[a][b]){ ok=true; break; }
                    if(!ok) return false; }
    return true;
  };
  std::vector<char> keepCl(cl.size(),1);
  for(std::size_t a=0;a<cl.size();++a) for(std::size_t b=0;b<cl.size();++b)
    if(a!=b && keepCl[a] && cl[a]!=cl[b] && disjImplies(cl[a],cl[b]))
      keepCl[b]=0;
  std::vector<std::vector<int>> M;
  for(std::size_t a=0;a<cl.size();++a) if(keepCl[a]) M.push_back(cl[a]);

  const int m = static_cast<int>(M.size());
  if(top) { st.n_cnf_conjuncts = m; st.lattice_size = (m<=30)?(1<<m):0; }
  if(m == 1) {
    // The CNF collapsed to a single clause: the sentence is logically that
    // clause, a pure disjunction of components (its conjunctive structure was
    // redundant -- the shape a self-join with an implication between its
    // components leaves behind).  There is no inclusion-exclusion to run, but
    // the simpler sentence may well compile; recurse on it unless it is the
    // sentence we already have (no progress -> decline).
    Sentence el;
    for(int lid : M[0]) el.push_back(literals[lid]);
    if(canonSentence(el) != canonSentence(s))
      return compile(el, top);
  }
  if(m < 2)
    throw MobiusDecline("Möbius: no inclusion-exclusion structure (M<2); "
                        "sentence has no separator and does not decompose");
  if(provsql_mobius_max_cnf > 0 && m > provsql_mobius_max_cnf)
    throw MobiusDecline("Möbius: CNF conjunct count exceeds the cap "
                        "(provsql.mobius_max_cnf)");

  // 3. Enumerate non-empty subsets s of [m]; ψ_s = ⋁ of literals across the
  //    clauses in s, minimised.  Accumulate coefficient (-1)^{|s|+1} per
  //    distinct ψ_s (keyed by its minimised literal-id set).
  std::map<std::vector<int>, long> coeff;
  for(unsigned mask=1; mask < (1u<<m); ++mask) {
    std::set<int> lits;
    for(int i=0;i<m;++i) if(mask & (1u<<i))
      lits.insert(M[i].begin(), M[i].end());
    std::vector<int> key = minimiseDisj(lits);
    const int pc = __builtin_popcount(mask);
    coeff[key] += (pc & 1) ? 1 : -1;
  }
  if(top) st.lattice_collapsed = static_cast<int>(coeff.size());

  // 4. Build the gate_mobius over the surviving elements.  Each element is a
  //    pure disjunction of its literal components.
  std::vector<pg_uuid_t> children;
  std::vector<long> coeffs;
  int cancelled = 0;
  bool cancelledHard = false;
  for(const auto &kv : coeff) {
    if(kv.second == 0) {
      ++cancelled;
      // Is this cancelled element #P-hard (no separator, no decomposition)?
      // Probe cheaply: a pure disjunction of these literals with shared
      // symbols and no separator is the hard term whose cancellation makes
      // the query tractable.
      if(!cancelledHard) {
        Sentence el;
        for(int lid : kv.first) el.push_back(literals[lid]);
        std::set<std::pair<unsigned,int>> occ;
        std::vector<std::set<long>> cv;
        // Symbol-connected and no separator => hard.
        if(el.size() >= 2 && !findSeparator(el, occ, cv)) {
          // check symbol-connected
          UF uf(static_cast<int>(el.size()));
          std::map<unsigned,int> firstRelDisj;
          for(std::size_t i=0;i<el.size();++i)
            for(const auto &a : el[i])
              { auto it=firstRelDisj.find(a.rel);
                if(it==firstRelDisj.end()) firstRelDisj[a.rel]=static_cast<int>(i);
                else uf.uni(it->second, static_cast<int>(i)); }
          std::set<int> r; for(std::size_t i=0;i<el.size();++i) r.insert(uf.find(static_cast<int>(i)));
          if(r.size()==1) cancelledHard = true;
        }
      }
      continue;
    }
    Sentence el;
    for(int lid : kv.first) el.push_back(literals[lid]);
    children.push_back(compile(el, false));
    coeffs.push_back(kv.second);
  }
  if(top) { st.n_cancelled = cancelled; st.cancelled_hard = cancelledHard;
            st.n_nonzero = static_cast<int>(children.size()); }

  if(children.empty())
    return mkConst(false);
  // The top-level Möbius step inlines the literal lineage onto its own gate so
  // the root gate_mobius carries it directly (single level); nested steps do
  // not (their values are discarded by the transparent-to-lineage passthrough).
  if(top && !pending_lineage.empty()) {
    lineage_consumed = true;
    return mkMobius(children, coeffs, pending_lineage);
  }
  return mkMobius(children, coeffs);
}

// --- main recursion --------------------------------------------------------

pg_uuid_t MobiusCompiler::compile(const Sentence &sentence0, bool top)
{
  CHECK_FOR_INTERRUPTS();

  // 1. Drop disjuncts containing an ABSENT ground atom (the conjunction is
  //    then FALSE).  Present ground atoms are kept: under TID a tuple in the
  //    relation is a probabilistic Bernoulli event (its input token), NOT a
  //    certainty, so it stays an atom and bottoms out as a token leaf.
  Sentence s;
  for(const Disjunct &d0 : sentence0) {
    bool dead = false;
    for(const MAtom &a : d0) {
      if(ground(a)) {
        std::vector<long> el;
        for(const auto &t : a.args) el.push_back(t.v);
        if(!fi.isPresent(a.rel, el)) { dead=true; break; }
      }
    }
    if(!dead && !d0.empty()) s.push_back(d0);
  }
  if(s.empty())
    return mkConst(false);

  // Base case: a single fully-ground disjunct is a conjunction of tuples ->
  // the independent AND of their input tokens (a certain / untracked tuple
  // contributes the identity gate_one).
  auto isGround = [&](const Disjunct &d){
    for(const auto &a : d) if(!ground(a)) return false;
    return true;
  };
  // A self-join can send two atoms onto the SAME tuple, so the token list is
  // deduplicated (by mkBool) before the AND is built: x ∧ x = x is what the
  // repetition means, whereas an *independent* AND over two copies of one
  // token would square its probability.  Two DISTINCT tuples are independent
  // by the TID restriction (one token per (relation, element tuple), no token
  // shared between two fact slots -- both enforced at gather time).
  auto tokenAnd = [&](const Disjunct &d)->pg_uuid_t {
    std::vector<pg_uuid_t> toks;
    for(const auto &a : d) {
      std::vector<long> el; for(const auto &t : a.args) el.push_back(t.v);
      const std::string *tk = fi.token(a.rel, el);
      if(tk==nullptr) return mkConst(false);     // absent (shouldn't reach here)
      if(tk->empty()) continue;                  // certain fact: identity for AND
      toks.push_back(string2uuid(*tk));
    }
    return mkBool(true, toks);                    // empty -> gate_one (all certain)
  };

  // Memoisation.
  const std::string key = canonSentence(s);
  {
    auto it = memo.find(key);
    if(it!=memo.end()) { ++st.memo_hits; return string2uuid(it->second); }
  }

  pg_uuid_t result;

  // 2. Independent union: group the disjuncts by overlapping fact footprints
  //    (disjoint footprints => disjoint tuples => independent).  On a
  //    reduced-form sentence this is the disjoint-vocabulary split; with
  //    constants around it also separates, say, S(a,y) from S(b,z).
  {
    const int n = static_cast<int>(s.size());
    UF uf(n);
    std::vector<Footprint> fps;
    fps.reserve(n);
    for(const auto &d : s) fps.push_back(footprintOf(d));
    for(int i=0;i<n;++i)
      for(int j=i+1;j<n;++j)
        if(!disjointFootprints(fps[i], fps[j])) uf.uni(i, j);
    std::map<int, Sentence> groups;
    for(int i=0;i<n;++i) groups[uf.find(i)].push_back(s[i]);
    if(groups.size() > 1) {
      std::vector<pg_uuid_t> ch;
      for(auto &kv : groups) ch.push_back(compile(kv.second, false));
      result = mkBool(false, ch);
      memo[key] = uuid2string(result);
      return result;
    }
  }

  // 3. A single group: every disjunct overlaps some other one.
  if(s.size()==1) {
    // Fully ground -> AND of tuple tokens (base case).
    if(isGround(s[0])) {
      result = tokenAnd(s[0]);
      memo[key] = uuid2string(result);
      return result;
    }
    // One disjunct (a CQ): split into variable-connected components and AND
    // them.  Independence of the product needs the components to read disjoint
    // tuples, certified by their fact footprints (disjoint relation symbols, or
    // a position pinned to different constants).  Components sharing a symbol
    // with no separating constant are a within-disjunct self-join: they are NOT
    // independent, so fall through -- the Möbius step below turns the
    // conjunction into the disjunctive detour P(c1)+P(c2)-P(c1∨c2).
    std::vector<Disjunct> comps = componentsOf(s[0]);
    if(comps.size() > 1) {
      std::vector<Footprint> fps;
      fps.reserve(comps.size());
      for(const auto &c : comps) fps.push_back(footprintOf(c));
      bool indep = true;
      for(std::size_t i=0;i<comps.size() && indep;++i)
        for(std::size_t j=i+1;j<comps.size();++j)
          if(!disjointFootprints(fps[i], fps[j])) { indep=false; break; }
      if(indep) {
        std::vector<pg_uuid_t> ch;
        for(auto &c : comps) { Sentence one{c}; ch.push_back(compile(one,false)); }
        result = mkBool(true, ch);
        memo[key] = uuid2string(result);
        return result;
      }
    }
    // A single connected CQ, or components that are not independent: the
    // separator branch below, then the Möbius step.
  }

  // 4. Separator (independent project) if one exists.
  {
    std::set<std::pair<unsigned,int>> occ;
    std::vector<std::set<long>> classVars;
    if(findSeparator(s, occ, classVars)) {
      // Active domain of the separator class.
      std::set<long> dom;
      for(const auto &rp : occ) {
        auto it = fi.domain.find(rp);
        if(it!=fi.domain.end()) dom.insert(it->second.begin(), it->second.end());
      }
      std::vector<pg_uuid_t> ch;
      for(long a : dom) {
        // Substitute every class variable (per disjunct) by the constant a.
        Sentence sub;
        sub.reserve(s.size());
        for(std::size_t di=0; di<s.size(); ++di) {
          Disjunct nd;
          for(const MAtom &at : s[di]) {
            MAtom na = at;
            for(auto &t : na.args)
              if(t.isVar && classVars[di].count(t.v)) { t.isVar=false; t.v=a; }
            nd.push_back(std::move(na));
          }
          sub.push_back(std::move(nd));
        }
        ch.push_back(compile(sub, false));   // pieces for distinct a independent
      }
      result = mkBool(false, ch);            // independent OR over the domain
      memo[key] = uuid2string(result);
      return result;
    }
  }

  // 5. No separator.  A single CONNECTED CQ here is genuinely #P-hard (no
  //    separator, nothing to decompose).  A single disjunct that did decompose
  //    but whose components are not independent (a within-disjunct self-join)
  //    goes to the Möbius step like a multi-disjunct sentence: its CNF is the
  //    unit clauses of its components, and the signed enumeration computes
  //    P(c1) + P(c2) - P(c1 ∨ c2) -- the disjunctive detour a self-join forces
  //    (Dalvi & Suciu, Example 3.1).
  if(s.size()==1 && componentsOf(s[0]).size() < 2)
    throw MobiusDecline("Möbius: non-hierarchical single CQ (no separator); "
                        "the query is #P-hard");
  result = mobiusStep(s, top);
  memo[key] = uuid2string(result);
  return result;
}

/// Compile @p s over @p fi through the shard normalization, returning the root
/// token.  Every entry point goes through here: ranking / shattering is a
/// no-op on an already reduced-form sentence (which is every query the array
/// decoder builds), so the common path pays only the scan that decides so.
pg_uuid_t compileNormalized(const Sentence &s, const FactIndex &fi,
                            MobiusStats &st, const std::string &lineage = "")
{
  Sentence  sn;
  FactIndex fin;
  const bool norm = normalizeShards(s, fi, sn, fin);
  if(norm && provsql_verbose >= 20)
    provsql_notice("ucq_mobius: ranking/shattering: %d disjuncts -> %d: %s",
                   static_cast<int>(s.size()), static_cast<int>(sn.size()),
                   canonSentence(sn).c_str());
  MobiusCompiler mc(norm ? fin : fi, st);
  return mc.compileTop(norm ? sn : s, lineage);
}

// ===========================================================================
// Argument decoding (mirrors ucq_joint's columnar convention).
// ===========================================================================

int checkedLen(ArrayType *a, const char *what) {
  if(a==NULL) return 0;
  if(ARR_NDIM(a)>1) provsql_error("ucq_mobius: %s must be 1-D", what);
  if(ARR_HASNULL(a)) provsql_error("ucq_mobius: %s must not contain NULLs", what);
  return ARR_NDIM(a)==0 ? 0 : ARR_DIMS(a)[0];
}
const int32 *intArr(FunctionCallInfo fcinfo, int n, const char *what, int &len){
  ArrayType *a = PG_ARGISNULL(n)?NULL:PG_GETARG_ARRAYTYPE_P(n);
  len = checkedLen(a, what);
  return (a==NULL||len==0)?NULL:(const int32*)ARR_DATA_PTR(a);
}

/// Build the top sentence from the raw query arrays.  Disjunct-local variables
/// are globalised by a per-disjunct offset (returned via @p base) so
/// unification across disjuncts is by (rel,pos), not by raw id, and so head
/// variables can be pinned per disjunct.
Sentence buildSentenceArrays(const int32 *d_nvars, int n_disj,
                             const int32 *a_disj, const int32 *a_rel,
                             const int32 *a_vars, int n_av,
                             const int32 *a_arity, int n_ad,
                             std::vector<long> &base)
{
  if(n_disj==0) provsql_error("ucq_mobius: the UCQ has no disjuncts");
  base.assign(n_disj, 0);
  long acc = 1;   // start at 1 so id 0 is unused (avoids any sentinel clash)
  for(int d=0; d<n_disj; ++d){ base[d]=acc; acc += d_nvars[d] + 1; }

  Sentence s(n_disj);
  int voff=0;
  for(int i=0;i<n_ad;++i){
    const int d=a_disj[i];
    if(d<0||d>=n_disj) provsql_error("ucq_mobius: atom disjunct out of range");
    const int ar=a_arity[i];
    if(ar<0||voff+ar>n_av)
      provsql_error("ucq_mobius: atom_vars shorter than arities");
    MAtom at; at.rel=static_cast<unsigned>(a_rel[i]);
    for(int k=0;k<ar;++k){
      Term t; t.isVar=true; t.v = base[d] + a_vars[voff+k];
      at.args.push_back(t);
    }
    voff+=ar;
    s[d].push_back(std::move(at));
  }
  return s;
}

/// Build a FactIndex (and, optionally, the text->dense-id map) from the raw
/// fact arrays.
FactIndex buildFactIndexArrays(const int32 *f_rel, int n_fr,
                               const int32 *f_elems, int n_fe,
                               const int32 *f_arity,
                               const pg_uuid_t *tok)
{
  // G3 (tuple independence): every present fact must be gated by a bare
  // gate_input.  A fact whose token is an internal gate (a view-derived /
  // reachability / repair_key lineage) is correlated -- out of scope for the
  // lifted-inference recursion, which assumes independence -- so decline and
  // let the caller fall back (the more general joint-width route already had
  // its chance).  The input-gate enum OID, fetched once.
  const Oid input_oid = get_constants(true).GATE_TYPE_TO_OID[gate_input];

  FactIndex fi;
  // Self-join overlap guard: which (rel, element) key first carried each token.
  // The same base tuple feeding two DISTINCT fact slots (a self-join whose
  // constant prefilters are not disjoint, so the slots share rows) is a
  // correlation the independence-assuming recursion cannot represent.
  std::map<std::string, std::pair<unsigned, std::vector<long>>> token_owner;
  int eoff=0;
  for(int i=0;i<n_fr;++i){
    const int ar=f_arity[i];
    if(ar<0||eoff+ar>n_fe)
      provsql_error("ucq_mobius: fact_elems shorter than arities");
    unsigned rel=static_cast<unsigned>(f_rel[i]);
    std::vector<long> el;
    for(int k=0;k<ar;++k){
      long e=static_cast<long>(f_elems[eoff+k]);
      el.push_back(e);
      fi.domain[{rel,k}].insert(e);
    }
    eoff+=ar;
    fi.present.insert({rel, el});
    bool nil=true;
    for(int b=0;b<16;++b) if(tok[i].data[b]!=0) nil=false;
    if(!nil) {
      pg_uuid_t tk = tok[i];
      const Datum gt = DirectFunctionCall1(get_gate_type, UUIDPGetDatum(&tk));
      if(static_cast<Oid>(DatumGetInt32(gt)) != input_oid)
        throw MobiusDecline("non-TID input: a fact token is not a bare "
                            "gate_input (correlated / derived lineage)");
    }
    // Bag multiplicity guard: the fact index is keyed by (rel, element tuple),
    // one token per key (the SJF-TID reduced form).  Two DISTINCT probabilistic
    // tuples that project to the same element tuple -- e.g. a relation column
    // not bound by the query, or a duplicate-bearing input -- would silently
    // collapse to one event (lose the OR), under-counting the probability.  The
    // lifted-inference recursion has no place to represent that disjunction, so
    // decline and let the joint-width route (which sees full multiplicity) take
    // it.  (A repeated key with the SAME token is the same tuple gathered twice
    // -- harmless, kept.)
    const std::string newtok = nil ? std::string() : uuid2string(tok[i]);
    auto existing = fi.tok.find({rel, el});
    if(existing != fi.tok.end() && existing->second != newtok)
      throw MobiusDecline("bag multiplicity: two distinct tuples share an "
                          "element tuple (non-reduced data) -- defer to "
                          "joint-width, which keeps the multiplicity");
    // The same token under a DIFFERENT (rel, element) key means one base tuple
    // feeds two fact slots -- an overlapping (non-disjoint) self-join, a
    // correlation the lifted recursion would wrongly treat as independent.
    // Decline (joint-width tracks the shared token correctly).  A self-join
    // WITHIN the query is not this case and passes: its atoms share one slot,
    // and ranking / shattering plus the Möbius step handle them.  What this
    // catches is two relation legs of the descriptor (one table scanned twice
    // under overlapping prefilters), which no shard split can separate -- the
    // shards partition a leg, not the legs themselves.
    if(!newtok.empty()) {
      auto own = token_owner.find(newtok);
      if(own != token_owner.end() && own->second != std::make_pair(rel, el))
        throw MobiusDecline("self-join overlap: one tuple feeds two fact slots "
                            "(non-disjoint self-join) -- defer to joint-width");
      token_owner[newtok] = {rel, el};
    }
    fi.tok[{rel, el}] = newtok;
  }
  return fi;
}

/// Decode the query arrays (0..4) into the top sentence (Boolean path).
Sentence decodeQuery(FunctionCallInfo fcinfo)
{
  int n_disj,n_ad,n_ar,n_av,n_aa;
  const int32 *d_nvars = intArr(fcinfo,0,"disjunct_nvars",n_disj);
  const int32 *a_disj  = intArr(fcinfo,1,"atom_disjunct",n_ad);
  const int32 *a_rel   = intArr(fcinfo,2,"atom_rel",n_ar);
  const int32 *a_vars  = intArr(fcinfo,3,"atom_vars",n_av);
  const int32 *a_arity = intArr(fcinfo,4,"atom_arity",n_aa);
  if(n_ad!=n_ar || n_ad!=n_aa)
    provsql_error("ucq_mobius: atom arrays must have the same length");
  std::vector<long> base;
  return buildSentenceArrays(d_nvars,n_disj,a_disj,a_rel,a_vars,n_av,
                             a_arity,n_ad,base);
}

/// Decode the fact arrays (5..8) into a FactIndex (Boolean path).
FactIndex decodeFacts(FunctionCallInfo fcinfo)
{
  int n_fr,n_fe,n_fa,n_ft;
  const int32 *f_rel   = intArr(fcinfo,5,"fact_rel",n_fr);
  const int32 *f_elems = intArr(fcinfo,6,"fact_elems",n_fe);
  const int32 *f_arity = intArr(fcinfo,7,"fact_arity",n_fa);
  ArrayType *toks = PG_ARGISNULL(8)?NULL:PG_GETARG_ARRAYTYPE_P(8);
  n_ft = checkedLen(toks,"fact_tokens");
  if(n_fr!=n_fa || n_fr!=n_ft)
    provsql_error("ucq_mobius: fact arrays must have the same length");
  const pg_uuid_t *tok = (toks&&n_ft)?(const pg_uuid_t*)ARR_DATA_PTR(toks):NULL;
  return buildFactIndexArrays(f_rel,n_fr,f_elems,n_fe,f_arity,tok);
}

// ===========================================================================
// Per-answer (free head variables): head-pin then compile, one circuit per
// output group.  On the first call of a query the facts are gathered once
// (ucq_joint_gather) and the value dictionary cached; each group binds its
// head variables to their values, the head positions are substituted to
// constants across every disjunct, and the Möbius circuit is compiled and
// cached.  Mirrors ucq_joint_provenance_answer (the per-group caching), but
// each answer is a separate compile rather than a single sweep.
// ===========================================================================

struct MobAnswerCache {
  bool ready = false;                 ///< gather succeeded
  Sentence sentence;                  ///< the UCQ template (global vars)
  std::vector<long> base;             ///< per-disjunct variable offset
  std::vector<int>  d_nvars;          ///< per-disjunct n_vars
  FactIndex fi;                       ///< the gathered facts
  std::map<std::string,long> val_to_id;   ///< text value -> dense id
  std::map<std::string,std::string> tokcache;  ///< head-key -> token uuid
};

void mobAnswerCacheDelete(void *arg) { delete reinterpret_cast<MobAnswerCache*>(arg); }

std::string mobHeadKey(const std::vector<std::string> &vals)
{
  std::string k;
  for(const auto &v : vals){ k += v; k.push_back('\x1f'); }
  return k;
}

/// Gather the facts + value dictionary once (via ucq_joint_gather) into @p c.
/// Returns false on any failure (the caller then declines to the fallback).
bool mobGather(Datum descriptor, MobAnswerCache *c)
{
  SPI_connect();
  Oid argt[1] = { JSONBOID };
  Datum argv[1] = { descriptor };
  char argn[1] = { ' ' };
  const int rc = SPI_execute_with_args(
    "SELECT * FROM provsql.ucq_joint_gather($1)", 1, argt, argv, argn, true, 1);
  if(rc != SPI_OK_SELECT || SPI_processed != 1) { SPI_finish(); return false; }

  bool ok = true;
  try {
    TupleDesc td = SPI_tuptable->tupdesc;
    HeapTuple row = SPI_tuptable->vals[0];
    bool isnull;
    auto ia = [&](int col, int &n)->const int32*{
      Datum d = SPI_getbinval(row, td, col, &isnull);
      if(isnull){ n=0; return nullptr; }
      ArrayType *a = DatumGetArrayTypeP(d);
      n = ArrayGetNItems(ARR_NDIM(a), ARR_DIMS(a));
      return (const int32*) ARR_DATA_PTR(a);
    };
    int n_dnv,n_adisj,n_arel,n_avars,n_aarity,n_frel,n_felems,n_farity;
    const int32 *dnv    = ia(1,n_dnv);
    const int32 *adisj  = ia(2,n_adisj);
    const int32 *arel   = ia(3,n_arel);
    const int32 *avars  = ia(4,n_avars);
    const int32 *aarity = ia(5,n_aarity);
    const int32 *frel   = ia(6,n_frel);
    const int32 *felems = ia(7,n_felems);
    const int32 *farity = ia(8,n_farity);
    Datum dtok = SPI_getbinval(row, td, 9, &isnull);
    ArrayType *atok = isnull ? nullptr : DatumGetArrayTypeP(dtok);
    const int n_ftok = atok ? ArrayGetNItems(ARR_NDIM(atok), ARR_DIMS(atok)) : 0;
    const pg_uuid_t *ftok = atok ? (const pg_uuid_t*) ARR_DATA_PTR(atok) : nullptr;
    if(n_frel != n_farity || n_frel != n_ftok)
      throw MobiusDecline("ucq_mobius: fact arrays length mismatch");

    c->sentence = buildSentenceArrays(dnv,n_dnv,adisj,arel,avars,n_avars,
                                      aarity,n_adisj,c->base);
    c->d_nvars.assign(dnv, dnv+n_dnv);
    c->fi = buildFactIndexArrays(frel,n_frel,felems,n_felems,farity,ftok);

    // The value dictionary: dense id -> text, inverted to text -> id.
    Datum dval = SPI_getbinval(row, td, 10, &isnull);
    if(!isnull) {
      ArrayType *aval = DatumGetArrayTypeP(dval);
      Datum *elems; bool *nulls; int nval;
      deconstruct_array(aval, TEXTOID, -1, false, TYPALIGN_INT,
                        &elems, &nulls, &nval);
      for(int i=0;i<nval;++i)
        if(!nulls[i])
          c->val_to_id[TextDatumGetCString(elems[i])] = i;
    }
  } catch(...) {
    ok = false;
  }
  SPI_finish();
  return ok;
}

}  // namespace

/**
 * @brief Materialise the safe-UCQ Möbius circuit and return its root token.
 *
 * Columnar arguments (mirrors @c ucq_joint_materialize_tracked, TID inputs):
 *   0..4 disjunct_nvars, atom_disjunct, atom_rel, atom_vars, atom_arity
 *   5..8 fact_rel, fact_elems, fact_arity, fact_tokens
 *   9    lineage (uuid, optional): the literal Boolean provenance of the query,
 *        carried on the root gate_mobius so the token still answers Shapley /
 *        semiring / PROV on the normal lineage (the Möbius combination is a
 *        probability-only shortcut layered over it).
 *
 * The root is a @c gate_mobius carrying the certified-independent Boolean
 * islands (the signed combination) and the lineage child; @c probability_evaluate
 * answers the \#P-hard UCQ in PTIME through the fast Möbius route, and every other
 * evaluator passes through to the lineage.  Declines raise an error so the SQL
 * wrapper falls back.
 */
Datum ucq_mobius_materialize_tracked(PG_FUNCTION_ARGS)
{
  try {
    Sentence s = decodeQuery(fcinfo);
    FactIndex fi = decodeFacts(fcinfo);
    std::string lineage;
    if(!PG_ARGISNULL(9))
      lineage = uuid2string(*PG_GETARG_UUID_P(9));
    MobiusStats st;
    pg_uuid_t root = compileNormalized(s, fi, st, lineage);
    pg_uuid_t *u = (pg_uuid_t*) palloc(sizeof(pg_uuid_t));
    *u = root;
    PG_RETURN_UUID_P(u);
  } catch(const MobiusDecline &e) {
    provsql_error("ucq_mobius: %s", e.what());
  } catch(const std::exception &e) {
    provsql_error("ucq_mobius: %s", e.what());
  } catch(...) {
    provsql_error("ucq_mobius: unknown exception");
  }
  PG_RETURN_NULL();
}

/**
 * @brief Compile the Möbius circuit and return the lattice statistics plus the
 *        probability (the demonstrability surface).  Same columnar arguments
 *        as @c ucq_mobius_materialize_tracked.
 */
Datum ucq_mobius_compile_stats(PG_FUNCTION_ARGS)
{
  try {
    Sentence s = decodeQuery(fcinfo);
    FactIndex fi = decodeFacts(fcinfo);
    MobiusStats st;
    pg_uuid_t root = compileNormalized(s, fi, st);
    st.probability = mobius_probability_of(root);

    TupleDesc tupdesc;
    if(get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
      provsql_error("ucq_mobius_compile_stats: expected composite return type");
    tupdesc = BlessTupleDesc(tupdesc);
    Datum values[9];
    bool nulls[9] = {false,false,false,false,false,false,false,false,false};
    values[0] = Float8GetDatum(st.probability);
    values[1] = Int32GetDatum(st.n_components);
    values[2] = Int32GetDatum(st.n_cnf_conjuncts);
    values[3] = Int32GetDatum(st.lattice_collapsed);
    values[4] = Int32GetDatum(st.n_nonzero);
    values[5] = Int32GetDatum(st.n_cancelled);
    values[6] = BoolGetDatum(st.cancelled_hard);
    values[7] = Int64GetDatum(static_cast<int64>(st.dd_size));
    values[8] = Int64GetDatum(static_cast<int64>(st.memo_hits));
    PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
  } catch(const std::exception &e) {
    provsql_error("ucq_mobius_compile_stats: %s", e.what());
  } catch(...) {
    provsql_error("ucq_mobius_compile_stats: unknown exception");
  }
  PG_RETURN_NULL();
}

/**
 * @brief Per-answer Möbius provenance (the planner-substituted entry point for
 *        a non-Boolean UCQ with free head variables).
 *
 * Arguments: (descriptor jsonb, head_vars int[], head_vals text[],
 * fallback uuid).  Called once per output group; on the first call the facts
 * are gathered once and the value dictionary cached, then each group pins its
 * head variables (the canonical head indices @p head_vars, in every disjunct)
 * to their values (@p head_vals, matched through the gather's text dictionary)
 * and compiles the head-pinned Möbius circuit, caching head-key -> token.  On
 * any decline (unsafe shape, head value absent, ...) returns @p fallback.
 */
Datum ucq_mobius_provenance_answer(PG_FUNCTION_ARGS)
{
  MobAnswerCache *cache =
    reinterpret_cast<MobAnswerCache*>(fcinfo->flinfo->fn_extra);

  if(cache == nullptr) {
    MemoryContext fnctx = fcinfo->flinfo->fn_mcxt;
    cache = new MobAnswerCache();
    MemoryContextCallback *cb = (MemoryContextCallback*)
      MemoryContextAllocZero(fnctx, sizeof(MemoryContextCallback));
    cb->func = mobAnswerCacheDelete;
    cb->arg = cache;
    MemoryContextRegisterResetCallback(fnctx, cb);
    fcinfo->flinfo->fn_extra = cache;

    if(!PG_ARGISNULL(0)) {
      // Gather inside a subtransaction so a SQL error declines gracefully.
      MemoryContext oldcxt = CurrentMemoryContext;
      ResourceOwner oldowner = CurrentResourceOwner;
      BeginInternalSubTransaction(NULL);
      PG_TRY();
      {
        cache->ready = mobGather(PG_GETARG_DATUM(0), cache);
        ReleaseCurrentSubTransaction();
        MemoryContextSwitchTo(oldcxt);
        CurrentResourceOwner = oldowner;
      }
      PG_CATCH();
      {
        MemoryContextSwitchTo(oldcxt);
        RollbackAndReleaseCurrentSubTransaction();
        MemoryContextSwitchTo(oldcxt);
        CurrentResourceOwner = oldowner;
        FlushErrorState();
        cache->ready = false;
      }
      PG_END_TRY();
    }
  }

  // The group's head values (text), and the head variable indices.
  if(cache->ready && !PG_ARGISNULL(1) && !PG_ARGISNULL(2)) {
    std::vector<int> head_vars;
    {
      ArrayType *a = PG_GETARG_ARRAYTYPE_P(1);
      const int32 *d = (const int32*) ARR_DATA_PTR(a);
      const int n = ArrayGetNItems(ARR_NDIM(a), ARR_DIMS(a));
      for(int i=0;i<n;++i) head_vars.push_back(d[i]);
    }
    std::vector<std::string> head_vals;
    {
      ArrayType *a = PG_GETARG_ARRAYTYPE_P(2);
      Datum *elems; bool *nulls; int n;
      deconstruct_array(a, TEXTOID, -1, false, TYPALIGN_INT, &elems, &nulls, &n);
      for(int i=0;i<n;++i)
        head_vals.push_back(nulls[i] ? std::string() : TextDatumGetCString(elems[i]));
    }

    if(head_vars.size() == head_vals.size()) {
      const std::string key = mobHeadKey(head_vals);
      auto it = cache->tokcache.find(key);
      if(it != cache->tokcache.end()) {
        pg_uuid_t *u = (pg_uuid_t*) palloc(sizeof(pg_uuid_t));
        *u = string2uuid(it->second);
        PG_RETURN_UUID_P(u);
      }
      try {
        // Map each head value to its dense id; pin the head variable (canonical
        // index hv, hence global base[d]+hv in every disjunct) to that constant.
        Sentence s = cache->sentence;
        bool resolved = true;
        for(std::size_t h=0; h<head_vars.size() && resolved; ++h) {
          auto vit = cache->val_to_id.find(head_vals[h]);
          if(vit == cache->val_to_id.end()) { resolved = false; break; }
          const long val = vit->second;
          const int hv = head_vars[h];
          for(std::size_t d=0; d<s.size(); ++d) {
            const long gid = cache->base[d] + hv;
            for(MAtom &at : s[d])
              for(Term &t : at.args)
                if(t.isVar && t.v == gid) { t.isVar=false; t.v=val; }
          }
        }
        if(resolved) {
          // The per-group literal lineage (the normal per-answer provenance,
          // argument 3) is carried on the gate_mobius so this answer's token
          // still answers Shapley / semiring on its normal lineage.
          std::string lineage;
          if(!PG_ARGISNULL(3))
            lineage = uuid2string(*PG_GETARG_UUID_P(3));
          MobiusStats st;
          pg_uuid_t root = compileNormalized(s, cache->fi, st, lineage);
          cache->tokcache[key] = uuid2string(root);
          pg_uuid_t *u = (pg_uuid_t*) palloc(sizeof(pg_uuid_t));
          *u = root;
          PG_RETURN_UUID_P(u);
        }
      } catch(const std::exception &e) {
        // decline this group -> fallback (reported at verbose_level >= 5, the
        // only place a decline is otherwise invisible)
        if(provsql_verbose >= 5)
          provsql_notice("ucq_mobius: per-answer decline: %s", e.what());
      } catch(...) {
        // decline this group -> fallback
      }
    }
  }

  if(PG_ARGISNULL(3))
    PG_RETURN_NULL();
  PG_RETURN_DATUM(PG_GETARG_DATUM(3));
}