fix(dcim): Prevent stale Cable Paths on Termination changes

Trace each replacement before retiring its superseded paths in
create_cablepaths(). Keep replacement and co-origin recovery in one
transaction. Use a worklist to restore origins whose current path is
removed, sharing parent expansion and connector/channel grouping.

Rebuild affected paths from current cable-end membership rather than
historical origin hops so candidate order cannot determine grouping.
Batch origin lookups, skip missing generic termination targets, and
stream trace_paths by cable end.

Have trace_paths report unsupported groups, continue processing
independent groups, and exit nonzero if any group failed.

Add regression coverage for shared origins, channelized parents,
rollback, candidate ordering, missing targets, generic prefetching,
and command failures.

Fixes #23121

docs/administration/management-commands.md

@@ -158,6 +158,30 @@ Generate any missing cable paths among all cable termination objects. This is us
 python3 netbox/manage.py trace_paths
 ```
 
+### Failures and upgrades
+
+Unsupported origin groups are reported while tracing continues for other groups. If any group cannot be traced,
+`trace_paths` exits with a nonzero status after processing the remaining groups. A selected cable end that resolves
+to no live terminations is also reported as a failure, rather than counted as successfully retraced. This can happen
+if membership changes while the command is running.
+
+The per-model summaries count selected endpoints, not paths written or every sibling retraced on a selected end.
+The failure summary also counts the groups that failed; the final error counts failed groups across all endpoint
+models.
+
+Cable-end membership and connector grouping are resolved from `CableTermination`. The command does not repair
+persisted endpoint fields such as `cable`, `cable_end`, `cable_connector`, or `cable_positions`. Inconsistent cached
+links or positions can still prevent tracing; repeatedly running the command does not repair those inconsistencies.
+Do not assume that every reported failure can be corrected by editing topology in the UI.
+
+When invoked during an upgrade, this failure stops the remaining upgrade steps. Investigate the reported topology
+or data inconsistency, correct its underlying cause, and rerun the upgrade. When an end changed concurrently,
+rerun tracing after the changes have completed.
+
+With `--force`, existing cable paths are deleted before rebuilding starts. A failed group can therefore remain
+without paths until the underlying problem is corrected and tracing is run again. A group-level rollback does not
+undo the initial deletion of all paths.
+
 ## webhook_receiver
 
 Start a simple HTTP listener that prints any requests it receives. This is a debugging aid for testing webhooks: point a webhook at the listener and inspect exactly what NetBox sends. It listens on port 9000 by default; pass `--port` to change it and `--no-headers` to suppress the request headers.

netbox/core/management/commands/upgrade.py

@@ -58,7 +58,7 @@ class Command(BaseCommand):
             out.write("Skipping cable path check.")
         else:
             out.write("Checking for missing cable paths...")
-            call_command('trace_paths', no_input=options['no_input'], stdout=out)
+            call_command('trace_paths', no_input=options['no_input'], stdout=out, stderr=self.stderr)
 
         # Documentation (filesystem; needs the documentation source tree)
         if options['readonly'] and options['build_docs']:

netbox/dcim/management/commands/trace_paths.py

@@ -1,10 +1,15 @@
-from django.core.management.base import BaseCommand
+from itertools import groupby, islice
+
+from django.core.management.base import BaseCommand, CommandError
 from django.core.management.color import no_style
 from django.db import connection
-from django.db.models import Q
+from django.db.models import F, Q
 
+from dcim.exceptions import UnsupportedCablePath
 from dcim.models import CablePath, ConsolePort, ConsoleServerPort, Interface, PowerFeed, PowerOutlet, PowerPort
-from dcim.signals import create_cablepaths
+from dcim.utils import create_cablepaths, get_cable_end_terminations
+
+ORIGIN_GROUP_BATCH_SIZE = 100
 
 ENDPOINT_MODELS = (
     ConsolePort,
@@ -36,6 +41,17 @@ class Command(BaseCommand):
         bar_size = int(percentage / 5)
         self.stdout.write(f"\r  [{'#' * bar_size}{' ' * (20 - bar_size)}] {int(percentage)}%", ending='')
 
+    @staticmethod
+    def group_key(obj):
+        """
+        Key an annotated endpoint by its cable end, falling back to its own PK when it has no termination row.
+        """
+        if obj._trace_cable_id is not None:
+            return (obj._trace_cable_id, obj._trace_cable_end)
+
+        # Wireless endpoints and missing CableTermination rows retain the singleton fallback.
+        return obj.pk
+
     def handle(self, *model_names, **options):
 
         # If --force was passed, first delete all existing CablePaths
@@ -66,7 +82,8 @@ class Command(BaseCommand):
                 for sql in sequence_sql:
                     cursor.execute(sql)
 
-        # Retrace paths
+        # Retrace paths. A failed group must roll back, but must not prevent repairing independent groups.
+        failures = 0
         for model in ENDPOINT_MODELS:
             params = Q(cable__isnull=False)
             if hasattr(model, 'wireless_link'):
@@ -79,12 +96,65 @@ class Command(BaseCommand):
                 self.stdout.write(f'Found no missing {model._meta.verbose_name} paths; skipping')
                 continue
             self.stdout.write(f'Retracing {origins_count} cabled {model._meta.verbose_name_plural}...')
-            i = 0
-            for i, obj in enumerate(origins, start=1):
-                create_cablepaths([obj])
-                if not i % 100:
-                    self.draw_progress_bar(i * 100 / origins_count)
+            # The unique termination relation gives each selected endpoint at most one authoritative end.
+            # Adjacent cable ends can be processed together without keeping a set of all previously seen ends.
+            origins = origins.annotate(
+                _trace_cable_id=F('cable_terminations__cable_id'),
+                _trace_cable_end=F('cable_terminations__cable_end'),
+            ).order_by('_trace_cable_id', '_trace_cable_end', 'pk')
+
+            grouped_origins = groupby(origins.iterator(chunk_size=1000), key=self.group_key)
+            retraced = i = model_failures = 0
+            while True:
+                batch = []
+                # Consume each group before advancing groupby: its iterators share the underlying stream.
+                # Keep only a representative and count, not all selected endpoints on each end.
+                for key, selected in islice(grouped_origins, ORIGIN_GROUP_BATCH_SIZE):
+                    first = next(selected)
+                    selected_count = 1 + sum(1 for _ in selected)
+                    batch.append((key, first, selected_count))
+                if not batch:
+                    break
+
+                cable_keys = [key for key, _, _ in batch if isinstance(key, tuple)]
+                cable_ends = get_cable_end_terminations(cable_keys)
+                for key, first, selected_count in batch:
+                    group = cable_ends[key] if isinstance(key, tuple) else [first]
+                    try:
+                        if not group:
+                            raise UnsupportedCablePath(
+                                'No current terminations remain for the selected cable end; '
+                                'its membership may have changed during tracing.'
+                            )
+                        create_cablepaths(group)
+                    except UnsupportedCablePath as error:
+                        # The helper's savepoint has unwound; other groups can still be repaired.
+                        model_failures += 1
+                        if isinstance(key, tuple):
+                            target = f'cable #{key[0]} end {key[1]}'
+                        else:
+                            target = f'{first._meta.label} #{first.pk}'
+                        self.stderr.write(self.style.ERROR(f'Unable to trace {target}: {error}'))
+                    else:
+                        retraced += selected_count
+
+                    # Advance progress for every selected endpoint, even in a failed or shared group.
+                    for completed in range((i // 100 + 1) * 100, i + selected_count + 1, 100):
+                        self.draw_progress_bar(completed * 100 / origins_count)
+                    i += selected_count
             self.draw_progress_bar(100)
-            self.stdout.write(self.style.SUCCESS(f'\n  Retraced {i} {model._meta.verbose_name_plural}'))
+            self.stdout.write(self.style.SUCCESS(f'\n  Retraced {retraced} {model._meta.verbose_name_plural}'))
+            if model_failures:
+                failed = origins_count - retraced
+                self.stdout.write(self.style.WARNING(
+                    f'  Failed to retrace {failed} selected {model._meta.verbose_name_plural} '
+                    f'in {model_failures} origin group(s)'
+                ))
+                failures += model_failures
 
+        if failures:
+            raise CommandError(
+                f'Unable to trace {failures} origin group(s) across all endpoint models. Other groups were processed; '
+                'correct the reported topology or data inconsistencies and rerun trace_paths.'
+            )
         self.stdout.write(self.style.SUCCESS('Finished.'))

netbox/dcim/tests/test_api.py

@@ -4741,6 +4741,29 @@ class CableTestCase(APIViewTestCases.APIViewTestCase):
             },
         ]
 
+    def test_repeated_put_does_not_accumulate_paths(self):
+        """
+        Repeating an identical PUT must leave the cable with the two paths its terminations trace.
+        """
+        self.add_permissions('dcim.change_cable')
+        cable = Cable.objects.get(label='Cable 1')
+        interface_a = Interface.objects.get(cable=cable, cable_end=CableEndChoices.SIDE_A)
+        interface_b = Interface.objects.get(cable=cable, cable_end=CableEndChoices.SIDE_B)
+        data = {
+            'status': cable.status,
+            'a_terminations': [{'object_type': 'dcim.interface', 'object_id': interface_a.pk}],
+            'b_terminations': [{'object_type': 'dcim.interface', 'object_id': interface_b.pk}],
+        }
+
+        for attempt in range(3):
+            with self.subTest(attempt=attempt):
+                response = self.client.put(self._get_detail_url(cable), data, format='json', **self.header)
+
+                self.assertHttpStatus(response, status.HTTP_200_OK)
+                for interface in (interface_a, interface_b):
+                    self.assertTrue(Interface.objects.get(pk=interface.pk)._path.is_complete)
+                self.assertEqual(CablePath.objects.filter(_nodes__contains=cable).count(), 2)
+
     def test_graphql_cable_termination_cached_filters(self):
         """
         Validate filtering cables by cached CableTermination relations via GraphQL:

netbox/dcim/tests/test_cablepath_rebuilds.py

@@ -0,0 +1,864 @@
+from io import StringIO
+from itertools import permutations
+from types import SimpleNamespace
+from unittest.mock import call, patch
+
+from django.contrib.contenttypes.models import ContentType
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from django.db import transaction
+from django.db.models import Case, IntegerField, Value, When
+
+from dcim import utils
+from dcim.choices import CableEndChoices, CableProfileChoices
+from dcim.exceptions import UnsupportedCablePath
+from dcim.management.commands import trace_paths
+from dcim.models import (
+    Cable,
+    CablePath,
+    CableTermination,
+    ConsolePort,
+    ConsoleServerPort,
+    FrontPort,
+    Interface,
+    PortMapping,
+    RearPort,
+)
+from dcim.tests.utils import BaseCablePathTestCase
+
+
+class CablePathRebuildTestCase(BaseCablePathTestCase):
+    """
+    Explicit rebuilds must use current membership, not the order or grouping of historical paths.
+    """
+
+    def _create_interfaces(self, *names):
+        return [Interface.objects.create(device=self.device, name=name) for name in names]
+
+    def _assert_joint_paths(self, cable, far, origins):
+        self.assertCurrentPathExists((far, cable, origins), is_complete=True, is_active=True)
+        path = self.assertPathExists((origins, cable, far), is_complete=True, is_active=True)
+        for origin in origins:
+            origin.refresh_from_db()
+            self.assertPathIsSet(origin, path)
+        self.assertEqual(CablePath.objects.filter(_nodes__contains=cable).count(), 2)
+
+    def test_rebuild_uses_current_membership_in_every_candidate_order(self):
+        for order in permutations(('far', 'joint', 'stale')):
+            with self.subTest(order=order), transaction.atomic():
+                far, first, second = self._create_interfaces('IF1', 'IF2', 'IF3')
+                cable = Cable(a_terminations=[far], b_terminations=[first, second])
+                cable.save()
+                for origin in (far, first, second):
+                    origin.refresh_from_db()
+                joint = first._path
+                stale = CablePath.from_origin([second])
+                stale.save()
+                joint.save()
+                pks = {'far': far._path_id, 'joint': joint.pk, 'stale': stale.pk}
+                node = utils.object_to_path_node(cable)
+                filter_paths = CablePath.objects.filter
+                candidate_orders = []
+
+                def ordered_candidates(*args, **kwargs):
+                    queryset = filter_paths(*args, **kwargs)
+                    # Order only candidate discovery, never the replacement helper's overlap lookups.
+                    if not args and kwargs in ({'_nodes__overlap': [node]}, {'_nodes__contains': cable}):
+                        queryset = queryset.order_by(Case(
+                            *[When(pk=pks[name], then=Value(index)) for index, name in enumerate(order)],
+                            output_field=IntegerField(),
+                        ))
+                        candidate_orders.append(list(queryset.values_list('pk', flat=True)))
+                    return queryset
+
+                with patch.object(CablePath.objects, 'filter', side_effect=ordered_candidates):
+                    utils.rebuild_paths([cable])
+
+                self.assertEqual(candidate_orders, [[pks[name] for name in order]])
+                self.assertFalse(CablePath.objects.filter(pk__in=pks.values()).exists())
+                self._assert_joint_paths(cable, far, [first, second])
+                self.assertEqual(CablePath.objects.count(), 2)
+
+                # Repeating the rebuild must preserve the same shape, without pinning any query order.
+                utils.rebuild_paths([cable])
+                self._assert_joint_paths(cable, far, [first, second])
+                self.assertEqual(CablePath.objects.count(), 2)
+                transaction.set_rollback(True)
+
+    def test_rebuild_resolves_a_stale_hop_spanning_current_cable_ends(self):
+        far1, origin1, far2, origin2 = self._create_interfaces('IF1', 'IF2', 'IF3', 'IF4')
+        cable1 = Cable(a_terminations=[far1], b_terminations=[origin1])
+        cable1.save()
+        cable2 = Cable(a_terminations=[far2], b_terminations=[origin2])
+        cable2.save()
+        far2.refresh_from_db()
+        untouched = far2._path_id
+        stale = CablePath(
+            path=[
+                [utils.object_to_path_node(origin1), utils.object_to_path_node(origin2)],
+                [utils.object_to_path_node(cable1)],
+                [utils.object_to_path_node(far1)],
+            ],
+            is_complete=True,
+            is_active=True,
+        )
+        stale.save()
+
+        utils.rebuild_paths([cable1])
+
+        self.assertFalse(CablePath.objects.filter(pk=stale.pk).exists())
+        for cable, far, origin in ((cable1, far1, origin1), (cable2, far2, origin2)):
+            self.assertCurrentPathExists((origin, cable, far), is_complete=True)
+            self.assertCurrentPathExists((far, cable, origin), is_complete=True)
+        far2.refresh_from_db()
+        self.assertEqual(far2._path_id, untouched)
+        self.assertEqual(CablePath.objects.count(), 4)
+
+    def test_rebuild_failure_restores_all_candidates_and_references(self):
+        interfaces = self._create_interfaces('IF1', 'IF2', 'IF3', 'IF4')
+        cables = [
+            Cable(a_terminations=[interfaces[0]], b_terminations=[interfaces[1]]),
+            Cable(a_terminations=[interfaces[2]], b_terminations=[interfaces[3]]),
+        ]
+        for cable in cables:
+            cable.save()
+        original_ids = set(CablePath.objects.values_list('pk', flat=True))
+        references = dict(Interface.objects.filter(pk__in=[obj.pk for obj in interfaces]).values_list('pk', '_path_id'))
+        traced = CablePath.from_origin
+        attempts = 0
+
+        def fail_after_a_replacement(origins):
+            nonlocal attempts
+            attempts += 1
+            if attempts == 2:
+                self.assertFalse(CablePath.objects.filter(pk__in=original_ids).exists())
+                self.assertTrue(CablePath.objects.exclude(pk__in=original_ids).exists())
+                raise UnsupportedCablePath('Simulated rebuild failure')
+            return traced(origins)
+
+        with patch.object(CablePath, 'from_origin', side_effect=fail_after_a_replacement):
+            with self.assertRaisesMessage(UnsupportedCablePath, 'Simulated rebuild failure'):
+                utils.rebuild_paths(cables)
+
+        self.assertEqual(attempts, 2)
+        self.assertEqual(set(CablePath.objects.values_list('pk', flat=True)), original_ids)
+        for interface in interfaces:
+            interface.refresh_from_db()
+            self.assertEqual(interface._path_id, references[interface.pk])
+
+    def test_rebuild_removes_rows_for_missing_and_disconnected_origins(self):
+        far, origin, detached = self._create_interfaces('IF1', 'IF2', 'Detached')
+        cable = Cable(a_terminations=[far], b_terminations=[origin])
+        cable.save()
+        # Encode a nonexistent interface without creating inconsistent current cable associations.
+        missing = Interface(pk=1000000000)
+        self.assertFalse(Interface.objects.filter(pk=missing.pk).exists())
+        obsolete_ids = []
+        for obj in (detached, missing):
+            path = CablePath(
+                path=[
+                    [utils.object_to_path_node(obj)],
+                    [utils.object_to_path_node(cable)],
+                    [utils.object_to_path_node(far)],
+                ],
+                is_complete=True,
+                is_active=True,
+            )
+            path.save()
+            obsolete_ids.append(path.pk)
+
+        utils.rebuild_paths([cable, cable])
+
+        self.assertFalse(CablePath.objects.filter(pk__in=obsolete_ids).exists())
+        self.assertCurrentPathExists((origin, cable, far), is_complete=True)
+        self.assertCurrentPathExists((far, cable, origin), is_complete=True)
+        detached.refresh_from_db()
+        self.assertPathIsNotSet(detached)
+        self.assertEqual(CablePath.objects.count(), 2)
+
+    def test_origin_group_uses_the_termination_not_cached_fields(self):
+        far, first, second = self._create_interfaces('IF1', 'IF2', 'IF3')
+        cable = Cable(a_terminations=[far], b_terminations=[first, second])
+        cable.save()
+        for cached_fields in ({'cable_end': CableEndChoices.SIDE_A}, {'cable': None, 'cable_end': None}):
+            with self.subTest(cached_fields=cached_fields):
+                Interface.objects.filter(pk=first.pk).update(**cached_fields)
+                first.refresh_from_db()
+                groups = utils.get_cablepath_origin_groups([first])
+                key, origins = next(iter(groups.items()))
+                self.assertEqual(key, (cable.pk, CableEndChoices.SIDE_B))
+                self.assertEqual(list(origins), [first, second])
+
+                # Membership resolution does not repair the endpoint's cached cable fields.
+                if cached_fields.get('cable', cable) is None:
+                    before = list(CablePath.objects.order_by('pk').values_list('pk', 'path'))
+                    with self.assertRaisesMessage(UnsupportedCablePath, 'same link'):
+                        utils.create_cablepaths(origins)
+                    self.assertEqual(list(CablePath.objects.order_by('pk').values_list('pk', 'path')), before)
+                else:
+                    # Legacy tracing resolves the opposite end through CableTermination, not cached cable_end.
+                    utils.create_cablepaths(origins)
+                    self._assert_joint_paths(cable, far, [first, second])
+                first.refresh_from_db()
+                self.assertEqual(first.cable_end, cached_fields['cable_end'])
+                if 'cable' in cached_fields:
+                    self.assertIsNone(first.cable_id)
+
+    def test_origin_connectors_use_termination_metadata_without_repairing_cached_fields(self):
+        near1, near2, far1, far2 = self._create_interfaces('Near1', 'Near2', 'Far1', 'Far2')
+        cable = Cable(
+            profile=CableProfileChoices.TRUNK_2C1P,
+            a_terminations=[near1, near2], b_terminations=[far1, far2],
+        )
+        cable.clean()
+        cable.save()
+        # Corrupt only the derived connector fields, not current membership or positions.
+        Interface.objects.filter(pk__in=[near1.pk, near2.pk]).update(cable_connector=None)
+        near1.refresh_from_db()
+        groups = utils.get_cablepath_origin_groups([near1])
+        origins = groups[(cable.pk, CableEndChoices.SIDE_A)]
+        self.assertEqual([(obj.pk, obj.cable_connector) for obj in origins], [(near1.pk, 1), (near2.pk, 2)])
+
+        with patch.object(CablePath, 'from_origin', wraps=CablePath.from_origin) as traced:
+            utils.create_cablepaths(origins)
+
+        self.assertEqual(traced.call_args_list, [call([near1]), call([near2])])
+        for near, far in ((near1, far1), (near2, far2)):
+            self.assertCurrentPathExists((near, cable, far), is_complete=True, is_active=True)
+            self.assertCurrentPathExists((far, cable, near), is_complete=True, is_active=True)
+        self.assertEqual(CablePath.objects.count(), 4)
+        self.assertEqual(
+            list(Interface.objects.filter(pk__in=[near1.pk, near2.pk]).values_list('cable_connector', flat=True)),
+            [None, None],
+        )
+
+        # The explicit rebuild must use the same grouping without persisting a cache repair.
+        utils.rebuild_paths([cable])
+        for near, far in ((near1, far1), (near2, far2)):
+            self.assertCurrentPathExists((near, cable, far), is_complete=True, is_active=True)
+            near.refresh_from_db()
+            self.assertIsNone(near.cable_connector)
+        self.assertEqual(CablePath.objects.count(), 4)
+
+    def test_rebuild_rejects_cached_link_drift_without_losing_candidates(self):
+        far, first, second = self._create_interfaces('Far', 'First', 'Second')
+        cable = Cable(a_terminations=[far], b_terminations=[first, second])
+        cable.save()
+        Interface.objects.filter(pk=first.pk).update(cable=None, cable_end=None)
+        paths = list(CablePath.objects.order_by('pk').values_list('pk', 'path'))
+        references = dict(Interface.objects.values_list('pk', '_path_id'))
+
+        with self.assertRaisesMessage(UnsupportedCablePath, 'same link'):
+            utils.rebuild_paths([cable])
+
+        self.assertEqual(list(CablePath.objects.order_by('pk').values_list('pk', 'path')), paths)
+        self.assertEqual(dict(Interface.objects.values_list('pk', '_path_id')), references)
+        first.refresh_from_db()
+        self.assertIsNone(first.cable_id)
+        self.assertIsNone(first.cable_end)
+
+    def test_origin_group_keeps_an_uncabled_object_as_a_singleton(self):
+        origin, = self._create_interfaces('Detached')
+        key, origins = next(iter(utils.get_cablepath_origin_groups([origin]).items()))
+        self.assertEqual(key, utils.object_to_path_node(origin))
+        self.assertEqual(list(origins), [origin])
+
+    def test_empty_rebuild_does_not_query_the_database(self):
+        with self.assertNumQueries(0):
+            utils.rebuild_paths(iter(()))
+
+    def test_rebuild_midspan_preserves_remote_joint_origins_and_unrelated_paths(self):
+        far, first, second, other_a, other_b, partial = self._create_interfaces(
+            'IF1', 'IF2', 'IF3', 'Unrelated A', 'Unrelated B', 'Partial'
+        )
+        fronts = [FrontPort.objects.create(device=self.device, name=f'FP{i}') for i in range(3)]
+        rears = [RearPort.objects.create(device=self.device, name=f'RP{i}') for i in range(3)]
+        for front, rear in zip(fronts, rears):
+            PortMapping.objects.create(
+                device=self.device,
+                front_port=front, front_port_position=1,
+                rear_port=rear, rear_port_position=1,
+            )
+        near_cable = Cable(a_terminations=[first, second], b_terminations=[fronts[0]])
+        near_cable.save()
+        far_cable = Cable(a_terminations=[fronts[1]], b_terminations=[far])
+        far_cable.save()
+        middle = Cable(a_terminations=[rears[0]], b_terminations=[rears[1]])
+        middle.save()
+        unrelated = Cable(a_terminations=[other_a], b_terminations=[other_b])
+        unrelated.save()
+        partial_cable = Cable(a_terminations=[partial], b_terminations=[fronts[2]])
+        partial_cable.save()
+        route = ([first, second], near_cable, fronts[0], rears[0], middle, rears[1], fronts[1], far_cable, far)
+        joint = self.assertPathExists(route, is_complete=True)
+        second.refresh_from_db()
+        stale = CablePath.from_origin([second])
+        stale.save()
+        joint.save()
+        preserved = {}
+        for obj in (other_a, other_b, partial):
+            obj.refresh_from_db()
+            preserved[obj.pk] = obj._path_id
+
+        utils.rebuild_paths([rears[0], fronts[1]])
+
+        self.assertFalse(CablePath.objects.filter(pk=stale.pk).exists())
+        joint = self.assertPathExists(route, is_complete=True, is_active=True)
+        for obj in (first, second):
+            obj.refresh_from_db()
+            self.assertPathIsSet(obj, joint)
+        self.assertCurrentPathExists(tuple(reversed(route)), is_complete=True, is_active=True)
+        self.assertCurrentPathExists((other_a, unrelated, other_b), pk=preserved[other_a.pk], is_complete=True)
+        self.assertCurrentPathExists((other_b, unrelated, other_a), pk=preserved[other_b.pk], is_complete=True)
+        self.assertCurrentPathExists(
+            (partial, partial_cable, fronts[2], rears[2]), pk=preserved[partial.pk], is_complete=False
+        )
+        self.assertEqual(CablePath.objects.count(), 5)
+
+    def test_rebuild_preserves_channel_and_connector_origins(self):
+        parent = Interface.objects.create(
+            device=self.device, name='et0', type='100gbase-x-qsfp28', channels=4
+        )
+        channels = [
+            Interface.objects.create(
+                device=self.device, name=f'et0/{i}', type='channel', parent=parent, channel_id=i
+            ) for i in range(1, 5)
+        ]
+        far = [
+            Interface.objects.create(device=self.device, name=f'xe{i}', type='10gbase-x-sfpp') for i in range(4)
+        ]
+        cable = Cable(
+            profile=CableProfileChoices.BREAKOUT_1C4P_4C1P,
+            a_terminations=[parent], b_terminations=far,
+        )
+        cable.clean()
+        cable.save()
+        channels[0].refresh_from_db()
+        key, origins = next(iter(utils.get_cablepath_origin_groups([channels[0]]).items()))
+        self.assertEqual(key, utils.object_to_path_node(channels[0]))
+        self.assertEqual(list(origins), [channels[0]])
+        stale = CablePath(
+            path=[
+                [utils.object_to_path_node(obj) for obj in channels],
+                [utils.object_to_path_node(cable)],
+                [utils.object_to_path_node(obj) for obj in far],
+            ],
+            is_complete=True,
+            is_active=True,
+        )
+        stale.save()
+
+        utils.rebuild_paths([cable])
+
+        self.assertFalse(CablePath.objects.filter(pk=stale.pk).exists())
+        self.assertEqual(CablePath.objects.count(), 8)
+        for channel, peer in zip(channels, far):
+            self.assertCurrentPathExists((channel, cable, peer), is_complete=True, is_active=True)
+            self.assertCurrentPathExists((peer, cable, channel), is_complete=True, is_active=True)
+        parent.refresh_from_db()
+        self.assertPathIsNotSet(parent)
+
+    def test_rebuild_does_not_trace_an_unrelated_connector(self):
+        near1, near2, far1, far2 = self._create_interfaces('Near1', 'Near2', 'Far1', 'Far2')
+        fronts = [FrontPort.objects.create(device=self.device, name=f'FP{i}') for i in range(2)]
+        rears = [RearPort.objects.create(device=self.device, name=f'RP{i}') for i in range(2)]
+        for front, rear in zip(fronts, rears):
+            PortMapping.objects.create(
+                device=self.device, front_port=front, front_port_position=1,
+                rear_port=rear, rear_port_position=1,
+            )
+        trunk = Cable(
+            profile=CableProfileChoices.TRUNK_2C1P,
+            a_terminations=[near1, near2], b_terminations=rears,
+        )
+        trunk.clean()
+        trunk.save()
+        drops = []
+        for front, far in zip(fronts, (far1, far2)):
+            cable = Cable(a_terminations=[front], b_terminations=[far])
+            cable.save()
+            drops.append(cable)
+        affected_route = (near1, trunk, rears[0], fronts[0], drops[0], far1)
+        other_route = (near2, trunk, rears[1], fronts[1], drops[1], far2)
+        affected = {
+            self.assertCurrentPathExists(affected_route, is_complete=True).pk,
+            self.assertCurrentPathExists(tuple(reversed(affected_route)), is_complete=True).pk,
+        }
+        preserved = {
+            near2.pk: self.assertCurrentPathExists(other_route, is_complete=True).pk,
+            far2.pk: self.assertCurrentPathExists(tuple(reversed(other_route)), is_complete=True).pk,
+        }
+        trace = CablePath.from_origin
+
+        def reject_unrelated_group(origins):
+            if near2 in origins or far2 in origins:
+                raise UnsupportedCablePath('The unrelated connector must not be traced')
+            return trace(origins)
+
+        with patch.object(CablePath, 'from_origin', side_effect=reject_unrelated_group) as traced:
+            utils.rebuild_paths([fronts[0]])
+
+        self.assertCountEqual(traced.call_args_list, [call([near1]), call([far1])])
+        self.assertFalse(CablePath.objects.filter(pk__in=affected).exists())
+        self.assertEqual(CablePath.objects.count(), 4)
+        self.assertCurrentPathExists(affected_route, is_complete=True, is_active=True)
+        self.assertCurrentPathExists(tuple(reversed(affected_route)), is_complete=True, is_active=True)
+        self.assertCurrentPathExists(other_route, pk=preserved[near2.pk], is_complete=True, is_active=True)
+        self.assertCurrentPathExists(
+            tuple(reversed(other_route)), pk=preserved[far2.pk], is_complete=True, is_active=True
+        )
+
+    def test_rebuild_expands_a_historical_parent_before_selecting_groups(self):
+        parent = Interface.objects.create(
+            device=self.device, name='et0', type='100gbase-x-qsfp28', channels=2
+        )
+        channels = [
+            Interface.objects.create(
+                device=self.device, name=f'et0/{i}', type='channel', parent=parent, channel_id=i
+            ) for i in range(1, 3)
+        ]
+        far = self._create_interfaces('Far1', 'Far2')
+        cable = Cable(
+            profile=CableProfileChoices.BREAKOUT_1C2P_2C1P, a_terminations=[parent], b_terminations=far,
+        )
+        cable.clean()
+        cable.save()
+        preserved = {}
+        for channel, peer in zip(channels, far):
+            peer.refresh_from_db()
+            preserved[peer.pk] = peer._path_id
+            channel.refresh_from_db()
+            channel._path.delete()
+        stale = CablePath(
+            path=[
+                [utils.object_to_path_node(parent)],
+                [utils.object_to_path_node(cable)],
+                [utils.object_to_path_node(peer) for peer in far],
+            ],
+            is_complete=True,
+            is_active=True,
+        )
+        stale.save()
+        self.assertEqual(list(CablePath.objects.filter(_nodes__contains=parent)), [stale])
+
+        with patch.object(CablePath, 'from_origin', wraps=CablePath.from_origin) as traced:
+            utils.rebuild_paths([parent])
+
+        self.assertCountEqual(traced.call_args_list, [call([channels[0]]), call([channels[1]])])
+        self.assertFalse(CablePath.objects.filter(pk=stale.pk).exists())
+        self.assertEqual(CablePath.objects.count(), 4)
+        parent.refresh_from_db()
+        self.assertPathIsNotSet(parent)
+        for channel, peer in zip(channels, far):
+            self.assertCurrentPathExists((channel, cable, peer), is_complete=True, is_active=True)
+            self.assertCurrentPathExists(
+                (peer, cable, channel), pk=preserved[peer.pk], is_complete=True, is_active=True
+            )
+
+    def test_rebuild_keeps_requested_connectors_in_one_recovery_worklist(self):
+        near1, near2, far1, far2, other1, other2 = self._create_interfaces(
+            'Near1', 'Near2', 'Far1', 'Far2', 'Other1', 'Other2'
+        )
+        cable = Cable(
+            profile=CableProfileChoices.TRUNK_2C1P,
+            a_terminations=[near1, near2], b_terminations=[far1, far2],
+        )
+        cable.clean()
+        cable.save()
+        unrelated = Cable(a_terminations=[other1], b_terminations=[other2])
+        unrelated.save()
+        preserved = {
+            obj.pk: self.assertCurrentPathExists((obj, unrelated, peer), is_complete=True).pk
+            for obj, peer in ((other1, other2), (other2, other1))
+        }
+        # This historical hop is not a rebuild candidate: it records a different cable. Replacing the first
+        # requested connector retires it and discovers the second, already-requested connector for recovery.
+        stale = CablePath(
+            path=[
+                [utils.object_to_path_node(obj) for obj in (near1, near2)],
+                [utils.object_to_path_node(unrelated)],
+                [utils.object_to_path_node(obj) for obj in (far1, far2)],
+            ],
+            is_complete=True,
+            is_active=True,
+        )
+        stale.save()
+        self.assertFalse(CablePath.objects.filter(pk=stale.pk, _nodes__contains=cable).exists())
+
+        with patch.object(CablePath, 'from_origin', wraps=CablePath.from_origin) as traced:
+            utils.rebuild_paths([cable])
+
+        self.assertCountEqual(traced.call_args_list, [call([obj]) for obj in (near1, near2, far1, far2)])
+        self.assertFalse(CablePath.objects.filter(pk=stale.pk).exists())
+        self.assertEqual(CablePath.objects.count(), 6)
+        for near, far in ((near1, far1), (near2, far2)):
+            self.assertCurrentPathExists((near, cable, far), is_complete=True, is_active=True)
+            self.assertCurrentPathExists((far, cable, near), is_complete=True, is_active=True)
+        for obj, peer in ((other1, other2), (other2, other1)):
+            self.assertCurrentPathExists((obj, unrelated, peer), pk=preserved[obj.pk], is_complete=True)
+
+    def test_origin_objects_are_fetched_once_per_content_type(self):
+        far, *origins = self._create_interfaces('Far', *(f'IF{i}' for i in range(32)))
+        cable = Cable(a_terminations=[far], b_terminations=origins)
+        cable.save()
+        nodes = [utils.object_to_path_node(obj) for obj in origins]
+        # ContentType IDs are already cached by object_to_path_node().
+        with self.assertNumQueries(1):
+            objects = utils._get_cablepath_origin_objects(nodes)
+            self.assertEqual([objects[node].link for node in nodes], [cable] * len(nodes))
+        self.assertEqual([objects[node] for node in nodes], origins)
+
+    def test_current_groups_are_resolved_in_bulk(self):
+        far, *origins = self._create_interfaces('Far', *(f'IF{i}' for i in range(16)))
+        cable = Cable(a_terminations=[far], b_terminations=origins)
+        cable.save()
+        ContentType.objects.get_for_model(Interface)
+
+        # Membership, end rows, generic targets, and cables are fetched in batches, not per origin.
+        with self.assertNumQueries(4):
+            groups = utils.get_cablepath_origin_groups(origins)
+        self.assertEqual(groups, {(cable.pk, CableEndChoices.SIDE_B): origins})
+
+    def test_known_cable_ends_are_loaded_together(self):
+        far, first, second = self._create_interfaces('IF1', 'IF2', 'IF3')
+        cable = Cable(a_terminations=[far], b_terminations=[first, second])
+        cable.save()
+        ContentType.objects.get_for_model(Interface)
+        keys = [(cable.pk, CableEndChoices.SIDE_A), (cable.pk, CableEndChoices.SIDE_B)]
+        with self.assertNumQueries(3):
+            groups = utils.get_cable_end_terminations(keys)
+        self.assertEqual(groups, {keys[0]: [far], keys[1]: [first, second]})
+        with self.assertNumQueries(0):
+            self.assertEqual([obj.cable for group in groups.values() for obj in group], [cable] * 3)
+
+    def test_bulk_end_lookup_does_not_include_the_opposite_ends(self):
+        first_a, first_b, second_a, second_b = self._create_interfaces('IF1', 'IF2', 'IF3', 'IF4')
+        first = Cable(a_terminations=[first_a], b_terminations=[first_b])
+        first.save()
+        second = Cable(a_terminations=[second_a], b_terminations=[second_b])
+        second.save()
+        first_key = (first.pk, CableEndChoices.SIDE_A)
+        second_key = (second.pk, CableEndChoices.SIDE_B)
+        groups = utils.get_cable_end_terminations([first_key, second_key])
+        self.assertEqual(groups, {first_key: [first_a], second_key: [second_b]})
+
+    def test_empty_origin_lookups_do_not_query_the_database(self):
+        with self.assertNumQueries(0):
+            self.assertEqual(utils._get_cablepath_origin_objects([]), {})
+            self.assertEqual(utils.get_cable_end_terminations([]), {})
+            self.assertEqual(utils.get_cablepath_origin_groups([]), {})
+
+    def test_missing_generic_target_is_ignored_when_retracing_a_cable_end(self):
+        far, origin = self._create_interfaces('IF1', 'IF2')
+        cable = Cable(a_terminations=[far], b_terminations=[origin])
+        cable.save()
+        origin.refresh_from_db()
+        origin._path.delete()
+        key = (cable.pk, CableEndChoices.SIDE_B)
+
+        # Bypass save-time validation to represent a stale generic reference, without deleting a live endpoint.
+        missing_id = Interface.objects.order_by('-pk').values_list('pk', flat=True).first() + 1
+        orphan = CableTermination(
+            cable=cable,
+            cable_end=CableEndChoices.SIDE_B,
+            termination_type=ContentType.objects.get_for_model(Interface),
+            termination_id=missing_id,
+        )
+        CableTermination.objects.bulk_create([orphan])
+        self.assertIsNone(CableTermination.objects.get(pk=orphan.pk).termination)
+        self.assertEqual(utils.get_cable_end_terminations([key]), {key: [origin]})
+
+        out, err = StringIO(), StringIO()
+        call_command('trace_paths', stdout=out, stderr=err, no_input=True)
+
+        self.assertCurrentPathExists((origin, cable, far), is_complete=True, is_active=True)
+        self.assertEqual(CablePath.objects.count(), 2)
+        self.assertIn('Finished.', out.getvalue())
+        self.assertEqual(err.getvalue(), '')
+        # Tracing tolerates the dangling reference; it does not silently delete the CableTermination row.
+        self.assertTrue(CableTermination.objects.filter(pk=orphan.pk).exists())
+
+    def test_cable_prefetch_traverses_different_generic_target_types(self):
+        far, origin = self._create_interfaces('IF1', 'IF2')
+        cable = Cable(a_terminations=[far], b_terminations=[origin])
+        cable.save()
+        console = ConsolePort.objects.create(device=self.device, name='Console')
+        server = ConsoleServerPort.objects.create(device=self.device, name='Console server')
+        console_cable = Cable(a_terminations=[console], b_terminations=[server])
+        console_cable.save()
+        keys = [(cable.pk, CableEndChoices.SIDE_B), (console_cable.pk, CableEndChoices.SIDE_A)]
+        for model in (Interface, ConsolePort):
+            ContentType.objects.get_for_model(model)
+
+        # End rows, two generic target types, then their shared cable relation: no per-origin queries.
+        with self.assertNumQueries(4):
+            groups = utils.get_cable_end_terminations(keys)
+        self.assertEqual(groups, {keys[0]: [origin], keys[1]: [console]})
+        with self.assertNumQueries(0):
+            self.assertEqual(
+                [obj.cable for key in keys for obj in groups[key]], [cable, console_cable]
+            )
+
+
+class TracePathsRecoveryTestCase(BaseCablePathTestCase):
+
+    def test_progress_bar_updates_for_an_already_retraced_cable_end(self):
+        class FakeQuerySet(list):
+            def filter(self, *args, **kwargs):
+                return self
+
+            def count(self):
+                return len(self)
+
+            def annotate(self, **kwargs):
+                return self
+
+            def order_by(self, *fields):
+                return self
+
+            def iterator(self, chunk_size):
+                return iter(self)
+
+        endpoint = SimpleNamespace(pk=1, cable_id=1, _trace_cable_id=1, _trace_cable_end='B')
+        origins = FakeQuerySet([endpoint] * 200)
+        model = SimpleNamespace(
+            objects=SimpleNamespace(filter=lambda *args, **kwargs: origins),
+            _meta=SimpleNamespace(verbose_name='interface', verbose_name_plural='interfaces'),
+        )
+        command = trace_paths.Command(stdout=StringIO())
+        with (
+            patch.object(trace_paths, 'ENDPOINT_MODELS', (model,)),
+            patch.object(trace_paths, 'get_cable_end_terminations', return_value={(1, 'B'): [endpoint]}) as fetch,
+            patch.object(trace_paths, 'create_cablepaths') as create,
+            patch.object(command, 'draw_progress_bar') as progress,
+        ):
+            command.handle(force=False, no_input=True)
+
+        create.assert_called_once_with([endpoint])
+        fetch.assert_called_once_with([(1, 'B')])
+        self.assertEqual(progress.call_args_list, [call(50), call(100), call(100)])
+
+    def test_group_batches_preserve_counts_progress_and_singleton_fallbacks(self):
+        class FakeQuerySet(list):
+            def filter(self, *args, **kwargs):
+                return self
+
+            def count(self):
+                return len(self)
+
+            def annotate(self, **kwargs):
+                return self
+
+            def order_by(self, *fields):
+                return self
+
+            def iterator(self, chunk_size):
+                return iter(self)
+
+        origins = FakeQuerySet()
+        memberships = {}
+        # More than one batch, and a group spanning a progress boundary. Counts must survive buffering.
+        for cable_id in range(1, trace_paths.ORIGIN_GROUP_BATCH_SIZE + 3):
+            members = [
+                SimpleNamespace(
+                    pk=len(origins) + offset, _trace_cable_id=cable_id, _trace_cable_end='B'
+                ) for offset in range(1, 102 if cable_id == 1 else 2)
+            ]
+            origins.extend(members)
+            memberships[(cable_id, 'B')] = members
+        fallbacks = [
+            SimpleNamespace(pk=len(origins) + offset, _trace_cable_id=None, _trace_cable_end=None)
+            for offset in (1, 2)
+        ]
+        origins.extend(fallbacks)
+        model = SimpleNamespace(
+            objects=SimpleNamespace(filter=lambda *args, **kwargs: origins),
+            _meta=SimpleNamespace(verbose_name='interface', verbose_name_plural='interfaces'),
+        )
+        out, err = StringIO(), StringIO()
+        command = trace_paths.Command(stdout=out, stderr=err)
+        failed_key = (trace_paths.ORIGIN_GROUP_BATCH_SIZE, 'B')
+
+        def fetch_memberships(keys):
+            return {key: memberships[key] for key in keys}
+
+        def trace_group(group):
+            if group is memberships[failed_key]:
+                raise UnsupportedCablePath('Simulated group failure')
+
+        with (
+            patch.object(trace_paths, 'ENDPOINT_MODELS', (model,)),
+            patch.object(trace_paths, 'get_cable_end_terminations', side_effect=fetch_memberships) as fetch,
+            patch.object(trace_paths, 'create_cablepaths', side_effect=trace_group) as trace,
+            patch.object(command, 'draw_progress_bar') as progress,
+        ):
+            with self.assertRaisesMessage(CommandError, 'Unable to trace 1 origin group(s)'):
+                command.handle(force=False, no_input=True)
+
+        keys = list(memberships)
+        self.assertEqual(fetch.call_args_list, [
+            call(keys[:trace_paths.ORIGIN_GROUP_BATCH_SIZE]), call(keys[trace_paths.ORIGIN_GROUP_BATCH_SIZE:]),
+        ])
+        self.assertEqual(trace.call_args_list, [
+            *[call(members) for members in memberships.values()], call([fallbacks[0]]), call([fallbacks[1]]),
+        ])
+        self.assertEqual(progress.call_args_list, [
+            *[call(count * 100 / len(origins)) for count in range(100, len(origins) + 1, 100)], call(100),
+        ])
+        self.assertIn(f'Retraced {len(origins) - 1} interfaces', out.getvalue())
+        self.assertIn('Failed to retrace 1 selected interfaces in 1 origin group(s)', out.getvalue())
+        self.assertNotIn('Finished.', out.getvalue())
+        self.assertEqual(err.getvalue().count('Unable to trace'), 1)
+
+    def test_batched_end_lookup_continues_real_repairs_after_a_failed_group(self):
+        groups = []
+        for index, names in enumerate((('A1', 'A2', 'Drifted'), ('B1', 'B2'), ('C1',))):
+            far = Interface.objects.create(device=self.device, name=f'Far{index}')
+            origins = [Interface.objects.create(device=self.device, name=name) for name in names]
+            cable = Cable(a_terminations=[far], b_terminations=origins)
+            cable.save()
+            origins[0].refresh_from_db()
+            origins[0]._path.delete()
+            groups.append((cable, far, origins))
+        bad, _, bad_origins = groups[0]
+        Interface.objects.filter(pk=bad_origins[-1].pk).update(cable=None)
+        out, err = StringIO(), StringIO()
+
+        with (
+            patch.object(trace_paths, 'ENDPOINT_MODELS', (Interface,)),
+            patch.object(trace_paths, 'ORIGIN_GROUP_BATCH_SIZE', 2),
+            patch.object(trace_paths, 'get_cable_end_terminations', wraps=utils.get_cable_end_terminations) as fetch,
+        ):
+            with self.assertRaisesMessage(CommandError, 'Unable to trace 1 origin group(s)'):
+                call_command('trace_paths', no_input=True, stdout=out, stderr=err)
+
+        keys = [(cable.pk, CableEndChoices.SIDE_B) for cable, _, _ in groups]
+        self.assertEqual(fetch.call_args_list, [call(keys[:2]), call(keys[2:])])
+        self.assertEqual(err.getvalue().count(f'Unable to trace cable #{bad.pk} end B:'), 1)
+        self.assertIn('Retraced 3 interfaces', out.getvalue())
+        self.assertIn('Failed to retrace 2 selected interfaces in 1 origin group(s)', out.getvalue())
+        self.assertNotIn('Finished.', out.getvalue())
+        for cable, far, origins in groups[1:]:
+            path = self.assertPathExists((origins, cable, far), is_complete=True, is_active=True)
+            for origin in origins:
+                origin.refresh_from_db()
+                self.assertPathIsSet(origin, path)
+            self.assertCurrentPathExists((far, cable, origins), is_complete=True, is_active=True)
+        for origin in bad_origins:
+            origin.refresh_from_db()
+            self.assertPathIsNotSet(origin)
+        self.assertEqual(CablePath.objects.count(), 5)
+
+    def test_empty_resolved_end_is_reported_without_preventing_later_repairs(self):
+        missing, missing_peer, good, good_peer = [
+            Interface.objects.create(device=self.device, name=name)
+            for name in ('Missing', 'Missing peer', 'Good', 'Good peer')
+        ]
+        missing_cable = Cable(a_terminations=[missing], b_terminations=[missing_peer])
+        missing_cable.save()
+        good_cable = Cable(a_terminations=[good], b_terminations=[good_peer])
+        good_cable.save()
+        Interface.objects.filter(pk__in=[missing.pk, good.pk]).update(_path=None)
+        missing_key = (missing_cable.pk, CableEndChoices.SIDE_A)
+        out, err = StringIO(), StringIO()
+        fetch_ends = utils.get_cable_end_terminations
+
+        def lose_membership(keys):
+            groups = fetch_ends(keys)
+            # Simulate membership disappearing between endpoint selection and the batched end lookup.
+            if missing_key in groups:
+                groups[missing_key] = []
+            return groups
+
+        with (
+            patch.object(trace_paths, 'ENDPOINT_MODELS', (Interface,)),
+            patch.object(trace_paths, 'get_cable_end_terminations', side_effect=lose_membership),
+            patch.object(trace_paths, 'create_cablepaths', wraps=utils.create_cablepaths) as traced,
+        ):
+            with self.assertRaisesMessage(CommandError, 'Unable to trace 1 origin group(s)'):
+                call_command('trace_paths', no_input=True, stdout=out, stderr=err)
+
+        traced.assert_called_once_with([good])
+        self.assertIn(f'Unable to trace cable #{missing_cable.pk} end A:', err.getvalue())
+        self.assertIn('No current terminations remain', err.getvalue())
+        self.assertIn('Retraced 1 interfaces', out.getvalue())
+        self.assertIn('Failed to retrace 1 selected interfaces in 1 origin group(s)', out.getvalue())
+        self.assertIn('100%', out.getvalue())
+        self.assertNotIn('Finished.', out.getvalue())
+        missing.refresh_from_db()
+        self.assertPathIsNotSet(missing)
+        self.assertCurrentPathExists((good, good_cable, good_peer), is_complete=True)
+        self.assertCurrentPathExists((good_peer, good_cable, good), is_complete=True)
+
+    def test_unsupported_group_does_not_prevent_later_repairs(self):
+        for force in (False, True):
+            with self.subTest(force=force), transaction.atomic():
+                bad_far, first, second, drifted, good_a, good_b = [
+                    Interface.objects.create(device=self.device, name=name)
+                    for name in ('A0', 'A1', 'A2', 'A3', 'Z0', 'Z1')
+                ]
+                bad = Cable(a_terminations=[bad_far], b_terminations=[first, second, drifted])
+                bad.save()
+                good = Cable(a_terminations=[good_a], b_terminations=[good_b])
+                good.save()
+                # The CableTermination survives; grouping the selected siblings exposes this cached-FK drift.
+                Interface.objects.filter(pk=drifted.pk).update(cable=None)
+                Interface.objects.filter(pk__in=[first.pk, second.pk, good_a.pk]).update(_path=None)
+                before_refs = dict(Interface.objects.filter(
+                    pk__in=[first.pk, second.pk, drifted.pk]
+                ).values_list('pk', '_path_id'))
+                before_paths = set(CablePath.objects.filter(_nodes__contains=bad).values_list('pk', flat=True))
+                out, err = StringIO(), StringIO()
+                create = utils.create_cablepaths
+                attempts = []
+
+                def record_attempt(origins):
+                    attempts.append(tuple(obj.pk for obj in origins))
+                    return create(origins)
+
+                with patch.object(trace_paths, 'create_cablepaths', side_effect=record_attempt):
+                    with self.assertRaisesMessage(CommandError, 'Unable to trace 1 origin group(s)'):
+                        call_command('trace_paths', force=force, no_input=True, stdout=out, stderr=err)
+
+                failed_group = (first.pk, second.pk, drifted.pk)
+                self.assertEqual(attempts.count(failed_group), 1)
+                self.assertLess(attempts.index(failed_group), attempts.index((good_a.pk,)))
+                self.assertEqual(err.getvalue().count(f'Unable to trace cable #{bad.pk} end B:'), 1)
+                self.assertNotIn('Finished.', out.getvalue())
+                self.assertCurrentPathExists((good_a, good, good_b), is_complete=True)
+                self.assertCurrentPathExists((good_b, good, good_a), is_complete=True)
+                if force:
+                    # --force's earlier global deletion is deliberately not restored by a failed group.
+                    for obj in (first, second, drifted):
+                        obj.refresh_from_db()
+                        self.assertPathIsNotSet(obj)
+                else:
+                    self.assertEqual(
+                        set(CablePath.objects.filter(_nodes__contains=bad).values_list('pk', flat=True)), before_paths
+                    )
+                    for obj in (first, second, drifted):
+                        obj.refresh_from_db()
+                        self.assertEqual(obj._path_id, before_refs[obj.pk])
+                    self.assertEqual(CablePath.objects.count(), 4)
+                transaction.set_rollback(True)
+
+    def test_force_rebuild_removes_stale_rows_and_preserves_the_joint_hop(self):
+        far, first, second = [
+            Interface.objects.create(device=self.device, name=f'IF{i}') for i in range(3)
+        ]
+        cable = Cable(a_terminations=[far], b_terminations=[first, second])
+        cable.save()
+        second.refresh_from_db()
+        CablePath.from_origin([second]).save()
+        self.assertEqual(CablePath.objects.count(), 3)
+        out = StringIO()
+
+        call_command('trace_paths', force=True, no_input=True, stdout=out)
+
+        self.assertIn('Finished.', out.getvalue())
+        self.assertEqual(CablePath.objects.count(), 2)
+        self.assertCurrentPathExists((far, cable, [first, second]), is_complete=True)
+        joint = self.assertPathExists(([first, second], cable, far), is_complete=True)
+        for origin in (first, second):
+            origin.refresh_from_db()
+            self.assertPathIsSet(origin, joint)

netbox/dcim/tests/test_cablepaths.py

@@ -1,8 +1,13 @@
+from io import StringIO
+
+from django.core.management import call_command
+
 from circuits.models import *
-from dcim.choices import LinkStatusChoices
+from dcim.choices import CableEndChoices, LinkStatusChoices
 from dcim.models import *
 from dcim.svg import CableTraceSVG
 from dcim.tests.utils import BaseCablePathTestCase
+from dcim.utils import create_cablepaths, object_to_path_node
 from utilities.exceptions import AbortRequest
 
 
@@ -2891,6 +2896,307 @@ class LegacyCablePathTestCase(BaseCablePathTestCase):
         # Verify _path is cleared on removed interface (#21127)
         interface3.refresh_from_db()
         self.assertPathIsNotSet(interface3)
+        self.assertEqual(CablePath.objects.count(), 2)
+
+    def test_304_replacing_a_termination_retires_superseded_paths(self):
+        """
+        [IF1] --C1-- [IF2] becomes [IF1] --C1-- [IF3], and back again
+
+        Each replacement must leave only the two paths the cable's current terminations trace.
+        """
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        interface2 = Interface.objects.create(device=self.device, name='Interface 2')
+        interface3 = Interface.objects.create(device=self.device, name='Interface 3')
+
+        cable1 = Cable(a_terminations=[interface1], b_terminations=[interface2])
+        cable1.save()
+        self.assertEqual(CablePath.objects.count(), 2)
+
+        for peer, detached in ((interface3, interface2), (interface2, interface3)):
+            with self.subTest(peer=peer.name):
+                cable1 = Cable.objects.get(pk=cable1.pk)
+                cable1.b_terminations = [peer]
+                cable1.full_clean()
+                cable1.save()
+
+                self.assertCurrentPathExists((interface1, cable1, peer), is_complete=True, is_active=True)
+                self.assertCurrentPathExists((peer, cable1, interface1), is_complete=True, is_active=True)
+                self.assertEqual(CablePath.objects.count(), 2)
+                detached.refresh_from_db()
+                self.assertIsNone(detached.cable)
+                self.assertPathIsNotSet(detached)
+
+    def test_305_adding_a_termination_retires_superseded_paths(self):
+        """
+        [IF1] --C1-- [IF2] gains a second B-side termination [IF3]
+
+        Extending an end must retire the paths whose destinations the extension supersedes.
+        """
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        interface2 = Interface.objects.create(device=self.device, name='Interface 2')
+        interface3 = Interface.objects.create(device=self.device, name='Interface 3')
+
+        cable1 = Cable(a_terminations=[interface1], b_terminations=[interface2])
+        cable1.save()
+        self.assertEqual(CablePath.objects.count(), 2)
+
+        cable1 = Cable.objects.get(pk=cable1.pk)
+        cable1.b_terminations = [interface2, interface3]
+        cable1.full_clean()
+        cable1.save()
+
+        self.assertCurrentPathExists(
+            (interface1, cable1, [interface2, interface3]), is_complete=True, is_active=True
+        )
+        path2 = self.assertPathExists(
+            ([interface2, interface3], cable1, interface1), is_complete=True, is_active=True
+        )
+        for interface in (interface2, interface3):
+            interface.refresh_from_db()
+            self.assertPathIsSet(interface, path2)
+        self.assertEqual(CablePath.objects.count(), 2)
+
+    def test_306_retracing_one_joint_origin_restores_the_whole_hop(self):
+        """
+        [IF1] --C1-- [IF2]
+                     [IF3]
+
+        trace_paths retraces a cable end as a unit, so repairing one co-origin restores the joint hop.
+        """
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        interface2 = Interface.objects.create(device=self.device, name='Interface 2')
+        interface3 = Interface.objects.create(device=self.device, name='Interface 3')
+
+        cable1 = Cable(a_terminations=[interface1], b_terminations=[interface2, interface3])
+        cable1.save()
+        self.assertEqual(CablePath.objects.count(), 2)
+
+        # trace_paths selects on a null _path
+        Interface.objects.filter(pk=interface2.pk).update(_path=None)
+
+        call_command('trace_paths', no_input=True, stdout=StringIO())
+
+        for interface in (interface1, interface2, interface3):
+            interface.refresh_from_db()
+            self.assertIsNotNone(interface._path_id, msg=f'{interface} left without a path')
+        self.assertPathExists(([interface2, interface3], cable1, interface1))
+        self.assertEqual(CablePath.objects.count(), 2)
+
+    def test_307_retracing_a_cable_end_retires_a_superseded_co_origin_path(self):
+        """
+        [IF1] --C1-- [IF2]
+                     [IF3]
+
+        A path superseded at a co-origin of the retraced end is retired, not left beside its replacement.
+        """
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        interface2 = Interface.objects.create(device=self.device, name='Interface 2')
+        interface3 = Interface.objects.create(device=self.device, name='Interface 3')
+
+        cable1 = Cable(a_terminations=[interface1], b_terminations=[interface2, interface3])
+        cable1.save()
+        self.assertEqual(CablePath.objects.count(), 2)
+
+        # Seed the stale row a pre-fix install would hold, then restore both origins to the shared path
+        interface3 = Interface.objects.get(pk=interface3.pk)
+        joint_path = interface3._path
+        superseded = CablePath.from_origin([interface3])
+        superseded.save()
+        joint_path.save()
+
+        # trace_paths selects on a null _path
+        Interface.objects.filter(pk=interface2.pk).update(_path=None)
+
+        call_command('trace_paths', no_input=True, stdout=StringIO())
+
+        self.assertFalse(
+            CablePath.objects.filter(pk=superseded.pk).exists(), msg='the superseded path survived the retrace'
+        )
+        self.assertPathExists(([interface2, interface3], cable1, interface1))
+        self.assertEqual(CablePath.objects.count(), 2)
+
+    def test_308_retracing_a_partial_origin_group_retires_its_co_origins_paths(self):
+        """
+        Retracing one origin of a shared path must also replace its co-origin's stale paths.
+        """
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        interface2 = Interface.objects.create(device=self.device, name='Interface 2')
+        interface3 = Interface.objects.create(device=self.device, name='Interface 3')
+        cable = Cable(a_terminations=[interface1], b_terminations=[interface2, interface3])
+        cable.save()
+
+        interface3.refresh_from_db()
+        joint_path = interface3._path
+        superseded = CablePath.from_origin([interface3])
+        superseded.save()
+        joint_path.save()
+
+        interface2.refresh_from_db()
+        create_cablepaths([interface2])
+
+        self.assertFalse(CablePath.objects.filter(pk=superseded.pk).exists())
+        self.assertFalse(CablePath.objects.filter(pk=joint_path.pk).exists())
+        self.assertCurrentPathExists((interface2, cable, interface1), is_complete=True)
+        self.assertCurrentPathExists((interface3, cable, interface1), is_complete=True)
+        self.assertCurrentPathExists((interface1, cable, [interface2, interface3]), is_complete=True)
+        self.assertEqual(CablePath.objects.count(), 3)
+
+    def test_309_retracing_preserves_another_current_origin_group(self):
+        """
+        An origin group whose references the deletion does not clear is left intact, rows and all.
+        """
+        interfaces = [
+            Interface.objects.create(device=self.device, name=f'Interface {i}') for i in range(1, 5)
+        ]
+        interface1, interface2, interface3, interface4 = interfaces
+        cable = Cable(a_terminations=[interface1], b_terminations=interfaces[1:])
+        cable.save()
+        for interface in interfaces:
+            interface.refresh_from_db()
+
+        # Overlapping groups from earlier partial retraces, with interface3 and interface4 on the second
+        interface2._path.delete()
+        first = CablePath.from_origin([interface2, interface3])
+        first.save()
+        second = CablePath.from_origin([interface3, interface4])
+        second.save()
+
+        create_cablepaths([interface2])
+
+        self.assertFalse(CablePath.objects.filter(pk=first.pk).exists())
+        self.assertCurrentPathExists((interface2, cable, interface1), is_complete=True)
+        self.assertTrue(
+            CablePath.objects.filter(pk=second.pk).exists(), msg='the untouched origin group was replaced'
+        )
+        second.refresh_from_db()
+        self.assertEqual(second.path[0], [object_to_path_node(interface3), object_to_path_node(interface4)])
+        for interface in (interface3, interface4):
+            interface.refresh_from_db()
+            self.assertPathIsSet(interface, second)
+        self.assertEqual(CablePath.objects.count(), 3)
+
+    def test_310_recovery_preserves_the_requested_connector_groups(self):
+        """
+        A recovered group must not replace a connector group already rebuilt by this call.
+        """
+        interfaces = [
+            Interface.objects.create(device=self.device, name=f'Interface {i}') for i in range(1, 5)
+        ]
+        interface1, interface2, interface3, interface4 = interfaces
+        cable = Cable(a_terminations=[interface1], b_terminations=interfaces[1:])
+        cable.save()
+        for interface in interfaces:
+            interface.refresh_from_db()
+
+        CablePath.from_origin([interface2, interface3]).save()
+        # Use a connector grouping that cannot be persisted to isolate requested-group precedence.
+        interface2.cable_connector = 1
+        interface3.cable_connector = 2
+        interface4.cable_connector = 2
+
+        create_cablepaths([interface2, interface3, interface4])
+
+        self.assertCurrentPathExists((interface2, cable, interface1), is_complete=True)
+        shared_path = self.assertPathExists(([interface3, interface4], cable, interface1), is_complete=True)
+        for interface in (interface3, interface4):
+            interface.refresh_from_db()
+            self.assertPathIsSet(interface, shared_path)
+        self.assertEqual(CablePath.objects.count(), 3)
+
+    def test_311_recovery_separates_origins_on_different_links(self):
+        """
+        A stale origin hop holding origins since moved apart is recovered as one group per current link.
+        """
+        interfaces = [
+            Interface.objects.create(device=self.device, name=f'Interface {i}') for i in range(1, 7)
+        ]
+        interface1, interface2, interface3, interface4, interface5, interface6 = interfaces
+        cable1 = Cable(
+            a_terminations=[interface1],
+            b_terminations=[interface2, interface3, interface4, interface5],
+        )
+        cable1.save()
+
+        # The hop as stored while all four shared cable1
+        interface2.refresh_from_db()
+        stale_nodes = interface2._path.path
+
+        # Move one origin through ordinary saves, so only the CablePath below is stale
+        cable1.b_terminations = [interface2, interface3, interface4]
+        cable1.save()
+        interface5.refresh_from_db()
+        cable2 = Cable(a_terminations=[interface6], b_terminations=[interface5])
+        cable2.save()
+
+        interface2.refresh_from_db()
+        interface5.refresh_from_db()
+        current_cable1 = interface2._path
+        current_cable2 = interface5._path
+
+        # Leave the obsolete row as their only originating path, so recovery gets the whole mixed hop
+        current_cable1.delete()
+        current_cable2.delete()
+        superseded = CablePath(path=stale_nodes, is_complete=True, is_active=True)
+        superseded.save()
+
+        create_cablepaths([Interface.objects.get(pk=interface2.pk)])
+
+        self.assertFalse(CablePath.objects.filter(pk=superseded.pk).exists())
+        self.assertCurrentPathExists((interface2, cable1, interface1), is_complete=True)
+        # The origins still sharing cable1 stay one group, and the moved one is recovered on its own cable
+        shared = self.assertPathExists(([interface3, interface4], cable1, interface1), is_complete=True)
+        for interface in (interface3, interface4):
+            interface.refresh_from_db()
+            self.assertPathIsSet(interface, shared)
+        self.assertCurrentPathExists((interface5, cable2, interface6), is_complete=True)
+        self.assertEqual(CablePath.objects.count(), 5)
+
+    def test_312_retracing_repairs_an_endpoint_whose_cable_end_drifted(self):
+        """
+        An endpoint whose denormalized cable_end no longer matches its CableTermination is still retraced.
+        """
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        interface2 = Interface.objects.create(device=self.device, name='Interface 2')
+        cable1 = Cable(a_terminations=[interface1], b_terminations=[interface2])
+        cable1.save()
+
+        # The termination row still says B, which is the drift this command exists to repair
+        Interface.objects.filter(pk=interface2.pk).update(
+            _path=None, cable_end=CableEndChoices.SIDE_A
+        )
+
+        call_command('trace_paths', no_input=True, stdout=StringIO())
+
+        self.assertCurrentPathExists((interface2, cable1, interface1), is_complete=True)
+        self.assertEqual(CablePath.objects.count(), 2)
+
+    def test_313_recovery_follows_a_chain_of_cleared_references(self):
+        """
+        Recovering an origin can clear another group's references, whose origins are recovered in turn.
+        """
+        interfaces = [
+            Interface.objects.create(device=self.device, name=f'Interface {i}') for i in range(1, 5)
+        ]
+        interface1, interface2, interface3, interface4 = interfaces
+        cable = Cable(a_terminations=[interface1], b_terminations=interfaces[1:])
+        cable.save()
+        for interface in interfaces:
+            interface.refresh_from_db()
+
+        # interface3 references the first group, so losing it starts the chain
+        interface2._path.delete()
+        first = CablePath.from_origin([interface2, interface3])
+        first.save()
+        second = CablePath.from_origin([interface3, interface4])
+        second.save()
+        first.save()
+
+        create_cablepaths([interface2])
+
+        self.assertFalse(CablePath.objects.filter(pk__in=[first.pk, second.pk]).exists())
+        for interface in interfaces[1:]:
+            self.assertCurrentPathExists((interface, cable, interface1), is_complete=True)
+        self.assertEqual(CablePath.objects.count(), 4)
 
     def test_401_exclude_midspan_devices(self):
         """

netbox/dcim/tests/test_cablepaths2.py

@@ -2785,3 +2785,68 @@ class CablePathTestCase(BaseCablePathTestCase):
             set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)),
             termination_pks
         )
+
+    def test_311_moving_a_midspan_termination_preserves_far_end_paths(self):
+        """
+        [IF1] --C1-- [FP1][RP1] --C3-- [RP2][FP2] --C2-- [IF2]
+        becomes
+        [IF1] --C1-- [FP1][RP1] --C3-- [RP3][FP3] --C4-- [IF3]
+
+        Rear ports originate no path, so moving a mid-span cable's end must retrace the rows which traverse it
+        rather than delete them: their origins cannot be recovered from the cable's own terminations.
+        """
+        interfaces = [
+            Interface.objects.create(device=self.device, name=f'Interface {i}') for i in range(1, 4)
+        ]
+        rear_ports = [
+            RearPort.objects.create(device=self.device, name=f'Rear Port {i}') for i in range(1, 4)
+        ]
+        front_ports = [
+            FrontPort.objects.create(device=self.device, name=f'Front Port {i}') for i in range(1, 4)
+        ]
+        for front_port, rear_port in zip(front_ports, rear_ports):
+            PortMapping.objects.create(
+                device=self.device,
+                front_port=front_port,
+                front_port_position=1,
+                rear_port=rear_port,
+                rear_port_position=1
+            )
+
+        cable1 = Cable(a_terminations=[interfaces[0]], b_terminations=[front_ports[0]])
+        cable1.clean()
+        cable1.save()
+        cable2 = Cable(a_terminations=[front_ports[1]], b_terminations=[interfaces[1]])
+        cable2.clean()
+        cable2.save()
+        cable3 = Cable(a_terminations=[rear_ports[0]], b_terminations=[rear_ports[1]])
+        cable3.clean()
+        cable3.save()
+        cable4 = Cable(a_terminations=[front_ports[2]], b_terminations=[interfaces[2]])
+        cable4.clean()
+        cable4.save()
+
+        before = (
+            interfaces[0], cable1, front_ports[0], rear_ports[0], cable3, rear_ports[1], front_ports[1], cable2,
+            interfaces[1],
+        )
+        self.assertCurrentPathExists(before, is_complete=True, is_active=True)
+        self.assertCurrentPathExists(tuple(reversed(before)), is_complete=True, is_active=True)
+        # Two for the completed link, one for the third interface stopping at its own rear port
+        self.assertEqual(CablePath.objects.count(), 3)
+
+        cable3 = Cable.objects.get(pk=cable3.pk)
+        cable3.b_terminations = [rear_ports[2]]
+        cable3.full_clean()
+        cable3.save()
+
+        after = (
+            interfaces[0], cable1, front_ports[0], rear_ports[0], cable3, rear_ports[2], front_ports[2], cable4,
+            interfaces[2],
+        )
+        self.assertCurrentPathExists(after, is_complete=True, is_active=True)
+        self.assertCurrentPathExists(tuple(reversed(after)), is_complete=True, is_active=True)
+        self.assertCurrentPathExists(
+            (interfaces[1], cable2, front_ports[1], rear_ports[1]), is_complete=False
+        )
+        self.assertEqual(CablePath.objects.count(), 3)

netbox/dcim/tests/test_channelization.py

@@ -12,6 +12,7 @@ from rest_framework.test import APIClient
 from core.choices import ObjectChangeActionChoices
 from core.models import ObjectChange
 from dcim.choices import CableProfileChoices, InterfaceTypeChoices
+from dcim.exceptions import UnsupportedCablePath
 from dcim.filtersets import InterfaceFilterSet
 from dcim.models import (
     Cable,
@@ -27,6 +28,7 @@ from dcim.models import (
 from dcim.svg import CableTraceSVG
 from dcim.svg.cables import Connector
 from dcim.tests.utils import BaseCablePathTestCase
+from dcim.utils import create_cablepaths, object_to_path_node
 from users.constants import TOKEN_PREFIX
 from users.models import Token, User
 from utilities.ordering import naturalize_interface
@@ -468,6 +470,243 @@ class ChannelizedCablePathTestCase(BaseCablePathTestCase):
         self.assertIsNone(channel.cable_positions)
         self.assertPathIsNotSet(channel)
 
+    def test_114_replacing_a_far_end_retires_the_channels_superseded_paths(self):
+        """
+        Replacing one far-end interface of a breakout cable must leave one path per channel in each direction. The
+        channels are the real origins, so the rows the retrace supersedes can only be found after the expansion.
+        """
+        parent, channels = self._create_channelized_interface('et0', 4)
+        far = [
+            Interface.objects.create(device=self.device, name=f'xe{i}', type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS)
+            for i in range(4)
+        ]
+        replacement = Interface.objects.create(
+            device=self.device, name='xe4', type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS
+        )
+
+        cable = Cable(
+            profile=CableProfileChoices.BREAKOUT_1C4P_4C1P,
+            a_terminations=[parent],
+            b_terminations=far,
+        )
+        cable.clean()
+        cable.save()
+        self.assertEqual(CablePath.objects.count(), 8)
+
+        cable = Cable.objects.get(pk=cable.pk)
+        cable.b_terminations = [*far[:3], replacement]
+        cable.clean()
+        cable.save()
+
+        self.assertEqual(CablePath.objects.count(), 8)
+        for channel, far_iface in zip(channels, [*far[:3], replacement]):
+            channel.refresh_from_db()
+            far_iface.refresh_from_db()
+            forward = self.assertPathExists((channel, cable, far_iface), is_complete=True, is_active=True)
+            reverse = self.assertPathExists((far_iface, cable, channel), is_complete=True, is_active=True)
+            self.assertPathIsSet(channel, forward)
+            self.assertPathIsSet(far_iface, reverse)
+        far[3].refresh_from_db()
+        self.assertIsNone(far[3].cable)
+        self.assertPathIsNotSet(far[3])
+
+    def test_115_failed_channel_trace_restores_the_replaced_paths(self):
+        """
+        A trace that raises partway through a channelized end must leave every channel's stored path in place.
+        """
+        parent, channels = self._create_channelized_interface('et0', 2)
+        far = [
+            Interface.objects.create(device=self.device, name=f'xe{i}', type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS)
+            for i in range(2)
+        ]
+        cable = Cable(
+            profile=CableProfileChoices.BREAKOUT_1C2P_2C1P,
+            a_terminations=[parent],
+            b_terminations=far,
+        )
+        cable.clean()
+        cable.save()
+        self.assertEqual(CablePath.objects.count(), 4)
+
+        stored_paths = {}
+        for channel in channels:
+            channel.refresh_from_db()
+            stored_paths[channel.pk] = channel._path
+
+        traced = CablePath.from_origin
+
+        def failing_from_origin(terminations):
+            if terminations and terminations[0] == channels[1]:
+                # The first channel's path must already be replaced, or the rollback proves nothing
+                self.assertFalse(CablePath.objects.filter(pk=stored_paths[channels[0].pk].pk).exists())
+                raise UnsupportedCablePath('Simulated trace error')
+            return traced(terminations)
+
+        with mock.patch.object(CablePath, 'from_origin', side_effect=failing_from_origin):
+            with self.assertRaises(UnsupportedCablePath):
+                create_cablepaths([parent])
+
+        # Reaching a query at all proves the block kept its savepoint
+        self.assertEqual(CablePath.objects.count(), 4)
+        for channel in channels:
+            channel.refresh_from_db()
+            self.assertPathIsSet(channel, stored_paths[channel.pk])
+
+    def test_116_recovery_preserves_channel_and_connector_groups(self):
+        """
+        Recovering a stale hop must keep both channels and breakout connectors in separate paths.
+        """
+        parent, channels = self._create_channelized_interface('et0', 4)
+        far = [
+            Interface.objects.create(device=self.device, name=f'xe{i}', type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS)
+            for i in range(4)
+        ]
+        cable = Cable(
+            profile=CableProfileChoices.BREAKOUT_1C4P_4C1P,
+            a_terminations=[parent],
+            b_terminations=far,
+        )
+        cable.clean()
+        cable.save()
+        self.assertEqual(CablePath.objects.count(), 8)
+
+        for side, origins, peers in (('channels', channels, far), ('connectors', far, channels)):
+            with self.subTest(side=side):
+                peer_paths = {}
+                for peer in peers:
+                    peer.refresh_from_db()
+                    peer_paths[peer.pk] = peer._path_id
+
+                # Keep the valid wiring, but replace this end's current paths with one obsolete origin hop.
+                for origin in origins:
+                    origin.refresh_from_db()
+                    origin._path.delete()
+                superseded = CablePath(
+                    path=[
+                        [object_to_path_node(origin) for origin in origins],
+                        [object_to_path_node(cable)],
+                        [object_to_path_node(peer) for peer in peers],
+                    ],
+                    is_complete=True,
+                    is_active=True,
+                )
+                superseded.save()
+
+                # Request just one origin so the other three must pass through recovery.
+                create_cablepaths([origins[0]])
+
+                self.assertFalse(CablePath.objects.filter(pk=superseded.pk).exists())
+                self.assertEqual(CablePath.objects.count(), 8)
+                for origin, peer in zip(origins, peers):
+                    self.assertCurrentPathExists((origin, cable, peer), is_complete=True, is_active=True)
+                    self.assertCurrentPathExists(
+                        (peer, cable, origin), pk=peer_paths[peer.pk], is_complete=True, is_active=True
+                    )
+
+    def test_117_failed_recovery_restores_the_previous_paths(self):
+        """
+        Failure after deleting a shared origin path must restore that row and all its origin references.
+        """
+        parent, channels = self._create_channelized_interface('et0', 2)
+        far = [
+            Interface.objects.create(device=self.device, name=f'xe{i}', type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS)
+            for i in range(2)
+        ]
+        cable = Cable(
+            profile=CableProfileChoices.BREAKOUT_1C2P_2C1P,
+            a_terminations=[parent],
+            b_terminations=far,
+        )
+        cable.clean()
+        cable.save()
+        for channel in channels:
+            channel.refresh_from_db()
+            channel._path.delete()
+        superseded = CablePath(
+            path=[
+                [object_to_path_node(channel) for channel in channels],
+                [object_to_path_node(cable)],
+                [object_to_path_node(peer) for peer in far],
+            ],
+            is_complete=True,
+            is_active=True,
+        )
+        superseded.save()
+        original_ids = set(CablePath.objects.values_list('pk', flat=True))
+        traced = CablePath.from_origin
+
+        def failing_from_origin(terminations):
+            if terminations and terminations[0] == channels[1]:
+                self.assertFalse(CablePath.objects.filter(pk=superseded.pk).exists())
+                raise UnsupportedCablePath('Simulated recovery error')
+            return traced(terminations)
+
+        with mock.patch.object(CablePath, 'from_origin', side_effect=failing_from_origin):
+            with self.assertRaises(UnsupportedCablePath):
+                create_cablepaths([channels[0]])
+
+        self.assertEqual(set(CablePath.objects.values_list('pk', flat=True)), original_ids)
+        for channel in channels:
+            channel.refresh_from_db()
+            self.assertPathIsSet(channel, superseded)
+
+    def test_118_empty_origins_do_not_query_the_database(self):
+        with self.assertNumQueries(0):
+            create_cablepaths([])
+
+    def test_119_recovery_expands_a_channelized_parent(self):
+        """
+        A recovered parent contributes its channels, without retracing a channel already handled by this call.
+        """
+        parent, channels = self._create_channelized_interface('et0', 2)
+        far = [
+            Interface.objects.create(device=self.device, name=f'xe{i}', type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS)
+            for i in range(2)
+        ]
+        cable = Cable(
+            profile=CableProfileChoices.BREAKOUT_1C2P_2C1P,
+            a_terminations=[parent],
+            b_terminations=far,
+        )
+        cable.clean()
+        cable.save()
+        peer_paths = {}
+        for peer in far:
+            peer.refresh_from_db()
+            peer_paths[peer.pk] = peer._path_id
+        for channel in channels:
+            channel.refresh_from_db()
+            channel._path.delete()
+
+        # Seed an obsolete hop naming the parent. Its recovery must supply the second channel's missing path.
+        superseded = CablePath(
+            path=[
+                [object_to_path_node(channels[0]), object_to_path_node(parent)],
+                [object_to_path_node(cable)],
+                [object_to_path_node(peer) for peer in far],
+            ],
+            is_complete=True,
+            is_active=True,
+        )
+        superseded.save()
+        parent.refresh_from_db()
+        self.assertPathIsSet(parent, superseded)
+
+        with mock.patch.object(CablePath, 'from_origin', wraps=CablePath.from_origin) as trace:
+            create_cablepaths([channels[0]])
+
+        # Expanding the parent must not retrace the explicitly requested first channel a second time.
+        self.assertEqual(trace.call_args_list, [mock.call([channels[0]]), mock.call([channels[1]])])
+        self.assertFalse(CablePath.objects.filter(pk=superseded.pk).exists())
+        self.assertEqual(CablePath.objects.count(), 4)
+        parent.refresh_from_db()
+        self.assertPathIsNotSet(parent)
+        for channel, peer in zip(channels, far):
+            self.assertCurrentPathExists((channel, cable, peer), is_complete=True, is_active=True)
+            self.assertCurrentPathExists(
+                (peer, cable, channel), pk=peer_paths[peer.pk], is_complete=True, is_active=True
+            )
+
 
 class ChannelizedInterfaceTestCase(TestCase):
     """

netbox/dcim/tests/test_management_commands.py

@@ -49,9 +49,18 @@ class TracePathsTestCase(TestCase):
         self.assertIn('Finished.', out.getvalue())
 
     def test_retraces_missing_cabled_endpoint_path(self):
-        endpoint = object()
+        endpoint = SimpleNamespace(pk=1, _trace_cable_id=None, _trace_cable_end=None)
 
         class FakeQuerySet(list):
+            def annotate(self, **kwargs):
+                return self
+
+            def order_by(self, *fields):
+                return self
+
+            def iterator(self, chunk_size):
+                return iter(self)
+
             def filter(self, *args, **kwargs):
                 return self
 
@@ -81,9 +90,20 @@ class TracePathsTestCase(TestCase):
         self.assertIn('Finished.', out.getvalue())
 
     def test_progress_bar_drawn_every_100_endpoints(self):
-        endpoints = [object() for _ in range(100)]
+        endpoints = [
+            SimpleNamespace(pk=i, _trace_cable_id=None, _trace_cable_end=None) for i in range(100)
+        ]
 
         class FakeQuerySet(list):
+            def annotate(self, **kwargs):
+                return self
+
+            def order_by(self, *fields):
+                return self
+
+            def iterator(self, chunk_size):
+                return iter(self)
+
             def filter(self, *args, **kwargs):
                 return self
 

netbox/dcim/tests/utils.py

@@ -64,6 +64,21 @@ class BaseCablePathTestCase(TestCase):
         cablepath = self._get_cablepath(nodes, **kwargs)
         self.assertIsNone(cablepath, msg='Unexpected CablePath found')
 
+    def assertCurrentPathExists(self, nodes, **kwargs):
+        """
+        Assert that the first node references a CablePath with the given route via _path, and return it.
+
+        :param nodes: Iterable of steps, the first being the originating path endpoint object
+        """
+        origin = type(nodes[0]).objects.get(pk=nodes[0].pk)
+        self.assertIsNotNone(origin._path_id, msg=f'No path set on originating endpoint {origin}')
+        # Matched on the route alone, so a flag mismatch does not report itself as a wrong route
+        cablepath = self._get_cablepath(nodes, pk=origin._path_id)
+        self.assertIsNotNone(cablepath, msg=f'Path #{origin._path_id} on {origin} does not match the expected route')
+        for attr, expected in kwargs.items():
+            self.assertEqual(getattr(cablepath, attr), expected, msg=f'Path #{cablepath.pk} on {origin}: {attr}')
+        return cablepath
+
     def assertPathIsSet(self, origin, cablepath, msg=None):
         """
         Assert that a specific CablePath instance is set as the path on the origin.

netbox/dcim/utils.py

@@ -1,8 +1,9 @@
-from collections import defaultdict
+from collections import defaultdict, deque
 
 from django.apps import apps
 from django.contrib.contenttypes.models import ContentType
 from django.db import router, transaction
+from django.db.models import Q
 from django.utils.translation import gettext as _
 
 from dcim.constants import MODULE_TOKEN
@@ -125,13 +126,98 @@ def path_node_to_object(repr):
     return ct.model_class().objects.filter(pk=object_id).first()
 
 
+def _get_cablepath_origin_objects(nodes):
+    """
+    Fetch fresh origins in bulk by content type, indexed by their encoded path nodes.
+    """
+    ids_by_type = defaultdict(set)
+    for node in nodes:
+        type_id, object_id = decompile_path_node(node)
+        ids_by_type[type_id].add(object_id)
+
+    objects = {}
+    for type_id, object_ids in ids_by_type.items():
+        model = ContentType.objects.get_for_id(type_id).model_class()
+        if model is not None:
+            # Recovery also reads link; loading its direct relations avoids another query for every origin.
+            relations = [
+                field.name for field in model._meta.fields
+                if field.many_to_one and field.name in ('cable', 'wireless_link')
+            ]
+            queryset = model.objects.select_related(*relations) if relations else model.objects.all()
+            for object_id, obj in queryset.in_bulk(object_ids).items():
+                objects[compile_path_node(type_id, object_id)] = obj
+    return objects
+
+
 def create_cablepaths(objects):
     """
-    Create CablePaths for all paths originating from the specified set of nodes.
+    Trace and replace paths for the supplied origins, recovering co-origins whose current paths are removed.
+
+    The supplied objects may span multiple connectors on one link. Grouping and recovery share one worklist.
 
     :param objects: Iterable of cabled objects (e.g. Interfaces)
     """
-    from dcim.models import CablePath, Interface
+    from dcim.models import CablePath, PathEndpoint
+
+    origin_groups = _group_cablepath_origins(objects)
+    pending = deque(origin_groups)
+    if not pending:
+        return
+    processed = set()
+
+    # Keep a savepoint so callers can catch tracing failures inside an outer transaction.
+    # This provides rollback, but does not serialize concurrent rebuilds.
+    with transaction.atomic(using=router.db_for_write(CablePath)):
+        while pending:
+            # Do not let recovery replace origins already rebuilt by this call.
+            origins = [
+                obj for obj in pending.popleft()
+                if object_to_path_node(obj) not in processed
+            ]
+            if not origins:
+                continue
+
+            # Trace first, so an unsupported topology raises before anything is deleted
+            path = CablePath.from_origin(origins)
+            nodes = {object_to_path_node(obj) for obj in origins}
+            processed.update(nodes)
+
+            # `overlap` takes the encoded nodes directly, and matches nothing for an empty set
+            for old_path in CablePath.objects.filter(_nodes__overlap=list(nodes)):
+                # `_nodes` matches a node anywhere in a path, including as another path's destination
+                if not old_path.path or not nodes.intersection(old_path.path[0]):
+                    continue
+
+                # Recover only origins whose current path is being deleted, keeping different links separate
+                remaining = [node for node in old_path.path[0] if node not in processed]
+                # Refresh immediately before this deletion; earlier iterations can change _path references.
+                origin_objects = _get_cablepath_origin_objects(remaining)
+                by_link = defaultdict(list)
+                for node in remaining:
+                    origin = origin_objects.get(node)
+                    if origin is None:
+                        continue
+                    # Absence of the back-reference is not evidence that a pointer was cleared
+                    if isinstance(origin, PathEndpoint) and origin._path_id != old_path.pk:
+                        continue
+                    if link := origin.link:
+                        by_link[link].append(origin)
+                for linked_origins in by_link.values():
+                    recovered_groups = _group_cablepath_origins(linked_origins)
+                    pending.extend(recovered_groups)
+
+                old_path.delete()
+
+            if path:
+                path.save()
+
+
+def _group_cablepath_origins(objects):
+    """
+    Expand channelized interfaces and group origins on one link by connector, keeping channels separate.
+    """
+    from dcim.models import Interface
 
     # Expand any channelized interface into its channel subinterfaces. A channelized parent originates no path of its
     # own; instead, each channel subinterface traces independently from the single connector position it occupies.
@@ -144,35 +230,137 @@ def create_cablepaths(objects):
         else:
             expanded.append(obj)
 
-    # Arrange objects by cable connector. All objects with a null connector are grouped together. Channel
-    # subinterfaces must each originate their own path, as sharing a connector would otherwise collapse a group of
-    # siblings into a single malformed path.
-    origins = defaultdict(list)
+    origin_groups = []
+    connectors = defaultdict(list)
     for obj in expanded:
         if isinstance(obj, Interface) and obj.channel_id:
-            if cp := CablePath.from_origin([obj]):
-                cp.save()
+            origin_groups.append([obj])
         else:
-            origins[obj.cable_connector].append(obj)
+            connectors[obj.cable_connector].append(obj)
+    origin_groups.extend(connectors.values())
+
+    return origin_groups
 
-    for connector, objects in origins.items():
-        if cp := CablePath.from_origin(objects):
-            cp.save()
+
+def get_cable_end_terminations(cable_ends):
+    """
+    Return the current termination objects for each (cable ID, end) key, preserving connector order.
+
+    Use the CableTermination's connector for grouping. This does not repair persisted endpoint fields;
+    inconsistent cached links or positions can still prevent tracing.
+    """
+    from dcim.models import CableTermination
+
+    groups = {key: [] for key in cable_ends}
+    if not groups:
+        return groups
+
+    cables_by_end = defaultdict(list)
+    for cable_id, cable_end in groups:
+        cables_by_end[cable_end].append(cable_id)
+    query = Q()
+    # Cable ends are A or B, so this needs at most two clauses regardless of the number of cables.
+    for cable_end, cable_ids in cables_by_end.items():
+        query |= Q(cable_id__in=cable_ids, cable_end=cable_end)
+    terminations = CableTermination.objects.filter(query).order_by(
+        'cable_id', 'cable_end', 'connector', 'pk'
+    ).prefetch_related('termination__cable')
+    for termination in terminations:
+        # A stale GenericForeignKey can still reference an object which no longer exists.
+        if (obj := termination.termination) is not None:
+            # Partition this fetched instance by its authoritative connector, without saving the endpoint.
+            obj.cable_connector = termination.connector
+            groups[(termination.cable_id, termination.cable_end)].append(obj)
+    return groups
+
+
+def get_cablepath_origin_groups(objects):
+    """
+    Resolve originating objects to current cable ends in bulk. Channels and uncabled origins remain singletons.
+    """
+    from dcim.models import CableTermination, Interface
+
+    objects = {object_to_path_node(obj): obj for obj in objects}
+    if not objects:
+        return {}
+
+    ids_by_type = defaultdict(list)
+    for node, obj in objects.items():
+        if not (isinstance(obj, Interface) and obj.channel_id):
+            type_id, object_id = decompile_path_node(node)
+            ids_by_type[type_id].append(object_id)
+
+    # Resolve membership from CableTermination, never from the endpoint's cached cable/end fields.
+    keys = {}
+    if ids_by_type:
+        query = Q()
+        for type_id, object_ids in ids_by_type.items():
+            query |= Q(termination_type_id=type_id, termination_id__in=object_ids)
+        for type_id, object_id, cable_id, cable_end in CableTermination.objects.filter(query).values_list(
+            'termination_type_id', 'termination_id', 'cable_id', 'cable_end'
+        ):
+            keys[compile_path_node(type_id, object_id)] = (cable_id, cable_end)
+
+    cable_ends = get_cable_end_terminations(keys.values())
+    groups = {}
+    for node, obj in objects.items():
+        if node in keys:
+            groups[keys[node]] = cable_ends[keys[node]]
+        else:
+            groups[node] = [obj]
+    return groups
 
 
 def rebuild_paths(terminations):
     """
-    Rebuild all CablePaths which traverse the specified nodes.
+    Rebuild paths traversing the given nodes from their origins' current cable-end membership.
     """
     from dcim.models import CablePath
 
-    for obj in terminations:
-        cable_paths = CablePath.objects.filter(_nodes__contains=obj)
+    nodes = [object_to_path_node(obj) for obj in terminations]
+    if not nodes:
+        return
 
-        with transaction.atomic(using=router.db_for_write(CablePath)):
-            for cp in cable_paths:
-                cp.delete()
-                create_cablepaths(cp.origins)
+    # Snapshot the entire operation before replacement can retire another candidate. The transaction includes
+    # candidate deletion as well as all replacements; a failure must restore both. It does not serialize rebuilds.
+    with transaction.atomic(using=router.db_for_write(CablePath)):
+        cable_paths = list(CablePath.objects.filter(_nodes__overlap=nodes))
+        origin_nodes = dict.fromkeys(
+            node for cable_path in cable_paths for node in (cable_path.path[0] if cable_path.path else ())
+        )
+        origin_objects = _get_cablepath_origin_objects(origin_nodes)
+        origins = [origin_objects[node] for node in origin_nodes if node in origin_objects]
+        origin_groups = get_cablepath_origin_groups(origins)
+
+        # A historical hop can name a parent which now originates paths through its channels.
+        affected_nodes = set()
+        for origin in origins:
+            expanded_groups = _group_cablepath_origins([origin])
+            for group in expanded_groups:
+                affected_nodes.update(object_to_path_node(obj) for obj in group)
+
+        # Current membership determines each complete group, but only candidate origins determine its scope.
+        # Do not retrace unrelated connectors on the same cable end.
+        rebuild_inputs = []
+        scheduled = set()
+        for current_origins in origin_groups.values():
+            selected_origins = []
+            current_groups = _group_cablepath_origins(current_origins)
+            for group in current_groups:
+                group_nodes = {object_to_path_node(obj) for obj in group}
+                if group_nodes & affected_nodes and not group_nodes.issubset(scheduled):
+                    selected_origins.extend(group)
+                    # A parent's channels can also appear as individual candidate origins.
+                    scheduled.update(group_nodes)
+            if selected_origins:
+                # Keep one call per end so all requested groups precede recovery in the same worklist.
+                rebuild_inputs.append(selected_origins)
+
+        # Use the model method to clear current endpoint references, including those on disconnected origins.
+        for cable_path in cable_paths:
+            cable_path.delete()
+        for origins in rebuild_inputs:
+            create_cablepaths(origins)
 
 
 def rebuild_cable_paths(cable):