T8989: wave4 vyos_command, dict_op refactorT8989_wave4

* T8989: vyos_command module

* T8989: vyos_command module UAT and SIT

* T8989: vyos_command changelog

* T8989: vyos_command linter

* T8989: vyos_config module

* T8989: vyos_config module changelog

* T8989: Wave 4 vyos_config module with integration and unit tests

* T8323: vyos_system module

* T8332: vyos_system SIT and UAT

* T8323: vyos_vlan module

* T8323: vyos_vlan module

* T8323: vyos_vlan module SIT and UAT

* T8323: vyos_system module

* T8989: Wave 4 vyos_vlan reworked with dict_op engine

* T8989: Fix dict_op single-value string list handling, add vyos_system integration tests

* T8989: logging_global refactor

* T8989: migrate ntp_global, logging_global, firewall_global to dict_op engine

* T8989: vyos_nat module for REST API collection

* T8989: vyos_nat module for REST API collection, linter fixes

* T8989: vyos_ha module for REST API collection

* T8989: vyos_ha module for REST API collection

* T8989: vyos_ha module sanity and linter fixes

* T8989: vyos_ha module sanity and linter fixes

* T8989: vyos_ha module linter fixes

* T8989: vyos.rest AI comment fixes

* T8323: vyos_nat AI comment fixes

* T8989 ai fixes

* T8989: vyos_bgp_address_family dict_op

* T8989: vyos_bgp_address_family vyos_bgp_global dict_op

* T8989: dict_op refactor for firewall_*, nat, user

* T8989: dict_op refactor for firewall_*, nat, user

* T8989: dict_op refactor for ntp_global, ha

* T8989: snmp_server dict_op refactor

* T8989: snmp_server dict_op refactor

* T8989: route_map dict_op refactor

16 files changed, 4255 insertions, 2059 deletions

diff --git a/plugins/modules/vyos_bgp_address_family.py b/plugins/modules/vyos_bgp_address_family.py
index 65f5810..b745bd5 100644
--- a/plugins/modules/vyos_bgp_address_family.py
+++ b/plugins/modules/vyos_bgp_address_family.py
@@ -234,24 +234,101 @@ gathered:
   type: dict
 saved:
   description: Whether the config was saved after changes.
-  returned: when changes are applied
+  returned: when changed
   type: bool
 response:
   description: Raw API response.
-  returned: when changes are applied
+  returned: always
   type: dict
 """
 
 from ansible.module_utils.basic import AnsibleModule
-from ansible_collections.vyos.rest.plugins.module_utils.vyos import VyOSModule
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    autoclean,
+    cast_by_spec,
+    dict_op,
+    from_device,
+    normalize_have,
+)
 
 
 _BASE = ["protocols", "bgp"]
 _AFI_MAP = {"ipv4": "ipv4-unicast", "ipv6": "ipv6-unicast"}
-_AFI_RMAP = {"ipv4-unicast": "ipv4", "ipv6-unicast": "ipv6"}
+_AFI_RMAP = {v: k for k, v in _AFI_MAP.items()}
+
+# Tag-node keys whose value dict_op must always see as a dict, never a
+# bare string/list -- VyOS's REST API collapses a single-child tag node
+# to a plain string (or a list for multiple), exactly like it does for
+# ordinary list leaves (see dict_op's own str->list coercion for that
+# case). Only genuine tag nodes with no other structure need this.
+_AF_TAG_KEYS = {"network", "redistribute"}
+
+# The only neighbor-AF options whose device shape isn't a direct
+# structural match for their argspec type. Every other key in this
+# level's argspec passes through autoclean()/from_device() untouched.
+_NEIGHBOR_AF_IRREGULAR = {"afi", "allowas_in", "capability", "soft_reconfiguration"}
+
+
+# ---------------------------------------------------------------------------
+# want -> device: structural reshaping only (networks/redistribute keyed
+# by prefix/protocol, AFI abbreviation, the 3 irregular neighbor-AF
+# options). Everything else is autoclean -- no field-name mapping.
+# ---------------------------------------------------------------------------
+
+
+def _global_af_to_device(af_list):
+    result = {}
+    for af in af_list or []:
+        entry = {}
+        networks = af.get("networks") or []
+        if networks:
+            entry["network"] = {
+                n["prefix"]: autoclean({k: v for k, v in n.items() if k != "prefix"})
+                for n in networks
+            }
+        redistribute = af.get("redistribute") or []
+        if redistribute:
+            entry["redistribute"] = {
+                r["protocol"]: autoclean({k: v for k, v in r.items() if k != "protocol"})
+                for r in redistribute
+            }
+        result[_AFI_MAP[af["afi"]]] = entry
+    return result
+
+
+def _neighbor_af_to_device(af_list):
+    result = {}
+    for af in af_list or []:
+        entry = autoclean({k: v for k, v in af.items() if k not in _NEIGHBOR_AF_IRREGULAR})
+
+        # allowas-in is a container node ({"number": N}), not a bare scalar.
+        if af.get("allowas_in") is not None:
+            entry["allowas_in"] = {"number": af["allowas_in"]}
+
+        # capability.orf: the chosen value becomes a dict KEY, not a leaf
+        # value (confirmed against vyos-1x: afi-capability-orf.xml.i).
+        orf = (af.get("capability") or {}).get("orf")
+        if orf:
+            entry["capability"] = {"orf": {"prefix-list": {orf: {}}}}
+
+        # soft_reconfiguration is a two-level presence node, not a flat one.
+        if af.get("soft_reconfiguration"):
+            entry["soft_reconfiguration"] = {"inbound": {}}
+
+        result[_AFI_MAP[af["afi"]]] = entry
+    return result
+
+
+# ---------------------------------------------------------------------------
+# device -> argspec (public have/gathered output)
+# ---------------------------------------------------------------------------
 
+_GLOBAL_AF_OPTIONS = None  # populated after ARGUMENT_SPEC is defined below
+_NEIGHBOR_AF_OPTIONS = None
 
-def _parse_global_af(raw_afs):
+
+def _global_af_from_device(raw_afs):
     if not raw_afs or not isinstance(raw_afs, dict):
         return []
     result = []
@@ -259,32 +336,28 @@ def _parse_global_af(raw_afs):
         afi = _AFI_RMAP.get(af_key)
         if not afi:
             continue
-        af_data = af_data or {}
-        entry = {"afi": afi}
-
-        nets = af_data.get("network", {})
-        if nets and isinstance(nets, dict):
-            entry["networks"] = [{"prefix": p} for p in sorted(nets.keys())]
-
-        redist = af_data.get("redistribute", {})
-        if redist and isinstance(redist, dict):
-            redist_list = []
-            for proto, rdata in sorted(redist.items()):
-                r = {"protocol": proto}
-                rdata = rdata or {}
-                if "metric" in rdata:
-                    r["metric"] = int(rdata["metric"])
-                if "route-map" in rdata:
-                    r["route_map"] = rdata["route-map"]
-                redist_list.append(r)
-            if redist_list:
-                entry["redistribute"] = redist_list
-
+        af_data = dict(af_data or {})
+        networks_raw = af_data.pop("network", None) or {}
+        redistribute_raw = af_data.pop("redistribute", None) or {}
+
+        entry = {"afi": afi, **from_device(af_data)}
+        if networks_raw:
+            entry["networks"] = [
+                {"prefix": prefix, **from_device(data or {})}
+                for prefix, data in sorted(networks_raw.items())
+            ]
+        if redistribute_raw:
+            entry["redistribute"] = [
+                {"protocol": proto, **from_device(data or {})}
+                for proto, data in sorted(redistribute_raw.items())
+            ]
+
+        cast_by_spec(entry, _GLOBAL_AF_OPTIONS)
         result.append(entry)
     return result
 
 
-def _parse_neighbor_af(raw_afs):
+def _neighbor_af_from_device(raw_afs):
     if not raw_afs or not isinstance(raw_afs, dict):
         return []
     result = []
@@ -292,71 +365,45 @@ def _parse_neighbor_af(raw_afs):
         afi = _AFI_RMAP.get(af_key)
         if not afi:
             continue
-        af_data = af_data or {}
-        entry = {"afi": afi}
-
-        if "nexthop-self" in af_data:
-            entry["nexthop_self"] = True
-        if "route-reflector-client" in af_data:
-            entry["route_reflector_client"] = True
-        if "route-server-client" in af_data:
-            entry["route_server_client"] = True
-        if "default-originate" in af_data:
-            entry["default_originate"] = True
-        if "maximum-prefix" in af_data:
-            entry["maximum_prefix"] = int(af_data["maximum-prefix"])
-        if "weight" in af_data:
-            entry["weight"] = int(af_data["weight"])
-        if "unsuppress-map" in af_data:
-            entry["unsuppress_map"] = af_data["unsuppress-map"]
-        if "allowas-in" in af_data:
-            ai = af_data["allowas-in"]
-            if isinstance(ai, dict) and "number" in ai:
-                entry["allowas_in"] = int(ai["number"])
-            else:
-                entry["allowas_in"] = 1
-
-        sc = af_data.get("soft-reconfiguration", {})
-        if sc and "inbound" in sc:
-            entry["soft_reconfiguration"] = True
+        af_data = dict(af_data or {})
+        allowas = af_data.pop("allowas-in", None)
+        orf = ((af_data.pop("capability", None) or {}).get("orf") or {}).get("prefix-list") or {}
+        soft = af_data.pop("soft-reconfiguration", None)
 
-        rm = af_data.get("route-map", {})
-        if rm:
-            entry["route_map"] = {}
-            if "import" in rm:
-                entry["route_map"]["import"] = rm["import"]
-            if "export" in rm:
-                entry["route_map"]["export"] = rm["export"]
-
-        pl = af_data.get("prefix-list", {})
-        if pl:
-            entry["prefix_list"] = {}
-            if "import" in pl:
-                entry["prefix_list"]["import"] = pl["import"]
-            if "export" in pl:
-                entry["prefix_list"]["export"] = pl["export"]
+        entry = {"afi": afi, **from_device(af_data)}
+        cast_by_spec(entry, _NEIGHBOR_AF_OPTIONS)
+
+        if isinstance(allowas, dict) and "number" in allowas:
+            entry["allowas_in"] = int(allowas["number"])
+        elif allowas is not None:
+            entry["allowas_in"] = 1
+
+        if "receive" in orf:
+            entry["capability"] = {"orf": "receive"}
+        elif "send" in orf:
+            entry["capability"] = {"orf": "send"}
+
+        if isinstance(soft, dict) and "inbound" in soft:
+            entry["soft_reconfiguration"] = True
 
         result.append(entry)
     return result
 
 
-def get_running_config(vyos):
-    raw = vyos.get_config(_BASE)
+def _device_to_argspec(raw):
     if not raw or not isinstance(raw, dict):
         return {}
     result = {}
-
     if "system-as" in raw:
         result["as_number"] = int(raw["system-as"])
 
-    global_afs = _parse_global_af(raw.get("address-family"))
+    global_afs = _global_af_from_device(raw.get("address-family"))
     if global_afs:
         result["address_family"] = global_afs
 
     neighbors = []
     for nb_id, nb_data in sorted((raw.get("neighbor") or {}).items()):
-        nb_data = nb_data or {}
-        nb_afs = _parse_neighbor_af(nb_data.get("address-family"))
+        nb_afs = _neighbor_af_from_device((nb_data or {}).get("address-family"))
         if nb_afs:
             neighbors.append({"neighbor_address": nb_id, "address_family": nb_afs})
     if neighbors:
@@ -365,119 +412,60 @@ def get_running_config(vyos):
     return result
 
 
-def _global_af_cmds(af, have_af):
-    cmds = []
-    afi = af["afi"]
-    af_key = _AFI_MAP[afi]
-    abase = _BASE + ["address-family", af_key]
-    have_af = have_af or {}
-
-    want_nets = {n["prefix"]: n for n in (af.get("networks") or [])}
-    have_nets = {n["prefix"]: n for n in (have_af.get("networks") or [])}
-    for prefix in want_nets:
-        if prefix not in have_nets:
-            cmds.append(("set", abase + ["network", prefix]))
-
-    want_redist = {r["protocol"]: r for r in (af.get("redistribute") or [])}
-    have_redist = {r["protocol"]: r for r in (have_af.get("redistribute") or [])}
-    for proto, entry in want_redist.items():
-        have_entry = have_redist.get(proto, {})
-        rbase = abase + ["redistribute", proto]
-        if proto not in have_redist:
-            cmds.append(("set", rbase))
-        if entry.get("metric") and entry["metric"] != have_entry.get("metric"):
-            cmds.append(("set", rbase + ["metric", str(entry["metric"])]))
-        if entry.get("route_map") and entry["route_map"] != have_entry.get("route_map"):
-            cmds.append(("set", rbase + ["route-map", entry["route_map"]]))
-
-    return cmds
-
-
-def _neighbor_af_cmds(nb_addr, af, have_af):
-    cmds = []
-    afi = af["afi"]
-    af_key = _AFI_MAP[afi]
-    nbase = _BASE + ["neighbor", nb_addr, "address-family", af_key]
-    have_af = have_af or {}
-
-    if af.get("soft_reconfiguration") and not have_af.get("soft_reconfiguration"):
-        cmds.append(("set", nbase + ["soft-reconfiguration", "inbound"]))
-    if af.get("nexthop_self") and not have_af.get("nexthop_self"):
-        cmds.append(("set", nbase + ["nexthop-self"]))
-    if af.get("route_reflector_client") and not have_af.get("route_reflector_client"):
-        cmds.append(("set", nbase + ["route-reflector-client"]))
-    if af.get("route_server_client") and not have_af.get("route_server_client"):
-        cmds.append(("set", nbase + ["route-server-client"]))
-    if af.get("default_originate") and not have_af.get("default_originate"):
-        cmds.append(("set", nbase + ["default-originate"]))
-    if af.get("maximum_prefix") and af["maximum_prefix"] != have_af.get("maximum_prefix"):
-        cmds.append(("set", nbase + ["maximum-prefix", str(af["maximum_prefix"])]))
-    if af.get("weight") and af["weight"] != have_af.get("weight"):
-        cmds.append(("set", nbase + ["weight", str(af["weight"])]))
-    if af.get("allowas_in") and af["allowas_in"] != have_af.get("allowas_in"):
-        cmds.append(("set", nbase + ["allowas-in", "number", str(af["allowas_in"])]))
-    if af.get("unsuppress_map") and af["unsuppress_map"] != have_af.get("unsuppress_map"):
-        cmds.append(("set", nbase + ["unsuppress-map", af["unsuppress_map"]]))
-
-    want_rm = af.get("route_map") or {}
-    have_rm = have_af.get("route_map") or {}
-    if want_rm.get("import") and want_rm["import"] != have_rm.get("import"):
-        cmds.append(("set", nbase + ["route-map", "import", want_rm["import"]]))
-    if want_rm.get("export") and want_rm["export"] != have_rm.get("export"):
-        cmds.append(("set", nbase + ["route-map", "export", want_rm["export"]]))
-
-    want_pl = af.get("prefix_list") or {}
-    have_pl = have_af.get("prefix_list") or {}
-    if want_pl.get("import") and want_pl["import"] != have_pl.get("import"):
-        cmds.append(("set", nbase + ["prefix-list", "import", want_pl["import"]]))
-    if want_pl.get("export") and want_pl["export"] != have_pl.get("export"):
-        cmds.append(("set", nbase + ["prefix-list", "export", want_pl["export"]]))
-
-    return cmds
-
-
-def build_commands(config, have, state):
-    cmds = []
+def get_running_config(vyos):
+    return vyos.get_config(_BASE) or {}
+
+
+# ---------------------------------------------------------------------------
+# Command building — dict_op scoped per owned subtree.
+#
+# "protocols bgp" is a shared root owned jointly with vyos_bgp_global, so
+# every dict_op call here is scoped to a subtree this module fully owns
+# (global address-family, or one neighbor's address-family) — never the
+# shared root, and never a whole "neighbor.<addr>" entry (which also
+# holds remote-as/timers/password etc. that belong to other modules).
+# ---------------------------------------------------------------------------
+
+
+def build_commands(config, raw_have, state):
     config = config or {}
+    raw_have = raw_have or {}
+    commands = []
 
-    if state == "deleted":
-        if have.get("address_family"):
-            cmds.append(("delete", _BASE + ["address-family"]))
-        for nb in have.get("neighbors") or []:
-            path = _BASE + ["neighbor", nb["neighbor_address"], "address-family"]
-            cmds.append(("delete", path))
-        return cmds
+    global_af_base = _BASE + ["address-family"]
+    raw_global_af = raw_have.get("address-family") or {}
+    raw_neighbors = raw_have.get("neighbor") or {}
 
-    if state == "replaced":
-        would_set = build_commands(config, {}, "merged")
-        have_set = build_commands(have, {}, "merged")
-        if would_set == have_set:
-            return []
-        if have.get("address_family"):
-            cmds.append(("delete", _BASE + ["address-family"]))
-        for nb in have.get("neighbors") or []:
-            path = _BASE + ["neighbor", nb["neighbor_address"], "address-family"]
-            cmds.append(("delete", path))
-        have = {}
-
-    # global address-family
-    have_global_af_map = {af["afi"]: af for af in (have.get("address_family") or [])}
-    for af in config.get("address_family") or []:
-        cmds += _global_af_cmds(af, have_global_af_map.get(af["afi"]))
-
-    # per-neighbor address-family
-    have_nb_map = {
-        n["neighbor_address"]: {af["afi"]: af for af in n.get("address_family", [])}
-        for n in (have.get("neighbors") or [])
+    want_global_af = _global_af_to_device(config.get("address_family") or [])
+    want_neighbors = {
+        nb["neighbor_address"]: _neighbor_af_to_device(nb.get("address_family") or [])
+        for nb in (config.get("neighbors") or [])
     }
 
-    for nb in config.get("neighbors") or []:
-        nb_addr = nb["neighbor_address"]
-        have_nb_afs = have_nb_map.get(nb_addr, {})
-        for af in nb.get("address_family") or []:
-            cmds += _neighbor_af_cmds(nb_addr, af, have_nb_afs.get(af["afi"]))
+    if state == "deleted":
+        if raw_global_af:
+            commands.append(("delete", global_af_base))
+        for nb_addr, nb_data in sorted(raw_neighbors.items()):
+            if (nb_data or {}).get("address-family"):
+                commands.append(("delete", _BASE + ["neighbor", nb_addr, "address-family"]))
+        return commands
+
+    norm_global_af = normalize_have(raw_global_af, _AF_TAG_KEYS)
+    if state == "replaced":
+        commands += dict_op(want_global_af, norm_global_af, global_af_base, op="purge")
+    commands += dict_op(want_global_af, norm_global_af, global_af_base, op="set")
+
+    for nb_addr in sorted(set(want_neighbors) | set(raw_neighbors)):
+        nb_base = _BASE + ["neighbor", nb_addr, "address-family"]
+        raw_nb_af = (raw_neighbors.get(nb_addr) or {}).get("address-family") or {}
+        norm_nb_af = normalize_have(raw_nb_af, _AF_TAG_KEYS)
+        want_nb_af = want_neighbors.get(nb_addr, {})
 
-    return cmds
+        if state == "replaced":
+            commands += dict_op(want_nb_af, norm_nb_af, nb_base, op="purge")
+        commands += dict_op(want_nb_af, norm_nb_af, nb_base, op="set")
+
+    return commands
 
 
 ARGUMENT_SPEC = dict(
@@ -594,6 +582,14 @@ ARGUMENT_SPEC = dict(
     ),
 )
 
+# Populated post-definition to avoid forward-reference ordering issues;
+# these back cast_by_spec so have-side int leaves are derived from the
+# spec itself rather than a hand-maintained field list.
+_GLOBAL_AF_OPTIONS = ARGUMENT_SPEC["config"]["options"]["address_family"]["options"]
+_NEIGHBOR_AF_OPTIONS = ARGUMENT_SPEC["config"]["options"]["neighbors"]["options"]["address_family"][
+    "options"
+]
+
 
 def main():
     module = AnsibleModule(ARGUMENT_SPEC, supports_check_mode=True)
@@ -602,12 +598,13 @@ def main():
     state = module.params["state"]
     config = module.params.get("config") or {}
 
-    have = get_running_config(vyos)
+    raw_have = get_running_config(vyos)
+    have = _device_to_argspec(raw_have)
 
     if state == "gathered":
-        module.exit_json(changed=False, gathered=have)
+        module.exit_json(changed=False, gathered=have, commands=[])
 
-    commands = build_commands(config, have, state)
+    commands = build_commands(config, raw_have, state)
 
     if module.check_mode:
         module.exit_json(changed=bool(commands), commands=commands, before=have)
@@ -618,7 +615,7 @@ def main():
         module.exit_json(
             changed=True,
             before=have,
-            after=get_running_config(vyos),
+            after=_device_to_argspec(get_running_config(vyos)),
             commands=commands,
             saved=saved,
             response=response,
diff --git a/plugins/modules/vyos_bgp_global.py b/plugins/modules/vyos_bgp_global.py
index a05e3b0..50d7a87 100644
--- a/plugins/modules/vyos_bgp_global.py
+++ b/plugins/modules/vyos_bgp_global.py
@@ -205,254 +205,197 @@ gathered:
   type: dict
 saved:
   description: Whether the config was saved after changes.
-  returned: when changes are applied
+  returned: when changed
   type: bool
 response:
   description: Raw API response.
-  returned: when changes are applied
+  returned: always
   type: dict
 """
 
 from ansible.module_utils.basic import AnsibleModule
-from ansible_collections.vyos.rest.plugins.module_utils.vyos import VyOSModule
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    autoclean,
+    cast_by_spec,
+    dict_op,
+    from_device,
+    normalize_have,
+    scope_to_spec,
+)
 
 
 _BASE = ["protocols", "bgp"]
 
+# "neighbor" and "peer-group" are genuine tag nodes (like network/
+# redistribute in vyos_bgp_address_family) that VyOS's REST API can
+# collapse to a bare string for a single entry with no other config.
+_TAG_KEYS = {"neighbor", "peer-group"}
 
-def _parse_parameters(raw):
-    if not raw or not isinstance(raw, dict):
-        return {}
-    result = {}
-    if "router-id" in raw:
-        result["router_id"] = raw["router-id"]
-    if "log-neighbor-changes" in raw:
-        result["log_neighbor_changes"] = True
-    if "no-ipv4-unicast" in raw:
-        result["no_ipv4_unicast"] = True
-    if "graceful-restart" in raw:
-        result["graceful_restart"] = True
-    bp = raw.get("bestpath", {}) or {}
-    if bp:
-        bestpath = {}
-        if "as-path" in bp:
-            bestpath["as_path"] = bp["as-path"]
-        if bestpath:
-            result["bestpath"] = bestpath
-    conf = raw.get("confederation", {}) or {}
-    if conf:
-        confederation = {}
-        if "identifier" in conf:
-            confederation["identifier"] = int(conf["identifier"])
-        if "peers" in conf:
-            peers = conf["peers"]
-            if isinstance(peers, list):
-                confederation["peers"] = [int(p) for p in peers]
-            else:
-                confederation["peers"] = [int(peers)]
-        if confederation:
-            result["confederation"] = confederation
+
+# ---------------------------------------------------------------------------
+# want -> device / device -> argspec
+#
+# Every leaf here is a direct structural match between argspec and device
+# shape (unlike vyos_bgp_address_family, this module has zero device-shape
+# exceptions) -- only the two tag-node reshapes (neighbors keyed by
+# address, peer_groups keyed by name) are unavoidable structural work.
+# ---------------------------------------------------------------------------
+
+
+def _neighbors_to_device(neighbors):
+    return {
+        nb["neighbor_address"]: autoclean(
+            {k: v for k, v in nb.items() if k != "neighbor_address"},
+        )
+        for nb in neighbors or []
+    }
+
+
+def _neighbors_from_device(raw):
+    result = []
+    for addr, data in sorted((raw or {}).items()):
+        scoped = scope_to_spec(data or {}, _NEIGHBOR_OPTIONS, exclude={"neighbor_address"})
+        entry = {"neighbor_address": addr, **from_device(scoped)}
+        cast_by_spec(entry, _NEIGHBOR_OPTIONS)
+        result.append(entry)
+    return result
+
+
+def _peer_groups_to_device(peer_groups):
+    return {
+        pg["peer_group"]: autoclean({k: v for k, v in pg.items() if k != "peer_group"})
+        for pg in peer_groups or []
+    }
+
+
+def _peer_groups_from_device(raw):
+    result = []
+    for name, data in sorted((raw or {}).items()):
+        scoped = scope_to_spec(data or {}, _PEER_GROUP_OPTIONS, exclude={"peer_group"})
+        entry = {"peer_group": name, **from_device(scoped)}
+        cast_by_spec(entry, _PEER_GROUP_OPTIONS)
+        result.append(entry)
     return result
 
 
-def _parse_neighbor(nb_id, data):
-    nb = {"neighbor_address": nb_id}
-    data = data or {}
-    if "remote-as" in data:
-        nb["remote_as"] = int(data["remote-as"])
-    if "description" in data:
-        nb["description"] = data["description"]
-    if "ebgp-multihop" in data:
-        nb["ebgp_multihop"] = int(data["ebgp-multihop"])
-    if "local-as" in data:
-        nb["local_as"] = int(data["local-as"])
-    if "password" in data:
-        nb["password"] = data["password"]
-    if "peer-group" in data:
-        nb["peer_group"] = data["peer-group"]
-    if "shutdown" in data:
-        nb["shutdown"] = True
-    if "update-source" in data:
-        nb["update_source"] = data["update-source"]
-    if "disable-connected-check" in data:
-        nb["disable_connected_check"] = True
-    timers = data.get("timers", {}) or {}
-    if timers:
-        t = {}
-        if "holdtime" in timers:
-            t["holdtime"] = int(timers["holdtime"])
-        if "keepalive" in timers:
-            t["keepalive"] = int(timers["keepalive"])
-        if t:
-            nb["timers"] = t
-    return nb
-
-
-def _parse_peer_group(pg_name, data):
-    pg = {"peer_group": pg_name}
-    data = data or {}
-    if "remote-as" in data:
-        pg["remote_as"] = int(data["remote-as"])
-    if "description" in data:
-        pg["description"] = data["description"]
-    if "ebgp-multihop" in data:
-        pg["ebgp_multihop"] = int(data["ebgp-multihop"])
-    if "password" in data:
-        pg["password"] = data["password"]
-    if "update-source" in data:
-        pg["update_source"] = data["update-source"]
-    timers = data.get("timers", {}) or {}
-    if timers:
-        t = {}
-        if "holdtime" in timers:
-            t["holdtime"] = int(timers["holdtime"])
-        if "keepalive" in timers:
-            t["keepalive"] = int(timers["keepalive"])
-        if t:
-            pg["timers"] = t
-    return pg
+def _want_to_device(config):
+    config = config or {}
+    result = {}
+    if config.get("as_number") is not None:
+        result["system_as"] = config["as_number"]
+    if config.get("parameters"):
+        result["parameters"] = autoclean(config["parameters"])
+    if config.get("neighbors"):
+        result["neighbor"] = _neighbors_to_device(config["neighbors"])
+    if config.get("peer_groups"):
+        result["peer_group"] = _peer_groups_to_device(config["peer_groups"])
+    return result
 
 
 def get_running_config(vyos):
-    raw = vyos.get_config(_BASE)
+    return vyos.get_config(_BASE) or {}
+
+
+def _device_to_argspec(raw):
     if not raw or not isinstance(raw, dict):
         return {}
     result = {}
-
     if "system-as" in raw:
         result["as_number"] = int(raw["system-as"])
-
-    params = _parse_parameters(raw.get("parameters"))
-    if params:
+    if raw.get("parameters"):
+        params = from_device(raw["parameters"])
+        cast_by_spec(params, _PARAMETERS_OPTIONS)
         result["parameters"] = params
-
-    neighbors = []
-    for nb_id, data in sorted((raw.get("neighbor") or {}).items()):
-        neighbors.append(_parse_neighbor(nb_id, data))
+    neighbors = _neighbors_from_device(raw.get("neighbor"))
     if neighbors:
         result["neighbors"] = neighbors
-
-    peer_groups = []
-    for pg_name, data in sorted((raw.get("peer-group") or {}).items()):
-        peer_groups.append(_parse_peer_group(pg_name, data))
+    peer_groups = _peer_groups_from_device(raw.get("peer-group"))
     if peer_groups:
         result["peer_groups"] = peer_groups
-
     return result
 
 
