From a2bc9aeab44cd7b861f1b481d2d0b05204f89a51 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 14 Jun 2026 21:20:52 +0100 Subject: [PATCH] Advertise CultNet transport profiles --- .../cultcache-py/src/cultmesh_py/daemon.py | 1 + .../cultcache-py/src/cultmesh_py/server.py | 22 +++++ .../src/cultnet_py/interop_peer.py | 22 +++++ .../cultcache-py/src/cultnet_py/messages.py | 2 + .../src/cultnet_py/schema_catalog.py | 6 ++ packages/cultcache-py/tests/test_cultcache.py | 28 +++++- .../contracts/cultnet.hello.schema.json | 6 ++ .../examples/cultnet_interop_peer.rs | 54 ++++++++++- packages/cultnet-rs/src/contracts.rs | 65 +++++++++++++ packages/cultnet-rs/tests/cultnet.rs | 27 ++++++ .../contracts/cultnet.hello.schema.json | 6 ++ packages/cultnet-ts/src/contracts.ts | 2 + packages/cultnet-ts/src/index.ts | 3 + packages/cultnet-ts/test/cultnet.test.ts | 15 +++ .../test/interop/cultnet-interop-peer.ts | 50 ++++++++++ .../test/interop/cultnet-interop-shared.ts | 22 +++++ .../test/interop/cultnet-interop.test.ts | 18 +++- .../Contracts/cultnet.hello.schema.json | 6 ++ .../CultNetSchemaMessages.cs | 92 +++++++++++++++++++ .../Program.cs | 57 ++++++++++++ .../NetworkingTests.cs | 42 +++++++++ 21 files changed, 539 insertions(+), 7 deletions(-) diff --git a/packages/cultcache-py/src/cultmesh_py/daemon.py b/packages/cultcache-py/src/cultmesh_py/daemon.py index d06df10d..087dc218 100644 --- a/packages/cultcache-py/src/cultmesh_py/daemon.py +++ b/packages/cultcache-py/src/cultmesh_py/daemon.py @@ -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, diff --git a/packages/cultcache-py/src/cultmesh_py/server.py b/packages/cultcache-py/src/cultmesh_py/server.py index 17c69713..dab652df 100644 --- a/packages/cultcache-py/src/cultmesh_py/server.py +++ b/packages/cultcache-py/src/cultmesh_py/server.py @@ -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, } diff --git a/packages/cultcache-py/src/cultnet_py/interop_peer.py b/packages/cultcache-py/src/cultnet_py/interop_peer.py index e9e5e9b2..36924558 100644 --- a/packages/cultcache-py/src/cultnet_py/interop_peer.py +++ b/packages/cultcache-py/src/cultnet_py/interop_peer.py @@ -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, } diff --git a/packages/cultcache-py/src/cultnet_py/messages.py b/packages/cultcache-py/src/cultnet_py/messages.py index 2735589b..4fa8d336 100644 --- a/packages/cultcache-py/src/cultnet_py/messages.py +++ b/packages/cultcache-py/src/cultnet_py/messages.py @@ -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 [], }, ) diff --git a/packages/cultcache-py/src/cultnet_py/schema_catalog.py b/packages/cultcache-py/src/cultnet_py/schema_catalog.py index f2e77476..35be3246 100644 --- a/packages/cultcache-py/src/cultnet_py/schema_catalog.py +++ b/packages/cultcache-py/src/cultnet_py/schema_catalog.py @@ -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": { diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 280dc76b..d170dca2 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -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") diff --git a/packages/cultnet-rs/contracts/cultnet.hello.schema.json b/packages/cultnet-rs/contracts/cultnet.hello.schema.json index bced7577..a557dc3f 100644 --- a/packages/cultnet-rs/contracts/cultnet.hello.schema.json +++ b/packages/cultnet-rs/contracts/cultnet.hello.schema.json @@ -30,6 +30,12 @@ "type": "array", "items": { "type": "string", "minLength": 1 } }, + "transportProfiles": { + "type": "array", + "items": { + "$ref": "./cultnet.transport-profile.schema.json" + } + }, "supportsSchemaCatalog": { "type": "boolean" } } } diff --git a/packages/cultnet-rs/examples/cultnet_interop_peer.rs b/packages/cultnet-rs/examples/cultnet_interop_peer.rs index 08a09909..5019ea2c 100644 --- a/packages/cultnet-rs/examples/cultnet_interop_peer.rs +++ b/packages/cultnet-rs/examples/cultnet_interop_peer.rs @@ -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, + #[serde(default, skip_serializing_if = "Option::is_none")] + transport_profiles: Option>, supports_schema_catalog: bool, }, } @@ -290,6 +298,7 @@ fn probe(options: &BTreeMap) -> Result<()> { tcp_port, wire_contract, supported_document_types, + transport_profiles, supports_schema_catalog, }) = rmp_serde::from_slice::(&buffer[..len]) { @@ -307,6 +316,7 @@ fn probe(options: &BTreeMap) -> 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) -> 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 { }] } +fn tcp_transport_profiles(runtime_id: &str, host: &str, port: u16) -> Vec { + 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()), diff --git a/packages/cultnet-rs/src/contracts.rs b/packages/cultnet-rs/src/contracts.rs index 24be7b48..ad5e8401 100644 --- a/packages/cultnet-rs/src/contracts.rs +++ b/packages/cultnet-rs/src/contracts.rs @@ -81,6 +81,68 @@ pub struct CultNetDocumentMutationContract { pub notes: Option>, } +#[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, +} + +#[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, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub port: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub path: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub discovery_group: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub wire_contracts: Option>, + pub channels: Vec, +} + +#[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, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_fragment_bytes: Option, +} + +#[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>, #[serde(default, skip_serializing_if = "Option::is_none")] + transport_profiles: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] supports_schema_catalog: Option, }, #[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")?; diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index d4d12edd..7bf34f64 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -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, } ); diff --git a/packages/cultnet-ts/contracts/cultnet.hello.schema.json b/packages/cultnet-ts/contracts/cultnet.hello.schema.json index bced7577..a557dc3f 100644 --- a/packages/cultnet-ts/contracts/cultnet.hello.schema.json +++ b/packages/cultnet-ts/contracts/cultnet.hello.schema.json @@ -30,6 +30,12 @@ "type": "array", "items": { "type": "string", "minLength": 1 } }, + "transportProfiles": { + "type": "array", + "items": { + "$ref": "./cultnet.transport-profile.schema.json" + } + }, "supportsSchemaCatalog": { "type": "boolean" } } } diff --git a/packages/cultnet-ts/src/contracts.ts b/packages/cultnet-ts/src/contracts.ts index aaabf3f5..eeaa8036 100644 --- a/packages/cultnet-ts/src/contracts.ts +++ b/packages/cultnet-ts/src/contracts.ts @@ -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": diff --git a/packages/cultnet-ts/src/index.ts b/packages/cultnet-ts/src/index.ts index 9fd02908..0b2b1700 100644 --- a/packages/cultnet-ts/src/index.ts +++ b/packages/cultnet-ts/src/index.ts @@ -26,6 +26,9 @@ export { type CultNetSnapshotRequestMessage, type CultNetSnapshotResponseMessage, type CultNetSnapshotResponseRawMessage, + type CultNetTransportChannel, + type CultNetTransportDescriptor, + type CultNetTransportProfile, type CultNetVerifyMessage, type CultNetWireContract, type GhostlightAgent, diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index e396250c..b478a17a 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -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(); diff --git a/packages/cultnet-ts/test/interop/cultnet-interop-peer.ts b/packages/cultnet-ts/test/interop/cultnet-interop-peer.ts index dd6fa8be..634cd4bf 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop-peer.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop-peer.ts @@ -143,6 +143,8 @@ async function serve(args: Map): Promise { 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, }; diff --git a/packages/cultnet-ts/test/interop/cultnet-interop-shared.ts b/packages/cultnet-ts/test/interop/cultnet-interop-shared.ts index 0f9359aa..18b524e2 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop-shared.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop-shared.ts @@ -232,6 +232,27 @@ export const discoveryProbeSchema = z.object({ export type DiscoveryProbe = z.infer; +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(), }); diff --git a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts index 739183c0..1b75ca90 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts @@ -711,13 +711,18 @@ async function expectProbePeers( expectedPeers: string[], ): Promise { const found = new Set(); + const profiles = new Map(); 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): Promise { const socket = connectTcp(port, "127.0.0.1"); const framer = new LengthPrefixedMessageFramer(); diff --git a/src/GameCult.Networking/Contracts/cultnet.hello.schema.json b/src/GameCult.Networking/Contracts/cultnet.hello.schema.json index bced7577..a557dc3f 100644 --- a/src/GameCult.Networking/Contracts/cultnet.hello.schema.json +++ b/src/GameCult.Networking/Contracts/cultnet.hello.schema.json @@ -30,6 +30,12 @@ "type": "array", "items": { "type": "string", "minLength": 1 } }, + "transportProfiles": { + "type": "array", + "items": { + "$ref": "./cultnet.transport-profile.schema.json" + } + }, "supportsSchemaCatalog": { "type": "boolean" } } } diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs index 18b6f9ef..9c8fcfbe 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -254,11 +254,103 @@ namespace GameCult.Networking /// [Key("supportedMessageVersions")] public string[]? SupportedMessageVersions { get; set; } /// + /// Gets or sets the transport profiles advertised by this peer. + /// + [Key("transportProfiles")] public CultNetTransportProfile[]? TransportProfiles { get; set; } + /// /// Gets or sets the supports schema catalog. /// [Key("supportsSchemaCatalog")] public bool SupportsSchemaCatalog { get; set; } } + /// + /// Describes the transports a runtime can speak. + /// + [MessagePackObject] + public class CultNetTransportProfile + { + /// + /// Gets or sets the schema version. + /// + [Key("schemaVersion")] public string SchemaVersion { get; set; } = "cultnet.transport_profile.v0"; + /// + /// Gets or sets the runtime id. + /// + [Key("runtimeId")] public string RuntimeId { get; set; } = string.Empty; + /// + /// Gets or sets the transport descriptors. + /// + [Key("transports")] public CultNetTransportDescriptor[] Transports { get; set; } = Array.Empty(); + } + + /// + /// Describes one transport endpoint. + /// + [MessagePackObject] + public class CultNetTransportDescriptor + { + /// + /// Gets or sets the transport id. + /// + [Key("transportId")] public string TransportId { get; set; } = string.Empty; + /// + /// Gets or sets the protocol name. + /// + [Key("protocol")] public string Protocol { get; set; } = string.Empty; + /// + /// Gets or sets the host. + /// + [Key("host")] public string? Host { get; set; } + /// + /// Gets or sets the port. + /// + [Key("port")] public int? Port { get; set; } + /// + /// Gets or sets the path. + /// + [Key("path")] public string? Path { get; set; } + /// + /// Gets or sets the discovery group. + /// + [Key("discoveryGroup")] public string? DiscoveryGroup { get; set; } + /// + /// Gets or sets the wire contracts. + /// + [Key("wireContracts")] public string[]? WireContracts { get; set; } + /// + /// Gets or sets the channels. + /// + [Key("channels")] public CultNetTransportChannel[] Channels { get; set; } = Array.Empty(); + } + + /// + /// Describes one transport channel. + /// + [MessagePackObject] + public class CultNetTransportChannel + { + /// + /// Gets or sets the channel id. + /// + [Key("channelId")] public string ChannelId { get; set; } = string.Empty; + /// + /// Gets or sets the delivery semantics. + /// + [Key("delivery")] public string Delivery { get; set; } = string.Empty; + /// + /// Gets or sets the ordering semantics. + /// + [Key("ordering")] public string Ordering { get; set; } = string.Empty; + /// + /// Gets or sets the maximum payload bytes. + /// + [Key("maxPayloadBytes")] public int? MaxPayloadBytes { get; set; } + /// + /// Gets or sets the maximum fragment bytes. + /// + [Key("maxFragmentBytes")] public int? MaxFragmentBytes { get; set; } + } + /// /// CultNet login request message. /// diff --git a/tests/GameCult.Networking.InteropPeer/Program.cs b/tests/GameCult.Networking.InteropPeer/Program.cs index 7b1e39df..ad9c05a9 100644 --- a/tests/GameCult.Networking.InteropPeer/Program.cs +++ b/tests/GameCult.Networking.InteropPeer/Program.cs @@ -162,6 +162,7 @@ static async Task ProbeAsync(Dictionary 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(); + [Key("transportProfiles")] public CultNetTransportProfile[]? TransportProfiles { get; set; } [Key("supportsSchemaCatalog")] public bool SupportsSchemaCatalog { get; set; } } diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 42b8b675..6520db76 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -181,6 +181,48 @@ namespace GameCult.Networking.Tests .With.InnerException.InstanceOf()); } + [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() {