fix(bridge): align static TG options loop with legacy bridge_master

Refresh TS1/TS2 static talkgroups on the 26s timer and options_config path
so user-activated bridges and OPTIONS strings stay consistent after reload.

src/adn_server/application/bridge/bridge_table.py

@@ -361,6 +361,16 @@ class BridgeTableMixin:
                 _tmout = 35791394
             new_ts1 = str(_options.get("TS1_STATIC") or "").strip()
             new_ts2 = str(_options.get("TS2_STATIC") or "").strip()
+            # Legacy options_config: malformed TS1/TS2 aborts the whole OPTIONS apply (continue).
+            if new_ts1 and re.search(r"[^\d,]", new_ts1):
+                return
+            if new_ts2 and re.search(r"[^\d,]", new_ts2):
+                return
+            merged = self._merged_static_tg_lists_for_master(system_name)
+            if merged is not None:
+                _tmout, ts1_nums, ts2_nums = merged
+                new_ts1 = ",".join(str(x) for x in ts1_nums)
+                new_ts2 = ",".join(str(x) for x in ts2_nums)
             if re.search(r"[^\d,]", new_ts1) or re.search(r"[^\d,]", new_ts2):
                 return
             # Peers may resend identical RPTO on every voice burst. YAML may already match parsed TS — do not
@@ -509,24 +519,77 @@ class BridgeTableMixin:
                 _tmout = 35791394
             ts1_list: list[int] = []
             ts2_list: list[int] = []
-            for tg_s in str(_options.get("TS1_STATIC") or "").split(","):
-                try:
-                    tg1 = int(tg_s.strip())
-                    if tg1 not in (0, 1, 2, 3, 4, 5, 9, 9990, 9991, 9992, 9993, 9994, 9995, 9996, 9997, 9998, 9999):
-                        ts1_list.append(tg1)
-                except ValueError:
-                    pass
-            for tg_s in str(_options.get("TS2_STATIC") or "").split(","):
-                try:
-                    tg2 = int(tg_s.strip())
-                    if 0 < tg2 < 16777215 and tg2 not in (9990, 9991, 9992, 9993, 9994, 9995, 9996, 9997, 9998, 9999):
-                        ts2_list.append(tg2)
-                except ValueError:
-                    pass
+            ts1_raw = str(_options.get("TS1_STATIC") or "").strip()
+            ts2_raw = str(_options.get("TS2_STATIC") or "").strip()
+            if ts1_raw and not re.search(r"[^\d,]", ts1_raw):
+                for tg_s in ts1_raw.split(","):
+                    try:
+                        tg1 = int(tg_s.strip())
+                        if tg1 not in (0, 1, 2, 3, 4, 5, 9, 9990, 9991, 9992, 9993, 9994, 9995, 9996, 9997, 9998, 9999):
+                            ts1_list.append(tg1)
+                    except ValueError:
+                        pass
+            if ts2_raw and not re.search(r"[^\d,]", ts2_raw):
+                for tg_s in ts2_raw.split(","):
+                    try:
+                        tg2 = int(tg_s.strip())
+                        if 0 < tg2 < 16777215 and tg2 not in (9990, 9991, 9992, 9993, 9994, 9995, 9996, 9997, 9998, 9999):
+                            ts2_list.append(tg2)
+                    except ValueError:
+                        pass
             return (_tmout, ts1_list, ts2_list)
         except Exception:
             return None
 
