Netbox Sync Device Inventory Task

task api name: sync_device_inventory

The sync_device_inventory task reconciles live hardware inventory collected from the Nornir service with NetBox device and module data.

Live inventory is collected with Nornir parse_ttp using get="inventory". The chassis inventory record updates dcim.device.serial. Non-chassis inventory records, including optics and transceivers, are managed as NetBox dcim.modules installed in device-level dcim.module_bays.

Parsed Inventory Records

The Nornir inventory parser and optional inventory transformer must return one list of records per device. Every record has exactly these fields:

- description: "ASR9K Route Switch Processor with 440G/slot Fabric and 6GB"
  slot: "module 0/RSP0/CPU0"
  module: "A9K-RSP440-TR"
  serial: "M9YXCZV9QF"

Each field value must be a string or null.

Device Serial Number

A record whose trimmed slot value is chassis, case-insensitive, is the device chassis record:

- description: "Cisco ASR 9006 Router"
  slot: "chassis"
  module: "ASR-9006"
  serial: "JCY98XR393D"

For this record, the task:

1. reads serial as the desired NetBox dcim.device.serial;
2. creates a synthetic chassis entry in the inventory diff;
3. updates the NetBox device serial when the live and NetBox values differ;
4. does not create a module bay, module type, or installed module for it.

The chassis module and description values do not participate in the device serial comparison. An empty or BUILTIN chassis serial is ignored. Multiple chassis records with different non-empty serial numbers produce a per-device error and that device is skipped.

Safety Controls

Module deletions are disabled by default. Set process_deletions=True to delete NetBox modules that are absent from live network inventory.

Missing module bays and module types are reported by default. Set create_module_bays=True or create_module_types=True when the task should extend NetBox modeling from live inventory.

Live records with empty module identity, empty serial numbers, or BUILTIN serial numbers are skipped and reported in res["errors"]. Skipped slots suppress deletion so incomplete live data does not remove existing NetBox modules.

Filtering Modules

The task can select or exclude modules using lists of case-sensitive glob patterns:

• filter_by_module: include normalized module type names matching any pattern.
• filter_by_slot: include normalized module bay names matching any pattern.
• ignore_modules: exclude normalized module type names matching any pattern.
• ignore_slots: exclude normalized module bay names matching any pattern.

Filters run after the optional Python transformer, inventory name mapping, and built-in live-data normalization. A pattern therefore matches the final module_type and slot values used for NetBox comparison, not necessarily the raw parser values.

Patterns within one argument use OR logic. When both include filters are provided, a module must match both dimensions. Ignore patterns take precedence: a match in either ignore list excludes the module.

The chassis record is never filtered, so device serial synchronization remains independent from module selection. The same comparison scope is applied to existing NetBox modules. A live slot excluded by a filter is also excluded from the NetBox side, preventing it from being treated as stale when process_deletions=True.

Example:

filter_by_module:
  - "A9K-*"
  - "SFP-10G-*"
filter_by_slot:
  - "module 0/*"
ignore_modules:
  - "*-BUILTIN"
ignore_slots:
  - "module 0/PS*"

Inventory Name Mapping

Use inventory_map when live module or slot names differ from existing NetBox module type models and module bay names. Supply either an inline mapping or an nf:// reference to a YAML file available through the NorFab File Service. The mapping is applied after an optional Python transformer and before inventory normalization and diff calculation.

module_types:
  Cisco:
    "ASR 9000 RSP440":
      - glob: "A9K-RSP440-*"
    "10GBASE-LR SFP+":
      - regex: "^SFP-10G-LR(=)?$"
    "ASR 9000 Fan Tray":
      - eval: "value == 'ASR-9006-FAN-V2'"

module_bays:
  Cisco:
    "ASR-9006":
      "0/RSP0":
        - glob: "module 0/RSP0/*"
      "0/1":
        - regex: "^module 0/1(/CPU0)?$"
      "TenGigE0/2/CPU0/0":
        - eval: "value == 'module mau TenGigE0/2/CPU0/0'"

File reference example:

inventory_map: "nf://netbox/inventory_maps/iosxr.yaml"

The downloaded YAML is parsed with yaml.safe_load and validated with the same Pydantic model used for inline mappings. Invalid YAML, invalid mapping conditions, and missing files fail the task before any NetBox writes.

Mapping Scope

• module_types is selected using the exact, case-sensitive NetBox device manufacturer name.
• A module type target key is the desired NetBox module type model.
• Module type conditions test the live record's module value.
• module_bays is selected using the exact NetBox manufacturer name followed by the exact NetBox device type model.
• A module bay target key is the desired NetBox module bay name.
• Module bay conditions test the live record's slot value.
• Chassis records are not mapped.

Each condition contains exactly one matcher:

• glob uses case-sensitive glob matching.
• regex performs a full regular-expression match.
• eval evaluates trusted Python configuration with the live string available as value.

Conditions under one target use OR logic. If one target matches, its key replaces the live value. If nothing matches, the original value is retained. If multiple targets match, the record is skipped and the ambiguity is added to res["errors"].

Warning

eval is trusted executable configuration, not a security sandbox. Only accept inventory mappings from trusted sources.

Python Inventory Transformer

Use inventory_transform for normalization that cannot be expressed with pattern mappings. Its value is an nf:// reference to a Python file available through the NorFab File Service:

inventory_transform: "nf://netbox/inventory_transformers/iosxr.py"

The file must export a function named transform with this contract:

