GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 4 Projects Releases Packages Wiki Activity Actions 4

Serve Python CultMesh discovery over CultNet

Browse source
This commit is contained in:
Meta Vens 2026-06-13 21:37:57 +01:00
commit 45f7db14b5
6 changed files with 437 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

14
packages/cultcache-py/README.md
View file

@ -159,12 +159,13 @@ python -m cultnet_py.interop_peer dial --runtime-id python-client --runtime-kind
python -m cultnet_py.interop_peer probe --runtime-id python-prober --discovery-port 4075 --discovery-group 239.77.44.11
```
`cultmesh_py` starts with a local cache-backed node and schema-v0 helpers for
the CultMesh Verse catalog and peer exchange wire messages:
`cultmesh_py` includes a local cache-backed node, schema-v0 helpers for the
CultMesh Verse catalog and peer exchange wire messages, local authority lease
checks, and stream transport negotiation:
```python
from cultcache_py import define_database_entry_type
from cultmesh_py import CultMeshPeerCatalog, create_node, peer_exchange_request
from cultmesh_py import CultMeshPeerCatalog, CultMeshStreamCatalog, create_node, peer_exchange_request
note_doc = define_database_entry_type("mesh.note", [("body", 0)])
node = create_node("mesh.cc", runtime_id="python-runtime")
@ -173,6 +174,8 @@ node.put(note_doc, "note:1", {"body": "hello"})
peers = CultMeshPeerCatalog()
response = peers.create_response(peer_exchange_request("pex-1", verse_id="local"))
streams = CultMeshStreamCatalog()
```
## Wire Parity
@ -191,6 +194,7 @@ Current receipts:
- `packages/cultnet-ts/test/interop/cultnet-interop.test.ts` includes Python in
the live TS/Rust/C#/Python schema-v0 peer ring: discovery, hello, schema
catalog, raw snapshot, raw document put, mutation receipt, and fire-command
receipt.
receipt. The same test asks the Python peer for CultMesh Verse catalog and
peer exchange responses over the CultNet pipe.
- `packages/cultcache-py/tests/test_cultcache.py` covers Python CultMesh
Verse catalog and peer exchange MessagePack wire shapes.
Verse catalog, peer exchange, authority lease, and stream negotiation behavior.

14
packages/cultcache-py/src/cultmesh_py/__init__.py
View file

@ -6,6 +6,13 @@ from .wire import (
VERSE_CATALOG_RESPONSE,
CultMeshPeerCard,
CultMeshPeerCatalog,
CultMeshAuthorityLease,
CultMeshAuthorityLeaseCatalog,
CultMeshStreamCatalog,
CultMeshStreamConsumerProfile,
CultMeshStreamDescriptor,
CultMeshStreamFrameHandle,
CultMeshStreamNegotiation,
CultMeshVerseCatalog,
CultMeshVerseCompatibility,
CultMeshVerseDescriptor,
@ -17,6 +24,13 @@ __all__ = [
"CultMeshNode",
"CultMeshPeerCard",
"CultMeshPeerCatalog",
"CultMeshAuthorityLease",
"CultMeshAuthorityLeaseCatalog",
"CultMeshStreamCatalog",
"CultMeshStreamConsumerProfile",
"CultMeshStreamDescriptor",
"CultMeshStreamFrameHandle",
"CultMeshStreamNegotiation",
"CultMeshVerseCatalog",
"CultMeshVerseCompatibility",
"CultMeshVerseDescriptor",

200
packages/cultcache-py/src/cultmesh_py/wire.py
View file

@ -1,6 +1,7 @@
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any
@ -134,6 +135,195 @@ class CultMeshPeerCatalog:
"peers": peers,
}
def find(self, verse_id: str, role: str | None = None) -> list[CultMeshPeerCard]:
require_non_empty(verse_id, "verse_id")
return [
peer
for peer in sorted(self._peers.values(), key=lambda item: item.peer_id)
if peer.verse_id == verse_id and (role is None or role in peer.roles)
]
@dataclass(frozen=True)
class CultMeshAuthorityLease:
lease_id: str
verse_id: str
peer_id: str
roles: tuple[str, ...]
valid_from: datetime
expires_at: datetime
shard_ids: tuple[str, ...] = ()
@dataclass
class CultMeshAuthorityLeaseCatalog:
_leases: dict[str, CultMeshAuthorityLease] = field(default_factory=dict)
def upsert(self, lease: CultMeshAuthorityLease) -> None:
require_non_empty(lease.lease_id, "lease.lease_id")
require_non_empty(lease.verse_id, "lease.verse_id")
require_non_empty(lease.peer_id, "lease.peer_id")
if lease.expires_at <= lease.valid_from:
raise ValueError("CultMesh authority lease expiry must be after valid_from")
self._leases[lease.lease_id] = lease
def is_authorized(
self,
peer: CultMeshPeerCard,
role: str,
shard_id: str | None = None,
at: datetime | None = None,
) -> bool:
require_non_empty(role, "role")
if not peer.authority_lease_id:
return False
lease = self._leases.get(peer.authority_lease_id)
if lease is None:
return False
checked_at = at or datetime.now(lease.valid_from.tzinfo)
return (
lease.valid_from <= checked_at < lease.expires_at
and lease.verse_id == peer.verse_id
and lease.peer_id == peer.peer_id
and role in lease.roles
and role in peer.roles
and (shard_id is None or not lease.shard_ids or shard_id in lease.shard_ids)
)
ZERO_COPY_TRANSPORTS = {
"shared-memory",
"shared-d3d12-texture",
"shared-d3d11-texture",
"dma-buf",
"iosurface",
"ahardwarebuffer",
}
@dataclass(frozen=True)
class CultMeshStreamDescriptor:
stream_id: str
verse_id: str
owner_peer_id: str
kind: str
clock: dict[str, Any]
preferred_transports: tuple[str, ...]
label: str | None = None
audio: dict[str, Any] | None = None
video: dict[str, Any] | None = None
required_access: str = "read"
max_in_flight_frames: int | None = None
metadata_schema_id: str | None = None
@dataclass(frozen=True)
class CultMeshStreamConsumerProfile:
peer_id: str
verse_id: str
supported_transports: tuple[str, ...]
accepted_kinds: tuple[str, ...] = ()
can_import_gpu_handles: bool = False
can_map_shared_memory: bool = False
max_in_flight_frames: int | None = None
@dataclass(frozen=True)
class CultMeshStreamNegotiation:
stream_id: str
producer_peer_id: str
consumer_peer_id: str
transport: str
access: str
max_in_flight_frames: int
copy_budget: str
@dataclass(frozen=True)
class CultMeshStreamFrameHandle:
stream_id: str
sequence: int
timestamp_ns: int
transport: str
duration_ns: int | None = None
byte_length: int | None = None
native_handle: str | None = None
resource_key: str | None = None
page_ref: str | None = None
fence_handle: str | None = None
fence_value: int | None = None
unavoidable_copy_count: int | None = None
metadata: dict[str, Any] | None = None
@dataclass
class CultMeshStreamCatalog:
_streams: dict[str, CultMeshStreamDescriptor] = field(default_factory=dict)
_latest_frames: dict[str, CultMeshStreamFrameHandle] = field(default_factory=dict)
@property
def streams(self) -> list[CultMeshStreamDescriptor]:
return [self._streams[key] for key in sorted(self._streams)]
def declare(self, stream: CultMeshStreamDescriptor) -> CultMeshStreamDescriptor:
require_non_empty(stream.stream_id, "stream.stream_id")
require_non_empty(stream.verse_id, "stream.verse_id")
require_non_empty(stream.owner_peer_id, "stream.owner_peer_id")
require_non_empty(str(stream.clock.get("clockDomainId", "")), "stream.clock.clockDomainId")
if not stream.preferred_transports:
raise ValueError("stream.preferred_transports must not be empty")
self._streams[stream.stream_id] = stream
return stream
def get(self, stream_id: str) -> CultMeshStreamDescriptor | None:
require_non_empty(stream_id, "stream_id")
return self._streams.get(stream_id)
def find(self, verse_id: str, kind: str | None = None) -> list[CultMeshStreamDescriptor]:
require_non_empty(verse_id, "verse_id")
return [
stream
for stream in self.streams
if stream.verse_id == verse_id and (kind is None or stream.kind == kind)
]
def negotiate(self, stream_id: str, consumer: CultMeshStreamConsumerProfile) -> CultMeshStreamNegotiation:
stream = self.get(stream_id)
if stream is None:
raise ValueError(f"Unknown CultMesh stream {stream_id!r}")
if consumer.verse_id != stream.verse_id:
raise ValueError("stream and consumer must belong to the same Verse")
if consumer.accepted_kinds and stream.kind not in consumer.accepted_kinds:
raise ValueError(f"consumer does not accept {stream.kind} streams")
transport = next(
(candidate for candidate in stream.preferred_transports if candidate in consumer.supported_transports),
None,
)
if transport is None:
raise ValueError("stream and consumer have no compatible body transport")
stream_max = stream.max_in_flight_frames if stream.max_in_flight_frames is not None else 2**53 - 1
consumer_max = consumer.max_in_flight_frames if consumer.max_in_flight_frames is not None else 2**53 - 1
return CultMeshStreamNegotiation(
stream_id=stream.stream_id,
producer_peer_id=stream.owner_peer_id,
consumer_peer_id=consumer.peer_id,
transport=transport,
access=stream.required_access,
max_in_flight_frames=min(stream_max, consumer_max),
copy_budget=copy_budget_for(transport),
)
def publish_frame(self, frame: CultMeshStreamFrameHandle) -> CultMeshStreamFrameHandle:
require_non_empty(frame.stream_id, "frame.stream_id")
if frame.stream_id not in self._streams:
raise ValueError(f"Unknown CultMesh stream {frame.stream_id!r}")
self._latest_frames[frame.stream_id] = frame
return frame
def latest_frame(self, stream_id: str) -> CultMeshStreamFrameHandle | None:
require_non_empty(stream_id, "stream_id")
return self._latest_frames.get(stream_id)
def verse_catalog_request(message_id: str, *, verse_ids: list[str] | None = None, transport_version: str | None = None) -> dict[str, Any]:
return {
@ -165,3 +355,13 @@ def peer_exchange_request(
def require_non_empty(value: str, name: str) -> None:
if not value or not value.strip():
raise ValueError(f"{name} must be non-empty")
def copy_budget_for(transport: str) -> str:
if transport in ZERO_COPY_TRANSPORTS:
return "zero-copy-target"
if transport == "cultcache-page":
return "one-copy-fallback"
if transport == "inline-bytes":
return "opaque-runtime"
return "opaque-runtime"

53
packages/cultcache-py/src/cultnet_py/interop_peer.py
View file

@ -23,6 +23,15 @@ from cultcache_py import (
define_document_type,
)
from cultcache_py.documents import DocumentDefinition
from cultmesh_py import (
PEER_EXCHANGE_REQUEST,
VERSE_CATALOG_REQUEST,
CultMeshPeerCard,
CultMeshPeerCatalog,
CultMeshVerseCatalog,
CultMeshVerseCompatibility,
CultMeshVerseDescriptor,
)
from .framing import read_frame, write_frame
@ -65,6 +74,8 @@ class PeerState:
note_schema_id: str
interop_schema: dict[str, Any]
interop_schema_json: str
verse_catalog: CultMeshVerseCatalog
peer_catalog: CultMeshPeerCatalog
def main(argv: list[str] | None = None) -> int:
@ -187,6 +198,10 @@ def handle_server_message(state: PeerState, message: dict[str, Any]) -> list[dic
return [raw_snapshot_response(state, message)]
if schema_version == "cultnet.document_put_raw.v0":
return handle_raw_put(state, message)
if schema_version == VERSE_CATALOG_REQUEST:
return [state.verse_catalog.create_response(message)]
if schema_version == PEER_EXCHANGE_REQUEST:
return [state.peer_catalog.create_response(message)]
return []
@ -373,9 +388,47 @@ def build_state(runtime_id: str, runtime_kind: str, display_name: str, agent_id:
note_schema_id=note_schema_id,
interop_schema=schema,
interop_schema_json=schema_json,
verse_catalog=default_verse_catalog(runtime_id),
peer_catalog=default_peer_catalog(runtime_id),
)
def default_verse_catalog(runtime_id: str) -> CultMeshVerseCatalog:
catalog = CultMeshVerseCatalog()
catalog.upsert(
CultMeshVerseDescriptor(
verse_id="python-interop",
display_name="Python Interop Verse",
authority_model="OperatorCluster",
compatibility=CultMeshVerseCompatibility(
transport_version="cultmesh.v0",
rules_hash="python-interop-rules",
required_plugin_ids=("core",),
),
discovery_endpoints=(f"cultnet://{runtime_id}",),
authority_runtime_ids=(runtime_id,),
description="Python CultMesh interop surface",
)
)
return catalog
def default_peer_catalog(runtime_id: str) -> CultMeshPeerCatalog:
catalog = CultMeshPeerCatalog()
catalog.upsert(
CultMeshPeerCard(
peer_id=runtime_id,
verse_id="python-interop",
endpoints=(f"cultnet://{runtime_id}",),
roles=("discovery", "read-replica", "shard-primary"),
shard_ids=("interop",),
region="local",
authority_lease_id=f"lease:{runtime_id}",
)
)
return catalog
def define_interop_documents(note_schema_id: str, note_schema_json: str) -> dict[str, Binding]:
return {
"note": binding(INTEROP_DOCUMENT_TYPE, note_schema_id, INTEROP_SCHEMA_VERSION, note_schema_json, note_slots, note_from_slots),

80
packages/cultcache-py/tests/test_cultcache.py
View file

@ -3,6 +3,7 @@
import tempfile
import unittest
from dataclasses import asdict, dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path
from cultcache_py import (
@ -18,6 +19,12 @@ from cultmesh_py import create_node
from cultmesh_py import (
CultMeshPeerCard,
CultMeshPeerCatalog,
CultMeshAuthorityLease,
CultMeshAuthorityLeaseCatalog,
CultMeshStreamCatalog,
CultMeshStreamConsumerProfile,
CultMeshStreamDescriptor,
CultMeshStreamFrameHandle,
CultMeshVerseCatalog,
CultMeshVerseCompatibility,
CultMeshVerseDescriptor,
@ -236,6 +243,79 @@ class CultCacheTests(unittest.TestCase):
self.assertIn("read-replica", decoded["peers"][0]["roles"])
self.assertEqual(decoded["peers"][0]["authorityLeaseId"], "lease-1")
def test_cultmesh_authority_lease_requires_live_matching_lease(self) -> None:
peer = CultMeshPeerCard(
peer_id="voidbot-local",
verse_id="local",
endpoints=("cultmesh://localhost",),
roles=("shard-primary",),
authority_lease_id="lease:voidbot-local",
)
leases = CultMeshAuthorityLeaseCatalog()
now = datetime.now(UTC)
self.assertFalse(leases.is_authorized(peer, "shard-primary", at=now))
leases.upsert(
CultMeshAuthorityLease(
lease_id="lease:voidbot-local",
verse_id="local",
peer_id="voidbot-local",
roles=("shard-primary",),
valid_from=now - timedelta(seconds=1),
expires_at=now + timedelta(seconds=1),
)
)
self.assertTrue(leases.is_authorized(peer, "shard-primary", at=now))
self.assertFalse(leases.is_authorized(peer, "read-replica", at=now))
def test_cultmesh_stream_catalog_negotiates_transport_and_latest_frame(self) -> None:
streams = CultMeshStreamCatalog()
streams.declare(
CultMeshStreamDescriptor(
stream_id="mimir:kiyo-pro",
verse_id="studio",
owner_peer_id="starfire",
kind="video",
label="Kiyo Pro",
clock={"clockDomainId": "starfire-qpc", "confidence": 0.25},
video={"width": 1920, "height": 1080, "pixelFormat": "YUY2", "framesPerSecond": 30},
preferred_transports=("shared-d3d12-texture", "shared-memory", "cultcache-page"),
max_in_flight_frames=3,
)
)
negotiation = streams.negotiate(
"mimir:kiyo-pro",
CultMeshStreamConsumerProfile(
peer_id="fensalir",
verse_id="studio",
supported_transports=("shared-d3d12-texture", "cultcache-page"),
accepted_kinds=("video",),
can_import_gpu_handles=True,
max_in_flight_frames=2,
),
)
self.assertEqual(negotiation.transport, "shared-d3d12-texture")
self.assertEqual(negotiation.max_in_flight_frames, 2)
self.assertEqual(negotiation.copy_budget, "zero-copy-target")
streams.publish_frame(
CultMeshStreamFrameHandle(
stream_id="mimir:kiyo-pro",
sequence=42,
timestamp_ns=1_000_000_000,
duration_ns=33_333_334,
transport="shared-d3d12-texture",
native_handle="0xfeed",
fence_handle="0xbeef",
fence_value=7,
unavoidable_copy_count=0,
)
)
self.assertEqual(streams.latest_frame("mimir:kiyo-pro").sequence, 42)
if __name__ == "__main__":
unittest.main()

94
packages/cultnet-ts/test/interop/cultnet-interop.test.ts
View file

@ -2,12 +2,14 @@ import assert from "node:assert/strict";
import { execFile, spawn, type ChildProcessByStdio } from "node:child_process";
import { once } from "node:events";
import { existsSync, rmSync } from "node:fs";
import { createServer } from "node:net";
import { connect as connectTcp, createServer } from "node:net";
import { homedir, networkInterfaces, tmpdir } from "node:os";
import { dirname, join, resolve } from "node:path";
import type { Readable } from "node:stream";
import { test } from "node:test";
import { promisify } from "node:util";
import { decode, encode } from "@msgpack/msgpack";
import { encodeFrame, LengthPrefixedMessageFramer } from "../../src/framing";
const execFileAsync = promisify(execFile);
@ -238,46 +240,63 @@ test("CultNet TS/Rust/C#/Python peers discover each other and exchange raw state
assert.ok(tsDialPython.mutatedNote.tags.includes("decorated:ts-python-client"));
assert.equal(tsDialPython.fireReceipt.ammoRemaining, 29);
const pythonVerseCatalog = await requestCultMeshFromPython(pythonPort, {
schemaVersion: "cultmesh.verse_catalog_request.v0",
messageId: "python-verses",
transportVersion: "cultmesh.v0",
});
assert.equal(pythonVerseCatalog.schemaVersion, "cultmesh.verse_catalog_response.v0");
assert.equal(pythonVerseCatalog.messageId, "python-verses");
assert.equal(pythonVerseCatalog.verses[0].verseId, "python-interop");
assert.equal(pythonVerseCatalog.verses[0].compatibility.transportVersion, "cultmesh.v0");
const pythonPeerExchange = await requestCultMeshFromPython(pythonPort, {
schemaVersion: "cultmesh.peer_exchange_request.v0",
messageId: "python-peers",
verseId: "python-interop",
roles: ["read-replica"],
});
assert.equal(pythonPeerExchange.schemaVersion, "cultmesh.peer_exchange_response.v0");
assert.equal(pythonPeerExchange.messageId, "python-peers");
assert.equal(pythonPeerExchange.peers[0].peerId, "python-peer");
assert.ok(pythonPeerExchange.peers[0].roles.includes("read-replica"));
const expectedPeers = ["csharp-peer", "python-peer", "rust-peer", "ts-peer"];
const tsProbe = await runJsonCommand("ts-probe", process.execPath, [
await expectProbePeers("ts-probe", process.execPath, [
tsPeerScript,
"probe",
"--runtime-id", "ts-prober",
"--discovery-port", String(discoveryPort),
"--discovery-group", discoveryGroup,
"--timeout-ms", "3000",
], cultNetTsRoot);
assert.deepEqual(tsProbe.peers.map((peer: { runtimeId: string }) => peer.runtimeId).sort(), expectedPeers);
], cultNetTsRoot, {}, expectedPeers);
const rustProbe = await runJsonCommand("rust-probe", rustBinaryPath, [
await expectProbePeers("rust-probe", rustBinaryPath, [
"probe",
"--runtime-id", "rust-prober",
"--discovery-port", String(discoveryPort),
"--discovery-group", discoveryGroup,
"--timeout-ms", "3000",
], cultnetRsRoot);
assert.deepEqual(rustProbe.peers.map((peer: { runtimeId: string }) => peer.runtimeId).sort(), expectedPeers);
], cultnetRsRoot, {}, expectedPeers);
const csharpProbe = await runJsonCommand("csharp-probe", dotnetCommand, [
await expectProbePeers("csharp-probe", dotnetCommand, [
csharpDllPath,
"probe",
"--runtime-id", "csharp-prober",
"--discovery-port", String(discoveryPort),
"--discovery-group", discoveryGroup,
"--timeout-ms", "3000",
], cultLibRoot);
assert.deepEqual(csharpProbe.peers.map((peer: { runtimeId: string }) => peer.runtimeId).sort(), expectedPeers);
], cultLibRoot, {}, expectedPeers);
const pythonProbe = await runJsonCommand("python-probe", pythonCommand, [
await expectProbePeers("python-probe", pythonCommand, [
"-m", "cultnet_py.interop_peer",
"probe",
"--runtime-id", "python-prober",
"--discovery-port", String(discoveryPort),
"--discovery-group", discoveryGroup,
"--timeout-ms", "3000",
], cultcachePyRoot, { PYTHONPATH: cultcachePySrc });
assert.deepEqual(pythonProbe.peers.map((peer: { runtimeId: string }) => peer.runtimeId).sort(), expectedPeers);
], cultcachePyRoot, { PYTHONPATH: cultcachePySrc }, expectedPeers);
});
async function buildInteropPeers(): Promise<void> {
@ -404,6 +423,55 @@ async function runJsonCommand(
}
}
async function expectProbePeers(
name: string,
command: string,
args: string[],
cwd: string,
env: NodeJS.ProcessEnv,
expectedPeers: string[],
): Promise<void> {
const found = new Set<string>();
for (let attempt = 0; attempt < 3; attempt += 1) {
const probe = await runJsonCommand(`${name}-${attempt + 1}`, command, args, cwd, env);
for (const peer of probe.peers as Array<{ runtimeId: string }>) {
found.add(peer.runtimeId);
}
if (expectedPeers.every((peer) => found.has(peer))) {
assert.deepEqual([...found].sort(), expectedPeers);
return;
}
}
assert.deepEqual([...found].sort(), expectedPeers);
}
async function requestCultMeshFromPython(port: number, request: Record<string, unknown>): Promise<any> {
const socket = connectTcp(port, "127.0.0.1");
const framer = new LengthPrefixedMessageFramer();
await once(socket, "connect");
return await new Promise((resolve, reject) => {
const timeout = setTimeout(() => {
socket.destroy();
reject(new Error(`Timed out waiting for Python CultMesh response to ${request.schemaVersion}.`));
}, 5000);
socket.on("data", (chunk) => {
for (const frame of framer.push(chunk)) {
clearTimeout(timeout);
socket.end();
resolve(decode(frame));
}
});
socket.once("error", (error) => {
clearTimeout(timeout);
reject(error);
});
socket.write(encodeFrame(encode(request)));
});
}
async function stopProcess(processState: RunningServeProcess): Promise<void> {
if (processState.child.killed || processState.child.exitCode !== null) {
return;