-def _neighbor_cmds(nb, have_nb):
-    cmds = []
-    nb_addr = nb["neighbor_address"]
-    nbase = _BASE + ["neighbor", nb_addr]
-    have_nb = have_nb or {}
-
-    if nb.get("remote_as") and nb["remote_as"] != have_nb.get("remote_as"):
-        cmds.append(("set", nbase + ["remote-as", str(nb["remote_as"])]))
-    if nb.get("description") and nb["description"] != have_nb.get("description"):
-        cmds.append(("set", nbase + ["description", nb["description"]]))
-    if nb.get("ebgp_multihop") and nb["ebgp_multihop"] != have_nb.get("ebgp_multihop"):
-        cmds.append(("set", nbase + ["ebgp-multihop", str(nb["ebgp_multihop"])]))
-    if nb.get("local_as") and nb["local_as"] != have_nb.get("local_as"):
-        cmds.append(("set", nbase + ["local-as", str(nb["local_as"])]))
-    if nb.get("password") and nb["password"] != have_nb.get("password"):
-        cmds.append(("set", nbase + ["password", nb["password"]]))
-    if nb.get("peer_group") and nb["peer_group"] != have_nb.get("peer_group"):
-        cmds.append(("set", nbase + ["peer-group", nb["peer_group"]]))
-    if nb.get("update_source") and nb["update_source"] != have_nb.get("update_source"):
-        cmds.append(("set", nbase + ["update-source", nb["update_source"]]))
-    if nb.get("shutdown") and not have_nb.get("shutdown"):
-        cmds.append(("set", nbase + ["shutdown"]))
-    if nb.get("disable_connected_check") and not have_nb.get("disable_connected_check"):
-        cmds.append(("set", nbase + ["disable-connected-check"]))
-
-    want_t = nb.get("timers") or {}
-    have_t = have_nb.get("timers") or {}
-    if want_t.get("holdtime") and want_t["holdtime"] != have_t.get("holdtime"):
-        cmds.append(("set", nbase + ["timers", "holdtime", str(want_t["holdtime"])]))
-    if want_t.get("keepalive") and want_t["keepalive"] != have_t.get("keepalive"):
-        cmds.append(("set", nbase + ["timers", "keepalive", str(want_t["keepalive"])]))
-
-    return cmds
-
-
-def _peer_group_cmds(pg, have_pg):
-    cmds = []
-    pg_name = pg["peer_group"]
-    pbase = _BASE + ["peer-group", pg_name]
-    have_pg = have_pg or {}
-
-    if pg.get("remote_as") and pg["remote_as"] != have_pg.get("remote_as"):
-        cmds.append(("set", pbase + ["remote-as", str(pg["remote_as"])]))
-    if pg.get("description") and pg["description"] != have_pg.get("description"):
-        cmds.append(("set", pbase + ["description", pg["description"]]))
-    if pg.get("ebgp_multihop") and pg["ebgp_multihop"] != have_pg.get("ebgp_multihop"):
-        cmds.append(("set", pbase + ["ebgp-multihop", str(pg["ebgp_multihop"])]))
-    if pg.get("password") and pg["password"] != have_pg.get("password"):
-        cmds.append(("set", pbase + ["password", pg["password"]]))
-    if pg.get("update_source") and pg["update_source"] != have_pg.get("update_source"):
-        cmds.append(("set", pbase + ["update-source", pg["update_source"]]))
-
-    want_t = pg.get("timers") or {}
-    have_t = have_pg.get("timers") or {}
-    if want_t.get("holdtime") and want_t["holdtime"] != have_t.get("holdtime"):
-        cmds.append(("set", pbase + ["timers", "holdtime", str(want_t["holdtime"])]))
-    if want_t.get("keepalive") and want_t["keepalive"] != have_t.get("keepalive"):
-        cmds.append(("set", pbase + ["timers", "keepalive", str(want_t["keepalive"])]))
-
-    return cmds
-
-
-def build_commands(config, have, state):
-    cmds = []
-
-    if state == "deleted":
-        if have:
-            cmds.append(("delete", _BASE))
-        return cmds
-
-    if state == "replaced":
-        would_set = build_commands(config, {}, "merged")
-        have_set = build_commands(have, {}, "merged")
-        if would_set == have_set:
-            return []
-        if have:
-            cmds.append(("delete", _BASE))
-        have = {}
-
-    config = config or {}
+# ---------------------------------------------------------------------------
+# Command building — dict_op scoped per owned subtree, with one exception.
+#
+# "protocols bgp" is a shared root with vyos_bgp_address_family, and each
+# neighbor entry mixes fields owned by *both* modules (this module owns
+# remote-as/timers/etc.; the sibling module owns the nested address-family
+# subtree). Every dict_op call for a neighbor or peer-group here first
+# goes through scope_to_spec() against this module's own ARGUMENT_SPEC, so
+# a foreign subtree like address-family is never visible to purge/set —
+# without hardcoding its name, since this module's argspec simply never
+# declared it.
+#
+# The one exception: removing system-as. VyOS rejects any commit that
+# leaves "protocols bgp" non-empty without an AS number defined, so that
+# specific transition can't be done with scoped/incremental commands --
+# see the short-circuit at the top of build_commands().
+# ---------------------------------------------------------------------------
+
+
+def build_commands(config, raw_have, state):
+    raw_have = raw_have or {}
+    # "deleted" is "replaced" with an empty desired state -- same scoped
+    # purge mechanics, no separate blanket-delete-the-whole-root logic
+    # (which would have wiped the sibling module's config too)...
+    want = _want_to_device({} if state == "deleted" else config)
+    effective_state = "replaced" if state == "deleted" else state
+
+    # ...EXCEPT for one case: VyOS requires system-as to be defined
+    # whenever "protocols bgp" has any content at all, and rejects the
+    # commit otherwise. So if system-as is being removed (present in
+    # have, absent from want) under replaced/deleted -- the only states
+    # that purge at all -- the only valid action is to delete the entire
+    # tree in one atomic commit, including address-family, which cannot
+    # validly exist without an AS number anyway. This is a real
+    # device-model cascade, not cross-module scope creep. It must never
+    # fire for "merged": an omitted config/as_number there is a no-op by
+    # definition, and merged's set-only dict_op flow below already
+    # leaves system-as untouched correctly on its own.
+    if effective_state == "replaced" and "system-as" in raw_have and "system_as" not in want:
+        return [("delete", _BASE)]
+
+    commands = []
+
+    norm_have = normalize_have(raw_have, _TAG_KEYS)
+
+    top_have = {k: v for k, v in raw_have.items() if k in ("system-as", "parameters")}
+    top_want = {k: v for k, v in want.items() if k in ("system_as", "parameters")}
+    if effective_state == "replaced":
+        commands += dict_op(top_want, top_have, _BASE, op="purge")
+    commands += dict_op(top_want, top_have, _BASE, op="set")
+
+    raw_neighbors = norm_have.get("neighbor") or {}
+    want_neighbors = want.get("neighbor", {})
+    for addr in sorted(set(want_neighbors) | set(raw_neighbors)):
+        nbase = _BASE + ["neighbor", addr]
+        have_scoped = scope_to_spec(
+            raw_neighbors.get(addr) or {},
+            _NEIGHBOR_OPTIONS,
+            exclude={"neighbor_address"},
+        )
+        want_entry = want_neighbors.get(addr, {})
+        if effective_state == "replaced":
+            commands += dict_op(want_entry, have_scoped, nbase, op="purge")
+        commands += dict_op(want_entry, have_scoped, nbase, op="set")
+
+    raw_peer_groups = norm_have.get("peer-group") or {}
+    want_peer_groups = want.get("peer_group", {})
+    for name in sorted(set(want_peer_groups) | set(raw_peer_groups)):
+        pbase = _BASE + ["peer-group", name]
+        have_scoped = scope_to_spec(
+            raw_peer_groups.get(name) or {},
+            _PEER_GROUP_OPTIONS,
+            exclude={"peer_group"},
+        )
+        want_entry = want_peer_groups.get(name, {})
+        if effective_state == "replaced":
+            commands += dict_op(want_entry, have_scoped, pbase, op="purge")
+        commands += dict_op(want_entry, have_scoped, pbase, op="set")
 
-    # system-as — must be first
-    if config.get("as_number") and config["as_number"] != have.get("as_number"):
-        cmds.append(("set", _BASE + ["system-as", str(config["as_number"])]))
-
-    # parameters
-    params = config.get("parameters") or {}
-    have_params = have.get("parameters") or {}
-    if params.get("router_id") and params["router_id"] != have_params.get("router_id"):
-        cmds.append(("set", _BASE + ["parameters", "router-id", params["router_id"]]))
-    if params.get("log_neighbor_changes") and not have_params.get("log_neighbor_changes"):
-        cmds.append(("set", _BASE + ["parameters", "log-neighbor-changes"]))
-    if params.get("no_ipv4_unicast") and not have_params.get("no_ipv4_unicast"):
-        cmds.append(("set", _BASE + ["parameters", "no-ipv4-unicast"]))
-    if params.get("graceful_restart") and not have_params.get("graceful_restart"):
-        cmds.append(("set", _BASE + ["parameters", "graceful-restart"]))
-    bp = params.get("bestpath") or {}
-    have_bp = have_params.get("bestpath") or {}
-    if bp.get("as_path") and bp["as_path"] != have_bp.get("as_path"):
-        cmds.append(("set", _BASE + ["parameters", "bestpath", "as-path", bp["as_path"]]))
-
-    # neighbors
-    have_nb_map = {n["neighbor_address"]: n for n in (have.get("neighbors") or [])}
-    for nb in config.get("neighbors") or []:
-        cmds += _neighbor_cmds(nb, have_nb_map.get(nb["neighbor_address"]))
-
-    # peer_groups
-    have_pg_map = {p["peer_group"]: p for p in (have.get("peer_groups") or [])}
-    for pg in config.get("peer_groups") or []:
-        cmds += _peer_group_cmds(pg, have_pg_map.get(pg["peer_group"]))
-
-    return cmds
+    return commands
 
 
 ARGUMENT_SPEC = dict(
@@ -535,6 +478,13 @@ ARGUMENT_SPEC = dict(
     ),
 )
 
+# Populated post-definition (avoids forward-reference ordering); backs
+# cast_by_spec/scope_to_spec so have-side casting and cross-module
+# protection are both derived from the spec itself.
+_PARAMETERS_OPTIONS = ARGUMENT_SPEC["config"]["options"]["parameters"]["options"]
+_NEIGHBOR_OPTIONS = ARGUMENT_SPEC["config"]["options"]["neighbors"]["options"]
+_PEER_GROUP_OPTIONS = ARGUMENT_SPEC["config"]["options"]["peer_groups"]["options"]
+
 
 def main():
     module = AnsibleModule(ARGUMENT_SPEC, supports_check_mode=True)
@@ -543,12 +493,13 @@ def main():
     state = module.params["state"]
     config = module.params.get("config") or {}
 
-    have = get_running_config(vyos)
+    raw_have = get_running_config(vyos)
+    have = _device_to_argspec(raw_have)
 
     if state == "gathered":
-        module.exit_json(changed=False, gathered=have)
+        module.exit_json(changed=False, gathered=have, commands=[])
 
-    commands = build_commands(config, have, state)
+    commands = build_commands(config, raw_have, state)
 
     if module.check_mode:
         module.exit_json(changed=bool(commands), commands=commands, before=have)
@@ -559,7 +510,7 @@ def main():
         module.exit_json(
             changed=True,
             before=have,
-            after=get_running_config(vyos),
+            after=_device_to_argspec(get_running_config(vyos)),
             commands=commands,
             saved=saved,
             response=response,
diff --git a/plugins/modules/vyos_command.py b/plugins/modules/vyos_command.py
new file mode 100644
index 0000000..d688251
--- /dev/null
+++ b/plugins/modules/vyos_command.py
@@ -0,0 +1,228 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# GNU General Public License v3.0+
+from __future__ import absolute_import, division, print_function
+
+
+__metaclass__ = type
+
+DOCUMENTATION = r"""
+---
+module: vyos_command
+short_description: Run show commands on VyOS devices using REST API
+description:
+  - Sends show commands to VyOS devices via the REST API C(/show) endpoint
+    and returns the output.
+  - Equivalent to C(vyos_command) in the CLI collection but uses the REST API.
+  - Uses REST API (C(connection=httpapi)) instead of CLI.
+version_added: "1.0.0"
+author:
+  - VyOS Community (@vyos)
+options:
+  commands:
+    description:
+      - List of show commands to run on the device.
+      - Each command is a list of path elements passed to the C(/show) endpoint.
+      - Commands may be specified as a string (space-separated) or a list.
+    type: list
+    elements: raw
+    required: true
+  wait_for:
+    description:
+      - Specifies what to evaluate from the output of the command and what
+        conditionals to apply. This argument will cause the task to wait for
+        a particular conditional to be true before moving forward.
+    type: list
+    elements: str
+    aliases: [waitfor]
+  match:
+    description:
+      - The C(match) argument is used in conjunction with the C(wait_for)
+        argument to specify the match policy.
+    type: str
+    choices: [any, all]
+    default: all
+  retries:
+    description:
+      - Specifies the number of retries a command should be run before it
+        is considered failed.
+    type: int
+    default: 10
+  interval:
+    description:
+      - Configures the interval in seconds to wait between retries of the
+        command.
+    type: int
+    default: 1
+notes:
+  - Requires C(ansible_connection=httpapi) with the VyOS httpapi plugin.
+  - C(ansible_network_os) must be set to C(vyos.rest.vyos).
+  - Only C(show) commands are supported via the REST API.
+  - Commands are passed as path lists to the C(/show) endpoint.
+"""
+
+EXAMPLES = r"""
+- name: Run show version
+  vyos.rest.vyos_command:
+    commands:
+      - - version
+  register: result
+
+- name: Run multiple show commands
+  vyos.rest.vyos_command:
+    commands:
+      - - interfaces
+      - - ip
+        - route
+      - - system
+        - uptime
+  register: result
+
+- name: Run show commands as strings
+  vyos.rest.vyos_command:
+    commands:
+      - "interfaces"
+      - "ip route"
+      - "version"
+  register: result
+
+- name: Wait for BGP to establish
+  vyos.rest.vyos_command:
+    commands:
+      - - ip
+        - bgp
+        - summary
+    wait_for:
+      - result[0] contains Established
+    retries: 10
+    interval: 5
+"""
+
+RETURN = r"""
+stdout:
+  description: List of output from each command.
+  returned: always
+  type: list
+  sample: ["VyOS 1.5.0\n...", "Interface    IP Address\n..."]
+stdout_lines:
+  description: List of output split into lines for each command.
+  returned: always
+  type: list
+failed_conditions:
+  description: List of conditions that failed.
+  returned: failed
+  type: list
+"""
+
+import time
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import VyOSModule
+
+
+def parse_command(cmd):
+    """Convert a command to a path list."""
+    if isinstance(cmd, list):
+        return cmd
+    elif isinstance(cmd, str):
+        return cmd.split()
+    return list(cmd)
+
+
+def run_commands(vyos, commands):
+    """Run show commands and return stdout list."""
+    stdout = []
+    for cmd in commands:
+        path = parse_command(cmd)
+        try:
+            result = vyos.show(path)
+            stdout.append(result if result else "")
+        except Exception as e:
+            stdout.append("ERROR: %s" % str(e))
+    return stdout
+
+
+def evaluate_conditions(stdout, wait_for, match):
+    """Evaluate wait_for conditions against stdout."""
+    failed = []
+    results = []
+
+    for condition in wait_for:
+        # Parse simple conditions: "result[N] contains STRING"
+        if " contains " in condition:
+            parts = condition.split(" contains ", 1)
+            ref = parts[0].strip()
+            value = parts[1].strip()
+            # Extract index from result[N]
+            if ref.startswith("result[") and ref.endswith("]"):
+                try:
+                    idx = int(ref[7:-1])
+                    matched = value in stdout[idx]
+                    results.append(matched)
+                    if not matched:
+                        failed.append(condition)
+                except (ValueError, IndexError):
+                    failed.append(condition)
+            else:
+                failed.append(condition)
+        else:
+            # Unsupported condition format
+            failed.append(condition)
+
+    if match == "any":
+        return not any(results), failed
+    return bool(failed), failed
+
+
+def main():
+    module = AnsibleModule(
+        argument_spec=dict(
+            commands=dict(type="list", elements="raw", required=True),
+            wait_for=dict(type="list", elements="str", aliases=["waitfor"]),
+            match=dict(type="str", default="all", choices=["any", "all"]),
+            retries=dict(type="int", default=10),
+            interval=dict(type="int", default=1),
+        ),
+        supports_check_mode=True,
+    )
+
+    vyos = VyOSModule(module)
+    commands = module.params["commands"]
+    wait_for = module.params["wait_for"] or []
+    match = module.params["match"]
+    retries = module.params["retries"]
+    interval = module.params["interval"]
+
+    stdout = []
+    failed_conditions = []
+
+    for attempt in range(retries):
+        stdout = run_commands(vyos, commands)
+
+        if not wait_for:
+            break
+
+        failed_check, failed_conditions = evaluate_conditions(stdout, wait_for, match)
+        if not failed_check:
+            break
+
+        if attempt < retries - 1:
+            time.sleep(interval)
+    else:
+        if failed_conditions:
+            module.fail_json(
+                msg="One or more conditional statements have not been satisfied",
+                failed_conditions=failed_conditions,
+            )
+
+    stdout_lines = [out.splitlines() for out in stdout]
+
+    module.exit_json(
+        changed=False,
+        stdout=stdout,
+        stdout_lines=stdout_lines,
+    )
+
+
+if __name__ == "__main__":
+    main()
diff --git a/plugins/modules/vyos_config.py b/plugins/modules/vyos_config.py
new file mode 100644
index 0000000..55d5fdc
--- /dev/null
+++ b/plugins/modules/vyos_config.py
@@ -0,0 +1,238 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# GNU General Public License v3.0+
+from __future__ import absolute_import, division, print_function
+
+
+__metaclass__ = type
+
+DOCUMENTATION = r"""
+---
+module: vyos_config
+short_description: Manage VyOS configuration using REST API
+description:
+  - Manages VyOS device configuration via the REST API.
+  - Accepts configuration commands in CLI C(set)/C(delete) string format
+    and applies them via the REST C(/configure) endpoint.
+  - Uses REST API (C(connection=httpapi)) instead of CLI.
+version_added: "1.0.0"
+author:
+  - VyOS Community (@vyos)
+options:
+  lines:
+    description:
+      - Ordered list of C(set) or C(delete) commands to apply.
+      - Commands should be in standard VyOS CLI format, e.g.
+        C(set system host-name router1) or C(delete protocols bgp).
+    type: list
+    elements: str
+  src:
+    description:
+      - Path to a file containing C(set)/C(delete) commands, one per line.
+      - Blank lines and lines starting with C(#) are ignored.
+      - Mutually exclusive with C(lines).
+    type: path
+  match:
+    description:
+      - Controls how commands are matched against the running configuration.
+      - C(line) checks each command against the running config and only
+        applies commands that would change the configuration.
+      - C(none) applies all commands without checking the running config.
+    type: str
+    default: line
+    choices: [line, none]
+  save:
+    description:
+      - Save the configuration to disk after applying changes.
+    type: bool
+    default: false
+notes:
+  - Requires C(ansible_connection=httpapi) with the VyOS httpapi plugin.
+  - C(ansible_network_os) must be set to C(vyos.rest.vyos).
+  - Unlike the CLI collection's C(vyos_config), this module does not support
+    C(backup), C(confirm), or C(comment) options as these are CLI-specific.
+  - Commands are parsed from CLI string format into REST API path arrays.
+"""
+
+EXAMPLES = r"""
+- name: Apply configuration lines
+  vyos.rest.vyos_config:
+    lines:
+      - set system host-name router1
+      - set system domain-name example.com
+      - set interfaces ethernet eth0 description "WAN"
+    save: true
+
+- name: Delete configuration
+  vyos.rest.vyos_config:
+    lines:
+      - delete protocols bgp
+    save: true
+
+- name: Apply config from file
+  vyos.rest.vyos_config:
+    src: /tmp/vyos_config.txt
+    match: none
+    save: true
+
+- name: Always apply without matching
+  vyos.rest.vyos_config:
+    lines:
+      - set system host-name router1
+    match: none
+"""
+
+RETURN = r"""
+commands:
+  description: List of commands applied to the device.
+  returned: always
+  type: list
+saved:
+  description: Whether the configuration was saved to disk.
+  returned: when save is true and changes were made
+  type: bool
+response:
+  description: Raw API response from the device.
+  returned: always
+  type: dict
+"""
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import VyOSModule
+
+
+def parse_line(line):
+    """Parse a CLI set/delete line into (op, path) tuple.
+
+    Args:
+        line (str): CLI command, e.g. "set system host-name router1"
+
+    Returns:
+        tuple: (op, path) where op is "set" or "delete" and path is a list,
+               or None if the line is not a valid command.
+    """
+    import shlex
+
+    line = line.strip()
+    if not line or line.startswith("#"):
+        return None
+    try:
+        tokens = shlex.split(line)
+    except ValueError:
+        tokens = line.split()
+    if len(tokens) < 2:
+        return None
+    op = tokens[0].lower()
+    if op not in ("set", "delete"):
+        return None
+    path = tokens[1:]
+    return (op, path)
+
+
+def load_lines(module):
+    """Load lines from either lines param or src file."""
+    if module.params["lines"]:
+        return module.params["lines"]
+    src = module.params["src"]
+    if src:
+        try:
+            with open(src) as f:
+                return f.readlines()
+        except IOError as e:
+            module.fail_json(msg="Unable to read src file: %s" % str(e))
+    return []
+
+
+def parse_commands(lines):
+    """Parse a list of CLI lines into (op, path) tuples."""
+    commands = []
+    for line in lines:
+        parsed = parse_line(line)
+        if parsed:
+            commands.append(parsed)
+    return commands
+
+
+def filter_commands(commands, vyos):
+    """Filter commands that would not change the running config.
+
+    For set commands, check if the path already has the desired value.
+    For delete commands, check if the path exists.
+    """
+    filtered = []
+    for op, path in commands:
+        if op == "set":
+            if len(path) >= 2:
+                # For leaf: path[-1] is the value, path[:-1] is the config path
+                # e.g. ["system","host-name","vyos150"] -> get ["system","host-name"]
+                # returns {"host-name": "vyos150"} -> unwrap -> "vyos150"
+                parent_path = path[:-1]
+                value = path[-1]
+                parent = vyos.get_config(parent_path)
+                if isinstance(parent, dict):
+                    # unwrap single-key dict (API wraps leaf values)
+                    if len(parent) == 1:
+                        actual = list(parent.values())[0]
+                    else:
+                        actual = parent.get(parent_path[-1])
+                    if actual == value:
+                        continue
+                    # value may be a key in the dict (tag node)
+                    if value in parent:
+                        continue
+                elif isinstance(parent, str) and parent == value:
+                    continue
+            filtered.append((op, path))
+        elif op == "delete":
+            current = vyos.get_config(path)
+            if current is not None and current != {}:
+                filtered.append((op, path))
+    return filtered
+
+
+def main():
+    module = AnsibleModule(
+        argument_spec=dict(
+            lines=dict(type="list", elements="str"),
+            src=dict(type="path"),
+            match=dict(type="str", default="line", choices=["line", "none"]),
+            save=dict(type="bool", default=False),
+        ),
+        mutually_exclusive=[["lines", "src"]],
+        supports_check_mode=True,
+    )
+
+    vyos = VyOSModule(module)
+
+    lines = load_lines(module)
+    commands = parse_commands(lines)
+
+    if not commands:
+        module.exit_json(changed=False, commands=[])
+
+    match = module.params["match"]
+    if match == "line":
+        commands = filter_commands(commands, vyos)
+
+    if not commands:
+        module.exit_json(changed=False, commands=[])
+
+    if module.check_mode:
+        module.exit_json(changed=True, commands=commands)
+
+    response = vyos.apply_commands(commands)
+
+    saved = False
+    if module.params["save"]:
+        saved = vyos.save_config()
+
+    module.exit_json(
+        changed=True,
+        commands=commands,
+        saved=saved,
+        response=response,
+    )
+
+
+if __name__ == "__main__":
+    main()
diff --git a/plugins/modules/vyos_firewall_global.py b/plugins/modules/vyos_firewall_global.py
index 66cdabc..337f364 100644
--- a/plugins/modules/vyos_firewall_global.py
+++ b/plugins/modules/vyos_firewall_global.py
@@ -181,21 +181,30 @@ gathered:
   type: dict
 saved:
   description: Whether the config was saved after changes.
-  returned: when changes are applied
+  returned: when changed
   type: bool
 response:
   description: Raw API response.
-  returned: when changes are applied
+  returned: always
   type: dict
 """
 
 from ansible.module_utils.basic import AnsibleModule
-from ansible_collections.vyos.rest.plugins.module_utils.vyos import VyOSModule
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    autoclean,
+    dict_op,
+    from_device,
+    normalize_have,
+)
 
 
 _BASE = ["firewall", "group"]
 
-# Map argspec key -> API key, value key
+# argspec_key -> (device_key, member_key). A genuinely minimal, unavoidable
+# mapping: VyOS's 5 group types have different kebab-case device names and
+# different member-list field names (address/network/port/interface), none
+# of which is a mechanical snake<->kebab transform of the other.
 _GROUP_TYPES = {
     "address_group": ("address-group", "address"),
     "network_group": ("network-group", "network"),
@@ -204,103 +213,81 @@ _GROUP_TYPES = {
     "ipv6_network_group": ("ipv6-network-group", "network"),
 }
 
+# Each of the 5 device_keys above is itself a tag node (keyed by group
+# name) that can collapse to a bare string for a single group with no
+# other config. The member fields (address/network/port/interface) are
+# NOT tag nodes -- confirmed against vyos-1x (leafNode with <multi/>) --
+# they're plain multi-value leaves, so dict_op's own native list handling
+# applies to them directly; no reshaping needed.
+_TAG_KEYS = {device_key for device_key, _member_key in _GROUP_TYPES.values()}
 
-def _parse_group_type(raw, val_key):
-    """Parse a group dict from API raw data."""
-    if not raw or not isinstance(raw, dict):
-        return []
-    result = []
-    for name, data in sorted(raw.items()):
-        entry = {"name": name}
-        data = data or {}
-        if data.get("description"):
-            entry["description"] = data["description"]
-        val = data.get(val_key)
-        if val is not None:
-            if isinstance(val, list):
-                entry[val_key.replace("-", "_")] = val
-            elif isinstance(val, str):
-                entry[val_key.replace("-", "_")] = [val]
-            elif isinstance(val, dict):
-                entry[val_key.replace("-", "_")] = list(val.keys())
-        result.append(entry)
-    return result
 
+def _group_to_device(g, member_key):
+    entry = autoclean({k: v for k, v in g.items() if k not in ("name", member_key)})
+    members = g.get(member_key)
+    if members:
+        entry[member_key] = [str(m) for m in members]
+    return entry
 
-def get_running_config(vyos):
-    raw = vyos.get_config(_BASE)
-    if not raw or not isinstance(raw, dict):
-        return {}
-    result = {"group": {}}
 
-    for arg_key, (api_key, val_key) in _GROUP_TYPES.items():
-        groups = _parse_group_type(raw.get(api_key), val_key)
-        if groups:
-            result["group"][arg_key] = groups
+def _groups_to_device(groups, member_key):
+    return {g["name"]: _group_to_device(g, member_key) for g in groups or []}
 
-    if not result["group"]:
-        return {}
-    return result
 
+def _want_to_device(config):
+    group = (config or {}).get("group") or {}
+    want = {}
+    for arg_key, (device_key, member_key) in _GROUP_TYPES.items():
+        groups = group.get(arg_key) or []
+        if groups:
+            want[device_key] = _groups_to_device(groups, member_key)
+    return want
 
-def _group_cmds(arg_key, groups, have_groups, state):
-    cmds = []
-    api_key, val_key = _GROUP_TYPES[arg_key]
-    have_map = {g["name"]: g for g in (have_groups or [])}
-    want_map = {g["name"]: g for g in (groups or [])}
 
-    if state == "replaced":
-        for name in set(have_map) - set(want_map):
-            cmds.append(("delete", _BASE + [api_key, name]))
+def _group_from_device(name, data, member_key):
+    data = dict(data or {})
+    members = data.pop(member_key, None)
+    entry = {"name": name, **from_device(data)}
+    if members is not None:
+        member_list = [members] if isinstance(members, str) else members
+        entry[member_key] = sorted(str(m) for m in member_list)
+    return entry
 
-    for name, group in want_map.items():
-        have_group = have_map.get(name, {})
-        gbase = _BASE + [api_key, name]
 
-        if group.get("description") and group["description"] != have_group.get("description"):
-            cmds.append(("set", gbase + ["description", group["description"]]))
+def _groups_from_device(raw_groups, member_key):
+    if not raw_groups or not isinstance(raw_groups, dict):
+        return []
+    return [_group_from_device(name, data, member_key) for name, data in sorted(raw_groups.items())]
 
-        # normalize val_key for argspec (underscores)
-        arg_val_key = val_key.replace("-", "_")
-        want_vals = set(group.get(arg_val_key) or [])
-        have_vals = set(have_group.get(arg_val_key) or [])
 
-        for val in want_vals - have_vals:
-            cmds.append(("set", gbase + [val_key, val]))
+def get_running_config(vyos):
+    return vyos.get_config(_BASE) or {}
 
-        if state == "replaced":
-            for val in have_vals - want_vals:
-                cmds.append(("delete", gbase + [val_key, val]))
 
-    return cmds
+def _device_to_argspec(raw):
+    if not raw or not isinstance(raw, dict):
+        return {}
+    group = {}
+    for arg_key, (device_key, member_key) in _GROUP_TYPES.items():
+        groups = _groups_from_device(raw.get(device_key), member_key)
+        if groups:
+            group[arg_key] = groups
+    return {"group": group} if group else {}
 
 
-def build_commands(config, have, state):
-    cmds = []
+def build_commands(config, raw_have, state):
+    raw_have = raw_have or {}
+    want = _want_to_device(config)
+    norm_have = normalize_have(raw_have, _TAG_KEYS)
 
     if state == "deleted":
-        if have:
-            cmds.append(("delete", _BASE))
-        return cmds
+        return [("delete", _BASE)] if raw_have else []
 
+    commands = []
     if state == "replaced":
-        # Check if anything differs
-        would_set = build_commands(config, {}, "merged")
-        have_set = build_commands(have, {}, "merged")
-        if would_set == have_set:
-            return []
-
-    config = config or {}
-    want_group = config.get("group") or {}
-    have_group = have.get("group") or {}
-
-    for arg_key in _GROUP_TYPES:
-        want_groups = want_group.get(arg_key) or []
-        have_groups = have_group.get(arg_key) or []
-        if want_groups or (state == "replaced" and have_groups):
-            cmds += _group_cmds(arg_key, want_groups, have_groups, state)
-
-    return cmds
+        commands += dict_op(want, norm_have, _BASE, op="purge")
+    commands += dict_op(want, norm_have, _BASE, op="set")
+    return commands
 
 
 ARGUMENT_SPEC = dict(
@@ -373,12 +360,13 @@ def main():
     state = module.params["state"]
     config = module.params.get("config") or {}
 
-    have = get_running_config(vyos)
+    raw_have = get_running_config(vyos)
+    have = _device_to_argspec(raw_have)
 
     if state == "gathered":
         module.exit_json(changed=False, gathered=have)
 
-    commands = build_commands(config, have, state)
+    commands = build_commands(config, raw_have, state)
 
     if module.check_mode:
         module.exit_json(changed=bool(commands), commands=commands, before=have)
@@ -389,7 +377,7 @@ def main():
         module.exit_json(
             changed=True,
             before=have,
-            after=get_running_config(vyos),
+            after=_device_to_argspec(get_running_config(vyos)),
             commands=commands,
             saved=saved,
             response=response,
diff --git a/plugins/modules/vyos_firewall_interfaces.py b/plugins/modules/vyos_firewall_interfaces.py
index 8769f54..4b4b7eb 100644
--- a/plugins/modules/vyos_firewall_interfaces.py
+++ b/plugins/modules/vyos_firewall_interfaces.py
@@ -168,197 +168,166 @@ gathered:
   type: list
 saved:
   description: Whether the config was saved after changes.
-  returned: when changes are applied
+  returned: when changed
   type: bool
 response:
   description: Raw API response.
-  returned: when changes are applied
+  returned: always
   type: dict
 """
 
 from ansible.module_utils.basic import AnsibleModule
-from ansible_collections.vyos.rest.plugins.module_utils.vyos import VyOSModule
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    autoclean,
+    dict_op,
+    from_device,
+    normalize_have,
+)
 
 
 _BASE = ["firewall"]
