package gudusoft.gsqlparser.demos.sqlguard;

import gudusoft.gsqlparser.EAggregateType;
import gudusoft.gsqlparser.EDbVendor;
import gudusoft.gsqlparser.EResolverType;
import gudusoft.gsqlparser.TCustomSqlStatement;
import gudusoft.gsqlparser.TGSqlParser;
import gudusoft.gsqlparser.catalog.diagnostic.CatalogDiagnostic;
import gudusoft.gsqlparser.catalog.diagnostic.CatalogDiagnosticSeverity;
import gudusoft.gsqlparser.catalog.input.CatalogInputException;
import gudusoft.gsqlparser.catalog.input.CatalogInputKind;
import gudusoft.gsqlparser.catalog.input.CatalogInputSource;
import gudusoft.gsqlparser.catalog.input.CatalogLoadOptions;
import gudusoft.gsqlparser.catalog.input.CatalogModelValidator;
import gudusoft.gsqlparser.catalog.input.CatalogValidationResult;
import gudusoft.gsqlparser.catalog.input.model.ColumnModel;
import gudusoft.gsqlparser.catalog.input.model.ConstraintModel;
import gudusoft.gsqlparser.catalog.input.model.TableModel;
import gudusoft.gsqlparser.catalog.input.model.UnifiedCatalogModel;
import gudusoft.gsqlparser.catalog.input.readers.JsonManifestCatalogInputReader;
import gudusoft.gsqlparser.dlineage.DataFlowAnalyzer;
import gudusoft.gsqlparser.dlineage.dataflow.model.Option;
import gudusoft.gsqlparser.dlineage.dataflow.model.xml.column;
import gudusoft.gsqlparser.dlineage.dataflow.model.xml.dataflow;
import gudusoft.gsqlparser.dlineage.dataflow.model.xml.relationship;
import gudusoft.gsqlparser.dlineage.dataflow.model.xml.sourceColumn;
import gudusoft.gsqlparser.dlineage.dataflow.model.xml.table;
import gudusoft.gsqlparser.ESortType;
import gudusoft.gsqlparser.nodes.TExpression;
import gudusoft.gsqlparser.nodes.TExpressionList;
import gudusoft.gsqlparser.nodes.TFunctionCall;
import gudusoft.gsqlparser.nodes.TGroupBy;
import gudusoft.gsqlparser.nodes.TGroupByItem;
import gudusoft.gsqlparser.nodes.TCTE;
import gudusoft.gsqlparser.nodes.TCTEList;
import gudusoft.gsqlparser.nodes.TObjectName;
import gudusoft.gsqlparser.nodes.TOrderBy;
import gudusoft.gsqlparser.nodes.TOrderByItem;
import gudusoft.gsqlparser.nodes.TParseTreeVisitor;
import gudusoft.gsqlparser.nodes.TPartitionClause;
import gudusoft.gsqlparser.nodes.TResultColumn;
import gudusoft.gsqlparser.nodes.TResultColumnList;
import gudusoft.gsqlparser.nodes.TTable;
import gudusoft.gsqlparser.nodes.TWindowDef;
import gudusoft.gsqlparser.nodes.TWindowFrame;
import gudusoft.gsqlparser.nodes.TWindowSpecification;
import gudusoft.gsqlparser.resolver2.TSQLResolverConfig;
import gudusoft.gsqlparser.resolver2.binding.BindingDiagnostic;
import gudusoft.gsqlparser.resolver2.binding.BindingDiagnosticSeverity;
import gudusoft.gsqlparser.resolver2.binding.BindingResult;
import gudusoft.gsqlparser.sqlenv.ESQLDataObjectType;
import gudusoft.gsqlparser.sqlenv.IdentifierService;
import gudusoft.gsqlparser.stmt.TSelectSqlStatement;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;

public class SqlGuardService {
    public static final int MAX_SQL_BYTES = 512 * 1024;
    public static final int MAX_CATALOG_BYTES = 512 * 1024;
    private static final int MAX_STATEMENTS = 50;

    public SqlGuardResponse check(SqlGuardRequest req) {
        if (req == null) {
            return SqlGuardResponse.error(null, "INVALID_REQUEST", "Invalid request.");
        }
        SqlGuardResponse baseErr = validate(req);
        if (baseErr != null) {
            return baseErr;
        }

        SqlGuardResponse r = new SqlGuardResponse();
        r.requestId = req.requestId;
        r.facts.dialect = req.dialect;
        audit(r, req);

        EDbVendor vendor = EDbVendor.fromAlias(req.dialect);
        UnifiedCatalogModel catalog;
        if ("inline".equalsIgnoreCase(req.catalogMode)) {
            // Pass the request vendor to BOTH the reader and the validator
            // so {@link CatalogModelValidator} can enforce
            // {@code options.vendor() == model.vendor()}. Without options,
            // a request with {@code dialect="postgresql"} would happily
            // pass an Oracle-shaped manifest, then bind columns under PG
            // identifier folding while the catalog was authored under
            // Oracle folding — silent governance miss.
            CatalogLoadOptions opts = CatalogLoadOptions.builder().vendor(vendor).build();
            try {
                CatalogInputSource src = CatalogInputSource.fromBytes(
                        req.catalog.getBytes(StandardCharsets.UTF_8),
                        CatalogInputKind.JSON_MANIFEST);
                catalog = new JsonManifestCatalogInputReader().read(src, opts);
            } catch (CatalogInputException ex) {
                return SqlGuardResponse.error(req.requestId, "INVALID_CATALOG_JSON",
                        "Inline catalog JSON could not be parsed: "
                                + sanitizeParserMessage(ex.getMessage()));
            }
            CatalogValidationResult vr = new CatalogModelValidator().validate(catalog, opts);
            if (!vr.ok()) {
                return SqlGuardResponse.error(req.requestId, "INVALID_CATALOG_JSON",
                        "Inline catalog failed validation: "
                                + firstCatalogErrorMessage(vr));
            }
        } else {
            catalog = SqlGuardCatalogSamples.sample(req.sampleCatalogId);
        }
        TGSqlParser parser = new TGSqlParser(vendor);
        parser.setResolverType(EResolverType.RESOLVER2);
        parser.setSqlEnv(new SqlGuardSampleEnv(vendor, catalog));
        TSQLResolverConfig cfg = parser.getResolver2Config();
        if (cfg == null) {
            cfg = new TSQLResolverConfig();
        }
        cfg.setEmitBindingDiagnostics(true);
        cfg.setBindingStrictCatalogValidation(true);
        parser.setResolver2Config(cfg);
        parser.sqltext = req.sql;

        int rc;
        try {
            rc = parser.parse();
        } catch (Exception e) {
            return parseError(r, parser);
        }
        if (rc != 0 || parser.sqlstatements == null || parser.sqlstatements.size() == 0) {
            return parseError(r, parser);
        }
        if (parser.sqlstatements.size() > MAX_STATEMENTS) {
            return SqlGuardResponse.error(req.requestId, "TOO_MANY_STATEMENTS", "Too many SQL statements.");
        }
        SqlGuardResponse bindingErr = bindingError(parser, r);
        if (bindingErr != null) {
            return bindingErr;
        }
        SqlGuardResponse semanticErr = validateSemantics(parser, r, catalog);
        if (semanticErr != null) {
            return semanticErr;
        }

        for (int i = 0; i < parser.sqlstatements.size(); i++) {
            analyzeStatement(parser.sqlstatements.get(i), i, req, r, catalog, vendor);
        }
        merge(r);
        // PRD §11 four-decision model — promote WARN to APPROVAL_REQUIRED
        // when any catalog-bound column carries the RESTRICTED policy tag
        // AND is exposed by the SQL (projection / SELECT * / window slot).
        // DENY-class findings always win; ALLOW stays ALLOW (no
        // restricted-column access at all).
        maybeRequireApproval(r);
        return r;
    }

    /**
     * Upgrade {@code r.decision} from "WARN" to "APPROVAL_REQUIRED" when
     * any {@link gudusoft.gsqlparser.demos.sqlguard.SqlGuardResponse.ColumnFact}
     * carries a RESTRICTED policy tag and is being exposed by the SQL.
     * Implements PRD §11 ("approval_required" — the fourth decision in
     * the live preview's decision model) without introducing a new
     * severity level at rule-evaluation time; the rules continue to fire
     * as {@code warn}, and this post-step decides whether the aggregate
     * decision is operationally "this is a warn the human MUST review"
     * versus "this is a warn you can clear yourself".
     *
     * <p>Why scan {@code facts.columns} instead of {@code violations}:
     * column facts already carry the resolved catalog tags. A
     * RESTRICTED-tagged column that triggered SENSITIVE_COLUMN_IN_OUTPUT
     * is identical at the column-fact layer to one that triggered
     * SELECT_STAR_ON_SENSITIVE_TABLE — both are "RESTRICTED is in the
     * output". Detecting at the fact layer means we don't have to teach
     * every individual rule about approval semantics.
     */
    private void maybeRequireApproval(SqlGuardResponse r) {
        if (!"WARN".equals(r.decision)) {
            return;
        }
        for (SqlGuardResponse.ColumnFact cf : r.facts.columns) {
            if (cf.policyTags == null || cf.policyTags.isEmpty()) continue;
            if (cf.exposure == null) continue;
            // Only output-bearing exposures count — predicate-only access
            // (filter / join) doesn't surface restricted values to the
            // caller, so it stays plain WARN.
            String e = cf.exposure;
            boolean outputBearing = "explicit_select".equals(e)
                    || "implicit_select_star".equals(e)
                    || "window_partition".equals(e)
                    || "window_order".equals(e);
            if (!outputBearing) continue;
            for (String t : cf.policyTags) {
                if ("RESTRICTED".equalsIgnoreCase(t)) {
                    r.decision = "APPROVAL_REQUIRED";
                    r.summary = "Restricted columns are exposed; human approval required before this SQL runs.";
                    r.suggestions.add("Remove the restricted column(s) from the projection, or route this query through your approval workflow before execution.");
                    return;
                }
            }
        }
    }

    private SqlGuardResponse validate(SqlGuardRequest req) {
        if (req.sql == null || req.sql.trim().length() == 0) {
            return SqlGuardResponse.error(req.requestId, "INVALID_REQUEST", "SQL is required.");
        }
        if (req.sql.getBytes(StandardCharsets.UTF_8).length > MAX_SQL_BYTES) {
            return SqlGuardResponse.error(req.requestId, "SQL_TOO_LARGE", "SQL exceeds 512 KB limit.");
        }
        if (req.catalog != null && req.catalog.getBytes(StandardCharsets.UTF_8).length > MAX_CATALOG_BYTES) {
            return SqlGuardResponse.error(req.requestId, "CATALOG_TOO_LARGE", "Catalog exceeds 512 KB limit.");
        }
        EDbVendor requestedVendor = EDbVendor.fromAlias(req.dialect);
        if (requestedVendor == null || !requestedVendor.isImplemented()) {
            return SqlGuardResponse.error(req.requestId, "UNSUPPORTED_DIALECT", "Unsupported SQL dialect.");
        }
        if ("sample".equalsIgnoreCase(req.catalogMode)) {
            if (!SqlGuardCatalogSamples.isSupportedSample(req.sampleCatalogId)) {
                return SqlGuardResponse.error(req.requestId, "UNSUPPORTED_SAMPLE_CATALOG", "Unsupported sample catalog.");
            }
        } else if ("inline".equalsIgnoreCase(req.catalogMode)) {
            if (req.catalog == null || req.catalog.trim().length() == 0) {
                return SqlGuardResponse.error(req.requestId, "INVALID_REQUEST",
                        "Inline catalog is required when catalogMode=inline.");
            }
        } else {
            return SqlGuardResponse.error(req.requestId, "UNSUPPORTED_CATALOG_MODE", "Unsupported catalog mode.");
        }
        if (!"all".equalsIgnoreCase(req.policyPreset)
                && !"pii-basic".equalsIgnoreCase(req.policyPreset)
                && !"destructive-sql-block".equalsIgnoreCase(req.policyPreset)) {
            return SqlGuardResponse.error(req.requestId, "UNSUPPORTED_POLICY_PRESET", "Unsupported policy preset.");
        }
        return null;
    }

    private SqlGuardResponse parseError(SqlGuardResponse r, TGSqlParser parser) {
        r.ok = false;
        r.error = new SqlGuardResponse.ErrorInfo();
        r.error.code = "SQL_PARSE_ERROR";
        String details = "";
        try {
            details = sanitizeParserMessage(parser == null ? null : parser.getErrormessage());
        } catch (Exception ignored) {
            details = "";
        }
        r.error.message = details.length() == 0
                ? "SQL syntax error: SQL could not be parsed."
                : "SQL syntax error: " + details;
        return r;
    }

    private SqlGuardResponse validateSemantics(TGSqlParser parser, SqlGuardResponse r, UnifiedCatalogModel cat) {
        for (int i = 0; i < parser.sqlstatements.size(); i++) {
            TCustomSqlStatement stmt = parser.sqlstatements.get(i);
            if (stmt instanceof TSelectSqlStatement) {
                SqlGuardResponse err = validateSelectSemantics((TSelectSqlStatement) stmt, r, cat);
                if (err != null) {
                    return err;
                }
            }
        }
        return null;
    }

    private SqlGuardResponse validateSelectSemantics(TSelectSqlStatement sel, SqlGuardResponse r, UnifiedCatalogModel cat) {
        TGroupBy groupBy = sel.getGroupByClause();
        if (groupBy == null || groupBy.isAllModifier() || groupBy.getItems() == null || groupBy.getItems().size() == 0) {
            return null;
        }
        LinkedHashSet<String> grouped = new LinkedHashSet<String>();
        for (int i = 0; i < groupBy.getItems().size(); i++) {
            TGroupByItem item = groupBy.getItems().getGroupByItem(i);
            if (item != null && item.getExpr() != null) {
                grouped.add(normalizeSemanticExpression(item.getExpr().toString()));
            }
        }
        TResultColumnList cols = sel.getResultColumnList();
        if (cols == null) {
            return null;
        }
        for (int i = 0; i < cols.size(); i++) {
            TResultColumn rc = cols.getResultColumn(i);
            String expr = resultExpressionText(rc);
            if (expr.length() == 0 || "*".equals(expr) || expr.endsWith(".*") || isAggregateColumn(rc)) {
                continue;
            }
            if (!grouped.contains(normalizeSemanticExpression(expr))) {
                return semanticError(r, "Column/expression " + expr
                        + " appears in SELECT but is not aggregated and is not listed in GROUP BY.");
            }
        }
        return null;
    }

    /**
     * Convert the first {@link BindingDiagnosticSeverity#ERROR}-level binding
     * diagnostic produced by resolver2 into a {@code SQL_SEMANTIC_ERROR}
     * envelope. Returns {@code null} when the SQL binds cleanly so the caller
     * proceeds to policy / lineage analysis. Replaces the regex-based
     * qualified-column scanner retired in S18 (plan §7.3 S18).
     */
    private SqlGuardResponse bindingError(TGSqlParser parser, SqlGuardResponse r) {
        BindingResult br = parser.getBindingResult();
        if (br == null || !br.hasErrors()) {
            return null;
        }
        for (BindingDiagnostic d : br.getDiagnostics()) {
            if (d.severity() == BindingDiagnosticSeverity.ERROR) {
                return semanticError(r, d.message());
            }
        }
        return null;
    }

