T8989: static_routes module

* T8989 static_routes

* T8989 static_routes

* T8989 added changelog fragment for static routes module

* t8989 static route changelog

* t8989 static_routes module

* T8989: static_routes SIT overridden and rtt

* T8989: SIT updated

* T8989: vyos_static_routes dict_op refactor

2 files changed, 821 insertions, 95 deletions

diff --git a/plugins/module_utils/vyos.py b/plugins/module_utils/vyos.py
index 82670e7..a37a0c2 100644
--- a/plugins/module_utils/vyos.py
+++ b/plugins/module_utils/vyos.py
@@ -5,15 +5,6 @@ Provides ``get_config()``, ``apply_commands()``, and ``save_config()``
 on top of ``VyOSRestClient``, so resource modules can work with simple
 ``("set", path)`` / ``("delete", path)`` command tuples rather than
 calling the REST client directly.
-
-Version-adaptive paths
-----------------------
-Some VyOS config paths changed between minor versions (e.g. the NTP
-``allow-client address`` node was removed in 1.5+).  ``apply_commands``
-handles this transparently: if a ``set`` command is rejected by the device
-it retries with the last path segment removed (one level up), covering
-the most common schema simplifications.  A ``delete`` that fails is
-treated as a no-op (already absent).
 """
 
 from __future__ import absolute_import, division, print_function
@@ -27,30 +18,361 @@ from ansible_collections.vyos.rest.plugins.module_utils.vyos_rest import (
 )
 
 
-class VyOSModule:
-    """Thin wrapper around VyOSRestClient for resource modules.
+# ---------------------------------------------------------------------------
+# Legacy dynamic config utilities (used by Wave 1-3 modules)
+# ---------------------------------------------------------------------------
+
+
+def _kebab_to_snake(s):
+    """Convert kebab-case string to snake_case."""
+    return s.replace("-", "_")
+
+
+def _snake_to_kebab(s):
+    """Convert snake_case string to kebab-case."""
+    return s.replace("_", "-")
+
+
+def normalize(raw):
+    """Recursively normalize an API response dict to snake_case keys."""
+    if isinstance(raw, dict):
+        return {_kebab_to_snake(k): normalize(v) for k, v in raw.items()}
+    if isinstance(raw, list):
+        return [normalize(v) for v in raw]
+    return raw
+
+
+def denormalize_path(path):
+    """Convert a snake_case path list to kebab-case for the API."""
+    return [_snake_to_kebab(p) for p in path]
+
+
+def _diff_value(want_val, have_val, path, cmds, delete_missing):
+    if isinstance(want_val, dict):
+        if not want_val:
+            if have_val is None:
+                cmds.append(("set", denormalize_path(path)))
+        else:
+            have_dict = have_val if isinstance(have_val, dict) else {}
+            _diff_dict(want_val, have_dict, path, cmds, delete_missing)
+    elif isinstance(want_val, list):
+        have_set = set(have_val) if isinstance(have_val, list) else set()
+        for item in want_val:
+            if item not in have_set:
+                cmds.append(("set", denormalize_path(path + [str(item)])))
+        if delete_missing:
+            want_set = set(str(i) for i in want_val)
+            for item in have_val or []:
+                if str(item) not in want_set:
+                    cmds.append(("delete", denormalize_path(path + [str(item)])))
+    else:
+        if want_val != have_val:
+            cmds.append(("set", denormalize_path(path + [str(want_val)])))
+
+
+def _diff_dict(want, have, path, cmds, delete_missing):
+    for key, want_val in want.items():
+        _diff_value(want_val, have.get(key), path + [key], cmds, delete_missing)
+    if delete_missing:
+        for key in have:
+            if key not in want:
+                cmds.append(("delete", denormalize_path(path + [key])))
+
+
+def diff_configs(want, have, base_path, delete_missing=False):
+    """Diff two normalized config dicts and return API command tuples.
 
     Args:
-        module: AnsibleModule instance.
+        want (dict): Desired configuration (snake_case keys).
+        have (dict): Current configuration (snake_case keys).
+        base_path (list): Base API path for commands.
+        delete_missing (bool): Generate delete commands for keys in
+            ``have`` absent from ``want``.
+
+    Returns:
+        list: Tuples of ``("set", path)`` or ``("delete", path)``.
     """