-_AFIS = ["ipv4", "ipv6"]
-_HOOKS = ["input", "output", "forward"]
-
-
-def _parse_rule(rule_num, data):
-    rule = {"number": int(rule_num)}
-    data = data or {}
-    if "action" in data:
-        rule["action"] = data["action"]
-    if "description" in data:
-        rule["description"] = data["description"]
-    if "disable" in data:
-        rule["disable"] = True
-    if "protocol" in data:
-        rule["protocol"] = data["protocol"]
-    if "state" in data:
-        rule["state"] = data["state"]
-    if "log" in data:
-        rule["log"] = True
-
-    for endpoint in ["source", "destination"]:
-        ep = data.get(endpoint, {}) or {}
-        if ep:
-            rule[endpoint] = {}
-            if "address" in ep:
-                rule[endpoint]["address"] = ep["address"]
-            if "port" in ep:
-                rule[endpoint]["port"] = ep["port"]
-
-    return rule
-
-
-def _parse_hook_filter(hook, data):
-    entry = {"hook": hook}
-    data = data or {}
-    filter_data = data.get("filter", {}) or {}
-    if "default-action" in filter_data:
-        entry["default_action"] = filter_data["default-action"]
-    if "description" in filter_data:
-        entry["description"] = filter_data["description"]
-    rules_raw = filter_data.get("rule", {}) or {}
-    if rules_raw and isinstance(rules_raw, dict):
-        rules = [
-            _parse_rule(num, rdata)
-            for num, rdata in sorted(
-                rules_raw.items(),
-                key=lambda x: int(x[0]),
-            )
-        ]
-        if rules:
-            entry["rules"] = rules
-    return entry
 
+# The only hook filter keys this module owns under firewall.<afi>. Sibling
+# top-level keys under the same afi (e.g. firewall.<afi>.name, owned by
+# vyos_firewall_rules) are never enumerated or touched -- this module
+# only ever builds paths as _BASE + [afi, hook, "filter", ...] for hook
+# drawn from this fixed set, never a blanket op at _BASE + [afi] itself.
+_HOOKS = ("input", "output", "forward")
+_AFIS = ("ipv4", "ipv6")
 
-def get_running_config(vyos):
+# "rule" is a genuine tag node (keyed by rule number) that VyOS's REST API
+# can collapse to a bare value for a single rule with no other config.
+_TAG_KEYS = {"rule"}
+
+
+# ---------------------------------------------------------------------------
+# want -> device / device -> argspec
+#
+# Every leaf here is a direct structural match between argspec and device
+# shape (protocol, description, disable, state, log, source/destination
+# both flowing through autoclean/from_device generically). The only
+# unavoidable structural work: the "rule" tag-node reshape (keyed by
+# number) and inserting the literal "filter" wrapper key that VyOS
+# requires one level under each hook but the argspec omits (hook_entry's
+# fields live directly on it, not nested under a "filter" key).
+# ---------------------------------------------------------------------------
+
+
+def _rules_to_device(rules):
+    return {
+        str(r["number"]): autoclean({k: v for k, v in r.items() if k != "number"})
+        for r in rules or []
+    }
+
+
+def _rules_from_device(raw):
     result = []
-    for afi in _AFIS:
-        raw = vyos.get_config(_BASE + [afi])
-        if not raw or not isinstance(raw, dict):
-            continue
-        hooks = []
-        for hook in _HOOKS:
-            if hook in raw:
-                parsed = _parse_hook_filter(hook, raw[hook])
-                if len(parsed) > 1:  # more than just hook key
-                    hooks.append(parsed)
-        if hooks:
-            result.append({"afi": afi, "hooks": hooks})
+    for num, data in sorted((raw or {}).items(), key=lambda kv: int(kv[0])):
+        entry = {"number": int(num), **from_device(data or {})}
+        result.append(entry)
     return result
 
 
-def _rule_cmds(afi, hook, rule, have_rule):
-    cmds = []
-    rbase = _BASE + [afi, hook, "filter", "rule", str(rule["number"])]
-    have_rule = have_rule or {}
+def _hook_filter_to_device(hook_entry):
+    entry = autoclean({k: v for k, v in hook_entry.items() if k not in ("hook", "rules")})
+    if hook_entry.get("rules"):
+        entry["rule"] = _rules_to_device(hook_entry["rules"])
+    return entry
 
-    if rule.get("action") and rule["action"] != have_rule.get("action"):
-        cmds.append(("set", rbase + ["action", rule["action"]]))
-    if rule.get("description") and rule["description"] != have_rule.get("description"):
-        cmds.append(("set", rbase + ["description", rule["description"]]))
-    if rule.get("disable") and not have_rule.get("disable"):
-        cmds.append(("set", rbase + ["disable"]))
-    if rule.get("protocol") and rule["protocol"] != have_rule.get("protocol"):
-        cmds.append(("set", rbase + ["protocol", rule["protocol"]]))
-    if rule.get("state") and rule["state"] != have_rule.get("state"):
-        cmds.append(("set", rbase + ["state", rule["state"]]))
-    if rule.get("log") and not have_rule.get("log"):
-        cmds.append(("set", rbase + ["log"]))
 
-    for endpoint in ["source", "destination"]:
-        want_ep = rule.get(endpoint) or {}
-        have_ep = have_rule.get(endpoint) or {}
-        if want_ep.get("address") and want_ep["address"] != have_ep.get("address"):
-            cmds.append(("set", rbase + [endpoint, "address", want_ep["address"]]))
-        if want_ep.get("port") and want_ep["port"] != have_ep.get("port"):
-            cmds.append(("set", rbase + [endpoint, "port", str(want_ep["port"])]))
+def _hook_filter_from_device(hook, filter_data):
+    filter_data = dict(filter_data or {})
+    rules_raw = filter_data.pop("rule", None) or {}
+    entry = {"hook": hook, **from_device(filter_data)}
+    if rules_raw:
+        entry["rules"] = _rules_from_device(rules_raw)
+    return entry
 
-    return cmds
 
+def _want_to_device(config):
+    result = {}
+    for entry in config or []:
+        afi = entry["afi"]
+        hooks = entry.get("hooks") or []
+        if not hooks:
+            continue
+        result[afi] = {h["hook"]: {"filter": _hook_filter_to_device(h)} for h in hooks}
+    return result
 
-def _hook_cmds(afi, hook_entry, have_hook, state):
-    cmds = []
-    hook = hook_entry["hook"]
-    hbase = _BASE + [afi, hook, "filter"]
-    have_hook = have_hook or {}
 
-    if hook_entry.get("default_action") and hook_entry["default_action"] != have_hook.get(
-        "default_action",
-    ):
-        cmds.append(("set", hbase + ["default-action", hook_entry["default_action"]]))
-    if hook_entry.get("description") and hook_entry["description"] != have_hook.get("description"):
-        cmds.append(("set", hbase + ["description", hook_entry["description"]]))
+def get_running_config(vyos):
+    return vyos.get_config(_BASE) or {}
 
-    have_rules = {r["number"]: r for r in (have_hook.get("rules") or [])}
-    want_rules = {r["number"]: r for r in (hook_entry.get("rules") or [])}
 
-    if state == "replaced":
-        for num in set(have_rules) - set(want_rules):
-            cmds.append(("delete", hbase + ["rule", str(num)]))
+def _device_to_argspec(raw):
+    raw = raw or {}
+    result = []
+    for afi in _AFIS:
+        afi_raw = raw.get(afi) or {}
+        hooks = []
+        for hook in _HOOKS:
+            filter_data = (afi_raw.get(hook) or {}).get("filter")
+            if filter_data:
+                hooks.append(_hook_filter_from_device(hook, filter_data))
+        if hooks:
+            result.append({"afi": afi, "hooks": hooks})
+    return result
 
-    for num, rule in want_rules.items():
-        cmds += _rule_cmds(afi, hook, rule, have_rules.get(num))
 
-    return cmds
+# ---------------------------------------------------------------------------
+# Command building — dict_op scoped to _BASE + [afi, hook, "filter"] only,
+# per hook, never a blanket op at _BASE + [afi] or _BASE itself (which
+# would risk vyos_firewall_rules's firewall.<afi>.name subtree, even
+# though today the keys happen to differ -- staying scoped to the exact
+# owned path is the same discipline established for the BGP modules).
+# ---------------------------------------------------------------------------
 
 
-def build_commands(config, have_list, state):
-    cmds = []
+def build_commands(config, raw_have, state):
+    raw_have = raw_have or {}
+    config = config or []
+    norm_have = normalize_have(raw_have, _TAG_KEYS)
 
     if state == "deleted":
-        if not config:
-            if have_list:
-                for entry in have_list:
-                    afi = entry["afi"]
-                    for hook_entry in entry.get("hooks", []):
-                        cmds.append(("delete", _BASE + [afi, hook_entry["hook"], "filter"]))
-        else:
-            have_map = {(e["afi"], h["hook"]): h for e in have_list for h in e.get("hooks", [])}
-            for entry in config:
-                afi = entry["afi"]
-                for hook_entry in entry.get("hooks") or []:
-                    if (afi, hook_entry["hook"]) in have_map:
-                        cmds.append(("delete", _BASE + [afi, hook_entry["hook"], "filter"]))
-        return cmds
-
-    have_map = {e["afi"]: {h["hook"]: h for h in e.get("hooks", [])} for e in have_list}
-
-    if state == "overridden":
-        want_keys = {(e["afi"], h["hook"]) for e in (config or []) for h in e.get("hooks", [])}
-        for e in have_list:
-            for h in e.get("hooks", []):
-                if (e["afi"], h["hook"]) not in want_keys:
-                    cmds.append(("delete", _BASE + [e["afi"], h["hook"], "filter"]))
+        commands = []
+        # No config given -> delete every hook filter currently present.
+        # Config given -> delete only the (afi, hook) pairs it names.
+        targets = (
+            [(afi, hook) for afi in _AFIS for hook in _HOOKS]
+            if not config
+            else [(e["afi"], h["hook"]) for e in config for h in (e.get("hooks") or [])]
+        )
+        for afi, hook in targets:
+            if ((raw_have.get(afi) or {}).get(hook) or {}).get("filter"):
+                commands.append(("delete", _BASE + [afi, hook, "filter"]))
+        return commands
 
-    for entry in config or []:
-        afi = entry["afi"]
-        have_afi = have_map.get(afi, {})
+    want = _want_to_device(config)
+    commands = []
 
-        for hook_entry in entry.get("hooks") or []:
-            hook = hook_entry["hook"]
-            have_hook = have_afi.get(hook)
+    if state == "overridden":
+        want_pairs = {(afi, hook) for afi, hooks in want.items() for hook in hooks}
+        for afi in _AFIS:
+            for hook in _HOOKS:
+                if (afi, hook) not in want_pairs and (
+                    (raw_have.get(afi) or {}).get(hook) or {}
+                ).get(
+                    "filter",
+                ):
+                    commands.append(("delete", _BASE + [afi, hook, "filter"]))
 
-            if state == "replaced" and have_hook:
-                want_cmds = _hook_cmds(afi, hook_entry, {}, "merged")
-                have_hook_entry = {
-                    "hook": hook,
-                    "default_action": have_hook.get("default_action"),
-                    "rules": have_hook.get("rules", []),
-                }
-                have_cmds = _hook_cmds(afi, have_hook_entry, {}, "merged")
-                if want_cmds != have_cmds:
-                    cmds.append(("delete", _BASE + [afi, hook, "filter"]))
-                    have_hook = None
+    for afi, hooks in want.items():
+        for hook, want_hook in hooks.items():
+            hbase = _BASE + [afi, hook, "filter"]
+            have_filter = ((norm_have.get(afi) or {}).get(hook) or {}).get("filter") or {}
+            want_filter = want_hook.get("filter", {})
 
-            effective_state = state if state not in ("replaced", "overridden") else "merged"
-            cmds += _hook_cmds(afi, hook_entry, have_hook, effective_state)
+            if state in ("replaced", "overridden"):
+                commands += dict_op(want_filter, have_filter, hbase, op="purge")
+            commands += dict_op(want_filter, have_filter, hbase, op="set")
 