    /**
     * Shared with {@link #aggregateFunctionsFromAst} so the GROUP BY
     * semantic check and the lineage code path agree on what counts as
     * an aggregate. Both routes walk the same AST visitor — there is no
     * lexical / textual path that could desync them.
     */
    private boolean isAggregateColumn(TResultColumn rc) {
        List<String> fns = aggregateFunctionsFromAst(rc);
        return fns != null && !fns.isEmpty();
    }

    private String normalizeSemanticExpression(String expr) {
        if (expr == null) {
            return "";
        }
        String s = expr.toLowerCase(Locale.ROOT).replace("\"", "").replace("`", "").trim();
        s = s.replaceAll("\\s+", "");
        return s;
    }

    private SqlGuardResponse semanticError(SqlGuardResponse r, String message) {
        r.ok = false;
        r.error = new SqlGuardResponse.ErrorInfo();
        r.error.code = "SQL_SEMANTIC_ERROR";
        r.error.message = "SQL semantic error: " + message;
        return r;
    }

    /**
     * Scrubs an error message for envelope use: collapses runs of
     * whitespace and truncates at 500 chars. Every C0 control
     * character ({@code 0x00..0x1F}) and DEL ({@code 0x7F}) becomes a
     * space — not just CR/LF — because
     * {@code gudusoft.gsqlparser.util.json.JSONSerializer} only
     * escapes the named-escape subset
     * ({@code "}, {@code \\}, {@code /}, {@code \b}, {@code \f},
     * {@code \n}, {@code \r}, {@code \t}). Unescaped controls
     * survive into the wire envelope and can poison BFF
     * console / syslog rendering (terminal escape sequences, line
     * separators that fool log parsers, NUL byte truncation in C-based
     * log shippers). C1 controls (0x80–0x9F) are intentionally
     * preserved because they are valid Unicode characters in any
     * UTF-8 caller string and the JSON serializer round-trips them
     * correctly; stripping them would mutate legitimate caller data.
     */
    private String sanitizeParserMessage(String msg) {
        if (msg == null) {
            return "";
        }
        StringBuilder sb = new StringBuilder(msg.length());
        for (int i = 0; i < msg.length(); i++) {
            char c = msg.charAt(i);
            sb.append(c < 0x20 || c == 0x7F ? ' ' : c);
        }
        String s = sb.toString().trim();
        while (s.indexOf("  ") >= 0) {
            s = s.replace("  ", " ");
        }
        if (s.length() > 500) {
            s = s.substring(0, 500) + "...";
        }
        return s;
    }

    private void audit(SqlGuardResponse r, SqlGuardRequest req) {
        r.audit.put("catalogMode", req.catalogMode);
        // Under inline mode the request-default ("ecommerce") for
        // sampleCatalogId is unused; record the literal sentinel
        // {@code "inline"} so (a) the wire shape is stable — the JSON
        // serializer drops null map values, which would silently
        // delete the key — and (b) the audit log doesn't claim a
        // sample catalog drove the decision. Pinned by
        // {@code inlineModeAuditWireFormatLocksInSentinel}.
        r.audit.put("sampleCatalogId",
                "inline".equalsIgnoreCase(req.catalogMode) ? "inline" : req.sampleCatalogId);
        r.audit.put("policyPreset", req.policyPreset);
        r.audit.put("engine", "gsp-java");
        r.audit.put("engineVersion", "local");
    }

    /**
     * Picks the first {@link CatalogDiagnosticSeverity#ERROR} message from
     * a failed {@link CatalogValidationResult}. The validator returns
     * every problem it found, but the worker only surfaces one in the
     * envelope so the BFF gets a single actionable line — same shape as
     * {@link #bindingError} picking the first resolver2 ERROR.
     */
    private String firstCatalogErrorMessage(CatalogValidationResult vr) {
        for (CatalogDiagnostic d : vr.diagnostics()) {
            if (d.severity() == CatalogDiagnosticSeverity.ERROR) {
                return sanitizeParserMessage(d.message());
            }
        }
        return "unknown validation error";
    }

    private void analyzeStatement(
            TCustomSqlStatement stmt,
            int index,
            SqlGuardRequest req,
            SqlGuardResponse r,
            UnifiedCatalogModel cat,
            EDbVendor vendor) {
        SqlGuardResponse.StatementResult sr = new SqlGuardResponse.StatementResult();
        sr.index = index;
        sr.sqlType = sqlType(stmt);
        r.statements.add(sr);

        FromScope scope = addTables(stmt, r, cat, vendor);
        int lineageBefore = r.facts.lineage.size();
        if (stmt instanceof TSelectSqlStatement) {
            analyzeSelect((TSelectSqlStatement) stmt, index, req, r, cat, scope, vendor);
        } else {
            addViolation(r, index, "DML_OR_DDL_BLOCKED", "error", "Only SELECT statements are allowed in this demo.", sr.sqlType);
        }
        sr.lineageCount = r.facts.lineage.size() - lineageBefore;
        applyStatementDecision(sr, r.violations, index);
    }

    /**
     * FROM-clause snapshot — list of resolved table names (lowercased,
     * catalog-canonical). Used by {@link #analyzeSelect} to pick a
     * default table for {@code SELECT *} expansion. The alias map that
     * used to live here is gone: dlineage now resolves alias →
     * source-table during lineage extraction, so sqlguard no longer
     * maintains its own alias index.
     */
    private static final class FromScope {
        final List<String> tableNames = new ArrayList<String>();
    }

    private String sqlType(TCustomSqlStatement s) {
        String n = s.getClass().getSimpleName().toUpperCase(Locale.ROOT);
        if (n.contains("SELECT")) return "SELECT";
        if (n.contains("DELETE")) return "DELETE";
        if (n.contains("UPDATE")) return "UPDATE";
        if (n.contains("INSERT")) return "INSERT";
        if (n.contains("CREATE") || n.contains("DROP") || n.contains("ALTER")) return "DDL";
        return n;
    }

    private FromScope addTables(TCustomSqlStatement stmt, SqlGuardResponse r,
                                 UnifiedCatalogModel cat, EDbVendor vendor) {
        FromScope scope = new FromScope();
        if (stmt.tables == null) {
            return scope;
        }
        for (int i = 0; i < stmt.tables.size(); i++) {
            TTable t = stmt.tables.getTable(i);
            String raw = t.toString();
            // Wire-form short name (no schema qualifier) for facts.tables
            // + scope. Vendor-aware normalization replaces the
            // unconditional .toLowerCase(Locale.ROOT) that was only
            // correct for PG-default-unquoted-ASCII. See
            // sqlguard/CLAUDE.md "Identifier comparison rule".
            String tableName = IdentifierService.normalizeStatic(
                    vendor, ESQLDataObjectType.dotTable, simple(raw));
            if (tableName == null || tableName.length() == 0) {
                continue;
            }
            scope.tableNames.add(tableName);
            // Catalog lookup passes the schema-qualified form so
            // findTable resolves the right schema when present, instead
            // of guessing. Catalog comparison uses the model's vendor
            // (catalog-stored canonical form), not the request vendor
            // (SQL parsing). See SqlGuardCatalogSamples "Vendor convention".
            boolean matched = SqlGuardCatalogSamples.findTable(
                    cat, qualifiedTableRef(raw)) != null;
            // findTable uses model.vendor() internally; see
            // SqlGuardCatalogSamples class Javadoc "Vendor convention".
            addTableFact(r, tableName, matched, vendor);
        }
        return scope;
    }

    private void addTableFact(SqlGuardResponse r, String n, boolean matched, EDbVendor vendor) {
        for (SqlGuardResponse.TableFact t : r.facts.tables) {
            if (IdentifierService.areEqualStatic(vendor, ESQLDataObjectType.dotTable, t.name, n)) {
                return;
            }
        }
        SqlGuardResponse.TableFact tf = new SqlGuardResponse.TableFact();
        tf.name = n;
        tf.schema = "public";
        tf.catalogMatched = matched;
        r.facts.tables.add(tf);
        if (!matched) {
            r.facts.diagnostics.add("MISSING_TABLE:" + n);
        }
    }

    /**
     * Per-statement lineage extraction. Delegates to the GSP {@code dlineage}
     * package (DataFlowAnalyzer) for the column-level relationship graph,
     * then layers sqlguard-specific processing on top: catalog binding,
     * policy tags, exposure classification, and DISTINCT-aware
     * {@code aggregateFunctions} (which dlineage does not surface).
     *
     * <p>See {@code sqlguard/CLAUDE.md} "Lineage extraction rule
     * (mandatory)": sqlguard MUST consume dlineage's in-memory model and
     * MUST NOT reimplement column-to-source-column tracing. This method
     * was rewritten to comply (replaces the legacy AST-visitor
     * {@code sourceColumnsFromAst} plus the diagnostic-only
     * {@code captureDlineageSignal} which threw away dlineage's output).
     */
    private void analyzeSelect(
            TSelectSqlStatement sel,
            int index,
            SqlGuardRequest req,
            SqlGuardResponse r,
            UnifiedCatalogModel cat,
            FromScope scope,
            EDbVendor vendor) {
        TResultColumnList astCols = sel.getResultColumnList();
        if (astCols == null) {
            return;
        }

        // Structural facts — pure-AST observations the BFF / frontend
        // surfaces to the reader ("what does this query do"). Independent
        // of dlineage and rule evaluation; called once per top-level
        // analyzeSelect entry. Blueprint §S6 (CTEs), §S12 (ORDER BY),
        // §S14/§S15 (set operations), §S16 (HAVING).
        collectStructuralFacts(sel, index, r, vendor);

        // S18 — SELECT_STAR_WITHOUT_LIMIT. Purely AST-driven (no dlineage
        // needed) so detect up-front: any `*` or `x.*` projection paired
        // with an absent {@link TSelectSqlStatement#getLimitClause()}
        // earns one warning per statement, regardless of how many star
        // refs the projection contains. Fires independently of
        // {@code SELECT_STAR_ON_SENSITIVE_TABLE} per blueprint §S18 — a
        // single SELECT * SQL can produce both findings. Fires under
        // every preset (matches the existing
        // {@code SELECT_STAR_ON_SENSITIVE_TABLE} convention).
        checkSelectStarWithoutLimit(sel, index, r, astCols);

        // S20 — UNQUALIFIED_COLUMN. Schema-stability hygiene; gated on
        // {@code policyPreset="all"} so the narrow guardrail presets
        // ({@code pii-basic} / {@code destructive-sql-block}) stay
        // single-purpose and existing requests against them are
        // byte-identical. Genuine ambiguity is already caught upstream
        // by {@link #bindingError} as {@code SQL_SEMANTIC_ERROR}.
        if ("all".equalsIgnoreCase(req.policyPreset)) {
            checkUnqualifiedColumns(sel, index, r, vendor);
        }

        // S25 — UNSUPPORTED_DIALECT_FUNCTION. AST-only walk over
        // TFunctionCall against the L5 blacklist
        // ({@link DialectFunctionDivergence}). Runs under every preset
        // — this is a safety floor (denying SQL that can't possibly
        // execute), not a hygiene preference. Independent of dlineage:
        // a function call in WHERE that doesn't participate in
        // projection lineage (e.g. WHERE DATE_TRUNC(...) > x) still
        // must fire.
        checkUnsupportedDialectFunction(sel, index, r, vendor);

        dataflow df = runDlineage(sel, vendor, r, index);

        // S22 — tenant-isolation policy gate. Fires before the
        // projection-tracing loop so the violation is reported even if
        // the projection branch later early-returns (e.g. when dlineage
        // produces an empty top resultset). Internally guards on
        // {@code userContext.tenantId} so existing tests that don't
        // declare a tenant context stay unaffected.
        checkTenantFilter(index, req, r, cat, df, vendor);

        // S24 — wrong join path. Walks dlineage joinCondition source
        // pairs against catalog FK metadata. Independent of tenant
        // filter (no shared state) but uses the same dlineage pass.
        // Always populates {@code facts.joins[]} so downstream consumers
        // can see the FK-match decision even when the rule is silent;
        // fires only when an FK exists between the joined tables but
        // the predicate doesn't align with it.
        checkWrongJoinPath(index, r, cat, df, vendor);

        table topRs = topResultset(df);
        if (df == null || topRs == null || topRs.getColumns() == null) {
            // Dlineage unavailable or empty. Fall back to AST-only star
            // expansion for SELECT *, but emit no lineage edges otherwise.
            // The diagnostic was already added by runDlineage().
            String defaultTable = scope.tableNames.isEmpty() ? null : scope.tableNames.get(0);
            for (int i = 0; i < astCols.size(); i++) {
                String text = resultExpressionText(astCols.getResultColumn(i));
                if ("*".equals(text) || text.endsWith(".*")) {
                    expandStar(index, r, cat, defaultTable, vendor);
                }
            }
            return;
        }

        // Index resultset IDs so the leaf-tracer can tell intermediate
        // function/CAST resultsets from real catalog tables.
        Set<String> resultsetIds = new HashSet<String>();
        for (table rs : df.getResultsets()) {
            if (rs.getId() != null) resultsetIds.add(rs.getId());
        }

        String defaultTable = scope.tableNames.isEmpty() ? null : scope.tableNames.get(0);
        int astIdx = 0;
        for (column rsCol : topRs.getColumns()) {
            if (isSystemColumn(rsCol)) {
                continue;
            }
            TResultColumn rc = astIdx < astCols.size() ? astCols.getResultColumn(astIdx) : null;
            astIdx++;
            String text = rc == null ? "" : resultExpressionText(rc);
            if ("*".equals(text) || text.endsWith(".*")) {
                expandStar(index, r, cat, defaultTable, vendor);
                continue;
            }

            // Result-column name is a column-class identifier. Sqlguard's
            // virtual "result." namespace is sqlguard's own convention,
            // but the projected column name itself follows the dialect's
            // identifier rules.
            String target = "result." + (rsCol.getName() == null ? ""
                    : IdentifierService.normalizeStatic(vendor, ESQLDataObjectType.dotColumn, rsCol.getName()));

            // Tier 1 — lineage edges come from dlineage. Walk the
            // relationship graph from the result column to its leaf
            // sources, recursing through intermediate (function / CAST)
            // resultsets.
            List<sourceColumn> leaves = new ArrayList<sourceColumn>();
            boolean[] traversedResultset = new boolean[] { false };
            Set<String> visited = new HashSet<String>();
            traceLeaves(df, resultsetIds, topRs.getId(), rsCol.getName(),
                    visited, leaves, traversedResultset, vendor);

            // Tier 3a — AST-side window detection (blueprint §1.5.B B1/B2).
            // Captured up-front so we can both decide whether to walk
            // dlineage's window-extra fdr edges below AND store the field
            // on the lineage edge for downstream consumers.
            List<SqlGuardResponse.WindowFn> winFns = windowFunctionsFromAst(rc);

            // Tier 1b — when a window aggregate participates in the
            // projection, PARTITION BY / ORDER BY columns also drive the
            // result and belong in sourceColumns. dlineage surfaces them
            // as {@code fdr} edges with {@code clauseType="selectList"}
            // (partition) and {@code "orderby"} (order) on the function
            // resultset; the regular {@link #traceLeaves} pass skips them
            // because they're not {@code fdd} data-dependency edges.
            //
            // Partition and order columns go into SEPARATE buckets so the
            // catalog-binding pass can label exposure correctly per role
            // ({@code window_partition} vs {@code window_order}). Folding
            // both into the main leaves list — as the original B1/B2
            // implementation did — mis-attributed partition-only PII as
            // {@code explicit_select} exposure with a misleading
            // {@code SENSITIVE_COLUMN_IN_OUTPUT} violation. The fix splits
            // the buckets and dispatches the right sensitive rule per
            // exposure (blueprint §B5: {@code SENSITIVE_COLUMN_IN_WINDOW_PARTITION}).
            List<sourceColumn> partitionExtras = new ArrayList<sourceColumn>();
            List<sourceColumn> orderExtras = new ArrayList<sourceColumn>();
            if (winFns != null && !winFns.isEmpty()) {
                Set<String> windowVisited = new HashSet<String>();
                traceWindowExtras(df, resultsetIds, topRs.getId(), rsCol.getName(),
                        windowVisited, partitionExtras, orderExtras, vendor);
            }

            // Tier 2 — sqlguard's catalog binding: translate dlineage's
            // (parent_name, column) into the fully-qualified
            // "schema.table.column" that the wire contract uses, dropping
            // any leaf that doesn't bind to a known catalog column.
            // dlineage's parent_name preserves the user-written casing
            // (e.g. "sbTransaction"); IdentifierService folds it per the
            // dialect's rules so catalog lookups succeed.
            //
            // sourceColumns (wire format) is the union of all three roles
            // in document order: function-arg leaves first (matching the
            // pre-B1 contract), then partition extras, then order extras.
            LinkedHashSet<String> functionArgSources = bindToCatalog(leaves, cat, vendor);
            LinkedHashSet<String> partitionSources = bindToCatalog(partitionExtras, cat, vendor);
            LinkedHashSet<String> orderSources = bindToCatalog(orderExtras, cat, vendor);

            // Role precedence: a column appearing in BOTH function-arg
            // (dlineage fdd) AND partition/order (dlineage fdr) is
            // legitimate when the AST shows the column is a real projection
            // dependency — either a real function argument like {@code
            // LAG(email) OVER (PARTITION BY email)} (LAG passes email
            // through to output AND partitions by it) or a scalar-operator
            // operand like {@code email || COUNT(*) OVER (PARTITION BY
            // email)} (the bare {@code email} ref is a real projection
            // source). It is dlineage over-attribution when the AST does
            // NOT reference the column outside the window spec (e.g.
            // {@code COUNT(*) OVER (PARTITION BY email)} — dlineage's
            // fdd edge includes email as a function "dependency" because
            // of the {@code *} glob, even though email isn't really a
            // projection source).
            //
            // The AST collector ({@link #collectAstProjectionDependencies})
            // gathers catalog-canonical references to columns OUTSIDE any
            // window spec, then we strip from functionArgSources only
            // those overlap columns that are NOT in that set. Without
            // this gate the COUNT(*) shape would incorrectly fire
            // {@code SENSITIVE_COLUMN_IN_OUTPUT} on a partition-only PII
            // column.
            if (!partitionSources.isEmpty() || !orderSources.isEmpty()) {
                Set<String> astProjectionDeps =
                        collectAstProjectionDependencies(rc, cat, vendor);
                stripOverAttributedSources(functionArgSources,
                        partitionSources, orderSources, astProjectionDeps);
            }

            // sourceColumns (wire format) is the union of all three roles
            // in document order: function-arg first (matching the pre-B1
            // contract), then partition, then order.
            LinkedHashSet<String> sourceColumns = new LinkedHashSet<String>();
            sourceColumns.addAll(functionArgSources);
            sourceColumns.addAll(partitionSources);
            sourceColumns.addAll(orderSources);
            List<String> srcList = new ArrayList<String>(sourceColumns);

            // Tier 3 — sqlguard-specific value-add. aggregateFunctions
            // (with DISTINCT modifier) is NOT surfaced by dlineage; the
            // AST visitor stays as the authoritative source for this
            // field. See sqlguard/CLAUDE.md Case 1 escalation about the
            // dlineage gap.
            List<String> aggFns = aggregateFunctionsFromAst(rc);

            // lineageType: window beats aggregate beats direct/expression.
            // A projection that mixes scalar + window calls is rare but
            // legal — we still classify it as window because the window
            // semantics (frame, partitioning) are the dominant signal.
            String ltype;
            if (winFns != null && !winFns.isEmpty()) {
                ltype = "window";
            } else if (aggFns != null && !aggFns.isEmpty()) {
                ltype = "aggregate";
            } else if (!traversedResultset[0] && leaves.size() == 1
                    && nameEquals(vendor, rsCol.getName(), leaves.get(0).getColumn())) {
                ltype = "direct";
            } else {
                ltype = "expression";
            }

            // Per-role column-fact emission. Each call dispatches the
            // right sensitive-rule via addColumnFact's exposure switch
            // — a column that appears in both projection AND partition
            // produces two column facts (one per role) and two distinct
            // violations, mirroring the dual-exposure reality.
            for (String sc : functionArgSources) {
                addColumnFactForSource(r, sc, "explicit_select", cat, vendor);
            }
            for (String sc : partitionSources) {
                addColumnFactForSource(r, sc, "window_partition", cat, vendor);
            }
            for (String sc : orderSources) {
                addColumnFactForSource(r, sc, "window_order", cat, vendor);
            }
            addLineage(r, index, target, srcList, ltype, aggFns, winFns);
        }

        // S23 — AGGREGATION_GRAIN_MISMATCH. Runs last so every aggregate
        // lineage edge for this statement is already in
        // {@code r.facts.lineage}. Self-silences when the request carries
        // no {@code question}, so existing callers stay byte-identical.
        checkAggregationGrainMismatch(index, req, r);
    }

