Add `open_topology_events()`, push-based discovery

Second half of layer B: an `@acm` yielding a `trio` receive-chan
of `TIPCNameEvent` fed by a nursery-spawned reader on a
`SOCK_SEQPACKET` conn to `TIPC_TOP_SRV`.

This is the bit that makes #378's "end game cluster proto" claim
real — the kernel *tells* us when any actor anywhere in the
cluster publishes or withdraws a service name, so a registrar
never has to poll `find_actor()`. Groundwork for the push
registry in `discovery/_registry.py` (gh #184, #216).

Deats,
- `filt` selects granularity; `TIPC_SUB_SERVICE` is one event
  per *name*, `TIPC_SUB_PORTS` one per *publisher* — the latter
  makes the §2.3 duplicate-name/round-robin crosstalk case
  externally observable, which is how a push-registry could
  ever detect it.
- a full event buf **drops** w/ a loud warning rather than
  blocking the reader; stalling it just backs up the kernel's
  own queue and loses the event less visibly.
- `SOCK_SEQPACKET` is fine here bc this sock never goes through
  `MsgpackTransport` — the contract's "`SOCK_STREAM` only" rule
  is about `MsgTransport` streams, not this.

XXX teardown order is load-bearing: cancel the nursery BEFORE
closing the fd. `.close()`ing out from under a pending
`.recv()` races — trio's retry can land on an already-freed fd
and raise a bare `OSError(EBADF)` instead of the
`ClosedResourceError` the reader guards for, which then escapes
the nursery as an eg.

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))

tests/ipc/test_tipc.py

@@ -43,6 +43,7 @@ from tractor.ipc._tipc import (
     TIPCAddress,
     instance_from_seed,
     is_tipc_available,
+    open_topology_events,
     start_listener,
 )
 
@@ -703,3 +704,108 @@ def test_decode_name_event_rejects_junk():
         stype=TRACTOR_STYPE,
         scope=TIPC_CLUSTER_SCOPE,
     ) is None
+
+
+@requires_tipc
+def test_topology_reports_publish_and_withdraw():
+    '''
+    "Publishing a bind IS registration" — but *observed from the
+    outside*, by the kernel pushing us the name-table transition.
+
+    This is the whole point of layer B: cluster-wide service
+    (de)registration with NO registrar actor and NO polling.
+
+    '''
+    async def main():
+        addr: TIPCAddress = TIPCAddress.get_random()
+        got: list = []
+
+        async with open_topology_events(
+            stype=addr._stype,
+        ) as events:
+            # publish..
+            lstnr = await start_listener(addr=addr)
+            with trio.fail_after(5):
+                got.append(await events.receive())
+
+            # ..then withdraw
+            lstnr.socket.close()
+            with trio.fail_after(5):
+                got.append(await events.receive())
+
+        kinds: list[str] = [ev.kind for ev in got]
+        assert kinds == ['published', 'withdrawn']
+
+        for ev in got:
+            assert ev.addr._instance == addr._instance
+            assert ev.addr._stype == addr._stype
+            assert isinstance(ev.ref, int)
+        # both transitions name the SAME publisher port
+        assert got[0].ref == got[1].ref
+        assert 'published' in repr(got[0])
+
+    trio.run(main)
+
+
+@requires_tipc
+def test_topology_acm_closes_cleanly():
+    '''
+    The `@acm` must tear down w/o leaking its reader task — exit
+    closes the sock, the reader's `.recv()` raises
+    `ClosedResourceError`, the nursery collapses.
+
+    Also assert an *unconsumed* subscription doesn't wedge exit.
+
+    '''
+    async def main():
+        addr: TIPCAddress = TIPCAddress.get_random()
+        with trio.fail_after(10):
+            async with open_topology_events(stype=addr._stype):
+                lstnr = await start_listener(addr=addr)
+                # deliberately DON'T read the event
+                await trio.sleep(0.2)
+                lstnr.socket.close()
+
+            # ..and a second open/close round still works
+            async with open_topology_events(
+                stype=addr._stype,
+            ) as events:
+                assert events is not None
+
+    trio.run(main)
+
+
+@requires_tipc
+def test_topology_sub_ports_reports_each_publisher():
+    '''
+    `TIPC_SUB_PORTS` gives one event per *publisher*, so the
+    duplicate-name/round-robin case (§2.3) is observable from the
+    topology feed — which is how a push-registry would ever
+    detect silent crosstalk.
+
+    '''
+    async def main():
+        addr: TIPCAddress = TIPCAddress.get_random()
+
+        async with open_topology_events(
+            stype=addr._stype,
+            filt=_tipc.TIPC_SUB_PORTS,
+        ) as events:
+            first = await start_listener(addr=addr)
+            second = await start_listener(addr=addr)
+
+            refs: set[int] = set()
+            with trio.fail_after(5):
+                for _ in range(2):
+                    ev = await events.receive()
+                    assert ev.kind == 'published'
+                    assert ev.addr._instance == addr._instance
+                    refs.add(ev.ref)
+
+            # two DISTINCT publisher ports on one service name
+            assert len(refs) == 2
+
+            first.socket.close()
+            second.socket.close()
+
+    trio.run(main)

tractor/ipc/_tipc.py