-    return cmds
+    return commands
 
 
 ARGUMENT_SPEC = dict(
@@ -444,12 +413,13 @@ def main():
     state = module.params["state"]
     config = module.params.get("config") or []
 
-    have = get_running_config(vyos)
+    raw_have = get_running_config(vyos)
+    have = _device_to_argspec(raw_have)
 
     if state == "gathered":
         module.exit_json(changed=False, gathered=have)
 
-    commands = build_commands(config, have, state)
+    commands = build_commands(config, raw_have, state)
 
     if module.check_mode:
         module.exit_json(changed=bool(commands), commands=commands, before=have)
@@ -460,7 +430,7 @@ def main():
         module.exit_json(
             changed=True,
             before=have,
-            after=get_running_config(vyos),
+            after=_device_to_argspec(get_running_config(vyos)),
             commands=commands,
             saved=saved,
             response=response,
diff --git a/plugins/modules/vyos_firewall_rules.py b/plugins/modules/vyos_firewall_rules.py
index a1c7c01..daf3915 100644
--- a/plugins/modules/vyos_firewall_rules.py
+++ b/plugins/modules/vyos_firewall_rules.py
@@ -127,6 +127,10 @@ notes:
   - C(ansible_network_os) must be set to C(vyos.rest.vyos).
   - Rule sets are identified by AFI and name. Deleting a rule set removes
     all its rules.
+  - The C(group) suboption can only reference an address-group. VyOS also
+    supports network-group/port-group/domain-group references, which this
+    module can read back (via C(gathered)) if already configured by other
+    means, but cannot create -- the argspec has no group-type discriminator.
 """
 
 EXAMPLES = r"""
@@ -185,248 +189,196 @@ gathered:
   type: list
 saved:
   description: Whether the config was saved after changes.
-  returned: when changes are applied
+  returned: when changed
   type: bool
 response:
   description: Raw API response.
-  returned: when changes are applied
+  returned: always
   type: dict
 """
 
 from ansible.module_utils.basic import AnsibleModule
-from ansible_collections.vyos.rest.plugins.module_utils.vyos import VyOSModule
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    autoclean,
+    dict_op,
+    from_device,
+    normalize_have,
+)
 
 
 _BASE = ["firewall"]
-_AFIS = ["ipv4", "ipv6"]
-
-
-def _parse_rule(rule_num, data):
-    rule = {"number": int(rule_num)}
-    data = data or {}
-    if "action" in data:
-        rule["action"] = data["action"]
-    if "description" in data:
-        rule["description"] = data["description"]
-    if "disable" in data:
-        rule["disable"] = True
-    if "protocol" in data:
-        rule["protocol"] = data["protocol"]
-    if "state" in data:
-        rule["state"] = data["state"]
-    if "log" in data:
-        rule["log"] = True
-
-    src = data.get("source", {}) or {}
-    if src:
-        rule["source"] = {}
-        if "address" in src:
-            rule["source"]["address"] = src["address"]
-        if "group" in src:
-            grp = src["group"]
-            if isinstance(grp, dict):
-                rule["source"]["group"] = list(grp.values())[0] if grp else None
-            else:
-                rule["source"]["group"] = grp
-        if "port" in src:
-            rule["source"]["port"] = src["port"]
-
-    dst = data.get("destination", {}) or {}
-    if dst:
-        rule["destination"] = {}
-        if "address" in dst:
-            rule["destination"]["address"] = dst["address"]
-        if "group" in dst:
-            grp = dst["group"]
-            if isinstance(grp, dict):
-                rule["destination"]["group"] = list(grp.values())[0] if grp else None
-            else:
-                rule["destination"]["group"] = grp
-        if "port" in dst:
-            rule["destination"]["port"] = dst["port"]
-
-    icmp = data.get("icmp", {}) or {}
-    if icmp:
-        rule["icmp"] = {}
-        if "type" in icmp:
-            rule["icmp"]["type"] = int(icmp["type"])
-        if "code" in icmp:
-            rule["icmp"]["code"] = int(icmp["code"])
-
-    return rule
-
-
-def _parse_rule_set(rs_name, data):
-    rs = {"name": rs_name}
-    data = data or {}
-    if "default-action" in data:
-        rs["default_action"] = data["default-action"]
-    if "description" in data:
-        rs["description"] = data["description"]
-    rules_raw = data.get("rule", {}) or {}
-    if rules_raw and isinstance(rules_raw, dict):
-        rules = [
-            _parse_rule(num, rdata)
-            for num, rdata in sorted(
-                rules_raw.items(),
-                key=lambda x: int(x[0]),
-            )
-        ]
-        if rules:
-            rs["rules"] = rules
-    return rs
+_AFIS = ("ipv4", "ipv6")
+
+# Tag nodes VyOS's REST API can collapse to a bare string/list for a
+# single entry with no other config -- "name" (rule sets, keyed by name)
+# and "rule" (rules, keyed by number).
+_TAG_KEYS = {"name", "rule"}
+
+
+# ---------------------------------------------------------------------------
+# want -> device / device -> argspec
+#
+# Every leaf here matches the device shape directly (action, description,
+# disable, protocol, state, log, icmp.type/code) except one: "group".
+# VyOS wraps a group reference under a literal group-kind key
+# (address-group/network-group/...), not a flat value -- see the module
+# note above on why this module can only ever *write* address-group.
+# Rule-set/rule tag-node reshaping (keyed by name/number) is the other
+# unavoidable structural work.
+# ---------------------------------------------------------------------------
+
+
+def _endpoint_to_device(ep):
+    entry = autoclean({k: v for k, v in ep.items() if k != "group"})
+    if ep.get("group"):
+        entry["group"] = {"address-group": ep["group"]}
+    return entry
+
+
+def _endpoint_from_device(data):
+    data = dict(data or {})
+    group = data.pop("group", None)
+    entry = from_device(data)
+    if isinstance(group, dict) and group:
+        entry["group"] = list(group.values())[0]
+    elif isinstance(group, str):
+        entry["group"] = group
+    return entry
+
+
+def _rules_to_device(rules):
+    result = {}
+    for r in rules or []:
+        entry = autoclean(
+            {k: v for k, v in r.items() if k not in ("number", "source", "destination")},
+        )
+        for endpoint in ("source", "destination"):
+            if r.get(endpoint):
+                entry[endpoint] = _endpoint_to_device(r[endpoint])
+        result[str(r["number"])] = entry
+    return result
 
 
-def get_running_config(vyos):
+def _rules_from_device(raw):
     result = []
-    for afi in _AFIS:
-        raw = vyos.get_config(_BASE + [afi, "name"])
-        if not raw or not isinstance(raw, dict):
-            continue
-        # unwrap "name" key if present
-        raw = raw.get("name", raw)
-        if not raw or not isinstance(raw, dict):
-            continue
-        rule_sets = [_parse_rule_set(name, data) for name, data in sorted(raw.items())]
-        if rule_sets:
-            result.append({"afi": afi, "rule_sets": rule_sets})
+    for num, data in sorted((raw or {}).items(), key=lambda kv: int(kv[0])):
+        data = dict(data or {})
+        src = data.pop("source", None)
+        dst = data.pop("destination", None)
+        entry = {"number": int(num), **from_device(data)}
+        if src:
+            entry["source"] = _endpoint_from_device(src)
+        if dst:
+            entry["destination"] = _endpoint_from_device(dst)
+        result.append(entry)
     return result
 
 
-def _rule_cmds(rs_name, afi, rule, have_rule):
-    cmds = []
-    rbase = _BASE + [afi, "name", rs_name, "rule", str(rule["number"])]
-    have_rule = have_rule or {}
-
-    if rule.get("action") and rule["action"] != have_rule.get("action"):
-        cmds.append(("set", rbase + ["action", rule["action"]]))
-    if rule.get("description") and rule["description"] != have_rule.get("description"):
-        cmds.append(("set", rbase + ["description", rule["description"]]))
-    if rule.get("disable") and not have_rule.get("disable"):
-        cmds.append(("set", rbase + ["disable"]))
-    if rule.get("protocol") and rule["protocol"] != have_rule.get("protocol"):
-        cmds.append(("set", rbase + ["protocol", rule["protocol"]]))
-    if rule.get("state") and rule["state"] != have_rule.get("state"):
-        cmds.append(("set", rbase + ["state", rule["state"]]))
-    if rule.get("log") and not have_rule.get("log"):
-        cmds.append(("set", rbase + ["log"]))
-
-    for endpoint in ["source", "destination"]:
-        want_ep = rule.get(endpoint) or {}
-        have_ep = have_rule.get(endpoint) or {}
-        if want_ep.get("address") and want_ep["address"] != have_ep.get("address"):
-            cmds.append(("set", rbase + [endpoint, "address", want_ep["address"]]))
-        if want_ep.get("port") and want_ep["port"] != have_ep.get("port"):
-            cmds.append(("set", rbase + [endpoint, "port", str(want_ep["port"])]))
-        if want_ep.get("group") and want_ep["group"] != have_ep.get("group"):
-            cmds.append(
-                (
-                    "set",
-                    rbase
-                    + [
-                        endpoint,
-                        "group",
-                        "address-group",
-                        want_ep["group"],
-                    ],
-                ),
-            )
+def _rule_set_to_device(rs):
+    entry = autoclean({k: v for k, v in rs.items() if k not in ("name", "rules")})
+    if rs.get("rules"):
+        entry["rule"] = _rules_to_device(rs["rules"])
+    return entry
 
-    icmp = rule.get("icmp") or {}
-    have_icmp = have_rule.get("icmp") or {}
-    if icmp.get("type") and icmp["type"] != have_icmp.get("type"):
-        cmds.append(("set", rbase + ["icmp", "type", str(icmp["type"])]))
-    if icmp.get("code") and icmp["code"] != have_icmp.get("code"):
-        cmds.append(("set", rbase + ["icmp", "code", str(icmp["code"])]))
 
-    return cmds
+def _rule_set_from_device(name, data):
+    data = dict(data or {})
+    rules_raw = data.pop("rule", None) or {}
+    entry = {"name": name, **from_device(data)}
+    if rules_raw:
+        entry["rules"] = _rules_from_device(rules_raw)
+    return entry
 
 
-def _rule_set_cmds(afi, rs, have_rs, state):
-    cmds = []
-    rs_name = rs["name"]
-    rsbase = _BASE + [afi, "name", rs_name]
-    have_rs = have_rs or {}
+def _want_to_device(config):
+    result = {}
+    for entry in config or []:
+        afi = entry["afi"]
+        rule_sets = entry.get("rule_sets") or []
+        if not rule_sets:
+            continue
+        result[afi] = {rs["name"]: _rule_set_to_device(rs) for rs in rule_sets}
+    return result
 
-    if rs.get("default_action") and rs["default_action"] != have_rs.get("default_action"):
-        cmds.append(("set", rsbase + ["default-action", rs["default_action"]]))
-    if rs.get("description") and rs["description"] != have_rs.get("description"):
-        cmds.append(("set", rsbase + ["description", rs["description"]]))
 
-    have_rules = {r["number"]: r for r in (have_rs.get("rules") or [])}
-    want_rules = {r["number"]: r for r in (rs.get("rules") or [])}
+def get_running_config(vyos):
+    """Fetch each AFI's rule-set subtree directly at firewall.<afi>.name --
+    the most targeted path available, deliberately not a broader fetch at
+    firewall.<afi> or firewall itself (which would pull in the hook-filter
+    and group subtrees owned by sibling modules for no benefit here).
+    """
+    result = {}
+    for afi in _AFIS:
+        raw = vyos.get_config(_BASE + [afi, "name"])
+        if raw and isinstance(raw, dict):
+            # Some VyOS REST responses wrap the result in an extra "name"
+            # key even when fetched at a path already ending in "name";
+            # unwrap defensively either way.
+            raw = raw.get("name", raw)
+            if raw and isinstance(raw, dict):
+                result[afi] = raw
+    return result
 
-    if state == "replaced":
-        for num in set(have_rules) - set(want_rules):
-            cmds.append(("delete", rsbase + ["rule", str(num)]))
 
-    for num, rule in want_rules.items():
-        cmds += _rule_cmds(rs_name, afi, rule, have_rules.get(num))
+def _device_to_argspec(raw):
+    raw = raw or {}
+    result = []
+    for afi in _AFIS:
+        afi_raw = raw.get(afi) or {}
+        rule_sets = [_rule_set_from_device(name, data) for name, data in sorted(afi_raw.items())]
+        if rule_sets:
+            result.append({"afi": afi, "rule_sets": rule_sets})
+    return result
 
-    return cmds
 
+# ---------------------------------------------------------------------------
+# Command building — dict_op scoped to _BASE + [afi, "name", rs_name] per
+# rule set, never a blanket op at _BASE + [afi] or _BASE itself (which
+# would risk vyos_firewall_interfaces's hook-filter subtree and
+# vyos_firewall_global's group subtree under the same "firewall" root).
+# ---------------------------------------------------------------------------
 
-def build_commands(config, have_list, state):
-    cmds = []
+
+def build_commands(config, raw_have, state):
+    raw_have = raw_have or {}
+    config = config or []
+    norm_have = {afi: normalize_have(data, _TAG_KEYS) for afi, data in raw_have.items()}
 
     if state == "deleted":
+        commands = []
         if not config:
-            if have_list:
-                cmds.append(("delete", _BASE))
+            for afi, rule_sets in raw_have.items():
+                for name in rule_sets:
+                    commands.append(("delete", _BASE + [afi, "name", name]))
         else:
-            have_map = {
-                (e["afi"], rs["name"]): rs for e in have_list for rs in e.get("rule_sets", [])
-            }
             for entry in config:
                 afi = entry["afi"]
                 for rs in entry.get("rule_sets") or []:
-                    if (afi, rs["name"]) in have_map:
-                        cmds.append(("delete", _BASE + [afi, "name", rs["name"]]))
-        return cmds
+                    if rs["name"] in (raw_have.get(afi) or {}):
+                        commands.append(("delete", _BASE + [afi, "name", rs["name"]]))
+        return commands
 
-    have_map = {e["afi"]: {rs["name"]: rs for rs in e.get("rule_sets", [])} for e in have_list}
+    want = _want_to_device(config)
+    commands = []
 
     if state == "overridden":
-        want_keys = {
-            (e["afi"], rs["name"]) for e in (config or []) for rs in e.get("rule_sets", [])
-        }
-        for e in have_list:
-            for rs in e.get("rule_sets", []):
-                if (e["afi"], rs["name"]) not in want_keys:
-                    cmds.append(("delete", _BASE + [e["afi"], "name", rs["name"]]))
+        want_keys = {(afi, name) for afi, rule_sets in want.items() for name in rule_sets}
+        for afi, rule_sets in raw_have.items():
+            for name in rule_sets:
+                if (afi, name) not in want_keys:
+                    commands.append(("delete", _BASE + [afi, "name", name]))
 
-    for entry in config or []:
-        afi = entry["afi"]
-        have_afi = have_map.get(afi, {})
-
-        for rs in entry.get("rule_sets") or []:
-            have_rs = have_afi.get(rs["name"])
-
-            if state == "replaced" and have_rs:
-                # delete and rebuild if different
-                want_cmds = _rule_set_cmds(afi, rs, {}, "merged")
-                have_cmds = _rule_set_cmds(
-                    afi,
-                    {
-                        "name": rs["name"],
-                        "default_action": have_rs.get("default_action"),
-                        "rules": have_rs.get("rules", []),
-                    },
-                    {},
-                    "merged",
-                )
-                if want_cmds != have_cmds:
-                    cmds.append(("delete", _BASE + [afi, "name", rs["name"]]))
-                    have_rs = None
-
-            cmds += _rule_set_cmds(
-                afi,
-                rs,
-                have_rs,
-                state if state not in ("replaced", "overridden") else "merged",
-            )
-
-    return cmds
+    for afi, rule_sets in want.items():
+        for name, want_rs in rule_sets.items():
+            rsbase = _BASE + [afi, "name", name]
+            have_rs = (norm_have.get(afi) or {}).get(name) or {}
+
+            if state in ("replaced", "overridden"):
+                commands += dict_op(want_rs, have_rs, rsbase, op="purge")
+            commands += dict_op(want_rs, have_rs, rsbase, op="set")
+
+    return commands
 
 
 ARGUMENT_SPEC = dict(
@@ -517,12 +469,13 @@ def main():
     state = module.params["state"]
     config = module.params.get("config") or []
 
-    have = get_running_config(vyos)
+    raw_have = get_running_config(vyos)
+    have = _device_to_argspec(raw_have)
 
     if state == "gathered":
         module.exit_json(changed=False, gathered=have)
 
-    commands = build_commands(config, have, state)
+    commands = build_commands(config, raw_have, state)
 
     if module.check_mode:
         module.exit_json(changed=bool(commands), commands=commands, before=have)
@@ -533,7 +486,7 @@ def main():
         module.exit_json(
             changed=True,
             before=have,
-            after=get_running_config(vyos),
+            after=_device_to_argspec(get_running_config(vyos)),
             commands=commands,
             saved=saved,
             response=response,
diff --git a/plugins/modules/vyos_ha.py b/plugins/modules/vyos_ha.py
new file mode 100644
index 0000000..ff1705f
--- /dev/null
+++ b/plugins/modules/vyos_ha.py
@@ -0,0 +1,783 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# GNU General Public License v3.0+
+from __future__ import absolute_import, division, print_function
+
+
+__metaclass__ = type
+
+DOCUMENTATION = r"""
+module: vyos_ha
+short_description: Manage VRRP and load balancer configuration on VyOS via REST API
+description:
+- Manages VRRP groups, global VRRP parameters, sync-groups, virtual servers, and LVS
+  real servers on VyOS devices via the REST API.
+- Uses REST API (C(connection=httpapi)) instead of CLI.
+- Targets VyOS 1.4+.
+version_added: 1.0.0
+author:
+- Evgeny Molotkov (@omnom62)
+options:
+  config:
+    description: High-availability configuration.
+    type: dict
+    suboptions:
+      disable:
+        description: Disable all high-availability configuration.
+        type: bool
+        default: false
+      virtual_servers:
+        description: List of load balancer virtual server definitions.
+        type: list
+        elements: dict
+        suboptions:
+          name:
+            type: str
+            required: true
+            description: Name.
+          address:
+            type: str
+            description: Address.
+          algorithm:
+            type: str
+            description: Algorithm.
+          delay_loop:
+            type: int
+            description: Delay loop.
+          forward_method:
+            type: str
+            choices:
+            - direct
+            - nat
+            description: Forward method.
+          fwmark:
+            type: int
+            description: Fwmark.
+          persistence_timeout:
+            type: int
+            description: Persistence timeout.
+          port:
+            type: int
+            description: Port.
+          protocol:
+            type: str
+            choices:
+            - tcp
+            - udp
+            description: Protocol.
+          real_server:
+            type: list
+            elements: dict
+            suboptions:
+              address:
+                type: str
+                required: true
+                description: Address.
+              port:
+                type: int
+                description: Port.
+              connection_timeout:
+                type: int
+                description: Connection timeout.
+              health_check_script:
+                type: str
+                description: Health check script.
+            description: Real server.
+      vrrp:
+        description: VRRP configuration.
+        type: dict
+        suboptions:
+          global_parameters:
+            type: dict
+            suboptions:
+              garp:
+                type: dict
+                suboptions:
+                  interval:
+                    type: int
+                    description: Interval.
+                  master_delay:
+                    type: int
+                    description: Master delay.
+                  master_refresh:
+                    type: int
+                    description: Master refresh.
+                  master_refresh_repeat:
+                    type: int
+                    description: Master refresh repeat.
+                  master_repeat:
+                    type: int
+                    description: Master repeat.
+                description: Garp.
+              startup_delay:
+                type: int
+                description: Startup delay.
+              version:
+                type: str
+                description: Version.
+            description: Global parameters.
+          groups:
+            type: list
+            elements: dict
+            suboptions:
+              name:
+                type: str
+                required: true
+                description: Name.
+              address:
+                type: list
+                elements: str
+                description: Address.
+              advertise_interval:
+                type: int
+                description: Advertise interval.
+              authentication:
+                type: dict
+                suboptions:
+                  password:
+                    type: str
+                    description: Password.
+                  type:
+                    type: str
+                    description: Type.
+                description: Authentication.
+              description:
+                type: str
+                description: Description.
+              disable:
+                type: bool
+                default: false
+                description: Disable.
+              excluded_address:
+                type: list
+                elements: str
+                description: Excluded address.
+              garp:
+                type: dict
+                suboptions:
+                  interval:
+                    type: int
+                    description: Interval.
+                  master_delay:
+                    type: int
+                    description: Master delay.
+                  master_refresh:
+                    type: int
+                    description: Master refresh.
+                  master_refresh_repeat:
+                    type: int
+                    description: Master refresh repeat.
+                  master_repeat:
+                    type: int
+                    description: Master repeat.
+                description: Garp.
+              health_check:
+                type: dict
+                suboptions:
+                  failure_count:
+                    type: int
+                    description: Failure count.
+                  interval:
+                    type: int
+                    description: Interval.
+                  ping:
+                    type: str
+                    description: Ping.
+                  script:
+                    type: str
+                    description: Script.
+                description: Health check.
+              hello_source_address:
+                type: str
+                description: Hello source address.
+              interface:
+                type: str
+                description: Interface.
+              no_preempt:
+                type: bool
+                default: false
+                description: No preempt.
+              peer_address:
+                type: str
+                description: Peer address.
+              preempt_delay:
+                type: int
+                description: Preempt delay.
+              priority:
+                type: int
+                description: Priority.
+              rfc3768_compatibility:
+                type: bool
+                default: false
+                description: Rfc3768 compatibility.
+              track:
+                type: dict
+                suboptions:
+                  exclude_vrrp_interface:
+                    type: bool
+                    description: Exclude vrrp interface.
+                  interface:
+                    type: list
+                    elements: str
+                    description: Interface.
+                description: Track.
+              transition_script:
+                type: dict
+                suboptions:
+                  backup:
+                    type: str
+                    description: Backup.
+                  fault:
+                    type: str
+                    description: Fault.
+                  master:
+                    type: str
+                    description: Master.
+                  stop:
+                    type: str
+                    description: Stop.
+                description: Transition script.
+              vrid:
+                type: int
+                description: Vrid.
+            description: Groups.
+          snmp:
+            type: str
+            choices:
+            - enabled
+            - disabled
+            description: Snmp.
+          sync_groups:
+            type: list
+            elements: dict
+            suboptions:
+              name:
+                type: str
+                required: true
+                description: Name.
+              health_check:
+                type: dict
+                suboptions:
+                  failure_count:
+                    type: int
+                    description: Failure count.
+                  interval:
+                    type: int
+                    description: Interval.
+                  ping:
+                    type: str
+                    description: Ping.
+                  script:
+                    type: str
+                    description: Script.
+                description: Health check.
+              member:
+                type: list
+                elements: str
+                description: Member.
+              transition_script:
+                type: dict
+                suboptions:
+                  backup:
+                    type: str
+                    description: Backup.
+                  fault:
+                    type: str
+                    description: Fault.
+                  master:
+                    type: str
+                    description: Master.
+                  stop:
+                    type: str
+                    description: Stop.
+                description: Transition script.
+            description: Sync groups.
+  state:
+    description: Desired end state of the configuration.
+    type: str
+    choices:
+    - merged
+    - replaced
+    - overridden
+    - deleted
+    - gathered
+    default: merged
+
+"""
+
+EXAMPLES = r"""
+- name: Merge VRRP configuration
+  vyos.rest.vyos_ha:
+    config:
+      vrrp:
+        global_parameters:
+          startup_delay: 30
+        groups:
+          - name: g1
+            interface: eth0
+            vrid: 20
+            priority: 100
+            address:
+              - 192.168.1.100/24
+        sync_groups:
+          - name: sg1
+            member: [g1]
+        snmp: enabled
+    state: merged
+
+- name: Delete all HA configuration
+  vyos.rest.vyos_ha:
+    state: deleted
+
+- name: Gather current HA configuration
+  vyos.rest.vyos_ha:
+    state: gathered
+"""
+
+RETURN = r"""
+before:
+  description: HA configuration before this module ran.
+  returned: always
+  type: dict
+after:
+  description: HA configuration after this module ran.
+  returned: when changed
+  type: dict
+commands:
+  description: List of API commands sent to the device.
+  returned: always
+  type: list
+gathered:
+  description: Current HA configuration as structured data.
+  returned: when state is gathered
+  type: dict
+saved:
+  description: Whether the config was saved after changes.
+  returned: when changed
+  type: bool
+"""
+
+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,
+    normalize_have,
+    to_tag_dict,
+)
+
+
+_BASE = ["high-availability"]
+
+ARGUMENT_SPEC = dict(
+    config=dict(
+        type="dict",
+        options=dict(
+            disable=dict(type="bool", default=False),
+            virtual_servers=dict(
+                type="list",
+                elements="dict",
+                options=dict(
+                    name=dict(type="str", required=True),
+                    address=dict(type="str"),
+                    algorithm=dict(type="str"),
+                    delay_loop=dict(type="int"),
+                    forward_method=dict(type="str", choices=["direct", "nat"]),
+                    fwmark=dict(type="int"),
+                    persistence_timeout=dict(type="int"),
+                    port=dict(type="int"),
+                    protocol=dict(type="str", choices=["tcp", "udp"]),
+                    real_server=dict(
+                        type="list",
+                        elements="dict",
+                        options=dict(
+                            address=dict(type="str", required=True),
+                            port=dict(type="int"),
+                            connection_timeout=dict(type="int"),
+                            health_check_script=dict(type="str"),
+                        ),
+                    ),
+                ),
+            ),
+            vrrp=dict(
+                type="dict",
+                options=dict(
+                    global_parameters=dict(
+                        type="dict",
+                        options=dict(
+                            garp=dict(
+                                type="dict",
+                                options=dict(
+                                    interval=dict(type="int"),
+                                    master_delay=dict(type="int"),
+                                    master_refresh=dict(type="int"),
+                                    master_refresh_repeat=dict(type="int"),
+                                    master_repeat=dict(type="int"),
+                                ),
+                            ),
+                            startup_delay=dict(type="int"),
+                            version=dict(type="str"),
+                        ),
+                    ),
+                    groups=dict(
+                        type="list",
+                        elements="dict",
+                        options=dict(
+                            name=dict(type="str", required=True),
+                            address=dict(type="list", elements="str"),
+                            advertise_interval=dict(type="int"),
+                            authentication=dict(
+                                type="dict",
+                                options=dict(
+                                    password=dict(type="str", no_log=True),
+                                    type=dict(type="str"),
+                                ),
+                            ),
+                            description=dict(type="str"),
+                            disable=dict(type="bool", default=False),
+                            excluded_address=dict(type="list", elements="str"),
+                            garp=dict(
+                                type="dict",
+                                options=dict(
+                                    interval=dict(type="int"),
+                                    master_delay=dict(type="int"),
+                                    master_refresh=dict(type="int"),
+                                    master_refresh_repeat=dict(type="int"),
+                                    master_repeat=dict(type="int"),
+                                ),
+                            ),
+                            health_check=dict(
+                                type="dict",
+                                options=dict(
+                                    failure_count=dict(type="int"),
+                                    interval=dict(type="int"),
+                                    ping=dict(type="str"),
+                                    script=dict(type="str"),
+                                ),
+                            ),
+                            hello_source_address=dict(type="str"),
+                            interface=dict(type="str"),
+                            no_preempt=dict(type="bool", default=False),
+                            peer_address=dict(type="str"),
+                            preempt_delay=dict(type="int"),
+                            priority=dict(type="int"),
+                            rfc3768_compatibility=dict(type="bool", default=False),
+                            track=dict(
+                                type="dict",
+                                options=dict(
+                                    exclude_vrrp_interface=dict(type="bool"),
+                                    interface=dict(type="list", elements="str"),
+                                ),
+                            ),
+                            transition_script=dict(
+                                type="dict",
+                                options=dict(
+                                    backup=dict(type="str"),
+                                    fault=dict(type="str"),
+                                    master=dict(type="str"),
+                                    stop=dict(type="str"),
+                                ),
+                            ),
+                            vrid=dict(type="int"),
+                        ),
+                    ),
+                    snmp=dict(type="str", choices=["enabled", "disabled"]),
+                    sync_groups=dict(
+                        type="list",
+                        elements="dict",
+                        options=dict(
+                            name=dict(type="str", required=True),
+                            health_check=dict(
+                                type="dict",
+                                options=dict(
+                                    failure_count=dict(type="int"),
+                                    interval=dict(type="int"),
+                                    ping=dict(type="str"),
+                                    script=dict(type="str"),
+                                ),
+                            ),
+                            member=dict(type="list", elements="str"),
+                            transition_script=dict(
+                                type="dict",
+                                options=dict(
+                                    backup=dict(type="str"),
+                                    fault=dict(type="str"),
+                                    master=dict(type="str"),
+                                    stop=dict(type="str"),
+                                ),
+                            ),
+                        ),
+                    ),
+                ),
+            ),
+        ),
+    ),
+    state=dict(
+        default="merged",
+        choices=["merged", "replaced", "overridden", "deleted", "gathered"],
+    ),
+)
+
+_TOP_OPTIONS = ARGUMENT_SPEC["config"]["options"]
+_VS_OPTIONS = _TOP_OPTIONS["virtual_servers"]["options"]
+_RS_OPTIONS = _VS_OPTIONS["real_server"]["options"]
+_VRRP_OPTIONS = _TOP_OPTIONS["vrrp"]["options"]
+_GROUP_OPTIONS = _VRRP_OPTIONS["groups"]["options"]
+_SYNC_GROUP_OPTIONS = _VRRP_OPTIONS["sync_groups"]["options"]
+
+# Tag nodes VyOS's REST API can collapse to a bare string/list for a
+# single entry with no other config. Split by section because "address"
+# means two different things depending on where it appears -- confirmed
+# against vyos-1x: vrrp.group.<name>.address is a genuine tagNode (VRRP
+# virtual IPs, each with real child structure), but virtual-server.
+# <name>.address is a flat scalar string (the load-balancer's own bind
+# address). A single blanket key-name-based coercion across the whole
+# raw tree would wrongly reshape the latter into a tag-node dict --
+# exactly the class of bug this split avoids.
+_VS_TAG_KEYS = {"virtual-server", "real-server"}
+_VRRP_TAG_KEYS = {"group", "sync-group", "address", "excluded-address"}
+
+# track.interface and sync-group.member are NOT included above --
+# confirmed <leafNode><multi/>, i.e. plain multi-value leaves, not tag
+# nodes; dict_op's own native list handling (which already corrects for
+# the same single-value-collapse quirk) applies to them directly, no
+# reshaping needed.
+
+
+# ---------------------------------------------------------------------------
+# Structural adapters — the only genuine exceptions, confirmed against
+# vyos-1x schema, not assumed:
+# 1. Named-object lists (virtual_servers, real_server, groups,
+#    sync_groups): argspec uses [{name: "x", ...}], device uses
+#    {"x": {...}}.
+# 2. address / excluded_address (VRRP group virtual IPs): genuine
+#    tagNodes (each has real child structure) -> {"a": {}, "b": {}}.
+# 3. snmp: argspec "enabled"/"disabled" string <-> device presence node
+#    (present) / absent. "disabled" has no device-side representation at
+#    all -- see the explicit delete in build_commands().
+# 4. health_check_script: argspec flat field <-> device nested under
+#    health-check.script.
+#
+# Everything else -- including track.interface and sync_group.member,
+# both plain multi-value leaves despite superficially looking like the
+# same shape as address/excluded_address -- flows through autoclean/
+# from_device untouched.
+# ---------------------------------------------------------------------------
+
+
+def _real_server_to_device(rs):
+    entry = autoclean(
+        {k: v for k, v in rs.items() if k not in ("address", "health_check_script")},
+    )
+    if rs.get("health_check_script"):
+        entry["health-check"] = {"script": rs["health_check_script"]}
+    return entry
+
+
+def _real_server_from_device(addr, data):
+    data = dict(data or {})
+    hc = data.pop("health-check", None) or {}
+    entry = {"address": addr, **from_device(data)}
+    if hc.get("script"):
+        entry["health_check_script"] = hc["script"]
+    cast_by_spec(entry, _RS_OPTIONS)
+    return entry
+
+
+def _virtual_server_to_device(vs):
+    entry = autoclean({k: v for k, v in vs.items() if k not in ("name", "real_server")})
+    if vs.get("real_server"):
+        entry["real-server"] = {
+            rs["address"]: _real_server_to_device(rs) for rs in vs["real_server"]
+        }
+    return entry
+
+
+def _virtual_server_from_device(name, data):
+    data = dict(data or {})
+    rs_raw = data.pop("real-server", None) or {}
+    entry = {"name": name, **from_device(data)}
+    cast_by_spec(entry, _VS_OPTIONS)
+    if rs_raw:
+        entry["real_server"] = [
+            _real_server_from_device(addr, rdata) for addr, rdata in sorted(rs_raw.items())
+        ]
+    return entry
+
+
+def _group_to_device(grp):
+    entry = autoclean(
+        {k: v for k, v in grp.items() if k not in ("name", "address", "excluded_address")},
+    )
+    if grp.get("address"):
+        entry["address"] = {a: {} for a in grp["address"]}
+    if grp.get("excluded_address"):
+        entry["excluded-address"] = {a: {} for a in grp["excluded_address"]}
+    return entry
+
+
+def _group_from_device(name, data):
+    data = dict(data or {})
+    addr_raw = data.pop("address", None)
+    excl_raw = data.pop("excluded-address", None)
+    entry = {"name": name, **from_device(data)}
+    cast_by_spec(entry, _GROUP_OPTIONS)
+    if addr_raw:
+        entry["address"] = sorted(to_tag_dict(addr_raw).keys())
+    if excl_raw:
+        entry["excluded_address"] = sorted(to_tag_dict(excl_raw).keys())
+    return entry
+
+
+def _sync_group_from_device(name, data):
+    entry = {"name": name, **from_device(data or {})}
+    cast_by_spec(entry, _SYNC_GROUP_OPTIONS)
+    return entry
+
+
+def _want_to_device(config):
+    if not config:
+        return {}
+    want = autoclean({k: v for k, v in config.items() if k not in ("virtual_servers", "vrrp")})
+
+    if config.get("virtual_servers"):
+        want["virtual-server"] = {
+            vs["name"]: _virtual_server_to_device(vs) for vs in config["virtual_servers"]
+        }
+
+    vrrp = config.get("vrrp") or {}
+    if vrrp:
+        vrrp_dev = autoclean(
+            {k: v for k, v in vrrp.items() if k not in ("groups", "sync_groups", "snmp")},
+        )
+        # snmp: "enabled" -> presence node; "disabled" has no device-side
+        # form at all (handled via an explicit delete in build_commands).
+        if vrrp.get("snmp") == "enabled":
+            vrrp_dev["snmp"] = {}
+        if vrrp.get("groups"):
+            vrrp_dev["group"] = {g["name"]: _group_to_device(g) for g in vrrp["groups"]}
+        if vrrp.get("sync_groups"):
+            vrrp_dev["sync-group"] = {
+                sg["name"]: autoclean({k: v for k, v in sg.items() if k != "name"})
+                for sg in vrrp["sync_groups"]
+            }
+        if vrrp_dev:
+            want["vrrp"] = vrrp_dev
+
+    return want
+
+
+def get_running_config(vyos):
+    return vyos.get_config(_BASE) or {}
+
+
+def _device_to_argspec(raw):
+    if not raw:
+        return {}
+    result = from_device({k: v for k, v in raw.items() if k not in ("virtual-server", "vrrp")})
+    cast_by_spec(result, _TOP_OPTIONS)
+
+    vs_raw = raw.get("virtual-server") or {}
+    if vs_raw:
+        result["virtual_servers"] = [
+            _virtual_server_from_device(name, data) for name, data in sorted(vs_raw.items())
+        ]
+
+    vrrp_raw = raw.get("vrrp") or {}
+    if vrrp_raw:
+        vrrp_arg = from_device(
+            {k: v for k, v in vrrp_raw.items() if k not in ("group", "sync-group", "snmp")},
+        )
+        cast_by_spec(vrrp_arg, _VRRP_OPTIONS)
+        if "snmp" in vrrp_raw:
+            vrrp_arg["snmp"] = "enabled"
+
+        grp_raw = vrrp_raw.get("group") or {}
+        if grp_raw:
+            vrrp_arg["groups"] = [
+                _group_from_device(name, data) for name, data in sorted(grp_raw.items())
+            ]
+
+        sg_raw = vrrp_raw.get("sync-group") or {}
+        if sg_raw:
+            vrrp_arg["sync_groups"] = [
+                _sync_group_from_device(name, data) for name, data in sorted(sg_raw.items())
+            ]
+
+        if vrrp_arg:
+            result["vrrp"] = vrrp_arg
+
+    return result
+
+
+def build_commands(config, raw_have, state):
+    raw_have = raw_have or {}
+    config = config or {}
+
+    if state == "deleted":
+        return [("delete", _BASE)] if raw_have else []
+
+    want = _want_to_device(config)
+    norm_have = {k: v for k, v in raw_have.items() if k not in ("virtual-server", "vrrp")}
+    if raw_have.get("virtual-server"):
+        norm_have["virtual-server"] = normalize_have(raw_have, _VS_TAG_KEYS)["virtual-server"]
+    if raw_have.get("vrrp"):
+        norm_have["vrrp"] = normalize_have(raw_have, _VRRP_TAG_KEYS)["vrrp"]
+
+    commands = []
+    if state == "overridden":
+        commands += dict_op(want, norm_have, _BASE, op="purge")
+    elif state == "replaced":
+        for section, section_want in want.items():
+            section_have = norm_have.get(section, {})
+            commands += dict_op(section_want, section_have, _BASE + [section], op="purge")
+    commands += dict_op(want, norm_have, _BASE, op="set")
+
+    # snmp "disabled" has no device-side value to compare against --
+    # it's the absence of the presence node, which dict_op's set/purge
+    # logic can't express as a "delete" on its own. Handled explicitly.
+    if (config.get("vrrp") or {}).get("snmp") == "disabled":
+        if "snmp" in (raw_have.get("vrrp") or {}):
+            commands.append(("delete", _BASE + ["vrrp", "snmp"]))
+
+    return commands
+
+
+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)
+
+    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 = _device_to_argspec(get_running_config(vyos))
+        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()
diff --git a/plugins/modules/vyos_logging_global.py b/plugins/modules/vyos_logging_global.py
index f18ba3f..4a89a12 100644
--- a/plugins/modules/vyos_logging_global.py
+++ b/plugins/modules/vyos_logging_global.py
@@ -1,7 +1,6 @@
 #!/usr/bin/python
 # -*- coding: utf-8 -*-
-# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
-
+# GNU General Public License v3.0+
 from __future__ import absolute_import, division, print_function
 
 
@@ -13,8 +12,7 @@ module: vyos_logging_global
 short_description: Manage syslog configuration on VyOS devices using REST API
 description:
   - Manages syslog (logging) configuration on VyOS devices via the REST API.
-  - Supports console, file, host, user, and global logging targets with
-    per-target facility and severity configuration.
+  - Targets VyOS 1.5+ syslog schema under C(system syslog).
   - Uses REST API (C(connection=httpapi)) instead of CLI.
 version_added: "1.0.0"
 author:
@@ -40,51 +38,12 @@ options:
               severity:
                 description: Minimum severity level to log (e.g. err, debug, all).
                 type: str
-      files:
-        description: Logging to local files.
-        type: list
-        elements: dict
-        suboptions:
-          path:
-            description: Path to the log file on the device.
-            type: str
-          archive:
-            description: Log file archive/rotation settings.
-            type: dict
-            suboptions:
-              file_num:
-                description: Number of archived log files to keep.
-                type: int
-              size:
-                description: Maximum size of log file in kilobytes before rotation.
-                type: int
-          facilities:
-            description: List of syslog facilities to log to this file.
-            type: list
-            elements: dict
-            suboptions:
-              facility:
-                description: Syslog facility name.
-                type: str
-              severity:
-                description: Minimum severity level to log.
-                type: str
       global_params:
-        description: Global syslog parameters (maps to C(system syslog global)).
+        description: Global syslog parameters (maps to C(system syslog local) on device).
         type: dict
         suboptions:
-          archive:
-            description: Global log archive/rotation settings.
-            type: dict
-            suboptions:
-              file_num:
-                description: Number of archived log files to keep.
-                type: int
-              size:
-                description: Maximum size of log file in kilobytes before rotation.
-                type: int
           facilities:
-            description: List of syslog facilities for global logging.
+            description: List of syslog facilities for local logging.
             type: list
             elements: dict
             suboptions:
@@ -101,7 +60,7 @@ options:
             description: Use the fully qualified domain name in syslog messages.
             type: bool
       hosts:
-        description: Logging to remote syslog hosts.
+        description: Logging to remote syslog hosts (maps to C(system syslog remote)).
         type: list
         elements: dict
         suboptions:
@@ -148,9 +107,6 @@ options:
                 description: Minimum severity level to send.
                 type: str
 
-  running_config:
-    description: Used only with state C(parsed).
-    type: str
 
   state:
     description:
@@ -163,8 +119,6 @@ options:
       - overridden
       - deleted
       - gathered
-      - rendered
-      - parsed
 """
 
 EXAMPLES = r"""
@@ -175,30 +129,18 @@ EXAMPLES = r"""
         facilities:
           - facility: local7
             severity: err
-      files:
-        - path: logFile
-          archive:
-            file_num: 2
-          facilities:
-            - facility: local6
-              severity: emerg
       hosts:
         - hostname: 172.16.0.1
-          port: 223
+          port: 514
           facilities:
             - facility: local7
               severity: all
-            - facility: all
-              protocol: udp
       users:
         - username: vyos
           facilities:
             - facility: local7
               severity: debug
       global_params:
-        archive:
-          file_num: 2
-          size: 111
         facilities:
           - facility: cron
             severity: debug
@@ -234,223 +176,184 @@ gathered:
   type: dict
 saved:
   description: Result of save_config after applying changes.
-  returned: when changes are applied
+  returned: when changed
   type: dict
 """
 
 from ansible.module_utils.basic import AnsibleModule
-from ansible_collections.vyos.rest.plugins.module_utils.vyos import VyOSModule
-
-
-# ------------------------------------------------------------
-# Normalization
-# ------------------------------------------------------------
-
-
-def normalize_config(cfg):
-    result = {
-        "console": {"facilities": {}},
-        "global": {"facilities": {}},
-        "hosts": {},
-        "files": {},
-        "users": {},
-    }
-
-    for f in cfg.get("console", {}).get("facilities", []):
-        result["console"]["facilities"][f["facility"]] = f.get("severity")
-
-    gp = cfg.get("global_params", {})
-    for f in gp.get("facilities", []):
-        result["global"]["facilities"][f["facility"]] = f.get("severity")
-    if gp.get("archive"):
-        result["global"]["archive"] = gp["archive"]
-    if gp.get("marker_interval"):
-        result["global"]["marker_interval"] = gp["marker_interval"]
-    if gp.get("preserve_fqdn"):
-        result["global"]["preserve_fqdn"] = True
-
-    for h in cfg.get("hosts", []):
-        host = {"port": h.get("port"), "facilities": {}}
-        for f in h.get("facilities", []):
-            host["facilities"][f["facility"]] = {k: v for k, v in f.items() if k != "facility"}
-        result["hosts"][h["hostname"]] = host
-
-    for f in cfg.get("files", []):
-        facilities = {x["facility"]: x.get("severity") for x in f.get("facilities", [])}
-        result["files"][f["path"]] = {
-            "archive": f.get("archive"),
-            "facilities": facilities,
-        }
-
-    for u in cfg.get("users", []):
-        result["users"][u["username"]] = {
-            "facilities": {f["facility"]: f.get("severity") for f in u.get("facilities", [])},
-        }
-
-    return result
-
-
-def normalize_running(raw):
-    result = {
-        "console": {"facilities": {}},
-        "global": {"facilities": {}},
-        "hosts": {},
-        "files": {},
-        "users": {},
-    }
-
-    if not raw:
-        return result
-
-    for f, data in raw.get("console", {}).get("facility", {}).items():
-        result["console"]["facilities"][f] = data.get("level")
-
-    g = raw.get("local", {})
-    for f, data in g.get("facility", {}).items():
-        result["global"]["facilities"][f] = data.get("level")
-    if "archive" in g:
-        result["global"]["archive"] = g["archive"]
-    if "marker" in g and "interval" in g["marker"]:
-        result["global"]["marker_interval"] = g["marker"]["interval"]
-    if "preserve-fqdn" in g:
-        result["global"]["preserve_fqdn"] = True
-
-    for host, data in raw.get("remote", {}).items():
-        h = {"port": data.get("port"), "facilities": {}}
-        for f, fd in data.get("facility", {}).items():
-            h["facilities"][f] = {
-                "severity": fd.get("level"),
-                "protocol": fd.get("protocol"),
-            }
-        result["hosts"][host] = h
-
-    for path, data in raw.get("file", {}).items():
-        facilities = {}
-        for f, fd in data.get("facility", {}).items():
-            facilities[f] = fd.get("level")
-        result["files"][path] = {
-            "archive": data.get("archive"),
-            "facilities": facilities,
-        }
-
-    for user, data in raw.get("user", {}).items():
-        facilities = {}
-        for f, fd in data.get("facility", {}).items():
-            facilities[f] = fd.get("level")
-        result["users"][user] = {"facilities": facilities}
-
-    return result
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    dict_op,
+)
 
 
-# ------------------------------------------------------------
-# Diff helpers
-# ------------------------------------------------------------
+_BASE = ["system", "syslog"]
 
 
-def diff_facilities(base, want, have, state):
-    cmds = []
-    want_keys = set(want)
-    have_keys = set(have)
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
 
-    for f in want_keys:
-        if f not in have_keys or want[f] != have[f]:
-            path = base + ["facility", f]
-            if want[f]:
-                path += ["level", want[f]]
-            cmds.append(("set", path))
 
-    if state in ["replaced", "deleted"]:
-        for f in have_keys - want_keys:
-            cmds.append(("delete", base + ["facility", f]))
-
-    return cmds
-
-
-def diff_map(base, want, have, state):
-    cmds = []
-    w = set(want)
-    h = set(have)
-
-    if state in ["merged", "replaced"]:
-        for k in w - h:
-            cmds.append(("set", base + [k]))
-
-    if state in ["replaced", "deleted"]:
-        for k in h - w:
-            cmds.append(("delete", base + [k]))
-
-    return cmds
-
-
-# ------------------------------------------------------------
-# Build commands
-# ------------------------------------------------------------
-
-
-def build_commands(want, have, state):
-    cmds = []
-
-    if state == "overridden":
-        cmds.append(("delete", ["system", "syslog"]))
-        state = "merged"
-
-    cmds += diff_facilities(
-        ["system", "syslog", "console"],
-        want["console"]["facilities"],
-        have["console"]["facilities"],
-        state,
-    )
-
-    cmds += diff_facilities(
-        ["system", "syslog", "local"],
-        want["global"]["facilities"],
-        have["global"]["facilities"],
-        state,
-    )
-
-    cmds += diff_map(
-        ["system", "syslog", "file"],
-        want["files"],
-        have["files"],
-        state,
-    )
-
-    cmds += diff_map(
-        ["system", "syslog", "remote"],
-        want["hosts"],
-        have["hosts"],
-        state,
-    )
-
-    cmds += diff_map(
-        ["system", "syslog", "user"],
-        want["users"],
-        have["users"],
-        state,
-    )
+def _fac_list_to_device(facilities):
+    """Convert [{facility, severity, protocol}] -> {"name": {"level": s, ...}}"""
+    result = {}
+    for fac in facilities or []:
+        name = fac["facility"]
+        entry = {}
+        if fac.get("severity"):
+            entry["level"] = fac["severity"]
+        if fac.get("protocol"):
+            entry["protocol"] = fac["protocol"]
+        result[name] = entry
+    return result
 
-    return cmds
 
+def _fac_device_to_list(raw_fac):
+    """Convert {"name": {"level": s}} -> [{facility, severity}]"""
+    if not raw_fac or not isinstance(raw_fac, dict):
+        return []
+    result = []
+    for name, data in sorted(raw_fac.items()):
+        entry = {"facility": name}
+        if isinstance(data, dict):
+            if data.get("level"):
+                entry["severity"] = data["level"]
+            if data.get("protocol"):
+                entry["protocol"] = data["protocol"]
+        result.append(entry)
+    return result
 
-# ------------------------------------------------------------
-# Running config
-# ------------------------------------------------------------
 
+# ---------------------------------------------------------------------------
+# Shape adapters
+# ---------------------------------------------------------------------------
+
+
+def _want_to_device(config):
+    """Convert argspec config to device shape for dict_op.
+
+    VyOS 1.5 syslog schema:
+      system syslog console facility <f> level <s>
+      system syslog local   facility <f> level <s>   (was: global)
+      system syslog remote  <host> facility <f> ...  (was: host)
+      system syslog user    <u> facility <f> ...
+      system syslog marker  interval <n>             (was: global marker)
+      system syslog preserve-fqdn                    (was: global preserve-fqdn)
+      NOTE: file and archive are removed in VyOS 1.5
+    """
+    if not config:
+        return {}
+    want = {}
+
+    # console
+    console = config.get("console") or {}
+    if console.get("facilities"):
+        want["console"] = {"facility": _fac_list_to_device(console["facilities"])}
+
+    # global_params -> local + top-level marker/preserve-fqdn
+    gp = config.get("global_params") or {}
+    if gp:
+        if gp.get("facilities"):
+            want["local"] = {"facility": _fac_list_to_device(gp["facilities"])}
+        if gp.get("marker_interval") is not None:
+            want["marker"] = {"interval": gp["marker_interval"]}
+        if gp.get("preserve_fqdn"):
+            want["preserve-fqdn"] = {}
+
+    # hosts -> remote (keyed by hostname)
+    for h in config.get("hosts") or []:
+        hd = {}
+        if h.get("port") is not None:
+            hd["port"] = h["port"]
+        if h.get("protocol"):
+            hd["protocol"] = h["protocol"]
+        if h.get("facilities"):
+            hd["facility"] = _fac_list_to_device(h["facilities"])
+        want.setdefault("remote", {})[h["hostname"]] = hd
+
+    # users -> user (keyed by username)
+    for u in config.get("users") or []:
+        ud = {}
+        if u.get("facilities"):
+            ud["facility"] = _fac_list_to_device(u["facilities"])
+        want.setdefault("user", {})[u["username"]] = ud
+
+    return want
+
+
+def _device_to_argspec(raw):
+    """Convert raw device response to argspec shape for before/after/gathered."""
+    if not raw:
+        return {}
+    result = {}
+
+    # console
+    console = raw.get("console") or {}
+    if console:
+        facs = _fac_device_to_list(console.get("facility"))
+        if facs:
+            result["console"] = {"facilities": facs}
+
+    # local -> global_params
+    local = raw.get("local") or {}
+    marker = raw.get("marker") or {}
+    preserve_fqdn = "preserve-fqdn" in raw
+    if local or marker or preserve_fqdn:
+        gp = {}
+        facs = _fac_device_to_list(local.get("facility") if isinstance(local, dict) else {})
+        if facs:
+            gp["facilities"] = facs
+        if isinstance(marker, dict) and "interval" in marker:
+            gp["marker_interval"] = marker["interval"]
+        if preserve_fqdn:
+            gp["preserve_fqdn"] = True
+        if gp:
+            result["global_params"] = gp
+
+    # remote -> hosts
+    remote_raw = raw.get("remote") or {}
+    if remote_raw and isinstance(remote_raw, dict):
+        hosts = []
+        for hostname, data in sorted(remote_raw.items()):
+            h = {"hostname": hostname}
+            if isinstance(data, dict):
+                if data.get("port") is not None:
+                    h["port"] = data["port"]
+                if data.get("protocol"):
+                    h["protocol"] = data["protocol"]
+                facs = _fac_device_to_list(data.get("facility"))
+                if facs:
+                    h["facilities"] = facs
+            hosts.append(h)
+        if hosts:
+            result["hosts"] = hosts
+
+    # user -> users
+    user_raw = raw.get("user") or {}
+    if user_raw and isinstance(user_raw, dict):
+        users = []
+        for username, data in sorted(user_raw.items()):
+            u = {"username": username}
+            if isinstance(data, dict):
+                facs = _fac_device_to_list(data.get("facility"))
+                if facs:
+                    u["facilities"] = facs
+            users.append(u)
+        if users:
+            result["users"] = users
 
-def get_running_config(vyos):
-    raw = vyos.get_config(["system", "syslog"])
-    return normalize_running(raw)
+    return result
 
 
-# ------------------------------------------------------------
+# ---------------------------------------------------------------------------
 # Main
-# ------------------------------------------------------------
+# ---------------------------------------------------------------------------
 
 
 def main():
     argument_spec = dict(
         config=dict(type="dict"),
-        running_config=dict(type="str"),
         state=dict(
             default="merged",
             choices=[
@@ -459,8 +362,6 @@ def main():
                 "overridden",
                 "deleted",
                 "gathered",
-                "rendered",
-                "parsed",
             ],
         ),
     )
@@ -471,22 +372,34 @@ def main():
     state = module.params["state"]
     config = module.params.get("config") or {}
 
+    raw_have = vyos.get_config(_BASE)
+    have = _device_to_argspec(raw_have)
+
     if state == "gathered":
-        module.exit_json(gathered=get_running_config(vyos))
+        module.exit_json(changed=False, gathered=have)
 
-    want = normalize_config(config)
-    have = get_running_config(vyos)
+    want_device = _want_to_device(config)
 
     if state == "deleted":
-        want = {
-            "console": {"facilities": {}},
-            "global": {"facilities": {}},
-            "hosts": {},
-            "files": {},
-            "users": {},
-        }
-
-    commands = build_commands(want, have, state)
+        commands = [("delete", _BASE)] if raw_have else []
+    elif state == "overridden":
+        commands = []
+        for section in list(raw_have.keys()):
+            if section not in want_device:
+                commands.append(("delete", _BASE + [section]))
+            else:
+                commands += dict_op(
+                    want_device[section],
+                    raw_have[section],
+                    _BASE + [section],
+                    op="purge",
+                )
+        commands += dict_op(want_device, raw_have, _BASE, op="set")
+    else:
+        commands = []
+        if state == "replaced":
+            commands += dict_op(want_device, raw_have, _BASE, op="purge")
+        commands += dict_op(want_device, raw_have, _BASE, op="set")
 
     if module.check_mode:
         module.exit_json(changed=bool(commands), commands=commands, before=have)
@@ -494,10 +407,11 @@ def main():
     if commands:
         response = vyos.apply_commands(commands)
         saved = vyos.save_config()
+        after = {} if state == "deleted" else _device_to_argspec(vyos.get_config(_BASE))
         module.exit_json(
             changed=True,
             before=have,
-            after=want,
+            after=after,
             commands=commands,
             saved=saved,
             response=response,
diff --git a/plugins/modules/vyos_nat.py b/plugins/modules/vyos_nat.py
new file mode 100644
index 0000000..ae8fea0
--- /dev/null
+++ b/plugins/modules/vyos_nat.py
@@ -0,0 +1,492 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# GNU General Public License v3.0+
+from __future__ import absolute_import, division, print_function
+
+
+__metaclass__ = type
+
+DOCUMENTATION = r"""
+---
+module: vyos_nat
+short_description: Manage NAT configuration on VyOS devices using REST API
+description:
+  - Manages NAT configuration on VyOS devices via the REST API.
+  - Supports source, destination, static, CGNAT, NAT64, and NAT66.
+  - Uses REST API (C(connection=httpapi)) instead of CLI.
+  - Targets VyOS 1.5+.
+version_added: "1.0.0"
+author:
+  - Evgeny Molotkov (@omnom62)
+options:
+  config:
+    description: NAT configuration.
+    type: dict
+  state:
+    description:
+      - The desired state of the NAT configuration.
+    type: str
+    default: merged
+    choices: [merged, replaced, overridden, deleted, gathered]
+"""
+
+EXAMPLES = r"""
+- name: Merge source NAT rule
+  vyos.rest.vyos_nat:
+    config:
+      nat:
+        source:
+          rule:
+            - id: 100
+              outbound_interface:
+                name: eth0
+              translation:
+                address: masquerade
+    state: merged
+
+- name: Delete all NAT
+  vyos.rest.vyos_nat:
+    state: deleted
+
+- name: Gather NAT configuration
+  vyos.rest.vyos_nat:
+    state: gathered
+"""
+
+RETURN = r"""
+before:
+  description: NAT configuration before this module ran.
+  returned: always
+  type: dict
+after:
+  description: NAT configuration after this module ran.
+  returned: when changed
+  type: dict
+commands:
+  description: List of API commands sent to the device.
+  returned: always
+  type: list
+gathered:
+  description: Current NAT configuration as structured data.
+  returned: when state is gathered
+  type: dict
+saved:
+  description: Whether the config was saved after changes.
+  returned: when changed
+  type: bool
+"""
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    autoclean,
+    dict_op,
+    from_device,
+    normalize_have,
+    to_tag_dict,
+)
+
+
+_NAT_TYPES = ("nat", "nat64", "nat66")
+
+# "rule" (any NAT rule set) and "backend" (load-balance) are genuine tag
+# nodes everywhere they appear -- unambiguous. "range" is NOT included
+# here: confirmed against vyos-1x that it means two different things
+# depending on parent -- cgnat.pool.external.<name>.range is a tagNode
+# (has a nested "seq" leaf), but cgnat.pool.internal.<name>.range is a
+# plain multi-value leafNode (<multi/>). Handling that context-sensitive
+# case generically by key name alone would silently corrupt one or the
+# other, so it's handled explicitly in _normalize_cgnat_have() instead.
+_TAG_KEYS = {"rule", "backend"}
+
+
+# ---------------------------------------------------------------------------
+# load_balance.backend / nat64 translation.pool — the only two genuine
+# structural exceptions in this module (confirmed tagNodes with nested
+# substructure). Every other field (destination/source/translation/
+# match, inbound/outbound-interface, exclude, disable, description,
+# protocol, packet_type, load_balance.hash) is a direct structural match
+# and flows through autoclean/from_device untouched. hash in particular
+# stays a plain list -- it's a multi-value leafNode, not a tag node, so
+# dict_op's own native list handling applies to it directly.
+# ---------------------------------------------------------------------------
+
+
+def _backend_to_device(backends):
+    return {b["ip"]: autoclean({k: v for k, v in b.items() if k != "ip"}) for b in backends or []}
+
+
+def _backend_from_device(raw):
+    result = []
+    for ip, data in sorted((raw or {}).items()):
+        entry = {"ip": ip, **from_device(data or {})}
+        if "weight" in entry:
+            entry["weight"] = int(entry["weight"])
+        result.append(entry)
+    return result
+
+
+def _pool_to_device(pools):
+    return {str(p["id"]): autoclean({k: v for k, v in p.items() if k != "id"}) for p in pools or []}
+
+
+def _pool_from_device(raw):
+    return [
+        {"id": int(pid), **from_device(data or {})}
+        for pid, data in sorted((raw or {}).items(), key=lambda kv: int(kv[0]))
+    ]
+
+
+def _rule_to_device(rule):
+    entry = autoclean(
+        {k: v for k, v in rule.items() if k not in ("id", "load_balance", "translation")},
+    )
+
+    lb = rule.get("load_balance")
+    if lb:
+        lb_entry = autoclean({k: v for k, v in lb.items() if k != "backend"})
+        if lb.get("backend"):
+            lb_entry["backend"] = _backend_to_device(lb["backend"])
+        entry["load_balance"] = lb_entry
+
+    translation = rule.get("translation")
+    if translation:
+        t_entry = autoclean({k: v for k, v in translation.items() if k != "pool"})
+        if translation.get("pool"):
+            t_entry["pool"] = _pool_to_device(translation["pool"])
+        entry["translation"] = t_entry
+
+    return entry
+
+
+def _rule_from_device(raw):
+    raw = raw or {}
+    entry = from_device({k: v for k, v in raw.items() if k not in ("load-balance", "translation")})
+
+    lb_raw = raw.get("load-balance")
+    if lb_raw:
+        lb_entry = from_device({k: v for k, v in lb_raw.items() if k != "backend"})
+        # "hash" is a multi-value leafNode; the device can collapse a
+        # single value to a bare string. from_device() only does
+        # kebab->snake translation, not type coercion, so fix that up
+        # explicitly here (there's no ARGUMENT_SPEC for cast_by_spec to
+        # derive this from -- config is a bare type=dict in this module).
+        if isinstance(lb_entry.get("hash"), str):
+            lb_entry["hash"] = [lb_entry["hash"]]
+        if lb_raw.get("backend"):
+            lb_entry["backend"] = _backend_from_device(lb_raw["backend"])
+        entry["load_balance"] = lb_entry
+
+    t_raw = raw.get("translation")
+    if t_raw:
+        t_entry = from_device({k: v for k, v in t_raw.items() if k != "pool"})
+        if t_raw.get("pool"):
+            t_entry["pool"] = _pool_from_device(t_raw["pool"])
+        entry["translation"] = t_entry
+
+    return entry
+
+
+def _rules_to_device(rules):
+    return {str(r["id"]): _rule_to_device(r) for r in rules or []}
+
+
+def _rules_from_device(raw):
+    return [
+        {"id": int(rid), **_rule_from_device(data or {})}
+        for rid, data in sorted((raw or {}).items(), key=lambda kv: int(kv[0]))
+    ]
+
+
+# ---------------------------------------------------------------------------
+# CGNAT — cgnat.pool.external.<name>.range is a genuine tag node
+# (confirmed: nested "seq" leaf); cgnat.pool.internal.<name>.range is a
+# plain multi-value leafNode (confirmed <multi/>). Fixing the real bug
+# here: the previous implementation only checked isinstance(str)/
+# isinstance(dict) for internal range and silently dropped it whenever
+# the device returned the actual real shape -- a plain list.
+# ---------------------------------------------------------------------------
+
+
+def _cgnat_pool_external_to_device(pools):
+    result = {}
+    for p in pools or []:
+        entry = autoclean({k: v for k, v in p.items() if k not in ("name", "range")})
+        if p.get("range"):
+            entry["range"] = {
+                r["value"]: ({"seq": r["seq"]} if r.get("seq") is not None else {})
+                for r in p["range"]
+            }
+        result[p["name"]] = entry
+    return result
+
+
+def _cgnat_pool_external_from_device(raw):
+    result = []
+    for name, data in sorted((raw or {}).items()):
+        data = data or {}
+        p = {"name": name, **from_device({k: v for k, v in data.items() if k != "range"})}
+        rng = data.get("range")
+        if rng:
+            rng_dict = to_tag_dict(rng)
+            p["range"] = [
+                (
+                    {"value": v, "seq": int(d["seq"])}
+                    if isinstance(d, dict) and d.get("seq")
+                    else {"value": v}
+                )
+                for v, d in sorted(rng_dict.items())
+            ]
+        result.append(p)
+    return result
+
+
+def _cgnat_pool_internal_to_device(pools):
+    return {p["name"]: {"range": list(p["range"])} for p in pools or [] if p.get("range")}
+
+
+def _cgnat_pool_internal_from_device(raw):
+    result = []
+    for name, data in sorted((raw or {}).items()):
+        rng = (data or {}).get("range")
+        p = {"name": name}
+        if rng:
+            # Confirmed real bug in the previous implementation: it only
+            # checked isinstance(str)/isinstance(dict) here, silently
+            # dropping "range" entirely whenever the device returned the
+            # actual real shape for >1 value -- a plain list.
+            p["range"] = [rng] if isinstance(rng, str) else list(rng)
+        result.append(p)
+    return result
+
+
+# The two known CGNAT pool kinds and their handlers, declared once. Both
+# are unavoidable exceptions -- "external" pool range is a tag node
+# (confirmed: nested "seq" leaf), "internal" pool range is a plain
+# multi-value leaf (confirmed <multi/>), same key name, genuinely
+# different device shape, not discoverable by walking the JSON alone.
+# What's NOT necessary is repeating "if pool.get(kind)" per kind inline
+# -- one table declares the exception, both directions read it.
+_CGNAT_POOL_KINDS = {
+    "external": (_cgnat_pool_external_to_device, _cgnat_pool_external_from_device),
+    "internal": (_cgnat_pool_internal_to_device, _cgnat_pool_internal_from_device),
+}
+
+
+def _cgnat_pool_to_device(pool):
+    return {
+        kind: to_fn(pool[kind])
+        for kind, (to_fn, _from_fn) in _CGNAT_POOL_KINDS.items()
+        if pool.get(kind)
+    }
+
+
+def _cgnat_pool_from_device(pool_raw):
+    return {
+        kind: from_fn(pool_raw[kind])
+        for kind, (_to_fn, from_fn) in _CGNAT_POOL_KINDS.items()
+        if pool_raw.get(kind)
+    }
+
+
+def _normalize_cgnat_have(cgnat_raw):
+    """Like normalize_have(), but external/internal pool "range" needs
+    different treatment despite sharing a key name -- see _TAG_KEYS.
+    """
+    if not cgnat_raw or not isinstance(cgnat_raw, dict):
+        return {}
+    result = normalize_have(cgnat_raw, _TAG_KEYS)
+    ext_raw = (cgnat_raw.get("pool") or {}).get("external")
+    if ext_raw:
+        ext_norm = {}
+        for name, data in ext_raw.items():
+            data = dict(data or {})
+            if "range" in data:
+                data["range"] = to_tag_dict(data["range"])
+            ext_norm[name] = data
+        result.setdefault("pool", {})["external"] = ext_norm
+    return result
+
+
+def _cgnat_to_device(cgnat):
+    if not cgnat:
+        return {}
+    entry = autoclean({k: v for k, v in cgnat.items() if k not in ("pool", "rule")})
+    pool = cgnat.get("pool") or {}
+    pool_entry = _cgnat_pool_to_device(pool)
+    if pool_entry:
+        entry["pool"] = pool_entry
+    if cgnat.get("rule"):
+        entry["rule"] = _rules_to_device(cgnat["rule"])
+    return entry
+
+
+def _cgnat_from_device(raw):
+    raw = raw or {}
+    entry = from_device({k: v for k, v in raw.items() if k not in ("pool", "rule")})
+    pool_entry = _cgnat_pool_from_device(raw.get("pool") or {})
+    if pool_entry:
+        entry["pool"] = pool_entry
+    if raw.get("rule"):
+        entry["rule"] = _rules_from_device(raw["rule"])
+    return entry
+
+
+# ---------------------------------------------------------------------------
+# want -> device / device -> argspec (top level)
+# ---------------------------------------------------------------------------
+
+
+# Which sections are valid under each NAT type, and whether it has a
+# cgnat subtree (only "nat" does) -- declared once so _want_to_device and
+# _device_to_argspec each need a single loop instead of three near-
+# identical hand-written blocks per NAT type.
+_NAT_TYPE_SECTIONS = {
+    "nat": ("destination", "source", "static"),
+    "nat64": ("source",),
+    "nat66": ("destination", "source"),
+}
+
+
+def _want_to_device(config):
+    if not config:
+        return {}
+    result = {}
+    for nat_type, sections in _NAT_TYPE_SECTIONS.items():
+        nat = config.get(nat_type) or {}
+        if not nat:
+            continue
+        nat_dev = {}
+        if nat_type == "nat" and nat.get("cgnat"):
+            nat_dev["cgnat"] = _cgnat_to_device(nat["cgnat"])
+        for section in sections:
+            rules = (nat.get(section) or {}).get("rule")
+            if rules:
+                nat_dev[section] = {"rule": _rules_to_device(rules)}
+        if nat_dev:
+            result[nat_type] = nat_dev
+    return result
+
+
+def _device_to_argspec(raw_all):
+    if not raw_all:
+        return {}
+    result = {}
+    for nat_type, sections in _NAT_TYPE_SECTIONS.items():
+        nat = raw_all.get(nat_type) or {}
+        if not nat:
+            continue
+        nat_arg = {}
+        if nat_type == "nat" and nat.get("cgnat"):
+            nat_arg["cgnat"] = _cgnat_from_device(nat["cgnat"])
+        for section in sections:
+            rules = (nat.get(section) or {}).get("rule")
+            if rules:
+                nat_arg[section] = {"rule": _rules_from_device(rules)}
+        if nat_arg:
+            result[nat_type] = nat_arg
+    return result
+
+
+# ---------------------------------------------------------------------------
+# Main
+# ---------------------------------------------------------------------------
+
+
+def _get_raw(vyos):
+    """Retrieve all NAT config from device."""
+    result = {}
+    for nat_type in _NAT_TYPES:
+        raw = vyos.get_config([nat_type])
+        if raw:
+            result[nat_type] = raw
+    return result
+
+
+def _normalize_nat_have(raw_have, nat_type):
+    """normalize_have() for a given NAT type's have data, with "nat"'s
+    cgnat section handled by the range-context-aware
+    _normalize_cgnat_have() instead of the generic pass (which would
+    mishandle internal-pool range -- see _TAG_KEYS).
+    """
+    nat_raw = raw_have.get(nat_type, {})
+    result = normalize_have(nat_raw, _TAG_KEYS)
+    if nat_type == "nat" and nat_raw.get("cgnat"):
+        result["cgnat"] = _normalize_cgnat_have(nat_raw["cgnat"])
+    return result
+
+
+def main():
+    argument_spec = dict(
+        config=dict(type="dict"),
+        state=dict(
+            default="merged",
+            choices=["merged", "replaced", "overridden", "deleted", "gathered"],
+        ),
+    )
+
+    module = AnsibleModule(argument_spec, supports_check_mode=True)
+    vyos = VyOSModule(module)
+
+    state = module.params["state"]
+    config = module.params.get("config") or {}
+
+    raw_have = _get_raw(vyos)
+    have = _device_to_argspec(raw_have)
+
+    if state == "gathered":
+        module.exit_json(changed=False, gathered=have)
+
+    want_device = _want_to_device(config)
+
+    if state == "deleted":
+        commands = []
+        if not config:
+            for nat_type in _NAT_TYPES:
+                if raw_have.get(nat_type):
+                    commands.append(("delete", [nat_type]))
+        else:
+            for nat_type in _NAT_TYPES:
+                if config.get(nat_type) and raw_have.get(nat_type):
+                    commands.append(("delete", [nat_type]))
+    elif state == "overridden":
+        commands = []
+        for nat_type in _NAT_TYPES:
+            nat_want = want_device.get(nat_type, {})
+            nat_have_norm = _normalize_nat_have(raw_have, nat_type)
+            base = [nat_type]
+            commands += dict_op(nat_want, nat_have_norm, base, op="purge")
+            commands += dict_op(nat_want, nat_have_norm, base, op="set")
+    else:
+        commands = []
+        for nat_type in _NAT_TYPES:
+            nat_want = want_device.get(nat_type, {})
+            nat_have_norm = _normalize_nat_have(raw_have, nat_type)
+            base = [nat_type]
+            if state == "replaced":
+                for section, section_want in nat_want.items():
+                    section_have = nat_have_norm.get(section, {})
+                    commands += dict_op(section_want, section_have, base + [section], op="purge")
+            commands += dict_op(nat_want, nat_have_norm, base, op="set")
+
+    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 = _device_to_argspec(_get_raw(vyos))
+        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()
diff --git a/plugins/modules/vyos_ntp_global.py b/plugins/modules/vyos_ntp_global.py
index dd5fe4e..e308035 100644
--- a/plugins/modules/vyos_ntp_global.py
+++ b/plugins/modules/vyos_ntp_global.py
@@ -63,12 +63,6 @@ options:
               - ptp
               - interleave
 
-  running_config:
-    description:
-      - Used only with state C(parsed).
-      - Provide the output of C(show configuration commands | grep ntp).
-    type: str
-
   state:
     description:
       - The desired state of the NTP configuration.
@@ -80,8 +74,6 @@ options:
       - overridden
       - deleted
       - gathered
-      - rendered
-      - parsed
 """
 
 EXAMPLES = r"""
@@ -137,194 +129,139 @@ gathered:
   description: Current NTP configuration as structured data.
   returned: when state is gathered
   type: dict
-rendered:
-  description: CLI commands generated for the provided config (offline).
-  returned: when state is rendered
-  type: list
-parsed:
-  description: Structured data parsed from running_config.
-  returned: when state is parsed
-  type: dict
 saved:
   description: Whether the config was saved after changes.
-  returned: when changes are applied
+  returned: when changed
   type: bool
 """
 
 from ansible.module_utils.basic import AnsibleModule
-from ansible_collections.vyos.rest.plugins.module_utils.utils import normalize_to_list
-from ansible_collections.vyos.rest.plugins.module_utils.vyos import VyOSModule
-
-
-def normalize_config(config):
-    result = {
-        "allow_clients": sorted(config.get("allow_clients") or []),
-        "listen_addresses": sorted(config.get("listen_addresses") or []),
-        "servers": {},
-    }
-    for s in config.get("servers") or []:
-        name = s["server"]
-        result["servers"][name] = sorted(s.get("options") or [])
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    dict_op,
+    normalize_have,
+    to_tag_dict,
+)
+
+
+_BASE = ["service", "ntp"]
+
+# "server" is a genuine tag node (keyed by server address) that VyOS's
+# REST API can collapse to a bare value for a single server with no
+# options set.
+_TAG_KEYS = {"server"}
+
+
+def _servers_to_device(servers):
+    """server[].options is the one genuine structural exception here:
+    the argspec wraps per-server options in a named "options" list
+    field, but confirmed against vyos-1x (service_ntp.xml.in) each
+    option (noselect/nts/pool/prefer/ptp/interleave) is a direct
+    valueless leafNode sibling under the server tagNode itself -- there
+    is no "options" wrapper node on the device side at all.
+    """
+    return {s["server"]: {opt: {} for opt in (s.get("options") or [])} for s in servers or []}
+
+
+def _servers_from_device(raw):
+    result = []
+    for name, data in sorted((raw or {}).items()):
+        entry = {"server": name}
+        if data:
+            entry["options"] = sorted(to_tag_dict(data).keys())
+        result.append(entry)
     return result
 
 
-def normalize_servers(value):
-    result = {}
-    if isinstance(value, dict):
-        for server, data in value.items():
-            if isinstance(data, dict):
-                result[server] = sorted(list(data.keys()))
-            elif isinstance(data, list):
-                result[server] = sorted(data)
-            elif isinstance(data, str):
-                result[server] = [data]
-            else:
-                result[server] = []
-    elif isinstance(value, list):
-        for server in value:
-            result[server] = []
-    elif isinstance(value, str):
-        result[value] = []
-    return result
+def _want_to_device(config):
+    want = {}
+    if config.get("allow_clients"):
+        # allow_clients is a flat argspec list, but confirmed against
+        # vyos-1x (allow-client.xml.i) the device nests the multi-value
+        # leaf one level deeper, under a literal "address" child --
+        # allow-client itself is a plain grouping node, not the leaf.
+        want["allow-client"] = {"address": list(config["allow_clients"])}
+    if config.get("listen_addresses"):
+        want["listen-address"] = list(config["listen_addresses"])
+    if config.get("servers"):
+        want["server"] = _servers_to_device(config["servers"])
+    return want
 
 
 def get_running_config(vyos):
-    raw = vyos.get_config(["service", "ntp"])
-    result = {
-        "allow_clients": [],
-        "listen_addresses": [],
-        "servers": {},
-    }
-    if not raw:
-        return result
-
-    # allow-client: handle both VyOS schemas
-    #   1.4:  {"allow-client": {"address": {"10.x.x.x/y": {}}}}
-    #   1.5+: {"allow-client": {"10.x.x.x/y": {}}}  (no address subnode)
-    allow_raw_outer = raw.get("allow-client", {})
-    if "address" in allow_raw_outer:
-        allow_raw = allow_raw_outer.get("address", [])
+    return vyos.get_config(_BASE) or {}
+
+
+def _device_to_argspec(raw):
+    raw = raw or {}
+    result = {"allow_clients": [], "listen_addresses": [], "servers": []}
+
+    # allow-client: handle both VyOS schema variants (this module
+    # targets VyOS 1.4+) --
+    #   1.4:  {"allow-client": {"address": {...}}}
+    #   1.5+: {"allow-client": {...}}  (no "address" subnode observed
+    #         on some REST responses)
+    # Confirmed current vyos-1x schema always declares the "address"
+    # child, but this stays defensive for older devices/REST variants.
+    allow_outer = raw.get("allow-client") or {}
+    if isinstance(allow_outer, dict) and "address" in allow_outer:
+        allow_raw = allow_outer["address"]
     else:
-        allow_raw = allow_raw_outer
-    result["allow_clients"] = sorted(normalize_to_list(allow_raw))
+        allow_raw = allow_outer
+    if allow_raw:
+        result["allow_clients"] = sorted(to_tag_dict(allow_raw).keys())
 
-    result["listen_addresses"] = sorted(
-        normalize_to_list(raw.get("listen-address", [])),
-    )
-    result["servers"] = normalize_servers(raw.get("server", {}))
+    listen_raw = raw.get("listen-address")
+    if listen_raw:
+        result["listen_addresses"] = sorted(to_tag_dict(listen_raw).keys())
+
+    result["servers"] = _servers_from_device(raw.get("server"))
     return result
 
 
-def build_commands(desired, existing, state):
-    cmds = []
+def _normalize_allow_client(raw_have):
+    """Ensure allow-client always presents the shape _want_to_device
+    emits and dict_op compares against -- a dict with a plain LIST
+    under "address" -- regardless of which VyOS schema/REST variant the
+    device actually returned (a missing "address" wrapper, or the
+    address values themselves collapsed to a dict-of-presence or a bare
+    string instead of a plain array). This keeps dict_op only ever
+    comparing list-vs-list for this field, the same well-exercised path
+    used throughout the rest of this collection, rather than needing
+    any change to the shared engine for a dict-vs-list case.
+    """
+    allow_outer = raw_have.get("allow-client")
+    if not allow_outer:
+        return raw_have
+
+    if isinstance(allow_outer, dict) and "address" in allow_outer:
+        address_raw = allow_outer["address"]
+    else:
+        address_raw = allow_outer
+
+    raw_have = dict(raw_have)
+    raw_have["allow-client"] = {"address": sorted(to_tag_dict(address_raw).keys())}
+    return raw_have
+
+
+def build_commands(config, raw_have, state):
+    raw_have = _normalize_allow_client(raw_have or {})
+    config = config or {}
 
     if state == "overridden":
         state = "replaced"
 
     if state == "deleted":
-        if existing["servers"] or existing["allow_clients"] or existing["listen_addresses"]:
-            cmds.append(("delete", ["service", "ntp"]))
-        return cmds
-
-    cmds += diff_list(
-        "allow-client",
-        "address",
-        desired["allow_clients"],
-        existing["allow_clients"],
-        state,
-    )
-    cmds += diff_list(
-        "listen-address",
-        None,
-        desired["listen_addresses"],
-        existing["listen_addresses"],
-        state,
-    )
-    cmds += diff_servers(desired["servers"], existing["servers"], state)
-    return cmds
-
-
-def diff_list(node, subnode, desired, existing, state):
-    cmds = []
-    desired = set(desired)
-    existing = set(existing)
-
-    if state in ("merged", "replaced"):
-        for v in desired - existing:
-            path = ["service", "ntp", node]
-            if subnode:
-                path += [subnode, v]
-            else:
-                path += [v]
-            cmds.append(("set", path))
-
-    if state in ("replaced", "deleted"):
-        for v in existing - desired:
-            path = ["service", "ntp", node]
-            if subnode:
-                path += [subnode, v]
-            else:
-                path += [v]
-            cmds.append(("delete", path))
-
-    return cmds
-
-
-def diff_servers(desired, existing, state):
-    cmds = []
-    desired_set = set(desired.keys())
-    existing_set = set(existing.keys())
-
-    if state in ("merged", "replaced"):
-        for server in desired_set:
-            desired_opts = set(desired[server])
-            existing_opts = set(existing.get(server, []))
-            if server not in existing_set:
-                cmds.append(("set", ["service", "ntp", "server", server]))
-            for opt in desired_opts - existing_opts:
-                cmds.append(("set", ["service", "ntp", "server", server, opt]))
-            if state == "replaced":
-                for opt in existing_opts - desired_opts:
-                    cmds.append(("delete", ["service", "ntp", "server", server, opt]))
-
-    if state in ("replaced", "deleted"):
-        for server in existing_set - desired_set:
-            cmds.append(("delete", ["service", "ntp", "server", server]))
-
-    return cmds
-
-
-def parse_running_config(text):
-    result = {"allow_clients": [], "listen_addresses": [], "servers": {}}
-    for line in text.splitlines():
-        parts = line.strip().split()
-        if len(parts) < 4:
-            continue
-        if parts[3] == "allow-clients":
-            result["allow_clients"].append(parts[-1])
-        elif parts[3] == "listen-address":
-            result["listen_addresses"].append(parts[-1])
-        elif parts[3] == "server":
-            server = parts[4]
-            if server not in result["servers"]:
-                result["servers"][server] = []
-            if len(parts) > 5:
-                result["servers"][server].append(parts[5])
-    return result
+        return [("delete", _BASE)] if raw_have else []
 
+    want = _want_to_device(config)
+    norm_have = normalize_have(raw_have, _TAG_KEYS)
 
-def render_commands(config):
-    cmds = []
-    for c in config["allow_clients"]:
-        cmds.append("set service ntp allow-client address {c}".format(c=c))
-    for la in config["listen_addresses"]:
-        cmds.append("set service ntp listen-address {la}".format(la=la))
-    for server, opts in config["servers"].items():
-        if not opts:
-            cmds.append("set service ntp server {s}".format(s=server))
-        for opt in opts:
-            cmds.append("set service ntp server {s} {o}".format(s=server, o=opt))
-    return cmds
+    commands = []
+    if state == "replaced":
+        commands += dict_op(want, norm_have, _BASE, op="purge")
+    commands += dict_op(want, norm_have, _BASE, op="set")
+    return commands
 
 
 def main():
@@ -357,7 +294,6 @@ def main():
                 ),
             ),
         ),
