#!/usr/bin/env python3
"""
provider-bundle-check.py -- focused, fail-closed QA for the Pattern A provider revert (D-060).

Asserts ONLY the post-revert (D-052/D-053 + Pattern A) provider invariants on a
Charmed-OpenStack bundle:
  1. exactly 12 API charms bind public -> provider-public; none remain on provider-vip
     (12 since BUNDLEFIX-056 2026-08-06 -- designate joined; was 11; vault stays OUT, metal-only)
  2. every clustered VIP is EITHER a v4 triple -- provider-public(10.12.4/22)
     admin(10.12.8/22) internal(10.12.12/22) -- OR a dual-family sextet appending
     the three v6 legs (R2, RULED 2026-07-27: dual-stack for both DCs). All legs
     share ONE host part, in 50-99 (D-020 amendment / R11, 2026-07-27), and the v6
     host part MIRRORS the v4 octet textually (D-136 ruling, same date).
     prefer-ipv6 and the v6 legs must land TOGETHER (see invariant 9).
     EXCEPTION -- vault (D-020 amendment 2026-08-05): vault is METAL-ONLY, the metal
     PAIR (metal-admin + metal-internal) with NO provider-public leg -- the internal
     PKI/CA root must not be tenant-facing and has no provider leg to host a VIP. It
     is the single sanctioned non-triple clustered VIP; still band- and
     octet-uniqueness-checked, just against the metal pair.
  3. ovn-chassis bridge-interface-mappings carries ALL FOUR chassis MACs, INCLUDING
     openstack0's -- the Pattern A revert re-adds the openstack0 MAC that D-057 trimmed
     for the now-dead provider-vip plane.

Structural invariants (absorbed from the retired scripts/review-bundle.py -- DOCFIX-070;
that linter's expectations were pre-D-052 and its NOT CLEAN verdict was pure noise):
  4. every relation side names an EXISTING app and carries an explicit :endpoint
  5. mysql-innodb-cluster deploys at num_units 3 (D-062 -- single-unit seed never bootstraps)
  6. no VIP last-octet is shared between applications
  7. keystone ships the D-051/D-064 policy IN the bundle (resources: policyd-override,
     use-policyd-override true), and the committed zip content matches
     policies/domain-manager-policy.yaml (DOCFIX-071 drift guard; content compare, not
     byte compare, so zip mtimes cannot false-fail it)
  8. R11 (D-020 AMENDMENT, 2026-07-27): a principal wired to an hacluster subordinate
     MUST carry a vip. Without one it binds a UNIT address and pacemaker has nothing
     to fail over to -- decorative HA.
  9. prefer-ipv6 IS ONLY LEGAL ON A CHARM THAT DECLARES IT, and where it is legal it
     stays COUPLED to dual-family VIP arity.
     REPLACED 2026-07-31 (D-101 RULING NOTE, GA-R5: "Yes -- keep the v6 legs, remove
     only the option"). The previous text of this invariant read "prefer-ipv6 makes
     HAProxy bind :::port in ADDITION to *:port, so the two must travel together".
     THAT MECHANISM CLAIM IS REFUTED BY MEASUREMENT
     (`docs/audit/stage5-prefer-ipv6-charm-research-20260731.txt`): in BOTH charm
     template families the `:::port` frontend bind is gated on
     `ipv6_enabled = not is_ipv6_disabled()` -- a read of the kernel
     `net.ipv6.conf.all.disable_ipv6` sysctl -- and pacemaker picks
     `ocf:heartbeat:IPv6addr` by per-address family detection. Neither consults the
     option. The one value prefer-ipv6 sets that resembles a bind address,
     `haproxy_host`, is consumed by NO template in any charm downloaded.
     THE INVARIANT IS REPLACED, NOT DELETED -- the L3-9 half it was built on is real
     and is retained. Three assertions now:
       9a  prefer-ipv6 present on a charm that does NOT declare it FAILS. This is the
           2026-07-31 attempt-1 defect: juju rejects the WHOLE bundle atomically
           (`unknown option "prefer-ipv6"` on barbican) rather than ignoring it, and
           `juju deploy --dry-run` does NOT validate option names, so nothing else
           in this repo can catch it.
       9b  RE-POINTED 2026-07-31 (D-101 RULING NOTE (b)): the option must be ABSENT
           from EVERY application until IPv6 is operational. It previously required the
           option and the v6 legs to travel together on a declaring charm; the ruling
           removes it everywhere, so that form would fail on the ruled artifact. The v6
           legs are RETAINED (arity is 9c). L3-9 cannot recur while the option is absent
           everywhere. This returns to the coupling form when v6 is operational.
       9c  arity itself: a vip is a v4 triple or a dual-family sextet, never other.
     PREFER_IPV6_CHARMS below is the measured authority for 9a/9b.
 10. HA ARITY (2026-07-29 gate hardening, the second half of R11's shape): an hacluster
     subordinate's `cluster_count` MUST equal its principal's `num_units`. Until this
     landed, `cluster_count` was asserted NOWHERE in scripts/ or tests/ -- so rewriting
     every declared value 3->1 produced a BYTE-IDENTICAL PASS while pacemaker was free to
     bootstrap the VIP off a single node: exactly the decorative HA D-121 exists to
     retire. A MISSING cluster_count also FAILS (an unrecognised state refuses; the
     charm default is not a substitute for the declared quorum).
 11. MACHINES-BLOCK DC IDENTITY (same hardening). ONE Office1 MAAS region sees BOTH
     DCs' nodes (D-104/D-123), so the machines block's `tags=` is the only thing
     keeping dc1's model off dc0's hardware. A machines overlay that PARSES but
     contributes nothing -- gutted, or keyed "01" where the base says "0" -- was a
     byte-identical PASS and would have allocated dc1 against `tags=openstack-vr1-dc0`.
     Three assertions: every machine carries a DC tag; the whole block carries exactly
     ONE distinct DC tag (a mis-keyed overlay yields base dc0 + overlay dc1 = mixed,
     and fails under EITHER --dc); and that tag matches --dc. Plus: a machine that no
     application places `to:` is an overlay entry that contributes nothing.
     RESIDUAL LIMIT, stated plainly: a gutted overlay is only caught because --dc
     DECLARES the intended DC. Validating a dc1 input while calling it dc0 is outside
     what any bundle-static check can see.
 12. EVERY APPLICATION CARRIES AN EXPLICIT `base:`, EQUAL TO `default-base` (RULED
     2026-07-31, GA-R5 option D). MEASURED, not reasoned -- capture
     docs/audit/stage5-dc0-deploy-attempt2-20260731.txt: a bundle whose applications rely
     SOLELY on `default-base` is deployable exactly ONCE. On a RE-RUN, for an application
     that ALREADY EXISTS, juju resolves the charm WITHOUT that default and lands on the
     ubuntu@24.04 revision (barbican 265 vs 261; mysql-router 1154 vs 1178), then refuses
     the downgrade -- so a partially-failed deploy cannot be resumed at all. A model-level
     `default-base` does NOT fix it (set, read back, re-run identical). The 56 redundant
     lines ARE the fix, and this invariant exists so they cannot be silently "simplified"
     away: without it, deleting them is a byte-identical PASS and the trap returns at the
     next partial failure, at any DC. Subordinates are included deliberately --
     mysql-router differs across bases, and hacluster happening to share a revision today
     is not a property to depend on.

It is the single deploy-gate: it REPLACED the retired scripts/d057-bundle-check.py (D-060)
and now also scripts/review-bundle.py (DOCFIX-070). FAIL -> exit 1. ASCII-only output.
(repo-lint: allow-stale-tokens -- guard checks name retired tokens by necessity)
"""
import sys, re, ipaddress, argparse
try:
    import yaml
