diff --git a/cli/internal/conformity/catalogue_test.go b/cli/internal/conformity/catalogue_test.go
index f85e4e86..02e72ff6 100644
--- a/cli/internal/conformity/catalogue_test.go
+++ b/cli/internal/conformity/catalogue_test.go
@@ -64,11 +64,15 @@ func TestCatalogue(t *testing.T) {
 }
 
 // The catalogue and the registered checks are the same set, so a check
-// ships with its trigger and near miss.
+// ships with its trigger and near miss; a two-pass check brings
+// rule-never-passes, which the engine reports for it.
 func TestEveryCatalogueRuleIsRegistered(t *testing.T) {
 	registered := map[string]bool{}
 	for _, c := range incoherenceChecks {
 		registered[c.code] = true
+		if c.twoPass {
+			registered["rule-never-passes"] = true
+		}
 	}
 	for _, r := range catalogue(t) {
 		if !registered[r.Code] {
diff --git a/cli/internal/conformity/rules.go b/cli/internal/conformity/rules.go
index d1c74903..9e58b7e1 100644
--- a/cli/internal/conformity/rules.go
+++ b/cli/internal/conformity/rules.go
@@ -1,6 +1,11 @@
 package conformity
 
 import (
+	"encoding/json"
+	"fmt"
+	"sort"
+	"strconv"
+	"strings"
 	"time"
 
 	"gopkg.in/yaml.v3"
@@ -16,7 +21,11 @@ import (
 type check struct {
 	code     string
 	severity Severity
-	run      func(c *ctx) []finding
+	// twoPass marks an error rule that the reference can tolerate: it is
+	// run once with tolerated constraints treated as absent, and when it
+	// fires only as written, rule-never-passes warns instead of an error.
+	twoPass bool
+	run     func(c *ctx) []finding
 }
 
 // finding is what a check reports, before it becomes a Diagnostic.
@@ -24,6 +33,7 @@ type finding struct {
 	node    *yaml.Node
 	path    string
 	message string
+	rules   []string // the rule ids it read, which rule-never-passes names when tolerated
 }
 
 var incoherenceChecks []check
@@ -32,10 +42,16 @@ func register(cs ...check) {
 	incoherenceChecks = append(incoherenceChecks, cs...)
 }
 
-// ctx is what a check sees: the lint date and the waivers.
+// ctx is what a check sees: the spec, the lint date, the declared rules
+// and the waivers.
 type ctx struct {
+	spec    *yaml.Node
 	today   time.Time
+	rules   map[string]*yaml.Node // declared rules by id, as declaredRules names them
 	waivers map[string]waiverInfo // the first waiver for each rule id
+	source  string                // kafka.source: brokers, or cr when empty
+	// ignoreTolerated is a two-pass check's first pass: a tolerated constraint reads as absent.
+	ignoreTolerated bool
 }
 
 type waiverInfo struct {
@@ -50,7 +66,13 @@ func (w waiverInfo) active(today time.Time) bool {
 }
 
 func newCtx(spec *yaml.Node, today time.Time) *ctx {
-	c := &ctx{today: today, waivers: map[string]waiverInfo{}}
+	c := &ctx{spec: spec, today: today, rules: map[string]*yaml.Node{}, waivers: map[string]waiverInfo{}}
+	for _, r := range declaredRules(spec) {
+		c.rules[r.id] = r.node
+	}
+	if s := lookup(spec, "kafka.source"); s != nil && s.Kind == yaml.ScalarNode {
+		c.source = s.Value
+	}
 	if ws := get(spec, "waivers"); ws != nil && ws.Kind == yaml.SequenceNode {
 		for i, w := range ws.Content {
 			rule, until := get(w, "rule"), get(w, "until")
@@ -69,6 +91,25 @@ func newCtx(spec *yaml.Node, today time.Time) *ctx {
 	return c
 }
 
+// tolerated says whether a declared rule cannot fail the check: its
+// severity is warn, or an active waiver names it.
+func (c *ctx) tolerated(id string) bool {
+	if w, ok := c.waivers[id]; ok && w.active(c.today) {
+		return true
+	}
+	sev := get(c.rules[id], "severity")
+	return sev != nil && sev.Value == "warn"
+}
+
+// node is the constraint a rule id declares, or nil when it is absent, or
+// tolerated during a two-pass check's first pass.
+func (c *ctx) node(id string) *yaml.Node {
+	if c.ignoreTolerated && c.tolerated(id) {
+		return nil
+	}
+	return c.rules[id]
+}
+
 // incoherence runs every registered check. It runs only on a reference
 // with no error so far.
 func incoherence(root *yaml.Node, today time.Time) []Diagnostic {
@@ -79,9 +120,193 @@ func incoherence(root *yaml.Node, today time.Time) []Diagnostic {
 	c := newCtx(spec, today)
 	var ds []Diagnostic
 	for _, chk := range incoherenceChecks {
+		if !chk.twoPass {
+			for _, f := range chk.run(c) {
+				ds = append(ds, at(f.node, chk.severity, chk.code, f.path, f.message))
+			}
+			continue
+		}
+		c.ignoreTolerated = true
+		hard := map[string]bool{}
 		for _, f := range chk.run(c) {
+			hard[f.path] = true
 			ds = append(ds, at(f.node, chk.severity, chk.code, f.path, f.message))
 		}
+		c.ignoreTolerated = false
+		for _, f := range chk.run(c) {
+			if !hard[f.path] {
+				ds = append(ds, at(f.node, Warning, "rule-never-passes", f.path, c.neverPasses(chk.code, f)))
+			}
+		}
 	}
 	return ds
 }
+
+// neverPasses explains a tolerated rule that cannot pass on any cluster.
+func (c *ctx) neverPasses(code string, f finding) string {
+	var how []string
+	for _, id := range f.rules {
+		if w, ok := c.waivers[id]; ok && w.active(c.today) {
+			how = append(how, fmt.Sprintf("spec.%s is waived until %s, a waiver no change to the cluster can retire", id, w.until.Format("2006-01-02")))
+		} else if c.tolerated(id) {
+			how = append(how, "spec."+id+" has severity warn, so the reference passes only while it warns")
+		}
+	}
+	return fmt.Sprintf("%s [%s]; it is tolerated, so the check does not fail, but it can never pass: %s",
+		strings.TrimSuffix(f.message, "."), code, strings.Join(how, "; "))
+}
+
+// jsonText is a scalar as the CLI sends it: an int, float or bool as
+// encoding/json writes the decoded value (2.0 is 2, 0x10 is 16, True is
+// true), anything else as its text.
+func jsonText(n *yaml.Node) string {
+	n = resolveAlias(n)
+	switch kind(n) {
+	case "integer", "number", "boolean":
+		var v any
+		if err := n.Decode(&v); err == nil {
+			if b, err := json.Marshal(v); err == nil {
+				return string(b)
+			}
+		}
+	}
+	return n.Value
+}
+
+// maxCount is the most an integer field allows: count, zones, skew and
+// perNode range over 0..maxCount when absent.
+const maxCount = 10000
+
+// ints is the values an integer constraint allows: a sorted list for a bare
+// value, equals or oneOf, else the range lo..hi; absent means 0..maxCount.
+type ints struct {
+	list   []int
+	lo, hi int
+}
+
+func intsOf(n *yaml.Node) ints {
+	n = resolveAlias(n)
+	whole := ints{lo: 0, hi: maxCount}
+	if !present(n) {
+		return whole
+	}
+	val := func(v *yaml.Node) (int, bool) {
+		if v == nil || v.Kind != yaml.ScalarNode {
+			return 0, false
+		}
+		i, err := strconv.Atoi(jsonText(v))
+		return i, err == nil
+	}
+	single := n
+	if n.Kind == yaml.MappingNode {
+		single = get(n, "equals")
+	}
+	if i, ok := val(single); ok {
+		return ints{list: []int{i}, lo: i, hi: i}
+	}
+	if list := get(n, "oneOf"); list != nil && list.Kind == yaml.SequenceNode {
+		var out ints
+		for _, item := range list.Content {
+			if i, ok := val(item); ok {
+				out.list = append(out.list, i)
+			}
+		}
+		sort.Ints(out.list)
+		if len(out.list) > 0 {
+			out.lo, out.hi = out.list[0], out.list[len(out.list)-1]
+			return out
+		}
+	}
+	if i, ok := val(get(n, "min")); ok {
+		whole.lo = i
+	}
+	if i, ok := val(get(n, "max")); ok {
+		whole.hi = i
+	}
+	return whole
+}
+
+// has says whether the constraint allows v.
+func (s ints) has(v int) bool {
+	if s.list != nil {
+		i := sort.SearchInts(s.list, v)
+		return i < len(s.list) && s.list[i] == v
+	}
+	return v >= s.lo && v <= s.hi
+}
+
+// any says whether some allowed value, tried in order, satisfies f.
+func (s ints) any(f func(int) bool) bool {
+	if s.list != nil {
+		for _, v := range s.list {
+			if f(v) {
+				return true
+			}
+		}
+		return false
+	}
+	for v := s.lo; v <= s.hi; v++ {
+		if f(v) {
+			return true
+		}
+	}
+	return false
+}
+
+func (s ints) String() string {
+	if s.list != nil {
+		parts := make([]string, len(s.list))
+		for i, v := range s.list {
+			parts[i] = strconv.Itoa(v)
+		}
+		if len(parts) == 1 {
+			return parts[0]
+		}
+		return "one of " + strings.Join(parts, ", ")
+	}
+	switch {
+	case s.lo == 0 && s.hi == maxCount:
+		return "any number"
+	case s.hi == maxCount:
+		return fmt.Sprintf("at least %d", s.lo)
+	case s.lo == 0:
+		return fmt.Sprintf("at most %d", s.hi)
+	}
+	return fmt.Sprintf("%d to %d", s.lo, s.hi)
+}
+
+// textsOf is the values a bare, equals or oneOf constraint allows, as the
+// CLI sends them; nil for min and max, or when it is absent.
+func textsOf(n *yaml.Node) []string {
+	n = resolveAlias(n)
+	if !present(n) {
+		return nil
+	}
+	if n.Kind == yaml.ScalarNode {
+		return []string{jsonText(n)}
+	}
+	if eq := get(n, "equals"); eq != nil && eq.Kind == yaml.ScalarNode {
+		return []string{jsonText(eq)}
+	}
+	if list := get(n, "oneOf"); list != nil && list.Kind == yaml.SequenceNode {
+		var out []string
+		for _, item := range list.Content {
+			if item = resolveAlias(item); item.Kind == yaml.ScalarNode {
+				out = append(out, jsonText(item))
+			}
+		}
+		return out
+	}
+	return nil
+}
+
+// only says whether a constraint is declared and allows nothing but v, in any case.
+func only(n *yaml.Node, v string) bool {
+	vals := textsOf(n)
+	for _, x := range vals {
+		if !strings.EqualFold(x, v) {
+			return false
+		}
+	}
+	return len(vals) > 0
+}
diff --git a/cli/internal/conformity/rules_durability.go b/cli/internal/conformity/rules_durability.go
new file mode 100644
index 00000000..0314766a
--- /dev/null
+++ b/cli/internal/conformity/rules_durability.go
@@ -0,0 +1,267 @@
+package conformity
+
+import (
+	"fmt"
+	"math"
+	"strconv"
+	"strings"
+	"unicode"
+
+	"gopkg.in/yaml.v3"
+)
+
+// The durability rules: broker settings the check can never read, or that
+// make replication promise less than it seems.
+
+func init() {
+	register(
+		check{code: "config-value-kafka-refuses", severity: Error, twoPass: true, run: kafkaRefuses},
+		check{code: "sensitive-config-from-brokers", severity: Error, twoPass: true, run: sensitiveFromBrokers},
+		check{code: "replication-factor-exceeds-brokers", severity: Warning, run: replicationExceedsBrokers},
+		check{code: "min-isr-not-below-rf", severity: Warning, run: minISRNotBelowRF},
+		check{code: "isr-lost-with-one-zone", severity: Warning, run: isrLostWithOneZone},
+		check{code: "unclean-election-required", severity: Warning, run: uncleanElection},
+	)
+}
+
+const unclean = "unclean.leader.election.enable"
+
+// kafkaAccepts is the integer range Kafka validates a broker key against.
+var kafkaAccepts = map[string][2]int64{
+	"min.insync.replicas":                              {1, math.MaxInt32},
+	"transaction.state.log.min.isr":                    {1, math.MaxInt32},
+	"offsets.topic.replication.factor":                 {1, math.MaxInt16},
+	"transaction.state.log.replication.factor":         {1, math.MaxInt16},
+	"share.coordinator.state.topic.replication.factor": {1, math.MaxInt16},
+	"share.coordinator.state.topic.min.isr":            {1, math.MaxInt16},
+	"default.replication.factor":                       {math.MinInt32, math.MaxInt32},
+}
+
+// configKeys is each kafka.config key once, in file order.
+func configKeys(c *ctx) []string {
+	config := lookup(c.spec, "kafka.config")
+	if config == nil || config.Kind != yaml.MappingNode {
+		return nil
+	}
+	var keys []string
+	seen := map[string]bool{}
+	for i := 0; i+1 < len(config.Content); i += 2 {
+		if k := config.Content[i].Value; !seen[k] {
+			seen[k] = true
+			keys = append(keys, k)
+		}
+	}
+	return keys
+}
+
+// refused says whether a constraint on a key Kafka validates admits no
+// value Kafka accepts, read as Integer.parseInt(text.trim()) does, or for
+// unclean.leader.election.enable as true or false.
+func refused(key string, n *yaml.Node) bool {
+	r, isInt := kafkaAccepts[key]
+	if !isInt && key != unclean {
+		return false
+	}
+	vals := textsOf(n)
+	if vals == nil && !isInt {
+		return true // min or max: a broker reports true or false, never a number
+	}
+	if vals == nil {
+		lo, hi, _ := nums(n)
+		return math.Max(math.Ceil(lo), float64(r[0])) > math.Min(math.Floor(hi), float64(r[1]))
+	}
+	for _, v := range vals {
+		v = strings.TrimFunc(v, func(ch rune) bool { return ch <= ' ' })
+		if strings.ContainsFunc(v, func(ch rune) bool { return ch > unicode.MaxASCII }) {
+			return false // Java also reads other scripts' digits; leave those be
+		}
+		if !isInt && (strings.EqualFold(v, "true") || strings.EqualFold(v, "false")) {
+			return false
+		}
+		if i, err := strconv.ParseInt(v, 10, 64); isInt && err == nil && i >= r[0] && i <= r[1] {
+			return false
+		}
+	}
+	return true
+}
+
+// kafkaRefuses: with source brokers, each broker's effective value is
+// compared, and Kafka never holds a value it refuses.
+func kafkaRefuses(c *ctx) []finding {
+	if c.source != "brokers" {
+		return nil
+	}
+	var fs []finding
+	for _, key := range configKeys(c) {
+		id := childPath("kafka.config", key)
+		if n := c.node(id); n != nil && refused(key, n) {
+			fs = append(fs, finding{n, "spec." + id, "admits no value Kafka accepts for the key, so no running broker can report one, and this rule can never pass.", []string{id}})
+		}
+	}
+	return fs
+}
+
+var sensitiveKeys = []string{"ssl.key.password", "ssl.keystore.password", "ssl.truststore.password", "ssl.keystore.key",
+	"ssl.keystore.certificate.chain", "ssl.truststore.certificates", "sasl.jaas.config", "password.encoder.secret",
+	"password.encoder.old.secret", "delegation.token.secret.key", "delegation.token.master.key"}
+
+// sensitiveFromBrokers: Kafka returns a password-type config without its
+// value, so with source brokers its rule is UNKNOWN or FAIL on every cluster.
+func sensitiveFromBrokers(c *ctx) []finding {
+	if c.source != "brokers" {
+		return nil
+	}
+	var fs []finding
+	for _, key := range configKeys(c) {
+		id := childPath("kafka.config", key)
+		n := c.node(id)
+		if n == nil {
+			continue
+		}
+		for _, s := range sensitiveKeys {
+			if key == s || strings.HasSuffix(key, "."+s) {
+				fs = append(fs, finding{n, "spec." + id, "with source brokers this key is sensitive, and Kafka never returns its value. " +
+					"The rule is always UNKNOWN or FAIL, so the reference can never pass. Drop the key, and keep secrets out of the reference.", []string{id}})
+				break
+			}
+		}
+	}
+	return fs
+}
+
+// nums is the least and greatest number a constraint admits: its values that
+// read as numbers, or its min and max, a missing one infinite. ok is false
+// when it is absent or admits no number. Rounding to float64 only widens the
+// range, so a check built on it fires less, never wrongly.
+func nums(n *yaml.Node) (lo, hi float64, ok bool) {
+	lo, hi = math.Inf(1), math.Inf(-1)
+	num := func(text string) (float64, bool) {
+		f, err := strconv.ParseFloat(strings.TrimSpace(text), 64)
+		return f, err == nil && !math.IsNaN(f)
+	}
+	if vals := textsOf(n); vals != nil {
+		for _, v := range vals {
+			if f, isNum := num(v); isNum {
+				lo, hi, ok = min(lo, f), max(hi, f), true
+			}
+		}
+		return lo, hi, ok
+	}
+	if !present(n) {
+		return lo, hi, false
+	}
+	lo, hi = math.Inf(-1), math.Inf(1)
+	if m := get(n, "min"); m != nil {
+		lo, ok = num(jsonText(m))
+	}
+	if m := get(n, "max"); m != nil {
+		var okHi bool
+		hi, okHi = num(jsonText(m))
+		ok = ok || okHi
+	}
+	return lo, hi, ok
+}
+
+// replicationFactors are the rules that set a topic's replicas.
+var replicationFactors = []string{
+	"kafka.config[default.replication.factor]", "kafka.config[offsets.topic.replication.factor]",
+	"kafka.config[transaction.state.log.replication.factor]", "kafka.config[share.coordinator.state.topic.replication.factor]",
+	"topics.replicationFactor",
+}
+
+// replicationExceedsBrokers: Kafka refuses a topic with more replicas than
+// registered brokers, so the topics these factors govern are never created.
+func replicationExceedsBrokers(c *ctx) []finding {
+	_, most, ok := nums(c.node("brokers.count"))
+	if !ok || math.IsInf(most, 1) {
+		return nil
+	}
+	var fs []finding
+	for _, id := range replicationFactors {
+		n := c.node(id)
+		if least, _, ok := nums(n); ok && least > most {
+			fs = append(fs, finding{node: n, path: "spec." + id, message: fmt.Sprintf("needs at least %v brokers, but spec.brokers.count allows at most %v: "+
+				"Kafka refuses to create a topic with more replicas than brokers.", least, most)})
+		}
+	}
+	return fs
+}
+
+// minISRs pair each min ISR rule with the rules that bound the in-sync
+// replicas it asks for: the replication factor, then the brokers.
+var minISRs = []struct {
+	isr    string
+	bounds []string
+}{
+	{"kafka.config[min.insync.replicas]", []string{"kafka.config[default.replication.factor]", "kafka.config[offsets.topic.replication.factor]", "brokers.count"}},
+	{"kafka.config[transaction.state.log.min.isr]", []string{"kafka.config[transaction.state.log.replication.factor]", "brokers.count"}},
+	{"topics.minInsyncReplicas", []string{"topics.replicationFactor", "brokers.count"}},
+}
+
+// minISRNotBelowRF: the ISR never exceeds the replication factor or the
+// brokers, so a min ISR above them refuses every acks=all write, and one
+// equal to them refuses those writes whenever a replica lags.
+func minISRNotBelowRF(c *ctx) []finding {
+	var fs []finding
+	for _, p := range minISRs {
+		n := c.node(p.isr)
+		lo, _, ok := nums(n)
+		if !ok {
+			continue
+		}
+		least := math.Ceil(lo)
+		bound, bhi := "", math.Inf(1)
+		for _, b := range p.bounds {
+			if _, hi, ok := nums(c.node(b)); ok && hi < bhi {
+				bound, bhi = b, hi
+			}
+		}
+		if bound == "" || least < bhi || least == bhi && bhi < 2 {
+			continue
+		}
+		outcome := "always refused"
+		if least == bhi {
+			outcome = "refused whenever one replica is out of sync, rolling restarts included"
+		}
+		fs = append(fs, finding{node: n, path: "spec." + p.isr, message: fmt.Sprintf("is at least %v, and spec.%s is at most %v: acks=all writes it governs are %s. Keep it below spec.%s.",
+			least, bound, bhi, outcome, bound)})
+	}
+	return fs
+}
+
+// isrLostWithOneZone: some zone holds ceil(R/Z) replicas of a partition
+// however they are placed, so if even the most zones and replicas allowed
+// leave fewer than the min ISR, a single-zone outage stops acks=all writes.
+func isrLostWithOneZone(c *ctx) []finding {
+	zn := c.node("brokers.zones")
+	zlo, z, ok := nums(zn)
+	if !ok || zlo < 2 || math.IsInf(z, 1) {
+		return nil
+	}
+	var fs []finding
+	for _, p := range minISRs {
+		_, rhi, okR := nums(c.node(p.bounds[0]))
+		mlo, _, okM := nums(c.node(p.isr))
+		r, m := math.Floor(rhi), math.Max(1, math.Ceil(mlo))
+		if !okR || !okM || math.IsInf(r, 1) || r < 2 || m >= r {
+			continue
+		}
+		if left := r - math.Ceil(r/z); left < m {
+			fs = append(fs, finding{node: zn, path: "spec.brokers.zones", message: fmt.Sprintf("allows at most %v zones: with spec.%s %v, one zone holds %v replicas of a partition, "+
+				"so losing it leaves %v, below spec.%s %v, and a single-zone outage stops acks=all writes. Allow more zones, or a higher replication factor.",
+				z, p.bounds[0], r, r-left, left, p.isr, m)})
+		}
+	}
+	return fs
+}
+
+// uncleanElection: an out-of-sync leader truncates writes acknowledged
+// under acks=all; harmless only on a single broker.
+func uncleanElection(c *ctx) []finding {
+	id := childPath("kafka.config", unclean)
+	n := c.node(id)
+	if _, most, ok := nums(c.node("brokers.count")); !only(n, "true") || ok && most < 2 {
+		return nil
+	}
+	return []finding{{node: n, path: "spec." + id, message: "asks for true: an out-of-sync replica can become leader, and writes acknowledged under acks=all can be lost."}}
+}
diff --git a/cli/internal/conformity/rules_placement.go b/cli/internal/conformity/rules_placement.go
new file mode 100644
index 00000000..5b835068
--- /dev/null
+++ b/cli/internal/conformity/rules_placement.go
@@ -0,0 +1,190 @@
+package conformity
+
+import "fmt"
+
+// The placement rules: where a role's pods can run. The check counts a
+// role's pods (count), the zones that host one (zones), the most minus the
+// fewest pods per zone (skew) and the pods on each node that has one
+// (perNode). zones and skew fail a pod that is not scheduled in a zone.
+
+func init() {
+	register(
+		check{code: "zones-unreachable", severity: Error, twoPass: true, run: zonesUnreachable},
+		check{code: "skew-unreachable", severity: Error, twoPass: true, run: skewUnreachable},
+		check{code: "pernode-cannot-place-count", severity: Error, twoPass: true, run: perNodeCannotPlace},
+		check{code: "controller-quorum-even", severity: Warning, run: quorumEven},
+		check{code: "controller-quorum-zone-loss", severity: Warning, run: quorumZoneLoss},
+	)
+}
+
+var roles = []string{"controllers", "brokers"}
+
+// zonesFit says whether some allowed count and zones fit: each zone that
+// counts hosts a pod, and with any pod every pod must be in a zone.
+func zonesFit(count, zones ints) bool {
+	return count.has(0) && zones.has(0) || zones.any(func(z int) bool { return z >= 1 && z <= count.hi })
+}
+
+func zonesUnreachable(c *ctx) []finding {
+	var fs []finding
+	for _, role := range roles {
+		ids := []string{role + ".count", role + ".zones"}
+		zn := c.node(ids[1])
+		count, zones := intsOf(c.node(ids[0])), intsOf(zn)
+		if zn == nil || zonesFit(count, zones) {
+			continue
+		}
+		fs = append(fs, finding{zn, "spec." + ids[1], fmt.Sprintf("allows %s, but spec.%s allows %s: each zone that counts hosts a pod, "+
+			"and with any pod every pod must run in a zone, so no cluster can meet both.", zones, ids[0], count), ids})
+	}
+	return fs
+}
+
+// skewFits says whether some allowed count n, split over exactly z allowed
+// zones with one pod or more in each, has an allowed skew: one zone gives 0;
+// two give n's parity, up to n-2; more give 0 when z divides n, else 1, and
+// every skew from 2 to n-z.
+func skewFits(count, zones, skew ints) bool {
+	reach := func(n, z, s int) bool {
+		switch {
+		case z == 1:
+			return s == 0
+		case s < 0 || s > n-z:
+			return false
+		case z == 2:
+			return s%2 == n%2
+		case s >= 2:
+			return true
+		}
+		return (s == 0) == (n%z == 0)
+	}
+	return count.any(func(n int) bool {
+		if n == 0 {
+			return zones.has(0) && skew.has(0)
+		}
+		top := min(n, zones.hi)
+		if skew.lo > 0 {
+			top = min(top, n-skew.lo) // a skew of s needs z <= n-s
+		}
+		for z := max(1, zones.lo); z <= top; z++ {
+			if !zones.has(z) {
+				continue
+			}
+			if skew.list != nil {
+				for _, s := range skew.list {
+					if reach(n, z, s) {
+						return true
+					}
+				}
+				continue
+			}
+			// In a range, the reachable skews show within three steps of its least.
+			for s := skew.lo; s <= min(skew.hi, skew.lo+2); s++ {
+				if reach(n, z, s) {
+					return true
+				}
+			}
+		}
+		return false
+	})
+}
+
+func skewUnreachable(c *ctx) []finding {
+	var fs []finding
+	for _, role := range roles {
+		ids := []string{role + ".count", role + ".zones", role + ".skew"}
+		sn := c.node(ids[2])
+		count, zones, skew := intsOf(c.node(ids[0])), intsOf(c.node(ids[1])), intsOf(sn)
+		if sn == nil || !zonesFit(count, zones) || skewFits(count, zones, skew) {
+			continue
+		}
+		fs = append(fs, finding{sn, "spec." + ids[2], fmt.Sprintf("allows %s, but no count and zones this reference allows split with that skew: "+
+			"N pods in Z zones have skew 0 only when Z divides N, and with 2 zones the skew has N's parity.", skew), ids})
+	}
+	return fs
+}
+
+// placeable says whether some allowed count of pods, every one scheduled,
+// fills at least as many nodes as some allowed zones with each node's count
+// in perNode. shortcut: a oneOf perNode reads as the range from its least to
+// its greatest member, so {oneOf: [2, 4]} seems to place 3 pods; it only
+// fires less, upgrade if such a perNode turns up.
+func placeable(count, zones, per ints) bool {
+	a, b := max(1, per.lo), per.hi
+	z1 := maxCount + 1
+	zones.any(func(z int) bool {
+		if z >= 1 {
+			z1 = z
+		}
+		return z >= 1
+	})
+	return count.any(func(n int) bool {
+		if n == 0 {
+			return zones.has(0)
+		}
+		most := n / a // the most nodes n pods fill at a or more each
+		return b >= a && z1 <= most && (n+b-1)/b <= most
+	})
+}
+
+func perNodeCannotPlace(c *ctx) []finding {
+	var fs []finding
+	for _, role := range roles {
+		ids := []string{role + ".perNode", role + ".count", role + ".zones", role + ".skew"}
+		forcing := ""
+		for _, id := range ids[2:] {
+			if forcing == "" && c.node(id) != nil {
+				forcing = id
+			}
+		}
+		// rackAwareness from the CR fails a broker that runs on no node.
+		if role == "brokers" && c.source != "brokers" {
+			ids = append(ids, "kafka.rackAwareness")
+			if forcing == "" && only(c.node("kafka.rackAwareness"), "true") {
+				forcing = "kafka.rackAwareness"
+			}
+		}
+		pn := c.node(ids[0])
+		count, zones, per := intsOf(c.node(ids[1])), intsOf(c.node(ids[2])), intsOf(pn)
+		if pn == nil || forcing == "" || !zonesFit(count, zones) || placeable(count, zones, per) {
+			continue
+		}
+		fs = append(fs, finding{pn, "spec." + ids[0], fmt.Sprintf("allows %s on each node that hosts one, but spec.%s requires every pod to run on a node, "+
+			"and no count of %s this reference allows splits that way over enough nodes for its zones.", per, forcing, role), ids})
+	}
+	return fs
+}
+
+// quorumEven warns of a KRaft quorum that only an even count can make.
+func quorumEven(c *ctx) []finding {
+	n := c.node("controllers.count")
+	cs := intsOf(n)
+	if n == nil || cs.lo < 2 || cs.list == nil && cs.lo != cs.hi || cs.any(func(v int) bool { return v%2 == 1 }) {
+		return nil
+	}
+	return []finding{{node: n, path: "spec.controllers.count", message: fmt.Sprintf("is %s: a KRaft quorum of an even size "+
+		"tolerates no more failed controllers than one controller fewer, and every commit waits on one more. Use an odd count.", cs)}}
+}
+
+// quorumZoneLoss warns when no allowed count and zones keep a KRaft
+// majority through the loss of the zone with the most controllers.
+func quorumZoneLoss(c *ctx) []finding {
+	zn := c.node("controllers.zones")
+	count, zones := intsOf(c.node("controllers.count")), intsOf(zn)
+	if zn == nil || zones.lo < 2 {
+		return nil
+	}
+	paired := false
+	survives := count.any(func(n int) bool {
+		return zones.any(func(z int) bool {
+			paired = paired || z <= n
+			return z <= n && n-(n+z-1)/z >= n/2+1
+		})
+	})
+	if !paired || survives {
+		return nil
+	}
+	return []finding{{node: zn, path: "spec.controllers.zones", message: "no count of controllers spec.controllers.count allows survives losing a zone: " +
+		"some zone holds ceil(N/Z) of the N controllers, and the rest are short of the N/2+1 a KRaft quorum needs. " +
+		"Use at least 3 zones, with a count they split evenly, such as 3 in 3 or 5 in 3."}}
+}
diff --git a/cli/internal/conformity/rules_test.go b/cli/internal/conformity/rules_test.go
new file mode 100644
index 00000000..98437ea8
--- /dev/null
+++ b/cli/internal/conformity/rules_test.go
@@ -0,0 +1,75 @@
+package conformity
+
+import "testing"
+
+// skewFits agrees with every way to split n pods over exactly z zones.
+func TestSkewFitsMatchesEverySplit(t *testing.T) {
+	var splits func(n, z, least int) [][]int
+	splits = func(n, z, least int) [][]int {
+		if z == 0 {
+			if n == 0 {
+				return [][]int{{}}
+			}
+			return nil
+		}
+		var out [][]int
+		for first := least; first <= n; first++ {
+			for _, rest := range splits(n-first, z-1, first) {
+				out = append(out, append([]int{first}, rest...))
+			}
+		}
+		return out
+	}
+	for n := 0; n <= 14; n++ {
+		for z := 0; z <= n+1; z++ {
+			seen := map[int]bool{}
+			for _, p := range splits(n, z, 1) {
+				if len(p) == 0 {
+					seen[0] = true
+				} else {
+					seen[p[len(p)-1]-p[0]] = true
+				}
+			}
+			for s := 0; s <= n+1; s++ {
+				if got := skewFits(just(n), just(z), just(s)); got != seen[s] {
+					t.Errorf("%d pods in %d zones, skew %d: skewFits says %v", n, z, s, got)
+				}
+				// As a range, it fits when any skew from s to s+3 does.
+				want := seen[s] || seen[s+1] || seen[s+2] || seen[s+3]
+				if got := skewFits(just(n), just(z), ints{lo: s, hi: s + 3}); got != want {
+					t.Errorf("%d pods in %d zones, skew %d to %d: skewFits says %v", n, z, s, s+3, got)
+				}
+			}
+		}
+	}
+}
+
+func just(v int) ints { return ints{list: []int{v}, lo: v, hi: v} }
+
+// Each role gets its own two-pass verdict: one hard error, one tolerated.
+func TestTwoPassJudgesEachFindingAlone(t *testing.T) {
+	ds := diagnose(t, `  brokers:
+    count: 2
+    zones: {min: 3}
+  controllers:
+    count: 1
+    zones: {min: 2, severity: warn}
+`)
+	var hard, tolerated []string
+	for _, d := range ds {
+		switch d.Code {
+		case "zones-unreachable":
+			hard = append(hard, d.Path)
+		case "rule-never-passes":
+			tolerated = append(tolerated, d.Path+": "+d.Message)
+		}
+	}
+	if len(hard) != 1 || hard[0] != "spec.brokers.zones" {
+		t.Errorf("errors at %v, want spec.brokers.zones", hard)
+	}
+	if len(tolerated) != 1 || tolerated[0] != "spec.controllers.zones: allows at least 2, but spec.controllers.count allows 1: "+
+		"each zone that counts hosts a pod, and with any pod every pod must run in a zone, so no cluster can meet both [zones-unreachable]; "+
+		"it is tolerated, so the check does not fail, but it can never pass: spec.controllers.zones has severity warn, so the reference passes only while it warns" {
+		t.Errorf("rule-never-passes: %q", tolerated)
+	}
+}
diff --git a/cli/internal/conformity/testdata/rules.json b/cli/internal/conformity/testdata/rules.json
index 56d20889..dc7a8483 100644
--- a/cli/internal/conformity/testdata/rules.json
+++ b/cli/internal/conformity/testdata/rules.json
@@ -5,5 +5,89 @@
     "title": "A waiver whose date has passed",
     "triggers": "# lint date 2026-10-19\nbrokers:\n  count: {min: 3}\nwaivers:\n  - {rule: brokers.count, reason: OPS-1, by: platform-team, until: 2026-01-31}",
     "nearMiss": "# lint date 2026-10-19\nbrokers:\n  count: {min: 3}\nwaivers:\n  - {rule: brokers.count, reason: OPS-1, by: platform-team, until: 2026-10-19}"
+  },
+  {
+    "code": "zones-unreachable",
+    "severity": "error",
+    "title": "Zones required that the pod count cannot reach",
+    "triggers": "brokers:\n  count: 2\n  zones: {min: 3}",
+    "nearMiss": "brokers:\n  count: {min: 2}\n  zones: {min: 3}"
+  },
+  {
+    "code": "skew-unreachable",
+    "severity": "error",
+    "title": "Skew impossible for the allowed count and zones",
+    "triggers": "brokers:\n  count: 4\n  zones: 3\n  skew: 0",
+    "nearMiss": "brokers:\n  count: 4\n  zones: {min: 3}\n  skew: 0"
+  },
+  {
+    "code": "pernode-cannot-place-count",
+    "severity": "error",
+    "title": "perNode cannot place the pods that must be scheduled",
+    "triggers": "brokers:\n  count: 3\n  zones: {min: 1}\n  perNode: {max: 0}",
+    "nearMiss": "brokers:\n  count: 3\n  perNode: {max: 0}"
+  },
+  {
+    "code": "controller-quorum-zone-loss",
+    "severity": "warning",
+    "title": "KRaft quorum cannot survive losing one zone",
+    "triggers": "controllers:\n  count: 3\n  zones: 2",
+    "nearMiss": "controllers:\n  count: 3\n  zones: {min: 2}"
+  },
+  {
+    "code": "controller-quorum-even",
+    "severity": "warning",
+    "title": "Even KRaft controller count",
+    "triggers": "controllers:\n  count: 4",
+    "nearMiss": "controllers:\n  count: {min: 4, max: 5}"
+  },
+  {
+    "code": "config-value-kafka-refuses",
+    "severity": "error",
+    "title": "Broker setting admits only values Kafka refuses",
+    "triggers": "kafka:\n  source: brokers\n  config:\n    min.insync.replicas: {max: 0}",
+    "nearMiss": "kafka:\n  source: brokers\n  config:\n    min.insync.replicas: 2.0"
+  },
+  {
+    "code": "sensitive-config-from-brokers",
+    "severity": "error",
+    "title": "With source brokers, a config key Kafka never returns",
+    "triggers": "kafka:\n  source: brokers\n  config:\n    ssl.keystore.password: changeit",
+    "nearMiss": "kafka:\n  source: brokers\n  config:\n    ssl.keystore.type: PKCS12"
+  },
+  {
+    "code": "replication-factor-exceeds-brokers",
+    "severity": "warning",
+    "title": "Replication factor above the most brokers the reference allows",
+    "triggers": "brokers:\n  count: 3\nkafka:\n  config:\n    offsets.topic.replication.factor: 4",
+    "nearMiss": "brokers:\n  count: {oneOf: [3, 5]}\nkafka:\n  config:\n    offsets.topic.replication.factor: 4"
+  },
+  {
+    "code": "min-isr-not-below-rf",
+    "severity": "warning",
+    "title": "min.insync.replicas not below the replication factor or broker count",
+    "triggers": "kafka:\n  config:\n    default.replication.factor: 3\n    min.insync.replicas: 3",
+    "nearMiss": "kafka:\n  config:\n    default.replication.factor: 3\n    min.insync.replicas: 2"
+  },
+  {
+    "code": "isr-lost-with-one-zone",
+    "severity": "warning",
+    "title": "Losing one zone drops partitions below min.insync.replicas",
+    "triggers": "brokers:\n  zones: 2\nkafka:\n  config:\n    default.replication.factor: 3\n    min.insync.replicas: 2",
+    "nearMiss": "brokers:\n  zones: 3\nkafka:\n  config:\n    default.replication.factor: 3\n    min.insync.replicas: 2"
+  },
+  {
+    "code": "unclean-election-required",
+    "severity": "warning",
+    "title": "Reference requires unclean leader election",
+    "triggers": "kafka:\n  config:\n    unclean.leader.election.enable: true",
+    "nearMiss": "brokers:\n  count: 1\nkafka:\n  config:\n    unclean.leader.election.enable: true"
+  },
+  {
+    "code": "rule-never-passes",
+    "severity": "warning",
+    "title": "A tolerated rule that can never pass",
+    "triggers": "# lint date 2026-10-19\nbrokers:\n  count: 2\n  zones: {min: 3}\nwaivers:\n  - {rule: brokers.zones, reason: OPS-412, by: platform-team, until: 2026-12-31}",
+    "nearMiss": "# lint date 2026-10-19\nbrokers:\n  count: 2\n  zones: {min: 3}\nwaivers:\n  - {rule: brokers.zones, reason: OPS-412, by: platform-team, until: 2026-01-31}"
   }
 ]
diff --git a/docs/book/10-cli-reference.md b/docs/book/10-cli-reference.md
index efdcedaf..6b727a6c 100644
--- a/docs/book/10-cli-reference.md
+++ b/docs/book/10-cli-reference.md
@@ -619,7 +619,7 @@ kates conformity check -f prod-3az.yaml -o json
 
 It exits `0` when every `fail` rule holds, `1` when one does not, and `2` when it could not tell or could not run. A `warn` rule never fails it. `--output-junit` also writes one test case per rule for a CI report: a failure for `FAIL`, an error for a `fail` rule that is `UNKNOWN`, and skipped for `WAIVED`. A `warn` rule's test case passes, with what it found in the output.
 
-Before it calls the API, it lints the file as `kates conformity lint` does and prints what lint finds to stderr. A warning, such as an expired waiver, does not stop it. An error stops it with exit `2`, and the API is never called. That includes a reference no cluster could pass, such as more zones required than brokers allowed.
+Before it calls the API, it lints the file as `kates conformity lint` does and prints what lint finds to stderr. A warning, such as an expired waiver, does not stop it. An error stops it with exit `2`, and the API is never called. That includes a reference no cluster could pass, such as more zones required than brokers allowed. When a rule at fault is `warn` or waived, the check can still pass, so lint warns instead, with `rule-never-passes`.
 
 #### conformity lint