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

Advertise CultNet transport profiles

Browse source
This commit is contained in:
Meta Vens 2026-06-14 21:20:52 +01:00
commit a2bc9aeab4
21 changed files with 539 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

1
packages/cultcache-py/src/cultmesh_py/daemon.py
View file

@ -161,6 +161,7 @@ def daemon_ready_document(server: CultMeshLocalServer) -> dict[str, Any]:
"supportedDocumentTypes": [document.type for document in server.node.documents],
"supportedMessageVersions": list(hello_response.get("supportedMessageVersions") or ()),
"supportedMutationContracts": list(hello_response.get("supportedMutationContracts") or ()),
"transportProfiles": list(hello_response.get("transportProfiles") or ()),
"snapshotLimits": {
"maxSnapshotDocuments": server.max_snapshot_documents,
"maxSnapshotBytes": server.max_snapshot_bytes,

22
packages/cultcache-py/src/cultmesh_py/server.py
View file

@ -308,6 +308,28 @@ class CultMeshLocalServer:
if self.observation_hub is not None
else []
),
"transportProfiles": [
{
"schemaVersion": "cultnet.transport_profile.v0",
"runtimeId": self.node.runtime_id,
"transports": [
{
"transportId": "cultmesh-local-tcp",
"protocol": "tcp_framed",
"host": self.host,
"port": self.port,
"wireContracts": ["cultnet.schema.v0"],
"channels": [
{
"channelId": "schema",
"delivery": "reliable",
"ordering": "ordered",
}
],
}
],
}
],
"supportsSchemaCatalog": True,
}

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

@ -569,6 +569,28 @@ def discovery_loop(sock: socket.socket, advertise_host: str, tcp_port: int, stat
"tcpHost": advertise_host,
"tcpPort": tcp_port,
"wireContract": INTEROP_WIRE_CONTRACT,
"transportProfiles": [
{
"schemaVersion": "cultnet.transport_profile.v0",
"runtimeId": state.runtime_id,
"transports": [
{
"transportId": "interop-tcp",
"protocol": "tcp_framed",
"host": advertise_host,
"port": tcp_port,
"wireContracts": [INTEROP_WIRE_CONTRACT],
"channels": [
{
"channelId": "schema",
"delivery": "reliable",
"ordering": "ordered",
}
],
}
],
}
],
"supportedDocumentTypes": [INTEROP_DOCUMENT_TYPE],
"supportsSchemaCatalog": True,
}

2
packages/cultcache-py/src/cultnet_py/messages.py
View file

@ -37,6 +37,7 @@ def hello(
supported_wire_contracts: list[str] | None = None,
supported_schema_versions: list[str] | None = None,
supported_mutation_contracts: list[dict[str, Any]] | None = None,
transport_profiles: list[dict[str, Any]] | None = None,
) -> CultNetMessage:
return CultNetMessage(
"cultnet.hello.v0",
@ -45,6 +46,7 @@ def hello(
"supportedWireContracts": supported_wire_contracts or ["cultnet.schema.v0"],
"supportedSchemaVersions": supported_schema_versions or [],
"supportedMutationContracts": supported_mutation_contracts or [],
"transportProfiles": transport_profiles or [],
},
)

6
packages/cultcache-py/src/cultnet_py/schema_catalog.py
View file

