Sync VLANs

The sync_vlans task reconciles VLAN names, descriptions, and tagged/untagged interface memberships from live devices into NetBox. It requests the normalized TTP vlans getter once for the validated device set, aggregates all successful worker results, then compares VLANs by VID and VLAN group. Names and descriptions are synchronized values and do not form part of a VLAN's identity.

Ordered vlan_map rules place matching VLANs into existing VLAN groups. VLANs which match no rule use the scalar vlan_group when supplied, otherwise they use their device site. Set require_vlan_group=True to skip and report VLANs which do not resolve through either group selection mechanism. VLAN groups are recommended for new deployments because direct VLAN-to-site assignment is deprecated in NetBox 4.4.

sync_device_interfaces creates and reconciles interface objects. Run it first when interfaces are missing, then run sync_vlans to reconcile VLAN attributes, memberships, and VLAN-derived interface mode using the live VLAN getter.

Inputs

Input	Default	Description
instance	Worker default	NetBox instance name.
dry_run	False	Calculate the diff without writing to NetBox.
with_approval	False	Present the prepared plan for approval before writing. Ignored during dry-run.
timeout	600	Timeout in seconds for host resolution and TTP parsing.
devices	None	Explicit NetBox and Nornir device names.
branch	None	NetBox Branching plugin branch name.
interface_map	None	Interface rename rules shared with interface sync, inline or in an nf:// YAML file.
vlan_group	None	Existing group for live VLANs not matched by vlan_map.
vlan_map	None	Ordered rules mapping live VLANs to existing groups, inline or in an nf:// YAML file.
require_vlan_group	False	Require every VLAN to resolve through vlan_map or vlan_group; unmatched VLANs are reported and skipped.
filter_by_vlan_ids	None	VLAN IDs or inclusive ranges such as 100 and 200-299.
preserve_description	None	Preserve NetBox descriptions only when live text is empty. Use True to always preserve or False to always use live text.
Nornir filters	None	FO, FB, FH, FC, FR, FG, FP, FL, FM, FX, and FN.

Provide devices or at least one Nornir host filter.

Existing NetBox VLAN resolution

For each live VLAN, the task fetches every NetBox VLAN with the same VID. VLAN name is not part of matching; it is a value that may need to be updated.

When vlan_map or vlan_group selects a group, that selection is authoritative:

1. Validate that the VID belongs to the selected group's VID ranges.
2. Validate the selected group scope against the device.
3. If either check fails, return an error and do not fall back.
4. Search only same-VID VLANs in that exact group.
5. If none exists, prepare a new VLAN in that group.

Without an explicit group selection, same-VID candidates are checked in this order:

1. VLAN in any device-compatible group whose VID ranges include the live VID, including an unscoped group.
2. VLAN assigned directly to the device site.
3. Global VLAN with neither a group nor a site.

Only the first non-empty level is considered. If that level contains more than one VLAN, matching is ambiguous: the live VLAN is skipped, an error lists the candidate NetBox VLAN IDs, and no arbitrary candidate is selected. If no candidate exists, a VLAN is prepared in the device site; global VLANs are only reused as existing fallbacks and are not created by default.

VLAN group scope matching

A scoped group must exactly match the corresponding direct device value:

Group scope	Device value
Site	Device site
Region	Device site's region
Site group	Device site's site group
Location	Device location
Rack	Device rack
Rack group	Device rack's rack group

For location matching, the assigned rack's direct location is used when the device location is empty. Parent regions, site groups, locations, and rack groups are not searched. If the required device value is empty, the candidate group is rejected. Cluster and cluster-group scopes are ignored.

VLAN group VID range matching

A grouped VLAN is eligible only when its VID is included in the group's inclusive vid_ranges, returned by NetBox as integer pairs:

vid_ranges:
  - [1, 20]
  - [50, 100]

The resolver checks the live VID against each interval directly instead of expanding the intervals into individual VLAN IDs. An automatically discovered out-of-range group VLAN is skipped while candidate search continues. An explicitly selected out-of-range group is an error and does not fall back to another group, the device site, or a global VLAN.

VLAN mapping

Each rule contains an exact NetBox VLAN group name. Additional matching criteria are optional:

- set_vlan_group: CAMPUS
  match_vlan_ids:
    - 100-199
  vlan_names:
    - USERS*
    - VOICE*
  match_device_names:
    - leaf-*
  match_interface_names:
    - Ethernet*

Rules are evaluated in list order and the first match wins for the entire VLAN and all its memberships. Values inside one criterion use OR logic. Populated criteria use AND logic. VLAN and device names use case-sensitive glob matching. VLAN ranges are inclusive and must remain within 1..4094. match_interface_names matches when any tagged or untagged interface on the VLAN matches the rule. For VLANs without interfaces, rules with interface-name criteria do not match. match_vlan_ids controls whether the rule selects its group; after selection, the group's own vid_ranges are validated separately. An unmatched VLAN uses vlan_group when supplied. Without either group match, it uses its device site unless require_vlan_group=True; strict mode reports and skips that VLAN instead.

interface_map uses the same device name, device type, match, and replacement rules as sync_device_interfaces. The task applies the first matching rename rule using substring containment before VLAN mapping and NetBox interface lookup. When different selected live names map to the same NetBox interface name, only the first name from a successfully resolved VLAN and its memberships are processed. Pass the same interface map to both tasks when live interface names are renamed.

The task resolves groups by exact name and does not create or update groups. All groups named by vlan_map or vlan_group are checked before live data is collected. Missing groups are logged and included in errors. VLANs selecting a missing group are skipped while other VLANs continue. A selected group is validated against both its VID ranges and the device scope. An incompatible selection is logged as an error, included in errors, and skipped without falling back to another VLAN; the error directs the operator to fix the group scope, VID ranges, or mapping.

