Add ruff linting, fix UDP protected access, fix behavioral bugs

- Add ruff with ARG, PLR6301, SLF001 rules; remove unused args,
  make methods static where applicable.
- UDP: add public properties (closed, uid, interfaces, tx_socks),
  public async_sendto and remove_subject_listener to eliminate
  all cross-class protected member access.
- Fix lage clamping to [-1, 35] matching reference LAGE_MIN/MAX.
- Fix gossip suppression: track periodic vs urgent gossip task,
  only suppress periodic (matching reference behavior).
- Validate gossip hash vs name on implicit topic creation.
- Remove unused function arguments across core modules.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: pavel-kirienko

CLAUDE.md

@@ -62,9 +62,10 @@ Tests are x10+ the size of source code and must provide full coverage of the cor
 Transport test coverage is more opportunistic.
 
 The library must ONLY be tested with Python versions starting from the minimum specified in `pyproject.toml`
-up to the current latest stable Python. TESTING ON UNSUPPORTED VERSIONS IS NOT ALLOWED.
+up to the current latest stable Python.
+TESTING ON UNSUPPORTED VERSIONS IS NOT ALLOWED.
 
-Work will not be accepted unless `nox` (without arguments) runs successfully.
+ACCEPTANCE CRITERIA: Work will not be accepted unless `nox` (without arguments) runs successfully.
 
 ### Logging
 

pyproject.toml

@@ -88,19 +88,24 @@ asyncio_mode = "auto"
 [tool.ruff]
 line-length = 120
 target-version = "py312"
+preview = true
 
 [tool.ruff.lint]
 # Only the checks the project needs right now; others disabled to avoid false positives.
 select = [
     "F401",    # unused imports
     "F811",    # redefinition of unused name
     "F841",    # local variable assigned but never used
+    "ARG001",  # unused function argument
+    "ARG002",  # unused method argument
+    "ARG005",  # unused lambda argument
+    "PLR6301", # method could be a function or static method
     "SLF001",  # private member accessed from outside its class/module
 ]
 
 [tool.ruff.lint.per-file-ignores]
 # Tests may access internals for white-box testing; allow SLF001 there.
-"tests/*" = ["SLF001"]
+"tests/*" = ["SLF001", "ARG", "PLR6301"]
 
 [tool.black]
 line-length = 120

src/pycyphal/_node.py

@@ -289,6 +289,7 @@ def __init__(self, node: NodeImpl, name: str, evictions: int, now: float) -> Non
         self.publish_futures: dict[int, PublishTracker] = {}
         self.request_futures: dict[int, ResponseStreamImpl] = {}  # tag -> ResponseStreamImpl
         self.gossip_task: asyncio.Task[None] | None = None
+        self.gossip_task_is_periodic = False
         self.gossip_counter = 0
 
     # -- Topic ABC --
@@ -313,10 +314,10 @@ def lage(self, now: float) -> int:
 
     def merge_lage(self, now: float, remote_lage: int) -> None:
         """Shift ts_origin backward if the remote claims an older origin."""
+        clamped = max(-1, min(35, remote_lage))  # LAGE_MIN=-1, LAGE_MAX=35
         my_lage = self.lage(now)
-        if remote_lage > my_lage:
-            # Shift origin backward to match the remote's claim.
-            self.ts_origin = now - (1 << remote_lage)
+        if clamped > my_lage:
+            self.ts_origin = now - (1 << clamped)
 
     def animate(self, ts: float) -> None:
         self.ts_animated = ts
@@ -516,7 +517,7 @@ def topic_allocate(self, topic: TopicImpl, new_evictions: int, now: float) -> No
                 t.release_transport_handles()
                 t.evictions = ev
                 t.ensure_listener()
-                self.schedule_gossip_urgent(t, now)
+                self.schedule_gossip_urgent(t)
                 continue
 
             modulus = self.transport.subject_id_modulus
@@ -532,7 +533,7 @@ def topic_allocate(self, topic: TopicImpl, new_evictions: int, now: float) -> No
                 t.evictions = ev
                 self.topics_by_subject_id[new_sid] = t
                 t.ensure_listener()
-                self.schedule_gossip_urgent(t, now)
+                self.schedule_gossip_urgent(t)
             elif left_wins(t.lage(now), t.hash, collider.lage(now), collider.hash):
                 # Our topic wins: take the slot, evict the collider.
                 t.release_transport_handles()