@ -332,6 +332,12 @@ def wire_message_schema_properties(schema_version: str) -> dict[str, Any]:
},
},
"supportedMessageVersions": string_array,
"transportProfiles": {
"type": "array",
"items": {
"$ref": TRANSPORT_PROFILE_SCHEMA_ID,
},
},
"supportsSchemaCatalog": {"type": "boolean"},
},
"cultnet.error.v0": {

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

@ -531,10 +531,27 @@ class CultCacheTests(unittest.TestCase):
self.assertEqual(CultNetSecret.decrypt_string(encrypted, nonce, security), "hello")
def test_cultnet_schema_message_frame_round_trip(self) -> None:
message = hello(runtime_id="python-test", supported_schema_versions=["cultnet.hello.v0"])
message = hello(
runtime_id="python-test",
supported_schema_versions=["cultnet.hello.v0"],
transport_profiles=[
{
"schemaVersion": "cultnet.transport_profile.v0",
"runtimeId": "python-test",
"transports": [
{
"transportId": "test",
"protocol": "tcp_framed",
"channels": [{"channelId": "schema", "delivery": "reliable", "ordering": "ordered"}],
}
],
}
],
)
parsed = parse_message(decode_frame(encode_frame(message.to_bytes())))
self.assertEqual(parsed.schema_version, "cultnet.hello.v0")
self.assertEqual(parsed.body["runtimeId"], "python-test")
self.assertEqual(parsed.body["transportProfiles"][0]["transports"][0]["protocol"], "tcp_framed")
def test_cultnet_database_subscription_helpers_match_schema_v0_shape(self) -> None:
subscribe = database_subscribe(
@ -1293,6 +1310,10 @@ class CultCacheTests(unittest.TestCase):
mutation_contract = hello_schema["properties"]["supportedMutationContracts"]["items"]
self.assertEqual(mutation_contract["type"], "object")
self.assertIn("operations", mutation_contract["properties"])
self.assertEqual(
hello_schema["properties"]["transportProfiles"]["items"]["$ref"],
"https://github.com/GameCult/cultnet-ts/contracts/cultnet.transport-profile.schema.json",
)
consensus_schema = json.loads(by_version["cultnet.simulation_consensus_candidate.v0"]["schemaJson"])
for required_field in ["witnessCount", "supportWeight", "totalWeight", "hasQuorum", "confidence"]:
@ -3428,8 +3449,11 @@ class CultCacheTests(unittest.TestCase):
self.assertIn("cultnet.schema_catalog_request.v0", hello_response["supportedMessageVersions"])
self.assertNotIn("cultnet.simulation_observation.v0", hello_response["supportedMessageVersions"])
self.assertEqual(hello_response["supportedDocumentTypes"], ["cultcache.interop-note"])
self.assertEqual(hello_response["transportProfiles"][0]["transports"][0]["protocol"], "tcp_framed")
self.assertEqual(hello_response["transportProfiles"][0]["transports"][0]["port"], int(ready["port"]))
self.assertEqual(ready["supportedMessageVersions"], hello_response["supportedMessageVersions"])
self.assertEqual(ready["supportedMutationContracts"], hello_response["supportedMutationContracts"])
self.assertEqual(ready["transportProfiles"], hello_response["transportProfiles"])
schema_response = client.fetch_schema_catalog(schema_ids=["cultcache.interop-note"], include_schema_json=True)
self.assertEqual(schema_response["schemas"][0]["schemaVersion"], INTEROP_SCHEMA_VERSION)
@ -4074,6 +4098,8 @@ class CultCacheTests(unittest.TestCase):
synced_shards = raw_client.sync_shard_catalog(synced_shard_catalog, schema_ids=["mesh.server_note.v1"])
shard_log = raw_client.fetch_shard_log(shard_id="notes", shard_epoch=1)
typed_shard_log = raw_client.fetch_shard_log_response(shard_id="notes", shard_epoch=1)
self.assertEqual(hello_response["transportProfiles"][0]["transports"][0]["protocol"], "tcp_framed")
self.assertEqual(hello_response["transportProfiles"][0]["transports"][0]["port"], server.port)
stale_shard_log = raw_client.fetch_shard_log(shard_id="notes", shard_epoch=0)
discovery_client = CultMesh.create_verse_discovery_client("127.0.0.1", server.port, timeout_seconds=2.0)
fetched_verses = discovery_client.fetch_verses(transport_version="cultmesh.v0")

6
packages/cultnet-rs/contracts/cultnet.hello.schema.json
View file

@ -30,6 +30,12 @@
"type": "array",
"items": { "type": "string", "minLength": 1 }
},
"transportProfiles": {
"type": "array",
"items": {
"$ref": "./cultnet.transport-profile.schema.json"
}
},
"supportsSchemaCatalog": { "type": "boolean" }
}
}

54
packages/cultnet-rs/examples/cultnet_interop_peer.rs
View file