except ImportError:
    sys.stderr.write("ERROR: PyYAML not installed (pip install pyyaml --break-system-packages)\n"); sys.exit(2)

# Per-DC VIP bands (D-101: only the network prefix moves per DC; the octet band is
# unchanged). --dc selects which set the VIP checks expect; default = the base
# bundle's dc0 values, so an un-parameterized run is byte-identical to before.
DC_BANDS = {
    "vr1-dc0": ("10.12.4.0/22", "10.12.8.0/22", "10.12.12.0/22"),
    "vr1-dc1": ("10.12.64.0/22", "10.12.68.0/22", "10.12.72.0/22"),
}
# The MAAS DC-scoping tag each --dc expects in the machines block's `tags=`.
# Written out per DC rather than glued from the flag ("openstack-" + dc): these are
# the literal tag strings carried by scripts/lib-hosts.sh (HOST_TAG) and
# overlays/vr1-dc1-machines.yaml, and a greppable table is what the next session reads.
DC_MACHINE_TAGS = {
    "vr1-dc0": "openstack-vr1-dc0",
    "vr1-dc1": "openstack-vr1-dc1",
}
# D-020 AMENDMENT / R11 (RULED 2026-07-27): the band widens 50-60 -> 50-99 to admit
# vault .61 and designate .62.
# EXPECT_PUBLIC_VIP = 12 (BUNDLEFIX-056, 2026-08-06): designate JOINS the public->
# provider-public count. The prior "deliberately STAYS 11" reasoning was DESCRIPTIVE OF A
# DEFECT -- it read designate's then-missing `public:` binding as intent. designate was ruled
# the FULL provider/admin/internal triple (.62, provider leg included) and its DEPLOYED charm
# DECLARES a `public` extra-binding (metadata.yaml on designate/0 verified); leaving it off
# provider-public orphaned the .4.62 leg and broke its TLS (F-CV1). Invariant 2 already
# REQUIRES designate's provider VIP triple, so invariant 1 asserting no public binding was
# internally inconsistent. vault STAYS OUT (metal-only, D-020 amendment 2026-08-05): the count
# is 12, NOT 13 -- bumping to 13 to admit vault remains WRONG and must still fail.
OCTET_LO, OCTET_HI = 50, 99
EXPECT_PUBLIC_VIP  = 12

# Invariant 9's measured authority: the charms that DECLARE a `prefer-ipv6` config
# option. Keyed by CHARM NAME, never by application name -- the application name is
# the deployer's choice and only the charm has a schema.
#
# MEASURED 2026-07-31 from the charm ARTIFACT at each pinned channel, downloaded for
# amd64/ubuntu@22.04 and read out of `config.yaml`. NOT from Charmhub's config-yaml API
# alone, and NOT from a repo comment. Capture:
# docs/audit/stage5-prefer-ipv6-charm-research-20260731.txt.
#
# SCOPE: ALL 33 charms in bundle.yaml, not only the thirteen that carry a VIP. The
# first pass measured the VIP charms alone and would have left this list WRONG -- five
# non-VIP charms declare the option, and `overlays/dc-dc-ipv6-family-matrix.yaml`
# already sets it on ceph-mon, an application with no VIP at all.
#
#   DECLARE (12)    ceph-mon 491, ceph-osd 953, ceph-radosgw 600 (all squid/stable);
#                   cinder 820, glance 681, keystone 857, neutron-api 710,
#                   nova-cloud-controller 823, nova-compute 894,
#                   openstack-dashboard 750 (all 2024.1/stable);
#                   hacluster 166 (2.4/stable); mysql-innodb-cluster 164 (8.0/stable)
#   DO NOT (21)     barbican 265, barbican-vault 99, cinder-backup 94, cinder-ceph 568,
#                   designate 418, designate-bind 266, glance-simplestreams-sync 152,
#                   magnum 96, magnum-dashboard 122, octavia 571, octavia-dashboard 168,
#                   octavia-diskimage-retrofit 257, placement 154 (all 2024.1/stable);
#                   ceph-rbd-mirror 62 (squid/stable); memcached 39 (latest/stable);
#                   mysql-router 1154 (8.0/stable); neutron-api-plugin-ovn 215;
#                   ovn-central 311, ovn-chassis 396 (24.03/stable);
#                   rabbitmq-server 295 (3.9/stable); vault 724 (1.8/stable)
#
# barbican returns NO at 2023.2/stable, 2023.1/stable AND ussuri/stable, so this is a
# SUBSET of the OpenStack charms rather than an option newer charms dropped.
#
# THIS IS A PINNED MEASUREMENT, NOT A STANDING TRUTH. A channel is a moving target: if
# a pin changes, re-measure against the new revision rather than trusting this list.
# scripts/render-dc-overlays.py READS this tuple with `ast` rather than restating it --
# a second copy is a second thing to drift (same pattern as APP_OCTET).
PREFER_IPV6_CHARMS = (
    "ceph-mon",
    "ceph-osd",
    "ceph-radosgw",
    "cinder",
    "glance",
    "hacluster",
    "keystone",
    "mysql-innodb-cluster",
    "neutron-api",
    "nova-cloud-controller",
    "nova-compute",
    "openstack-dashboard",
)

