diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md
index 10e4d18b..57bc9f7d 100644
--- a/packages/cultcache-py/README.md
+++ b/packages/cultcache-py/README.md
@@ -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.
diff --git a/packages/cultcache-py/src/cultmesh_py/__init__.py b/packages/cultcache-py/src/cultmesh_py/__init__.py
index a843a01c..c2382243 100644
--- a/packages/cultcache-py/src/cultmesh_py/__init__.py
+++ b/packages/cultcache-py/src/cultmesh_py/__init__.py
@@ -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",
diff --git a/packages/cultcache-py/src/cultmesh_py/wire.py b/packages/cultcache-py/src/cultmesh_py/wire.py
index ecddbec0..eebf1616 100644
--- a/packages/cultcache-py/src/cultmesh_py/wire.py
+++ b/packages/cultcache-py/src/cultmesh_py/wire.py
@@ -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"
diff --git a/packages/cultcache-py/src/cultnet_py/interop_peer.py b/packages/cultcache-py/src/cultnet_py/interop_peer.py
index e14451e3..12eaa0c7 100644
--- a/packages/cultcache-py/src/cultnet_py/interop_peer.py
+++ b/packages/cultcache-py/src/cultnet_py/interop_peer.py
@@ -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),
diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py
index f2e7503b..3243faaf 100644
--- a/packages/cultcache-py/tests/test_cultcache.py
+++ b/packages/cultcache-py/tests/test_cultcache.py
@@ -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()
diff --git a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts
index 420c6280..b7e56668 100644
--- a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts
+++ b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts
@@ -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;