-        running_config=dict(type="str"),
         state=dict(
             default="merged",
             choices=[
@@ -366,8 +302,6 @@ def main():
                 "overridden",
                 "deleted",
                 "gathered",
-                "rendered",
-                "parsed",
             ],
         ),
     )
@@ -378,40 +312,30 @@ def main():
     state = module.params["state"]
     config = module.params.get("config") or {}
 
-    if state == "parsed":
-        module.exit_json(parsed=parse_running_config(module.params["running_config"]))
-
-    desired = normalize_config(config)
-
-    if state == "rendered":
-        module.exit_json(rendered=render_commands(desired))
-
-    existing = get_running_config(vyos)
+    raw_have = get_running_config(vyos)
+    have = _device_to_argspec(raw_have)
 
     if state == "gathered":
-        module.exit_json(gathered=existing)
-
-    if state == "deleted":
-        desired = {"allow_clients": [], "listen_addresses": [], "servers": {}}
+        module.exit_json(gathered=have)
 
-    commands = build_commands(desired, existing, state)
+    commands = build_commands(config, raw_have, state)
 
     if module.check_mode:
-        module.exit_json(changed=bool(commands), commands=commands, before=existing)
+        module.exit_json(changed=bool(commands), commands=commands, before=have)
 
     if commands:
         response = vyos.apply_commands(commands)
         saved = vyos.save_config()
         module.exit_json(
             changed=True,
-            before=existing,
-            after=desired,
+            before=have,
+            after=_device_to_argspec(get_running_config(vyos)),
             commands=commands,
             saved=saved,
             response=response,
         )
 