# v6 VIP bands are NOT hardcoded: D-136 option (D) (ADOPTED 2026-07-27) makes the
# NetBox apex the source for v6 prefixes. Same record and same (role, kind) keying
# as scripts/dc-plane-ipam.sh:102-131 and netbox/dc-plane-apex-import.py.
APEX_ENV  = "APEX_RECORD"
APEX_GLOB = "netbox/draft/vr1-office1-current-*.json"
EXPECT_CHASSIS_MACS = {
    "52:54:00:3d:fd:54",   # openstack0 -- re-added by the Pattern A revert (D-057 had trimmed it)
    "52:54:00:9d:63:77",   # openstack1
    "52:54:00:89:7f:ce",   # openstack2
    "52:54:00:99:fc:c2",   # openstack3
}
MAC_RE = re.compile(r"[0-9a-f]{2}(?::[0-9a-f]{2}){5}")
ROLES = ("control", "compute", "storage")

def _deep_merge(a, b):
    if isinstance(a, dict) and isinstance(b, dict):
        out = dict(a)
        for k, v in b.items():
            out[k] = _deep_merge(a[k], v) if k in a else v
        return out
    return b

def merge_overlay(base, over):
    """Juju-like overlay merge: applications/machines deep-merge key-by-key (a None
    overlay value deletes that key), relations append, other top-level keys override.
    NOTE: this MIRRORS juju's documented merge so the checker validates the effective
    deploy input; the machines-block merge in particular is CONFIRMED at
    `juju deploy --dry-run` before deploy (no offline juju merger exists)."""
    base, over = base or {}, over or {}
    out = dict(base)
    for section in ("applications", "machines"):
        b = dict(base.get(section) or {})
        for k, v in (over.get(section) or {}).items():
            if v is None:
                b.pop(k, None)
            elif isinstance(b.get(k), dict) and isinstance(v, dict):
                b[k] = _deep_merge(b[k], v)
            else:
                b[k] = v
        if b:
            out[section] = b
    rels = list(base.get("relations") or []) + list(over.get("relations") or [])
    if rels:
        out["relations"] = rels
    for k, v in over.items():
        if k not in ("applications", "machines", "relations"):
            out[k] = v
    return out

def _v6_family(x):
    """'ula' for fc00::/7, else 'gua'. Mirrors render-dc-overlays.py:v6_family."""
    n = ipaddress.ip_network(x, strict=False) if "/" in str(x) else \
        ipaddress.ip_network(str(x) + "/128", strict=False)
    return "ula" if n.subnet_of(ipaddress.ip_network("fc00::/7")) else "gua"


def _apex_v6_bands(dc, leg_addrs=None):
    """The three v6 VIP-leg /64s for `dc`, read from the NetBox apex record.

    Returns (bands, None) on success or (None, reason) -- a reason is a REFUSAL
    (exit 2), never a pass: "could not look" is never "nothing there".

    The PROVIDER leg lives in the DEDICATED GUA VIP /64 (the apex marks it with
    "VIP" in the description), NOT the provider node /64; admin and internal use
    their own plane /64s. The record's shape differs between the live API and the
    repo dumps (live returns nested scope.slug/scope.name, dumps flatten to
    scope_site + a bare role string), so both are handled -- writing against one
    shape and matching zero objects on the other is a measured trap.
    """
    import glob, json, os
    rec = os.environ.get(APEX_ENV)
    if not rec:
        root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        cand = sorted(glob.glob(os.path.join(root, APEX_GLOB)))
        rec = cand[-1] if cand else None
    if not rec or not os.path.isfile(rec):
        return None, "no readable apex record (looked for %s; set %s)" % (APEX_GLOB, APEX_ENV)
    try:
        doc = json.load(open(rec, encoding="utf-8"))
    except Exception as e:
        return None, "apex record %s unreadable: %s" % (rec, e)
    found = {}
    for p in (doc.get("ipam/prefixes") or []):
        pre = str(p.get("prefix", ""))
        if ":" not in pre or not pre.endswith("/64"):
            continue                       # /60 and /48 are parent blocks, not planes
        sc = p.get("scope") or {}
        if (sc.get("slug") or sc.get("name") or p.get("scope_site") or "") != dc:
            continue
        role = p.get("role") or ""
        role = role.get("slug") if isinstance(role, dict) else role
        kind = "vip" if "VIP" in str(p.get("description") or "") else "plane"
        # REFUSE a duplicate rather than take the last. This reader and the renderer's
        # share the record, so last-writer-wins here would make the gate AGREE with a
        # wrongly-rendered overlay: measured in the 2026-07-29 chain audit, adding ONE
        # spare prefix to the apex re-homed every dc0 VIP into a different /64 and the
        # whole chain went GREEN. Which one won depended on JSON array order, which is
        # not a contract.
        fam = _v6_family(pre)
        if (role, kind, fam) in found and found[(role, kind, fam)] != pre:
            return None, ("apex record %s has TWO %s %s/%s /64s for %s (%s and %s) -- refusing "
                          "to pick one by array order"
                          % (rec, fam, role, kind, dc, found[(role, kind, fam)], pre))
        found[(role, kind, fam)] = pre
    # Per-LEG family resolution (2026-08-01). Mid-D-139 the three legs legitimately sit
    # in DIFFERENT families: provider-public's VIP /64 is GUA in both worlds (:11::), while
    # metal-admin/metal-internal are ULA until their DC's step-1 push lands. Inferring ONE
    # family for all three -- which a first cut of this fix did, from the provider leg --
    # asked for GUA bands on a still-ULA DC and refused every dc1 fixture. Caught by the
    # harness, not by review.
    want = (("provider", ("provider-public", "vip")),
            ("admin",    ("metal-admin", "plane")),
            ("internal", ("metal-internal", "plane")))
    bands, missing = {}, []
    for i, (leg, key) in enumerate(want):
        fams = sorted({f for (r, k, f) in found if (r, k) == key})
        if leg_addrs and i < len(leg_addrs):
            try:
                fam = _v6_family(leg_addrs[i])          # the family of THIS leg
            except Exception:
                fam = None
        else:
            fam = None
        if fam is None:
            if len(fams) > 1:
                return None, ("apex record %s has BOTH families for %s/%s in %s (%s) -- the "
                              "expected D-139 transition state; the caller must supply the "
                              "leg being checked. Refusing to pick one by array order"
                              % (rec, key[0], key[1], dc, ", ".join(fams)))
            fam = fams[0] if fams else None
        if fam is not None and (key[0], key[1], fam) in found:
            bands[leg] = ipaddress.ip_network(found[(key[0], key[1], fam)])
        else:
            missing.append("%s/%s%s" % (key[0], key[1], (" (%s)" % fam) if fam else ""))
    if missing:
        return None, ("apex record %s carries no v6 prefix for %s in %s -- refusing to "
                      "guess a VIP band" % (rec, ", ".join(missing), dc))
    return bands, None


