From 0251816eb1780d3343853dbcbc01b7a9078a46ff Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 14 Jun 2026 06:18:58 +0100 Subject: [PATCH] Add Python CultMesh discovery helpers --- packages/cultcache-py/PARITY.md | 7 +- .../cultcache-py/src/cultmesh_py/client.py | 95 ++++++++++++++++++- packages/cultcache-py/tests/test_cultcache.py | 24 ++++- 3 files changed, 120 insertions(+), 6 deletions(-) diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index 33289da1..58d54a37 100644 --- a/packages/cultcache-py/PARITY.md +++ b/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, diff --git a/packages/cultcache-py/src/cultmesh_py/client.py b/packages/cultcache-py/src/cultmesh_py/client.py index 91fa9b82..914b7ce4 100644 --- a/packages/cultcache-py/src/cultmesh_py/client.py +++ b/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 diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 0a937cfd..253a3885 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/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(