+    cmds = []
+    _diff_dict(want, have, base_path, cmds, delete_missing)
+    return cmds
+
+
+# ---------------------------------------------------------------------------
+# Generic dict diff engine (used by Wave 4+ modules)
+#
+# Design principles:
+#   - want uses snake_case (from YAML/argspec)
+#   - have uses kebab-case (from device API)
+#   - Conversion between - and _ happens here, once, in the core
+#   - Modules only need _BASE path — no key mapping anywhere
+#   - want is the reference dataset — drives all operations
+# ---------------------------------------------------------------------------
+
+
+def owned_config(have, argspec):
+    """Filter raw device config to only keys owned by this module.
+
+    Ownership is declared by the module's argspec — the single source
+    of truth for what this module manages. Keys in have not present in
+    argspec (after normalization) are excluded from before/after output.
+
+    Args:
+        have (dict): Raw device config (kebab-case keys).
+        argspec (dict): Module argument_spec dict.
+
+    Returns:
+        dict: Filtered have with only module-owned keys.
+    """
+    owned = set(argspec.keys()) - {"state"}
+    return {k: v for k, v in have.items() if k.replace("-", "_") in owned}
+
+
+# ---------------------------------------------------------------------------
+# Generic, field-name-agnostic helpers shared by every dict_op-based module.
+#
+# Key-case translation (snake_case <-> kebab-case) is dict_op's own job on
+# the want/have side fed to it directly; these helpers only handle what
+# dict_op *can't* infer on its own: Python bool <-> device presence-node,
+# tag-node string/list collapse, argspec-driven type casting for the
+# public have/gathered output, and keeping one module's dict_op calls from
+# reaching into a subtree owned by another module sharing the same root.
+# ---------------------------------------------------------------------------
+
+
+def autoclean(d):
+    """want-side cleanup: drop None/False, True -> presence node ({}),
+    recurse into dicts. Keys are left exactly as given -- dict_op does
+    the snake_case/kebab-case translation itself when it builds paths.
+    """
+    if not isinstance(d, dict):
+        return d
+    result = {}
+    for k, v in d.items():
+        if v is None or v is False:
+            continue
+        if v is True:
+            result[k] = {}
+        elif isinstance(v, dict):
+            cleaned = autoclean(v)
+            if cleaned:
+                result[k] = cleaned
+        else:
+            result[k] = v
+    return result
+
+
+def from_device(d):
+    """have-side inverse of autoclean, for building the public argspec
+    output: kebab-case -> snake_case keys, presence node -> True, recurse.
+    """
+    if not isinstance(d, dict):
+        return d
+    result = {}
+    for k, v in d.items():
+        snake_k = k.replace("-", "_")
+        if isinstance(v, dict):
+            result[snake_k] = True if not v else from_device(v)
+        else:
+            result[snake_k] = v
+    return result
+
+
+def to_tag_dict(value):
+    """Coerce a VyOS tag-node value (bare str/list, or already a dict)
+    to the {key: {}} shape dict_op always expects for a dict-typed key.
+    """
+    if not value:
+        return {}
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, str):
+        return {value: {}}
+    if isinstance(value, list):
+        return {str(v): {} for v in value}
+    return {}
+
+
+def normalize_have(raw, tag_keys=()):
+    """Coerce the given tag-node keys' subtrees so dict_op only ever sees
+    dicts for them -- VyOS's REST API collapses a single-child tag node to
+    a plain string (or a list for multiple), the same class of quirk
+    dict_op itself already corrects for ordinary list leaves. Every other
+    key is already a plain scalar/dict leaf and passes through untouched.
+    """
+    if not raw or not isinstance(raw, dict):
+        return {}
+    result = {}
+    for k, v in raw.items():
+        if isinstance(v, dict):
+            result[k] = normalize_have(v, tag_keys)
+        elif isinstance(v, (list, str)) and v:
+            result[k] = to_tag_dict(v) if k in tag_keys else v
+        else:
+            result[k] = v
+    return result
+
+
+def cast_by_spec(entry, options):
+    """Cast have-side string leaves to their ARGUMENT_SPEC-declared type.
+
+    Entirely argspec-driven -- no per-field name knowledge. This is what
+    lets from_device() stay purely structural (kebab->snake only) while
+    the public have/gathered output still reports ints as ints, without
+    a hand-maintained list of "which leaves happen to be numeric".
+    Handles list-of-dicts (elements="dict") and scalar-element lists
+    (elements="int"/etc, including VyOS's single-value collapse) alike.
+    """
+    if not isinstance(entry, dict):
+        return entry
+    for key, spec in (options or {}).items():
+        if key not in entry or entry[key] is None:
+            continue
+        spec_type = spec.get("type")
+        if spec_type == "int":
+            entry[key] = int(entry[key])
+        elif spec_type == "dict":
+            cast_by_spec(entry[key], spec.get("options"))
+        elif spec_type == "list":
+            val = entry[key]
+            if not isinstance(val, list):
+                val = [val]
+            elements = spec.get("elements")
+            if elements == "dict":
+                for item in val:
+                    cast_by_spec(item, spec.get("options"))
+            elif elements == "int":
+                val = [int(v) for v in val]
+            entry[key] = val
+    return entry
+
+
+def scope_to_spec(have, options, exclude=()):
+    """Filter have's keys down to ones a module's own argspec actually
+    declares (in kebab-case form), so dict_op purge/set calls never touch
+    a subtree owned by a different module sharing the same device-tree
+    root (e.g. a neighbor's nested address-family, owned by a sibling
+    *_address_family module, is invisible to a module whose argspec never
+    declared it) -- this protects against any such foreign subtree, present
+    or future, without hardcoding its name.
+    """
+    if not isinstance(have, dict):
+        return {}
+    owned = {k.replace("_", "-") for k in (options or {}) if k not in exclude}
+    return {k: v for k, v in have.items() if k in owned}
+
+
+def dict_op(want, have, base_path, op="set"):
+    """Generic dict diff engine for VyOS REST API.
+
+    Compares want (snake_case, from YAML) against have (kebab-case, from
+    device) and generates API command tuples. All key normalization between
+    snake_case and kebab-case happens here — modules never need to convert.
+
+    Set operations on the two datasets:
+        op="set"    want - have       present: apply what is missing
+        op="delete" want ∩ have       absent:  remove what exists
+        op="purge"  have - want       replaced: remove what is extra
+
+    Args:
+        want (dict): Desired config, snake_case keys (from YAML/argspec).
+        have (dict): Current config, kebab-case keys (raw from device API).
+        base_path (list): Base API path — the only module-specific knowledge.
+        op (str): "set", "delete", or "purge".
+
+    Returns:
+        list: Tuples of ("set", path) or ("delete", path).
+    """
+    cmds = []
+
+    # Index have by normalized key for O(1) lookup.
+    # Preserves original kebab-case key for use in API paths.
+    have_idx = {k.replace("-", "_"): (k, v) for k, v in (have or {}).items()}
+
+    if op == "purge":
+        # have - want: delete have keys not present in want
+        # Scoped naturally by _BASE — only this subtree is in have
+        want_keys = {k.replace("-", "_") for k in (want or {})}
+        for norm_k, (orig_k, have_v) in have_idx.items():
+            if norm_k not in want_keys:
+                cmds.append(("delete", base_path + [orig_k]))
+            elif isinstance(have_v, dict):
+                want_nested = (want or {}).get(norm_k) or (want or {}).get(orig_k) or {}
+                if isinstance(want_nested, dict):
+                    cmds += dict_op(want_nested, have_v, base_path + [orig_k], op="purge")
+            elif isinstance(have_v, (list, str)):
+                # List-valued leaf (e.g. a multi-value leafNode): purge
+                # extra have-only items not present in want's list, the
+                # same way op="set"/"delete" already diff list values.
+                # Device may return a single value as a string instead
+                # of a list -- same quirk correction as the list branch
+                # below.
+                want_nested = (want or {}).get(norm_k, (want or {}).get(orig_k))
+                if isinstance(want_nested, list):
+                    have_list = [have_v] if isinstance(have_v, str) else have_v
+                    want_set = {str(i) for i in want_nested}
+                    for item in have_list:
+                        if str(item) not in want_set:
+                            cmds.append(("delete", base_path + [orig_k, str(item)]))
+        return cmds
+
+    for key, want_val in (want or {}).items():
+        if want_val is None:
+            continue
+
+        # Normalize want key for lookup, get original device key for path
+        norm_key = key.replace("-", "_")
+        orig_key, have_val = have_idx.get(norm_key, (key.replace("_", "-"), None))
+        path = base_path + [orig_key]
+
+        if isinstance(want_val, dict):
+            if not want_val:
+                # Presence node
+                if op == "set" and have_val is None:
+                    cmds.append(("set", path))
+                elif op == "delete" and have_val is not None:
+                    cmds.append(("delete", path))
+            else:
+                # Recurse — have_val passed raw, conversion happens recursively
+                cmds += dict_op(want_val, have_val or {}, path, op)
+
+        elif isinstance(want_val, list):
+            # Device may return a single value as a string instead of a list
+            if isinstance(have_val, str):
+                have_val = [have_val]
+            have_set = set(str(i) for i in (have_val or []))
+            if op == "set":
+                # want - have: add missing items
+                for item in want_val:
+                    if str(item) not in have_set:
+                        cmds.append(("set", path + [str(item)]))
+            elif op == "delete":
+                # want ∩ have: remove items that exist
+                for item in want_val:
+                    if str(item) in have_set:
+                        cmds.append(("delete", path + [str(item)]))
+
+        else:
+            # Scalar leaf
+            have_str = str(have_val) if have_val is not None else ""
+            if op == "set" and str(want_val) != have_str:
+                cmds.append(("set", path + [str(want_val)]))
+            elif op == "delete" and have_val is not None:
+                cmds.append(("delete", path))
+
+    return cmds
+
+
+class VyOSModule:
+    """Thin wrapper around VyOSRestClient for resource modules."""
 
     def __init__(self, module):
         self._module = module
         self._client = VyOSRestClient(module)
 