def _host_part(addr):
    """The trailing label of an address: the v4 last octet, or the final v6 group.

    The 2026-07-27 ruling mirrors the v4 octet DIGITS into the v6 host part
    (keystone .50 -> ...::50), so the two are compared as TEXT and never
    numerically -- `printf '%x' 50` is 32, which is exactly the plausible-looking
    wrong band that ruling exists to avoid.
    """
    ip = ipaddress.ip_address(addr)
    s = str(ip)
    return s.rsplit(".", 1)[-1] if ip.version == 4 else s.rsplit(":", 1)[-1]


def _tags_of(spec):
    """The `tags=` set from a machine spec's constraints string ({} when absent)."""
    m = re.search(r"tags=(\S+)", str((spec or {}).get("constraints", "")))
    return set(m.group(1).split(",")) if m else set()


def _role_of(spec):
    """Extract control/compute/storage from a machine spec's `constraints: tags=...`."""
    tags = _tags_of(spec)
    for r in ROLES:
        if r in tags:
            return r
    return None


def _dc_tag_of(spec):
    """The DC-scoping tag (openstack-vr1-dcN) from a machine spec, or None.

    Matched against the DC_MACHINE_TAGS table rather than a pattern, so a typo'd
    tag reads as ABSENT (and fails) instead of being accepted as some new DC.
    """
    tags = _tags_of(spec)
    hits = sorted(t for t in tags if t in set(DC_MACHINE_TAGS.values()))
    return hits[0] if len(hits) == 1 else None

