Add Python CultMesh discovery helpers

packages/cultcache-py/PARITY.md

@@ -66,8 +66,11 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti.
   8-slot witness artifact payload while preserving dict helpers for existing
   wire callers.
 - Python CultMesh discovery client:
-  Fetches Verse catalogs and peer exchange responses from the live Python
-  endpoint over CultNet frames.
+  Fetches Verse catalogs and peer exchange responses from live endpoints over
+  CultNet frames. It can also parse `cultnet://host:port` endpoints, discover
+  from endpoint lists, de-duplicate discovery targets, and upsert results into
+  local Verse and peer catalogs with the same count-returning ergonomics as the
+  C# discovery clients.
 - Python CultMesh facade:
   Exposes C#-matching client entrypoints for Verse discovery, peer exchange,
   raw CultNet client construction, local nodes, catalogs, sessions, streams,

packages/cultcache-py/src/cultmesh_py/client.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 from typing import Any
+from urllib.parse import urlparse
 
 from cultnet_py import CultNetRawClient
 
@@ -98,6 +99,21 @@ class CultMeshDiscoveryClient:
         response = self.request_verse_catalog(verse_ids=verse_ids, transport_version=transport_version)
         return catalog.apply_response(response)
 
+    def discover_verse_catalog(
+        self,
+        catalog: CultMeshVerseCatalog,
+        *,
+        endpoints: list[str] | None = None,
+        transport_version: str | None = None,
+    ) -> int:
+        if endpoints is None:
+            return len(self.sync_verse_catalog(catalog, transport_version=transport_version))
+        count = 0
+        for endpoint in _distinct_non_empty(endpoints):
+            client = type(self).from_endpoint(endpoint, timeout_seconds=self.timeout_seconds)
+            count += client.discover_verse_catalog(catalog, transport_version=transport_version)
+        return count
+
     def sync_peer_catalog(
         self,
         catalog: CultMeshPeerCatalog,
@@ -115,10 +131,85 @@ class CultMeshDiscoveryClient:
         )
         return catalog.apply_response(response)
 
+    def discover_peer_catalog(
+        self,
+        catalog: CultMeshPeerCatalog,
+        *,
+        verse_id: str,
+        endpoints: list[str] | None = None,
+        roles: list[str] | None = None,
+        limit: int | None = None,
+    ) -> int:
+        if endpoints is None:
+            known_peer_ids = [peer.peer_id for peer in catalog.find(verse_id)]
+            return len(self.sync_peer_catalog(
+                catalog,
+                verse_id=verse_id,
+                roles=roles,
+                known_peer_ids=known_peer_ids,
+                limit=limit,
+            ))
+        count = 0
+        for endpoint in _distinct_non_empty(endpoints):
+            client = type(self).from_endpoint(endpoint, timeout_seconds=self.timeout_seconds)
+            count += client.discover_peer_catalog(catalog, verse_id=verse_id, roles=roles, limit=limit)
+        return count
+
+    @classmethod
+    def from_endpoint(cls, endpoint: str, *, timeout_seconds: float = 4.0) -> "CultMeshDiscoveryClient":
+        host, port = _parse_endpoint(endpoint)
+        return cls(host, port, timeout_seconds=timeout_seconds)
+
 
 class CultMeshVerseDiscoveryClient(CultMeshDiscoveryClient):
-    pass
+    def discover(
+        self,
+        catalog: CultMeshVerseCatalog,
+        *,
+        endpoints: list[str] | None = None,
+        transport_version: str | None = None,
+    ) -> int:
+        return self.discover_verse_catalog(
+            catalog,
+            endpoints=endpoints,
+            transport_version=transport_version,
+        )
 
 
 class CultMeshPeerExchangeClient(CultMeshDiscoveryClient):
-    pass
+    def discover(
+        self,
+        catalog: CultMeshPeerCatalog,
+        *,
+        verse_id: str,
+        endpoints: list[str] | None = None,
+        roles: list[str] | None = None,
+        limit: int | None = None,
+    ) -> int:
+        return self.discover_peer_catalog(
+            catalog,
+            verse_id=verse_id,
+            endpoints=endpoints,
+            roles=roles,
+            limit=limit,
+        )
+
+
+def _parse_endpoint(endpoint: str) -> tuple[str, int]:
+    if not endpoint or not endpoint.strip():
+        raise ValueError("endpoint must be non-empty")
+    parsed = urlparse(endpoint if "://" in endpoint else f"cultnet://{endpoint}")
+    if not parsed.hostname or parsed.port is None:
+        raise ValueError(f"CultMesh endpoint must include host and port: {endpoint!r}")
+    return parsed.hostname, parsed.port
+
+
+def _distinct_non_empty(values: list[str]) -> list[str]:
+    clean: list[str] = []
+    seen: set[str] = set()
+    for value in values:
+        if not value or not value.strip() or value in seen:
+            continue
+        clean.append(value)
+        seen.add(value)
+    return clean

packages/cultcache-py/tests/test_cultcache.py

@@ -1996,9 +1996,9 @@ class CultCacheTests(unittest.TestCase):
                     server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                     server.bind(("127.0.0.1", 0))
                     port_holder.append(server.getsockname()[1])
-                    server.listen(2)
+                    server.listen(6)
                     ready.set()
-                    for _ in range(4):
+                    for _ in range(6):
                         connection, _ = server.accept()
                         with connection:
                             stream = connection.makefile("rwb")
@@ -2028,6 +2028,22 @@ class CultCacheTests(unittest.TestCase):
         local_peers = CultMeshPeerCatalog()
         synced_verses = client.sync_verse_catalog(local_verses, transport_version="cultmesh.v0")
         synced_peers = client.sync_peer_catalog(local_peers, verse_id="aetheria-main", roles=["read-replica"])
+        discovered_verses = CultMeshVerseCatalog()
+        discovered_peers = CultMeshPeerCatalog()
+        endpoint = f"cultnet://127.0.0.1:{port_holder[0]}"
+        verse_discovery = CultMeshVerseDiscoveryClient.from_endpoint(endpoint, timeout_seconds=2.0)
+        peer_discovery = CultMeshPeerExchangeClient.from_endpoint(endpoint, timeout_seconds=2.0)
+        discovered_verse_count = verse_discovery.discover(
+            discovered_verses,
+            endpoints=[endpoint, endpoint, ""],
+            transport_version="cultmesh.v0",
+        )
+        discovered_peer_count = peer_discovery.discover(
+            discovered_peers,
+            verse_id="aetheria-main",
+            endpoints=[endpoint, endpoint, ""],
+            roles=["read-replica"],
+        )
 
         thread.join(2.0)
         self.assertFalse(server_error)
@@ -2038,6 +2054,10 @@ class CultCacheTests(unittest.TestCase):
         self.assertEqual(synced_verses[0].verse_id, "aetheria-main")
         self.assertEqual(local_peers.find("aetheria-main", role="read-replica")[0].peer_id, "peer-a")
         self.assertEqual(synced_peers[0].roles, ("read-replica",))
+        self.assertEqual(discovered_verse_count, 1)
+        self.assertEqual(discovered_peer_count, 1)
+        self.assertEqual(discovered_verses.get("aetheria-main").display_name, "Aetheria")
+        self.assertEqual(discovered_peers.find("aetheria-main", role="read-replica")[0].peer_id, "peer-a")
 
     def test_cultmesh_local_server_serves_node_and_catalogs_over_clients(self) -> None:
         document = define_database_entry_type(