-    # ------------------------------------------------------------------
-    # Read
-    # ------------------------------------------------------------------
-
     def get_config(self, path=None):
         """Retrieve the configuration subtree at *path*.
 
-        Args:
-            path (list, optional): Config path tokens. ``[]`` or ``None``
-                returns the entire running configuration.
-
-        Returns:
-            dict: Configuration subtree, or ``{}`` if the path doesn't exist.
+        Returns raw device dict with kebab-case keys.
         """
         try:
             result = self._client.retrieve_show_config(path or [])
@@ -58,63 +380,6 @@ class VyOSModule:
         except VyOSRestError:
             return {}
 
-    # ------------------------------------------------------------------
-    # Write
-    # ------------------------------------------------------------------
-
-    # def apply_commands(self, commands):
-    #     """Execute a list of ``(op, path)`` command tuples against the device.
-
-    #     Args:
-    #         commands (list): Each item is a 2-tuple ``(op, path)`` where:
-    #             - *op* is ``"set"`` or ``"delete"``
-    #             - *path* is a list of config path tokens
-
-    #             Optionally a 3-tuple ``(op, path, value)`` for leaf nodes
-    #             that carry a value.
-
-    #     Returns:
-    #         list: Results from each command (dicts with ``op`` and ``path``).
-
-    #     Raises:
-    #         VyOSRestError: If a command fails and cannot be recovered.
-
-    #     Notes:
-    #         **Version-adaptive set:** if a ``set`` is rejected by the device,
-    #         the method retries once with the last path segment removed.
-    #         This handles schema simplifications between VyOS releases
-    #         (e.g. ``allow-client address <prefix>`` -> ``allow-client <prefix>``
-    #         in VyOS 1.5+).
-
-    #         **Idempotent delete:** ``delete`` failures are silently ignored
-    #         (the path is already absent).
-    #     """
-    #     results = []
-
-    #     for cmd in commands:
-    #         # Support two command formats:
-    #         # Tuple: ("set", ["path", "tokens"])  or  ("set", ["path"], "value")
-    #         # Dict:  {"op": "set", "path": ["path", "tokens"]}
-    #         if isinstance(cmd, dict):
-    #             op = cmd["op"]
-    #             path = list(cmd["path"])
-    #             value = cmd.get("value")
-    #         else:
-    #             op = cmd[0]
-    #             path = list(cmd[1])
-    #             value = cmd[2] if len(cmd) > 2 else None
-
-    #         if op == "set":
-    #             results.append(self._apply_set(path, value))
-    #         elif op == "delete":
-    #             results.append(self._apply_delete(path))
-    #         else:
-    #             self._module.fail_json(
-    #                 msg="Unknown command op '{op}' in apply_commands".format(op=op),
-    #             )
-
-    #     return results
-
     def apply_commands(self, commands):
         if not commands:
             return []