Store the same YAML list in the File Sharing service and pass its URL when the rules are reused or maintained separately:

vlan_map: nf://netbox/vlan_map.yaml

Live data

The task runs Nornir parse_ttp with get="vlans", which returns:

- vid: 100
  name: USERS
  description: User access VLAN
  tagged_interfaces:
    - Ethernet5
  untagged_interfaces:
    - Ethernet6

The getter must supply both interface lists, including empty lists. Names and descriptions are trimmed, null descriptions become an empty string, and case is preserved. filter_by_vlan_ids removes out-of-range records from both the complete live device dataset and NetBox before comparison.

For existing VLANs, the default preserve_description=None keeps the NetBox description when live text is empty and otherwise uses the live value. Set it to True to always retain the NetBox description, or False to always apply the live value, including an empty string. Newly created VLANs always use the live description.

Identical live records from multiple devices in one scope are collapsed. Live VLANs with the same VID but different names or descriptions are reported as source conflicts. An automatically derived name matching VLAN<VID> yields to the first different name reported by a device. If all live observations retain the automatic name, an existing NetBox VLAN name is preserved. Otherwise, the first device in sorted device-name order supplies the value to synchronize. Each conflicting device is identified in errors. A conflict does not fail or skip that VLAN. Results returned for the same device by multiple Nornir workers are aggregated before identical live records are collapsed.

Interface memberships

VLAN attributes and interface assignments use separate snapshots. VLAN state is keyed by NetBox scope and VID. Interface state is keyed by device and interface name, with mode, tagged_vlans, and untagged_vlan fields. VLAN references use scope/VID, for example site:NORFAB-LAB/110 or group:CAMPUS/210.

The task adds reported memberships and preserves existing NetBox memberships. Empty live membership lists do not clear tagged or untagged assignments. Assignments on other devices and unresolved or unselected VLANs are also preserved. Missing referenced interfaces are reported and omitted from the interface diff while VLAN processing continues.

An interface may carry the same VLAN both tagged and untagged, or use different VLANs for tagged and untagged traffic. A new untagged assignment replaces the current assignment even when the old VLAN is outside the VID filter. The interface diff reports the old and new untagged_vlan references directly. This native VLAN replacement is the only membership removal performed by the task.

An interface with any tagged VLAN uses tagged mode, including an interface that also has an untagged VLAN. An interface with only an untagged VLAN uses access mode. Removing every VLAN membership does not change the existing mode. Q-in-Q service VLAN assignments are not currently synchronized.

Output

Results have two top-level keys. vlans contains VLAN attribute actions keyed by scope, such as group:CAMPUS, site:NORFAB-LAB, or global. interfaces contains mode and membership actions keyed by device and interface name.

Dry-run returns both standard sync diffs:

{
  "vlans": {
    "site:NORFAB-LAB": {
      "create": [110],
      "create_details": {
        "110": {
          "name": "VOICE",
          "description": ""
        }
      },
      "update": {
        "210": {
          "name": {
            "old_value": "VLAN_210",
            "new_value": "USERS"
          }
        }
      },
      "delete": [],
      "in_sync": [310]
    }
  },
  "interfaces": {
    "leaf-1": {
      "create": [],
      "update": {
        "Ethernet6": {
          "mode": {
            "old_value": "tagged",
            "new_value": "access"
          },
          "untagged_vlan": {
            "old_value": "site:NORFAB-LAB/100",
            "new_value": "site:NORFAB-LAB/110"
          }
        },
        "Ethernet5": {
          "tagged_vlans": {
            "old_value": [],
            "new_value": ["site:NORFAB-LAB/110"]
          }
        }
      },
      "delete": [],
      "in_sync": []
    }
  }
}

Live runs use completed-action verbs. The prepared plan remains available in the top-level diff field:

{
  "vlans": {
    "site:NORFAB-LAB": {
      "created": [110],
      "updated": [210],
      "deleted": [],
      "in_sync": [310]
    }
  },
  "interfaces": {
    "leaf-1": {
      "created": [],
      "updated": ["Ethernet5", "Ethernet6"],
      "deleted": [],
      "in_sync": []
    }
  }
}

An existing VLAN with a stale name is updated directly because name is not part of identity matching.

create_details displays the desired attributes of new VLANs alongside the standard make_diff actions. Membership-only changes appear under interfaces and do not mark the VLAN object as updated. Dry-run and approval show both diffs before any writes. The same preview is retained in diff.

Deletions

The task does not delete VLANs. Live parsing does not provide a reliable way to identify which additional NetBox VLANs are stale. The standard result shape therefore retains empty delete and deleted lists.

Branches

The task obtains its NetBox client with the requested branch. Reads and writes therefore remain inside that branch. The NetBox Branching plugin must be installed and configured for branch use.

Bulk Request Batching

List-based NetBox writes are sent as sequential requests containing at most batch_size objects. The default is 1000; set any integer greater than zero to tune the request size. Each batch emits matching progress event and log messages. If a request fails, the task stops and returns the results recorded for earlier successful batches.

Examples

CLIPython

Preview site-scoped VLAN changes:

nf# netbox sync vlans devices fn-ceos-lf-1 dry-run

Select devices with a Nornir filter and restrict VLAN IDs:

nf# netbox sync vlans FC leaf vlan-ids 100 200-299

Clear existing descriptions when live descriptions are empty:

nf# netbox sync vlans FC leaf vlan-ids 100 200-299 preserve-description false

Place all unmatched VLANs into one existing group:

nf# netbox sync vlans devices fn-ceos-lf-1 vlan-group CAMPUS

Require every VLAN to select a VLAN group:

nf# netbox sync vlans devices fn-ceos-lf-1 vlan-map nf://netbox/vlan_map.yaml require-vlan-group

from norfab.core.nfapi import NorFab

with NorFab(inventory="./inventory.yaml") as nf:
    client = nf.make_client()
    result = client.run_job(
        "netbox",
        "sync_vlans",
        workers="any",
        kwargs={
            "devices": ["fn-ceos-lf-1", "fn-ceos-lf-2"],
            "dry_run": True,
            "filter_by_vlan_ids": ["100-399"],
            "preserve_description": None,
            "vlan_map": [
                {
                    "set_vlan_group": "CAMPUS",
                    "match_vlan_ids": ["100-199"],
                    "vlan_names": ["TEST_L*"],
                    "match_device_names": ["fn-ceos-lf-*"],
                }
            ],
        },
    )
    print(result)

Troubleshooting

Missing parser data

Confirm the device platform is supported by the TTP vlans getter and that the Nornir worker can run the getter's command. Missing or malformed results are reported as errors; valid results from other devices and workers still proceed.

Live VLAN conflicts

Select devices that share one authoritative VLAN definition or correct their name and description differences. The first device in sorted order is used and each later conflicting device is listed in errors. The task continues to synchronize the first device's values.

VLAN group resolution

Check that the scalar vlan_group and every group named in vlan_map exist exactly as written. A missing group does not abort the task; each live VLAN that selects it is skipped and reported in errors. Also confirm that the selected group's site, region, site group, location, rack, or rack group scope exactly matches the device's direct assignment. Scope mismatches are reported and never fall back to an unrelated same-VID VLAN.

NetBox bulk failures

The task completes VLAN creation before updating VLAN attributes or interface assignments. A failed creation request stops the task immediately with an error. Successful creates from earlier requests remain recorded in created; there is no rollback across requests. Correct the reported validation or dependency error and rerun the dry-run.

Task command shell reference

nf# man tree netbox.sync.vlans

R - required field, M - supports multiline input, D - dynamic key

root
└── netbox:    Netbox service
    └── sync:    Sync Netbox data
        └── vlans:    Sync live VLAN configuration with NetBox
            ├── instance:    Netbox instance name to target
            ├── dry-run:    Calculate the VLAN diff without writing to NetBox, default 'False'
            ├── branch:    NetBox branching plugin branch name to use
            ├── FO:    Filter hosts using Filter Object
            ├── FB:    Filter hosts by name using Glob Patterns
            ├── FH:    Filter hosts by hostname
            ├── FC:    Filter hosts containment of pattern in name
            ├── FR:    Filter hosts by name using Regular Expressions
            ├── FG:    Filter hosts by group
            ├── FP:    Filter hosts by hostname using IP Prefix
            ├── FL:    Filter hosts by names list
            ├── FM:    Filter hosts by platform
            ├── FX:    Filter hosts excluding them by name
            ├── FN:    Negate the match
            ├── devices:    List of NetBox devices to collect VLANs from
            ├── timeout:    Job timeout
            ├── with-approval:    Preview VLAN changes and ask for review before writing to NetBox, default 'False'
            ├── interface-map:    Interface name mapping rules shared with interface sync
            ├── vlan-group:    Exact group name for live VLANs not matched by vlan-map
            ├── vlan-map:    Ordered rules mapping live VLANs to NetBox VLAN groups
            ├── require-vlan-group:    Require every live VLAN to resolve to a VLAN group, default 'False'
            ├── vlan-ids:    VLAN IDs or inclusive ranges to reconcile
            ├── preserve-description:    Preserve NetBox descriptions always (true), when live text is empty (null), or never (false)
            ├── workers:    Filter worker to target, default 'any'
            ├── verbose-result:    Control output details, default 'False'
            └── nowait:    Do not wait for job to complete, default 'False'
nf#

Python API reference

Synchronize live VLAN attributes and interface memberships with NetBox.

VLANs are mapped by the first matching vlan_map rule. Rule criteria match VLAN IDs, VLAN names, device names, and each interface name; populated criteria are combined with AND. VLANs which match no rule use vlan_group when supplied. Otherwise, they use their device site unless require_vlan_group=True.

Name conflicts with NetBox or between proposed VLANs in the same scope are reported and skipped. The first proposed name is retained; skipped creations are also removed from interface membership targets.

Parameters:

Name	Type	Description	Default
job	Job	NorFab job object.	required
instance	Union[None, str]	NetBox instance name. Uses the default instance when omitted.	None
dry_run	bool	Return the calculated diff without writing to NetBox.	False
with_approval	bool	Ask for approval before applying the prepared diff.	False
timeout	int	Timeout in seconds for Nornir host resolution and parsing.	600
devices	Union[None, list]	Explicit NetBox and Nornir device names.	None
branch	Union[None, str]	NetBox Branching plugin branch name.	None
interface_map	Union[None, str, list]	Interface rename rules shared with interface sync.	None
vlan_group	Union[None, str]	Group for VLANs not matched by vlan_map.	None
vlan_map	Union[None, str, list]	Ordered VLAN-to-group mapping rules, or an nf:// YAML file containing them.	None
require_vlan_group	bool	Require every VLAN to resolve to a VLAN group instead of falling back to its device site.	False
filter_by_vlan_ids	Union[None, list[str]]	VLAN IDs or inclusive ranges to reconcile.	None
preserve_description	Union[None, bool]	Description preservation policy. None preserves NetBox text when the live description is empty, True always preserves NetBox text, and False always uses live text.	None
**kwargs	Any	Nornir FFun host filters.	{}

Returns:

Name	Type	Description
Result	Result	VLAN actions keyed by scope and interface actions keyed by device.

