From d4737e957fab5f08dcc9d0226e429a0886184fc7 Mon Sep 17 00:00:00 2001
From: goodboy <goodboy_foss@protonmail.com>
Date: Fri, 14 Aug 2026 09:34:41 -0400
Subject: [PATCH] Pin `Endpoint` addr-reconciliation for tcp/uds

Guard test for `.start_listener()`s post-bind
`getsockname()`-vs-`.addr` round-trip, landed *before* that
reconciliation gets gated on an opt-out `ClassVar`.

- tcp: a `port=0` bind MUST still learn the kernel-picked
  port, since the reconciliation is the only path that ever
  does.
- uds: the sock-file path must survive the `.from_addr()`
  round-trip unchanged.

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))
---
 tests/ipc/test_server.py | 69 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 69 insertions(+)

diff --git a/tests/ipc/test_server.py b/tests/ipc/test_server.py
index 1d63bd1b..0e1fd61d 100644
--- a/tests/ipc/test_server.py
+++ b/tests/ipc/test_server.py
@@ -14,6 +14,7 @@ from tractor import (
 from tractor._testing.addr import (
     get_rando_addr,
 )
+from tractor.ipc._tcp import TCPAddress
 # TODO, use/check-roundtripping with some of these wrapper types?
 #
 # from .._addr import Address
@@ -70,3 +71,71 @@ def test_basic_ipc_server(
         pdb=debug_mode,
     ):
         trio.run(main)
+
+
+@pytest.mark.parametrize(
+    '_tpt_proto',
+    ['uds', 'tcp']
+)
+def test_ep_addr_reconciled_from_sockname(
+    _tpt_proto: str,
+    debug_mode: bool,
+):
+    '''
+    Guard `Endpoint.start_listener()`'s post-bind reconciliation of
+    `.addr` against the listener's `socket.getsockname()`.
+
+    For `tcp` that reconciliation is the ONLY way a kernel-assigned
+    port (from a `port=0` bind) is ever learned, so it must keep
+    firing; for `uds` the sock-file path must survive the
+    round-trip through `.from_addr()` unchanged.
+
+    Both are pinned here *before* the reconciliation gets gated on
+    an `Address.rebind_from_sockname` opt-out (for backends whose
+    `getsockname()` reports something other than what was bound).
+
+    '''
+    async def main():
+        async with ipc._server.open_ipc_server() as server:
+
+            accept_addr: tuple[str, int|str]
+            match _tpt_proto:
+                # XXX the whole point: ask the kernel to pick.
+                case 'tcp':
+                    accept_addr = (
+                        TCPAddress.def_bindspace,
+                        0,
+                    )
+                case 'uds':
+                    accept_addr = get_rando_addr(
+                        tpt_proto=_tpt_proto,
+                    )
+
+            eps: list[ipc._server.Endpoint] = await server.listen_on(
+                accept_addrs=[accept_addr],
+                stream_handler_nursery=None,
+            )
+            assert len(eps) == 1
+            ep: ipc._server.Endpoint = eps[0]
+            sockname = ep._listener.socket.getsockname()
+
+            match _tpt_proto:
+                case 'tcp':
+                    # the bind req was for "any port"..
+                    assert accept_addr[1] == 0
+                    # ..and the ep learned the real one.
+                    assert ep.addr._port != 0
+                    assert ep.addr.unwrap() == tuple(sockname[:2])
+
+                case 'uds':
+                    # sock-file path is stable across the
+                    # `.from_addr()` round-trip.
+                    assert ep.addr.unwrap() == accept_addr
+                    assert str(ep.addr.sockpath) == sockname
+
+            server._parent_tn.cancel_scope.cancel()
+
+    with devx.maybe_open_crash_handler(
+        pdb=debug_mode,
+    ):
+        trio.run(main)