@@ -45,15 +45,20 @@ Normative refs are the kernel sources (the tipc.io docs are stale),
 '''
 from __future__ import annotations
 from contextlib import (
+    asynccontextmanager as acm,
     contextmanager as cm,
 )
 import errno
 from hashlib import blake2b
 import os
 import socket
-from socket import SOCK_STREAM
+from socket import (
+    SOCK_SEQPACKET,
+    SOCK_STREAM,
+)
 import struct
 from typing import (
+    AsyncGenerator,
     Callable,
     ClassVar,
     Literal,
@@ -945,3 +950,149 @@ def _decode_name_event(
         node=node,
         ref=ref,
     )
+
+
+async def _stream_name_events(
+    sock,
+    stype: int,
+    scope: int,
+    tx: trio.MemorySendChannel,
+) -> None:
+    '''
+    Read `struct tipc_event`s off a topology-server socket until
+    it's closed, forwarding decoded ones to `tx`.
+
+    '''
+    try:
+        while True:
+            raw: bytes = await sock.recv(_EVENT_SIZE)
+            if not raw:
+                return
+
+            if (ev := _decode_name_event(
+                raw,
+                stype=stype,
+                scope=scope,
+            )) is None:
+                continue
+
+            try:
+                tx.send_nowait(ev)
+            except trio.WouldBlock:
+                # XXX drop rather than block: stalling this reader
+                # backs up the kernel's own queue and we'd lose the
+                # event anyway, just less visibly.
+                log.warning(
+                    f'TIPC topology event buffer full, dropping!\n'
+                    f'{ev}\n'
+                    f' |_raise `buf_size` or consume faster\n'
+                )
+
+    except (
+        trio.ClosedResourceError,
+        trio.BrokenResourceError,
+    ):
+        # normal `@acm` teardown: the socket was closed under us
+        return
+
+    finally:
+        tx.close()
+
+
+@acm
+async def open_topology_events(
+    stype: int = TRACTOR_STYPE,
+    lower: int = 0,
+    upper: int = 0xFFFF_FFFF,
+    filt: int = TIPC_SUB_SERVICE,
+    scope: int = TIPC_CLUSTER_SCOPE,
+    timeout: int = TIPC_WAIT_FOREVER,
+    buf_size: int = 64,
+) -> AsyncGenerator[
+    trio.MemoryReceiveChannel[TIPCNameEvent],
+    None,
+]:
+    '''
+    Subscribe to kernel name-table events for `stype` and yield a
+    `trio` receive-channel of `TIPCNameEvent`.
+
+    This is *push-based* service discovery: the kernel tells us
+    when any actor in the cluster publishes or withdraws a name,
+    so a registrar never has to poll `find_actor()`.
+
+    `filt` selects the granularity,
+    - `TIPC_SUB_SERVICE`: one event per *name* becoming
+      (un)available — "does anyone serve this?"
+    - `TIPC_SUB_PORTS`: one event per *publisher*, so N binders on
+      one name give N events. Verified.
+
+    NOTE this socket is `SOCK_SEQPACKET` and never goes through
+    `MsgpackTransport` — the contract's "`SOCK_STREAM` only"
+    constraint is about `MsgTransport` streams, not this.
+
+    '''
+    topsrv_addr = TIPCAddress(
+        _stype=TIPC_TOP_SRV,
+        _instance=TIPC_TOP_SRV,
+        _scope=scope,
+    )
+    sock = trio_socket.socket(
+        AF_TIPC,
+        SOCK_SEQPACKET,
+    )
+    with _close_on_error(sock):
+        with _reraise_as_connerr(
+            src_excs=(OSError,),
+            addr=topsrv_addr,
+        ):
+            await sock.connect((
+                TIPC_ADDR_NAME,
+                TIPC_TOP_SRV,
+                TIPC_TOP_SRV,
+                0,  # domain: 0 == "anywhere in scope"
+            ))
+            await sock.send(
+                _mk_subscr(
+                    stype=stype,
+                    lower=lower,
+                    upper=upper,
+                    filt=filt,
+                    timeout=timeout,
+                )
+            )
+
+    log.info(
+        f'Subscribed to TIPC name-table events\n'
+        f'[>\n'
+        f' |_stype: 0x{stype:08x}\n'
+        f' |_range: [{lower}, {upper}]\n'
+        f' |_filter: {filt}\n'
+    )
+    tx: trio.MemorySendChannel
+    rx: trio.MemoryReceiveChannel
+    tx, rx = trio.open_memory_channel(buf_size)
+    try:
+        async with trio.open_nursery() as tn:
+            tn.start_soon(
+                _stream_name_events,
+                sock,
+                stype,
+                scope,
+                tx,
+            )
+            try:
+                yield rx
+            finally:
+                # XXX cancel BEFORE closing the fd!
+                #
+                # `.close()`ing out from under a pending
+                # `.recv()` races: trio's retry can land on an
+                # already-freed fd and raise a bare
+                # `OSError(EBADF)` instead of the
+                # `ClosedResourceError` the reader guards for —
+                # which then escapes as an eg from the nursery.
+                # Cancelling first makes teardown deterministic.
+                tn.cancel_scope.cancel()
+    finally:
+        sock.close()
+        await rx.aclose()