Source code in norfab\workers\netbox_worker\vlan_tasks.py

@Task(
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()},
    input=SyncVlansInput,
    output=SyncVlansResult,
    mcp={
        "annotations": {
            "title": "Sync VLANs",
            "readOnlyHint": False,
            "destructiveHint": True,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def sync_vlans(
    self,
    job: Job,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    with_approval: bool = False,
    timeout: int = 600,
    devices: Union[None, list] = None,
    branch: Union[None, str] = None,
    interface_map: Union[None, str, list] = None,
    vlan_group: Union[None, str] = None,
    vlan_map: Union[None, str, list] = None,
    require_vlan_group: bool = False,
    filter_by_vlan_ids: Union[None, list[str]] = None,
    preserve_description: Union[None, bool] = None,
    batch_size: int = 1000,
    **kwargs: Any,
) -> Result:
    """Synchronize live VLAN attributes and interface memberships with NetBox.

    VLANs are mapped by the first matching ``vlan_map`` rule. Rule criteria
    match VLAN IDs, VLAN names, device names, and each interface name;
    populated criteria are combined with AND. VLANs which match no rule use
    ``vlan_group`` when supplied. Otherwise, they use their device site
    unless ``require_vlan_group=True``.

    Name conflicts with NetBox or between proposed VLANs in the same scope
    are reported and skipped. The first proposed name is retained; skipped
    creations are also removed from interface membership targets.

    Args:
        job: NorFab job object.
        instance: NetBox instance name. Uses the default instance when omitted.
        dry_run: Return the calculated diff without writing to NetBox.
        with_approval: Ask for approval before applying the prepared diff.
        timeout: Timeout in seconds for Nornir host resolution and parsing.
        devices: Explicit NetBox and Nornir device names.
        branch: NetBox Branching plugin branch name.
        interface_map: Interface rename rules shared with interface sync.
        vlan_group: Group for VLANs not matched by ``vlan_map``.
        vlan_map: Ordered VLAN-to-group mapping rules, or an ``nf://`` YAML
            file containing them.
        require_vlan_group: Require every VLAN to resolve to a VLAN group
            instead of falling back to its device site.
        filter_by_vlan_ids: VLAN IDs or inclusive ranges to reconcile.
        preserve_description: Description preservation policy. ``None`` preserves
            NetBox text when the live description is empty, ``True`` always
            preserves NetBox text, and ``False`` always uses live text.
        **kwargs: Nornir FFun host filters.

    Returns:
        Result: VLAN actions keyed by scope and interface actions keyed by device.
    """
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:sync_vlans",
        result=SyncVlansResultPayload().model_dump(),
        resources=[instance],
        dry_run=dry_run,
        diff={"vlans": {}, "interfaces": {}},
    )

    message = f"starting VLAN sync using NetBox instance '{instance}'"
    job.event(message)
    log.info(message)
    nb = self._get_pynetbox(instance, branch=branch, job=job)
    selected_devices = set(devices or [])
    if kwargs:
        selected_devices.update(self.get_nornir_hosts(kwargs, timeout))
    devices = sorted(selected_devices)
    if not devices:
        message = "no devices specified"
        job.event(message, severity="ERROR")
        log.error(message)
        ret.errors.append(message)
        ret.failed = True
        return ret
    # Resolve the requested devices once; later stages use this same object map.
    netbox_devices = self.bulk_filter(
        nb.dcim.devices,
        name=devices,
        fields="id,name,site,location,rack,device_type",
    )
    nb_devices = {str(device.name): device for device in netbox_devices}
    for device in devices:
        if device not in nb_devices:
            message = f"device '{device}' not found in NetBox"
            job.event(message, severity="ERROR")
            log.error(message)
            ret.errors.append(message)
    if not nb_devices:
        ret.failed = True
        return ret
    device_scopes = load_device_vlan_scopes(
        nb_devices.values(), nb, self.bulk_filter
    )
    message = f"loaded VLAN scopes for {len(nb_devices)} NetBox device(s)"
    job.event(message)
    log.info(message)
    selected_vids = set()
    for value in filter_by_vlan_ids or []:
        selected_vids.update(
            int(vid) for vid in expand_alphanumeric_range(f"[{value}]")
        )
    try:
        if self.is_url(vlan_map):
            vlan_map = TypeAdapter(list[VlanMapRule]).validate_python(
                yaml.safe_load(self.fetch_file(vlan_map, raise_on_fail=True)) or []
            )
        if self.is_url(interface_map):
            interface_map = TypeAdapter(list[InterfaceMapRule]).validate_python(
                yaml.safe_load(self.fetch_file(interface_map, raise_on_fail=True))
                or []
            )
    except Exception as exc:
        message = f"failed to load VLAN sync mapping: {exc}"
        job.event(message, severity="ERROR")
        log.error(message)
        ret.errors.append(message)
        ret.failed = True
        return ret
    interface_map = [dict(rule) for rule in interface_map or []]
    # Expand rule VID ranges once and discard missing groups from later lookups.
    rules = []
    for rule in vlan_map or []:
        rule = dict(rule)
        expanded_vlan_ids = set()
        for value in rule.get("match_vlan_ids") or []:
            expanded_vlan_ids.update(
                int(vid) for vid in expand_alphanumeric_range(f"[{value}]")
            )
        rule["expanded_vlan_ids"] = expanded_vlan_ids or None
        rules.append(rule)
    configured_groups = {rule["set_vlan_group"] for rule in rules}
    if vlan_group:
        configured_groups.add(vlan_group)
    group_records = []
    if configured_groups:
        group_records = self.bulk_filter(
            nb.ipam.vlan_groups,
            name=sorted(configured_groups),
            fields="id,name,vid_ranges,scope_type,scope_id,scope",
        )
    groups_by_name = {str(group.name): group for group in group_records}
    for name in sorted(configured_groups - groups_by_name.keys()):
        message = f"vlan group '{name}' does not exist in NetBox"
        job.event(message, severity="ERROR")
        log.error(message)
        ret.errors.append(message)
    groups = {group.id: group for group in groups_by_name.values()}
    message = (
        f"validated {len(rules)} VLAN map rule(s) and loaded {len(groups)} group(s)"
    )
    job.event(message)
    log.info(message)

    # Collect all worker results before choosing names, mappings, and memberships.
    message = f"collecting live VLANs from {len(nb_devices)} device(s)"
    job.event(message)
    log.info(message)
    parsed = self.client.run_job(
        "nornir",
        "parse_ttp",
        workers="all",
        timeout=timeout,
        kwargs={"get": "vlans", "FL": sorted(nb_devices)},
    )
    live_by_device = {}
    for worker, response in parsed.items():
        resources_failed = response.get("resources_failed") or []
        if resources_failed:
            ret.resources_failed = sorted(
                set(ret.resources_failed) | set(resources_failed)
            )
            message = (
                f"{worker} failed to fetch VLAN data from devices "
                f"{', '.join(sorted(resources_failed))}"
            )
            job.event(message, severity="ERROR")
            log.error(message)
            ret.errors.append(message)
        if response.get("failed"):
            message = f"worker '{worker}' failed to collect live VLAN data"
            job.event(message, severity="ERROR")
            log.error(message)
            ret.errors.append(message)
            continue
        for device, records in response["result"].items():
            live_by_device.setdefault(device, []).extend(records)
    if not live_by_device:
        message = "no live VLAN data collected"
        job.event(message, severity="ERROR")
        log.error(message)
        ret.errors.append(message)
        ret.failed = True
        return ret

    # Produce one record per device and VID while preserving parser order.
    live_vlans = []
    interface_names_by_device = {}
    for device, records in sorted(live_by_device.items()):
        device_model = str(nb_devices[device].device_type.model)
        interface_names = {}
        device_vlans = {}
        for record in records:
            vid = record["vid"]
            if selected_vids and vid not in selected_vids:
                continue

            for field in VLAN_MEMBERSHIP_FIELDS:
                for live_interface in record[field]:
                    if live_interface in interface_names:
                        continue
                    interface_names[live_interface] = map_interface_name(
                        live_interface,
                        interface_map,
                        device,
                        device_model,
                    )

            name = record["name"].strip()
            description = (record["description"] or "").strip()
            vlan = device_vlans.setdefault(
                vid,
                {
                    "device_name": device,
                    "vid": vid,
                    "name": name,
                    "description": description,
                    "tagged_interfaces": [],
                    "untagged_interfaces": [],
                },
            )
            if (
                vlan["name"].casefold() == f"vlan{vid}".casefold()
                and name.casefold() != f"vlan{vid}".casefold()
            ):
                vlan["name"] = name
            # Prefer useful text when duplicate records from one device differ.
            if description and not vlan["description"]:
                vlan["description"] = description
            for field in VLAN_MEMBERSHIP_FIELDS:
                for name in record[field]:
                    if name not in vlan[field]:
                        vlan[field].append(name)

        for vlan in device_vlans.values():
            mapped_interfaces = []
            for field in VLAN_MEMBERSHIP_FIELDS:
                for name in vlan[field]:
                    mapped_interfaces.append(interface_names[name])
            mapped = match_vlan_map(
                rules,
                vlan["vid"],
                vlan["name"],
                device,
                mapped_interfaces,
            )
            group_name = mapped or vlan_group
            group = groups_by_name.get(group_name)
            if group_name and group is None:
                continue
            if require_vlan_group and not group_name:
                message = (
                    f"skipping VLAN {vlan['vid']} from device '{device}': "
                    "no VLAN group mapping found"
                )
                job.event(message, severity="ERROR")
                log.error(message)
                ret.errors.append(message)
                continue
            vlan["selected_group_id"] = group.id if group else None
            live_vlans.append(vlan)
        interface_names_by_device[device] = interface_names
    message = (
        f"normalized {len(live_vlans)} live VLAN record(s) from "
        f"{len(live_by_device)} device(s)"
    )
    job.event(message)
    log.info(message)

    # Fetch all same-VID candidates because scope, rather than name, selects identity.
    vids = sorted({vlan["vid"] for vlan in live_vlans})
    candidates = []
    if vids:
        candidates = self.bulk_filter(
            nb.ipam.vlans, vid=vids, fields="id,vid,name,description,site,group"
        )
    candidate_group_ids = {vlan.group.id for vlan in candidates if vlan.group}
    missing_groups = sorted(candidate_group_ids - groups.keys())
    if missing_groups:
        candidate_groups = self.bulk_filter(
            nb.ipam.vlan_groups,
            id=missing_groups,
            fields="id,name,vid_ranges,scope_type,scope_id,scope",
        )
        for group in candidate_groups:
            groups[group.id] = group

    # Resolve VLAN attributes and interface memberships from the same observations.
    scope_payloads = {}
    vlan_live = {}
    vlan_current = {}
    vlan_objects = {}
    name_sources = {}
    description_sources = {}
    interface_targets = {}
    claimed_interfaces = {}
    for resolved in resolve_live_vlans(
        live_vlans, candidates, groups, device_scopes
    ):
        observation, existing = resolved["live"], resolved["vlan"]
        device, vid = observation["device_name"], observation["vid"]
        if resolved["error"]:
            message = (
                f"vlan {vid} from device '{device}' skipped: {resolved['error']}"
            )
            if observation["selected_group_id"]:
                message += (
                    "; fix the VLAN group scope, VID ranges, or group mapping"
                )
            job.event(message, severity="ERROR")
            log.error(message)
            ret.errors.append(message)
            continue
        scope = resolved["scope"]
        scope_payloads[scope] = resolved["scope_payload"]
        key = (scope, vid)
        if key not in name_sources:
            name_sources[key] = (device, observation["name"])
            description_sources[key] = (device, observation["description"])
            vlan_live.setdefault(scope, {})[vid] = {
                "name": observation["name"],
                "description": observation["description"],
            }
            if existing:
                vlan_objects[key] = existing
                vlan_current.setdefault(scope, {})[vid] = {
                    "name": str(existing.name).strip(),
                    "description": str(existing.description or "").strip(),
                }
                vlan_live[scope][vid]["description"] = apply_description_policy(
                    observation["description"],
                    vlan_current[scope][vid]["description"],
                    preserve_description,
                )
            else:
                vlan_current.setdefault(scope, {})
        elif name_sources[key][1] != observation["name"]:
            source_device, source_name = name_sources[key]
            conflict_device, conflict_name = device, observation["name"]
            automatic_name = f"VLAN{vid}".casefold()
            source_is_automatic = source_name.casefold() == automatic_name
            conflict_is_automatic = observation["name"].casefold() == automatic_name
            # Replace an autogenerated source name with a descriptive one.
            if source_is_automatic and not conflict_is_automatic:
                vlan_live[scope][vid]["name"] = observation["name"]
                name_sources[key] = (device, observation["name"])
                conflict_device, conflict_name = source_device, source_name
                source_device = device
            # Only different descriptive names represent a genuine conflict.
            if not source_is_automatic and not conflict_is_automatic:
                message = (
                    f"{scope} VLAN {vid} source conflict (name): using VLAN name "
                    f"'{vlan_live[scope][vid]['name']}' from device "
                    f"'{source_device}'; conflicting device '{conflict_device}' "
                    f"reports VLAN name '{conflict_name}'"
                )
                if message not in ret.errors:
                    job.event(message, severity="ERROR")
                    log.error(message)
                    ret.errors.append(message)
        source_description = description_sources[key][1]
        # Let the first non-empty device description replace an empty source.
        if observation["description"] and not source_description:
            description_sources[key] = (device, observation["description"])
            netbox_description = (
                vlan_current.get(scope, {}).get(vid, {}).get("description", "")
            )
            vlan_live[scope][vid]["description"] = apply_description_policy(
                observation["description"],
                netbox_description,
                preserve_description,
            )
        # Empty descriptions are ignored; only differing useful text conflicts.
        elif (
            source_description
            and observation["description"]
            and source_description != observation["description"]
        ):
            source_device = description_sources[key][0]
            message = (
                f"{scope} VLAN {vid} source conflict (description): using live "
                f"description from device '{source_device}'; conflicting device "
                f"'{device}' reports a different description"
            )
            if message not in ret.errors:
                job.event(message, severity="ERROR")
                log.error(message)
                ret.errors.append(message)
        vlan_reference = f"{scope}/{vid}"
        for field in VLAN_MEMBERSHIP_FIELDS:
            for live_interface in observation[field]:
                interface = interface_names_by_device[device][live_interface]
                interface_key = (device, interface)
                owner = claimed_interfaces.setdefault(interface_key, live_interface)
                if owner != live_interface:
                    continue
                target = interface_targets.setdefault(
                    interface_key,
                    {"tagged_vlans": set(), "untagged_vlan": None},
                )
                if field == "tagged_interfaces":
                    target["tagged_vlans"].add(vlan_reference)
                else:
                    target["untagged_vlan"] = vlan_reference

    # Keep a descriptive NetBox name when live data only supplies VLAN<VID>.
    for scope, vid in vlan_objects:
        live_name = vlan_live[scope][vid]["name"]
        if live_name.upper() == f"VLAN{vid}".upper():
            vlan_live[scope][vid]["name"] = vlan_current[scope][vid]["name"]

    # NetBox requires VLAN names to be unique within their group or site.
    # Validate proposed creations and renames before sending a bulk write so
    # one conflict does not reject every VLAN in the batch.
    proposed_names = {
        values["name"] for vlans in vlan_live.values() for values in vlans.values()
    }
    name_matches = []
    if proposed_names:
        name_matches = self.bulk_filter(
            nb.ipam.vlans,
            name=sorted(proposed_names),
            fields="id,vid,name,site,group",
        )
    for existing in name_matches:
        if existing.group:
            scope = f"group:{existing.group.name}"
        elif existing.site:
            scope = f"site:{existing.site.name}"
        else:
            scope = "global"
        for vid, values in list(vlan_live.get(scope, {}).items()):
            if values["name"] != str(existing.name):
                continue
            current = vlan_current.get(scope, {}).get(vid)
            if current and vlan_objects[(scope, vid)].id == existing.id:
                continue
            action = "update" if current else "create"
            if current:
                vlan_live[scope][vid] = dict(current)
            else:
                del vlan_live[scope][vid]
                vlan_reference = f"{scope}/{vid}"
                for target in interface_targets.values():
                    target["tagged_vlans"].discard(vlan_reference)
                    if target["untagged_vlan"] == vlan_reference:
                        target["untagged_vlan"] = None
            message = (
                f"VLAN {vid} name '{values['name']}' overlaps with VLAN "
                f"{existing.vid} in scope '{scope}'; skipping VLAN {action}"
            )
            job.event(message, severity="ERROR")
            log.error(message)
            ret.errors.append(message)

    # Also reject duplicate names among changes not yet present in NetBox.
    for scope, vlans in vlan_live.items():
        seen_names = {}
        for vid, values in list(vlans.items()):
            name = values["name"]
            if name not in seen_names:
                seen_names[name] = vid
                continue
            current = vlan_current.get(scope, {}).get(vid)
            action = "update" if current else "create"
            if current:
                vlans[vid] = dict(current)
                seen_names[current["name"]] = vid
            else:
                del vlans[vid]
                vlan_reference = f"{scope}/{vid}"
                for target in interface_targets.values():
                    target["tagged_vlans"].discard(vlan_reference)
                    if target["untagged_vlan"] == vlan_reference:
                        target["untagged_vlan"] = None
            message = (
                f"vlan {vid} name '{name}' overlaps with VLAN "
                f"{seen_names[name]} in scope '{scope}'; skipping VLAN {action}"
            )
            job.event(message, severity="ERROR")
            log.error(message)
            ret.errors.append(message)

    message = (
        f"resolved {sum(len(vlans) for vlans in vlan_live.values())} VLAN(s) "
        f"across {len(vlan_live)} NetBox scope(s)"
    )
    job.event(message)
    log.info(message)

    vlan_diff = self.make_diff(vlan_live, vlan_current)
    vlan_preview = {}
    for scope, actions in sorted(vlan_diff.items()):
        updates = {}
        for vid, changes in actions["update"].items():
            updates[str(vid)] = changes
        create_details = {}
        for vid in actions["create"]:
            create_details[str(vid)] = vlan_live[scope][vid]
        vlan_preview[scope] = {
            **actions,
            "update": updates,
            "create_details": create_details,
        }

    # Build interface state only for interfaces referenced by live VLAN data.
    netbox_interfaces = []
    if interface_targets:
        target_devices = sorted({device for device, _ in interface_targets})
        target_names = sorted({name for _, name in interface_targets})
        target_device_ids = [nb_devices[device].id for device in target_devices]
        netbox_interfaces = self.bulk_filter(
            nb.dcim.interfaces,
            device_id=target_device_ids,
            name=target_names,
            fields="id,name,device,mode,tagged_vlans,untagged_vlan",
        )
    interfaces = {}
    for interface in netbox_interfaces:
        key = (str(interface.device.name), str(interface.name))
        if key in interface_targets:
            interfaces[key] = interface
    message = f"loaded {len(interfaces)} NetBox interface(s) for VLAN membership"
    job.event(message)
    log.info(message)
    valid_interface_keys = interface_targets.keys() & interfaces.keys()
    missing_interfaces = interface_targets.keys() - valid_interface_keys
    for device, name in sorted(missing_interfaces):
        message = f"interface '{device}:{name}' not found in NetBox"
        job.event(message, severity="ERROR")
        log.error(message)
        ret.errors.append(message)

    assigned_vlan_ids = set()
    for key in valid_interface_keys:
        interface = interfaces[key]
        assigned_vlan_ids.update(vlan.id for vlan in interface.tagged_vlans)
        if interface.untagged_vlan:
            assigned_vlan_ids.add(interface.untagged_vlan.id)
    membership_vlans = {vlan.id: vlan for vlan in candidates}
    missing_vlan_ids = assigned_vlan_ids - membership_vlans.keys()
    if missing_vlan_ids:
        assigned_vlans = self.bulk_filter(
            nb.ipam.vlans,
            id=sorted(missing_vlan_ids),
            fields="id,vid,site,group",
        )
        for vlan in assigned_vlans:
            membership_vlans[vlan.id] = vlan

    vlan_id_by_reference = {}
    vlan_reference_by_id = {}
    for vlan in membership_vlans.values():
        if vlan.group:
            scope = f"group:{vlan.group.name}"
        elif vlan.site:
            scope = f"site:{vlan.site.name}"
        else:
            scope = "global"
        vlan_reference = f"{scope}/{vlan.vid}"
        vlan_id_by_reference[vlan_reference] = vlan.id
        vlan_reference_by_id[vlan.id] = vlan_reference

    interface_live = {}
    interface_current = {}
    for device, name in sorted(valid_interface_keys):
        interface = interfaces[(device, name)]
        tagged_references = sorted(
            vlan_reference_by_id[vlan.id] for vlan in interface.tagged_vlans
        )
        untagged_reference = None
        if interface.untagged_vlan:
            untagged_reference = vlan_reference_by_id[interface.untagged_vlan.id]
        current = {
            "mode": interface.mode.value if interface.mode else None,
            "tagged_vlans": tagged_references,
            "untagged_vlan": untagged_reference,
        }
        target = interface_targets[(device, name)]
        tagged_vlans = set(current["tagged_vlans"])
        tagged_vlans.update(target["tagged_vlans"])
        untagged_vlan = current["untagged_vlan"]
        if target["untagged_vlan"] is not None:
            untagged_vlan = target["untagged_vlan"]
        mode = current["mode"]
        if tagged_vlans:
            mode = "tagged"
        elif untagged_vlan:
            mode = "access"
        interface_current.setdefault(device, {})[name] = current
        interface_live.setdefault(device, {})[name] = {
            "mode": mode,
            "tagged_vlans": sorted(tagged_vlans),
            "untagged_vlan": untagged_vlan,
        }

    message = "calculating VLAN and interface diffs"
    job.event(message)
    log.info(message)
    interface_diff = self.make_diff(interface_live, interface_current)
    preview = {"vlans": vlan_preview, "interfaces": interface_diff}
    ret.diff = preview
    ret.result = preview
    create_count = sum(len(actions["create"]) for actions in vlan_diff.values())
    vlan_update_count = sum(
        len(actions["update"]) for actions in vlan_diff.values()
    )
    interface_update_count = sum(
        len(actions["update"]) for actions in interface_diff.values()
    )
    message = (
        f"prepared {create_count} VLAN creation(s), {vlan_update_count} VLAN "
        f"update(s), and {interface_update_count} interface update(s)"
    )
    job.event(message)
    log.info(message)
    if dry_run:
        message = "dry-run requested, returning VLAN sync diff without changes"
        job.event(message)
        log.info(message)
        return ret

    vlan_result = {
        scope: SyncActionSummary(in_sync=actions["in_sync"]).model_dump()
        for scope, actions in vlan_diff.items()
    }
    interface_result = {
        device: SyncActionSummary(in_sync=actions["in_sync"]).model_dump()
        for device, actions in interface_diff.items()
    }
    ret.result = SyncVlansResultPayload(
        vlans=vlan_result, interfaces=interface_result
    ).model_dump()
    if not sync_diff_has_changes(preview):
        job.event("no VLAN sync changes required")
        return ret
    if with_approval and not review_sync_task_result(job, "vlan sync", preview):
        ret.status = "skipped"
        ret.result = preview
        ret.dry_run = True
        ret.messages.append("review declined; changes were not applied")
        return ret

    # Complete all creations before any VLAN updates or interface writes.
    create_items = []
    for scope, actions in sorted(vlan_diff.items()):
        for vid in actions["create"]:
            create_items.append(
                (
                    (scope, vid),
                    {
                        "vid": vid,
                        "name": vlan_live[scope][vid]["name"],
                        "description": vlan_live[scope][vid]["description"],
                        **scope_payloads[scope],
                    },
                )
            )
    if create_items:
        total_batches = (len(create_items) + batch_size - 1) // batch_size
        for batch_start in range(0, len(create_items), batch_size):
            batch = create_items[batch_start : batch_start + batch_size]
            batch_number = batch_start // batch_size + 1
            message = f"creating VLAN batch {batch_number}/{total_batches} ({len(batch)} VLAN(s))"
            job.event(message)
            log.info(message)
            try:
                created = nb.ipam.vlans.create([payload for _, payload in batch])
            except Exception as exc:
                message = f"failed to create VLAN batch {batch_number}/{total_batches}: {exc}"
                job.event(message, severity="ERROR")
                log.error(message)
                ret.errors.append(message)
                ret.failed = True
                return ret
            for (key, _), vlan in zip(batch, created):
                scope, vid = key
                vlan_objects[key] = vlan
                vlan_id_by_reference[f"{scope}/{vid}"] = vlan.id
                ret.result["vlans"][scope]["created"].append(vid)
        message = f"created {len(create_items)} VLAN(s)"
        job.event(message)
        log.info(message)

    vlan_updates = []
    for scope, actions in sorted(vlan_diff.items()):
        for vid, changes in actions["update"].items():
            values = {}
            for field in ("name", "description"):
                if field in changes:
                    values[field] = vlan_live[scope][vid][field]
            if values:
                vlan_updates.append(
                    (
                        (scope, vid),
                        {"id": vlan_objects[(scope, vid)].id, **values},
                    )
                )
    if vlan_updates:
        total_batches = (len(vlan_updates) + batch_size - 1) // batch_size
        for batch_start in range(0, len(vlan_updates), batch_size):
            batch = vlan_updates[batch_start : batch_start + batch_size]
            batch_number = batch_start // batch_size + 1
            message = f"updating VLAN batch {batch_number}/{total_batches} ({len(batch)} VLAN(s))"
            job.event(message)
            log.info(message)
            try:
                nb.ipam.vlans.update([payload for _, payload in batch])
            except Exception as exc:
                message = f"failed to update VLAN batch {batch_number}/{total_batches}: {exc}"
                job.event(message, severity="ERROR")
                log.error(message)
                ret.errors.append(message)
                ret.failed = True
                return ret
            for (scope, vid), _ in batch:
                ret.result["vlans"][scope]["updated"].append(vid)
        message = f"updated {len(vlan_updates)} VLAN object(s)"
        job.event(message)
        log.info(message)

    interface_updates = []
    for device, actions in sorted(interface_diff.items()):
        for name in sorted(actions["update"]):
            desired = interface_live[device][name]
            tagged_vlan_ids = [
                vlan_id_by_reference[reference]
                for reference in desired["tagged_vlans"]
            ]
            untagged_vlan_id = None
            if desired["untagged_vlan"]:
                untagged_vlan_id = vlan_id_by_reference[desired["untagged_vlan"]]
            interface_updates.append(
                (
                    (device, name),
                    {
                        "id": interfaces[(device, name)].id,
                        "mode": desired["mode"],
                        "tagged_vlans": tagged_vlan_ids,
                        "untagged_vlan": untagged_vlan_id,
                    },
                )
            )
    if interface_updates:
        total_batches = (len(interface_updates) + batch_size - 1) // batch_size
        for batch_start in range(0, len(interface_updates), batch_size):
            batch = interface_updates[batch_start : batch_start + batch_size]
            batch_number = batch_start // batch_size + 1
            message = f"updating VLAN membership batch {batch_number}/{total_batches} ({len(batch)} interface(s))"
            job.event(message)
            log.info(message)
            try:
                nb.dcim.interfaces.update([payload for _, payload in batch])
            except Exception as exc:
                message = f"failed to update VLAN membership batch {batch_number}/{total_batches}: {exc}"
                job.event(message, severity="ERROR")
                log.error(message)
                ret.errors.append(message)
                ret.failed = True
                return ret
            for (device, name), _ in batch:
                ret.result["interfaces"][device]["updated"].append(name)
        message = (
            f"updated VLAN membership on {len(interface_updates)} interface(s)"
        )
        job.event(message)
        log.info(message)
    message = "vlan sync complete"
    job.event(message)
    log.info(message)
    return ret