+    def _merged_static_tg_lists_for_master(
+        self, system_name: str
+    ) -> tuple[float, list[int], list[int]] | None:
+        """Union static TG ids from runtime YAML and every connected peer RPTO (inject proxy)."""
+        sys_cfg = self._config.get("SYSTEMS", {}).get(system_name, {})
+        if sys_cfg.get("MODE") != "MASTER":
+            return None
+        tmout = float(sys_cfg.get("DEFAULT_UA_TIMER", 10))
+        if tmout <= 0:
+            tmout = 35791394.0
+        ts1_set: set[int] = set()
+        ts2_set: set[int] = set()
+        runtime = self._static_tg_lists_from_runtime_cfg(sys_cfg)
+        if runtime is not None:
+            tmout, ts1_list, ts2_list = runtime
+            ts1_set.update(ts1_list)
+            ts2_set.update(ts2_list)
+        protocols = self._get_protocols() if self._get_protocols else {}
+        proto = protocols.get(system_name)
+        peers = getattr(proto, "_peers", {}) if proto is not None else {}
+        if isinstance(peers, dict):
+            for peer in peers.values():
+                if not isinstance(peer, dict) or peer.get("CONNECTION") != "YES":
+                    continue
+                opt = peer.get("OPTIONS")
+                if opt is None:
+                    continue
+                if isinstance(opt, bytes):
+                    opt_str = opt.decode("utf8", errors="replace")
+                else:
+                    opt_str = str(opt)
+                parsed = self._parse_options_static_tgs(opt_str, sys_cfg)
+                if parsed is None:
+                    continue
+                peer_tmout, ts1_list, ts2_list = parsed
+                if peer_tmout > 0:
+                    tmout = max(tmout, peer_tmout)
+                ts1_set.update(ts1_list)
+                ts2_set.update(ts2_list)
+        if not ts1_set and not ts2_set and "OPTIONS" in sys_cfg:
+            parsed = self._parse_options_static_tgs(sys_cfg["OPTIONS"], sys_cfg)
+            if parsed is not None:
+                tmout, ts1_list, ts2_list = parsed
+                ts1_set.update(ts1_list)
+                ts2_set.update(ts2_list)
+        if not ts1_set and not ts2_set:
+            return None
+        return (tmout, sorted(ts1_set), sorted(ts2_set))
+
     def apply_static_tg_to_bridge(self, tg_int: int) -> None:
         """When a bridge was just created from OBP, mark MASTER systems that have this TG in static TS1/TS2 (runtime lists or OPTIONS) ACTIVE so the first OBP traffic reaches them."""
         systems_cfg = self._config.get("SYSTEMS", {})
@@ -535,10 +598,7 @@ class BridgeTableMixin:
                 continue
             if not systems_cfg.get(_system, {}).get("ENABLED", True):
                 continue
-            sys_cfg = systems_cfg[_system]
-            parsed = self._static_tg_lists_from_runtime_cfg(sys_cfg)
-            if parsed is None and "OPTIONS" in sys_cfg:
-                parsed = self._parse_options_static_tgs(sys_cfg["OPTIONS"], sys_cfg)
+            parsed = self._merged_static_tg_lists_for_master(_system)
             if not parsed:
                 continue
             _tmout, ts1_list, ts2_list = parsed
@@ -707,6 +767,11 @@ class BridgeTableMixin:
                 cur_ts2 = systems_cfg[_system].get("TS2_STATIC") or ""
                 new_ts1 = _options.get("TS1_STATIC") or ""
                 new_ts2 = _options.get("TS2_STATIC") or ""
+                merged = self._merged_static_tg_lists_for_master(_system)
+                if merged is not None:
+                    _tmout, ts1_nums, ts2_nums = merged
+                    new_ts1 = ",".join(str(x) for x in ts1_nums) if ts1_nums else ""
+                    new_ts2 = ",".join(str(x) for x in ts2_nums) if ts2_nums else ""
                 if str(new_ts1) != str(cur_ts1) or ua_timer_changed:
                     logger.debug("(OPTIONS) %s TS1 static TGs changed, updating", _system)
                     if cur_ts1:

src/adn_server/application/bridge/timers.py

@@ -99,7 +99,9 @@ class BridgeTimerMixin:
                             bridge_used = True
                             _debug_msgs.append('(ROUTER) Conference Bridge ACTIVE (no change): System: %s Bridge: %s, TS: %s, TGID: %s' % (system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b""))))
                     else:
-                        if not is_obp or (is_obp and to_type == "STAT"):
+                        # OBP legs from make_single_bridge use TO_TYPE NONE (legacy parity); keep the
+                        # bridge table while the OBP source row is ACTIVE (do not trim as "unused").
+                        if not is_obp or (is_obp and (to_type == "STAT" or active)):
                             bridge_used = True
                         _debug_msgs.append('(ROUTER) Conference Bridge NO ACTION: System: %s, Bridge: %s, TS: %s, TGID: %s' % (system_name, bridge_key, sys_entry.get("TS"), int_id(sys_entry.get("TGID", b""))))
 

tests/bridge/test_static_tg_options.py

@@ -2,9 +2,9 @@
 
 from __future__ import annotations
 
-from tests.harness.deterministic import DeterministicScenario, active_bridge
+from tests.harness.deterministic import DeterministicScenario, active_bridge, add_openbridge_system
 
-from adn_server.domain import bytes_3
+from adn_server.domain import bytes_3, bytes_4
 
 
 def _master_with_options(options: str) -> DeterministicScenario:
@@ -85,6 +85,55 @@ def test_bridge_reset_restores_prohibited_static_legs() -> None:
     assert leg["TO_TYPE"] == "NONE"
 
 
+def test_options_merge_static_tgs_from_all_connected_peers() -> None:
+    """Last peer RPTO must not drop other hotspots' static TG bridges (inject proxy)."""
+    scenario = DeterministicScenario()
+    add_openbridge_system(scenario.config)
+    scenario._wire_protocols_from_config()
+    scenario.bridge._get_protocols = lambda: scenario.protocols  # noqa: SLF001
+    master = scenario.config["SYSTEMS"]["MASTER-A"]
+    master["OPTIONS"] = "TS2=8730444;TIMER=300"
+    proto = scenario.protocols["MASTER-A"]
+    proto._peers = {
+        bytes_4(730039101): {
+            "CONNECTION": "YES",
+            "OPTIONS": b"TS2=730444;TIMER=300",
+        },
+        bytes_4(730266501): {
+            "CONNECTION": "YES",
+            "OPTIONS": b"TS2=8730444;TIMER=300",
+        },
+    }
+
+    scenario.bridge.options_config_for_system("MASTER-A")
+
+    sys_cfg = scenario.config["SYSTEMS"]["MASTER-A"]
+    assert "730444" in sys_cfg["TS2_STATIC"]
+    assert "8730444" in sys_cfg["TS2_STATIC"]
+    bridges = scenario.bridge.get_bridges()
+    assert "730444" in bridges
+    assert "8730444" in bridges
+    leg_730444 = next(e for e in bridges["730444"] if e["SYSTEM"] == "MASTER-A" and e["TS"] == 2)
+    assert leg_730444["ACTIVE"] is True
+
+
+def test_rule_timer_keeps_bridge_with_active_obp_none_leg() -> None:
+    """Static TG bridges must survive rule_timer when OBP uses TO_TYPE NONE (make_single_bridge)."""
+    scenario = DeterministicScenario()
+    add_openbridge_system(scenario.config)
+    scenario._wire_protocols_from_config()
+    scenario.bridge.make_single_bridge(bytes_3(730444), "MASTER-A", 2, 300.0)
+    bridges = scenario.bridge.get_bridges()
+    for entry in bridges["730444"]:
+        if entry["SYSTEM"] == "OBP-CL":
+            entry["ACTIVE"] = True
+            entry["TO_TYPE"] = "NONE"
+
+    scenario.bridge.rule_timer_loop()
+
+    assert "730444" in scenario.bridge.get_bridges()
+
+
 def test_options_applies_valid_static_tg_once() -> None:
     scenario = _master_with_options("TS2=52090;TIMER=10")
     scenario.bridge.options_config_for_system("MASTER-A")