-    module.exit_json(changed=False, before=existing, after=existing, commands=[])
+    module.exit_json(changed=False, before=have, after=have, commands=[])
 
 
 if __name__ == "__main__":
diff --git a/plugins/modules/vyos_route_maps.py b/plugins/modules/vyos_route_maps.py
index 607d1ab..347c58d 100644
--- a/plugins/modules/vyos_route_maps.py
+++ b/plugins/modules/vyos_route_maps.py
@@ -13,6 +13,12 @@ short_description: Manage route-map configuration on VyOS devices using REST API
 description:
   - Manages route maps on VyOS via the REST API.
   - Uses REST API (C(connection=httpapi)) instead of CLI.
+  - >-
+    Covers the commonly used match/set fields (as documented below). VyOS's
+    route-map schema is considerably larger than this (EVPN attributes,
+    extended communities, RPKI matching, on-match goto/next, route-source,
+    source-peer, source-vrf, and more) -- those are not modeled by this
+    module and are a real, documented limitation, not an oversight.
 version_added: "1.0.0"
 author:
   - VyOS Community (@vyos)
@@ -39,6 +45,7 @@ options:
           action:
             description: Permit or deny.
             type: str
+            choices: [permit, deny]
           description:
             description: Rule description.
             type: str
@@ -49,11 +56,166 @@ options:
             description: Continue at a different sequence number.
             type: int
           match:
-            description: Match conditions (passed through to VyOS API).
+            description: Match conditions.
             type: dict
+            suboptions:
+              interface:
+                description: Interface to match.
+                type: str
+              metric:
+                description: Metric of route to match.
+                type: int
+              origin:
+                description: BGP origin code to match.
+                type: str
+                choices: [egp, igp, incomplete]
+              peer:
+                description: Peer address to match.
+                type: str
+              protocol:
+                description: Match protocol via which the route was learnt.
+                type: str
+                choices:
+                  [
+                    babel, bgp, connected, isis, kernel, ospf, ospfv3,
+                    rip, ripng, static, table, vnc,
+                  ]
+              prefix_list:
+                description: IPv4 prefix-list to match.
+                type: str
+              prefix_list6:
+                description: IPv6 prefix-list to match.
+                type: str
+              ip:
+                description: IPv4 next-hop match parameters.
+                type: dict
+                suboptions:
+                  nexthop_address:
+                    description: IPv4 next-hop address to match.
+                    type: str
+                  nexthop_prefix_list:
+                    description: IPv4 next-hop prefix-list to match.
+                    type: str
+              ipv6:
+                description: IPv6 next-hop match parameters.
+                type: dict
+                suboptions:
+                  nexthop_address:
+                    description: IPv6 next-hop address to match.
+                    type: str
           set:
-            description: Route parameters to set (passed through to VyOS API).
+            description: Route parameters to set.
             type: dict
+            suboptions:
+              metric:
+                description: Metric of route.
+                type: int
+              metric_type:
+                description: Metric type.
+                type: str
+              origin:
+                description: BGP origin code to set.
+                type: str
+                choices: [egp, igp, incomplete]
+              originator_id:
+                description: BGP originator ID.
+                type: str
+              src:
+                description: Source address for route.
+                type: str
+              tag:
+                description: Route tag value.
+                type: int
+              weight:
+                description: BGP weight.
+                type: int
+              distance:
+                description: Locally significant administrative distance.
+                type: int
+              table:
+                description: Non-main kernel routing table.
+                type: int
+              local_preference:
+                description: BGP local preference.
+                type: int
+              ip_next_hop:
+                description: IPv4 next-hop address to set.
+                type: str
+              atomic_aggregate:
+                description: Set the BGP atomic aggregate attribute.
+                type: bool
+              as_path_exclude:
+                description: AS number(s) to remove from the as-path attribute.
+                type: str
+              as_path_prepend:
+                description: AS number(s) to prepend to the as-path attribute.
+                type: str
+              as_path_prepend_last_as:
+                description: Number of times to prepend the last AS number in the as-path.
+                type: int
+              aggregator:
+                description: BGP aggregator attribute.
+                type: dict
+                suboptions:
+                  as_:
+                    description: AS number of an aggregation.
+                    type: int
+                    aliases: [as]
+                  ip:
+                    description: IP address of an aggregation.
+                    type: str
+              community:
+                description: BGP community attribute.
+                type: dict
+                suboptions:
+                  add:
+                    description: Communities to add to a prefix.
+                    type: list
+                    elements: str
+                  replace:
+                    description: Communities to set for a prefix.
+                    type: list
+                    elements: str
+                  none:
+                    description: Completely remove the communities attribute from a prefix.
+                    type: bool
+                  delete:
+                    description: Remove communities defined in a list from a prefix.
+                    type: str
+              large_community:
+                description: BGP large community attribute.
+                type: dict
+                suboptions:
+                  add:
+                    description: Large communities to add to a prefix.
+                    type: list
+                    elements: str
+                  replace:
+                    description: Large communities to set for a prefix.
+                    type: list
+                    elements: str
+                  none:
+                    description: Completely remove the large-community attribute from a prefix.
+                    type: bool
+                  delete:
+                    description: Remove large communities defined in a list from a prefix.
+                    type: str
+              ipv6_next_hop:
+                description: IPv6 next-hop to set.
+                type: dict
+                suboptions:
+                  global:
+                    description: Nexthop IPv6 global address.
+                    type: str
+                  local:
+                    description: Nexthop IPv6 local address.
+                    type: str
+                  peer_address:
+                    description: Use the peer address (BGP only) as the nexthop.
+                    type: bool
+                  prefer_global:
+                    description: Prefer the global address as the nexthop.
+                    type: bool
 
   state:
     description:
@@ -70,7 +232,6 @@ options:
 notes:
   - Requires C(ansible_connection=httpapi) with the VyOS httpapi plugin.
   - C(ansible_network_os) must be set to C(vyos.rest.vyos).
-  - Input validation is delegated to the VyOS API.
 """
 
 EXAMPLES = r"""
@@ -84,10 +245,10 @@ EXAMPLES = r"""
             match:
               peer: 192.0.2.32
             set:
-              metric: "5"
+              metric: 5
               as_path_exclude: "111"
               aggregator:
-                as: 100
+                as_: 100
     state: merged
 
 - name: Delete all route maps
@@ -110,320 +271,526 @@ before:
   description: Route map configuration before this module ran.
   returned: always
   type: list
-
 after:
   description: Route map 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 route map 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
+  returned: when changed
   type: bool
-
 response:
   description: Raw API response.
-  returned: when changes are applied
+  returned: always
   type: dict
 """
 
 from ansible.module_utils.basic import AnsibleModule
-from ansible_collections.vyos.rest.plugins.module_utils.vyos import VyOSModule
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    autoclean,
+    cast_by_spec,
+    dict_op,
+    from_device,
+    to_tag_dict,
+)
 
 
 _BASE = ["policy", "route-map"]
 
-_SET_MAP = {
-    "as_path_prepend": ["as-path", "prepend"],
-    "as_path_exclude": ["as-path", "exclude"],
-    "as_path_prepend_last_as": ["as-path", "prepend-last-as"],
-    "ip_next_hop": ["ip-next-hop"],
-    "local_preference": ["local-preference"],
-    "metric": ["metric"],
-    "metric_type": ["metric-type"],
-    "origin": ["origin"],
-    "originator_id": ["originator-id"],
-    "src": ["src"],
-    "tag": ["tag"],
-    "weight": ["weight"],
-    "distance": ["distance"],
-    "table": ["table"],
-}
-
-_MATCH_MAP = {
-    "interface": ["interface"],
-    "metric": ["metric"],
-    "origin": ["origin"],
-    "peer": ["peer"],
-    "protocol": ["protocol"],
-}
 
+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: both named-list sections in this
+    argspec (route maps, rules) already mark exactly one suboption
+    required=True. Deriving it here means the key field is asserted to
+    exist by the argspec itself, not duplicated in a place that could
+    drift out of sync with it.
+    """
+    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
+    supplies whatever else is genuinely irreducible for a given section
+    (a nested reshape) -- defaulting to the generic recursive walker.
+    """
+    entry_transform = entry_transform or autoclean
+    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 get_running_config(vyos):
-    raw = vyos.get_config(["policy", "route-map"])
-    if not raw or not isinstance(raw, dict):
-        return []
 
-    rm_data = raw.get("route-map", raw)
-    if not isinstance(rm_data, dict):
-        return []
+def _keyed_list_from_device(raw, key_field, entry_transform=None, key_cast=None):
+    entry_transform = entry_transform or from_device
+    key_cast = key_cast or (lambda k: k)
+    return [
+        {key_field: key_cast(key), **entry_transform(data or {})}
+        for key, data in sorted(to_tag_dict(raw).items())
+    ]
 
-    result = []
-    for rm_name, rm_info in sorted(rm_data.items()):
-        entry = {"route_map": rm_name, "entries": []}
-        rm_info = rm_info or {}
-
-        for seq, rule_data in sorted(
-            (rm_info.get("rule") or {}).items(),
-            key=lambda x: int(x[0]),
-        ):
-            rule_data = rule_data or {}
-            rule = {"sequence": int(seq)}
-            if rule_data.get("action"):
-                rule["action"] = rule_data["action"]
-            if rule_data.get("description"):
-                rule["description"] = rule_data["description"]
-            if rule_data.get("call"):
-                rule["call"] = rule_data["call"]
-            if rule_data.get("continue"):
-                rule["continue_sequence"] = int(rule_data["continue"])
-            if rule_data.get("match"):
-                rule["match"] = rule_data["match"]
-            if rule_data.get("set"):
-                rule["set"] = rule_data["set"]
-            entry["entries"].append(rule)
-
-        result.append(entry)
 
-    return result
+# ---------------------------------------------------------------------------
+# match.prefix_list / match.prefix_list6 -- confirmed genuine structural
+# insertions: the argspec has these as flat fields, but the device
+# nests them two levels down (ip.address.prefix-list /
+# ipv6.address.prefix-list).
+#
+# match.ip.nexthop_address / nexthop_prefix_list, match.ipv6.
+# nexthop_address -- confirmed the device nests these ONE level
+# deeper than the argspec (ip.nexthop.address / ip.nexthop.prefix-list),
+# under a "nexthop" node the argspec doesn't have (commented out in
+# vyos-1x itself, T3304/T3976, since a plain leaf there would collide
+# with the node).
+# ---------------------------------------------------------------------------
 
 
-def _match_cmds(rbase, match):
-    cmds = []
+def _match_to_device(match):
     if not match:
-        return cmds
-
-    mbase = rbase + ["match"]
-
-    for key, path_suffix in _MATCH_MAP.items():
-        if match.get(key) is not None:
-            cmds.append(("set", mbase + path_suffix + [str(match[key])]))
+        return {}
+    exclude = {"prefix_list", "prefix_list6", "ip", "ipv6"}
+    device = autoclean({k: v for k, v in match.items() if k not in exclude})
 
     if match.get("prefix_list"):
-        cmds.append(("set", mbase + ["ip", "address", "prefix-list", match["prefix_list"]]))
+        device.setdefault("ip", {}).setdefault("address", {})["prefix-list"] = match["prefix_list"]
     if match.get("prefix_list6"):
-        cmds.append(("set", mbase + ["ipv6", "address", "prefix-list", match["prefix_list6"]]))
+        ipv6_addr = device.setdefault("ipv6", {}).setdefault("address", {})
+        ipv6_addr["prefix-list"] = match["prefix_list6"]
 
     ip = match.get("ip") or {}
     if ip.get("nexthop_address"):
-        cmds.append(("set", mbase + ["ip", "nexthop", "address", ip["nexthop_address"]]))
+        device.setdefault("ip", {}).setdefault("nexthop", {})["address"] = ip["nexthop_address"]
     if ip.get("nexthop_prefix_list"):
-        cmds.append(("set", mbase + ["ip", "nexthop", "prefix-list", ip["nexthop_prefix_list"]]))
+        ip_nh = device.setdefault("ip", {}).setdefault("nexthop", {})
+        ip_nh["prefix-list"] = ip["nexthop_prefix_list"]
 
     ipv6 = match.get("ipv6") or {}
     if ipv6.get("nexthop_address"):
-        cmds.append(("set", mbase + ["ipv6", "nexthop", "address", ipv6["nexthop_address"]]))
+        device.setdefault("ipv6", {}).setdefault("nexthop", {})["address"] = ipv6["nexthop_address"]
 
-    return cmds
+    return device
 
 
-def _set_cmds(rbase, setv):
-    cmds = []
-    if not setv:
-        return cmds
+def _match_from_device(data):
+    if not data:
+        return {}
+    ip_raw = data.get("ip") or {}
+    ipv6_raw = data.get("ipv6") or {}
+    exclude = {"ip", "ipv6"}
+    entry = from_device({k: v for k, v in data.items() if k not in exclude})
+
+    prefix_list = (ip_raw.get("address") or {}).get("prefix-list")
+    if prefix_list:
+        entry["prefix_list"] = prefix_list
+    prefix_list6 = (ipv6_raw.get("address") or {}).get("prefix-list")
+    if prefix_list6:
+        entry["prefix_list6"] = prefix_list6
+
+    ip_nexthop = ip_raw.get("nexthop") or {}
+    ip_sub = {}
+    if ip_nexthop.get("address"):
+        ip_sub["nexthop_address"] = ip_nexthop["address"]
+    if ip_nexthop.get("prefix-list"):
+        ip_sub["nexthop_prefix_list"] = ip_nexthop["prefix-list"]
+    if ip_sub:
+        entry["ip"] = ip_sub
+
+    ipv6_nexthop = ipv6_raw.get("nexthop") or {}
+    if ipv6_nexthop.get("address"):
+        entry["ipv6"] = {"nexthop_address": ipv6_nexthop["address"]}
+
+    return entry
+
+
+# ---------------------------------------------------------------------------
+# set.as_path_* -- confirmed structural collapse: three flat argspec
+# keys (as_path_exclude/prepend/prepend_last_as) collapse onto one
+# nested device node (as-path.{exclude,prepend,prepend-last-as}) with
+# different sub-key names -- no mechanical transform gets from
+# "as_path_exclude" to that shape.
+#
+# set.community / set.large_community / set.ipv6_next_hop are fully
+# generic once modeled as real nested dicts (confirmed against schema:
+# community/large-community are add/replace/none/delete nodes;
+# ipv6-next-hop is global/local/peer-address/prefer-global) -- no
+# entry-transform needed for them at all, the top-level community_to_
+# device call handles them via ordinary recursion.
+# ---------------------------------------------------------------------------
+_AS_PATH_FIELDS = {
+    "as_path_exclude": "exclude",
+    "as_path_prepend": "prepend",
+    "as_path_prepend_last_as": "prepend-last-as",
+}
 
-    sbase = rbase + ["set"]
 
-    for key, path_suffix in _SET_MAP.items():
-        if setv.get(key) is not None:
-            cmds.append(("set", sbase + path_suffix + [str(setv[key])]))
+# Both renames in this module are position-specific -- confirmed
+# against vyos-1x: "as" and "continue" are Python keywords and can't
+# be used as dict() kwargs at all, so "as_"/"continue_sequence" are
+# unavoidable argspec names, renamed to the device's real leaf names
+# "as"/"continue". Neither fits a shared flat rename map: "as_" is
+# nested inside "aggregator" specifically, and "continue_sequence" is
+# a rule-level field, not a set-level one -- each is handled directly
+# at its own point below instead.
 
-    if setv.get("atomic_aggregate"):
-        cmds.append(("set", sbase + ["atomic-aggregate"]))
 
-    comm = setv.get("community") or {}
-    if comm.get("value"):
-        cmds.append(("set", sbase + ["community", comm["value"]]))
+def _set_to_device(setv):
+    if not setv:
+        return {}
+    exclude = set(_AS_PATH_FIELDS) | {"aggregator"}
+    device = autoclean({k: v for k, v in setv.items() if k not in exclude})
+
+    as_path = {
+        device_key: setv[arg_key]
+        for arg_key, device_key in _AS_PATH_FIELDS.items()
+        if setv.get(arg_key) is not None
+    }
+    if as_path:
+        device["as-path"] = as_path
+
+    agg = setv.get("aggregator")
+    if agg:
+        agg_device = autoclean({k: v for k, v in agg.items() if k != "as_"})
+        if agg.get("as_") is not None:
+            agg_device["as"] = agg["as_"]
+        if agg_device:
+            device["aggregator"] = agg_device
+
+    return device
+
+
+def _set_from_device(data):
+    if not data:
+        return {}
+    as_path_raw = data.get("as-path") or {}
+    agg_raw = data.get("aggregator") or {}
+    exclude = {"as-path", "aggregator"}
+    entry = from_device({k: v for k, v in data.items() if k not in exclude})
+
+    for arg_key, device_key in _AS_PATH_FIELDS.items():
+        if as_path_raw.get(device_key) is not None:
+            entry[arg_key] = as_path_raw[device_key]
+    if "as_path_prepend_last_as" in entry:
+        entry["as_path_prepend_last_as"] = int(entry["as_path_prepend_last_as"])
+
+    if agg_raw:
+        agg_entry = from_device({k: v for k, v in agg_raw.items() if k != "as"})
+        if agg_raw.get("as") is not None:
+            agg_entry["as_"] = int(agg_raw["as"])
+        if agg_entry:
+            entry["aggregator"] = agg_entry
+
+    return entry
+
+
+# ---------------------------------------------------------------------------
+# Rules (keyed by sequence) and route maps (keyed by name) -- both are
+# named-list sections like any other in this collection, so they go
+# through the same _keyed_list_to_device/_keyed_list_from_device
+# mechanic as everything else, with key_field derived from ARGSPEC
+# rather than hand-declared, instead of the hand-rolled loops this had
+# before. _ROUTE_MAP_KEY/_RULE_KEY are derived after ARGUMENT_SPEC is
+# built (near the bottom of this file) since they need it to exist.
+# ---------------------------------------------------------------------------
+
+
+def _rule_entry_to_device(rest):
+    exclude = {"match", "set", "continue_sequence"}
+    device = autoclean({k: v for k, v in rest.items() if k not in exclude})
+    if rest.get("continue_sequence") is not None:
+        device["continue"] = rest["continue_sequence"]
+    if rest.get("match"):
+        m = _match_to_device(rest["match"])
+        if m:
+            device["match"] = m
+    if rest.get("set"):
+        s = _set_to_device(rest["set"])
+        if s:
+            device["set"] = s
+    return device
+
+
+def _rule_entry_from_device(data):
+    data = dict(data or {})
+    continue_raw = data.pop("continue", None)
+    match_raw = data.pop("match", None)
+    set_raw = data.pop("set", None)
+    entry = from_device(data)
+    if continue_raw is not None:
+        entry["continue_sequence"] = int(continue_raw)
+    match = _match_from_device(match_raw)
+    if match:
+        entry["match"] = match
+    setv = _set_from_device(set_raw)
+    if setv:
+        entry["set"] = setv
+    return entry
+
+
+def _route_map_entry_to_device(rest):
+    entries = rest.get("entries") or []
+    if not entries:
+        return {}
+    return {"rule": _keyed_list_to_device(entries, _RULE_KEY, _rule_entry_to_device)}
 
-    large_comm = setv.get("large_community")
-    if large_comm is not None:
-        cmds.append(("set", sbase + ["large-community", str(large_comm)]))
 
-    agg = setv.get("aggregator") or {}
-    agg_as = agg.get("as") or agg.get("as_")
-    if agg_as and agg.get("ip"):
-        cmds.append(("set", sbase + ["aggregator", "as", str(agg_as), "address", agg["ip"]]))
-    elif agg_as:
-        cmds.append(("set", sbase + ["aggregator", "as", str(agg_as)]))
+def _route_map_entry_from_device(data):
+    rule_raw = (data or {}).get("rule")
+    if not rule_raw:
+        return {"entries": []}
+    entries = _keyed_list_from_device(rule_raw, _RULE_KEY, _rule_entry_from_device, key_cast=int)
+    # _keyed_list_from_device sorts by the raw device key as a string,
+    # which orders sequence numbers wrong across a digit-count boundary
+    # (e.g. "10" < "9" lexicographically) -- re-sort numerically now
+    # that key_cast has already converted each key to a real int.
+    return {"entries": sorted(entries, key=lambda e: e[_RULE_KEY])}
 
-    nh6 = setv.get("ipv6_next_hop") or {}
-    if nh6.get("value"):
-        ip_type = nh6.get("ip_type") or "global"
-        cmds.append(("set", sbase + ["ipv6-next-hop", ip_type, nh6["value"]]))
 
-    return cmds
+def _want_to_device(config):
+    with_entries = [rm for rm in (config or []) if rm.get("entries")]
+    return _keyed_list_to_device(with_entries, _ROUTE_MAP_KEY, _route_map_entry_to_device)
 
 
-def _want_to_api_set(setv):
-    if not setv:
-        return {}
-    api = {}
-    for key, path in _SET_MAP.items():
-        if setv.get(key) is not None:
-            d = api
-            for p in path[:-1]:
-                d = d.setdefault(p, {})
-            d[path[-1]] = str(setv[key])
-    agg = setv.get("aggregator") or {}
-    agg_as = agg.get("as") or agg.get("as_")
-    if agg_as:
-        api.setdefault("aggregator", {})["as"] = str(agg_as)
-    large_comm = setv.get("large_community")
-    if large_comm is not None:
-        api["large-community"] = {str(large_comm): {}}
-    return api
-
-
-def _want_to_api_match(match):
-    if not match:
-        return {}
-    api = {}
-    for key in _MATCH_MAP:
-        if match.get(key) is not None:
-            api[key] = str(match[key])
-    if match.get("prefix_list"):
-        api.setdefault("ip", {}).setdefault("address", {})["prefix-list"] = match["prefix_list"]
-    if match.get("prefix_list6"):
-        api.setdefault("ipv6", {}).setdefault("address", {})["prefix-list"] = match["prefix_list6"]
-    return api
-
-
-def _rule_cmds(rm_name, rule, have_rule, state="merged"):
-    cmds = []
-    seq = str(rule["sequence"])
-    rbase = _BASE + [rm_name, "rule", seq]
-
-    if rule.get("action") and rule["action"] != have_rule.get("action"):
-        cmds.append(("set", rbase + ["action", rule["action"]]))
-    if rule.get("description") and rule["description"] != have_rule.get("description"):
-        cmds.append(("set", rbase + ["description", rule["description"]]))
-    if rule.get("call") and rule["call"] != have_rule.get("call"):
-        cmds.append(("set", rbase + ["call", rule["call"]]))
-    if rule.get("continue_sequence") is not None and rule["continue_sequence"] != have_rule.get(
-        "continue_sequence",
-    ):
-        cmds.append(("set", rbase + ["continue", str(rule["continue_sequence"])]))
-
-    want_match_api = _want_to_api_match(rule.get("match"))
-    have_match = have_rule.get("match") or {}
-    changed_match = {k: v for k, v in want_match_api.items() if have_match.get(k) != v}
-    if changed_match:
-        match = rule.get("match") or {}
-        changed_keys = set(changed_match.keys())
-        partial_match = {
-            k: v
-            for k, v in match.items()
-            if k in changed_keys
-            or (k == "prefix_list" and "ip" in changed_keys)
-            or (k == "prefix_list6" and "ipv6" in changed_keys)
-        }
-        if not partial_match:
-            partial_match = match
-        cmds += _match_cmds(rbase, partial_match)
-
-    want_set_api = _want_to_api_set(rule.get("set"))
-    have_set = have_rule.get("set") or {}
-    if state in ("replaced", "overridden"):
-        if want_set_api != have_set:
-            cmds += _set_cmds(rbase, rule.get("set"))
-    else:
-        have_subset = {k: have_set[k] for k in want_set_api if k in have_set}
-        if want_set_api != have_subset:
-            cmds += _set_cmds(rbase, rule.get("set"))
-
-    return cmds
-
-
-def build_commands(config, have_raw, state):
-    cmds = []
+def get_running_config(vyos):
+    raw = vyos.get_config(_BASE) or {}
+    if isinstance(raw, dict) and "route-map" in raw:
+        return raw["route-map"] or {}
+    return raw
+
+
+def _device_to_argspec(raw):
+    if not raw:
+        return []
+    return _keyed_list_from_device(raw, _ROUTE_MAP_KEY, _route_map_entry_from_device)
+
+
+def _seed_route_map_placeholders(want, have):
+    """dict_op's fallback guesses a translated device key whenever a
+    want key is missing from have entirely (a brand-new route map or
+    rule). That guess is correct for a schema field name but wrong for
+    a route-map name, which is an opaque value that may legitimately
+    contain an underscore (confirmed against vyos-1x: "Name of
+    route-map can only contain alpha-numeric letters, hyphen and
+    underscores") -- confirmed as a real bug via direct reproduction,
+    the same class found in vyos_snmp_server's "admin_user" case:
+    "my_route_map" was silently becoming "my-route-map" in the
+    generated command on first creation.
+
+    Seeds an empty placeholder into have (mutated in place) for every
+    route-map name present in want but not yet in have, using the
+    exact verbatim value -- dict_op's own unmodified exact-match lookup
+    then finds it directly and never reaches its guessing fallback.
+    Also seeds each rule's own tag-node level with None (not {}), since
+    a rule with no other fields set is a presence-only entry -- seeding
+    {} there would make dict_op think it already matches and skip
+    emitting the needed set command (the same mistake caught and fixed
+    once already this session).
+    """
+    for rm_name, rm_val in (want or {}).items():
+        rm_have = have.setdefault(rm_name, {})
+        if not isinstance(rm_have, dict):
+            continue
+        rule_want = (rm_val or {}).get("rule") or {}
+        if rule_want:
+            rule_have = rm_have.setdefault("rule", {})
+            if isinstance(rule_have, dict):
+                for seq in rule_want:
+                    if seq not in rule_have:
+                        rule_have[seq] = None
+
+
+def build_commands(config, raw_have, state):
+    raw_have = raw_have or {}
+    config = config or []
 
     if state == "deleted":
         if not config:
-            if have_raw:
-                cmds.append(("delete", _BASE))
-        else:
-            for rm in config:
+            return [("delete", _BASE)] if raw_have else []
+        cmds = []
+        for rm in config:
+            if rm.get("route_map") in raw_have:
                 cmds.append(("delete", _BASE + [rm["route_map"]]))
         return cmds
 
-    have_map = {e["route_map"]: e for e in have_raw}
+    want = _want_to_device(config)
+    norm_have = _want_to_device(_device_to_argspec(raw_have))
+    _seed_route_map_placeholders(want, norm_have)
 
+    commands = []
     if state == "overridden":
-        want_names = {rm["route_map"] for rm in config}
-        for name in set(have_map) - want_names:
-            cmds.append(("delete", _BASE + [name]))
-
-    for rm in config:
-        rm_name = rm["route_map"]
-        have_rm = have_map.get(rm_name, {})
-
-        if state == "replaced" and rm_name in have_map:
-            # Only delete and rebuild if something actually differs
-            have_entries = {str(r["sequence"]): r for r in (have_rm.get("entries") or [])}
-            want_seqs = {str(r["sequence"]) for r in (rm.get("entries") or [])}
-            extra_seqs = set(have_entries) - want_seqs
-            test_cmds = []
-            for rule in rm.get("entries") or []:
-                have_rule = have_entries.get(str(rule["sequence"]), {})
-                test_cmds += _rule_cmds(rm_name, rule, have_rule, state)
-            if test_cmds or extra_seqs:
-                cmds.append(("delete", _BASE + [rm_name]))
-                have_rm = {}
-            else:
-                continue  # already matches — idempotent
-
-        have_entries = {str(r["sequence"]): r for r in (have_rm.get("entries") or [])}
-
-        for rule in rm.get("entries") or []:
-            have_rule = have_entries.get(str(rule["sequence"]), {})
-            cmds += _rule_cmds(rm_name, rule, have_rule, state)
-
-    return cmds
+        commands += dict_op(want, norm_have, _BASE, op="purge")
+    elif state == "replaced":
+        want_names = {rm.get("route_map") for rm in config if rm.get("route_map")}
+        for name in want_names:
+            section_want = want.get(name, {})
+            section_have = norm_have.get(name, {})
+            commands += dict_op(section_want, section_have, _BASE + [name], op="purge")
+    commands += dict_op(want, norm_have, _BASE, op="set")
+    return commands
 
 
 ARGUMENT_SPEC = dict(
-    config=dict(type="list", elements="dict"),
+    config=dict(
+        type="list",
+        elements="dict",
+        options=dict(
+            route_map=dict(type="str", required=True),
+            entries=dict(
+                type="list",
+                elements="dict",
+                options=dict(
+                    sequence=dict(type="int", required=True),
+                    action=dict(type="str", choices=["permit", "deny"]),
+                    description=dict(type="str"),
+                    call=dict(type="str"),
+                    continue_sequence=dict(type="int"),
+                    match=dict(
+                        type="dict",
+                        options=dict(
+                            interface=dict(type="str"),
+                            metric=dict(type="int"),
+                            origin=dict(type="str", choices=["egp", "igp", "incomplete"]),
+                            peer=dict(type="str"),
+                            protocol=dict(
+                                type="str",
+                                choices=[
+                                    "babel",
+                                    "bgp",
+                                    "connected",
+                                    "isis",
+                                    "kernel",
+                                    "ospf",
+                                    "ospfv3",
+                                    "rip",
+                                    "ripng",
+                                    "static",
+                                    "table",
+                                    "vnc",
+                                ],
+                            ),
+                            prefix_list=dict(type="str"),
+                            prefix_list6=dict(type="str"),
+                            ip=dict(
+                                type="dict",
+                                options=dict(
+                                    nexthop_address=dict(type="str"),
+                                    nexthop_prefix_list=dict(type="str"),
+                                ),
+                            ),
+                            ipv6=dict(
+                                type="dict",
+                                options=dict(
+                                    nexthop_address=dict(type="str"),
+                                ),
+                            ),
+                        ),
+                    ),
+                    set=dict(
+                        type="dict",
+                        options=dict(
+                            metric=dict(type="int"),
+                            metric_type=dict(type="str"),
+                            origin=dict(type="str", choices=["egp", "igp", "incomplete"]),
+                            originator_id=dict(type="str"),
+                            src=dict(type="str"),
+                            tag=dict(type="int"),
+                            weight=dict(type="int"),
+                            distance=dict(type="int"),
+                            table=dict(type="int"),
+                            local_preference=dict(type="int"),
+                            ip_next_hop=dict(type="str"),
+                            atomic_aggregate=dict(type="bool"),
+                            as_path_exclude=dict(type="str"),
+                            as_path_prepend=dict(type="str"),
+                            as_path_prepend_last_as=dict(type="int"),
+                            aggregator=dict(
+                                type="dict",
+                                options=dict(
+                                    as_=dict(type="int", aliases=["as"]),
+                                    ip=dict(type="str"),
+                                ),
+                            ),
+                            community=dict(
+                                type="dict",
+                                options=dict(
+                                    add=dict(type="list", elements="str"),
+                                    replace=dict(type="list", elements="str"),
+                                    none=dict(type="bool"),
+                                    delete=dict(type="str"),
+                                ),
+                            ),
+                            large_community=dict(
+                                type="dict",
+                                options=dict(
+                                    add=dict(type="list", elements="str"),
+                                    replace=dict(type="list", elements="str"),
+                                    none=dict(type="bool"),
+                                    delete=dict(type="str"),
+                                ),
+                            ),
+                            ipv6_next_hop=dict(
+                                type="dict",
+                                options={
+                                    "global": dict(type="str"),
+                                    "local": dict(type="str"),
+                                    "peer_address": dict(type="bool"),
+                                    "prefer_global": dict(type="bool"),
+                                },
+                            ),
+                        ),
+                    ),
+                ),
+            ),
+        ),
+    ),
     state=dict(
         default="merged",
         choices=["merged", "replaced", "overridden", "deleted", "gathered"],
     ),
 )
 
+_ENTRY_OPTIONS = ARGUMENT_SPEC["config"]["options"]["entries"]["options"]
+_ROUTE_MAP_KEY = _derive_key_field(ARGUMENT_SPEC["config"]["options"])
+_RULE_KEY = _derive_key_field(_ENTRY_OPTIONS)
+
 
 def main():
     module = AnsibleModule(ARGUMENT_SPEC, supports_check_mode=True)
-
     vyos = VyOSModule(module)
 
     state = module.params["state"]
     config = module.params.get("config") or []
 
-    have = get_running_config(vyos)
+    raw_have = get_running_config(vyos)
+    have = _device_to_argspec(raw_have)
+    for rm in have:
+        for entry in rm.get("entries") or []:
+            cast_by_spec(entry, _ENTRY_OPTIONS)
 
     if state == "gathered":
         module.exit_json(changed=False, gathered=have)
 
-    commands = build_commands(config, have, state)
+    commands = build_commands(config, raw_have, state)
 
     if module.check_mode:
         module.exit_json(changed=bool(commands), commands=commands, before=have)
@@ -431,10 +798,15 @@ def main():
     if commands:
         response = vyos.apply_commands(commands)
         saved = vyos.save_config()
+        after_raw = get_running_config(vyos)
+        after = _device_to_argspec(after_raw)
+        for rm in after:
+            for entry in rm.get("entries") or []:
+                cast_by_spec(entry, _ENTRY_OPTIONS)
         module.exit_json(
             changed=True,
             before=have,
-            after=get_running_config(vyos),
+            after=after,
             commands=commands,
             saved=saved,
             response=response,
diff --git a/plugins/modules/vyos_snmp_server.py b/plugins/modules/vyos_snmp_server.py
index 1079402..db0a285 100644
--- a/plugins/modules/vyos_snmp_server.py
+++ b/plugins/modules/vyos_snmp_server.py
@@ -54,7 +54,9 @@ options:
         description: System location.
         type: str
       smux_peer:
-        description: Register a subtree for SMUX-based processing.
+        description: >-
+          Register a subtree for SMUX-based processing. The device supports
+          multiple values here; this module manages a single value only.
         type: str
       trap_source:
         description: SNMP trap source address.
@@ -72,12 +74,15 @@ options:
             description: UDP port (default 161).
             type: int
       trap_target:
-        description: SNMP trap target.
+        description: >-
+          SNMP (v2) trap target. The device supports multiple trap targets;
+          this module manages a single one only.
         type: dict
         suboptions:
           address:
             description: IP address of the trap target host.
             type: str
+            required: true
           community:
             description: Community name to use for traps.
             type: str
@@ -128,7 +133,9 @@ options:
                     description: Authentication algorithm.
                     type: str
                   encrypted_key:
-                    description: Encrypted authentication key (stored as encrypted-password on device).
+                    description: >-
+                      Encrypted authentication key (stored as encrypted-password
+                      on device).
                     type: str
                   plaintext_key:
                     description: Plaintext authentication key (device encrypts it).
@@ -164,6 +171,7 @@ options:
               address:
                 description: IP address of the SNMPv3 trap target.
                 type: str
+                required: true
               port:
                 description: UDP port on the trap target host.
                 type: int
@@ -202,7 +210,10 @@ options:
                     description: Plaintext privacy key.
                     type: str
           views:
-            description: SNMPv3 view configuration.
+            description: >-
+              SNMPv3 view configuration. The device supports multiple OIDs
+              (each with its own exclude/mask) per view; this module manages
+              a single OID entry per view only.
             type: list
             elements: dict
             suboptions:
@@ -278,7 +289,7 @@ after:
   returned: when changed
   type: dict
 commands:
-  description: List of API command dicts sent to the device.
+  description: List of API command tuples sent to the device.
   returned: always
   type: list
 gathered:
@@ -287,463 +298,418 @@ gathered:
   type: dict
 saved:
   description: Whether the config was saved after changes.
-  returned: when changes are applied
+  returned: when changed
   type: bool
 """
 
 from ansible.module_utils.basic import AnsibleModule
-from ansible_collections.vyos.rest.plugins.module_utils.vyos import VyOSModule
-
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    autoclean,
+    cast_by_spec,
+    dict_op,
+    from_device,
+    to_tag_dict,
+)
 
-SNMP_BASE = ["service", "snmp"]
 
-SCALAR_FIELDS = {
-    "contact": "contact",
-    "description": "description",
-    "location": "location",
-    "smux_peer": "smux-peer",
-    "trap_source": "trap-source",
+_BASE = ["service", "snmp"]
+
+# ---------------------------------------------------------------------------
+# The one thing a purely structural walk of ARGUMENT_SPEC can never
+# infer: a handful of field names that mean something different on the
+# device than in the argspec, and aren't a mechanical kebab<->snake
+# conversion dict_op could handle itself. Declared once, here, as a
+# flat value map -- not embedded in ARGUMENT_SPEC (keeping that 100%
+# standard Ansible), and not scattered across per-section transform
+# functions. Confirmed against vyos-1x for every entry:
+#   - authorization_type/clients/networks: communities' fields don't
+#     match their device leaf names at all ("authorization"/"client"/
+#     "network").
+#   - authentication/encrypted_key/plaintext_key: shared by v3 users
+#     and v3 trap-targets (both nest under a device "auth" node with
+#     "encrypted-password"/"plaintext-password" leaves). This is also
+#     a real bug fix -- the previous implementation used "plaintext-
+#     key", which does not exist on the device at all.
+#   - engine_id: "engineid" on the device is one word, so there's no
+#     hyphen for the mechanical conversion to split on.
+# None of these names are reused elsewhere in this argspec with a
+# different intended device mapping (confirmed by inspection), so one
+# flat map is safe here -- a module with a genuine name collision
+# across nesting levels (this one doesn't have one) would need the map
+# scoped by path instead.
+_DEVICE_RENAMES = {
+    "communities": "community",
+    "listen_addresses": "listen-address",
+    "snmp_v3": "v3",
+    "authorization_type": "authorization",
+    "clients": "client",
+    "networks": "network",
+    "authentication": "auth",
+    "encrypted_key": "encrypted-password",
+    "plaintext_key": "plaintext-password",
+    "engine_id": "engineid",
+    "groups": "group",
+    "users": "user",
+    "views": "view",
+    "trap_targets": "trap-target",
 }
 
 
-def to_list(value):
-    if value is None:
-        return []
-    if isinstance(value, list):
-        return value
-    if isinstance(value, str):
-        return [value]
-    if isinstance(value, dict):
-        return list(value.keys())
-    return [str(value)]
-
-
-def _cmd(op, path):
-    return {"op": op, "path": path}
-
-
-def _set(path):
-    return _cmd("set", path)
-
-
-def _delete(path):
-    return _cmd("delete", path)
+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 such section in this argspec
+    already marks exactly one suboption required=True (you can't
+    create a community without a name, a user without a username).
+    Deriving it here means the key field is asserted to exist by the
+    argspec itself, not duplicated in a place that could drift out of
+    sync with it.
+    """
+    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 _parse_communities(raw):
-    if not raw or not isinstance(raw, dict):
-        return []
-    result = []
-    for name, data in sorted(raw.items()):
-        entry = {"name": name}
-        if not isinstance(data, dict):
-            result.append(entry)
+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
+    supplies whatever else is genuinely irreducible for a given section
+    (a nested reshape) -- defaulting to the generic recursive walker.
+    """
+    entry_transform = entry_transform or autoclean
+    result = {}
+    for item in items or []:
+        if not item.get(key_field):
             continue
-        if "authorization" in data:
-            entry["authorization_type"] = data["authorization"]
-        if "client" in data:
-            entry["clients"] = sorted(to_list(data["client"]))
-        if "network" in data:
-            entry["networks"] = sorted(to_list(data["network"]))
-        result.append(entry)
+        rest = {k: v for k, v in item.items() if k != key_field}
+        result[item[key_field]] = entry_transform(rest)
     return result
 
 
-def _parse_listen_addresses(raw):
-    if not raw or not isinstance(raw, dict):
-        return []
-    result = []
-    for addr, data in sorted(raw.items()):
-        entry = {"address": addr}
-        if isinstance(data, dict) and "port" in data:
-            entry["port"] = int(data["port"])
-        result.append(entry)
-    return result
+def _keyed_list_from_device(raw, key_field, entry_transform=None):
+    entry_transform = entry_transform or from_device
+    return [
+        {key_field: key, **entry_transform(data or {})}
+        for key, data in sorted(to_tag_dict(raw).items())
+    ]
 
 
-def _parse_trap_target(raw):
-    if not raw:
-        return None
-    if isinstance(raw, str):
-        return {"address": raw}
+def _single_to_device(obj, key_field):
+    """trap_target (v2): confirmed a genuine tagNode keyed by address,
+    but the argspec models only a single object (documented
+    limitation, preserved as-is: the device supports multiple trap
+    targets, this module manages one). Reuses the same keyed-list
+    mechanic above as "a list capped to one entry" rather than a
+    bespoke pair of functions.
+    """
+    if not obj or not obj.get(key_field):
+        return {}
+    return _keyed_list_to_device([obj], key_field)
+
+
+def _single_from_device(raw, key_field):
+    entries = _keyed_list_from_device(raw, key_field)
+    return entries[0] if entries else None
+
+
+# ---------------------------------------------------------------------------
+# v3 views — the confirmed structural bug fix. "oid" is a genuine tag
+# node (keyed by the OID value itself) with its own "exclude"/"mask"
+# children -- the previous implementation treated "oid" as a flat leaf
+# and read exclude/mask from the wrong nesting level entirely (directly
+# under the view, when they actually live under view.oid.<value>). This
+# is a genuine arity change (three sibling scalar fields collapse into
+# one nested tag node), not a rename -- it can't be expressed through
+# _DEVICE_RENAMES, so it's the one section needing a real override
+# instead of the generic recursive walker. The device also supports
+# multiple OIDs per view and multiple excludes per OID (both <multi/> /
+# tagNode); the argspec only models one of each -- a documented
+# limitation, preserved as-is, not expanded here.
+# ---------------------------------------------------------------------------
+
+
+def _view_entry_to_device(rest):
+    entry = autoclean({k: v for k, v in rest.items() if k not in ("oid", "exclude", "mask")})
+    if rest.get("oid"):
+        oid_entry = {}
+        if rest.get("exclude"):
+            oid_entry["exclude"] = [rest["exclude"]]
+        if rest.get("mask"):
+            oid_entry["mask"] = rest["mask"]
+        entry["oid"] = {rest["oid"]: oid_entry}
+    return entry
+
+
+def _view_entry_from_device(data):
     entry = {}
-    if "address" in raw:
-        entry["address"] = raw["address"]
-    if "community" in raw:
-        entry["community"] = raw["community"]
-    if "port" in raw:
-        entry["port"] = int(raw["port"])
-    return entry if entry else None
-
-
-def _parse_v3_auth_privacy(raw, key):
-    block = raw.get(key) if isinstance(raw, dict) else None
-    if not block:
-        return None
-    result = {}
-    if "type" in block:
-        result["type"] = block["type"]
-    if "encrypted-password" in block:
-        result["encrypted_key"] = block["encrypted-password"]
-    if "plaintext-key" in block:
-        result["plaintext_key"] = block["plaintext-key"]
-    return result if result else None
-
-
-def _parse_v3_users(raw):
-    if not raw or not isinstance(raw, dict):
-        return []
-    result = []
-    for username, data in sorted(raw.items()):
-        entry = {"user": username}
-        auth = _parse_v3_auth_privacy(data, "auth")
-        if auth:
-            entry["authentication"] = auth
-        priv = _parse_v3_auth_privacy(data, "privacy")
-        if priv:
-            entry["privacy"] = priv
-        if isinstance(data, dict):
-            if "group" in data:
-                entry["group"] = data["group"]
-            if "mode" in data:
-                entry["mode"] = data["mode"]
-            if "tsm-key" in data:
-                entry["tsm_key"] = data["tsm-key"]
-        result.append(entry)
-    return result
+    oid_raw = (data or {}).get("oid")
+    if oid_raw:
+        oid_dict = to_tag_dict(oid_raw)
+        oid_value, oid_data = sorted(oid_dict.items())[0]
+        entry["oid"] = oid_value
+        oid_data = oid_data or {}
+        excl_raw = oid_data.get("exclude")
+        if excl_raw:
+            excl_list = (
+                [excl_raw] if isinstance(excl_raw, str) else sorted(to_tag_dict(excl_raw).keys())
+            )
+            entry["exclude"] = excl_list[0]
+        if oid_data.get("mask"):
+            entry["mask"] = oid_data["mask"]
+    return entry
+
+
+# Sections needing something other than the generic recursive walker,
+# keyed by the argspec field name -- a second small value map, kept
+# separate from _DEVICE_RENAMES because it answers a different
+# question (how to build/parse each entry, not what to call a field).
+# Every other named-list section in this module (communities,
+# listen_addresses, v3 groups/users/trap_targets) needs neither: their
+# member fields either match the device 1:1 or are covered by
+# _DEVICE_RENAMES, so the generic walker handles them with no entry
+# here at all.
+_ENTRY_OVERRIDES = {
+    "views": (_view_entry_to_device, _view_entry_from_device),
+}
 
 
-def _parse_v3_groups(raw):
-    if not raw or not isinstance(raw, dict):
-        return []
-    result = []
-    for name, data in sorted(raw.items()):
-        entry = {"group": name}
-        if isinstance(data, dict):
-            for key in ("mode", "seclevel", "view"):
-                if key in data:
-                    entry[key] = data[key]
-        result.append(entry)
-    return result
+# ---------------------------------------------------------------------------
+# The generic recursive walker. Driven entirely by ARGUMENT_SPEC's own
+# structure (type=dict -> recurse; type=list with options -> a named
+# list, keyed by _derive_key_field; type=list with no options -> a
+# plain multi-value leaf, left to dict_op's own list handling) plus the
+# two small value maps above for the handful of cases structure alone
+# can't resolve. This is what replaced a hand-written to-device/from-
+# device function pair for every single section in this module.
+# ---------------------------------------------------------------------------
 
 
-def _parse_v3_views(raw):
-    if not raw or not isinstance(raw, dict):
-        return []
-    result = []
-    for name, data in sorted(raw.items()):
-        entry = {"view": name}
-        if isinstance(data, dict) and "oid" in data:
-            oid_data = data["oid"]
-            if isinstance(oid_data, dict) and oid_data:
-                entry["oid"] = str(list(oid_data.keys())[0])
-            elif isinstance(oid_data, str):
-                entry["oid"] = oid_data
-        if isinstance(data, dict):
-            if "exclude" in data:
-                entry["exclude"] = data["exclude"]
-            if "mask" in data:
-                entry["mask"] = data["mask"]
-        result.append(entry)
-    return result
-
-
-def _parse_v3_trap_targets(raw):
-    if not raw or not isinstance(raw, dict):
-        return []
-    result = []
-    for addr, data in sorted(raw.items()):
-        entry = {"address": addr}
-        if isinstance(data, dict):
-            if "port" in data:
-                entry["port"] = int(data["port"])
-            if "protocol" in data:
-                entry["protocol"] = data["protocol"]
-            if "type" in data:
-                entry["type"] = data["type"]
-            auth = _parse_v3_auth_privacy(data, "auth")
-            if auth:
-                entry["authentication"] = auth
-            priv = _parse_v3_auth_privacy(data, "privacy")
-            if priv:
-                entry["privacy"] = priv
-        result.append(entry)
+def _spec_to_device(value, options_spec):
+    if not isinstance(value, dict):
+        return value
+    result = {}
+    for arg_key, sub_spec in options_spec.items():
+        val = value.get(arg_key)
+        if val is None or val is False:
+            continue
+        device_key = _DEVICE_RENAMES.get(arg_key, arg_key)
+        sub_type = sub_spec.get("type")
+        sub_options = sub_spec.get("options")
+        if sub_type == "dict" and sub_options:
+            converted = _spec_to_device(val, sub_options)
+            if converted:
+                result[device_key] = converted
+        elif sub_type == "list" and sub_options:
+            key_field = _derive_key_field(sub_options)
+            entry_to, _entry_from = _ENTRY_OVERRIDES.get(arg_key, (None, None))
+            entry_transform = entry_to or (
+                lambda rest, spec=sub_options: _spec_to_device(rest, spec)
+            )
+            result[device_key] = _keyed_list_to_device(val, key_field, entry_transform)
+        elif val is True:
+            result[device_key] = {}
+        elif sub_type == "list":
+            result[device_key] = list(val)
+        else:
+            result[device_key] = val
     return result
 
 
-def parse_snmp_config(raw):
+def _device_to_spec(raw, options_spec):
     if not raw or not isinstance(raw, dict):
         return {}
+    have_idx = {k.replace("-", "_"): k for k in raw}
     result = {}
-    for argspec_key, api_key in SCALAR_FIELDS.items():
-        if api_key in raw:
-            result[argspec_key] = raw[api_key]
-    communities = _parse_communities(raw.get("community"))
-    if communities:
-        result["communities"] = communities
-    listen = _parse_listen_addresses(raw.get("listen-address"))
-    if listen:
-        result["listen_addresses"] = listen
-    trap = _parse_trap_target(raw.get("trap-target"))
-    if trap:
-        result["trap_target"] = trap
-    v3_raw = raw.get("v3")
-    if v3_raw and isinstance(v3_raw, dict):
-        v3 = {}
-        if "engineid" in v3_raw:
-            v3["engine_id"] = v3_raw["engineid"]
-        groups = _parse_v3_groups(v3_raw.get("group"))
-        if groups:
-            v3["groups"] = groups
-        users = _parse_v3_users(v3_raw.get("user"))
-        if users:
-            v3["users"] = users
-        views = _parse_v3_views(v3_raw.get("view"))
-        if views:
-            v3["views"] = views
-        trap_targets = _parse_v3_trap_targets(v3_raw.get("trap-target"))
-        if trap_targets:
-            v3["trap_targets"] = trap_targets
-        if v3:
-            result["snmp_v3"] = v3
+    for arg_key, sub_spec in options_spec.items():
+        device_key = _DEVICE_RENAMES.get(arg_key, arg_key)
+        orig_key = device_key if device_key in raw else have_idx.get(arg_key)
+        if orig_key is None:
+            continue
+        raw_val = raw[orig_key]
+        sub_type = sub_spec.get("type")
+        sub_options = sub_spec.get("options")
+        if sub_type == "dict" and sub_options:
+            converted = _device_to_spec(raw_val, sub_options)
+            if converted:
+                result[arg_key] = converted
+        elif sub_type == "list" and sub_options:
+            key_field = _derive_key_field(sub_options)
+            _entry_to, entry_from = _ENTRY_OVERRIDES.get(arg_key, (None, None))
+            entry_transform = entry_from or (lambda d, spec=sub_options: _device_to_spec(d, spec))
+            entries = _keyed_list_from_device(raw_val, key_field, entry_transform)
+            if entries:
+                result[arg_key] = entries
+        elif sub_type == "list":
+            if raw_val:
+                result[arg_key] = sorted(to_tag_dict(raw_val).keys())
+        elif isinstance(raw_val, dict) and not raw_val:
+            result[arg_key] = True
+        else:
+            result[arg_key] = raw_val
     return result
 
 
+def _want_to_device(config):
+    config = config or {}
+    want = _spec_to_device(
+        {k: v for k, v in config.items() if k != "trap_target"},
+        _TOP_OPTIONS,
+    )
+    if config.get("trap_target"):
+        tt = _single_to_device(config["trap_target"], _derive_key_field(_TRAP_TARGET_OPTIONS))
+        if tt:
+            want["trap-target"] = tt
+    return want
+
+
 def get_running_config(vyos):
     try:
-        raw = vyos.get_config(SNMP_BASE)
+        return vyos.get_config(_BASE) or {}
     except Exception as e:
         if "Configuration under specified path is empty" in str(e):
             return {}
         raise
-    return parse_snmp_config(raw)
-
-
-def _build_scalar_commands(want, have, state):
-    cmds = []
-    for argspec_key, api_key in SCALAR_FIELDS.items():
-        want_val = want.get(argspec_key)
-        have_val = have.get(argspec_key)
-        path = SNMP_BASE + [api_key]
-        if state in ("replaced", "overridden"):
-            if have_val and want_val != have_val:
-                cmds.append(_delete(path))
-        if state in ("merged", "replaced", "overridden"):
-            if want_val and want_val != have_val:
-                cmds.append(_set(path + [want_val]))
-    return cmds
-
-
-def _build_community_commands(want_list, have_list, state):
-    cmds = []
-    want_map = {c["name"]: c for c in (want_list or [])}
-    have_map = {c["name"]: c for c in (have_list or [])}
-    if state in ("replaced", "overridden"):
-        for name in have_map:
-            if name not in want_map:
-                cmds.append(_delete(SNMP_BASE + ["community", name]))
-    for name, want_comm in want_map.items():
-        have_comm = have_map.get(name, {})
-        base = SNMP_BASE + ["community", name]
-        want_auth = want_comm.get("authorization_type")
-        have_auth = have_comm.get("authorization_type")
-        if state in ("replaced", "overridden") and have_auth and want_auth != have_auth:
-            cmds.append(_delete(base + ["authorization"]))
-        if want_auth and want_auth != have_auth:
-            cmds.append(_set(base + ["authorization", want_auth]))
-        want_clients = set(want_comm.get("clients") or [])
-        have_clients = set(have_comm.get("clients") or [])
-        for c in want_clients - have_clients:
-            cmds.append(_set(base + ["client", c]))
-        if state in ("replaced", "overridden"):
-            for c in have_clients - want_clients:
-                cmds.append(_delete(base + ["client", c]))
-        want_nets = set(want_comm.get("networks") or [])
-        have_nets = set(have_comm.get("networks") or [])
-        for n in want_nets - have_nets:
-            cmds.append(_set(base + ["network", n]))
-        if state in ("replaced", "overridden"):
-            for n in have_nets - want_nets:
-                cmds.append(_delete(base + ["network", n]))
-    return cmds
-
-
-def _build_listen_address_commands(want_list, have_list, state):
-    cmds = []
-    want_map = {e["address"]: e for e in (want_list or [])}
-    have_map = {e["address"]: e for e in (have_list or [])}
-    base = SNMP_BASE + ["listen-address"]
-    if state in ("replaced", "overridden"):
-        for addr in have_map:
-            if addr not in want_map:
-                cmds.append(_delete(base + [addr]))
-    for addr, want_entry in want_map.items():
-        have_entry = have_map.get(addr, {})
-        want_port = want_entry.get("port")
-        have_port = have_entry.get("port")
-        if addr not in have_map:
-            if want_port:
-                cmds.append(_set(base + [addr, "port", str(want_port)]))
-            else:
-                cmds.append(_set(base + [addr]))
-        elif want_port != have_port:
-            cmds.append(_delete(base + [addr]))
-            if want_port:
-                cmds.append(_set(base + [addr, "port", str(want_port)]))
-            else:
-                cmds.append(_set(base + [addr]))
-    return cmds
-
-
-def _build_trap_target_commands(want, have, state):
-    cmds = []
-    base = SNMP_BASE + ["trap-target"]
-    if state in ("merged", "replaced", "overridden"):
-        if want:
-            want_addr = want.get("address")
-            have_addr = have.get("address") if have else None
-            if want_addr and want_addr != have_addr:
-                cmds.append(_set(base + [want_addr]))
-            if want.get("community"):
-                cmds.append(_set(base + [want_addr, "community", want["community"]]))
-            if want.get("port"):
-                cmds.append(_set(base + [want_addr, "port", str(want["port"])]))
-    if state in ("replaced", "overridden"):
-        if have and (not want or have.get("address") != (want or {}).get("address")):
-            cmds.append(_delete(base))
-    return cmds
-
-
-def _build_v3_auth_privacy_commands(base, want_block, have_block, api_key):
-    cmds = []
-    if not want_block:
-        return cmds
-    block_base = base + [api_key]
-    have_block = have_block or {}
-    if want_block.get("type") and want_block["type"] != have_block.get("type"):
-        cmds.append(_set(block_base + ["type", want_block["type"]]))
-    if want_block.get("encrypted_key") and want_block["encrypted_key"] != have_block.get(
-        "encrypted_key",
-    ):
-        cmds.append(_set(block_base + ["encrypted-password", want_block["encrypted_key"]]))
-    if want_block.get("plaintext_key"):
-        cmds.append(_set(block_base + ["plaintext-key", want_block["plaintext_key"]]))
-    return cmds
-
-
-def _build_v3_user_commands(want_list, have_list, state):
-    cmds = []
-    want_map = {u["user"]: u for u in (want_list or [])}
-    have_map = {u["user"]: u for u in (have_list or [])}
-    base = SNMP_BASE + ["v3", "user"]
-    if state in ("replaced", "overridden"):
-        for username in have_map:
-            if username not in want_map:
-                cmds.append(_delete(base + [username]))
-    for username, want_user in want_map.items():
-        have_user = have_map.get(username, {})
-        user_base = base + [username]
-        cmds += _build_v3_auth_privacy_commands(
-            user_base,
-            want_user.get("authentication"),
-            have_user.get("authentication"),
-            "auth",
-        )
-        cmds += _build_v3_auth_privacy_commands(
-            user_base,
-            want_user.get("privacy"),
-            have_user.get("privacy"),
-            "privacy",
-        )
-        if want_user.get("group") and want_user["group"] != have_user.get("group"):
-            cmds.append(_set(user_base + ["group", want_user["group"]]))
-        if want_user.get("mode") and want_user["mode"] != have_user.get("mode"):
-            cmds.append(_set(user_base + ["mode", want_user["mode"]]))
-        if want_user.get("tsm_key") and want_user["tsm_key"] != have_user.get("tsm_key"):
-            cmds.append(_set(user_base + ["tsm-key", want_user["tsm_key"]]))
-    return cmds
-
-
-def _build_v3_group_commands(want_list, have_list, state):
-    cmds = []
-    want_map = {g["group"]: g for g in (want_list or [])}
-    have_map = {g["group"]: g for g in (have_list or [])}
-    base = SNMP_BASE + ["v3", "group"]
-    if state in ("replaced", "overridden"):
-        for name in have_map:
-            if name not in want_map:
-                cmds.append(_delete(base + [name]))
-    for name, want_group in want_map.items():
-        have_group = have_map.get(name, {})
-        group_base = base + [name]
-        for key, api_key in [("mode", "mode"), ("seclevel", "seclevel"), ("view", "view")]:
-            want_val = want_group.get(key)
-            have_val = have_group.get(key)
-            if want_val and want_val != have_val:
-                cmds.append(_set(group_base + [api_key, want_val]))
-            if state in ("replaced", "overridden") and have_val and want_val != have_val:
-                cmds.append(_delete(group_base + [api_key]))
-    return cmds
-
-
-def _build_v3_view_commands(want_list, have_list, state):
-    cmds = []
-    want_map = {v["view"]: v for v in (want_list or [])}
-    have_map = {v["view"]: v for v in (have_list or [])}
-    base = SNMP_BASE + ["v3", "view"]
-    if state in ("replaced", "overridden"):
-        for name in have_map:
-            if name not in want_map:
-                cmds.append(_delete(base + [name]))
-    for name, want_view in want_map.items():
-        have_view = have_map.get(name, {})
-        view_base = base + [name]
-        want_oid = str(want_view["oid"]) if want_view.get("oid") else None
-        have_oid = str(have_view.get("oid")) if have_view.get("oid") else None
-        if want_oid and want_oid != have_oid:
-            cmds.append(_set(view_base + ["oid", want_oid]))
-        if state in ("replaced", "overridden") and have_oid and want_oid != have_oid:
-            cmds.append(_delete(view_base + ["oid", have_oid]))
-        for key in ("exclude", "mask"):
-            want_val = want_view.get(key)
-            have_val = have_view.get(key)
-            if want_val and want_val != have_val:
-                cmds.append(_set(view_base + [key, want_val]))
-    return cmds
-
-
-def _build_v3_commands(want_v3, have_v3, state):
-    cmds = []
-    want_v3 = want_v3 or {}
-    have_v3 = have_v3 or {}
-    want_eid = want_v3.get("engine_id")
-    have_eid = have_v3.get("engine_id")
-    if want_eid and want_eid != have_eid:
-        cmds.append(_set(SNMP_BASE + ["v3", "engineid", want_eid]))
-    if state in ("replaced", "overridden") and have_eid and want_eid != have_eid:
-        cmds.append(_delete(SNMP_BASE + ["v3", "engineid"]))
-    cmds += _build_v3_group_commands(want_v3.get("groups"), have_v3.get("groups"), state)
-    cmds += _build_v3_user_commands(want_v3.get("users"), have_v3.get("users"), state)
-    cmds += _build_v3_view_commands(want_v3.get("views"), have_v3.get("views"), state)
-    return cmds
-
-
-def build_commands(want, have, state):
-    if state == "deleted":
-        if have:
-            return [_delete(SNMP_BASE)]
-        return []
-    cmds = []
-    cmds += _build_scalar_commands(want, have, state)
-    cmds += _build_community_commands(want.get("communities"), have.get("communities"), state)
-    cmds += _build_listen_address_commands(
-        want.get("listen_addresses"),
-        have.get("listen_addresses"),
-        state,
+
+
+def _device_to_argspec(raw):
+    if not raw:
+        return {}
+    result = _device_to_spec(
+        {k: v for k, v in raw.items() if k != "trap-target"},
+        _TOP_OPTIONS,
     )
-    cmds += _build_trap_target_commands(want.get("trap_target"), have.get("trap_target"), state)
-    cmds += _build_v3_commands(want.get("snmp_v3"), have.get("snmp_v3"), state)
-    return cmds
+    if raw.get("trap-target"):
+        tt = _single_from_device(raw["trap-target"], _derive_key_field(_TRAP_TARGET_OPTIONS))
+        if tt:
+            result["trap_target"] = tt
+    cast_by_spec(result, _TOP_OPTIONS)
+    return result
+
+
+# Device key names (as they appear in want/have, underscore-normalized)
+# whose child dict is a tag node keyed by an opaque value -- a username,
+# a community name, any user-supplied identifier -- rather than a schema
+# field name.
+_VERBATIM_KEYS = {"community", "listen_address", "group", "user", "view", "trap_target"}
+
+
+def _seed_tag_node_placeholders(want, have, verbatim_keys):
+    """dict_op's own key lookup falls back to guessing a translated
+    device key whenever a want key is missing from have entirely (a
+    brand-new entry). That guess is correct for a schema field name
+    (e.g. "trap_source" -> "trap-source" on first set) but wrong for a
+    tag-node key, which is an opaque value, not a schema name --
+    confirmed as a real bug: a username like "admin_user" was silently
+    becoming "admin-user" in the generated command the first time that
+    user was created (any tag-node key with an underscore would trigger
+    the same, since dict_op can't otherwise tell a schema name from a
+    value that merely happens to contain one).
+
+    Rather than teach the shared engine that distinction, this seeds an
+    empty placeholder into have (mutated in place) for every tag-node
+    entry present in want but not yet in have, keyed by the exact,
+    verbatim value from want. dict_op's own unmodified exact-match
+    lookup then finds it directly and never reaches its guessing
+    fallback at all -- the fix lives entirely in this module, not in
+    the shared engine, and every field the entry declares still
+    correctly shows up as "missing from have" and gets set, since the
+    placeholder is empty.
+    """
+    if not isinstance(want, dict):
+        return
+    have_idx = {k.replace("-", "_"): k for k in have}
+    for key, want_val in want.items():
+        if not isinstance(want_val, dict):
+            continue
+        norm_key = key.replace("-", "_")
+        orig_key = have_idx.get(norm_key, key)
+        have_val = have.setdefault(orig_key, {})
+        if not isinstance(have_val, dict):
+            continue
+        if norm_key in verbatim_keys:
+            for entry_key in want_val:
+                if entry_key not in have_val:
+                    have_val[entry_key] = None
+        else:
+            _seed_tag_node_placeholders(want_val, have_val, verbatim_keys)
+
+
+_CREDENTIAL_LEAVES = {"encrypted-password", "plaintext-password"}
+
+
+def _protect_credentials_from_purge(want, have):
+    """ "replaced"/"overridden" purge deletes anything in have that
+    isn't re-specified in want -- correct for ordinary config, but
+    wrong for a write-only credential leaf: the user can never read
+    back the current encrypted-password to re-supply it, so its
+    absence from a new config must not be read as "remove it".
+    Confirmed as a real device-rejected commit: VyOS requires an
+    auth/privacy node to carry an encrypted-password or plaintext-
+    password whenever the node exists at all, so purging the existing
+    hash out from under an unrelated field-level change (e.g. updating
+    "type") broke the commit entirely, not just the password.
+
+    Copies have's password leaf into want (mutating want in place)
+    wherever want doesn't already supply its own -- purge then sees it
+    as unchanged and never deletes it, while a genuinely new
+    plaintext_key/encrypted_key the user did provide still overrides
+    normally, since this only fills in what's missing.
+    """
+    if not isinstance(want, dict) or not isinstance(have, dict):
+        return
+    have_idx = {k.replace("-", "_"): k for k in have}
+    for key, want_val in want.items():
+        if not isinstance(want_val, dict):
+            continue
+        norm_key = key.replace("-", "_")
+        have_val = have.get(have_idx.get(norm_key, key))
+        if not isinstance(have_val, dict):
+            continue
+        if norm_key in ("auth", "privacy") and not (_CREDENTIAL_LEAVES & set(want_val)):
+            for cred in _CREDENTIAL_LEAVES:
+                if cred in have_val:
+                    want_val[cred] = have_val[cred]
+        _protect_credentials_from_purge(want_val, have_val)
+
+
+def build_commands(config, raw_have, state):
+    raw_have = raw_have or {}
+    config = config or {}
+
+    if state == "deleted":
+        return [("delete", _BASE)] if raw_have else []
+
+    want = _want_to_device(config)
+    # Rather than a generic key-name-based normalize_have, round-trip
+    # raw_have through the same structural converters used for want.
+    # This module has several keys that mean genuinely different things
+    # at different nesting depths (community/view/group are each both a
+    # tag node at one level and an unrelated scalar leaf at another) --
+    # a blanket tag_keys set would wrongly coerce the scalar occurrences
+    # into presence-dicts. Going through _device_to_argspec/
+    # _want_to_device instead resolves each occurrence with full
+    # knowledge of its actual position in the tree, not just its name.
+    norm_have = _want_to_device(_device_to_argspec(raw_have))
+    _seed_tag_node_placeholders(want, norm_have, _VERBATIM_KEYS)
+    _protect_credentials_from_purge(want, norm_have)
+
+    commands = []
+    if state == "overridden":
+        commands += dict_op(want, norm_have, _BASE, op="purge")
+    elif state == "replaced":
+        for section, section_want in want.items():
+            if not isinstance(section_want, dict):
+                continue
+            section_have = norm_have.get(section, {})
+            commands += dict_op(section_want, section_have, _BASE + [section], op="purge")
+    commands += dict_op(want, norm_have, _BASE, op="set")
+    return commands
 
 
 def _auth_privacy_spec():
@@ -784,7 +750,7 @@ ARGUMENT_SPEC = dict(
             trap_target=dict(
                 type="dict",
                 options=dict(
-                    address=dict(type="str"),
+                    address=dict(type="str", required=True),
                     community=dict(type="str"),
                     port=dict(type="int"),
                 ),
@@ -819,7 +785,7 @@ ARGUMENT_SPEC = dict(
                         type="list",
                         elements="dict",
                         options=dict(
-                            address=dict(type="str"),
+                            address=dict(type="str", required=True),
                             port=dict(type="int"),
                             protocol=dict(type="str", choices=["tcp", "udp"]),
                             type=dict(type="str", choices=["inform", "trap"]),
@@ -848,6 +814,9 @@ ARGUMENT_SPEC = dict(
     ),
 )
 
+_TOP_OPTIONS = ARGUMENT_SPEC["config"]["options"]
+_TRAP_TARGET_OPTIONS = _TOP_OPTIONS["trap_target"]["options"]
+
 
 def main():
     module = AnsibleModule(argument_spec=ARGUMENT_SPEC, supports_check_mode=True)
@@ -855,13 +824,13 @@ def main():
     state = module.params["state"]
     config = module.params.get("config") or {}
 
-    have = get_running_config(vyos)
+    raw_have = get_running_config(vyos)
+    have = _device_to_argspec(raw_have)
 
     if state == "gathered":
         module.exit_json(changed=False, gathered=have)
 
-    want = config
-    commands = build_commands(want, have, state)
+    commands = build_commands(config, raw_have, state)
 
     if module.check_mode:
         module.exit_json(changed=bool(commands), commands=commands, before=have)
@@ -872,7 +841,7 @@ def main():
         module.exit_json(
             changed=True,
             before=have,
-            after=want,
+            after=_device_to_argspec(get_running_config(vyos)),
             commands=commands,
             saved=saved,
             response=response,
diff --git a/plugins/modules/vyos_system.py b/plugins/modules/vyos_system.py
new file mode 100644
index 0000000..e2bbed5
--- /dev/null
+++ b/plugins/modules/vyos_system.py
@@ -0,0 +1,146 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# GNU General Public License v3.0+
+from __future__ import absolute_import, division, print_function
+
+
+__metaclass__ = type
+
+DOCUMENTATION = r"""
+---
+module: vyos_system
+short_description: Manage system settings on VyOS devices using REST API
+description:
+  - Manages basic system settings on VyOS devices via the REST API.
+  - Uses REST API (C(connection=httpapi)) instead of CLI.
+version_added: "1.0.0"
+author:
+  - VyOS Community (@vyos)
+options:
+  host_name:
+    description: Device hostname.
+    type: str
+  domain_name:
+    description: Device domain name.
+    type: str
+  name_server:
+    description: List of DNS name servers.
+    type: list
+    elements: str
+    aliases: [name_servers]
+  domain_search:
+    description: List of domain search suffixes.
+    type: list
+    elements: str
+  state:
+    description:
+      - C(present) applies the configuration.
+      - C(absent) removes the configuration.
+    type: str
+    choices: [present, absent]
+    default: present
+notes:
+  - Requires C(ansible_connection=httpapi) with the VyOS httpapi plugin.
+  - C(ansible_network_os) must be set to C(vyos.rest.vyos).
+"""
+
+EXAMPLES = r"""
+- name: Configure hostname and domain
+  vyos.rest.vyos_system:
+    host_name: router1
+    domain_name: example.com
+    name_server:
+      - 8.8.8.8
+      - 8.8.4.4
+    state: present
+
+- name: Remove domain name and name servers
+  vyos.rest.vyos_system:
+    domain_name: example.com
+    name_server:
+      - 8.8.8.8
+    state: absent
+"""
+
+RETURN = r"""
+before:
+  description: Module-owned system configuration before this module ran.
+  returned: always
+  type: dict
+after:
+  description: Module-owned system configuration after this module ran.
+  returned: when changed
+  type: dict
+commands:
+  description: List of API command tuples sent to the device.
+  returned: always
+  type: list
+saved:
+  description: Whether the config was saved after changes.
+  returned: when changed
+  type: bool
+response:
+  description: Raw API response.
+  returned: always
+  type: dict
+"""
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    dict_op,
+    owned_config,
+)
+
+
+_BASE = ["system"]
+
+ARGUMENT_SPEC = dict(
+    host_name=dict(type="str"),
+    domain_name=dict(type="str"),
+    name_server=dict(type="list", elements="str", aliases=["name_servers"]),
+    domain_search=dict(type="list", elements="str"),
+    state=dict(type="str", default="present", choices=["present", "absent"]),
+)
+
+
+def main():
+    module = AnsibleModule(ARGUMENT_SPEC, supports_check_mode=True)
+    vyos = VyOSModule(module)
+
+    state = module.params["state"]
+
+    # want: snake_case keys from YAML, nulls removed
+    _CANONICAL_KEYS = set(ARGUMENT_SPEC.keys()) - {"state"}
+    want = {k: v for k, v in module.params.items() if k in _CANONICAL_KEYS and v is not None}
+
+    # have: raw kebab-case keys from device, scoped by _BASE
+    have = vyos.get_config(_BASE)
+
+    # before/after: only keys owned by this module (declared in argspec)
+    before = owned_config(have, ARGUMENT_SPEC)
+
+    op = "set" if state == "present" else "delete"
+    commands = dict_op(want, have, _BASE, op=op)
+
+    if module.check_mode:
+        module.exit_json(changed=bool(commands), commands=commands, before=before)
+
+    if commands:
+        response = vyos.apply_commands(commands)
+        saved = vyos.save_config()
+        after = owned_config(vyos.get_config(_BASE), ARGUMENT_SPEC)
+        module.exit_json(
+            changed=True,
+            before=before,
+            after=after,
+            commands=commands,
+            saved=saved,
+            response=response,
+        )
+
+    module.exit_json(changed=False, before=before, after=before, commands=[])
+
+
+if __name__ == "__main__":
+    main()
diff --git a/plugins/modules/vyos_user.py b/plugins/modules/vyos_user.py
index 8ccb148..4413887 100644
--- a/plugins/modules/vyos_user.py
+++ b/plugins/modules/vyos_user.py
@@ -126,102 +126,131 @@ gathered:
   type: list
 saved:
   description: Whether the config was saved after changes.
-  returned: when changes are applied
+  returned: when changed
   type: bool
 response:
   description: Raw API response.
-  returned: when changes are applied
+  returned: always
   type: dict
 """
 
 from ansible.module_utils.basic import AnsibleModule
-from ansible_collections.vyos.rest.plugins.module_utils.vyos import VyOSModule
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    autoclean,
+    dict_op,
+    from_device,
+    normalize_have,
+)
 
 
 _BASE = ["system", "login", "user"]
 