def transform(
    device_name: str,
    parsed_data: list,
    worker: object,
    device_platform: str | None = None,
    device_manufacturer: str | None = None,
    device_type: str | None = None,
) -> list[dict]:
    """Normalize parsed inventory records for one device."""
    transformed_data = []

    for record in parsed_data:
        slot = record["slot"]

        if slot.startswith("module mau "):
            record["slot"] = slot.removeprefix("module mau ")

        if (
            device_manufacturer == "Cisco"
            and device_type == "ASR-9006"
            and record["module"] == "A9K-RSP440-TR"
        ):
            record["module"] = "ASR 9000 RSP440"

        transformed_data.append(record)

    return transformed_data

The arguments are:

• device_name: current NetBox and Nornir device name.
• parsed_data: parsed records for that device, list of dictionaries with slot, serial, description, module keys.
• worker: active NetBox worker object, including its configuration and all worker helper methods.
• device_platform: NetBox device platform name, or None when no platform is assigned.
• device_manufacturer: NetBox device type manufacturer name.
• device_type: NetBox device type model.

The return value must be a list of dictionaries. Every dictionary must contain exactly description, slot, module, and serial, with each value set to a string or None. Returning an empty list means there is no usable inventory for that device. Invalid output or an exception skips that device and records the validation or execution error in result's errors.

The transformer is loaded once per task and called once per device. It runs before inventory_map, so pattern conditions test the transformed module and slot values. The device metadata arguments are read from NetBox before the transformer runs, so they describe the current NetBox device, not the live inventory record.

Warning

Transformer files are executed as trusted Python code. Store them only in controlled File Service locations.

Branching Support

This task is branch aware and can push updates to a NetBox branch when the NetBox Branching plugin is installed. Use the branch argument to target a branch.

Result Structure

Dry-run mode (dry_run=True) returns the raw diff without writing to NetBox:

{
    "<device>": {
        "create": ["module 0/RSP0/CPU0", "module mau 0/1/0/0"],
        "update": {
            "chassis": {
                "serial": {"old_value": "OLD123", "new_value": "JCY98XR393D"}
            }
        },
        "delete": ["module stale"],
        "in_sync": []
    }
}

Live-run mode (dry_run=False, default) applies changes and returns a per-device action summary:

{
    "<device>": {
        "created": [
            "Cisco A9K-RSP440-TR",
            "Cisco SFP-10G-LR",
            "module 0/RSP0/CPU0",
            "module mau 0/1/0/0"
        ],
        "updated": ["chassis"],
        "deleted": [],
        "in_sync": []
    }
}

The created list includes created module bays, module types, and installed modules. In live-run mode res["diff"] is also populated with the raw diff. Missing module bays, missing module types, failed writes, and ignored live records are reported in res["errors"].

Examples

NFCLIPython

Preview chassis serial and module changes for one device:

nf#netbox sync device-inventory devices iosxr1 dry-run

Sync a device using only existing NetBox module bays and module types:

nf#netbox sync device-inventory devices iosxr1

Create missing module bays from live slot names, but require module types to already exist:

nf#netbox sync device-inventory devices iosxr1 create-module-bays

Create missing module bays and module types, then install modules:

nf#netbox sync device-inventory devices iosxr1 create-module-bays create-module-types

Sync multiple devices:

nf#netbox sync device-inventory devices iosxr1 iosxr2 create-module-bays create-module-types

Delete stale NetBox modules that are absent from live inventory:

nf#netbox sync device-inventory devices iosxr1 create-module-bays create-module-types process-deletions

Sync into a NetBox branch:

nf#netbox sync device-inventory devices iosxr1 branch inventory-sync-branch create-module-bays create-module-types

Add a NetBox changelog message to write operations:

nf#netbox sync device-inventory devices iosxr1 message "sync inventory from live device" create-module-bays create-module-types

Use Nornir host filters instead of explicit device names:

nf#netbox sync device-inventory FC iosxr create-module-bays create-module-types

Target a specific NetBox worker and keep detailed output:

nf#netbox sync device-inventory workers netbox-worker-1 devices iosxr1 verbose-result

Normalize parsed inventory with a Python transformer:

nf#netbox sync device-inventory devices iosxr1 inventory-transform nf://netbox/inventory_transformers/iosxr.py dry-run

Load inventory mappings from a YAML file:

nf#netbox sync device-inventory devices iosxr1 inventory-map nf://netbox/inventory_maps/iosxr.yaml dry-run

Sync only RSP and line-card modules in slots below module 0:

nf#netbox sync device-inventory devices iosxr1 filter-by-module "A9K-RSP*" "A9K-MOD*" filter-by-slot "module 0/*" dry-run

Ignore optics and power-module slots:

nf#netbox sync device-inventory devices iosxr1 ignore-modules "SFP-*" ignore-slots "power-module *" dry-run

from norfab.core.nfapi import NorFab

nf = NorFab(inventory="./inventory.yaml")
nf.start()
client = nf.make_client()

# dry run - preview raw create/update/delete/in_sync diff
result = client.run_job(
    "netbox",
    "sync_device_inventory",
    workers="any",
    kwargs={
        "devices": ["iosxr1"],
        "dry_run": True,
    },
)

# sync using existing NetBox module bays and module types only
result = client.run_job(
    "netbox",
    "sync_device_inventory",
    workers="any",
    kwargs={
        "devices": ["iosxr1"],
    },
)