@ -14,6 +14,12 @@ use cultnet_rs::CultNetMutationAuthority;
use cultnet_rs::CultNetSchemaKind;
use cultnet_rs::CultNetSchemaRegistration;
use cultnet_rs::CultNetSchemaRegistry;
use cultnet_rs::CultNetTransportChannel;
use cultnet_rs::CultNetTransportDelivery;
use cultnet_rs::CultNetTransportDescriptor;
use cultnet_rs::CultNetTransportOrdering;
use cultnet_rs::CultNetTransportProfile;
use cultnet_rs::CultNetTransportProtocol;
use cultnet_rs::CultNetWireContract;
use cultnet_rs::builtin_schema_registry;
use cultnet_rs::decode_cultnet_message_from_slice;
@ -168,6 +174,8 @@ enum DiscoveryMessage {
tcp_port: u16,
wire_contract: String,
supported_document_types: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
transport_profiles: Option<Vec<CultNetTransportProfile>>,
supports_schema_catalog: bool,
},
}
@ -290,6 +298,7 @@ fn probe(options: &BTreeMap<String, String>) -> Result<()> {
tcp_port,
wire_contract,
supported_document_types,
transport_profiles,
supports_schema_catalog,
}) = rmp_serde::from_slice::<DiscoveryMessage>(&buffer[..len])
{
@ -307,6 +316,7 @@ fn probe(options: &BTreeMap<String, String>) -> Result<()> {
"tcpPort": tcp_port,
"wireContract": wire_contract,
"supportedDocumentTypes": supported_document_types,
"transportProfiles": transport_profiles,
"supportsSchemaCatalog": supports_schema_catalog,
}),
);
@ -364,6 +374,7 @@ fn dial(config: DialConfig) -> Result<()> {
supported_document_types: Some(vec![INTEROP_DOCUMENT_TYPE.to_string()]),
supported_mutation_contracts: Some(interaction_contracts()),
supported_message_versions: Some(vec![INTEROP_SCHEMA_VERSION.to_string()]),
transport_profiles: None,
supports_schema_catalog: Some(true),
},
)?;
@ -533,6 +544,11 @@ fn start_udp_discovery_server(config: Arc<PeerConfig>) -> Result<()> {
tcp_port: config.tcp_port,
wire_contract: "cultnet.schema.v0".to_string(),
supported_document_types: vec![INTEROP_DOCUMENT_TYPE.to_string()],
transport_profiles: Some(tcp_transport_profiles(
&config.runtime_id,
&config.advertise_host,
config.tcp_port,
)),
supports_schema_catalog: true,
};
if let Ok(payload) = rmp_serde::to_vec_named(&announce) {
@ -612,11 +628,16 @@ fn handle_connection(
agent_id: Some(config.agent_id.clone()),
role: None,
display_name: Some(config.display_name.clone()),
supported_document_types: Some(vec![INTEROP_DOCUMENT_TYPE.to_string()]),
supported_mutation_contracts: Some(interaction_contracts()),
supported_message_versions: Some(vec![INTEROP_SCHEMA_VERSION.to_string()]),
supports_schema_catalog: Some(true),
},
supported_document_types: Some(vec![INTEROP_DOCUMENT_TYPE.to_string()]),
supported_mutation_contracts: Some(interaction_contracts()),
supported_message_versions: Some(vec![INTEROP_SCHEMA_VERSION.to_string()]),
transport_profiles: Some(tcp_transport_profiles(
&config.runtime_id,
&config.advertise_host,
config.tcp_port,
)),
supports_schema_catalog: Some(true),
},
)?;
}
request @ CultNetMessage::SchemaCatalogRequest { .. } => {
@ -814,6 +835,29 @@ fn interaction_contracts() -> Vec<CultNetDocumentMutationContract> {
}]
}
fn tcp_transport_profiles(runtime_id: &str, host: &str, port: u16) -> Vec<CultNetTransportProfile> {
vec![CultNetTransportProfile {
schema_version: "cultnet.transport_profile.v0".to_string(),
runtime_id: runtime_id.to_string(),
transports: vec![CultNetTransportDescriptor {
transport_id: "interop-tcp".to_string(),
protocol: CultNetTransportProtocol::TcpFramed,
host: Some(host.to_string()),
port: Some(port),
path: None,
discovery_group: None,
wire_contracts: Some(vec!["cultnet.schema.v0".to_string()]),
channels: vec![CultNetTransportChannel {
channel_id: "schema".to_string(),
delivery: CultNetTransportDelivery::Reliable,
ordering: CultNetTransportOrdering::Ordered,
max_payload_bytes: None,
max_fragment_bytes: None,
}],
}],
}]
}
fn response_options(config: &PeerConfig, tag: &str) -> CultNetDocumentPutOptions {
CultNetDocumentPutOptions {
source_runtime_id: Some(config.runtime_id.clone()),

65
packages/cultnet-rs/src/contracts.rs
View file

@ -81,6 +81,68 @@ pub struct CultNetDocumentMutationContract {
pub notes: Option<Vec<String>>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultNetTransportProfile {
pub schema_version: String,
pub runtime_id: String,
pub transports: Vec<CultNetTransportDescriptor>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultNetTransportDescriptor {
pub transport_id: String,
pub protocol: CultNetTransportProtocol,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub host: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub port: Option<u16>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub path: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub discovery_group: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub wire_contracts: Option<Vec<String>>,
pub channels: Vec<CultNetTransportChannel>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CultNetTransportProtocol {
TcpFramed,
Litenetlib,
Websocket,
Rudp,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultNetTransportChannel {
pub channel_id: String,
pub delivery: CultNetTransportDelivery,
pub ordering: CultNetTransportOrdering,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub max_payload_bytes: Option<u32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub max_fragment_bytes: Option<u32>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CultNetTransportDelivery {
Reliable,
Unreliable,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CultNetTransportOrdering {
Ordered,
Unordered,
Sequenced,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultNetRawDocumentRecord {
@ -137,6 +199,8 @@ pub enum CultNetMessage {
#[serde(default, skip_serializing_if = "Option::is_none")]
supported_message_versions: Option<Vec<String>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
transport_profiles: Option<Vec<CultNetTransportProfile>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
supports_schema_catalog: Option<bool>,
},
#[serde(rename = "cultnet.login.v0", rename_all = "camelCase")]
@ -287,6 +351,7 @@ fn validate_message(message: &CultNetMessage) -> Result<()> {
supported_document_types,
supported_mutation_contracts,
supported_message_versions,
transport_profiles: _,
supports_schema_catalog: _,
} => {
require_non_empty(runtime_id, "runtimeId")?;

27
packages/cultnet-rs/tests/cultnet.rs
View file

@ -15,6 +15,12 @@ use cultnet_rs::CultNetSchemaKind;
use cultnet_rs::CultNetSchemaRegistry;
use cultnet_rs::CultNetSecret;
use cultnet_rs::CultNetServerSecurityOptions;
use cultnet_rs::CultNetTransportChannel;
use cultnet_rs::CultNetTransportDelivery;
use cultnet_rs::CultNetTransportDescriptor;
use cultnet_rs::CultNetTransportOrdering;
use cultnet_rs::CultNetTransportProfile;
use cultnet_rs::CultNetTransportProtocol;
use cultnet_rs::CultNetWireContract;
use cultnet_rs::LengthPrefixedMessageFramer;
use cultnet_rs::builtin_schema_registry;
@ -89,6 +95,26 @@ fn cultnet_schema_messages_round_trip_through_messagepack_frames() -> Result<()>
]),
}]),
supported_message_versions: None,
transport_profiles: Some(vec![CultNetTransportProfile {
schema_version: "cultnet.transport_profile.v0".to_string(),
runtime_id: "voidbot-main".to_string(),
transports: vec![CultNetTransportDescriptor {
transport_id: "direct-pipe".to_string(),
protocol: CultNetTransportProtocol::TcpFramed,
host: None,
port: None,
path: None,
discovery_group: None,
wire_contracts: Some(vec!["cultnet.schema.v0".to_string()]),
channels: vec![CultNetTransportChannel {
channel_id: "schema".to_string(),
delivery: CultNetTransportDelivery::Reliable,
ordering: CultNetTransportOrdering::Ordered,
max_payload_bytes: None,
max_fragment_bytes: None,
}],
}],
}]),
supports_schema_catalog: Some(true),
};
let payload = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?;
@ -146,6 +172,7 @@ fn rust_decodes_typescript_generated_cultnet_frames() -> Result<()> {
supported_document_types: Some(vec!["ghostlight.agent-state".to_string()]),
supported_mutation_contracts: None,
supported_message_versions: None,
transport_profiles: None,
supports_schema_catalog: None,
}
);

6
packages/cultnet-ts/contracts/cultnet.hello.schema.json
View file

@ -30,6 +30,12 @@
"type": "array",
"items": { "type": "string", "minLength": 1 }
},
"transportProfiles": {
"type": "array",
"items": {
"$ref": "./cultnet.transport-profile.schema.json"
}
},
"supportsSchemaCatalog": { "type": "boolean" }
}
}

2
packages/cultnet-ts/src/contracts.ts
View file

@ -102,6 +102,7 @@ export interface CultNetHelloMessage {
supportedDocumentTypes?: string[];
supportedMutationContracts?: CultNetDocumentMutationContract[];
supportedMessageVersions?: string[];
transportProfiles?: CultNetTransportProfile[];
supportsSchemaCatalog?: boolean;
}
@ -955,6 +956,7 @@ function normalizeCultNetOptionalNulls(
"supportedDocumentTypes",
"supportedMutationContracts",
"supportedMessageVersions",
"transportProfiles",
]);
return;
case "cultnet.snapshot_request.v0":