@@ -138,10 +403,6 @@ class VyOSModule:
             self._client.configure_set(path, value)
             return {"op": "set", "path": path, "status": "ok"}
         except VyOSRestError as exc:
-            # Retry without the second-to-last segment (version-adaptive).
-            # Example: ["service","ntp","allow-client","address","10.0.0.0/24"]
-            #       -> ["service","ntp","allow-client","10.0.0.0/24"]
-            # Only attempt if path is long enough to have an intermediate node.
             if len(path) >= 3:
                 short_path = path[:-2] + [path[-1]]
                 try:
@@ -161,20 +422,18 @@ class VyOSModule:
         except VyOSRestError:
             return {"op": "delete", "path": path, "status": "noop"}
 
-    # ------------------------------------------------------------------
-    # Persist
-    # ------------------------------------------------------------------
-
-    def save_config(self, file_path=None):
-        """Save the running configuration to disk.
-
-        Args:
-            file_path (str, optional): Destination path on the device.
-                Defaults to ``/config/config.boot``.
+    def show(self, path):
+        """Run an operational show command via the /show endpoint.
 
-        Returns:
-            bool: ``True`` if saved successfully, ``False`` otherwise.
+        Raises VyOSRestError on failure; callers that need per-command
+        error reporting (e.g. vyos_command) rely on this propagating
+        rather than being indistinguishable from a valid empty response.
         """
+        result = self._client.show(path)
+        return result.get("data") or ""
+
+    def save_config(self, file_path=None):
+        """Save the running configuration to disk."""
         try:
             self._client.config_file_save(file_path)
             return True
diff --git a/plugins/modules/vyos_static_routes.py b/plugins/modules/vyos_static_routes.py
new file mode 100644
index 0000000..3276588
--- /dev/null
+++ b/plugins/modules/vyos_static_routes.py
@@ -0,0 +1,467 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+
+
+__metaclass__ = type
+
+DOCUMENTATION = r"""
+---
+module: vyos_static_routes
+short_description: Manage static routes on VyOS devices via REST API.
+description:
+  - Manages IPv4 and IPv6 static routes on VyOS devices using the HTTPS REST API.
+  - >-
+    Covers blackhole routes (distance) and next-hop routes (distance,
+    disable, outgoing interface). VyOS's static-route schema is
+    considerably larger than this -- reject routes (an ICMP-unreachable
+    counterpart to blackhole), a top-level per-route interface (a route
+    resolved via an outgoing interface with no next-hop address at
+    all), route tags, route descriptions, ECMP segment weighting, VRF
+    leaking, and BFD monitoring on next-hops are not modeled here.
+    That is a real, documented limitation, not an oversight.
+version_added: "1.0.0"
+author:
+  - VyOS Community (@vyos)
+options:
+  config:
+    description: List of static route configurations grouped by address family.
+    type: list
+    elements: dict
+    suboptions:
+      afi:
+        description: Address family indicator.
+        type: str
+        choices: [ipv4, ipv6]
+        required: true
+      routes:
+        description: List of static route entries.
+        type: list
+        elements: dict
+        suboptions:
+          dest:
+            description: Destination prefix in CIDR notation.
+            type: str
+            required: true
+          blackhole_config:
+            description: Blackhole route configuration (silently discard matching packets).
+            type: dict
+            suboptions:
+              distance:
+                description: Administrative distance (1-255).
+                type: int
+          next_hops:
+            description: List of next-hop addresses.
+            type: list
+            elements: dict
+            suboptions:
+              forward_router_address:
+                description: Next-hop IP address.
+                type: str
+                required: true
+              admin_distance:
+                description: Administrative distance for this next-hop (1-255).
+                type: int
+              enabled:
+                description: Whether this next-hop is enabled.
+                type: bool
+                default: true
+              interface:
+                description: Outgoing interface name.
+                type: str
+  state:
+    description:
+      - C(merged) - Add routes without removing existing ones.
+      - C(replaced) - Replace each named route (by afi + dest) exactly as specified.
+      - C(overridden) - Replace the entire static route table.
+      - C(deleted) - Remove listed or all static routes.
+      - C(gathered) - Read static routes from device without changes.
+    type: str
+    choices: [merged, replaced, overridden, deleted, gathered]
+    default: merged
+seealso:
+  - module: vyos.vyos.vyos_static_routes
+"""
+
+EXAMPLES = r"""
+- name: Merge IPv4 and IPv6 static routes
+  vyos.rest.vyos_static_routes:
+    config:
+      - afi: ipv4
+        routes:
+          - dest: 192.0.2.0/24
+            next_hops:
+              - forward_router_address: 10.0.0.1
+          - dest: 203.0.113.0/24
+            blackhole_config:
+              distance: 200
+      - afi: ipv6
+        routes:
+          - dest: 2001:db8::/32
+            next_hops:
+              - forward_router_address: 2001:db8::1
+    state: merged
+
+- name: Delete all static routes
+  vyos.rest.vyos_static_routes:
+    state: deleted
+
+- name: Gather current static routes
+  vyos.rest.vyos_static_routes:
+    state: gathered
+"""
+
+RETURN = r"""
+before:
+  description: Static route configuration before this module ran.
+  returned: always
+  type: list
+after:
+  description: Static route configuration after this module ran.
+  returned: when changed
+  type: list
+commands:
+  description: List of API command tuples sent to the device.
+  returned: always
+  type: list
+gathered:
+  description: Current static route configuration as structured data.
+  returned: when state is gathered
+  type: list
+saved:
+  description: Whether the config was saved after changes.
+  returned: when changes are applied
+  type: bool
+response:
+  description: Raw API response.
+  returned: when changes are applied
+  type: dict
+"""
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    autoclean,
+    cast_by_spec,
+    dict_op,
+    from_device,
+    to_tag_dict,
+)
+
+
+_BASE = ["protocols", "static"]
+_ROUTE_KEY = {"ipv4": "route", "ipv6": "route6"}
+
+
+def _derive_key_field(options_spec):
+    """The field identifying each entry in a named-list section is
+    never inferable from a generic walk alone -- but it doesn't need
+    to be hand-declared either: every named-list section in this
+    argspec (routes, next_hops) already marks exactly one suboption
+    required=True.
+    """
+    required = [k for k, spec in options_spec.items() if spec.get("required")]
+    if len(required) != 1:
+        raise ValueError(
+            "expected exactly one required suboption to serve as the key field, "
+            "found: {0}".format(required),
+        )
+    return required[0]
+
+
+def _keyed_list_to_device(items, key_field, entry_transform=None):
+    """A list of dicts, each identified by key_field's value, becomes a
+    device dict keyed by that value -- the one structural mechanic
+    every named-list section in this module needs.
+    """
+    entry_transform = entry_transform or (lambda rest: rest)
+    result = {}
+    for item in items or []:
+        if not item.get(key_field):
+            continue
+        rest = {k: v for k, v in item.items() if k != key_field}
+        result[str(item[key_field])] = entry_transform(rest)
+    return result
+
+
+def _keyed_list_from_device(raw, key_field, entry_transform=None):
+    entry_transform = entry_transform or (lambda d: d or {})
+    return [
+        {key_field: key, **entry_transform(data or {})}
+        for key, data in sorted(to_tag_dict(raw).items())
+    ]
+
+
+# ---------------------------------------------------------------------------
+# blackhole_config -- confirmed against vyos-1x: the device node has
+# ONLY "distance" (1-255) and "tag" (not modeled, see module scope
+# note). There is no "type" leaf anywhere in the schema -- the
+# original module's blackhole_config.type field was a genuine,
+# confirmed hallucination and has been removed.
+# ---------------------------------------------------------------------------
+
+
+def _blackhole_to_device(bc):
+    return autoclean(bc or {})
+
+
+def _blackhole_from_device(data):
+    if data is None:
+        return None
+    return from_device(data if isinstance(data, dict) else {})
+
+
+# ---------------------------------------------------------------------------
+# next_hops -- confirmed against vyos-1x: keyed by the next-hop
+# address, with disable (presence), distance (1-255), and interface
+# (plain leaf) as children. segments/vrf/bfd are real but out of scope
+# (see module docstring).
+# ---------------------------------------------------------------------------
+
+
+# admin_distance -> distance is a plain rename (not mechanical kebab).
+# enabled -> disable is a genuine, irreducible boolean inversion: the
+# device tracks presence of "disable" for an OFF next-hop, the
+# argspec tracks an "enabled" bool defaulting True -- autoclean's own
+# bool handling (True -> presence) doesn't fit an inverted, default-
+# true flag, so it's handled explicitly rather than forced through
+# the generic path. "interface" is a plain, direct-match leaf and
+# goes through autoclean/from_device like everywhere else.
+_NEXT_HOP_RENAMES = {"admin_distance": "distance"}
+
+
+def _next_hop_entry_to_device(rest):
+    exclude = set(_NEXT_HOP_RENAMES) | {"enabled"}
+    device = autoclean({k: v for k, v in rest.items() if k not in exclude})
+    for arg_key, device_key in _NEXT_HOP_RENAMES.items():
+        if rest.get(arg_key) is not None:
+            device[device_key] = rest[arg_key]
+    if rest.get("enabled") is False:
+        device["disable"] = {}
+    return device
+
+
+def _next_hop_entry_from_device(data):
+    exclude = set(_NEXT_HOP_RENAMES.values()) | {"disable"}
+    entry = from_device({k: v for k, v in data.items() if k not in exclude})
+    for arg_key, device_key in _NEXT_HOP_RENAMES.items():
+        if data.get(device_key) is not None:
+            entry[arg_key] = int(data[device_key])
+    if "disable" in data:
+        entry["enabled"] = False
+    return entry
+
+
+def _route_entry_to_device(rest):
+    device = {}
+    bc = rest.get("blackhole_config")
+    if bc is not None:
+        device["blackhole"] = _blackhole_to_device(bc)
+    next_hops = rest.get("next_hops") or []
+    if next_hops:
+        device["next-hop"] = _keyed_list_to_device(
+            next_hops,
+            "forward_router_address",
+            _next_hop_entry_to_device,
+        )
+    return device
+
+
+def _route_entry_from_device(data):
+    entry = {}
+    bh = _blackhole_from_device(data.get("blackhole"))
+    if bh is not None:
+        entry["blackhole_config"] = bh
+    nh_raw = data.get("next-hop")
+    if nh_raw:
+        entry["next_hops"] = _keyed_list_from_device(
+            nh_raw,
+            "forward_router_address",
+            _next_hop_entry_from_device,
+        )
+    return entry
+
+
+def _want_to_device(config):
+    result = {}
+    for entry in config or []:
+        afi = entry.get("afi")
+        route_key = _ROUTE_KEY.get(afi)
+        if not route_key:
+            continue
+        routes = entry.get("routes") or []
+        if not routes:
+            continue
+        result[route_key] = _keyed_list_to_device(routes, "dest", _route_entry_to_device)
+    return result
+
+
+def get_running_config(vyos):
+    return vyos.get_config(_BASE) or {}
+
+
+def _device_to_argspec(raw):
+    if not raw:
+        return []
+    result = []
+    for afi, route_key in _ROUTE_KEY.items():
+        raw_routes = raw.get(route_key)
+        if not raw_routes:
+            continue
+        routes = _keyed_list_from_device(raw_routes, "dest", _route_entry_from_device)
+        if routes:
+            result.append({"afi": afi, "routes": routes})
+    return result
+
+
+def build_commands(config, raw_have, state):
+    raw_have = raw_have or {}
+    config = config or []
+
+    if state == "deleted":
+        if not config:
+            return [("delete", _BASE)] if raw_have else []
+        cmds = []
+        for entry in config:
+            route_key = _ROUTE_KEY.get(entry.get("afi"))
+            if not route_key:
+                continue
+            have_routes = raw_have.get(route_key) or {}
+            routes = entry.get("routes") or []
+            if not routes:
+                if route_key in raw_have:
+                    cmds.append(("delete", _BASE + [route_key]))
+                continue
+            for route in routes:
+                dest = route.get("dest")
+                if dest and dest in have_routes:
+                    cmds.append(("delete", _BASE + [route_key, dest]))
+        return cmds
+
+    want = _want_to_device(config)
+    norm_have = _want_to_device(_device_to_argspec(raw_have))
+
+    commands = []
+    if state == "overridden":
+        commands += dict_op(want, norm_have, _BASE, op="purge")
+    elif state == "replaced":
+        # Scoped per individual route (afi + dest), matching every
+        # other module's "replaced only touches what's named" semantic
+        # -- not per address-family, which would incorrectly behave
+        # like overridden for the whole AFI. This also directly fixes
+        # the confirmed bug in the original implementation: dict_op's
+        # purge naturally detects and clears any omitted attribute
+        # (a next-hop's distance/interface, a route's blackhole
+        # entirely), rather than the original's hand-rolled "does
+        # anything differ" heuristic, which only inspected generated
+        # set-commands and therefore could never notice an omitted
+        # (cleared) value, since clearing never produced a command in
+        # the first place.
+        for entry in config:
+            route_key = _ROUTE_KEY.get(entry.get("afi"))
+            if not route_key:
+                continue
+            for route in entry.get("routes") or []:
+                dest = route.get("dest")
+                if not dest:
+                    continue
+                section_want = (want.get(route_key) or {}).get(dest, {})
+                section_have = (norm_have.get(route_key) or {}).get(dest, {})
+                commands += dict_op(
+                    section_want,
+                    section_have,
+                    _BASE + [route_key, dest],
+                    op="purge",
+                )
+    commands += dict_op(want, norm_have, _BASE, op="set")
+    return commands
+
+
+_NEXT_HOP_OPTIONS = dict(
+    forward_router_address=dict(type="str", required=True),
+    admin_distance=dict(type="int"),
+    enabled=dict(type="bool", default=True),
+    interface=dict(type="str"),
+)
+
+_ROUTE_OPTIONS = dict(
+    dest=dict(type="str", required=True),
+    blackhole_config=dict(
+        type="dict",
+        options=dict(
+            distance=dict(type="int"),
+        ),
+    ),
+    next_hops=dict(
+        type="list",
+        elements="dict",
+        options=_NEXT_HOP_OPTIONS,
+    ),
+)
+
+ARGUMENT_SPEC = dict(
+    config=dict(
+        type="list",
+        elements="dict",
+        options=dict(
+            afi=dict(type="str", required=True, choices=["ipv4", "ipv6"]),
+            routes=dict(
+                type="list",
+                elements="dict",
+                options=_ROUTE_OPTIONS,
+            ),
+        ),
+    ),
+    state=dict(
+        type="str",
+        default="merged",
+        choices=["merged", "replaced", "overridden", "deleted", "gathered"],
+    ),
+)
+
+
+def main():
+    module = AnsibleModule(ARGUMENT_SPEC, supports_check_mode=True)
+    vyos = VyOSModule(module)
+
+    state = module.params["state"]
+    config = module.params.get("config") or []
+
+    raw_have = get_running_config(vyos)
+    have = _device_to_argspec(raw_have)
+    for entry in have:
+        for route in entry.get("routes") or []:
+            cast_by_spec(route, _ROUTE_OPTIONS)
+
+    if state == "gathered":
+        module.exit_json(changed=False, gathered=have)
+
+    commands = build_commands(config, raw_have, state)
+
+    if module.check_mode:
+        module.exit_json(changed=bool(commands), commands=commands, before=have)
+
+    if commands:
+        response = vyos.apply_commands(commands)
+        saved = vyos.save_config()
+        after_raw = get_running_config(vyos)
+        after = _device_to_argspec(after_raw)
+        for entry in after:
+            for route in entry.get("routes") or []:
+                cast_by_spec(route, _ROUTE_OPTIONS)
+        module.exit_json(
+            changed=True,
+            before=have,
+            after=after,
+            commands=commands,
+            saved=saved,
+            response=response,
+        )
+
+    module.exit_json(changed=False, before=have, after=have, commands=[])
+
+
+if __name__ == "__main__":
+    main()