def main():
    ap = argparse.ArgumentParser(description="Pattern A / D-052-D-053 + placement bundle invariants")
    ap.add_argument("path", nargs="?", default="bundle.yaml", help="base bundle (default bundle.yaml)")
    ap.add_argument("--overlay", action="append", default=[], metavar="FILE",
                    help="overlay bundle(s), applied in order (juju-like merge) to validate the effective deploy input")
    ap.add_argument("--dc", choices=sorted(DC_BANDS), default="vr1-dc0",
                    help="DC whose VIP bands to expect (default vr1-dc0 = the base bundle's own values)")
    args = ap.parse_args()
    path = args.path
    try:
        doc = yaml.safe_load(open(path, encoding="utf-8"))
    except Exception as e:
        sys.stderr.write("ERROR: cannot parse %s: %s\n" % (path, e)); return 2
    for ov in args.overlay:
        try:
            doc = merge_overlay(doc, yaml.safe_load(open(ov, encoding="utf-8")))
        except Exception as e:
            sys.stderr.write("ERROR: cannot merge overlay %s: %s\n" % (ov, e)); return 2
    PROVIDER, ADMIN, INTERNAL = (ipaddress.ip_network(x) for x in DC_BANDS[args.dc])
    apps = (doc or {}).get("applications", {}) or {}
    machines = (doc or {}).get("machines", {}) or {}
    roles = {str(mid): _role_of(spec) for mid, spec in machines.items()}
    role_sep = any(roles.values())   # role-separated (VR1) vs hyperconverged base (VR0)
    fails, oks = [], []

    def pub(s): return ((s or {}).get("bindings", {}) or {}).get("public")
    on_vip    = sorted(n for n, s in apps.items() if pub(s) == "provider-vip")
    on_public = sorted(n for n, s in apps.items() if pub(s) == "provider-public")
    if on_vip:
        fails.append("public still on provider-vip (must be reverted to provider-public): %s" % ", ".join(on_vip))
    if len(on_public) != EXPECT_PUBLIC_VIP:
        fails.append("public->provider-public count=%d (expect %d): %s" % (len(on_public), EXPECT_PUBLIC_VIP, ", ".join(on_public)))
    else:
        oks.append("%d charms bind public->provider-public; none on provider-vip" % len(on_public))

    # -- 2 + 9. VIP legs, arity, family coupling, band, and octet uniqueness.
    #    A vip is a v4 TRIPLE (3) or, under R2's ruled dual-stack, a dual-family
    #    SEXTET (6) appending the v6 provider/admin/internal legs.
    v6_bands, v6_refusal = None, None
    _bandcache = {}
    vip_ok = vip_dual = 0
    octet_owner = {}

    # 9a runs over EVERY application, not only the VIP-carrying ones. juju's `unknown
    # option` rejection is fatal for any application, and it aborts the WHOLE bundle
    # atomically. Written inside the VIP loop first, which would have missed the case
    # the repo ALREADY has: overlays/dc-dc-ipv6-family-matrix.yaml sets prefer-ipv6 on
    # ceph-mon, an application with no vip -- outside every check in this file. That
    # overlay is a LATER deploy step, so the miss would not have shown up in attempt 2;
    # it would have shown up at the step after, looking like a new fault.
    # Asserted on PRESENCE, not truthiness: `prefer-ipv6: false` is the same fatal
    # `unknown option` to juju as `true`.
    for n in sorted(apps):
        spec = apps[n] or {}
        if "prefer-ipv6" not in ((spec.get("options") or {})):
            continue
        ch = str(spec.get("charm") or "")
        if ch not in PREFER_IPV6_CHARMS:
            fails.append("%s sets prefer-ipv6 but its charm %r does NOT declare that "
                         "option -- juju rejects the whole bundle (2026-07-31 attempt 1). "
                         "The v6 VIP legs do NOT need it: the ':::port' bind is gated on "
                         "the kernel disable_ipv6 sysctl, not on this option (D-101 "
                         "RULING NOTE 2026-07-31)" % (n, ch))

    for n in sorted(apps):
        opts = ((apps[n] or {}).get("options") or {})
        vip = opts.get("vip")
        if not vip:
            continue
        prefer6 = bool(opts.get("prefer-ipv6"))
        parts = str(vip).split()
        # D-020 amendment 2026-08-05: vault is METAL-ONLY -- exactly the metal PAIR
        # (metal-admin + metal-internal), NO provider-public leg. vault is the internal
        # PKI/CA root: it must not be tenant-facing, and has no provider leg to host a VIP
        # (a provider VIP findif-fails on its containers). Family/DC-aware via the --dc
        # PROVIDER/ADMIN/INTERNAL bands, NOT a hardcoded dc0 prefix (a `10.12.4.` literal
        # was blind to dc1's 10.12.64.x provider leg). It stays the v4 metal pair THROUGH a
        # future D-141 rule-3 dual-stack promotion until a new amendment rules on vault's v6
        # metal legs. The metal pair is STILL subject to the band + octet-uniqueness gates
        # (an early `continue` here must NOT disarm them for vault's octet -- if it did, .61
        # would go un-band-checked AND never register in octet_owner, so a collision on .61
        # with any other app would pass silently: a gate that cannot fail).
        if n == "vault":
            if len(parts) != 2:
                fails.append("vault vip must be the metal PAIR (metal-admin + metal-internal); "
                             "got %d address(es): %r (D-020 amendment 2026-08-05)" % (len(parts), vip))
                continue
            try:
                a_ip, i_ip = (ipaddress.ip_address(p) for p in parts)
            except ValueError as e:
                fails.append("vault bad vip ip: %s" % e); continue
            if a_ip in PROVIDER or i_ip in PROVIDER:
                fails.append("vault vip carries a provider-public leg: %r -- vault is METAL-ONLY, "
                             "the PKI/CA root must not be tenant-facing (D-020 amendment "
                             "2026-08-05)" % vip); continue
            if not (a_ip in ADMIN and i_ip in INTERNAL):
                fails.append("vault vip legs are not [metal-admin, metal-internal] for --dc %s: "
                             "%r (D-020 amendment 2026-08-05)" % (args.dc, vip)); continue
            hosts = {_host_part(p) for p in parts}
            if len(hosts) != 1:
                fails.append("vault vip host parts differ across legs: %r" % vip); continue
            h = hosts.pop()
            if not h.isdigit():
                fails.append("vault vip host part %r is not the decimal v4-octet mirror" % h); continue
            o = int(h)
            if not (OCTET_LO <= o <= OCTET_HI):
                fails.append("vault vip octet .%d outside %d-%d" % (o, OCTET_LO, OCTET_HI)); continue
            if h in octet_owner:
                fails.append("VIP last octet .%s shared by %s and %s" % (h, octet_owner[h], n)); continue
            octet_owner[h] = n
            oks.append("vault vip is metal-only (metal-admin + metal-internal, no provider), "
                       "octet .%d in %d-%d -- D-020 amendment 2026-08-05" % (o, OCTET_LO, OCTET_HI))
            continue
        if len(parts) not in (3, 6):
            fails.append("%s vip is neither a v4 triple (3) nor a dual-family sextet (6) "
                         "-- got %d address(es): %r" % (n, len(parts), vip)); continue
        dual = (len(parts) == 6)
        charm = str((apps[n] or {}).get("charm") or "")
        # 9a is asserted above, over every application. If it already flagged this app,
        # do not also run 9b on it -- 9b would report the coupling as the cause, sending
        # the reader to add v6 legs when the fix is to remove the option.
        if "prefer-ipv6" in opts and charm not in PREFER_IPV6_CHARMS:
            continue
        # 9b -- RE-POINTED 2026-07-31 (D-101 RULING NOTE (b), GA-R5: "Set it false on the
        # seven, keep every v6 VIP leg"). It previously required the option and the v6 legs
        # to travel TOGETHER on a declaring charm (the L3-9 merge-order defect). The ruling
        # removes the option from EVERY application until IPv6 is operational, so that
        # coupling would now fail on the ruled artifact. The invariant is REPLACED, not
        # deleted: the option must be ABSENT everywhere.
        #
        # Measured cause of the ruling: the LXD containers the API charms run in hold only
        # a link-local v6, and `get_relation_ip()` returns early with `get_ipv6_addr()[0]`
        # when the option is true -- so it raises and the install hook dies. The v6 legs
        # themselves are UNAFFECTED and are retained; arity is asserted by 9c below.
        #
        # WHEN IPv6 BECOMES OPERATIONAL and the option is reinstated, this returns to the
        # coupling form and the ruling note is the record of why it left. L3-9 cannot
        # recur while the option is absent everywhere -- there is nothing to keep while
        # legs are dropped.
        if "prefer-ipv6" in opts:
            fails.append("%s sets prefer-ipv6, which is RULED OFF for every application "
                         "until IPv6 is operational (D-101 RULING NOTE (b), 2026-07-31). "
                         "The v6 VIP legs stay; the option does not -- the LXD containers "
                         "hold only a link-local v6, so a charm that advertises v6 for "
                         "every relation dies in its install hook" % n); continue
        if dual:
            # Resolved PER APPLICATION and PER LEG, not once for the whole bundle. The
            # gate's question is "is this leg inside the apex band for its role", and
            # mid-D-139 legs legitimately differ in family -- provider-public's VIP /64 is
            # GUA in both worlds while metal-admin/internal are ULA until their DC's step-1
            # push lands. A FIRST cut of this fix resolved the bands ONCE from the first
            # app's legs and reused them for all thirteen; that silently graded every later
            # app against the first app's families and broke seven dc1 harness cases.
            # Caught by the harness, not by review. Cached on the family triple so the
            # record is still read once per distinct shape, not once per app.
            _famkey = tuple(_v6_family(x) if x else None for x in parts[3:6])
            if _famkey in _bandcache:
                v6_bands, v6_refusal = _bandcache[_famkey]
            else:
                v6_bands, v6_refusal = _apex_v6_bands(args.dc, parts[3:6])
                _bandcache[_famkey] = (v6_bands, v6_refusal)
        if dual and v6_bands is None:
            sys.stderr.write("ERROR: cannot evaluate %s's dual-family vip: %s\n" % (n, v6_refusal))
            return 2
        legs = [("provider", parts[0], PROVIDER),
                ("admin",    parts[1], ADMIN),
                ("internal", parts[2], INTERNAL)]
        if dual:
            legs += [("provider v6", parts[3], v6_bands["provider"]),
                     ("admin v6",    parts[4], v6_bands["admin"]),
                     ("internal v6", parts[5], v6_bands["internal"])]
        bad = False
        for lbl, addr, net in legs:
            try:
                ip = ipaddress.ip_address(addr)
            except ValueError as e:
                fails.append("%s bad vip ip: %s" % (n, e)); bad = True; break
            if ip.version != net.version:
                fails.append("%s %s leg %s is IPv%d, expected IPv%d"
                             % (n, lbl, addr, ip.version, net.version)); bad = True; break
            if ip not in net:
                fails.append("%s %s leg %s not in %s" % (n, lbl, addr, net)); bad = True; break
        if bad:
            continue
        hosts = {_host_part(p) for p in parts}
        if len(hosts) != 1:
            fails.append("%s vip host parts differ across legs -- the v6 host part must "
                         "MIRROR the v4 octet (2026-07-27 ruling): %r" % (n, vip)); continue
        h = hosts.pop()
        if not h.isdigit():
            fails.append("%s vip host part %r is not the decimal v4-octet mirror" % (n, h)); continue
        o = int(h)
        if not (OCTET_LO <= o <= OCTET_HI):
            fails.append("%s vip octet .%d outside %d-%d" % (n, o, OCTET_LO, OCTET_HI)); continue
        if h in octet_owner:
            fails.append("VIP last octet .%s shared by %s and %s" % (h, octet_owner[h], n)); continue
        octet_owner[h] = n
        vip_ok += 1
        vip_dual += 1 if dual else 0
    if vip_ok:
        oks.append("%d clustered VIP(s) are provider/admin/internal, octet %d-%d "
                   "(%d dual-family)" % (vip_ok, OCTET_LO, OCTET_HI, vip_dual))

    for n, s in apps.items():
        if (s or {}).get("charm") != "ovn-chassis":
            continue
        bim = str(((s or {}).get("options", {}) or {}).get("bridge-interface-mappings", ""))
        if not bim:
            continue
        macs = set(MAC_RE.findall(bim.lower()))
        if role_sep:
            # VR1 role-separated: the VR0 openstack0-3 Pattern-A set (D-057/D-060) does not
            # apply -- dc chassis are the compute-node provider MACs. Verify present + well-formed.
            if macs:
                oks.append("%s bridge-interface-mappings: %d well-formed MAC(s) (role-sep; VR0 set N/A)" % (n, len(macs)))
            else:
                fails.append("%s bridge-interface-mappings has no valid MAC" % n)
        else:
            missing = EXPECT_CHASSIS_MACS - macs
            if missing:
                fails.append("%s missing chassis MAC(s): %s" % (n, ", ".join(sorted(missing))))
            else:
                oks.append("%s bridge-interface-mappings: all 4 chassis MACs present (incl openstack0)" % n)

    # -- 4. relations: existing apps, explicit endpoints (the magnum-shared-db class) --
    rels = (doc or {}).get("relations", []) or []
    rel_bad = 0
    for r in rels:
        if not isinstance(r, list) or len(r) != 2:
            fails.append("relation not a 2-list: %r" % (r,)); rel_bad += 1; continue
        for side in r:
            side = str(side)
            if ":" not in side:
                fails.append("relation side lacks explicit :endpoint: %r" % side); rel_bad += 1; continue
            if side.split(":")[0] not in apps:
                fails.append("relation references unknown app: %r" % side); rel_bad += 1
    if rels and not rel_bad:
        oks.append("%d relations well-formed (explicit endpoints, all apps exist)" % len(rels))

    # -- 5. D-062: mysql-innodb-cluster at target count 3 --
    mi = apps.get("mysql-innodb-cluster") or {}
    if mi.get("num_units") != 3:
        fails.append("mysql-innodb-cluster num_units=%r (D-062 requires 3: single-unit seed never bootstraps)" % mi.get("num_units"))
    else:
        oks.append("mysql-innodb-cluster num_units=3 (D-062)")

    # -- 6. VIP octet uniqueness: folded into the family-aware loop above, which
    #       compares the HOST PART (v4 octet or final v6 group) rather than
    #       rsplit(".")-ing every token -- on a v6 literal that returned the whole
    #       address, so every v6 leg was trivially "unique" and unchecked.

    # -- 8. R11 (D-020 AMENDMENT, 2026-07-27): an hacluster principal with no VIP.
    #       Decorative HA -- the principal binds a UNIT address and pacemaker has
    #       nothing to fail over to. Nothing in scripts/ or tests/ asserts
    #       cluster_count, so this shape passed every gate until now.
    ha_principals = set()
    ha_pairs = set()          # (principal, its hacluster subordinate app)
    for r in rels:
        if not isinstance(r, list) or len(r) != 2:
            continue
        sides = [str(x) for x in r]
        if not all(x.endswith(":ha") for x in sides):
            continue
        for i, side in enumerate(sides):
            app, peer = side.split(":")[0], sides[1 - i].split(":")[0]
            if app in apps and (apps.get(peer) or {}).get("charm") == "hacluster":
                ha_principals.add(app)
                ha_pairs.add((app, peer))
    no_vip = sorted(a for a in ha_principals
                    if not ((apps.get(a) or {}).get("options") or {}).get("vip"))
    if no_vip:
        fails.append("hacluster relation but no vip: %s -- clustered principal binds a unit "
                     "address with nothing to fail over to (R11 / D-020 amendment)"
                     % ", ".join(no_vip))
    elif ha_principals:
        oks.append("%d hacluster principal(s) all carry a VIP (R11)" % len(ha_principals))

    # -- 10. HA ARITY: cluster_count == the principal's num_units.
    #       The hacluster charm configures its resources only once cluster_count peers
    #       have joined; at 3 units with cluster_count still 1 pacemaker can bootstrap
    #       the VIP off ONE node, which is decorative HA wearing a 3-unit costume.
    #       Equality catches BOTH directions -- a stale 1 under a scaled-up overlay,
    #       and a 3 that outlives a scale-down.
    cc_pf0 = len(fails)
    for principal, sub in sorted(ha_pairs):
        opts = ((apps.get(sub) or {}).get("options") or {})
        if "cluster_count" not in opts:
            fails.append("%s (hacluster for %s) declares NO cluster_count -- the quorum "
                         "size must be explicit; a charm default is not a declaration"
                         % (sub, principal)); continue
        cc = opts.get("cluster_count")
        # `num_units` ABSENT means juju deploys exactly ONE unit. Written out rather
        # than defaulted silently, because "1" here is a juju behaviour, not a guess.
        nu = (apps.get(principal) or {}).get("num_units", 1)
        if not isinstance(cc, int) or isinstance(cc, bool):
            fails.append("%s cluster_count=%r is not an integer -- refusing to compare "
                         "it with %s num_units" % (sub, cc, principal)); continue
        if not isinstance(nu, int) or isinstance(nu, bool):
            fails.append("%s num_units=%r is not an integer -- refusing to compare it "
                         "with %s cluster_count" % (principal, nu, sub)); continue
        if cc != nu:
            fails.append("%s cluster_count=%d but principal %s num_units=%d -- DECORATIVE "
                         "HA: pacemaker forms (or waits for) the wrong quorum and can bind "
                         "the VIP off a single node (D-121 / R11)" % (sub, cc, principal, nu))
    if ha_pairs and len(fails) == cc_pf0:
        oks.append("%d hacluster subordinate(s) declare cluster_count == principal "
                   "num_units" % len(ha_pairs))

    # -- 7. DOCFIX-071: keystone policy ships in-bundle, zip content matches source --
    import os, zipfile
    ks = apps.get("keystone") or {}
    res = ((ks.get("resources") or {}).get("policyd-override"))
    upo = ((ks.get("options") or {}).get("use-policyd-override"))
    if not upo:
        fails.append("keystone use-policyd-override is not true (D-051)")
    if not res:
        fails.append("keystone has no resources: policyd-override (DOCFIX-071: policy unreachable on redeploy)")
    else:
        base = os.path.dirname(os.path.abspath(path))
        zp = os.path.normpath(os.path.join(base, str(res)))
        src = os.path.join(base, "policies", "domain-manager-policy.yaml")
        if not os.path.isfile(zp):
            fails.append("policyd-override zip missing at %s" % zp)
        elif not os.path.isfile(src):
            fails.append("policy source missing at %s" % src)
        else:
            try:
                with zipfile.ZipFile(zp) as z:
                    inzip = z.read("domain-manager-policy.yaml")
                ondisk = open(src, "rb").read()
                if inzip != ondisk:
                    fails.append("policyd-override zip content DIFFERS from policies/domain-manager-policy.yaml (rebuild + recommit the zip)")
                else:
                    oks.append("keystone policyd-override wired in-bundle; zip content matches source (DOCFIX-071)")
            except KeyError:
                fails.append("zip lacks top-level domain-manager-policy.yaml (keystone reads the top-level name)")
            except Exception as e:
                fails.append("cannot read policyd zip: %s" % e)

    # -- 11. MACHINES-BLOCK DC IDENTITY. The machines `tags=` is the ONLY thing keeping
    #        one DC's model off the other DC's hardware (one MAAS region, both DCs --
    #        D-104/D-123). A machines overlay that parses but contributes nothing
    #        (gutted, or keyed "01" where the base says "0") used to be a byte-identical
    #        PASS.
    #        DELIBERATELY OUTSIDE the role_sep gate below, and keyed on `machines` alone.
    #        Keying it on role tags was ITSELF a bypass: an overlay that rewrites every
    #        machine's constraints to a bare `arch=amd64` strips the role tags AND the DC
    #        tag together, role_sep goes False, and the whole placement block -- these
    #        checks included -- self-skipped behind an [ok]. A bundle with NO machines
    #        block still skips (zero iterations), which is the VR0 shape.

    # ---- invariant 12: every application carries an explicit base == default-base ----
    # RULED 2026-07-31 (GA-R5 option D). Without this, deleting the 56 `base:` lines is a
    # byte-identical PASS and the once-only-deployable trap returns at the next partial
    # failure. Keyed on the bundle's OWN default-base rather than a literal, so a future
    # base change moves one line and this still holds. REFUSES if default-base is absent:
    # there would be nothing to compare against, and "could not look" is not "nothing
    # there".
    want_base = doc.get("default-base")
    if not want_base:
        fails.append("bundle declares no top-level default-base -- invariant 12 cannot be "
                     "evaluated, and an application base that matches nothing is not a pass")
    else:
        nobase = sorted(n for n in apps if not ((apps[n] or {}).get("base")))
        wrongbase = sorted(n for n in apps
                           if (apps[n] or {}).get("base")
                           and (apps[n] or {}).get("base") != want_base)
        if nobase:
            fails.append("%d application(s) carry NO explicit base -- a bundle relying on "
                         "default-base alone is deployable exactly ONCE and cannot be "
                         "resumed after a partial failure (2026-07-31 attempt 2): %s"
                         % (len(nobase), ", ".join(nobase)))
        if wrongbase:
            fails.append("%d application(s) carry a base that is not the bundle's "
                         "default-base %r: %s"
                         % (len(wrongbase), want_base,
                            ", ".join("%s=%r" % (n, apps[n]["base"]) for n in wrongbase)))
        if not nobase and not wrongbase:
            print("  [ok]   all %d application(s) carry an explicit base == default-base "
                  "%r (re-runnable after a partial deploy)" % (len(apps), want_base))

    if machines:
        want_tag = DC_MACHINE_TAGS[args.dc]
        dc_tags = {}
        for mid, spec in machines.items():
            t = _dc_tag_of(spec)
            if t is None:
                fails.append("machine %s carries no single DC tag in its constraints "
                             "(expected one of %s) -- an untagged machine can be "
                             "allocated from EITHER DC" % (mid, "/".join(sorted(DC_MACHINE_TAGS.values()))))
            else:
                dc_tags[str(mid)] = t
        seen = sorted(set(dc_tags.values()))
        if len(seen) > 1:
            fails.append("machines block MIXES DC tags %s -- the signature of a mis-keyed "
                         "machines overlay whose entries land beside the base's instead of "
                         "over them (both sets present, only one referenced)" % ", ".join(seen))
        elif seen and seen[0] != want_tag:
            fails.append("machines block is tagged %s but --dc %s was requested -- deploying "
                         "this input would allocate %s against the OTHER DC's nodes (the "
                         "machines overlay is absent, gutted, or not merged)"
                         % (seen[0], args.dc, args.dc))
        elif seen:
            oks.append("machines block: all %d machine(s) tagged %s, matching --dc %s"
                       % (len(dc_tags), want_tag, args.dc))

    # -- 8. Placement / anti-affinity (role separation). Activates ONLY when the
    #       machines block carries control/compute/storage role tags; a base or
    #       un-rendered bundle has none, so the checks self-skip (base stays green).
    #       Run on the MERGED bundle (base + overlays) to see the scaled apps' real
    #       num_units/to:. Catches the decorative-HA paths juju itself won't error on.
    if role_sep:
        pf0 = len(fails)
        n_ctl = sum(1 for r in roles.values() if r == "control")
        n_cmp = sum(1 for r in roles.values() if r == "compute")
        n_stg = sum(1 for r in roles.values() if r == "storage")
        placed = set()
        for n, s in apps.items():
            s = s or {}
            tos = [str(t) for t in (s.get("to") or [])]
            if not tos:
                continue
            ids = [t.split(":")[-1] for t in tos]
            placed.update(ids)
            for t, mid in zip(tos, ids):                       # a. no dangling placement
                if mid not in roles:
                    fails.append("%s to: references undefined machine %r" % (n, t))
            want = ("storage" if n == "ceph-osd" else "compute" if n == "nova-compute"
                    else "control" if any(t.startswith("lxd:") for t in tos) else None)
            if want:                                           # b. role placement
                off = [t for t, mid in zip(tos, ids) if roles.get(mid) != want]
                if off:
                    fails.append("%s placed off-role (want %s node): %s" % (n, want, ", ".join(off)))
            nu = s.get("num_units")                            # d. anti-affinity
            if isinstance(nu, int) and nu >= 2 and len(set(ids)) < nu:
                fails.append("%s num_units=%d but to: has %d distinct machine(s) -- anti-affinity defeat: %s"
                             % (n, nu, len(set(ids)), ids))
        orphans = sorted(m for m in map(str, machines) if m not in placed)
        if orphans:                                        # e. no inert machine entries
            fails.append("machine(s) %s are DEFINED but no application places `to:` them "
                         "-- an overlay whose machine keys do not match the base's adds "
                         "inert entries and overrides NOTHING" % ", ".join(orphans))
        for n, want, lbl in (("ceph-osd", n_stg, "storage"), ("nova-compute", n_cmp, "compute")):
            nu = (apps.get(n) or {}).get("num_units")          # c. bare-metal role counts
            if isinstance(nu, int) and nu != want:
                fails.append("%s num_units=%d != %d %s node(s)" % (n, nu, want, lbl))
        if len(fails) == pf0:
            oks.append("placement: role-separated (%d control/%d compute/%d storage); "
                       "anti-affinity + role placement + counts OK" % (n_ctl, n_cmp, n_stg))
    else:
        oks.append("placement: no role tags in machines block (base/un-rendered bundle) -- checks skipped")

    for o in oks:   print("  [ok]   %s" % o)
    for f in fails: print("  [FAIL] %s" % f)
    print("\n%s: Pattern A / D-052-D-053 bundle invariants (%s)" % ("PASS" if not fails else "FAIL", path))
    return 1 if fails else 0

if __name__ == "__main__":
    sys.exit(main())