@@ -544,15 +545,16 @@ def topic_allocate(self, topic: TopicImpl, new_evictions: int, now: float) -> No
                 del self.topics_by_subject_id[new_sid]
                 self.topics_by_subject_id[new_sid] = t
                 t.ensure_listener()
-                self.schedule_gossip_urgent(t, now)
+                self.schedule_gossip_urgent(t)
                 # Schedule collider for reallocation.
                 collider.release_transport_handles()
                 work.append((collider, collider.evictions + 1))
             else:
                 # Our topic loses: increment evictions and retry.
                 work.append((t, ev + 1))
 
-    def couple_topic_root(self, topic: TopicImpl, root: SubscriberRoot) -> None:
+    @staticmethod
+    def couple_topic_root(topic: TopicImpl, root: SubscriberRoot) -> None:
         """Create a coupling between a topic and a subscriber root if not already coupled."""
         for c in topic.couplings:
             if c.root is root:
@@ -587,25 +589,29 @@ def schedule_gossip(self, topic: TopicImpl) -> None:
         """Start periodic gossip for an explicit topic."""
         if topic.gossip_task is not None:
             return  # already scheduled
+        topic.gossip_task_is_periodic = True
         topic.gossip_task = self.loop.create_task(self.gossip_loop(topic))
 
-    def schedule_gossip_urgent(self, topic: TopicImpl, now: float) -> None:
+    def schedule_gossip_urgent(self, topic: TopicImpl) -> None:
         """Cancel current gossip and reschedule with short delay."""
         if topic.gossip_task is not None:
             topic.gossip_task.cancel()
+        topic.gossip_task_is_periodic = False
         topic.gossip_task = self.loop.create_task(self.gossip_urgent(topic))
 
     async def gossip_urgent(self, topic: TopicImpl) -> None:
         try:
             await asyncio.sleep(random.random() * GOSSIP_URGENT_DELAY_MAX)
             await self.send_gossip(topic, broadcast=True)
             # Resume periodic gossip.
+            topic.gossip_task_is_periodic = True
             topic.gossip_task = self.loop.create_task(self.gossip_loop(topic))
         except asyncio.CancelledError:
             pass
 
     async def gossip_loop(self, topic: TopicImpl) -> None:
         try:
+            topic.gossip_task_is_periodic = True
             while not self._closed and not topic.is_implicit:
                 dither = GOSSIP_PERIOD / GOSSIP_PERIOD_DITHER_RATIO
                 delay = GOSSIP_PERIOD + random.uniform(-dither, dither)