# create missing module bays from live inventory slot names
result = client.run_job(
    "netbox",
    "sync_device_inventory",
    workers="any",
    kwargs={
        "devices": ["iosxr1"],
        "create_module_bays": True,
    },
)

# create missing module bays and module types, then install modules
result = client.run_job(
    "netbox",
    "sync_device_inventory",
    workers="any",
    kwargs={
        "devices": ["iosxr1"],
        "create_module_bays": True,
        "create_module_types": True,
    },
)

# sync multiple devices
result = client.run_job(
    "netbox",
    "sync_device_inventory",
    workers="any",
    kwargs={
        "devices": ["iosxr1", "iosxr2"],
        "create_module_bays": True,
        "create_module_types": True,
    },
)

# delete stale NetBox modules absent from live inventory
result = client.run_job(
    "netbox",
    "sync_device_inventory",
    workers="any",
    kwargs={
        "devices": ["iosxr1"],
        "create_module_bays": True,
        "create_module_types": True,
        "process_deletions": True,
    },
)

# sync into a NetBox branch and attach a changelog message
result = client.run_job(
    "netbox",
    "sync_device_inventory",
    workers="any",
    kwargs={
        "devices": ["iosxr1"],
        "branch": "inventory-sync-branch",
        "message": "sync inventory from live device",
        "create_module_bays": True,
        "create_module_types": True,
    },
)

# use Nornir host filters instead of explicit device names
result = client.run_job(
    "netbox",
    "sync_device_inventory",
    workers="any",
    kwargs={
        "FC": "iosxr",
        "create_module_bays": True,
        "create_module_types": True,
    },
)

# filter and ignore normalized module type and slot names
result = client.run_job(
    "netbox",
    "sync_device_inventory",
    workers="any",
    kwargs={
        "devices": ["iosxr1"],
        "filter_by_module": ["A9K-RSP*", "A9K-MOD*"],
        "filter_by_slot": ["module 0/*"],
        "ignore_modules": ["*-BUILTIN"],
        "ignore_slots": ["module 0/PS*"],
        "dry_run": True,
    },
)

# map live module and slot names to existing NetBox names
inventory_map = {
    "module_types": {
        "Cisco": {
            "ASR 9000 RSP440": [
                {"glob": "A9K-RSP440-*"},
            ],
            "10GBASE-LR SFP+": [
                {"regex": r"^SFP-10G-LR(=)?$"},
            ],
        },
    },
    "module_bays": {
        "Cisco": {
            "ASR-9006": {
                "0/RSP0": [
                    {"glob": "module 0/RSP0/*"},
                ],
                "TenGigE0/2/CPU0/0": [
                    {
                        "eval": (
                            "value == "
                            "'module mau TenGigE0/2/CPU0/0'"
                        )
                    },
                ],
            },
        },
    },
}
result = client.run_job(
    "netbox",
    "sync_device_inventory",
    workers="any",
    kwargs={
        "devices": ["iosxr1"],
        "inventory_map": inventory_map,
        "dry_run": True,
    },
)

# load the same mapping structure from the NorFab File Service
result = client.run_job(
    "netbox",
    "sync_device_inventory",
    workers="any",
    kwargs={
        "devices": ["iosxr1"],
        "inventory_map": "nf://netbox/inventory_maps/iosxr.yaml",
        "dry_run": True,
    },
)

# run a transformer, then apply mappings to its returned records
result = client.run_job(
    "netbox",
    "sync_device_inventory",
    workers="any",
    kwargs={
        "devices": ["iosxr1"],
        "inventory_transform": (
            "nf://netbox/inventory_transformers/iosxr.py"
        ),
        "inventory_map": inventory_map,
        "dry_run": True,
    },
)

nf.destroy()

Inline inventory_map data is most conveniently supplied through the Python API or workflow task kwargs. NFCLI supports the file-backed inventory-map and inventory-transform references shown above.

NORFAB Netbox Sync Device Inventory Command Shell Reference

NorFab shell supports these command options for the NetBox sync_device_inventory task:

nf# man tree netbox.sync.device-inventory
root
└── netbox:    Netbox service
    └── sync:    Sync Netbox data
        └── device-inventory:    Sync device inventory facts e.g. serial number
            ├── timeout:    Job timeout
            ├── workers:    Filter worker to target, default 'any'
            ├── verbose-result:    Control output details, default 'False'
            ├── nowait:    Do not wait for job to complete, default 'False'
            ├── instance:    Netbox instance name to target
            ├── dry-run:    Do not commit to database
            ├── branch:    NetBox branching plugin branch name to use
            ├── devices:    List of NetBox devices to sync inventory for
            ├── process-deletions:    Delete NetBox modules present in module bays but absent from live inventory
            ├── create-module-types:    Create missing NetBox module types from live inventory model data
            ├── create-module-bays:    Create missing NetBox module bays using the live inventory slot names
            ├── inventory-map:    Pattern mappings or nf:// YAML file reference
            ├── inventory-transform:    nf:// Python transformer file containing a transform function
            ├── filter-by-module:    Glob patterns selecting normalized module type names
            ├── filter-by-slot:    Glob patterns selecting normalized module bay names
            ├── ignore-modules:    Glob patterns excluding normalized module type names
            ├── ignore-slots:    Glob patterns excluding normalized module bay names
            ├── message:    Changelog message recorded on NetBox writes
            ├── 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
nf#

Python API Reference

Synchronize chassis serial and installed module inventory into NetBox.

