refactor(family): adopt template claim resolution

docs/adr/0013-centralize-template-claim-ambiguity.md

@@ -26,7 +26,10 @@ Message construction belongs to the primitive so callers cannot drift in wording
 The primitive does not log. Callers emit its messages through their own logger,
 which preserves the engine warning source and keeps the decision free of side effects.
 
-Increment 1 adopts the primitive only in the engine.
-Installed-family adoption belongs to increment 2.
+The engine and installed-family discovery both use the primitive.
+Installed-family discovery passes the complete historical base relation and emits
+collision messages through its own logger with the label kind `family base`.
+Current resolved bases remain unconditional and precede accepted historical bases.
+Historical base extraction and duplicate handling remain unchanged.
 The `_singly_claimed` rule counts member primary keys across complete family
 candidates and remains separate.

netbox_interface_name_rules/family/conversion.py

@@ -126,7 +126,7 @@ def plan_module_conversions(
     channelization_supported = supports_channelization()
     plans = []
     claimed = set()
-    for base_name, source_base in flat_family_bases(rule, variables, interfaces, catalog):
+    for base_name, source_base in flat_family_bases(module, rule, variables, interfaces, catalog):
         names = family_names_for(rule, variables, base_name, source_base)
         if names is None:
             continue

netbox_interface_name_rules/family/installed.py

@@ -9,6 +9,7 @@
 
 from ..choices import BreakoutModeChoices
 from ..naming import evaluate_name_template
+from .claims import TemplateClaim, resolve_template_claims
 from .domain import (
     FamilyStatus,
     FamilyTopology,
@@ -73,44 +74,41 @@ def _historical_bases(rule, variables, template, interfaces):  # pragma: no cove
     return tuple(sorted(bases))
 
 
-def _ambiguous_bases(historical_by_template):
-    """Return every historical base that more than one template could claim."""
-    ambiguous = {base_name for bases in historical_by_template.values() if len(bases) > 1 for base_name in bases}
-    single_claims: dict[str, int] = {}
-    for bases in historical_by_template.values():
-        if len(bases) == 1:  # pragma: no cover - historical matchers require VC token support
-            single_claims[bases[0]] = single_claims.get(bases[0], 0) + 1
-    ambiguous.update(base_name for base_name, count in single_claims.items() if count > 1)
-    return ambiguous
+def _source_bases(template, historical_bases):
+    """Return the template's own base and its accepted historical bases."""
+    return (template.resolved, *historical_bases)
 
 
-def _source_bases(template, historical_bases, ambiguous_bases):
-    """Return the template's own base and every historical base it alone claims."""
-    unambiguous = tuple(
-        base_name for base_name in historical_bases if len(historical_bases) == 1 and base_name not in ambiguous_bases
-    )
-    return (template.resolved, *unambiguous)
-
-
-def flat_family_bases(rule, variables, interfaces, catalog):
+def flat_family_bases(module, rule, variables, interfaces, catalog):
     """Return ``(template base, source base)`` for every base a flat family could be named from.
 
     The template base is the name the rule resolves for this module now; the source base is the one
-    an installed family still spells, which differs after a virtual-chassis renumber.  A historical
-    base more than one template could claim is dropped: the rows it names are not certainly one
-    family's, and neither renaming nor converting them is this plugin's guess to make.
+    an installed family still spells, which differs after a virtual-chassis renumber.
+    A historical base that more than one template could claim is dropped.
+    Every historical base of a template that claims multiple bases is also dropped.
+    These claims do not identify one family with certainty, so the plugin does not rename or convert them.
     """
     if rule.channel_count <= 0:
         return ()
     templates = catalog.get()
     historical_by_template = {
         template.pk: _historical_bases(rule, variables, template, interfaces) for template in templates
     }
-    ambiguous_bases = _ambiguous_bases(historical_by_template)
+    accepted, messages = resolve_template_claims(
+        tuple(
+            TemplateClaim(template.pk, template.template_name, historical_by_template[template.pk])
+            for template in templates
+        ),
+        module=module,
+        label_kind="family base",
+    )
+    for message in messages:
+        logger.warning(message)
+    accepted_by_template = {pk: (base,) for pk, base in accepted}
     return tuple(
         (template.resolved, source_base)
         for template in templates
-        for source_base in _source_bases(template, historical_by_template[template.pk], ambiguous_bases)
+        for source_base in _source_bases(template, accepted_by_template.get(template.pk, ()))
     )
 
 
@@ -135,7 +133,7 @@ def _singly_claimed(candidates):
     return [candidate for candidate in candidates if all(claims[member.pk] == 1 for member in candidate[2])]
 
 
-def flat_family_candidates(rule, variables, interfaces, catalog):
+def flat_family_candidates(module, rule, variables, interfaces, catalog):
     """Return complete, unambiguous flat-family candidates on this module.
 
     A flat family carries the names the rule's channel range spells, and a flat rule and the
@@ -146,7 +144,7 @@ def flat_family_candidates(rule, variables, interfaces, catalog):
     if not by_name:
         return []
     candidates = []
-    for base_name, source_base in flat_family_bases(rule, variables, interfaces, catalog):
+    for base_name, source_base in flat_family_bases(module, rule, variables, interfaces, catalog):
         names = family_names_for(rule, variables, base_name, source_base)
         if names is None:
             continue
@@ -159,11 +157,11 @@ def flat_family_candidates(rule, variables, interfaces, catalog):
     return _singly_claimed(candidates)
 
 
-def _flat_candidates(rule, variables, interfaces, catalog):
+def _flat_candidates(module, rule, variables, interfaces, catalog):
     """Return the flat families a flat-mode rule owns on this module."""
     if rule.breakout_mode != BreakoutModeChoices.FLAT:
         return []
-    return flat_family_candidates(rule, variables, interfaces, catalog)
+    return flat_family_candidates(module, rule, variables, interfaces, catalog)
 
 
 def _flat_plan(module, target_names, interfaces):
@@ -346,7 +344,7 @@ def plan_installed_families(module, rule, variables, interfaces=None) -> Install
     plans = _channelized_plans(module, rule, variables, interfaces, catalog)
     plans.extend(
         _flat_plan(module, target_names, members)
-        for _base_name, target_names, members in _flat_candidates(rule, variables, interfaces, catalog)
+        for _base_name, target_names, members in _flat_candidates(module, rule, variables, interfaces, catalog)
     )
     plans.sort(key=lambda plan: plan.member_pks[0])
     return InstalledFamilyPlanSet(module_id=module.pk, plans=tuple(plans))

netbox_interface_name_rules/tests/test_vc_drift.py

@@ -810,7 +810,46 @@ def test_a_family_matcher_that_captures_two_bases_is_not_offered(self):
         self._join(VirtualChassis.objects.create(name="vcconv-vc2"), 5, device=standalone)
         self._switch_to_channelized(rule)
 
-        self.assertEqual(find_convertible_families(rule).candidates, ())
+        with self.assertLogs("netbox_interface_name_rules.family.installed", level="WARNING") as logs:
+            self.assertEqual(find_convertible_families(rule).candidates, ())
+
+        warnings = " ".join(logs.output)
+        self.assertIn("Family base", warnings)
+        self.assertIn("xe-{vc_position:0}/0/{module}", warnings)
+        self.assertIn("xe-{vc_position:9}/0/{module}", warnings)
+        self.assertIn(str(module), warnings)
+
+    def test_an_unrelated_family_survives_overlapping_historical_claims(self):
+        """A multi-base claim rejects its shared base but leaves an unrelated family available."""
+        module_type = _token_module_type(
+            self.manufacturer,
+            "VcConv-OVERLAP",
+            "xe-{vc_position}/0/{module}",
+            "xe-1/{vc_position}/{module}",
+            "et-{vc_position}/0/{module}",
+        )
+        rule = self._flat_rule(module_type, "brk-{base}:{channel}")
+        module, _ = self._install_on(self.device, module_type, "3")
+        for channel in range(4):
+            rename_out_of_band(
+                Interface.objects.get(module=module, name=f"brk-xe-1/1/3:{channel}"),
+                f"brk-xe-2/0/3:{channel}",
+            )
+        self._renumber(5)
+        self._switch_to_channelized(rule)
+
+        with self.assertLogs("netbox_interface_name_rules.family.installed", level="WARNING") as logs:
+            candidates = find_convertible_families(rule).candidates
+
+        self.assertEqual(len(candidates), 1)
+        self.assertTrue(candidates[0].convertible, candidates[0].reason)
+        self.assertEqual(list(candidates[0].current_names), [f"brk-et-1/0/3:{channel}" for channel in range(4)])
+        self.assertEqual(candidates[0].new_names[0], "et-0/0/3")
+        self.assertIn(
+            f"Family base 'xe-1/0/3' on {module} could be the drifted name of any of the templates "
+            "['xe-1/{vc_position}/{module}', 'xe-{vc_position}/0/{module}']",
+            " ".join(logs.output),
+        )
 
     def test_a_rule_without_a_base_is_identified_after_a_renumber(self):
         """Drift-immune by construction — asserted, not assumed, so the fix cannot regress it."""