+# "public-keys" is a tag node (keyed by key identifier) that could in
+# principle collapse to a bare value for a single entry; defensive only
+# -- "key" is required by the argspec so a real collapse is unlikely,
+# but the guard costs nothing and matches the pattern used everywhere
+# else a tag node is involved.
+_TAG_KEYS = {"public-keys"}
+
+# Users this module will never delete under state=absent, no matter what
+# the playbook asks for -- "vyos" is required for REST API access itself,
+# so deleting it would lock out every subsequent module call.
+_PROTECTED_USERS = {"vyos"}
+
+
+def _public_keys_to_device(keys):
+    return {
+        k["name"]: autoclean({kk: vv for kk, vv in k.items() if kk != "name"}) for k in keys or []
+    }
+
+
+def _public_keys_from_device(raw):
+    return [{"name": name, **from_device(data or {})} for name, data in sorted((raw or {}).items())]
+
+
+def _user_to_device(user):
+    """password/update_password are deliberately excluded here and
+    handled entirely outside dict_op in build_commands() -- "password"
+    (plaintext, write-only) and have's "encrypted-password" are
+    structurally different data with no valid equality comparison
+    between them, so whether to set it is a policy decision
+    (update_password), never a diff. public_keys nests under a literal
+    "authentication" wrapper the argspec doesn't have.
+    """
+    entry = autoclean(
+        {
+            k: v
+            for k, v in user.items()
+            if k not in ("name", "password", "update_password", "public_keys")
+        },
+    )
+    if user.get("public_keys"):
+        entry["authentication"] = {"public_keys": _public_keys_to_device(user["public_keys"])}
+    return entry
+
+
+def _user_from_device(name, data):
+    data = dict(data or {})
+    auth = data.pop("authentication", None) or {}
+    entry = {"name": name, **from_device(data)}
+    if auth.get("encrypted-password"):
+        entry["encrypted_password"] = auth["encrypted-password"]
+    pub_keys_raw = auth.get("public-keys")
+    if pub_keys_raw:
+        entry["public_keys"] = _public_keys_from_device(pub_keys_raw)
+    return entry
+
 
 def get_running_config(vyos):
-    raw = vyos.get_config(_BASE)
+    raw = vyos.get_config(_BASE) or {}
+    if isinstance(raw, dict):
+        raw = raw.get("user", raw)
+    return raw if isinstance(raw, dict) else {}
+
+
+def _device_to_argspec(raw):
     if not raw or not isinstance(raw, dict):
         return []
-    raw = raw.get("user", raw)
-    result = []
-    for username, data in sorted(raw.items()):
-        user = {"name": username}
-        data = data or {}
-        if data.get("full-name"):
-            user["full_name"] = data["full-name"]
-        auth = data.get("authentication", {}) or {}
-        if auth.get("encrypted-password"):
-            user["encrypted_password"] = auth["encrypted-password"]
-        pub_keys = auth.get("public-keys", {}) or {}
-        if pub_keys and isinstance(pub_keys, dict):
-            keys = []
-            for key_name, key_data in sorted(pub_keys.items()):
-                key_data = key_data or {}
-                k = {"name": key_name}
-                if key_data.get("key"):
-                    k["key"] = key_data["key"]
-                if key_data.get("type"):
-                    k["type"] = key_data["type"]
-                keys.append(k)
-            if keys:
-                user["public_keys"] = keys
-        result.append(user)
-    return result
-
-
-def build_commands(users, have_list, state):
-    cmds = []
-    have_map = {u["name"]: u for u in have_list}
+    return [_user_from_device(name, data) for name, data in sorted(raw.items())]
+
+
+def build_commands(users, raw_have, state):
+    raw_have = raw_have or {}
+    users = users or []
 
     if state == "absent":
+        commands = []
         for user in users:
             name = user["name"]
-            if name in have_map:
-                cmds.append(("delete", _BASE + [name]))
-        return cmds
-
-    # state == "present"
+            if name in _PROTECTED_USERS:
+                continue
+            if name in raw_have:
+                commands.append(("delete", _BASE + [name]))
+        return commands
+
+    # state == "present": additive-only, matches the original module's
+    # scope exactly -- existing fields/keys not mentioned in a user's
+    # config are left alone, never removed (there's no "replaced" state
+    # here to make a full-model rewrite meaningful).
+    commands = []
+    norm_have = normalize_have(raw_have, _TAG_KEYS)
     for user in users:
         name = user["name"]
-        have = have_map.get(name, {})
+        is_new = name not in raw_have
         ubase = _BASE + [name]
-        is_new = name not in have_map
+        have_user = norm_have.get(name) or {}
 
-        # full_name
-        if user.get("full_name") and user["full_name"] != have.get("full_name"):
-            cmds.append(("set", ubase + ["full-name", user["full_name"]]))
+        commands += dict_op(_user_to_device(user), have_user, ubase, op="set")
 
-        # password
         if user.get("password"):
-            update_pw = user.get("update_password", "always")
-            if update_pw == "always" or is_new:
-                cmds.append(
-                    (
-                        "set",
-                        ubase
-                        + [
-                            "authentication",
-                            "plaintext-password",
-                            user["password"],
-                        ],
-                    ),
+            update_policy = user.get("update_password", "always")
+            if update_policy == "always" or is_new:
+                commands.append(
+                    ("set", ubase + ["authentication", "plaintext-password", user["password"]]),
                 )
 
-        # public_keys
-        want_keys = {k["name"]: k for k in (user.get("public_keys") or [])}
-        have_keys = {k["name"]: k for k in (have.get("public_keys") or [])}
-        for key_name, key_data in want_keys.items():
-            have_key = have_keys.get(key_name, {})
-            kbase = ubase + ["authentication", "public-keys", key_name]
-            if key_data.get("key") and key_data["key"] != have_key.get("key"):
-                cmds.append(("set", kbase + ["key", key_data["key"]]))
-            if key_data.get("type") and key_data["type"] != have_key.get("type"):
-                cmds.append(("set", kbase + ["type", key_data["type"]]))
-
-    return cmds
+    return commands
 
 
 ARGUMENT_SPEC = dict(
@@ -273,12 +302,13 @@ def main():
     state = module.params["state"]
     users = module.params.get("users") or []
 
-    have = get_running_config(vyos)
+    raw_have = get_running_config(vyos)
+    have = _device_to_argspec(raw_have)
 
     if state == "gathered":
         module.exit_json(changed=False, gathered=have)
 
-    commands = build_commands(users, have, state)
+    commands = build_commands(users, raw_have, state)
 
     if module.check_mode:
         module.exit_json(changed=bool(commands), commands=commands, before=have)
@@ -289,7 +319,7 @@ def main():
         module.exit_json(
             changed=True,
             before=have,
-            after=get_running_config(vyos),
+            after=_device_to_argspec(get_running_config(vyos)),
             commands=commands,
             saved=saved,
             response=response,
diff --git a/plugins/modules/vyos_vlan.py b/plugins/modules/vyos_vlan.py
new file mode 100644
index 0000000..3c66122
--- /dev/null
+++ b/plugins/modules/vyos_vlan.py
@@ -0,0 +1,241 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# GNU General Public License v3.0+
+from __future__ import absolute_import, division, print_function
+
+
+__metaclass__ = type
+
+DOCUMENTATION = r"""
+---
+module: vyos_vlan
+short_description: Manage VLAN (vif) configuration on VyOS devices using REST API
+description:
+  - Manages VLAN sub-interface configuration on VyOS Ethernet interfaces
+    via the REST API.
+  - Uses REST API (C(connection=httpapi)) instead of CLI.
+version_added: "1.0.0"
+author:
+  - VyOS Community (@vyos)
+options:
+  config:
+    description: List of VLAN configurations.
+    type: list
+    elements: dict
+    suboptions:
+      vlan_id:
+        description: VLAN ID (0-4094).
+        type: int
+        required: true
+      description:
+        description: VLAN description.
+        type: str
+      address:
+        description: IP address for the VLAN interface.
+        type: str
+      interfaces:
+        description: List of Ethernet interfaces to configure this VLAN on.
+        type: list
+        elements: str
+        required: true
+  state:
+    description:
+      - C(present) creates or updates VLANs.
+      - C(absent) removes VLANs.
+      - C(gathered) returns current VLAN configuration.
+    type: str
+    choices: [present, absent, gathered]
+    default: present
+notes:
+  - Requires C(ansible_connection=httpapi) with the VyOS httpapi plugin.
+  - C(ansible_network_os) must be set to C(vyos.rest.vyos).
+"""
+
+EXAMPLES = r"""
+- name: Configure VLANs
+  vyos.rest.vyos_vlan:
+    config:
+      - vlan_id: 10
+        description: VLAN10
+        address: 192.168.10.1/24
+        interfaces:
+          - eth1
+      - vlan_id: 20
+        description: VLAN20
+        interfaces:
+          - eth1
+          - eth2
+    state: present
+
+- name: Remove a VLAN
+  vyos.rest.vyos_vlan:
+    config:
+      - vlan_id: 10
+        interfaces:
+          - eth1
+    state: absent
+
+- name: Gather VLAN configuration
+  vyos.rest.vyos_vlan:
+    state: gathered
+"""
+
+RETURN = r"""
+before:
+  description: VLAN configuration before this module ran.
+  returned: always
+  type: list
+after:
+  description: VLAN 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 VLAN configuration as structured data.
+  returned: when state is gathered
+  type: list
+saved:
+  description: Whether the config was saved after changes.
+  returned: when changed
+  type: bool
+response:
+  description: Raw API response.
+  returned: always
+  type: dict
+"""
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    dict_op,
+)
+
+
+_BASE = ["interfaces", "ethernet"]
+
+# Keys that define structure (path construction) vs passthrough (diff engine)
+_STRUCTURAL_KEYS = {"vlan_id", "interfaces"}
+
+
+def get_running_config(vyos):
+    raw = vyos.get_config(_BASE)
+    if not raw or not isinstance(raw, dict):
+        return []
+
+    eth_data = raw.get("ethernet", raw)
+    if not isinstance(eth_data, dict):
+        return []
+
+    # Structural reshape: ethernet.<iface>.vif.<id> -> flat list per vlan_id
+    # Raw device keys preserved — dict_op handles - <-> _ normalization
+    vlan_map = {}
+    for iface_name, iface_data in sorted(eth_data.items()):
+        iface_data = iface_data or {}
+        vif_data = iface_data.get("vif", {}) or {}
+        for vlan_id_str, vif_cfg in sorted(
+            vif_data.items(),
+            key=lambda x: int(x[0]),
+        ):
+            vif_cfg = vif_cfg or {}
+            vlan_id = int(vlan_id_str)
+            if vlan_id not in vlan_map:
+                vlan_map[vlan_id] = {"vlan_id": vlan_id, "interfaces": [], "_raw": {}}
+            vlan_map[vlan_id]["interfaces"].append(iface_name)
+            vlan_map[vlan_id]["_raw"].update(vif_cfg)
+
+    result = []
+    for vid, entry in sorted(vlan_map.items()):
+        item = {"vlan_id": entry["vlan_id"], "interfaces": entry["interfaces"]}
+        for k, v in entry["_raw"].items():
+            item[k] = v[0] if isinstance(v, list) and len(v) == 1 else v
+        result.append(item)
+    return result
+
+
+def build_commands(config, have_list, state):
+    cmds = []
+
+    have_map = {(e["vlan_id"], iface): e for e in have_list for iface in e.get("interfaces", [])}
+
+    for want in config or []:
+        vlan_id = str(want["vlan_id"])
+        for iface in want.get("interfaces") or []:
+            vif_base = _BASE + [iface, "vif", vlan_id]
+            have_entry = have_map.get((want["vlan_id"], iface), {})
+
+            if state == "absent":
+                if have_entry:
+                    cmds.append(("delete", vif_base))
+                continue
+
+            # Passthrough fields — dict_op handles - <-> _ normalization
+            want_vif = {
+                k: v for k, v in want.items() if k not in _STRUCTURAL_KEYS and v is not None
+            }
+            have_vif = {k: v for k, v in have_entry.items() if k not in _STRUCTURAL_KEYS}
+
+            new_cmds = dict_op(want_vif, have_vif, vif_base, op="set")
+            if not new_cmds and not have_entry:
+                cmds.append(("set", vif_base))
+            else:
+                cmds += new_cmds
+
+    return cmds
+
+
+ARGUMENT_SPEC = dict(
+    config=dict(
+        type="list",
+        elements="dict",
+        options=dict(
+            vlan_id=dict(type="int", required=True),
+            description=dict(type="str"),
+            address=dict(type="str"),
+            interfaces=dict(type="list", elements="str", required=True),
+        ),
+    ),
+    state=dict(
+        type="str",
+        default="present",
+        choices=["present", "absent", "gathered"],
+    ),
+)
+
+
+def main():
+    module = AnsibleModule(ARGUMENT_SPEC, supports_check_mode=True)
+    vyos = VyOSModule(module)
+
+    state = module.params["state"]
+    config = module.params.get("config") or []
+
+    have = get_running_config(vyos)
+
+    if state == "gathered":
+        module.exit_json(changed=False, gathered=have)
+
+    commands = build_commands(config, 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()
+        module.exit_json(
+            changed=True,
+            before=have,
+            after=get_running_config(vyos),
+            commands=commands,
+            saved=saved,
+            response=response,
+        )
+
+    module.exit_json(changed=False, before=have, after=have, commands=[])
+
+
+if __name__ == "__main__":
+    main()