    /**
     * Walk one statement's AST and append four kinds of structural fact
     * to the response: CTE names, top-level ORDER BY items, HAVING clauses,
     * and set-operation operators. Pure-AST observation — no dlineage, no
     * resolver state, no catalog. Each fact carries {@code statementIndex}
     * so multi-statement requests stay disambiguated.
     *
     * <p>For combined queries ({@code A UNION B INTERSECT C}) the entry
     * {@code sel} is the root combined node. CTEs and the top-level ORDER
     * BY live there; HAVING and per-leaf set operators live deeper. The
     * walker recurses through {@link TSelectSqlStatement#getLeftStmt()} /
     * {@link TSelectSqlStatement#getRightStmt()} and processes each leaf.
     *
     * <p>Blueprint refs: §S6 (CTEs), §S12 (ORDER BY), §S14/§S15 (set
     * operations), §S16 (HAVING).
     */
    private void collectStructuralFacts(TSelectSqlStatement sel, int index,
                                         SqlGuardResponse r, EDbVendor vendor) {
        if (sel == null) {
            return;
        }
        // CTEs are on the outermost node; safe to read at the root only.
        collectCtes(sel, r);
        // Top-level ORDER BY applies to the whole result; read at the root.
        collectOrderBy(sel, index, r);
        // HAVING + set operators may be nested inside combined queries.
        // Vendor threads through so EXCEPT-vs-MINUS DISTINCT disambiguates
        // by dialect (the parser's int constants collide at value 11).
        collectHavingAndSetOps(sel, index, r, vendor);
    }

    private void collectCtes(TSelectSqlStatement sel, SqlGuardResponse r) {
        TCTEList cteList = sel.getCteList();
        if (cteList == null) return;
        for (int i = 0; i < cteList.size(); i++) {
            TCTE cte = cteList.getCTE(i);
            if (cte == null) continue;
            TObjectName name = cte.getTableName();
            if (name == null) continue;
            String s = name.toString();
            if (s == null || s.isEmpty()) continue;
            r.facts.ctes.add(s.toLowerCase(Locale.ROOT));
        }
    }

    private void collectOrderBy(TSelectSqlStatement sel, int index, SqlGuardResponse r) {
        TOrderBy ob = sel.getOrderbyClause();
        if (ob == null || ob.getItems() == null) return;
        for (int i = 0; i < ob.getItems().size(); i++) {
            TOrderByItem item = ob.getItems().getOrderByItem(i);
            if (item == null || item.getSortKey() == null) continue;
            SqlGuardResponse.OrderByFact fact = new SqlGuardResponse.OrderByFact();
            fact.statementIndex = index;
            fact.column = item.getSortKey().toString().toLowerCase(Locale.ROOT);
            ESortType sort = item.getSortOrder();
            if (sort == ESortType.asc) {
                fact.direction = "ASC";
            } else if (sort == ESortType.desc) {
                fact.direction = "DESC";
            } else {
                fact.direction = "NONE";
            }
            r.facts.orderBy.add(fact);
        }
    }

    private void collectHavingAndSetOps(TSelectSqlStatement sel, int index,
                                         SqlGuardResponse r, EDbVendor vendor) {
        if (sel == null) return;
        if (sel.isCombinedQuery()) {
            // GSP represents `A op1 B op2 C` as a LEFT-recursive tree: the
            // ROOT carries op2, and op1 lives on the left subtree. To emit
            // operators in TEXTUAL order, recurse left FIRST, then add the
            // current node's op, then recurse right. The earlier order
            // (emit, left, right) reported `[op2, op1]` for chained queries,
            // which misled consumers reading the array as the operator
            // sequence the SQL wrote. Codex review §P2.
            collectHavingAndSetOps(sel.getLeftStmt(), index, r, vendor);
            String op = setOperatorName(sel.getSetOperator(), vendor);
            if (op != null) {
                SqlGuardResponse.SetOperationFact f = new SqlGuardResponse.SetOperationFact();
                f.statementIndex = index;
                f.op = op;
                r.facts.setOperations.add(f);
            }
            collectHavingAndSetOps(sel.getRightStmt(), index, r, vendor);
            return;
        }
        // Non-combined leaf: capture HAVING if present. HAVING hangs off
        // the GROUP BY clause in GSP's AST (matching the SQL grammar
        // requirement that HAVING requires GROUP BY); a present HAVING
        // without GROUP BY is impossible from a valid parse tree.
        // TGroupBy.getHavingClause() returns the TExpression directly —
        // the {@link THavingClause} node type is internal to the parser
        // and is not exposed at the GROUP BY level.
        TGroupBy gb = sel.getGroupByClause();
        TExpression having = gb != null ? gb.getHavingClause() : null;
        if (having != null) {
            String text = having.toString();
            if (text != null && !text.trim().isEmpty()) {
                SqlGuardResponse.AggregationFact a = new SqlGuardResponse.AggregationFact();
                a.statementIndex = index;
                a.having = text.trim().toLowerCase(Locale.ROOT);
                r.facts.aggregations.add(a);
            }
        }
    }

    /**
     * Map a {@code TSelectSqlStatement.setOperator_*} integer constant to a
     * canonical operator name. {@code vendor} disambiguates the {@code = 11}
     * collision between {@code SET_OPERATOR_EXCEPTDISTINCT} and
     * {@code SET_OPERATOR_MINUSDISTINCT}: in vendors that don't grammar-support
     * EXCEPT (Oracle, MySQL), value 11 must be MINUS DISTINCT; everywhere else
     * the SQL-standard EXCEPT DISTINCT spelling wins. Codex review §P2.
     */
    private static String setOperatorName(int code, EDbVendor vendor) {
        switch (code) {
            case TSelectSqlStatement.setOperator_union:        return "UNION";
            case TSelectSqlStatement.setOperator_unionall:     return "UNION ALL";
            case TSelectSqlStatement.SET_OPERATOR_UNIONDISTINCT: return "UNION DISTINCT";
            case TSelectSqlStatement.setOperator_intersect:    return "INTERSECT";
            case TSelectSqlStatement.setOperator_intersectall: return "INTERSECT ALL";
            case TSelectSqlStatement.SET_OPERATOR_INTERSECTDISTINCT: return "INTERSECT DISTINCT";
            case TSelectSqlStatement.setOperator_minus:        return "MINUS";
            case TSelectSqlStatement.setOperator_minusall:     return "MINUS ALL";
            case TSelectSqlStatement.setOperator_except:       return "EXCEPT";
            case TSelectSqlStatement.setOperator_exceptall:    return "EXCEPT ALL";
            case TSelectSqlStatement.SET_OPERATOR_MINUSDISTINCT:
                // Also matches SET_OPERATOR_EXCEPTDISTINCT — both = 11.
                // Vendor decides the spelling: Oracle/MySQL grammars use
                // MINUS, everyone else uses EXCEPT (SQL standard).
                return isMinusDialect(vendor) ? "MINUS DISTINCT" : "EXCEPT DISTINCT";
            default: return null;
        }
    }

    private static boolean isMinusDialect(EDbVendor vendor) {
        return vendor == EDbVendor.dbvoracle || vendor == EDbVendor.dbvmysql;
    }

    /**
     * Blueprint §S18 — emit {@code SELECT_STAR_WITHOUT_LIMIT} when the
     * statement projects a star ({@code SELECT *} or {@code t.*}) and
     * the SELECT has no {@code LIMIT} clause. Independent of
     * {@code SELECT_STAR_ON_SENSITIVE_TABLE}; both can fire on the same
     * SQL (the blueprint's gold S18 SQL produces exactly that pair).
     *
     * <p>One violation per statement regardless of how many star refs.
     * Two stars in the same projection are still one unbounded result.
     */
    private void checkSelectStarWithoutLimit(TSelectSqlStatement sel, int index,
                                              SqlGuardResponse r, TResultColumnList astCols) {
        if (sel.getLimitClause() != null) {
            return;
        }
        for (int i = 0; i < astCols.size(); i++) {
            String text = resultExpressionText(astCols.getResultColumn(i));
            if ("*".equals(text) || text.endsWith(".*")) {
                addViolation(r, index, "SELECT_STAR_WITHOUT_LIMIT", "warn",
                        "Unbounded SELECT *; add LIMIT or restrict columns.",
                        "SELECT * without LIMIT clause");
                return;
            }
        }
    }

    /**
     * Blueprint §S20 — emit {@code UNQUALIFIED_COLUMN} for column
     * references that resolved cleanly but lack a table qualifier in
     * the SQL ({@code SELECT col FROM t1 JOIN t2 …} where only one of
     * {@code t1} / {@code t2} contains {@code col}). The query works
     * today; the warning is the schema-stability signal — if a future
     * schema change adds {@code col} to a second in-scope table, the
     * resolver flips to {@code AMBIGUOUS_COLUMN} (which sqlguard's
     * {@link #bindingError} surfaces as {@code SQL_SEMANTIC_ERROR}
     * upstream of this method).
     *
     * <p>Genuinely-ambiguous references therefore never reach this
     * method — they short-circuit in {@code bindingError()}. The
     * filter chain ({@code dotColumn} object type, empty
     * {@link TObjectName#getTableString()}, non-null
     * {@link TObjectName#getSourceTable()}, {@code !isAmbiguous()})
     * picks off exactly the S20 case.
     *
     * <p>Dedup uses the {@code TObjectName.toString()} text so the same
     * column referenced twice in one SELECT produces one finding. The
     * resolved-table comparison is identifier-aware via
     * {@link IdentifierService}.
     */
    private void checkUnqualifiedColumns(TSelectSqlStatement sel, int index,
                                          SqlGuardResponse r, EDbVendor vendor) {
        UnqualifiedColumnCollector v = new UnqualifiedColumnCollector(vendor);
        sel.acceptChildren(v);
        for (Map.Entry<String, String> e : v.findings.entrySet()) {
            addViolation(r, index, "UNQUALIFIED_COLUMN", "warn",
                    "Unqualified column resolved to " + e.getValue()
                            + "; qualify it for stability across schema changes.",
                    e.getKey() + " (resolves to " + e.getValue() + ")");
        }
    }

