# `checks.module-eval-swarm-otel-identity` — see ./lib.nix for the shared
# rationale (why this suite exists, naming convention, "evaluates
# not executes").
{
  pkgs,
  lib,
  self,
  nixosSystem,
  shared,
}:
let
  inherit
    (import ./lib.nix {
      inherit
        pkgs
        lib
        self
        nixosSystem
        ;
    })
    hive
    runGroup
    otelSettings
    ;

  # A collector holding no store identity at all. Unlike Grafana's mirror
  # image, this is not a refused shape: the collector still receives every
  # hive's telemetry with nothing to push authenticated with, which is the
  # already-supported degrade `haveCollectorSecret` names above the module's
  # `let`. What this fixture is for is checking the reading unit itself does
  # not render, rather than rendering with an env var nothing filled in.
  otelNoIdentity = hive {
    deploy.swarm-otel.enable = true;
    swarm.authelia.url = "https://auth.example.invalid";
    deploy.forgejo.sso.clientSecretFile = "/var/lib/forgejo-oidc/by-hand.secret";
  };

  # 🩸 The shape the degrade above must not swallow: a collector on a host that
  # demonstrably READS the store — it holds the hive's own leaf — and is missing
  # only the collector's own pair. Before the four-way split this host rendered
  # the reading unit off `clientCertFile` alone, so the silence it gets now is a
  # regression rather than the supported degrade `otelNoIdentity` stands for. The
  # two fixtures differ in exactly that one pair, which is what lets the arms
  # below separate a refusal from a degrade.
  inherit (shared) otelReaderMissingLeaf;

  # The same deployment named in full, which must stay buildable. Pairs with the
  # fixture above so the refusal is pinned to the omission and not to reading a
  # remote store at all.
  inherit (shared) otelReaderNamedInFull;

  # Did the module refuse this host, and over which option. An assertion is a
  # config VALUE until something forces it — `.config` never throws — so a
  # fixture in a refused state stays evaluable and the refusal reads back as
  # data. Matched on the option name the message names rather than on its prose:
  # the option name is the part an operator has to act on, and a message that
  # stopped naming it would be the actual defect.
  refusedOver = m: option: lib.any (a: !a.assertion && lib.hasInfix option a.message) m.assertions;

  # authelia somewhere else, the credential delivered by hand. Whether this
  # collector authenticates must follow the credential, never another
  # service's placement.
  #
  # The `swarm.otel.clientSecretFile` below is the PRE-RENAME path. It predates
  # the split and is deliberately left spelled that way: it makes this fixture
  # the old-path case for that option too, so dropping its rename entry fails
  # the eval here rather than only in a real operator's config.
  otelRemoteAuthelia = hive {
    deploy.swarm-otel.enable = true;
    deploy.authelia.enable = false;
    # Where that elsewhere IS. Running no IdP does not mean knowing no IdP:
    # the authenticator this fixture exists to render puts this address in its
    # `token_url`, so a hive with a secret and no URL has a credential it can
    # present nowhere.
    swarm.authelia.url = "https://auth.example.invalid";
    swarm.otel.clientSecretFile = "/var/lib/swarm-otel-oidc/by-hand.secret";
  };

  # A collector whose ONLY scrape work is published: loopback targets forced
  # empty, published jobs only. Unreachable in a real deploy today —
  # the module seeds `scrapeTargets.collector` under its own `enable`, so the
  # loopback set is never empty on its own — which is exactly why the arm
  # below needs a fixture that takes that seeding away. `mkForce` is what
  # does it, and it leaves the collector itself enabled: the state under test
  # is a running collector with no self-scrape, not an absent one.
  otelOnlyPublished = hive {
    deploy.swarm-otel.enable = true;
    swarm.otel.scrapeTargets = lib.mkForce { };
    swarm.otel.publishedScrapeTargets.remote = "https://remote.t.local/metrics";
  };

  # Two hives in the roster, which no other fixture here has: every one of
  # them declares `swarm.hives.h1` alone, so a per-hive arm written against
  # one of those passes on a hardcoded literal.
  otelTwoHives = hive {
    deploy.swarm-otel.enable = true;
    deploy.authelia.enable = true;
    swarm.hives.h2.domain = "h2.t.local";
  };

  # authelia with neither the forge nor the collector beside it.
  inherit (shared) autheliaOnly;

  # Every swarm service on one host, the collector and authelia included.
  inherit (shared) allLocal;

  forgeMetricsUrl = "https://forge.t.local/metrics";
  autheliaMetricsUrl = "https://auth.t.local/metrics";
  logsMetricsUrl = "https://logs.t.local/metrics";
  vmMetricsUrl = "https://metrics.t.local/metrics";

  # The policy authelia applies to `subject` requesting `https://<domain><path>`,
  # where `path` is what authelia matches `resources` against: the cleaned path
  # plus `?<query>` when there is one.
  # It is the first rule whose domain, resources and subject all match, else
  # `default_policy`. ⚠️ A rule with a `subject` list does NOT match a subject
  # outside it — authelia moves on to the next rule — which is the semantics a
  # check of the rule's own fields cannot see.
  autheliaPolicy =
    m: subject: domain: path:
    let
      ac = m.containers.swarm-authelia.config.services.authelia.instances.swarm.settings.access_control;
      matches =
        r:
        r.domain == domain
        && lib.any (re: builtins.match re path != null) (r.resources or [ ".*" ])
        && (!(r ? subject) || builtins.elem subject r.subject);
      hit = lib.findFirst matches null ac.rules;
    in
    if hit == null then ac.default_policy else hit.policy;

  cases = [
    {
      # The collector's non-assertion, the deliberate mirror of Grafana's
      # assertion two cases up: a host with no store identity is a supported,
      # merely degraded shape here, so the reading unit simply does not exist
      # rather than refusing the build. `haveCollectorSecret` is what the
      # degrade already reads, unchanged by this slice.
      name = "a collector with no store identity renders no reading unit, and is not refused";
      ok =
        !(otelNoIdentity.systemd.services ? swarm-bao-otel-oidc)
        && otelNoIdentity.services.hyperhive.deploy.swarm-otel.clientSecretFile == null
        && !(lib.any (a: !a.assertion) otelNoIdentity.assertions);
    }
    {
      # The collector's half of the same split, and a different arm from the
      # authenticator case below: this one reads the PATH the unit loads, so a
      # reader left on a source that is non-null but wrong still fails. The
      # fixture spells the option its pre-rename way, so it covers the rename
      # entry at the same time.
      name = "a config written against the pre-rename otel secret path still loads it as a credential";
      ok =
        lib.any (c: lib.hasInfix "/var/lib/swarm-otel-oidc/by-hand.secret" c)
          otelRemoteAuthelia.containers.swarm-otel.config.systemd.services.opentelemetry-collector.serviceConfig.LoadCredential;
    }
    {
      # The collector authenticates because it HOLDS a credential, not because
      # authelia happens to share its host. Gating on the other service's
      # placement renders a collector that pushes unauthenticated wherever
      # authelia lives elsewhere — one of the supported shapes.
      name = "a collector with a hand-delivered secret authenticates without authelia beside it";
      ok =
        let
          s = otelSettings otelRemoteAuthelia;
        in
        (s.exporters."otlphttp/victoriametrics" ? auth)
        && builtins.elem "oauth2client/victoriametrics" s.service.extensions;
    }
    {
      # Defining a receiver and attaching it are two separate lists, and the
      # two gates were spelled differently: the receiver appeared for either
      # scrape option, the pipeline only for the loopback one. A published-
      # only collector therefore rendered scrape configs that reached no
      # pipeline — requested, parsed, delivered nowhere, and valid enough to
      # deploy. The receiver clause is what stops the arm passing for the
      # wrong reason, by an empty `prometheus` never rendering at all.
      name = "a published-only collector attaches its prometheus receiver to the swarm pipeline";
      ok =
        let
          s = otelSettings otelOnlyPublished;
        in
        otelOnlyPublished.services.hyperhive.swarm.otel.scrapeTargets == { }
        && otelOnlyPublished.services.hyperhive.swarm.otel.publishedScrapeTargets != { }
        && (s.receivers ? prometheus)
        && builtins.elem "prometheus" s.service.pipelines."metrics/swarm".receivers;
    }
    {
      # The collector reads its own host's target list, so a forge entry
      # declared only where the forge runs is a job no split deploy renders.
      # `otelRemoteAuthelia` runs neither the forge nor authelia.
      name = "a collector on a host without the forge scrapes the forge by name";
      ok =
        let
          s = otelSettings otelRemoteAuthelia;
          job = lib.findFirst (j: j.job_name == "forgejo") null s.receivers.prometheus.config.scrape_configs;
        in
        !otelRemoteAuthelia.services.hyperhive.deploy.forgejo.enable
        && job != null
        && job.static_configs == [ { targets = [ "forge.t.local" ]; } ]
        && job.metrics_path == "/metrics"
        && job.oauth2.endpoint_params.audience == forgeMetricsUrl
        && builtins.elem "prometheus" s.service.pipelines."metrics/swarm".receivers;
    }
    {
      # The other half: authelia refuses a token for an audience its client
      # was not registered for, and registers it from its OWN host's list.
      name = "authelia on a host without the forge or the collector admits the collector's forge scrape";
      ok =
        let
          h = autheliaOnly.services.hyperhive;
          client = lib.findFirst (c: c.id == h.swarm.otel.clientId) null h.swarm.authelia.oidc.clients;
          rules =
            autheliaOnly.containers.swarm-authelia.config.services.authelia.instances.swarm.settings.access_control.rules;
        in
        !h.deploy.forgejo.enable
        && !h.deploy.swarm-otel.enable
        && client != null
        && builtins.elem forgeMetricsUrl client.audience
        && builtins.elem {
          domain = "forge.t.local";
          resources = [ "^/metrics([?].*)?$" ];
          policy = "one_factor";
          subject = [ "oauth2:client:${h.swarm.otel.clientId}" ];
        } rules;
    }
    {
      # Authelia's half of the same move: its target is published by name, so
      # a collector anywhere in the swarm has a job for it.
      name = "a collector on a host without authelia scrapes authelia by name";
      ok =
        let
          s = otelSettings otelRemoteAuthelia;
          job = lib.findFirst (j: j.job_name == "authelia") null s.receivers.prometheus.config.scrape_configs;
        in
        !otelRemoteAuthelia.services.hyperhive.deploy.authelia.enable
        && job != null
        && job.static_configs == [ { targets = [ "auth.t.local" ]; } ]
        && job.metrics_path == "/metrics"
        && job.oauth2.endpoint_params.audience == autheliaMetricsUrl;
    }
    {
      # The location the collector's job requests: gated, and proxying to the
      # loopback listener that has no authentication of its own.
      name = "authelia's vhost serves /metrics only behind the collector's auth subrequest";
      ok =
        let
          h = autheliaOnly.services.hyperhive;
          locs = autheliaOnly.services.nginx.virtualHosts.${h.swarm.authelia.domain}.locations;
          authz = locs."= /__authelia_metrics_authz";
        in
        locs."= /metrics".proxyPass == "http://127.0.0.1:${toString h.swarm.authelia.metricsPort}/metrics"
        && lib.hasInfix "auth_request /__authelia_metrics_authz;" locs."= /metrics".extraConfig
        && authz.proxyPass == "https://${h.swarm.authelia.domain}/api/authz/auth-request"
        && lib.hasInfix "internal;" authz.extraConfig;
    }
    {
      # The store's half of the same move, on a collector host that does not
      # run the store.
      name = "a collector on a host without the metrics store scrapes the store by name";
      ok =
        let
          s = otelSettings otelRemoteAuthelia;
          job = lib.findFirst (
            j: j.job_name == "victoriametrics"
          ) null s.receivers.prometheus.config.scrape_configs;
        in
        !otelRemoteAuthelia.services.hyperhive.deploy.victoriametrics.enable
        && job != null
        && job.static_configs == [ { targets = [ "metrics.t.local" ]; } ]
        && job.metrics_path == "/metrics"
        && job.oauth2.endpoint_params.audience == vmMetricsUrl;
    }
    {
      # The metrics vhost is default-deny; `/metrics` reaches the store only
      # through its own exact location, gated like the push route. The `/`
      # clauses pin that the catch-all still refuses every other path.
      name = "the metrics store's vhost serves /metrics only behind the collector's auth subrequest";
      ok =
        let
          h = allLocal.services.hyperhive;
          locs = allLocal.services.nginx.virtualHosts.${h.swarm.victoriametrics.domain}.locations;
          m = locs."= /metrics" or null;
        in
        m != null
        && m.proxyPass == "http://127.0.0.1:${toString h.swarm.victoriametrics.port}/metrics"
        && lib.hasInfix "auth_request /__metrics_push_authz;" m.extraConfig
        && !(lib.hasInfix "error_page" m.extraConfig)
        && lib.hasInfix "internal;" locs."= /__metrics_push_authz".extraConfig
        && locs."/".return == "403"
        && locs."/".proxyPass == null;
    }
    {
      # The metrics store's name admits the collector's client on every path
      # and refuses every other subject. The rules carry no `resources`, so no
      # path or query spelling misses them and reaches `default_policy`, and
      # they come before the apex rule. That is only right while the push
      # route and `/metrics` are the vhost's sole authelia-gated locations,
      # which the last clause pins.
      name = "authelia admits only the collector anywhere on the metrics store's name";
      ok =
        let
          h = allLocal.services.hyperhive;
          collector = "oauth2:client:${h.swarm.otel.clientId}";
          rules =
            allLocal.containers.swarm-authelia.config.services.authelia.instances.swarm.settings.access_control.rules;
          indexed = lib.imap0 (i: r: r // { inherit i; }) rules;
          storeRules = lib.filter (r: r.domain == "metrics.t.local") indexed;
          apex = lib.findFirst (r: r.domain == "t.local") null indexed;
          policy = subject: autheliaPolicy allLocal subject "metrics.t.local";
          paths = [
            "/opentelemetry/api/v1/push"
            "/metrics"
            "/metrics?x=1"
            "/api/v1/query"
          ];
          locs = allLocal.services.nginx.virtualHosts.${h.swarm.victoriametrics.domain}.locations;
        in
        map (r: removeAttrs r [ "i" ]) storeRules == [
          {
            domain = "metrics.t.local";
            policy = "one_factor";
            subject = [ collector ];
          }
          {
            domain = "metrics.t.local";
            policy = "deny";
          }
        ]
        && apex != null
        && lib.all (r: r.i < apex.i) storeRules
        && lib.all (p: policy collector p == "one_factor") paths
        && lib.all (p: policy "user:someone" p == "deny") paths
        && lib.all (p: policy "group:admins" p == "deny") paths
        &&
          lib.hasInfix "auth_request /__metrics_push_authz;"
            locs."= /opentelemetry/api/v1/push".extraConfig
        &&
          lib.attrNames (lib.filterAttrs (_: l: lib.hasInfix "auth_request" (l.extraConfig or "")) locs) == [
            "= /metrics"
            "= /opentelemetry/api/v1/push"
          ];
    }
    {
      # Every published target admits the collector's client and refuses any
      # other authenticated subject, with or without a query string. Read
      # through authelia's first-match semantics: a client-scoped rule alone
      # lets an operator's session fall through to `default_policy = one_factor`,
      # and so does a request the rule's regex does not match.
      name = "every published metrics URL admits the collector and refuses a plain session";
      ok =
        let
          h = autheliaOnly.services.hyperhive;
          collector = "oauth2:client:${h.swarm.otel.clientId}";
          targets = lib.concatMap (
            url:
            let
              t = builtins.match "https://([^/]+)(/.*)" url;
            in
            map
              (q: [
                (builtins.elemAt t 0)
                "${builtins.elemAt t 1}${q}"
              ])
              [
                ""
                "?x=1"
              ]
          ) (lib.attrValues h.swarm.otel.publishedScrapeTargets);
          policy =
            subject: t: autheliaPolicy autheliaOnly subject (builtins.elemAt t 0) (builtins.elemAt t 1);
        in
        lib.sort lib.lessThan (lib.attrValues h.swarm.otel.publishedScrapeTargets) == [
          autheliaMetricsUrl
          forgeMetricsUrl
          logsMetricsUrl
          vmMetricsUrl
        ]
        && lib.length targets == 8
        && lib.all (t: policy collector t == "one_factor") targets
        && lib.all (t: policy "user:someone" t == "deny") targets
        && lib.all (t: policy "group:admins" t == "deny") targets;
    }
    {
      # Which request forms reach `= /metrics`, and whether the rule's regex
      # matches the path authelia derives from each. `nginx` records the
      # location each form lands in (measured against a stand-alone nginx);
      # `path` is authelia's cleaned path plus query for that form. The
      # invariant: every form the gated location serves is matched by the rule,
      # so none of them reaches `default_policy`.
      name = "the metrics rule matches every request form the gated location serves";
      ok =
        let
          rules =
            autheliaOnly.containers.swarm-authelia.config.services.authelia.instances.swarm.settings.access_control.rules;
          metricsRules = lib.filter (r: r ? resources) rules;
          re = lib.head (lib.head metricsRules).resources;
          table = [
            {
              request = "/metrics";
              nginx = "= /metrics";
              path = "/metrics";
            }
            {
              request = "/metrics?x=1";
              nginx = "= /metrics";
              path = "/metrics?x=1";
            }
            {
              request = "//metrics";
              nginx = "= /metrics";
              path = "/metrics";
            }
            {
              request = "/metrics/";
              nginx = "= /metrics/ (forge: 404) or /";
              path = "/metrics/";
            }
            {
              request = "/metricsx";
              nginx = "/";
              path = "/metricsx";
            }
          ];
          matched = row: builtins.match re row.path != null;
        in
        lib.length metricsRules == 6
        && lib.all (r: r.resources == [ re ]) metricsRules
        && re == "^/metrics([?].*)?$"
        && lib.all (row: matched row == (row.nginx == "= /metrics")) table;
    }
    {
      # Forgejo strips trailing slashes before routing, so `/metrics/` reaches
      # its metrics handler unless nginx refuses it first.
      name = "the forge refuses /metrics/ rather than proxying it to forgejo's metrics";
      ok =
        let
          locs =
            allLocal.services.nginx.virtualHosts.${allLocal.services.hyperhive.swarm.forge.domain}.locations;
        in
        (locs."= /metrics/".return or null) == "404";
    }
    {
      # Single host: authelia is scraped once, by its published name. A
      # loopback job beside it would scrape the same listener twice — and share
      # its job name, which ./swarm-otel.nix refuses.
      name = "a host running authelia and the collector scrapes authelia once, by name";
      ok =
        let
          o = allLocal.services.hyperhive.swarm.otel;
          jobs =
            lib.filter (j: j.job_name == "authelia")
              (otelSettings allLocal).receivers.prometheus.config.scrape_configs;
        in
        allLocal.services.hyperhive.deploy.authelia.enable
        && allLocal.services.hyperhive.deploy.swarm-otel.enable
        && !(o.scrapeTargets ? authelia)
        && (o.publishedScrapeTargets.authelia or null) == autheliaMetricsUrl
        && lib.length jobs == 1
        && (lib.head jobs).static_configs == [ { targets = [ "auth.t.local" ]; } ];
    }
    {
      # Same for the metrics store: one job, by its published name.
      name = "a host running the metrics store and the collector scrapes the store once, by name";
      ok =
        let
          o = allLocal.services.hyperhive.swarm.otel;
          jobs =
            lib.filter (j: j.job_name == "victoriametrics")
              (otelSettings allLocal).receivers.prometheus.config.scrape_configs;
        in
        allLocal.services.hyperhive.deploy.victoriametrics.enable
        && allLocal.services.hyperhive.deploy.swarm-otel.enable
        && !(o.scrapeTargets ? victoriametrics)
        && (o.publishedScrapeTargets.victoriametrics or null) == vmMetricsUrl
        && lib.length jobs == 1
        && (lib.head jobs).static_configs == [ { targets = [ "metrics.t.local" ]; } ];
    }
    {
      # The case above reads the options; this one reads the assertion they
      # feed, which is what refuses an operator's `nixos-rebuild`.
      name = "a single-host swarm passes the loopback-vs-published job name assertion";
      ok = !(refusedOver allLocal "is declared as both a loopback");
    }
    {
      name = "a collector on a host without the log store scrapes it by name";
      ok =
        let
          s = otelSettings otelRemoteAuthelia;
          job = lib.findFirst (
            j: j.job_name == "victorialogs"
          ) null s.receivers.prometheus.config.scrape_configs;
        in
        !otelRemoteAuthelia.services.hyperhive.deploy.victorialogs.enable
        && job != null
        && job.static_configs == [ { targets = [ "logs.t.local" ]; } ]
        && job.metrics_path == "/metrics"
        && job.oauth2.endpoint_params.audience == logsMetricsUrl;
    }
    {
      # `= /metrics` must carry the bare authelia check and nothing else: the
      # agents' `/__swarm_logs_auth` would admit an agent's bearer, and the
      # browser block's `error_page 401 =302` hands a scraper a login page.
      # The other routes keep their own checks.
      name = "the log store's vhost serves /metrics behind authelia alone, its other routes unchanged";
      ok =
        let
          h = allLocal.services.hyperhive;
          locs = allLocal.services.nginx.virtualHosts.${h.swarm.victorialogs.domain}.locations;
          checksOf = l: lib.filter (lib.hasPrefix "auth_request") (lib.splitString "\n" l.extraConfig);
        in
        h.deploy.victorialogs.enable
        && locs ? "= /metrics"
        && locs."= /metrics".proxyPass == "http://127.0.0.1:${toString h.swarm.victorialogs.port}/metrics"
        && checksOf locs."= /metrics" == [ "auth_request /__hive_authelia;" ]
        && !(lib.hasInfix "error_page" locs."= /metrics".extraConfig)
        && lib.hasInfix "error_page 401 =302" locs."/".extraConfig
        && lib.elem "auth_request /__hive_authelia;" (checksOf locs."/")
        && checksOf locs."^~ /select/logsql/" == [ "auth_request /__swarm_logs_auth;" ];
    }
    {
      # The `deny` that confines the log store's `/metrics` is scoped to that
      # path, and comes after the collector's rule: every other path on the
      # name still reaches `default_policy`, as it did with no rule at all.
      name = "authelia confines the log store's /metrics to the collector and leaves its other paths alone";
      ok =
        let
          h = autheliaOnly.services.hyperhive;
          collector = "oauth2:client:${h.swarm.otel.clientId}";
          rules =
            autheliaOnly.containers.swarm-authelia.config.services.authelia.instances.swarm.settings.access_control.rules;
          logsRules = lib.filter (r: r.domain == "logs.t.local") rules;
          policy = subject: autheliaPolicy autheliaOnly subject "logs.t.local";
        in
        map (r: r.policy) logsRules == [
          "one_factor"
          "deny"
        ]
        && lib.all (r: r.resources == [ "^/metrics([?].*)?$" ]) logsRules
        && (lib.head logsRules).subject == [ collector ]
        && policy collector "/metrics" == "one_factor"
        && policy "user:someone" "/metrics" == "deny"
        && policy "user:someone" "/metrics?x=1" == "deny"
        && policy "user:someone" "/" == "one_factor"
        && policy "user:someone" "/select/logsql/query" == "one_factor"
        && policy "user:someone" "/insert/opentelemetry/v1/logs" == "one_factor";
    }
    {
      # Single host: the store is scraped once, by its published name.
      name = "a host running the log store and the collector scrapes it once, by name";
      ok =
        let
          o = allLocal.services.hyperhive.swarm.otel;
          jobs =
            lib.filter (j: j.job_name == "victorialogs")
              (otelSettings allLocal).receivers.prometheus.config.scrape_configs;
        in
        allLocal.services.hyperhive.deploy.victorialogs.enable
        && allLocal.services.hyperhive.deploy.swarm-otel.enable
        && !(o.scrapeTargets ? victorialogs)
        && lib.length jobs == 1
        && (lib.head jobs).static_configs == [ { targets = [ "logs.t.local" ]; } ];
    }
    {
      # Read against the roster the fixture declares rather than against
      # names spelled here: an arm naming `h1` passes on a single-hive
      # config however the mapping is written. The length clause is what
      # makes the `all` mean anything — over an empty roster it holds
      # vacuously.
      name = "the swarm collector routes every hive's logs, not just one";
      ok =
        let
          p = (otelSettings otelTwoHives).service.pipelines;
          hives = lib.attrNames otelTwoHives.services.hyperhive.swarm.hives;
        in
        lib.length hives == 2
        && lib.all (h: (p ? "logs/${h}") && p."logs/${h}".receivers == [ "otlp/${h}" ]) hives;
    }
    {
      # The same split as the metrics case above — defining an exporter and
      # naming it are two lists — plus the half one shared list cannot have:
      # the metrics store's exporter renders perfectly well inside a logs
      # pipeline and posts journal records at an ingest route that is not
      # for them.
      name = "every logs pipeline sends to the log store and to no metrics one";
      ok =
        let
          s = otelSettings otelTwoHives;
          logPipes = lib.filterAttrs (n: _: lib.hasPrefix "logs/" n) s.service.pipelines;
          used = lib.unique (lib.concatMap (p: p.exporters) (lib.attrValues logPipes));
        in
        logPipes != { }
        && builtins.elem "otlphttp/victorialogs" used
        && !(builtins.elem "otlphttp/victoriametrics" used)
        && lib.all (e: s.exporters ? ${e}) used;
    }
    {
      # 🩸 The refusal, and the only way this module's omission is visible at
      # all: the reading unit is gated on its own leaf, so the regression is a
      # green build with the unit absent — and the missing unit is the evidence.
      # Read as a refusal naming both options, so an operator acts on the
      # message without opening the nix.
      name = "a store reader missing the collector's own leaf is refused, naming both options";
      ok =
        refusedOver otelReaderMissingLeaf "otelOidcClientCertFile"
        && refusedOver otelReaderMissingLeaf "otelOidcClientKeyFile";
    }
    {
      # ⚠️ The arm that keeps the refusal from eating the degrade beside it. A
      # collector on a host with NO store identity still receives every hive's
      # telemetry and is a supported deployment — `otelNoIdentity` is that
      # fixture, and it differs from the refused one only in the hive's own
      # leaf. Widening the gate to the collector's leaf alone would reject it,
      # which is what this pins.
      name = "a collector with no store identity at all is not refused over the collector's leaf";
      ok =
        !(refusedOver otelNoIdentity "otelOidcClientCertFile")
        && !(refusedOver otelNoIdentity "otelOidcClientKeyFile");
    }
    {
      # The other two shapes that must stay buildable: the same remote reader
      # named in full, and a host that mints its own PKI and therefore has all
      # eight as defaults from ./host-modules/glue-bao-tls.nix.
      name = "neither a fully-named remote reader nor a store host is refused";
      ok =
        !(refusedOver otelReaderNamedInFull "otelOidcClientCertFile")
        && !(refusedOver otelRemoteAuthelia "otelOidcClientCertFile");
    }
    {
      # The store's units exist only on its host; `Requires=` on an absent one
      # fails the fetch's start. The arm where the store is here is in
      # ./swarm-services-switch.nix.
      name = "the remote reader's fetch names none of the store's units";
      ok =
        let
          u = otelReaderNamedInFull.systemd.services.swarm-bao-otel-oidc;
          storeUnits = [
            "swarm-bao-pki.service"
            "container@${otelReaderNamedInFull.services.hyperhive.swarm.bao.machine}.service"
          ];
        in
        !(lib.any (n: lib.elem n (u.after ++ u.wants ++ u.requires)) storeUnits);
    }
  ];
in
runGroup "swarm-otel-identity" cases