Live inventory is collected from the Nornir service using parse_ttp(get="inventory"). The chassis record updates the NetBox device serial. All non-chassis records with a usable module identity and serial are reconciled as NetBox modules installed in device-level module bays.

Module deletions are disabled by default. Missing module bays and module types are reported unless explicitly created with create_module_bays or create_module_types.

Parameters:

Name	Type	Description	Default
job	Job	NorFab job object used for progress events.	required
instance	Union[None, str]	NetBox instance name to target. Defaults to the worker's default NetBox instance.	None
dry_run	bool	Return the planned inventory diff without writing changes to NetBox.	False
timeout	int	Timeout in seconds for the Nornir parse_ttp job.	60
devices	Union[None, list]	NetBox device names to synchronize.	None
branch	str	NetBox Branching plugin branch name.	None
process_deletions	bool	Delete stale NetBox modules when they are absent from live inventory.	False
create_module_types	bool	Create missing NetBox module types from live module model data.	False
create_module_bays	bool	Create missing NetBox module bays from live inventory slot names.	False
inventory_map	Union[None, str, dict, InventoryPatternMap]	Manufacturer and device type scoped pattern mappings, or an nf:// YAML file containing them.	None
inventory_transform	Union[None, str]	nf:// Python file containing transform(device_name, parsed_data, worker, **context). Supported context keyword arguments are device_platform, device_manufacturer, and device_type from NetBox.	None
filter_by_module	Union[None, list]	Glob patterns selecting normalized module type names.	None
filter_by_slot	Union[None, list]	Glob patterns selecting normalized module bay names.	None
ignore_modules	Union[None, list]	Glob patterns excluding normalized module type names.	None
ignore_slots	Union[None, list]	Glob patterns excluding normalized module bay names.	None
message	Union[None, str]	NetBox changelog message to attach to write operations.	None
**kwargs	Any	Additional Nornir filters and execution options forwarded to parse_ttp.	{}

Returns:

Type	Description
Result	Result object containing the normalized diff, per-device sync
Result	outcome, and any partial errors.

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