3
packages/cultnet-ts/src/index.ts
View file

@ -26,6 +26,9 @@ export {
type CultNetSnapshotRequestMessage,
type CultNetSnapshotResponseMessage,
type CultNetSnapshotResponseRawMessage,
type CultNetTransportChannel,
type CultNetTransportDescriptor,
type CultNetTransportProfile,
type CultNetVerifyMessage,
type CultNetWireContract,
type GhostlightAgent,

15
packages/cultnet-ts/test/cultnet.test.ts
View file

@ -139,6 +139,20 @@ test("CultNet peer frames and decodes typed messages over a direct pipe", async
agentId: "void",
displayName: "Void",
supportedDocumentTypes: ["ghostlight.agent-state"],
transportProfiles: [
{
schemaVersion: "cultnet.transport_profile.v0",
runtimeId: "voidbot-main",
transports: [
{
transportId: "direct-pipe",
protocol: "tcp_framed",
wireContracts: ["cultnet.schema.v0"],
channels: [{ channelId: "schema", delivery: "reliable", ordering: "ordered" }],
},
],
},
],
});
});
@ -146,6 +160,7 @@ test("CultNet peer frames and decodes typed messages over a direct pipe", async
if (message.schemaVersion === "cultnet.hello.v0") {
assert.equal(message.runtimeId, "voidbot-main");
assert.equal(message.agentId, "void");
assert.equal(message.transportProfiles?.[0]?.transports[0]?.protocol, "tcp_framed");
}
sender.close();

50
packages/cultnet-ts/test/interop/cultnet-interop-peer.ts
View file

@ -143,6 +143,8 @@ async function serve(args: Map<string, string>): Promise<void> {
runtimeKind,
displayName,
agentId,
advertiseHost,
tcpPort,
cache,
documentRegistry,
customSchemas,
@ -217,6 +219,8 @@ async function handleServerMessage(input: {
runtimeKind: string;
displayName: string;
agentId: string;
advertiseHost: string;
tcpPort: number;
cache: CultCache;
documentRegistry: CultNetDocumentRegistry;
customSchemas: CultNetSchemaRegistry;
@ -229,6 +233,8 @@ async function handleServerMessage(input: {
runtimeKind,
displayName,
agentId,
advertiseHost,
tcpPort,
cache,
documentRegistry,
customSchemas,
@ -253,6 +259,28 @@ async function handleServerMessage(input: {
receiptDocumentTypes: [INTEROP_MUTATION_RECEIPT_DOCUMENT_TYPE, INTEROP_FIRE_RECEIPT_DOCUMENT_TYPE],
}],
supportedMessageVersions: [INTEROP_SCHEMA_VERSION],
transportProfiles: [
{
schemaVersion: "cultnet.transport_profile.v0",
runtimeId,
transports: [
{
transportId: "interop-tcp",
protocol: "tcp_framed",
host: advertiseHost,
port: tcpPort,
wireContracts: [INTEROP_WIRE_CONTRACT],
channels: [
{
channelId: "schema",
delivery: "reliable",
ordering: "ordered",
},
],
},
],
},
],
supportsSchemaCatalog: true,
});
break;
@ -409,6 +437,28 @@ async function respondToProbe(input: {
tcpPort,
wireContract: INTEROP_WIRE_CONTRACT,
supportedDocumentTypes: [INTEROP_DOCUMENT_TYPE],
transportProfiles: [
{
schemaVersion: "cultnet.transport_profile.v0",
runtimeId,
transports: [
{
transportId: "interop-tcp",
protocol: "tcp_framed",
host: advertiseHost,
port: tcpPort,
wireContracts: [INTEROP_WIRE_CONTRACT],
channels: [
{
channelId: "schema",
delivery: "reliable",
ordering: "ordered",
},
],
},
],
},
],
supportsSchemaCatalog: true,
};

22
packages/cultnet-ts/test/interop/cultnet-interop-shared.ts
View file

@ -232,6 +232,27 @@ export const discoveryProbeSchema = z.object({
export type DiscoveryProbe = z.infer<typeof discoveryProbeSchema>;
export const discoveryTransportProfileSchema = z.object({
schemaVersion: z.literal("cultnet.transport_profile.v0"),
runtimeId: z.string().min(1),
transports: z.array(z.object({
transportId: z.string().min(1),
protocol: z.enum(["tcp_framed", "litenetlib", "websocket", "rudp"]),
host: z.string().min(1).nullable().optional(),
port: z.number().int().min(1).max(65535).nullable().optional(),
path: z.string().min(1).nullable().optional(),
discoveryGroup: z.string().min(1).nullable().optional(),
wireContracts: z.array(z.string().min(1)).nullable().optional(),
channels: z.array(z.object({
channelId: z.string().min(1),
delivery: z.enum(["reliable", "unreliable"]),
ordering: z.enum(["ordered", "unordered", "sequenced"]),
maxPayloadBytes: z.number().int().min(1).nullable().optional(),
maxFragmentBytes: z.number().int().min(1).nullable().optional(),
})),
})),
});
export const discoveryAnnounceSchema = z.object({
schemaVersion: z.literal(DISCOVERY_ANNOUNCE_SCHEMA_VERSION),
messageId: z.string().min(1),
@ -243,6 +264,7 @@ export const discoveryAnnounceSchema = z.object({
tcpPort: z.number().int().positive().max(65535),
wireContract: z.literal(INTEROP_WIRE_CONTRACT),
supportedDocumentTypes: z.array(z.string().min(1)),
transportProfiles: z.array(discoveryTransportProfileSchema).optional(),
supportsSchemaCatalog: z.boolean(),
});

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

@ -711,13 +711,18 @@ async function expectProbePeers(
expectedPeers: string[],
): Promise<void> {
const found = new Set<string>();
const profiles = new Map<string, unknown>();
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 }>) {
for (const peer of probe.peers as Array<{ runtimeId: string; transportProfiles?: unknown }>) {
found.add(peer.runtimeId);
profiles.set(peer.runtimeId, peer.transportProfiles);
}
if (expectedPeers.every((peer) => found.has(peer))) {
assert.deepEqual([...found].sort(), expectedPeers);
for (const peerId of expectedPeers) {
assertTcpFramedTransportProfile(peerId, profiles.get(peerId));
}
return;
}
}
@ -725,6 +730,17 @@ async function expectProbePeers(
assert.deepEqual([...found].sort(), expectedPeers);
}
function assertTcpFramedTransportProfile(peerId: string, value: unknown): void {
assert.ok(Array.isArray(value), `${peerId} did not advertise transport profiles`);
const firstProfile = value[0] as any;
assert.equal(firstProfile?.schemaVersion, "cultnet.transport_profile.v0");
assert.equal(firstProfile?.runtimeId, peerId);
const transport = firstProfile?.transports?.[0];
assert.equal(transport?.protocol, "tcp_framed");
assert.equal(transport?.channels?.[0]?.delivery, "reliable");
assert.equal(transport?.channels?.[0]?.ordering, "ordered");
}
async function requestCultMeshFromPython(port: number, request: Record<string, unknown>): Promise<any> {
const socket = connectTcp(port, "127.0.0.1");
const framer = new LengthPrefixedMessageFramer();

6
src/GameCult.Networking/Contracts/cultnet.hello.schema.json
View file

@ -30,6 +30,12 @@
"type": "array",
"items": { "type": "string", "minLength": 1 }
},
"transportProfiles": {
"type": "array",
"items": {
"$ref": "./cultnet.transport-profile.schema.json"
}
},
"supportsSchemaCatalog": { "type": "boolean" }
}
}

92
src/GameCult.Networking/CultNetSchemaMessages.cs
View file

@ -254,11 +254,103 @@ namespace GameCult.Networking
/// </summary>
[Key("supportedMessageVersions")] public string[]? SupportedMessageVersions { get; set; }
/// <summary>
/// Gets or sets the transport profiles advertised by this peer.
/// </summary>
[Key("transportProfiles")] public CultNetTransportProfile[]? TransportProfiles { get; set; }
/// <summary>
/// Gets or sets the supports schema catalog.
/// </summary>
[Key("supportsSchemaCatalog")] public bool SupportsSchemaCatalog { get; set; }
}
/// <summary>
/// Describes the transports a runtime can speak.
/// </summary>
[MessagePackObject]
public class CultNetTransportProfile
{
/// <summary>
/// Gets or sets the schema version.
/// </summary>
[Key("schemaVersion")] public string SchemaVersion { get; set; } = "cultnet.transport_profile.v0";
/// <summary>
/// Gets or sets the runtime id.
/// </summary>
[Key("runtimeId")] public string RuntimeId { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the transport descriptors.
/// </summary>
[Key("transports")] public CultNetTransportDescriptor[] Transports { get; set; } = Array.Empty<CultNetTransportDescriptor>();
}
/// <summary>
/// Describes one transport endpoint.
/// </summary>
[MessagePackObject]
public class CultNetTransportDescriptor
{
/// <summary>
/// Gets or sets the transport id.
/// </summary>
[Key("transportId")] public string TransportId { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the protocol name.
/// </summary>
[Key("protocol")] public string Protocol { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the host.
/// </summary>
[Key("host")] public string? Host { get; set; }
/// <summary>
/// Gets or sets the port.
/// </summary>
[Key("port")] public int? Port { get; set; }
/// <summary>
/// Gets or sets the path.
/// </summary>
[Key("path")] public string? Path { get; set; }
/// <summary>
/// Gets or sets the discovery group.
/// </summary>
[Key("discoveryGroup")] public string? DiscoveryGroup { get; set; }
/// <summary>
/// Gets or sets the wire contracts.
/// </summary>
[Key("wireContracts")] public string[]? WireContracts { get; set; }
/// <summary>
/// Gets or sets the channels.
/// </summary>
[Key("channels")] public CultNetTransportChannel[] Channels { get; set; } = Array.Empty<CultNetTransportChannel>();
}
/// <summary>
/// Describes one transport channel.
/// </summary>
[MessagePackObject]
public class CultNetTransportChannel
{
/// <summary>
/// Gets or sets the channel id.
/// </summary>
[Key("channelId")] public string ChannelId { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the delivery semantics.
/// </summary>
[Key("delivery")] public string Delivery { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the ordering semantics.
/// </summary>
[Key("ordering")] public string Ordering { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the maximum payload bytes.
/// </summary>
[Key("maxPayloadBytes")] public int? MaxPayloadBytes { get; set; }
/// <summary>
/// Gets or sets the maximum fragment bytes.
/// </summary>
[Key("maxFragmentBytes")] public int? MaxFragmentBytes { get; set; }
}
/// <summary>
/// CultNet login request message.
/// </summary>

57
tests/GameCult.Networking.InteropPeer/Program.cs
View file

@ -162,6 +162,7 @@ static async Task ProbeAsync(Dictionary<string, string> options)
tcpPort = peer.TcpPort,
wireContract = peer.WireContract,
supportedDocumentTypes = peer.SupportedDocumentTypes,
transportProfiles = peer.TransportProfiles,
supportsSchemaCatalog = peer.SupportsSchemaCatalog
})
.ToArray()
@ -241,6 +242,7 @@ static async Task DialAsync(DialConfig config)
displayName = remoteHello.DisplayName,
supportedDocumentTypes = remoteHello.SupportedDocumentTypes,
supportedMessageVersions = remoteHello.SupportedMessageVersions,
transportProfiles = remoteHello.TransportProfiles,
supportsSchemaCatalog = remoteHello.SupportsSchemaCatalog
},
hasInteropSchema,
@ -362,6 +364,33 @@ static async Task RunDiscoveryServerAsync(Socket socket, ServeConfig config, Can
TcpPort = config.TcpPort,
WireContract = CultNetWireContracts.SchemaV0,
SupportedDocumentTypes = [InteropPeerShared.InteropDocumentType],
TransportProfiles =
[
new CultNetTransportProfile
{
RuntimeId = config.RuntimeId,
Transports =
[
new CultNetTransportDescriptor
{
TransportId = "interop-tcp",
Protocol = "tcp_framed",
Host = config.AdvertiseHost,
Port = config.TcpPort,
WireContracts = [CultNetWireContracts.SchemaV0],
Channels =
[
new CultNetTransportChannel
{
ChannelId = "schema",
Delivery = "reliable",
Ordering = "ordered"
}
]
}
]
}
],
SupportsSchemaCatalog = true
};
var announceBytes = MessagePackSerializer.Serialize(announce, CultNetSchemaMessageSerialization.Options);
@ -433,6 +462,33 @@ static async Task RunTcpServerAsync(
SupportedDocumentTypes = [InteropPeerShared.InteropDocumentType],
SupportedMutationContracts = [InteropPeerShared.InteractionContract()],
SupportedMessageVersions = [InteropPeerShared.InteropSchemaVersion],
TransportProfiles =
[
new CultNetTransportProfile
{
RuntimeId = config.RuntimeId,
Transports =
[
new CultNetTransportDescriptor
{
TransportId = "interop-tcp",
Protocol = "tcp_framed",
Host = config.AdvertiseHost,
Port = config.TcpPort,
WireContracts = [CultNetWireContracts.SchemaV0],
Channels =
[
new CultNetTransportChannel
{
ChannelId = "schema",
Delivery = "reliable",
Ordering = "ordered"
}
]
}
]
}
],
SupportsSchemaCatalog = true
}, cancellationToken);
break;
@ -955,5 +1011,6 @@ public sealed class DiscoveryAnnounceMessage
[Key("tcpPort")] public int TcpPort { get; set; }
[Key("wireContract")] public string WireContract { get; set; } = CultNetWireContracts.SchemaV0;
[Key("supportedDocumentTypes")] public string[] SupportedDocumentTypes { get; set; } = Array.Empty<string>();
[Key("transportProfiles")] public CultNetTransportProfile[]? TransportProfiles { get; set; }
[Key("supportsSchemaCatalog")] public bool SupportsSchemaCatalog { get; set; }
}

42
tests/GameCult.Networking.Tests/NetworkingTests.cs
View file

@ -181,6 +181,48 @@ namespace GameCult.Networking.Tests
.With.InnerException.InstanceOf<MessagePack.MessagePackSerializationException>());
}
[Test]
public void CultNetSchemaMessageSerialization_RoundTrips_HelloTransportProfile()
{
var message = new CultNetHelloMessage
{
RuntimeId = "csharp-test",
RuntimeKind = "dotnet",
TransportProfiles =
[
new CultNetTransportProfile
{
RuntimeId = "csharp-test",
Transports =
[
new CultNetTransportDescriptor
{
TransportId = "test-pipe",
Protocol = "tcp_framed",
WireContracts = [CultNetWireContracts.SchemaV0],
Channels =
[
new CultNetTransportChannel
{
ChannelId = "schema",
Delivery = "reliable",
Ordering = "ordered"
}
]
}
]
}
]
};
var payload = CultNetSchemaMessageSerialization.Serialize(message);
var roundTrip = (CultNetHelloMessage)CultNetSchemaMessageSerialization.Deserialize(payload);
Assert.That(roundTrip.TransportProfiles, Is.Not.Null);
Assert.That(roundTrip.TransportProfiles![0].Transports[0].Protocol, Is.EqualTo("tcp_framed"));
Assert.That(roundTrip.TransportProfiles[0].Transports[0].Channels[0].Ordering, Is.EqualTo("ordered"));
}
[Test]
public void CultNetSchemaMessageSerialization_RoundTrips_RawSnapshotResponse()
{