    /**
     * AST visitor for {@link #checkUnqualifiedColumns}. Collects each
     * column reference whose SQL text has no table qualifier but whose
     * resolved {@link TObjectName#getSourceTable()} is non-null and
     * unambiguous. {@code findings} keys on the column's textual form
     * so a repeated reference (e.g. in SELECT list and GROUP BY) yields
     * one finding. Value is the catalog-canonical resolved table name.
     */
    private static final class UnqualifiedColumnCollector extends TParseTreeVisitor {
        final LinkedHashMap<String, String> findings = new LinkedHashMap<String, String>();
        private final EDbVendor vendor;

        UnqualifiedColumnCollector(EDbVendor vendor) {
            this.vendor = vendor;
        }

        @Override
        public void preVisit(gudusoft.gsqlparser.nodes.TObjectName node) {
            if (node == null) return;
            if (node.getDbObjectType() != gudusoft.gsqlparser.EDbObjectType.column) return;
            String col = node.getColumnNameOnly();
            if (col == null || col.isEmpty() || "*".equals(col)) return;
            // Qualified in the SQL? Skip — the rule is specifically about
            // unqualified references that the resolver had to disambiguate.
            String tableQualifier = node.getTableString();
            if (tableQualifier != null && tableQualifier.length() > 0) return;
            // Resolver couldn't pin it down → bindingError() already
            // surfaced this as SQL_SEMANTIC_ERROR. Defensive guard.
            if (node.isAmbiguous()) return;
            gudusoft.gsqlparser.nodes.TTable src = node.getSourceTable();
            if (src == null || src.getName() == null) return;
            // Skip CTE / subquery sources — they have no catalog identity
            // and "resolves to RS-1" is not actionable advice. Real
            // tables have an empty subquery and CTE pointer; that's the
            // signal we need.
            if (src.getSubquery() != null || src.getCTE() != null) return;
            String resolvedTable = IdentifierService.normalizeStatic(
                    vendor, ESQLDataObjectType.dotTable, src.getName().toString());
            if (resolvedTable == null || resolvedTable.isEmpty()) return;
            String key = node.toString();
            findings.putIfAbsent(key, resolvedTable);
        }
    }

    /**
     * Blueprint §S25 — emit {@code UNSUPPORTED_DIALECT_FUNCTION} when
     * the SQL calls a function name that's blacklisted for the
     * request dialect by {@link DialectFunctionDivergence}.
     *
     * <p>AST-only ({@link TFunctionCall} visitor) rather than going
     * through {@code dlineage.relationship.getFunction()} so functions
     * outside projection lineage still fire — e.g.
     * {@code SELECT 1 FROM t WHERE DATE_TRUNC('day', created_at) > NOW()}
     * dies the same as {@code SELECT DATE_TRUNC(...) FROM t}.
     *
     * <p>Dedup is left to {@link #addViolation}'s existing
     * {@code (statementIndex, code, evidence)} guard: the evidence
     * carries the function name as written, so the same name twice in
     * one statement produces one violation. Functions not in the
     * blacklist (UDFs, vendor-specific extensions we haven't gated,
     * unknown names) are silently allowed — the blacklist is
     * explicit-only by design, see
     * {@link DialectFunctionDivergence} Javadoc.
     *
     * <p>Severity is {@code error → DENY}: blacklist entries are
     * cases where the SQL genuinely cannot execute on the target
     * dialect, not soft preferences. The remediation hint stored in
     * the blacklist is appended to the violation message so the user
     * gets actionable advice.
     */
    private void checkUnsupportedDialectFunction(TSelectSqlStatement sel, int index,
                                                  SqlGuardResponse r, EDbVendor vendor) {
        UnsupportedDialectFunctionCollector v =
                new UnsupportedDialectFunctionCollector(vendor);
        sel.acceptChildren(v);
        for (Map.Entry<String, String> e : v.findings.entrySet()) {
            String fnName = e.getKey();
            String hint = e.getValue();
            String message = fnName + " is not supported by " + vendor.toString()
                    + (hint == null ? "." : ". " + hint);
            addViolation(r, index, "UNSUPPORTED_DIALECT_FUNCTION", "error",
                    message, fnName);
        }
    }

    /**
     * AST visitor for {@link #checkUnsupportedDialectFunction}.
     * Collects each {@link TFunctionCall} whose function name is
     * blacklisted for the request dialect by
     * {@link DialectFunctionDivergence}. {@code findings} is keyed on
     * the as-written function name (preserves source casing for
     * evidence) → remediation hint, so the same call repeated yields
     * one finding.
     *
     * <p>Skips window / FILTER forms by the same logic as
     * {@link AggregateCallCollector}: a window or FILTER form's
     * <i>aggregate</i> isn't what's being challenged — the wrapping
     * window function is. But for the dialect-function check the
     * unsupported call IS the issue, not the wrapping. So we walk all
     * TFunctionCalls regardless of window/FILTER context — if
     * {@code DATE_TRUNC} appears inside an OVER() partition, it's
     * still unsupported on MySQL.
     */
    private static final class UnsupportedDialectFunctionCollector extends TParseTreeVisitor {
        final LinkedHashMap<String, String> findings = new LinkedHashMap<String, String>();
        private final EDbVendor vendor;

        UnsupportedDialectFunctionCollector(EDbVendor vendor) {
            this.vendor = vendor;
        }

        @Override
        public void preVisit(TFunctionCall node) {
            if (node == null || node.getFunctionName() == null) {
                return;
            }
            String raw = node.getFunctionName().toString();
            if (raw == null || raw.isEmpty()) {
                return;
            }
            // Strip any quoting / brackets the lexer left on the name.
            // SqlGuardCatalogSamples.strip handles `"..."`, `[...]`, and
            // backtick forms — same helper AggregateCallCollector uses.
            String name = SqlGuardCatalogSamples.strip(raw);
            if (name == null || name.isEmpty()) {
                return;
            }
            String hint = DialectFunctionDivergence.lookup(vendor, name);
            if (hint == null) {
                return;
            }
            // Key on the as-written name (preserves casing in evidence)
            // and upper-case for the dedup compare so two appearances
            // of `date_trunc` and `DATE_TRUNC` in the same statement
            // collapse to one finding.
            String key = name.toUpperCase(Locale.ROOT);
            if (!findings.containsKey(key)) {
                findings.put(key, hint);
            }
        }
    }

    /**
     * Intent-keyword whitelist for blueprint §S23. Lower-case substring
     * match against the natural-language {@code question} text — present
     * → caller is asking for a DISTINCT-grain answer. The list is
     * intentionally short and English-only; this is a soft WARN rule,
     * not a contract enforcer. Add new keywords here when the false-
     * negative rate matters; resist the urge to swap in regex / NLP —
     * the simplicity is the design.
     */
    private static final String[] DISTINCT_INTENT_KEYWORDS = new String[] {
            "distinct", "unique", "different", "deduplicat",
            "how many kinds", "how many types"
    };

    /**
     * Blueprint §S23 — emit {@code AGGREGATION_GRAIN_MISMATCH} when the
     * caller's natural-language {@link SqlGuardRequest#question} text
     * implies a DISTINCT-grain answer (keyword whitelist match) but the
     * SQL's aggregate edge for the answering column uses a non-DISTINCT
     * call, OR vice-versa (plain-count intent answered with
     * {@code COUNT(DISTINCT)}).
     *
     * <p>Self-silences when {@code question} is null/blank — the
     * question itself is the opt-in. No preset gate (parallels
     * {@link #checkTenantFilter}'s {@code userContext.tenantId} opt-in,
     * not {@link #checkUnqualifiedColumns}'s preset gate): the rule is
     * useful whenever the caller declared intent, regardless of which
     * hygiene preset they picked. The intent heuristic is a tiny
     * keyword whitelist by design (see
     * {@link #DISTINCT_INTENT_KEYWORDS}); NLP is explicitly out of
     * scope, false positives surface as soft WARNs only.
     *
     * <p>Signal B (SQL grain) comes from
     * {@link SqlGuardResponse.LineageEdge#aggregateFunctions}, populated
     * by S2's AST visitor. An edge whose list contains any
     * {@code "FUNC(DISTINCT)"} entry is DISTINCT-grain; otherwise plain.
     * One violation per disagreeing aggregate edge — non-aggregate
     * edges carry no grain signal and are skipped. Multi-aggregate
     * edges (e.g. {@code COUNT(DISTINCT a) / COUNT(*)}) read as
     * DISTINCT-grain by the {@code any-DISTINCT} rule, matching S2's
     * "list grows naturally" contract.
     */
    private void checkAggregationGrainMismatch(int index, SqlGuardRequest req, SqlGuardResponse r) {
        if (req == null || req.question == null) {
            return;
        }
        String question = req.question.toLowerCase(Locale.ROOT);
        if (question.trim().isEmpty()) {
            return;
        }
        boolean intentDistinct = false;
        for (String needle : DISTINCT_INTENT_KEYWORDS) {
            if (question.contains(needle)) {
                intentDistinct = true;
                break;
            }
        }
        for (SqlGuardResponse.LineageEdge edge : r.facts.lineage) {
            if (edge.statementIndex != index) continue;
            if (edge.aggregateFunctions == null || edge.aggregateFunctions.isEmpty()) continue;

            boolean sqlDistinct = false;
            for (String fn : edge.aggregateFunctions) {
                if (fn != null && fn.endsWith("(DISTINCT)")) {
                    sqlDistinct = true;
                    break;
                }
            }
            if (intentDistinct == sqlDistinct) continue;

            String target = edge.targetColumn == null ? "?" : edge.targetColumn;
            String fnList = joinList(edge.aggregateFunctions);
            String message;
            String evidence;
            if (intentDistinct) {
                message = "Question implies a DISTINCT aggregation but SQL uses non-distinct "
                        + fnList + "; result may double-count rows.";
                evidence = target + ": question implies DISTINCT, SQL aggregates with " + fnList;
            } else {
                message = "Question implies a plain aggregation but SQL uses "
                        + fnList + "; DISTINCT may silently under-count rows.";
                evidence = target + ": question implies plain aggregation, SQL aggregates with " + fnList;
            }
            addViolation(r, index, "AGGREGATION_GRAIN_MISMATCH", "warn", message, evidence);
        }
    }