@Task(
    input=SyncDeviceInventoryInput,
    output=SyncDeviceInventoryResult,
    fastapi={"methods": ["PATCH"], "schema": NetboxFastApiArgs.model_json_schema()},
    mcp={
        "annotations": {
            "title": "Sync Device Inventory",
            "readOnlyHint": False,
            "destructiveHint": True,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def sync_device_inventory(
    self,
    job: Job,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    timeout: int = 60,
    devices: Union[None, list] = None,
    branch: str = None,
    process_deletions: bool = False,
    create_module_types: bool = False,
    create_module_bays: bool = False,
    inventory_map: Union[None, str, dict, InventoryPatternMap] = None,
    inventory_transform: Union[None, str] = None,
    filter_by_module: Union[None, list] = None,
    filter_by_slot: Union[None, list] = None,
    ignore_modules: Union[None, list] = None,
    ignore_slots: Union[None, list] = None,
    message: Union[None, str] = None,
    **kwargs: Any,
) -> Result:
    """Synchronize chassis serial and installed module inventory into NetBox.

    Live inventory is collected from the Nornir service using
    ``parse_ttp(get="inventory")``. The chassis record updates the NetBox
    device serial. All non-chassis records with a usable module identity
    and serial are reconciled as NetBox modules installed in device-level
    module bays.

    Module deletions are disabled by default. Missing module bays and
    module types are reported unless explicitly created with
    ``create_module_bays`` or ``create_module_types``.

    Args:
        job: NorFab job object used for progress events.
        instance: NetBox instance name to target. Defaults to the worker's
            default NetBox instance.
        dry_run: Return the planned inventory diff without writing changes
            to NetBox.
        timeout: Timeout in seconds for the Nornir ``parse_ttp`` job.
        devices: NetBox device names to synchronize.
        branch: NetBox Branching plugin branch name.
        process_deletions: Delete stale NetBox modules when they are absent
            from live inventory.
        create_module_types: Create missing NetBox module types from live
            module model data.
        create_module_bays: Create missing NetBox module bays from live
            inventory slot names.
        inventory_map: Manufacturer and device type scoped pattern mappings,
            or an ``nf://`` YAML file containing them.
        inventory_transform: ``nf://`` Python file containing
            ``transform(device_name, parsed_data, worker, **context)``.
            Supported context keyword arguments are ``device_platform``,
            ``device_manufacturer``, and ``device_type`` from NetBox.
        filter_by_module: Glob patterns selecting normalized module type
            names.
        filter_by_slot: Glob patterns selecting normalized module bay
            names.
        ignore_modules: Glob patterns excluding normalized module type
            names.
        ignore_slots: Glob patterns excluding normalized module bay names.
        message: NetBox changelog message to attach to write operations.
        **kwargs: Additional Nornir filters and execution options forwarded
            to ``parse_ttp``.

    Returns:
        Result object containing the normalized diff, per-device sync
        outcome, and any partial errors.
    """
    devices = devices or []
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:sync_device_inventory",
        result={},
        resources=[instance],
        dry_run=dry_run,
        diff={},
    )
    if self.is_url(inventory_map):
        inventory_map = yaml.safe_load(
            self.fetch_file(inventory_map, raise_on_fail=True)
        )
    inventory_patterns = InventoryPatternMap.model_validate(inventory_map or {})
    transform_function = None
    if self.is_url(inventory_transform):
        function_text = self.fetch_file(inventory_transform, raise_on_fail=True)
        globals_dict = {}
        exec(function_text, globals_dict, globals_dict)  # nosec B102
        transform_function = globals_dict["transform"]

    nb = self._get_pynetbox(instance, branch=branch)

    if message:
        job.event("setting NetBox changelog message for inventory sync")
        nb.http_session.headers["X-Changelog-Message"] = message

    if kwargs:
        job.event("resolving devices from Nornir filters")
        nornir_hosts = self.get_nornir_hosts(kwargs, timeout)
        for host in nornir_hosts:
            if host not in devices:
                devices.append(host)
        job.event(
            f"resolved {len(nornir_hosts)} device(s) from Nornir filters, "
            f"{len(devices)} total device(s) selected"
        )

    if not devices:
        msg = "no devices specified"
        job.event(msg, severity="ERROR")
        ret.errors.append(msg)
        ret.failed = True
        return ret

    devices = sorted(set(devices))
    log.info(
        f"{self.name} - Sync device inventory for {len(devices)} device(s) in '{instance}', dry_run={dry_run}"
    )
    job.event(
        f"syncing device inventory for {len(devices)} device(s), dry_run={dry_run}"
    )

    # Fetch and validate NetBox devices.
    job.event(f"validating {len(devices)} device(s) exist in NetBox")
    nb_devices_data = {}
    for device in self.bulk_filter(
        nb.dcim.devices,
        "name",
        devices,
        fields="id,name,serial,device_type,platform",
    ):
        device_type = device.device_type
        manufacturer = device_type.manufacturer
        nb_devices_data[device.name] = {
            "id": int(device.id),
            "name": device.name,
            "serial": str(device.serial or ""),
            "manufacturer": manufacturer.name,
            "device_type": device_type.model,
            "platform": device.platform.name if device.platform else None,
        }

    for device_name in list(devices):
        if device_name not in nb_devices_data:
            msg = f"{device_name} - device not found in Netbox"
            log.error(msg)
            job.event(msg, severity="ERROR")
            ret.errors.append(msg)
            devices.remove(device_name)

    if not devices:
        job.event(
            "no valid NetBox devices remain after validation", severity="ERROR"
        )
        ret.failed = True
        return ret
    job.event(f"validated {len(devices)} device(s) in NetBox")

    # Fetch current module bays and installed modules.
    job.event("fetching current module bay data from NetBox")
    nb_module_bays = {device_name: {} for device_name in devices}
    for module_bay in self.bulk_filter(nb.dcim.module_bays, "device", devices):
        device_name = module_bay.device.name
        bay_name = module_bay.name
        nb_module_bays[device_name][bay_name] = {
            "id": int(module_bay.id),
            "name": bay_name,
        }

    job.event("fetching installed module data from NetBox")
    nb_modules = {device_name: {} for device_name in devices}
    nb_module_ids = {device_name: {} for device_name in devices}
    for module in self.bulk_filter(nb.dcim.modules, "device", devices):
        device_name = module.device.name
        bay_name = module.module_bay.name
        nb_modules[device_name][bay_name] = normalise_netbox_module(module)
        nb_module_ids[device_name][bay_name] = int(module.id)

    # Collect live inventory data from Network.
    job.event(f"retrieving live inventory for {len(devices)} device(s)")
    nornir_kwargs = dict(kwargs)
    nornir_kwargs["get"] = "inventory"
    nornir_kwargs["strict"] = False
    if devices:
        nornir_kwargs["FL"] = devices
    parse_data = self.client.run_job(
        "nornir",
        "parse_ttp",
        workers="all",
        timeout=timeout,
        kwargs=nornir_kwargs,
    )
    if parse_data is None:
        msg = "nornir parse_ttp inventory returned no data"
        ret.errors.append(msg)
        ret.failed = True
        job.event(msg, severity="ERROR")
        return ret

    live_records_by_device = {device_name: [] for device_name in devices}
    parse_worker_errors = []
    for worker_name, worker_data in parse_data.items():
        if worker_data["failed"]:
            msg = f"{worker_name} - failed to parse inventory data from devices"
            parse_worker_errors.append(msg)
            log.warning(f"{msg}: {worker_data['errors']}")
            job.event(msg, severity="WARNING")
            continue
        for device_name, host_inventory in worker_data["result"].items():
            if device_name not in live_records_by_device:
                continue
            live_records_by_device[device_name].extend(host_inventory)

    # Normalize live and NetBox state.
    job.event("normalising live and NetBox inventory data")
    normalised_live_all = {}
    normalised_nb_all = {}
    skipped_slots_by_device = {}
    ignored_error_messages = set()

    for device_name in devices:
        records = live_records_by_device[device_name]
        if transform_function:
            try:
                device_data = nb_devices_data[device_name]
                records = DeviceInventoryRecords.model_validate(
                    transform_function(
                        device_name=device_name,
                        parsed_data=records,
                        worker=self,
                        device_platform=device_data["platform"],
                        device_manufacturer=device_data["manufacturer"],
                        device_type=device_data["device_type"],
                    )
                ).model_dump()
            except Exception as exc:
                msg = (
                    f"{device_name} - inventory transformer failed, "
                    f"skipping device: {exc}"
                )
                ret.errors.append(msg)
                log.error(msg)
                job.event(msg, severity="ERROR")
                continue

        if not records:
            msg = f"{device_name} - parsing returned no inventory data, skipping device"
            ret.errors.append(msg)
            log.error(msg)
            job.event(msg, severity="ERROR")
            continue

        live_state = {}
        skipped_slots = set()
        chassis_serials = set()
        manufacturer = nb_devices_data[device_name]["manufacturer"]
        device_type = nb_devices_data[device_name]["device_type"]
        module_type_targets = inventory_patterns.module_types.get(
            manufacturer,
            {},
        )
        module_bay_targets = inventory_patterns.module_bays.get(
            manufacturer, {}
        ).get(
            device_type,
            {},
        )

        for record in records:
            raw_slot = str(record["slot"] or "").strip()
            raw_module_name = str(record["module"] or "").strip()
            slot = raw_slot
            module_name = raw_module_name
            serial = str(record["serial"] or "").strip()
            description = str(record["description"] or "")
            slot_lower = slot.lower()
            is_chassis = slot_lower == "chassis"

            if not is_chassis:
                try:
                    module_type_matches = find_inventory_pattern_matches(
                        module_name,
                        module_type_targets,
                    )
                    module_bay_matches = find_inventory_pattern_matches(
                        slot,
                        module_bay_targets,
                    )
                except ValueError as exc:
                    skipped_slots.add(slot)
                    msg = (
                        f"{device_name}:{slot or 'unknown'} - "
                        f"inventory mapping failed: {exc}"
                    )
                    ret.errors.append(msg)
                    log.error(msg)
                    job.event(msg, severity="ERROR")
                    continue

                if len(module_bay_matches) > 1:
                    skipped_slots.add(slot)
                    skipped_slots.update(module_bay_matches)
                    msg = (
                        f"{device_name}:{slot or 'unknown'} - live slot "
                        f"matches multiple NetBox module bays: "
                        f"{sorted(module_bay_matches)}"
                    )
                    ret.errors.append(msg)
                    log.error(msg)
                    job.event(msg, severity="ERROR")
                    continue
                if module_bay_matches:
                    slot = module_bay_matches[0]

                if len(module_type_matches) > 1:
                    skipped_slots.add(slot)
                    msg = (
                        f"{device_name}:{raw_slot or 'unknown'} - live module "
                        f"'{module_name}' matches multiple NetBox module types: "
                        f"{sorted(module_type_matches)}"
                    )
                    ret.errors.append(msg)
                    log.error(msg)
                    job.event(msg, severity="ERROR")
                    continue
                if module_type_matches:
                    module_name = module_type_matches[0]

            ignored_slot = f"{raw_slot} -> {slot}" if raw_slot != slot else slot
            ignored_reason = ""

            if serial.upper() == "BUILTIN":
                serial = ""

            if is_chassis:
                if serial:
                    chassis_serials.add(serial)
                else:
                    ignored_slot = slot or "chassis"
                    ignored_reason = "chassis serial is empty"
            elif not slot:
                ignored_reason = "slot is empty"
            elif not module_name or module_name.upper() in {"N/A", "NA", "NONE"}:
                ignored_reason = "module identity is empty or N/A"
                skipped_slots.add(slot)
            elif not serial:
                ignored_reason = "serial is empty or BUILTIN"
                skipped_slots.add(slot)

            if ignored_reason:
                msg = (
                    f"{device_name}:{ignored_slot or 'unknown'} - "
                    f"ignored inventory record, {ignored_reason}"
                )
                if msg not in ignored_error_messages:
                    ignored_error_messages.add(msg)
                    ret.errors.append(msg)
                    log.warning(msg)
                    job.event(msg, severity="WARNING")
                continue

            if is_chassis:
                continue

            live_state[slot] = {
                "slot": slot,
                "inventory_type": "module",
                "manufacturer": manufacturer,
                "module_type": module_name,
                "serial": serial,
                "description": description,
                "status": "active",
            }

        skipped_slots_by_device[device_name] = skipped_slots

        if len(chassis_serials) > 1:
            msg = f"{device_name} - multiple chassis records found, skipping device"
            ret.errors.append(msg)
            log.error(msg)
            job.event(msg, severity="ERROR")
            continue
        if chassis_serials:
            live_state["chassis"] = {
                "slot": "chassis",
                "inventory_type": "chassis",
                "serial": next(iter(chassis_serials)),
            }

        if "chassis" not in live_state:
            msg = f"{device_name} - no chassis serial found"
            ret.errors.append(msg)
            log.warning(msg)
            job.event(msg, severity="WARNING")

        normalised_live_all[device_name] = live_state
        normalised_nb_all[device_name] = dict(nb_modules[device_name])
        if "chassis" in live_state:
            normalised_nb_all[device_name]["chassis"] = {
                "slot": "chassis",
                "inventory_type": "chassis",
                "serial": nb_devices_data[device_name]["serial"],
            }

    for device_name, live_state in normalised_live_all.items():
        excluded_live_slots = {
            slot
            for slot, record in live_state.items()
            if not inventory_record_matches_filters(
                record,
                filter_by_module=filter_by_module,
                filter_by_slot=filter_by_slot,
                ignore_modules=ignore_modules,
                ignore_slots=ignore_slots,
            )
        }
        selected_live_slots = set(live_state) - excluded_live_slots

        normalised_live_all[device_name] = {
            slot: record
            for slot, record in live_state.items()
            if slot in selected_live_slots
        }
        normalised_nb_all[device_name] = {
            slot: record
            for slot, record in normalised_nb_all[device_name].items()
            if slot not in excluded_live_slots
            and (
                slot in selected_live_slots
                or inventory_record_matches_filters(
                    record,
                    filter_by_module=filter_by_module,
                    filter_by_slot=filter_by_slot,
                    ignore_modules=ignore_modules,
                    ignore_slots=ignore_slots,
                )
            )
        }

    if not normalised_live_all:
        msg = "no inventory parsing results collected for devices"
        ret.errors.append(msg)
        ret.errors.extend(parse_worker_errors)
        ret.failed = True
        job.event(msg, severity="ERROR")
        return ret
    ret.messages.extend(parse_worker_errors)

    # Diff desired live state against current NetBox state.
    job.event("calculating inventory sync diff")
    full_diff = self.make_diff(normalised_live_all, normalised_nb_all)

    # Suppress deletes for slots where live inventory existed but was incomplete.
    for device_name, skipped_slots in skipped_slots_by_device.items():
        if device_name not in full_diff:
            continue

        safe_delete_slots = []
        for slot in full_diff[device_name]["delete"]:
            if slot in skipped_slots:
                continue
            safe_delete_slots.append(slot)
        full_diff[device_name]["delete"] = safe_delete_slots

    create_count = 0
    update_count = 0
    delete_count = 0
    in_sync_count = 0
    for actions in full_diff.values():
        create_count += len(actions["create"])
        update_count += len(actions["update"])
        delete_count += len(actions["delete"])
        in_sync_count += len(actions["in_sync"])
    job.event(
        "inventory sync diff complete: "
        f"{create_count} create, {update_count} update, "
        f"{delete_count} delete, {in_sync_count} in sync"
    )

    if dry_run is True:
        job.event(
            "dry-run requested, returning inventory sync diff without changes"
        )
        ret.result = full_diff
        ret.dry_run = True
        return ret
    else:
        ret.diff = full_diff

    missing_module_bay_keys = set()
    if not create_module_bays:
        job.event("validating module bays before module writes")
        for device_name, actions in full_diff.items():
            for slot in actions["create"]:
                if slot == "chassis":
                    continue
                if slot in nb_module_bays[device_name]:
                    continue
                missing_module_bay_keys.add((device_name, slot))
                msg = f"{device_name}:{slot} - module bay not found"
                ret.errors.append(msg)
                log.error(msg)
                job.event(msg, severity="ERROR")

    module_type_lookup_cache = {}
    missing_module_type_keys = set()
    required_module_types = {}
    for device_name, actions in full_diff.items():
        required_slots = set(actions["create"])
        for slot, changes in actions["update"].items():
            if "module_type" in changes or "manufacturer" in changes:
                required_slots.add(slot)

        for slot in required_slots:
            if slot == "chassis":
                continue
            desired = normalised_live_all[device_name][slot]
            lookup_key = (slugify(desired["manufacturer"]), desired["module_type"])
            module_type_data = required_module_types.setdefault(
                lookup_key,
                {
                    "manufacturer": desired["manufacturer"],
                    "model": desired["module_type"],
                    "locations": set(),
                },
            )
            module_type_data["locations"].add(f"{device_name}:{slot}")

    module_type_names = sorted(
        module_type_data["model"]
        for module_type_data in required_module_types.values()
    )
    if module_type_names:
        job.event(
            "validating module types by model and part number before module writes"
        )
        for module_type in self.bulk_filter(
            nb.dcim.module_types,
            "model",
            module_type_names,
        ):
            module_type_id = int(module_type.id)
            manufacturer_slug = module_type.manufacturer.slug
            module_type_lookup_cache[(manufacturer_slug, module_type.model)] = (
                module_type_id
            )
            if module_type.part_number:
                module_type_lookup_cache[
                    (manufacturer_slug, module_type.part_number)
                ] = module_type_id

        for module_type in self.bulk_filter(
            nb.dcim.module_types,
            "part_number",
            module_type_names,
        ):
            module_type_id = int(module_type.id)
            manufacturer_slug = module_type.manufacturer.slug
            module_type_lookup_cache[(manufacturer_slug, module_type.model)] = (
                module_type_id
            )
            if module_type.part_number:
                module_type_lookup_cache[
                    (manufacturer_slug, module_type.part_number)
                ] = module_type_id

    if not create_module_types:
        for lookup_key, module_type_data in required_module_types.items():
            if module_type_lookup_cache.get(lookup_key):
                continue
            missing_module_type_keys.add(lookup_key)
            locations = ", ".join(sorted(module_type_data["locations"]))
            module_type_label = (
                f"{module_type_data['manufacturer']} {module_type_data['model']}"
            )
            msg = f"{locations} - module type '{module_type_label}' not found"
            ret.errors.append(msg)
            log.error(msg)
            job.event(msg, severity="ERROR")

    device_results = {}
    for device_name, actions in full_diff.items():
        device_results[device_name] = {
            "created": [],
            "updated": [],
            "deleted": [],
            "in_sync": actions["in_sync"],
        }
    ret.result = device_results

    # Device serial updates from the synthetic chassis slot.
    job.event("applying chassis serial updates")
    for device_name, actions in full_diff.items():
        chassis_diff = actions["update"].get("chassis", {})
        if "serial" not in chassis_diff:
            continue
        new_serial = chassis_diff["serial"]["new_value"]
        if not new_serial:
            continue
        try:
            nb.dcim.devices.update(
                [{"id": nb_devices_data[device_name]["id"], "serial": new_serial}]
            )
            device_results[device_name]["updated"].append("chassis")
            job.event(f"{device_name} chassis serial updated")
        except Exception as exc:
            msg = f"{device_name} - failed to update chassis serial: {exc}"
            ret.errors.append(msg)
            log.error(msg)
            job.event(msg, severity="ERROR")

    # Optional module bay creation.
    if create_module_bays:
        job.event("creating missing module bays")
        for device_name, actions in full_diff.items():
            for slot in actions["create"]:
                if slot == "chassis":
                    continue
                if slot in nb_module_bays[device_name]:
                    continue
                try:
                    created = nb.dcim.module_bays.create(
                        device=nb_devices_data[device_name]["id"],
                        name=slot,
                        label=slot,
                    )
                    nb_module_bays[device_name][slot] = {
                        "id": int(created.id),
                        "name": slot,
                    }
                    device_results[device_name]["created"].append(slot)
                    job.event(f"{device_name}:{slot} module bay created")
                except Exception as exc:
                    msg = (
                        f"{device_name}:{slot} - failed to create module bay: {exc}"
                    )
                    ret.errors.append(msg)
                    log.error(msg)
                    job.event(msg, severity="ERROR")

    # Module creates.
    job.event("creating missing modules")
    for device_name, actions in full_diff.items():
        for slot in actions["create"]:
            if slot == "chassis":
                continue
            desired = normalised_live_all[device_name][slot]
            if (device_name, slot) in missing_module_bay_keys:
                continue
            if slot not in nb_module_bays[device_name]:
                msg = f"{device_name}:{slot} - module bay not found"
                ret.errors.append(msg)
                log.error(msg)
                job.event(msg, severity="ERROR")
                continue
            lookup_key = (slugify(desired["manufacturer"]), desired["module_type"])
            if lookup_key in missing_module_type_keys:
                continue

            module_type_id, module_type_label = get_or_create_module_type_id(
                nb,
                job,
                ret,
                device_name,
                slot,
                desired,
                create_module_types,
                module_type_lookup_cache,
            )
            if not module_type_id:
                continue
            if module_type_label and create_module_types:
                device_results[device_name]["created"].append(module_type_label)

            payload = {
                "device": nb_devices_data[device_name]["id"],
                "module_bay": nb_module_bays[device_name][slot]["id"],
                "module_type": module_type_id,
                "status": desired["status"],
                "serial": desired["serial"],
            }
            if desired.get("description"):
                payload["description"] = desired["description"]

            try:
                created = nb.dcim.modules.create(**payload)
                nb_module_ids[device_name][slot] = int(created.id)
                device_results[device_name]["created"].append(slot)
                job.event(f"{device_name}:{slot} module created")
            except Exception as exc:
                msg = f"{device_name}:{slot} - failed to create module: {exc}"
                ret.errors.append(msg)
                log.error(msg)
                job.event(msg, severity="ERROR")

    # Module updates.
    job.event("updating changed modules")
    for device_name, actions in full_diff.items():
        for slot, changes in actions["update"].items():
            if slot == "chassis":
                continue
            desired = normalised_live_all[device_name][slot]
            module_id = nb_module_ids[device_name].get(slot)
            if not module_id:
                continue

            payload = {"id": module_id}
            if "serial" in changes:
                payload["serial"] = desired["serial"]
            if "status" in changes:
                payload["status"] = desired["status"]
            if "description" in changes:
                payload["description"] = desired["description"]
            module_identity_changed = (
                "module_type" in changes or "manufacturer" in changes
            )
            if module_identity_changed:
                lookup_key = (
                    slugify(desired["manufacturer"]),
                    desired["module_type"],
                )
                if lookup_key in missing_module_type_keys:
                    continue
                module_type_id, module_type_label = get_or_create_module_type_id(
                    nb,
                    job,
                    ret,
                    device_name,
                    slot,
                    desired,
                    create_module_types,
                    module_type_lookup_cache,
                )
                if not module_type_id:
                    continue
                if module_type_label and create_module_types:
                    device_results[device_name]["created"].append(module_type_label)
                payload["module_type"] = module_type_id

            module_update_has_changes = len(payload) > 1
            if not module_update_has_changes:
                continue

            try:
                nb.dcim.modules.update([payload])
                device_results[device_name]["updated"].append(slot)
                job.event(f"{device_name}:{slot} module updated")
            except Exception as exc:
                msg = f"{device_name}:{slot} - failed to update module: {exc}"
                ret.errors.append(msg)
                log.error(msg)
                job.event(msg, severity="ERROR")

    # Optional module deletions.
    if process_deletions:
        job.event("deleting stale modules")
        for device_name, actions in full_diff.items():
            for slot in actions["delete"]:
                if slot == "chassis":
                    continue
                module_id = nb_module_ids[device_name].get(slot)
                if not module_id:
                    continue
                try:
                    module = nb.dcim.modules.get(id=module_id)
                    module.delete()
                    device_results[device_name]["deleted"].append(slot)
                    job.event(f"{device_name}:{slot} module deleted")
                except Exception as exc:
                    msg = f"{device_name}:{slot} - failed to delete module: {exc}"
                    ret.errors.append(msg)
                    log.error(msg)
                    job.event(msg, severity="ERROR")
    elif delete_count:
        job.event(
            f"skipping {delete_count} module deletion(s), process_deletions=False"
        )

    # Keep result lists deterministic and unique.
    for device_result in device_results.values():
        for key in ("created", "updated", "deleted", "in_sync"):
            device_result[key] = sorted(set(device_result[key]))

    job.event("device inventory sync complete")
    return ret