@@ -696,15 +702,15 @@ def dispatch_arrival(self, arrival: TransportArrival, *, subject_id: int | None,
         payload = msg[HEADER_SIZE:]
 
         if isinstance(hdr, (MsgBeHeader, MsgRelHeader)):
-            self.on_msg(arrival, hdr, payload, unicast)
+            self.on_msg(arrival, hdr, payload)
         elif isinstance(hdr, (MsgAckHeader, MsgNackHeader)):
             self.on_msg_ack(arrival, hdr)
         elif isinstance(hdr, (RspBeHeader, RspRelHeader)):
             self.on_rsp(arrival, hdr, payload)
         elif isinstance(hdr, (RspAckHeader, RspNackHeader)):
             self.on_rsp_ack(arrival, hdr)
         elif isinstance(hdr, GossipHeader):
-            self.on_gossip(arrival, hdr, payload, unicast, subject_id)
+            self.on_gossip(hdr, payload, unicast, subject_id)
         elif isinstance(hdr, ScoutHeader):
             self.on_scout(arrival, hdr, payload)
 
@@ -713,7 +719,6 @@ def on_msg(
         arrival: TransportArrival,
         hdr: MsgBeHeader | MsgRelHeader,
         payload: bytes,
-        unicast: bool,
     ) -> None:
         topic = self.topics_by_hash.get(hdr.topic_hash)
         if topic is None:
@@ -754,8 +759,8 @@ def on_msg(
             )
             self.deliver_to_subscribers(topic, arrival, breadcrumb, payload, hdr.tag)
 
+    @staticmethod
     def deliver_to_subscribers(
-        self,
         topic: TopicImpl,
         arrival: TransportArrival,
         breadcrumb: Breadcrumb,
@@ -873,7 +878,6 @@ async def do_send() -> None:
 
     def on_gossip(
         self,
-        arrival: TransportArrival,
         hdr: GossipHeader,
         payload: bytes,
         unicast: bool,
@@ -912,19 +916,26 @@ def on_gossip_known(
             win = my_lage > lage or (my_lage == lage and topic.evictions > evictions)
             topic.merge_lage(now, lage)
             if win:
-                self.schedule_gossip_urgent(topic, now)
+                self.schedule_gossip_urgent(topic)
             else:
                 self.topic_allocate(topic, evictions, now)
                 if topic.evictions == evictions:
                     self.schedule_gossip(topic)
         else:
             topic.merge_lage(now, lage)
-            # Suppress gossip if we can't contribute newer states.
+            # Suppress gossip if we can't contribute newer states to the consensus process.
+            # Only suppress if the current gossip task is periodic (not urgent) or the scope is broadcast.
+            # Unicast gossips are seen by a small subset of nodes so they do not suppress.
             is_broadcast = subject_id == self.broadcast_subject_id
             is_sharded = not unicast and not is_broadcast
-            suppress = (is_broadcast or is_sharded) and (topic.lage(now) == lage)
+            suppress = (
+                (is_broadcast or is_sharded)
+                and (topic.lage(now) == lage)
+                and (topic.gossip_task_is_periodic or is_broadcast)
+            )
             if suppress and topic.gossip_task is not None:
                 topic.gossip_task.cancel()
+                topic.gossip_task_is_periodic = True
                 topic.gossip_task = self.loop.create_task(self.gossip_loop(topic))
             topic.ensure_listener()
 
@@ -936,18 +947,23 @@ def on_gossip_unknown(self, topic_hash: int, evictions: int, lage: int, now: flo
             return
         win = left_wins(mine.lage(now), mine.hash, lage, topic_hash)
         if win:
-            self.schedule_gossip_urgent(mine, now)
+            self.schedule_gossip_urgent(mine)
         else:
             self.topic_allocate(mine, mine.evictions + 1, now)
 
     def topic_subscribe_if_matching(self, name: str, topic_hash: int, evictions: int, lage: int) -> TopicImpl | None:
         """Create an implicit topic if any pattern subscriber matches the name."""
+        # Validate that the hash matches the name to prevent corrupt gossip from creating inconsistencies.
+        if rapidhash(name) != topic_hash:
+            _logger.debug("Gossip hash mismatch for '%s': got %016x, expected %016x", name, topic_hash, rapidhash(name))
+            return None
         for pattern, root in self.sub_roots_pattern.items():
             subs = match_pattern(pattern, name)
             if subs is not None:
                 now = time.monotonic()
+                clamped_lage = max(0, min(35, lage))
                 topic = TopicImpl(self, name, evictions, now)
-                topic.ts_origin = now - max(1, (1 << max(0, lage)))
+                topic.ts_origin = now - max(1, (1 << clamped_lage))
                 self.topics_by_name[name] = topic
                 self.topics_by_hash[topic_hash] = topic
                 self.topic_allocate(topic, evictions, now)

src/pycyphal/_subscriber.py

@@ -140,7 +140,7 @@ def deliver(self, arrival: Arrival, tag: int, remote_id: int) -> None:
             # Out-of-order but within capacity: intern.
             if lin_tag not in state.interned:
                 state.interned[lin_tag] = InternedMsg(arrival=arrival, tag=tag, remote_id=remote_id, lin_tag=lin_tag)
-                self._arm_reorder_timeout(state, key)
+                self._arm_reorder_timeout(state)
         else:
             # Too far ahead: force-eject and resequence.
             self._force_eject_all(state)
@@ -176,7 +176,7 @@ def _force_eject_all(self, state: ReorderingState) -> None:
             state.timeout_handle.cancel()
             state.timeout_handle = None
 
-    def _arm_reorder_timeout(self, state: ReorderingState, key: tuple[int, int]) -> None:
+    def _arm_reorder_timeout(self, state: ReorderingState) -> None:
         """Arm or rearm the reordering timeout."""
         if state.timeout_handle is not None:
             return  # already armed

src/pycyphal/udp.py

@@ -348,7 +348,7 @@ def __init__(self, transport: UDPTransport, subject_id: int) -> None:
     async def __call__(self, deadline: Instant, priority: Priority, message: bytes | memoryview) -> None:
         if self._closed:
             raise SendError("Writer closed")
-        if self._transport._closed:
+        if self._transport.closed:
             raise SendError("Transport closed")
 
         mcast_ip, port = make_subject_endpoint(self._subject_id)
@@ -358,12 +358,12 @@ async def __call__(self, deadline: Instant, priority: Priority, message: bytes |
 
         errors: list[Exception] = []
         success_count = 0
-        for i, iface in enumerate(self._transport._interfaces):
+        for i, iface in enumerate(self._transport.interfaces):
             mtu = iface.mtu_cyphal
-            frames = _segment_transfer(priority, transfer_id, self._transport._uid, message, mtu)
+            frames = _segment_transfer(priority, transfer_id, self._transport.uid, message, mtu)
             try:
                 for frame in frames:
-                    await self._transport._async_sendto(self._transport._tx_socks[i], frame, (mcast_ip, port), deadline)
+                    await self._transport.async_sendto(self._transport.tx_socks[i], frame, (mcast_ip, port), deadline)
                 success_count += 1
             except (OSError, SendError) as e:
                 errors.append(e)
@@ -400,21 +400,7 @@ def close(self) -> None:
             return
         self._closed = True
         _logger.info("Subject listener closed for subject %d", self._subject_id)
-        handlers = self._transport._subject_handlers.get(self._subject_id, [])
-        if self._handler in handlers:
-            handlers.remove(self._handler)
-        if not handlers:
-            # No more listeners for this subject -- clean up sockets and tasks
-            self._transport._subject_handlers.pop(self._subject_id, None)
-            self._transport._reassemblers.pop(self._subject_id, None)
-            for i in range(len(self._transport._interfaces)):
-                key = (self._subject_id, i)
-                task = self._transport._mcast_rx_tasks.pop(key, None)
-                if task is not None:
-                    task.cancel()
-                sock = self._transport._mcast_socks.pop(key, None)
-                if sock is not None:
-                    sock.close()
+        self._transport.remove_subject_listener(self._subject_id, self._handler)
 
 
 # =====================================================================================================================
@@ -544,9 +530,48 @@ def _create_mcast_socket(subject_id: int, iface: Interface) -> socket.socket:
         _logger.info("Multicast socket for subject %d on %s (%s:%d)", subject_id, iface.address, mcast_ip, port)
         return sock
 
+    # -- Public accessors for internal classes --
+
+    @property
+    def closed(self) -> bool:
+        return self._closed
+
+    @property
+    def uid(self) -> int:
+        assert self._uid is not None
+        return self._uid
+
+    @property
+    def interfaces(self) -> list[Interface]:
+        return self._interfaces
+
+    @property
+    def tx_socks(self) -> list[socket.socket]:
+        return self._tx_socks
+
+    def remove_subject_listener(self, subject_id: int, handler: Callable[[TransportArrival], None]) -> None:
+        """
+        Remove a handler for a subject; clean up sockets/tasks if no handlers remain.
+        Internal use only.
+        """
+        handlers = self._subject_handlers.get(subject_id, [])
+        if handler in handlers:
+            handlers.remove(handler)
+        if not handlers:
+            self._subject_handlers.pop(subject_id, None)
+            self._reassemblers.pop(subject_id, None)
+            for i in range(len(self._interfaces)):
+                key = (subject_id, i)
+                task = self._mcast_rx_tasks.pop(key, None)
+                if task is not None:
+                    task.cancel()
+                sock = self._mcast_socks.pop(key, None)
+                if sock is not None:
+                    sock.close()
+
     # -- Async sendto helper --
 
-    async def _async_sendto(self, sock: socket.socket, data: bytes, addr: tuple[str, int], deadline: Instant) -> None:
+    async def async_sendto(self, sock: socket.socket, data: bytes, addr: tuple[str, int], deadline: Instant) -> None:
         """Send a UDP datagram, suspending until writable or deadline exceeded."""
         remaining_ns = deadline.ns - Instant.now().ns
         if remaining_ns <= 0:
@@ -600,7 +625,7 @@ async def unicast(self, deadline: Instant, priority: Priority, remote_id: int, m
             frames = _segment_transfer(priority, transfer_id, self._uid, message, iface.mtu_cyphal)
             try:
                 for frame in frames:
-                    await self._async_sendto(self._tx_socks[i], frame, ep, deadline)
+                    await self.async_sendto(self._tx_socks[i], frame, ep, deadline)
                 success_count += 1
             except (OSError, SendError) as e:
                 errors.append(e)