    private static String joinList(List<String> items) {
        if (items == null || items.isEmpty()) return "";
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < items.size(); i++) {
            if (i > 0) sb.append(", ");
            sb.append(items.get(i));
        }
        return sb.toString();
    }

    /**
     * Blueprint §S22 — emit {@code MISSING_TENANT_FILTER} when any
     * in-scope table carries a {@code TENANT_KEY}-tagged column but no
     * WHERE / HAVING predicate references that column.
     *
     * <p>Gated on {@code userContext.tenantId} being supplied — the
     * blueprint pins the rule to "caller declared tenant context", so
     * silence-by-default keeps existing requests (and the S1 / S2 seed
     * tests, which run preset {@code "all"} on broker SQL without
     * tenant filters) byte-identical until the caller opts in. Severity
     * is {@code warn} at all preset levels; a future {@code error}
     * escalation under a dedicated tenant-aware preset is the natural
     * extension (TODO item #2 ramp-up).
     *
     * <p>In-scope tables come from {@code df.getTables()}, NOT
     * {@link TCustomSqlStatement#tables}: CTE-wrapped SQL (the blueprint
     * S22 fixture is one) only puts the CTE alias in
     * {@code stmt.tables}, while dlineage transparently surfaces the
     * inner real tables. The check would silently no-op on the canonical
     * S22 fixture without this.
     *
     * <p>WHERE / HAVING coverage uses dlineage's {@code fdr} edges with
     * {@code sourceColumn.getClauseType()} in {@code {"where",
     * "having"}}. {@code joinCondition} is intentionally NOT accepted as
     * a tenant filter — {@code ON a.tenant_id = b.tenant_id} correlates
     * two tenant columns to each other but does not pin them to any
     * single tenant, so a join-only "filter" still returns cross-tenant
     * rows.
     */
    private void checkTenantFilter(int index, SqlGuardRequest req, SqlGuardResponse r,
                                    UnifiedCatalogModel cat, dataflow df, EDbVendor vendor) {
        if (df == null) {
            return;
        }
        if (req == null || req.userContext == null) {
            return;
        }
        Object tenantId = req.userContext.get("tenantId");
        if (!(tenantId instanceof String) || ((String) tenantId).isEmpty()) {
            return;
        }

        // Required: each TENANT_KEY-tagged column on every in-scope catalog
        // table. Keyed by catalog-canonical table name so the diff against
        // `filtered` (also keyed by catalog name) folds correctly.
        Map<String, LinkedHashSet<String>> required = new LinkedHashMap<String, LinkedHashSet<String>>();
        if (df.getTables() != null) {
            for (table t : df.getTables()) {
                if (t == null || t.getName() == null) {
                    continue;
                }
                SqlGuardCatalogSamples.TableMatch tm =
                        SqlGuardCatalogSamples.findTable(cat, t.getName());
                if (tm == null) {
                    continue;
                }
                LinkedHashSet<String> tenantCols = null;
                for (ColumnModel col : tm.table.columns()) {
                    if (SqlGuardCatalogSamples.policyTagNames(col).contains("TENANT_KEY")) {
                        if (tenantCols == null) {
                            tenantCols = new LinkedHashSet<String>();
                        }
                        tenantCols.add(col.name());
                    }
                }
                if (tenantCols != null && !required.containsKey(tm.table.name())) {
                    required.put(tm.table.name(), tenantCols);
                }
            }
        }
        if (required.isEmpty()) {
            return;
        }

        // Filtered: (table, tenantCol) pairs that appear in a WHERE or
        // HAVING predicate per dlineage's fdr edges. Catalog-bound so a
        // tenant column referenced under its catalog identifier matches
        // regardless of the SQL author's casing.
        Map<String, LinkedHashSet<String>> filtered = new LinkedHashMap<String, LinkedHashSet<String>>();
        if (df.getRelationships() != null) {
            for (relationship rel : df.getRelationships()) {
                if (!"fdr".equals(rel.getType()) || rel.getSources() == null) {
                    continue;
                }
                for (sourceColumn src : rel.getSources()) {
                    String clause = src.getClauseType();
                    if (!"where".equals(clause) && !"having".equals(clause)) {
                        continue;
                    }
                    if (src.getParent_name() == null || src.getColumn() == null) {
                        continue;
                    }
                    SqlGuardCatalogSamples.TableMatch tm =
                            SqlGuardCatalogSamples.findTable(cat, src.getParent_name());
                    if (tm == null) {
                        continue;
                    }
                    ColumnModel col = SqlGuardCatalogSamples.findColumn(tm, src.getColumn());
                    if (col == null) {
                        continue;
                    }
                    LinkedHashSet<String> cols = filtered.get(tm.table.name());
                    if (cols == null) {
                        cols = new LinkedHashSet<String>();
                        filtered.put(tm.table.name(), cols);
                    }
                    cols.add(col.name());
                }
            }
        }

        // Diff → ordered evidence string ("tableA (col1), tableB (col2, col3)").
        StringBuilder evidence = new StringBuilder();
        for (Map.Entry<String, LinkedHashSet<String>> entry : required.entrySet()) {
            LinkedHashSet<String> filteredCols = filtered.get(entry.getKey());
            LinkedHashSet<String> missing = new LinkedHashSet<String>();
            for (String tenantCol : entry.getValue()) {
                if (filteredCols == null || !filteredCols.contains(tenantCol)) {
                    missing.add(tenantCol);
                }
            }
            if (missing.isEmpty()) {
                continue;
            }
            if (evidence.length() > 0) {
                evidence.append(", ");
            }
            evidence.append(entry.getKey()).append(" (");
            boolean firstCol = true;
            for (String col : missing) {
                if (!firstCol) {
                    evidence.append(", ");
                }
                firstCol = false;
                evidence.append(col);
            }
            evidence.append(")");
        }
        if (evidence.length() == 0) {
            return;
        }
        addViolation(r, index, "MISSING_TENANT_FILTER", "warn",
                "Tables with TENANT_KEY columns are referenced without a tenant-scoped predicate; query returns cross-tenant rows.",
                evidence.toString());
    }

    /**
     * S24 — {@code WRONG_JOIN_PATH}. Walks dlineage's {@code fdr} edges
     * whose sources carry {@code clauseType="joinCondition"}, pairs the
     * sources by appearance order (one fdr per ON clause; multi-predicate
     * ON clauses put all sources into the same {@code sources[]} array,
     * left/right interleaved), catalog-binds each side, and compares the
     * predicate against the catalog's FK constraints in either direction.
     *
     * <p>For every successfully bound pair, emits a {@link SqlGuardResponse.JoinFact}
     * with {@code joinPath} and {@code joinPathMatchesForeignKey}. When
     * an FK exists between the two joined tables but the predicate's
     * columns don't line up with it, the JoinFact also gets
     * {@code expectedForeignKeyPath} (the correct predicate) and one
     * {@code WRONG_JOIN_PATH} violation fires.</p>
     *
     * <p><b>Two-pass scan per ON clause</b> handles mixed-predicate cases.
     * dlineage labels every column that appears anywhere in an ON clause
     * with {@code clauseType="joinCondition"} — including the column-side
     * of non-equi predicates like {@code A.x &gt; B.y} and {@code A.z
     * &gt; B.start_ts}. A naive single-pass would treat
     * {@code (A.x, B.y, A.z, B.start_ts)} as two equi-join pairs and
     * misfire WRONG_JOIN_PATH on the second pair (the inequality). We
     * avoid that by first scanning every pair to see whether ANY pair
     * matched an FK; if yes, suppress WRONG_JOIN_PATH for all other
     * pairs between the same two tables in this ON clause (they're
     * additional filters / multi-column equi-joins / non-equi clauses,
     * not wrong-path joins).</p>
     *
     * <p><b>Silent paths</b> (intentional, not bugs):</p>
     * <ul>
     *   <li>No FK exists between the joined tables at all → no violation
     *       (we can't second-guess novel joins; only flag mismatches
     *       against KNOWN constraints).</li>
     *   <li>Predicate columns don't bind to the catalog (e.g. CTE or
     *       subquery alias) → no violation (we can't verify what we
     *       can't see).</li>
     *   <li>Pure non-equi joins where dlineage hasn't paired both sides
     *       (e.g. {@code UPPER(A.x) = B.y} — the function side comes
     *       back with {@code clauseType="unknown"} rather than
     *       {@code "joinCondition"} and is silently dropped).</li>
     * </ul>
     *
     * <p><b>Single-column FK only (V1).</b> Composite FKs in the catalog
     * are ignored — they need AND-chained predicate correlation within
     * a single ON clause, which is shape work deferred to blueprint B8
     * (TODO item #15). Until then, an SQL query that correctly joins
     * via a composite FK gets {@code joinPathMatchesForeignKey:false}
     * but no violation fires (the catalog has no single-column FK between
     * the tables, so the rule stays silent).</p>
     *
     * <p>Severity is {@code warn}: a wrong join path produces meaningful
     * but plausible-looking output, so the rule warns rather than DENYs
     * — escalation is up to the BFF / human reviewer.</p>
     */
    private void checkWrongJoinPath(int index, SqlGuardResponse r,
                                    UnifiedCatalogModel cat, dataflow df, EDbVendor vendor) {
        if (df == null || df.getRelationships() == null) {
            return;
        }
        for (relationship rel : df.getRelationships()) {
            if (!"fdr".equals(rel.getType()) || rel.getSources() == null) {
                continue;
            }
            // Collect join-condition sources within this relationship in
            // appearance order. dlineage emits one fdr per ON clause;
            // multi-predicate ONs put all sources into the same sources[].
            List<sourceColumn> joinSources = new ArrayList<sourceColumn>();
            for (sourceColumn src : rel.getSources()) {
                if ("joinCondition".equals(src.getClauseType())) {
                    joinSources.add(src);
                }
            }
            // Two-pass scan. Pass 1: bind every pair to the catalog and
            // record whether it matches an FK. Pass 2: emit JoinFact for
            // each bound pair; fire WRONG_JOIN_PATH only when no pair
            // between the SAME table pair already matched an FK.
            //
            // Suppression is keyed by (leftTable, rightTable) identity
            // (order-independent), not by the whole ON clause. This way
            // a correct equi-join on (A, B) doesn't accidentally silence
            // a wrong-path join on (C, B) in the same ON clause — rare
            // but possible when an ON references a third table via a
            // correlated subquery or old-style multi-relation predicate.
            List<BoundJoinPair> pairs = new ArrayList<BoundJoinPair>();
            Set<TablePair> tablePairsWithFkMatch = new HashSet<TablePair>();
            for (int i = 0; i + 1 < joinSources.size(); i += 2) {
                BoundJoinPair bp = bindJoinPair(
                        joinSources.get(i), joinSources.get(i + 1), cat, vendor);
                if (bp == null) {
                    continue;  // Unbound side; can't reason about it.
                }
                bp.matchedFk = matchFkInEitherDirection(bp, cat, vendor);
                if (bp.matchedFk != null) {
                    tablePairsWithFkMatch.add(new TablePair(bp.leftTm.table, bp.rightTm.table));
                }
                pairs.add(bp);
            }

            for (BoundJoinPair bp : pairs) {
                SqlGuardResponse.JoinFact fact = new SqlGuardResponse.JoinFact();
                fact.joinPath = bp.joinPath;
                fact.joinPathMatchesForeignKey = bp.matchedFk != null;

                boolean samePairAlreadyMatched = tablePairsWithFkMatch.contains(
                        new TablePair(bp.leftTm.table, bp.rightTm.table));
                if (bp.matchedFk == null && !samePairAlreadyMatched) {
                    // No FK match anywhere in this ON clause — check whether
                    // an FK between these tables exists and surface a
                    // Did-you-mean.
                    ConstraintModel candidate = anySingleColumnFkBetween(
                            bp.leftTm.table, bp.rightTm.table, cat, vendor);
                    if (candidate == null) {
                        candidate = anySingleColumnFkBetween(
                                bp.rightTm.table, bp.leftTm.table, cat, vendor);
                    }
                    if (candidate != null) {
                        String expected = expectedFkPath(
                                candidate, bp.leftTm.table, bp.rightTm.table);
                        fact.expectedForeignKeyPath = expected;
                        addViolation(r, index, "WRONG_JOIN_PATH", "warn",
                                "Join condition does not match the known foreign key "
                                        + expected + ". Likely incorrect join.",
                                bp.joinPath);
                    }
                }
                r.facts.joins.add(fact);
            }
        }
    }

    /**
     * Catalog-bound representation of one ON-predicate pair. Used for the
     * two-pass scan in {@link #checkWrongJoinPath} — pass 1 builds the
     * list with {@code matchedFk} populated, pass 2 decides JoinFact /
     * violation emission with full knowledge of every pair in the ON.
     */
    private static final class BoundJoinPair {
        final SqlGuardCatalogSamples.TableMatch leftTm;
        final ColumnModel leftCol;
        final SqlGuardCatalogSamples.TableMatch rightTm;
        final ColumnModel rightCol;
        final String joinPath;
        ConstraintModel matchedFk;
        BoundJoinPair(SqlGuardCatalogSamples.TableMatch leftTm, ColumnModel leftCol,
                      SqlGuardCatalogSamples.TableMatch rightTm, ColumnModel rightCol) {
            this.leftTm = leftTm;
            this.leftCol = leftCol;
            this.rightTm = rightTm;
            this.rightCol = rightCol;
            this.joinPath = leftTm.table.name() + "." + leftCol.name()
                    + " = " + rightTm.table.name() + "." + rightCol.name();
        }
    }

    private BoundJoinPair bindJoinPair(sourceColumn left, sourceColumn right,
                                       UnifiedCatalogModel cat, EDbVendor vendor) {
        SqlGuardCatalogSamples.TableMatch leftTm =
                SqlGuardCatalogSamples.findTable(cat, left.getParent_name());
        SqlGuardCatalogSamples.TableMatch rightTm =
                SqlGuardCatalogSamples.findTable(cat, right.getParent_name());
        if (leftTm == null || rightTm == null) return null;
        ColumnModel leftCol = SqlGuardCatalogSamples.findColumn(leftTm, left.getColumn());
        ColumnModel rightCol = SqlGuardCatalogSamples.findColumn(rightTm, right.getColumn());
        if (leftCol == null || rightCol == null) return null;
        return new BoundJoinPair(leftTm, leftCol, rightTm, rightCol);
    }

    private ConstraintModel matchFkInEitherDirection(BoundJoinPair bp,
                                                     UnifiedCatalogModel cat,
                                                     EDbVendor vendor) {
        ConstraintModel m = matchSingleColumnFk(
                bp.leftTm.table, bp.leftCol.name(),
                bp.rightTm, bp.rightCol.name(), cat, vendor);
        if (m != null) return m;
        return matchSingleColumnFk(
                bp.rightTm.table, bp.rightCol.name(),
                bp.leftTm, bp.leftCol.name(), cat, vendor);
    }

    /**
     * Does {@code fkOwner} have a single-column FK from {@code fkColumn}
     * to {@code referencedTm.column}? Returns the matching constraint or
     * null. The referenced side is matched by resolving the FK's
     * {@code referencedTable} string through {@link SqlGuardCatalogSamples#findTable}
     * and comparing {@link TableModel} object identity — same instance,
     * same table. This is the multi-schema-safe comparison for
     * schema-qualified or catalog-qualified FK targets ({@code "hr.dept"},
     * {@code "ORCL.HR.DEPT"}).
     *
     * <p><b>Known limitation</b> — bare FK {@code referencedTable} values
     * (e.g. {@code "dept"} with no schema prefix) resolve via the
     * catalog-wide "first table wins" search in
     * {@link SqlGuardCatalogSamples#findTable}, while
     * {@link gudusoft.gsqlparser.catalog.input.CatalogModelValidator}
     * resolves bare FK targets current-schema-first. Catalog fixtures
     * shipped today are single-schema so this divergence is latent;
     * when a multi-schema sample catalog lands, the right fix is to
     * thread the FK owner's schema through to the resolver. Until then
     * the validator's dangling-reference check is the safety net —
     * an FK whose {@code referencedTable} doesn't resolve in either
     * resolver gets flagged at catalog-load time.</p>
     */
    private ConstraintModel matchSingleColumnFk(TableModel fkOwner,
                                                String fkColumn,
                                                SqlGuardCatalogSamples.TableMatch referencedTm,
                                                String referencedColumn,
                                                UnifiedCatalogModel cat,
                                                EDbVendor vendor) {
        if (fkOwner == null || referencedTm == null) return null;
        for (ConstraintModel cs : fkOwner.constraints()) {
            if (!isForeignKey(cs.type())) continue;
            if (cs.columns().size() != 1 || cs.referencedColumns().size() != 1) continue;
            if (!IdentifierService.areEqualStatic(vendor, ESQLDataObjectType.dotColumn,
                    cs.columns().get(0), fkColumn)) continue;
            SqlGuardCatalogSamples.TableMatch fkTargetTm =
                    SqlGuardCatalogSamples.findTable(cat, cs.referencedTable());
            if (fkTargetTm == null || fkTargetTm.table != referencedTm.table) continue;
            if (!IdentifierService.areEqualStatic(vendor, ESQLDataObjectType.dotColumn,
                    cs.referencedColumns().get(0), referencedColumn)) continue;
            return cs;
        }
        return null;
    }

    /**
     * Any single-column FK on {@code fkOwner} that targets {@code referencedTable}
     * (regardless of which exact columns). Used to drive the "Did-you-mean"
     * expectedForeignKeyPath when the predicate joined the right tables
     * via the wrong columns. Uses TableModel identity for target matching
     * (same multi-schema safety as {@link #matchSingleColumnFk}).
     */
    private ConstraintModel anySingleColumnFkBetween(TableModel fkOwner,
                                                     TableModel referencedTable,
                                                     UnifiedCatalogModel cat,
                                                     EDbVendor vendor) {
        if (fkOwner == null || referencedTable == null) return null;
        for (ConstraintModel cs : fkOwner.constraints()) {
            if (!isForeignKey(cs.type())) continue;
            if (cs.columns().size() != 1 || cs.referencedColumns().size() != 1) continue;
            SqlGuardCatalogSamples.TableMatch fkTargetTm =
                    SqlGuardCatalogSamples.findTable(cat, cs.referencedTable());
            if (fkTargetTm != null && fkTargetTm.table == referencedTable) {
                return cs;
            }
        }
        return null;
    }

    /**
     * Canonical "expected" path string for a {@link SqlGuardResponse.JoinFact},
     * ordered to match the SQL author's join orientation so the BFF Did-you-mean
     * reads naturally ({@code leftTable.fkCol = rightTable.refCol} when the
     * FK lives on the left, vice versa otherwise).
     */
    private String expectedFkPath(ConstraintModel fk, TableModel leftTable, TableModel rightTable) {
        // FK owner is either leftTable or rightTable (anySingleColumnFkBetween
        // only returns FKs whose owner is one of the two joined tables).
        boolean fkOnLeft = fkLivesOnTable(fk, leftTable);
        TableModel owner = fkOnLeft ? leftTable : rightTable;
        TableModel target = fkOnLeft ? rightTable : leftTable;
        String ownerCol = fk.columns().get(0);
        String targetCol = fk.referencedColumns().get(0);
        return owner.name() + "." + ownerCol + " = " + target.name() + "." + targetCol;
    }

    /**
     * Order-independent identity key for a (TableModel, TableModel) pair.
     * Used by {@link #checkWrongJoinPath}'s suppression set: a pair seen
     * as {@code (A, B)} must collide with the same pair seen as
     * {@code (B, A)}.
     *
     * <p>Implementation note: avoids canonicalizing pair order at
     * construction. A previous version sorted by {@link System#identityHashCode}
     * but that breaks under identity-hash collisions — two distinct objects
     * sharing a hash would canonicalize inconsistently between
     * {@code (a,b)} and {@code (b,a)}, so {@code equals} would falsely
     * report inequality. Instead, {@code equals} compares both orderings
     * with {@code ==} (collision-proof — only true object identity
     * matches), and {@code hashCode} XORs the two identity hashes for
     * an order-independent bucket value.</p>
     */
    private static final class TablePair {
        private final TableModel a;
        private final TableModel b;
        TablePair(TableModel a, TableModel b) {
            this.a = a;
            this.b = b;
        }
        @Override public boolean equals(Object o) {
            if (this == o) return true;
            if (!(o instanceof TablePair)) return false;
            TablePair other = (TablePair) o;
            return (a == other.a && b == other.b)
                    || (a == other.b && b == other.a);
        }
        @Override public int hashCode() {
            return System.identityHashCode(a) ^ System.identityHashCode(b);
        }
    }

    private static boolean fkLivesOnTable(ConstraintModel fk, TableModel t) {
        if (t == null) return false;
        // Reference equality is safe because the FK was obtained from one of
        // the joined tables' own constraints() lists in checkWrongJoinPath
        // — same TableModel instance, same constraint instance. The
        // alternative (value-equality on ConstraintModel) would also work
        // but matters only if two distinct ConstraintModel instances
        // happen to be equals() — which doesn't happen in our call paths.
        for (ConstraintModel cs : t.constraints()) {
            if (cs == fk) return true;
        }
        return false;
    }

    private static boolean isForeignKey(String type) {
        if (type == null) return false;
        // ASCII case-insensitive match for FK / FOREIGN KEY / FOREIGN_KEY.
        // Hand-coded to mirror CatalogModelValidator's helper (forbidden-apis
        // bans equalsIgnoreCase inside catalog/**; sqlguard reuses the same
        // discipline for parity).
        return asciiEqualsIgnoreCase(type, "FK")
                || asciiEqualsIgnoreCase(type, "FOREIGN KEY")
                || asciiEqualsIgnoreCase(type, "FOREIGN_KEY");
    }

    private static boolean asciiEqualsIgnoreCase(String a, String b) {
        if (a == null || b == null) return a == b;
        int len = a.length();
        if (b.length() != len) return false;
        for (int i = 0; i < len; i++) {
            char ca = a.charAt(i);
            char cb = b.charAt(i);
            if (ca == cb) continue;
            if (ca >= 'A' && ca <= 'Z') ca = (char) (ca + 32);
            if (cb >= 'A' && cb <= 'Z') cb = (char) (cb + 32);
            if (ca != cb) return false;
        }
        return true;
    }

    /**
     * Pick the top-level resultset that holds the SELECT's projected
     * columns. dlineage names it {@code "RS-1"} (one resultset per
     * statement when we run DataFlowAnalyzer per-SELECT). Intermediate
     * resultsets carry function names ({@code "COUNT"}, {@code "AVG"},
     * {@code "CAST"}) and are NOT the answer here.
     */
    private table topResultset(dataflow df) {
        if (df == null || df.getResultsets() == null) {
            return null;
        }
        for (table rs : df.getResultsets()) {
            if (rs.getName() != null && rs.getName().startsWith("RS-")) {
                return rs;
            }
        }
        return df.getResultsets().isEmpty() ? null : df.getResultsets().get(0);
    }

    /**
     * dlineage attaches a synthetic {@code RelationRows} column to every
     * resultset to carry filter / join / where-clause column references.
     * It is NOT part of the SELECT projection and must be skipped before
     * position-matching against the AST's result columns.
     */
    private boolean isSystemColumn(column c) {
        return c != null && ("RelationRows".equals(c.getName()) || "system".equals(c.getSource()));
    }

    /**
     * Walk the {@code fdd} (data-dependency) relationships from a
     * resultset column down to its leaf source columns, recursing
     * through intermediate function / CAST resultsets. Sets
     * {@code traversedResultset[0]=true} if any hop went through an
     * intermediate, which the caller uses to distinguish {@code direct}
     * from {@code expression} lineage type.
     *
     * <p>{@code fdr} (filter-dependency) and join-condition
     * relationships are not followed — those represent the influence of
     * filter / join predicates on row presence, not the data lineage
     * of the projected column value.
     */
    private void traceLeaves(dataflow df, Set<String> resultsetIds,
                             String parentId, String colName,
                             Set<String> visited, List<sourceColumn> out,
                             boolean[] traversedResultset, EDbVendor vendor) {
        if (parentId == null || colName == null) {
            return;
        }
        String key = parentId + ":" + colName;
        if (!visited.add(key) || visited.size() > 128) {
            return;
        }
        if (df.getRelationships() == null) {
            return;
        }
        for (relationship rel : df.getRelationships()) {
            if (!"fdd".equals(rel.getType())) continue;
            if (rel.getTarget() == null) continue;
            if (!parentId.equals(rel.getTarget().getParent_id())) continue;
            if (!nameEquals(vendor, colName, rel.getTarget().getColumn())) continue;
            if (rel.getSources() == null) continue;
            for (sourceColumn src : rel.getSources()) {
                String srcPid = src.getParent_id();
                if (srcPid != null && resultsetIds.contains(srcPid)) {
                    // Intermediate resultset (function / CAST / subquery).
                    traversedResultset[0] = true;
                    traceLeaves(df, resultsetIds, srcPid, src.getColumn(),
                            visited, out, traversedResultset, vendor);
                } else {
                    out.add(src);
                }
            }
        }
    }

    /**
     * Collect a result column's PARTITION BY / ORDER BY source columns
     * from window aggregates in its expression tree, into two SEPARATE
     * output lists so callers can label exposure correctly.
     *
     * <p>dlineage exposes window-spec columns as {@code fdr} edges with
     * {@code clauseType="selectList"} (partition) and {@code "orderby"}
     * (order) on the windowed function's intermediate resultset. The
     * regular {@link #traceLeaves} skips them because they are not
     * {@code fdd} data-dependency edges. Blueprint §B1 / §B2 require
     * those columns in {@code sourceColumns} because they determine
     * which rows the window aggregate reads — promoting them here keeps
     * the wire shape consistent with the seed's expected output.
     *
     * <p><b>Why two output lists.</b> A PARTITION BY column and an
     * ORDER BY column play different governance roles. PII in
     * {@code PARTITION BY x} is a k-anonymity leak (small partitions
     * identify individuals); PII in {@code ORDER BY x} is a sort-pattern
     * leak (rank reveals attribute order). Sqlguard fires distinct
     * violations and assigns distinct {@code exposure} labels in
     * {@code facts.columns[]} for each role, which requires keeping the
     * two source sets distinguishable. Folding both into one list (as
     * the original B1/B2 implementation did) collapses the signal and
     * led to PII in window partition being mis-attributed as
     * {@code explicit_select} exposure with a misleading
     * {@code SENSITIVE_COLUMN_IN_OUTPUT} violation.
     *
     * <p>Strategy: walk {@code fdd} edges from the result column to find
     * every intermediate (function / CAST / subquery) resultset that
     * feeds it. The window-extra {@code fdr selectList} / {@code orderby}
     * edges live ONLY on intermediate resultsets (empirically verified
     * across dlineage's current behavior — top-level resultsets carry
     * neither clause type for outer ORDER BY / GROUP BY). Collecting on
     * intermediates only avoids over-promoting unrelated outer-clause
     * edges from the top resultset.
     *
     * <p>Window-extra edges are scoped per function-resultset
     * (dlineage emits separate intermediate resultsets per
     * {@code TFunctionCall} with a window spec), so two window calls in
     * one projection don't cross-contaminate — each result-column trace
     * only descends into its own function resultsets. Pinned by
     * {@code separatelyProjectedWindowEdgesDoNotCrossContaminate}.
     *
     * <p>Recursion is the same bounded depth ({@code visited.size() &gt;
     * 128}) as {@link #traceLeaves}; identical visited keying so a
     * single statement can't blow up if dlineage produces a cyclic
     * graph.
     */
    private void traceWindowExtras(dataflow df, Set<String> resultsetIds,
                                    String parentId, String colName,
                                    Set<String> visited,
                                    List<sourceColumn> partitionOut,
                                    List<sourceColumn> orderOut,
                                    EDbVendor vendor) {
        if (parentId == null || colName == null) {
            return;
        }
        String key = parentId + ":" + colName;
        if (!visited.add(key) || visited.size() > 128) {
            return;
        }
        if (df.getRelationships() == null) {
            return;
        }
        // Walk down through fdd edges. Each intermediate resultset
        // encountered is the source of its own window-extra fdr edges
        // (if it's a function resultset for a windowed call); leaf
        // sources (real tables) are ignored here — the regular
        // traceLeaves pass already collected them via fdd.
        for (relationship rel : df.getRelationships()) {
            if (!"fdd".equals(rel.getType())) continue;
            if (rel.getTarget() == null) continue;
            if (!parentId.equals(rel.getTarget().getParent_id())) continue;
            if (!nameEquals(vendor, colName, rel.getTarget().getColumn())) continue;
            if (rel.getSources() == null) continue;
            for (sourceColumn src : rel.getSources()) {
                String srcPid = src.getParent_id();
                String srcCol = src.getColumn();
                if (srcPid == null) continue;
                if (resultsetIds.contains(srcPid)) {
                    collectWindowExtras(df, srcPid, srcCol, partitionOut, orderOut, vendor);
                    traceWindowExtras(df, resultsetIds, srcPid, srcCol,
                            visited, partitionOut, orderOut, vendor);
                }
            }
        }
    }

    /**
     * Append {@code fdr} edges on ({@code parentId}, {@code colName})
     * to the right output list per {@code clauseType}:
     * {@code "selectList"} → {@code partitionOut} (PARTITION BY),
     * {@code "orderby"} → {@code orderOut} (ORDER BY). Helper for
     * {@link #traceWindowExtras}.
     *
     * <p><b>Scoping</b> is intentionally both {@code parent_id}-keyed
     * AND target-column-keyed. The {@code parent_id} alone is not
     * sufficient on multi-column intermediates (subquery resultsets
     * carry one column per inner projection): if a sibling column on
     * the same intermediate happens to bear an unrelated
     * {@code selectList} / {@code orderby} fdr edge (a future dlineage
     * shape we can't rule out today), the per-column scope keeps it
     * out of the current trace's extras.
     *
     * <p>Empirically (current dlineage), only function-call resultsets
     * ever emit these clause types and they're scoped to the single
     * function column — so {@code parent_id} alone would also be
     * correct today. The tighter gate is defensive against the gap
     * dlineage may close inconsistently (MantisBT #4467) and pinned
     * by {@code separatelyProjectedWindowEdgesDoNotCrossContaminate}.
     */
    private void collectWindowExtras(dataflow df, String parentId, String colName,
                                      List<sourceColumn> partitionOut,
                                      List<sourceColumn> orderOut,
                                      EDbVendor vendor) {
        if (df.getRelationships() == null) return;
        for (relationship rel : df.getRelationships()) {
            if (!"fdr".equals(rel.getType())) continue;
            if (rel.getTarget() == null) continue;
            if (!parentId.equals(rel.getTarget().getParent_id())) continue;
            if (colName != null
                    && !nameEquals(vendor, colName, rel.getTarget().getColumn())) {
                continue;
            }
            if (rel.getSources() == null) continue;
            for (sourceColumn src : rel.getSources()) {
                String clause = src.getClauseType();
                if ("selectList".equals(clause)) {
                    partitionOut.add(src);
                } else if ("orderby".equals(clause)) {
                    orderOut.add(src);
                }
            }
        }
    }

    /**
     * Identifier-aware equality on column-class identifiers from
     * dlineage's relationship model. Dlineage preserves the
     * user-written casing on `parent_name` / `column` fields; this
     * comparison routes through {@link IdentifierService} so the
     * answer is correct under any dialect's case-folding rules.
     */
    private static boolean nameEquals(EDbVendor vendor, String a, String b) {
        if (a == null) return b == null;
        if (b == null) return false;
        return IdentifierService.areEqualStatic(vendor, ESQLDataObjectType.dotColumn, a, b);
    }

    /**
     * Run {@link DataFlowAnalyzer} on a single SELECT statement and
     * return the in-memory {@code dataflow} model. Emits diagnostics
     * to the response for downstream observability — the wire contract
     * preserves the {@code DLINEAGE_*} markers that pre-dated this
     * refactor so existing consumers continue to see them.
     */
    private dataflow runDlineage(TSelectSqlStatement sel, EDbVendor vendor,
                                  SqlGuardResponse r, int index) {
        try {
            Option option = new Option();
            option.setVendor(vendor);
            option.setSimpleOutput(false);
            option.setOutput(false);
            DataFlowAnalyzer da = new DataFlowAnalyzer(sel.toScript(), option);
            da.generateDataFlow();
            dataflow df = da.getDataFlow();
            if (df != null && df.getRelationships() != null && !df.getRelationships().isEmpty()) {
                r.facts.diagnostics.add("DLINEAGE_RELATIONSHIPS_PRESENT:statement=" + index);
            } else {
                r.facts.diagnostics.add("DLINEAGE_NO_RELATIONSHIP:statement=" + index);
            }
            return df;
        } catch (Exception e) {
            r.facts.diagnostics.add("DLINEAGE_UNAVAILABLE:statement=" + index);
            return null;
        }
    }

    private void expandStar(int index, SqlGuardResponse r, UnifiedCatalogModel cat,
                             String tableName, EDbVendor vendor) {
        SqlGuardCatalogSamples.TableMatch tm =
                SqlGuardCatalogSamples.findTable(cat, tableName);
        if (tm == null) {
            return;
        }
        for (ColumnModel c : tm.table.columns()) {
            List<String> tagNames = SqlGuardCatalogSamples.policyTagNames(c);
            addColumnFact(r, tm.table.name(), c.name(), tagNames, "implicit_select_star", vendor);
            addLineage(r, index, "result." + c.name(),
                    Collections.singletonList(tm.schema + "." + tm.table.name() + "." + c.name()), "direct");
        }
        addViolation(r, index, "SELECT_STAR_ON_SENSITIVE_TABLE", "warn",
                "SELECT * may expose sensitive columns.", "sensitive columns are present in selected table");
    }

    /**
     * Collect the catalog-bound canonical forms
     * ({@code schema.table.column}) of column references that appear in
     * the result-column expression OUTSIDE any window
     * {@code OVER (...)} spec. These are the columns the SQL genuinely
     * projects — function arguments, scalar operator operands
     * ({@code email || '...'}), CASE branches, COALESCE wrappers, etc.
     * Anything inside a {@link TWindowDef} / {@link TWindowSpecification}
     * (PARTITION BY / ORDER BY / frame) is skipped because those columns
     * are the partition/order role's job to surface, not the
     * projection's.
     *
     * <p>The role-precedence pass in {@code analyzeSelect} uses this set
     * to distinguish "column is really a projection dependency"
     * (legitimate dual-role like {@code email || COUNT(*) OVER
     * (PARTITION BY email)} where the bare {@code email} ref is a real
     * projection source) from "dlineage over-attributed the column to
     * an fdd edge" (COUNT(*) shape where the partition column leaks
     * into the function's fdd edge but is not a real projection
     * source).
     *
     * <p>Bound via the same {@link SqlGuardCatalogSamples} helpers as
     * the dlineage-source binding pass, so the comparison in
     * {@link #stripOverAttributedSources} is a clean Set difference on
     * canonical strings — qualifier-aware and dialect-aware. Codex
     * round-1 review (B5 work, 2026-05-20) flagged the previous
     * last-segment comparison as fragile under joins where two tables
     * share a column name; canonical-form comparison closes that gap.
     */
    private Set<String> collectAstProjectionDependencies(TResultColumn rc,
                                                          UnifiedCatalogModel cat,
                                                          EDbVendor vendor) {
        if (rc == null || rc.getExpr() == null) {
            return Collections.emptySet();
        }
        ProjectionDependencyCollector v = new ProjectionDependencyCollector(cat, vendor);
        rc.getExpr().acceptChildren(v);
        return v.canonicalRefs;
    }

    /**
     * AST visitor that gathers catalog-canonical references to columns
     * that appear in the result expression OUTSIDE any window spec.
     * Tracks an {@code insideWindow} depth counter so PARTITION BY /
     * ORDER BY / frame subtrees are properly skipped — even when nested
     * inside other expressions (e.g. {@code CASE WHEN COUNT(*) OVER
     * (PARTITION BY ...) > 1 THEN ...}).
     *
     * <p>Uses {@link TObjectName#getSourceTable()} (resolver-supplied)
     * + {@link TObjectName#getColumnNameOnly()} to build the canonical
     * form via {@link SqlGuardCatalogSamples#findTable} /
     * {@link SqlGuardCatalogSamples#findColumn} — same path the dlineage-
     * source catalog binding uses. References that fail to resolve are
     * silently dropped (the rule still fires correctly for resolved
     * columns; unresolved refs aren't catalog-bound anywhere in
     * sqlguard).
     */
    private static final class ProjectionDependencyCollector extends TParseTreeVisitor {
        final Set<String> canonicalRefs = new LinkedHashSet<String>();
        private int insideWindow = 0;
        private final UnifiedCatalogModel cat;
        private final EDbVendor vendor;

        ProjectionDependencyCollector(UnifiedCatalogModel cat, EDbVendor vendor) {
            this.cat = cat;
            this.vendor = vendor;
        }

        @Override
        public void preVisit(gudusoft.gsqlparser.nodes.TWindowDef node) { insideWindow++; }

        @Override
        public void postVisit(gudusoft.gsqlparser.nodes.TWindowDef node) { insideWindow--; }

        @Override
        public void preVisit(gudusoft.gsqlparser.nodes.TWindowSpecification node) { insideWindow++; }

        @Override
        public void postVisit(gudusoft.gsqlparser.nodes.TWindowSpecification node) { insideWindow--; }

        @Override
        public void preVisit(TObjectName node) {
            if (insideWindow > 0 || node == null) return;
            // Skip {@code *} and unresolved refs — they can't catalog-bind.
            String col = node.getColumnNameOnly();
            if (col == null || col.isEmpty() || "*".equals(col)) return;
            gudusoft.gsqlparser.nodes.TTable tbl = node.getSourceTable();
            if (tbl == null || tbl.getTableName() == null) return;
            String tableRef = tbl.getTableName().toString();
            String tableKey = IdentifierService.normalizeStatic(
                    vendor, ESQLDataObjectType.dotTable, tableRef);
            SqlGuardCatalogSamples.TableMatch tm =
                    SqlGuardCatalogSamples.findTable(cat, tableKey);
            ColumnModel c = SqlGuardCatalogSamples.findColumn(tm, col);
            if (tm != null && c != null) {
                canonicalRefs.add(tm.schema + "." + tm.table.name() + "." + c.name());
            }
        }
    }

    /**
     * Strip from {@code functionArgSources} any column that ALSO
     * appears in {@code partitionSources} / {@code orderSources} AND
     * is NOT a real AST projection dependency (per
     * {@code astProjectionDeps}). The remaining columns in
     * {@code functionArgSources} are exactly those genuinely referenced
     * as function arguments or scalar operands in the projection —
     * passthrough patterns like {@code LAG(email)} or {@code email ||
     * COUNT(*) OVER (PARTITION BY email)} preserve their "appears in
     * output" role even when the same column is also a partition key.
     */
    private void stripOverAttributedSources(LinkedHashSet<String> functionArgSources,
                                             LinkedHashSet<String> partitionSources,
                                             LinkedHashSet<String> orderSources,
                                             Set<String> astProjectionDeps) {
        Iterator<String> it = functionArgSources.iterator();
        while (it.hasNext()) {
            String fq = it.next();
            boolean overlapsPartition = partitionSources.contains(fq);
            boolean overlapsOrder = orderSources.contains(fq);
            if (!overlapsPartition && !overlapsOrder) continue;
            if (!astProjectionDeps.contains(fq)) {
                // dlineage over-attributed this column to fdd; partition/order is
                // the only real role.
                it.remove();
            }
        }
    }

    /**
     * Catalog-bind a list of dlineage {@link sourceColumn}s into the
     * canonical {@code "schema.table.column"} wire form, dropping any
     * leaf that doesn't resolve to a known catalog column. Preserves
     * insertion order and dedups via {@link LinkedHashSet}.
     *
     * <p>Identical resolution logic to the original inline pass in
     * {@code analyzeSelect} — factored out so the per-role buckets
     * (function-arg leaves, partition extras, order extras) can each
     * run it without code duplication.
     */
    private LinkedHashSet<String> bindToCatalog(List<sourceColumn> leaves,
                                                 UnifiedCatalogModel cat,
                                                 EDbVendor vendor) {
        LinkedHashSet<String> out = new LinkedHashSet<String>();
        for (sourceColumn leaf : leaves) {
            String tableKey = leaf.getParent_name() == null
                    ? null : IdentifierService.normalizeStatic(
                            vendor, ESQLDataObjectType.dotTable, leaf.getParent_name());
            String colName = leaf.getColumn();
            if (tableKey == null || colName == null) continue;
            SqlGuardCatalogSamples.TableMatch tm =
                    SqlGuardCatalogSamples.findTable(cat, tableKey);
            ColumnModel c = SqlGuardCatalogSamples.findColumn(tm, colName);
            if (tm != null && c != null) {
                out.add(tm.schema + "." + tm.table.name() + "." + c.name());
            }
        }
        return out;
    }

    private void addColumnFactForSource(SqlGuardResponse r, String fq, String exposure,
                                         UnifiedCatalogModel cat, EDbVendor vendor) {
        String[] p = fq.split("\\.");
        if (p.length < 3) {
            return;
        }
        SqlGuardCatalogSamples.TableMatch tm =
                SqlGuardCatalogSamples.findTable(cat, p[1]);
        ColumnModel c = SqlGuardCatalogSamples.findColumn(tm, p[2]);
        addColumnFact(r, p[1], p[2], SqlGuardCatalogSamples.policyTagNames(c), exposure, vendor);
    }

    private void addColumnFact(SqlGuardResponse r, String table, String name, List<String> tags,
                                String exposure, EDbVendor vendor) {
        for (SqlGuardResponse.ColumnFact cf : r.facts.columns) {
            // Identifier-aware dedup: a fact for (sbcustomer, sbcustcountry,
            // explicit_select) is the same regardless of whether the inputs
            // were folded by dialect-specific rules. Exposure is a sqlguard
            // tag (`explicit_select` / `implicit_select_star` /
            // `window_partition` / `window_order`), not an SQL identifier,
            // so plain .equals(). A column that appears in BOTH a projection
            // AND a window PARTITION BY produces two separate facts (one
            // per role) so the dual-exposure reality is visible to the BFF.
            if (IdentifierService.areEqualStatic(vendor, ESQLDataObjectType.dotTable, cf.table, table)
                    && IdentifierService.areEqualStatic(vendor, ESQLDataObjectType.dotColumn, cf.name, name)
                    && cf.exposure.equals(exposure)) {
                mergeTags(cf.policyTags, tags);
                return;
            }
        }
        SqlGuardResponse.ColumnFact cf = new SqlGuardResponse.ColumnFact();
        cf.table = table;
        cf.name = name;
        cf.policyTags.addAll(tags);
        cf.exposure = exposure;
        r.facts.columns.add(cf);
        if (isSensitive(tags)) {
            emitSensitiveViolation(r, exposure, table, name);
        }
    }

    /**
     * Dispatch the right "sensitive column" violation code based on how
     * the column is exposed in the SQL. Each role gets its own rule so
     * the BFF / audit log can distinguish "PII in the output" from
     * "PII in a window partition" (k-anonymity leak) from "PII in a
     * window ORDER BY" (sort-pattern leak).
     *
     * <ul>
     *   <li>{@code explicit_select} / {@code implicit_select_star} —
     *       {@code SENSITIVE_COLUMN_IN_OUTPUT} (existing, blueprint
     *       §S21).</li>
     *   <li>{@code window_partition} —
     *       {@code SENSITIVE_COLUMN_IN_WINDOW_PARTITION} (new,
     *       blueprint §1.5.B B5).</li>
     *   <li>{@code window_order} — no violation today. The blueprint
     *       does not pin a window-ORDER rule; PII in ORDER BY is a
     *       softer leak (rank reveals attribute order, not partition
     *       membership). Surface only as a column fact for now;
     *       a future {@code SENSITIVE_COLUMN_IN_WINDOW_ORDER} can ship
     *       once a seed needs it.</li>
     * </ul>
     */
    private void emitSensitiveViolation(SqlGuardResponse r, String exposure,
                                         String table, String name) {
        int idx = currentStatement(r);
        if ("window_partition".equals(exposure)) {
            addViolation(r, idx, "SENSITIVE_COLUMN_IN_WINDOW_PARTITION", "warn",
                    "Sensitive column appears in window PARTITION BY; "
                            + "small partitions may identify individuals.",
                    table + "." + name);
            return;
        }
        if ("window_order".equals(exposure)) {
            // Reserved exposure label — no violation today; see Javadoc.
            return;
        }
        // Default: explicit_select / implicit_select_star → in-output rule.
        addViolation(r, idx, "SENSITIVE_COLUMN_IN_OUTPUT", "warn",
                "Sensitive column appears in query output.",
                table + "." + name);
    }

    private void mergeTags(List<String> existing, List<String> incoming) {
        for (String tag : incoming) {
            if (!existing.contains(tag)) {
                existing.add(tag);
            }
        }
    }

    /**
     * Policy gate for {@code SENSITIVE_COLUMN_IN_OUTPUT}.
     *
     * <p>Triggers on {@code PII} or {@code RESTRICTED} only. {@code
     * FINANCIAL} is left out deliberately: aggregating a financial
     * column (e.g. {@code SUM(amount)}) hides individual values and
     * is structurally different from leaking SSN / PII. The catalog
     * still carries the {@code FINANCIAL} tag for downstream
     * lineage / audit consumers, and a future {@code info}-severity
     * tier (blueprint §3.3) is the right home for non-blocking
     * FINANCIAL exposure surfacing — at which point this rule can
     * fan out per severity rather than per tag.</p>
     */
    private boolean isSensitive(List<String> tags) {
        return tags.contains("PII") || tags.contains("RESTRICTED");
    }

    private int currentStatement(SqlGuardResponse r) {
        return r.statements.isEmpty() ? 0 : r.statements.get(r.statements.size() - 1).index;
    }

    private void addLineage(SqlGuardResponse r, int idx, String target, List<String> sources, String type) {
        addLineage(r, idx, target, sources, type, null, null);
    }

    private void addLineage(SqlGuardResponse r, int idx, String target, List<String> sources, String type,
                            List<String> aggregateFunctions,
                            List<SqlGuardResponse.WindowFn> windowFunctions) {
        for (SqlGuardResponse.LineageEdge existing : r.facts.lineage) {
            if (existing.statementIndex == idx
                    && existing.targetColumn.equals(target)
                    && existing.sourceColumns.equals(sources)
                    && existing.lineageType.equals(type)
                    && listEquals(existing.aggregateFunctions, aggregateFunctions)
                    && windowListEquals(existing.windowFunctions, windowFunctions)) {
                return;
            }
        }
        SqlGuardResponse.LineageEdge e = new SqlGuardResponse.LineageEdge();
        e.statementIndex = idx;
        e.targetColumn = target;
        e.sourceColumns.addAll(sources);
        e.lineageType = type;
        e.aggregateFunctions = aggregateFunctions;
        e.windowFunctions = windowFunctions;
        r.facts.lineage.add(e);
    }

    private static boolean listEquals(List<String> a, List<String> b) {
        return a == null ? b == null : a.equals(b);
    }

    /**
     * Structural equality on {@code List<WindowFn>}. We don't make
     * {@code WindowFn} / {@code OrderByItem} implement {@code equals}
     * because the dedup compare is the only caller that needs it; an
     * implicit {@code equals} on a public wire-format POJO would invite
     * downstream consumers to rely on it.
     */
    private static boolean windowListEquals(List<SqlGuardResponse.WindowFn> a,
                                             List<SqlGuardResponse.WindowFn> b) {
        if (a == null) return b == null;
        if (b == null) return false;
        if (a.size() != b.size()) return false;
        for (int i = 0; i < a.size(); i++) {
            SqlGuardResponse.WindowFn x = a.get(i);
            SqlGuardResponse.WindowFn y = b.get(i);
            if (!eq(x.functionName, y.functionName)) return false;
            if (!eq(x.partitionBy, y.partitionBy)) return false;
            if (!eq(x.frame, y.frame)) return false;
            if (x.orderBy.size() != y.orderBy.size()) return false;
            for (int j = 0; j < x.orderBy.size(); j++) {
                SqlGuardResponse.OrderByItem ox = x.orderBy.get(j);
                SqlGuardResponse.OrderByItem oy = y.orderBy.get(j);
                if (!eq(ox.column, oy.column)) return false;
                if (!eq(ox.direction, oy.direction)) return false;
            }
        }
        return true;
    }

    private static boolean eq(Object a, Object b) {
        return a == null ? b == null : a.equals(b);
    }

    private String resultExpressionText(TResultColumn rc) {
        if (rc.getExpr() != null) {
            return rc.getExpr().toString().trim();
        }
        return rc.toString().trim();
    }

    /**
     * Closed set of scalar aggregate function names this worker
     * recognizes. Lowercased so the case-insensitive comparison in the
     * visitor is a single set lookup. The list intentionally mirrors
     * the SQL-92 core aggregates ({@code COUNT}, {@code SUM},
     * {@code AVG}, {@code MIN}, {@code MAX}) — extending the set
     * (e.g. {@code STDDEV}, {@code VAR_POP}, {@code PERCENTILE_CONT},
     * vendor-specific {@code GROUP_CONCAT}) is deferred until a seed
     * needs them, so the schema for the extended shapes can be
     * designed against concrete SQL rather than speculatively.
     *
     * <p>Only consulted by the scalar {@link AggregateCallCollector};
     * {@link WindowCallCollector} (blueprint §B1/B2) accepts any
     * function call carrying a window spec, regardless of whether the
     * function appears in this set — {@code LAG}, {@code LEAD},
     * {@code ROW_NUMBER}, etc. are window-only and would never make
     * sense in {@code AGGREGATE_FUNCTIONS}.
     */
    private static final Set<String> AGGREGATE_FUNCTIONS;
    static {
        Set<String> s = new HashSet<String>();
        s.add("count");
        s.add("sum");
        s.add("avg");
        s.add("min");
        s.add("max");
        AGGREGATE_FUNCTIONS = Collections.unmodifiableSet(s);
    }

    /**
     * Derive {@code lineageType} from the pre-computed aggregate-function
     * list. Aggregate edges are anything with at least one detected
     * function call; non-aggregate edges fall through to the
     * bare-identifier check that distinguishes {@code direct} from
     * {@code expression}.
     *
     * <p>The {@code direct} vs {@code expression} discriminator still
     * inspects the textual form because both shapes carry the same
     * lineage edge topology — telling them apart needs only "is this
     * the literal name of one column?", which a regex answers cleanly
     * without the SQL identifier extraction failure modes that motivate
     * the AST-walking aggregate detector. See
     * {@code sqlguard/CLAUDE.md} for the rule.
     */
    private String lineageType(String text, List<String> aggregateFunctions) {
        if (aggregateFunctions != null && !aggregateFunctions.isEmpty()) {
            return "aggregate";
        }
        String l = text.toLowerCase(Locale.ROOT);
        return l.matches(identifierPattern() + "(\\." + identifierPattern() + "){0,2}") ? "direct" : "expression";
    }

    /**
     * Walk the AST under a result-column expression and collect every
     * aggregate function call, in document order, deduplicated. The
     * return shape is the same as the public {@code aggregateFunctions}
     * field on {@link SqlGuardResponse.LineageEdge}: {@code null} when
     * no aggregate call is detected (so the JSON envelope omits the
     * key on direct/expression edges) or a list of uppercase function
     * names — bare {@code "COUNT"} / {@code "AVG"} or
     * {@code "FUNC(DISTINCT)"} when the call uses the DISTINCT modifier.
     *
     * <p>Why AST instead of regex: SQL identifier extraction is the
     * parser's job, not the worker's. The parser has already resolved
     * whitespace ({@code COUNT (DISTINCT x)}), block comments
     * ({@code COUNT}/&#42;hint&#42;/{@code (DISTINCT x)}), single-quoted
     * literals (a literal {@code 'count(x)'} is a {@link TConstant},
     * not a {@link TFunctionCall}), and the DISTINCT modifier (exposed
     * as {@link EAggregateType}) by the time the AST exists. Walking
     * the AST trades one regex + one literal-stripping pattern +
     * special-case handling for one visitor pattern that is correct
     * by construction across every dialect GSP supports. See
     * {@code sqlguard/CLAUDE.md} "Identifier extraction rule" for the
     * project-wide rule.
     *
     * <p>Scope: scalar aggregates only. Window aggregates
     * ({@code AVG(x) OVER (...)}) live on the sibling
     * {@link #windowFunctionsFromAst} visitor (blueprint §1.5.B B1/B2,
     * shipped 2026-05-20); FILTER aggregates ({@code COUNT(x) FILTER
     * (WHERE ...)}) are still deferred to MantisBT #4468. We detect
     * those forms by peeking at
     * {@link TFunctionCall#getWindowSpecification()} /
     * {@link TFunctionCall#getWindowDef()} /
     * {@link TFunctionCall#getFilterClause()}; when any is non-null we
     * skip the node instead of misclassifying it as a scalar aggregate.
     *
     * <p>Blueprint §S2.
     */
    private List<String> aggregateFunctionsFromAst(TResultColumn rc) {
        if (rc == null || rc.getExpr() == null) {
            return null;
        }
        AggregateCallCollector collector = new AggregateCallCollector();
        rc.getExpr().acceptChildren(collector);
        return collector.functions.isEmpty() ? null : collector.functions;
    }

    /**
     * Walk the AST under a result-column expression and collect every
     * window-function call (any {@link TFunctionCall} carrying a non-null
     * {@link TFunctionCall#getWindowDef()} or
     * {@link TFunctionCall#getWindowSpecification()}). Activated by
     * blueprint §1.5.B B1 / B2.
     *
     * <p>Returns {@code null} when no window call is found so the JSON
     * envelope can omit the {@code windowFunctions} key on non-window
     * edges (preserves wire-format compatibility — see
     * {@link SqlGuardResponse.LineageEdge#windowFunctions} Javadoc).
     *
     * <p>This is the L3 AST-fallback per
     * {@code sqlguard/CLAUDE.md} "Lineage extraction rule": dlineage
     * does not surface window-spec metadata as typed fields
     * (MantisBT #4467). When that lands, the visitor migrates to
     * {@code relationship.getWindowSpec()} without changing the wire
     * shape.
     */
    private List<SqlGuardResponse.WindowFn> windowFunctionsFromAst(TResultColumn rc) {
        if (rc == null || rc.getExpr() == null) {
            return null;
        }
        WindowCallCollector collector = new WindowCallCollector();
        rc.getExpr().acceptChildren(collector);
        return collector.functions.isEmpty() ? null : collector.functions;
    }

    /**
     * AST visitor that collects window-function calls and their
     * partition / order / frame metadata. Mirrors
     * {@link AggregateCallCollector}'s shape but inspects the window
     * portion of {@link TFunctionCall} instead of the aggregate
     * modifier.
     *
     * <p>Recursion: {@code acceptChildren} on the root expression
     * descends into nested arguments, so a window call inside e.g.
     * {@code COALESCE(SUM(x) OVER (...), 0)} or
     * {@code CASE WHEN LAG(x) OVER (...) = y THEN ... END} still
     * surfaces.
     *
     * <p><b>Named-window references (e.g. {@code OVER w} with
     * {@code WINDOW w AS (PARTITION BY ...)})</b> are partially
     * supported: the visitor produces a {@code WindowFn} (so
     * {@code lineageType="window"} still fires) but with empty
     * {@code partitionBy} / {@code orderBy} and {@code frame=null}
     * because those clauses live on the SELECT's {@code WINDOW} clause
     * definition, not on the inline {@code OVER} reference. Resolving
     * the reference requires walking
     * {@link TSelectSqlStatement#getWindowClause()} which is out of
     * scope for this pass; flagged for follow-up if a seed needs it.
     *
     * <p><b>Partition / order expression text</b> is captured via
     * {@code TExpression.toString()} — so {@code PARTITION BY date_col
     * + 1} produces {@code partitionBy=["date_col + 1"]} rather than
     * extracting the {@code TObjectName} children. This matches the
     * field's contract ("raw column expressions as written"). The
     * catalog-bound source columns for that expression still flow
     * through dlineage's {@code fdr} edges into
     * {@link SqlGuardResponse.LineageEdge#sourceColumns}.
     */
    private static final class WindowCallCollector extends TParseTreeVisitor {
        final List<SqlGuardResponse.WindowFn> functions = new ArrayList<SqlGuardResponse.WindowFn>();

        @Override
        public void preVisit(TFunctionCall node) {
            if (node == null || node.getFunctionName() == null) {
                return;
            }
            TWindowDef wdef = node.getWindowDef();
            TWindowSpecification wspec = node.getWindowSpecification();
            if (wdef == null && wspec == null) {
                return;
            }
            String raw = node.getFunctionName().toString();
            if (raw == null) {
                return;
            }
            String upper = SqlGuardCatalogSamples.strip(raw).toUpperCase(Locale.ROOT);
            EAggregateType agg = node.getAggregateType();
            String entry = agg == EAggregateType.distinct ? upper + "(DISTINCT)" : upper;

            SqlGuardResponse.WindowFn wf = new SqlGuardResponse.WindowFn();
            wf.functionName = entry;

            TPartitionClause part = wdef != null ? wdef.getPartitionClause()
                    : wspec.getPartitionClause();
            if (part != null && part.getExpressionList() != null) {
                TExpressionList el = part.getExpressionList();
                for (int i = 0; i < el.size(); i++) {
                    TExpression e = el.getExpression(i);
                    if (e == null) continue;
                    String t = e.toString();
                    if (t != null && t.length() > 0) {
                        wf.partitionBy.add(t.toLowerCase(Locale.ROOT));
                    }
                }
            }

            TOrderBy ob = wdef != null ? wdef.getOrderBy() : wspec.getOrderBy();
            if (ob != null && ob.getItems() != null) {
                for (int i = 0; i < ob.getItems().size(); i++) {
                    TOrderByItem item = ob.getItems().getOrderByItem(i);
                    if (item == null || item.getSortKey() == null) continue;
                    SqlGuardResponse.OrderByItem out = new SqlGuardResponse.OrderByItem();
                    out.column = item.getSortKey().toString().toLowerCase(Locale.ROOT);
                    ESortType sort = item.getSortOrder();
                    if (sort == ESortType.asc) {
                        out.direction = "ASC";
                    } else if (sort == ESortType.desc) {
                        out.direction = "DESC";
                    } else {
                        out.direction = "NONE";
                    }
                    wf.orderBy.add(out);
                }
            }

            TWindowFrame frame = wdef != null ? wdef.getWindowFrame()
                    : wspec.getWindowFrame();
            if (frame != null) {
                String ft = frame.toString();
                if (ft != null && !ft.trim().isEmpty()) {
                    wf.frame = ft.trim();
                }
            }

            functions.add(wf);
        }
    }

    /**
     * AST visitor that collects scalar aggregate calls. Window / FILTER
     * forms are intentionally skipped (see Javadoc on
     * {@link #aggregateFunctionsFromAst}). The visitor uses
     * {@code preVisit(TFunctionCall)} which is the standard hook from
     * {@link TParseTreeVisitor}; recursion through nested arguments
     * happens via {@code acceptChildren} on the root expression, so a
     * call inside {@code COALESCE(SUM(a), AVG(b))} surfaces every
     * aggregate without manual traversal.
     */
    private static final class AggregateCallCollector extends TParseTreeVisitor {
        final List<String> functions = new ArrayList<String>();

        @Override
        public void preVisit(TFunctionCall node) {
            if (node == null || node.getFunctionName() == null) {
                return;
            }
            // Skip window aggregates (OVER (...)) and FILTER (WHERE …)
            // forms — they're separate lineage shapes owned by Tier 2.
            if (node.getWindowSpecification() != null || node.getWindowDef() != null
                    || node.getFilterClause() != null) {
                return;
            }
            String raw = node.getFunctionName().toString();
            if (raw == null) {
                return;
            }
            // Closed-set membership against well-known SQL aggregate
            // function names. NOT an SQL identifier comparison in the
            // sqlguard/CLAUDE.md "Identifier comparison rule" sense:
            // COUNT/SUM/AVG/MIN/MAX are built-in function names whose
            // identity is case-insensitive in every supported dialect.
            // Plain .toLowerCase() is correct and equivalent to
            // areEqualStatic for any vendor on these specific tokens.
            String name = SqlGuardCatalogSamples.strip(raw).toLowerCase(Locale.ROOT);
            if (!AGGREGATE_FUNCTIONS.contains(name)) {
                return;
            }
            String upper = name.toUpperCase(Locale.ROOT);
            // EAggregateType: none, all, distinct, unique. `all` and
            // `unique` fold into the bare form — `ALL` is SQL's default
            // and `UNIQUE` is an Oracle synonym for DISTINCT in the
            // grammar but isn't semantically meaningful enough to be
            // worth a third surface form in the worker contract.
            EAggregateType agg = node.getAggregateType();
            String entry = agg == EAggregateType.distinct ? upper + "(DISTINCT)" : upper;
            if (!functions.contains(entry)) {
                functions.add(entry);
            }
        }
    }

    private String identifierPattern() {
        return "([a-zA-Z_][\\w]*|\\\"[^\\\"]+\\\"|`[^`]+`)";
    }

    /**
     * Last dot-segment of a table reference, with quoting stripped.
     * Used for the wire-form short name on {@code r.facts.tables[].name}
     * and {@code scope.tableNames} — both of which intentionally drop
     * the schema/catalog qualifier. Callers that need the qualifier
     * (catalog lookup) should use {@link #qualifiedTableRef} instead so
     * {@link SqlGuardCatalogSamples#findTable} can match against the
     * schema name.
     */
    private String simple(String s) {
        String qualified = qualifiedTableRef(s);
        if (qualified.isEmpty()) {
            return "";
        }
        if (qualified.indexOf('.') >= 0) {
            qualified = qualified.substring(qualified.lastIndexOf('.') + 1);
        }
        return SqlGuardCatalogSamples.strip(qualified);
    }

    /**
     * Cleaned but still-qualified table reference (e.g.
     * {@code "public.sbcustomer"}). Strips outer quoting, trailing
     * punctuation, and any subquery / aliased fragments while preserving
     * the schema/catalog segments so
     * {@link SqlGuardCatalogSamples#findTable} can resolve the right
     * schema instead of guessing.
     */
    private String qualifiedTableRef(String s) {
        s = SqlGuardCatalogSamples.strip(s);
        if (s == null) {
            return "";
        }
        s = s.trim().replaceAll("[;,]$", "");
        if (s.indexOf('(') >= 0) {
            return "";
        }
        if (s.indexOf(' ') >= 0) {
            s = s.substring(0, s.indexOf(' '));
        }
        return s;
    }

    private void addViolation(SqlGuardResponse r, int idx, String code, String sev, String msg, String ev) {
        for (SqlGuardResponse.Violation v : r.violations) {
            if (v.statementIndex == idx && v.code.equals(code) && String.valueOf(v.evidence).equals(ev)) {
                return;
            }
        }
        SqlGuardResponse.Violation v = new SqlGuardResponse.Violation();
        v.statementIndex = idx;
        v.code = code;
        v.severity = sev;
        v.message = msg;
        v.evidence = ev;
        r.violations.add(v);
        if (code.equals("SELECT_STAR_ON_SENSITIVE_TABLE")) {
            r.suggestions.add("Replace SELECT * with explicit non-sensitive columns.");
        }
    }

    private void applyStatementDecision(SqlGuardResponse.StatementResult sr, List<SqlGuardResponse.Violation> vs, int idx) {
        int maxRisk = 0;
        for (SqlGuardResponse.Violation v : vs) {
            if (v.statementIndex == idx && "error".equals(v.severity)) {
                sr.decision = "DENY";
                sr.riskScore = 90;
                sr.summary = "SQL denied by policy.";
                return;
            }
            if (v.statementIndex == idx) {
                maxRisk = Math.max(maxRisk, "warn".equals(v.severity) ? 60 : 30);
            }
        }
        if (maxRisk > 0) {
            sr.decision = "WARN";
            sr.riskScore = maxRisk;
            sr.summary = "SQL has policy warnings.";
        }
    }

    private void merge(SqlGuardResponse r) {
        for (SqlGuardResponse.StatementResult s : r.statements) {
            r.riskScore = Math.max(r.riskScore, s.riskScore);
            if ("DENY".equals(s.decision)) {
                r.decision = "DENY";
                r.summary = "One or more statements are denied.";
            } else if (!"DENY".equals(r.decision) && "WARN".equals(s.decision)) {
                r.decision = "WARN";
                r.summary = "One or more statements have warnings.";
            }
        }
    }
}