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

Emit Python CultMesh node raw mutations

Browse source
This commit is contained in:
Meta Vens 2026-06-14 00:02:37 +01:00
commit 6b0caf2b21
4 changed files with 176 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -220,6 +220,8 @@ node = create_node("mesh.cc", runtime_id="python-runtime")
node.register_document(note_doc)
node.put(note_doc, "note:1", {"body": "hello"})
live_note = node.get_required(note_doc, "note:1")
put_message = node.put_raw_message(note_doc, "note:2", {"body": "wire me"}, shard_id="interop", shard_epoch=1)
delete_message = node.delete_raw_message(note_doc, "note:2", shard_id="interop", shard_epoch=1)
peers = CultMeshPeerCatalog()
response = peers.create_response(peer_exchange_request("pex-1", verse_id="local"))

61
packages/cultcache-py/src/cultmesh_py/node.py
View file

@ -4,9 +4,17 @@ from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from cultcache_py import CultCache, SingleFileMessagePackBackingStore
from cultcache_py import CultCache, CultCacheEnvelope, SingleFileMessagePackBackingStore
from cultcache_py.documents import DocumentDefinition
from cultnet_py import CultNetAppliedRecord, CultNetRawClient, apply_raw_snapshot, apply_shard_log_response
from cultnet_py import (
CultNetAppliedRecord,
CultNetMessage,
CultNetRawClient,
apply_raw_snapshot,
apply_shard_log_response,
document_delete,
document_put_raw,
)
@dataclass
@ -22,6 +30,36 @@ class CultMeshNode:
def put(self, document: DocumentDefinition[Any], key: str, value: Any) -> None:
self.cache.put(document, key, value)
def put_raw_message(
self,
document: DocumentDefinition[Any],
key: str,
value: Any,
*,
message_id: str = "",
shard_id: str | None = None,
shard_epoch: int | None = None,
) -> CultNetMessage:
catalog_entry = document.catalog_entry()
envelope = CultCacheEnvelope.create(
key=key,
type=document.type,
payload=document.encode_payload(value),
schema_id=catalog_entry.schema_id,
catalog_entry=catalog_entry,
)
self.cache.put_envelope(document, envelope)
return document_put_raw(
message_id=message_id,
key=key,
schema_id=envelope.schema_id or catalog_entry.schema_id,
stored_at=envelope.stored_at,
payload=envelope.payload,
source_runtime_id=self.runtime_id,
shard_id=shard_id,
shard_epoch=shard_epoch,
)
def get(self, document: DocumentDefinition[Any], key: str) -> Any:
return self.cache.get(document, key)
@ -31,6 +69,25 @@ class CultMeshNode:
def delete(self, document: DocumentDefinition[Any], key: str) -> None:
self.cache.delete(document, key)
def delete_raw_message(
self,
document: DocumentDefinition[Any],
key: str,
*,
message_id: str = "",
shard_id: str | None = None,
shard_epoch: int | None = None,
) -> CultNetMessage:
schema_id = document.catalog_entry().schema_id
self.delete(document, key)
return document_delete(
message_id=message_id,
schema_id=schema_id,
record_key=key,
shard_id=shard_id,
shard_epoch=shard_epoch,
)
def snapshot(self) -> dict[str, dict[str, Any]]:
return self.cache.snapshot()

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

@ -517,6 +517,51 @@ class CultCacheTests(unittest.TestCase):
self.assertEqual(log_changes[0].change_kind, "updated")
self.assertEqual(node.get_required(document, "item:node").value, 9)
def test_cultmesh_node_emits_raw_put_and_delete_messages_for_local_writes(self) -> None:
document = define_database_entry_type(
"mesh.emit_item",
[
("name", 0),
("category", 1),
("value", 2, 0),
],
cls=Item,
)
node = create_node(runtime_id="python-node")
node.register_document(document)
put = node.put_raw_message(
document,
"item:emit",
Item("wand", "gear", 4),
message_id="put-emit",
shard_id="mesh",
shard_epoch=1,
).to_wire()
self.assertEqual(node.get_required(document, "item:emit").name, "wand")
self.assertEqual(put["schemaVersion"], "cultnet.document_put_raw.v0")
self.assertEqual(put["messageId"], "put-emit")
self.assertEqual(put["document"]["schemaId"], document.catalog_entry().schema_id)
self.assertEqual(put["document"]["recordKey"], "item:emit")
self.assertEqual(put["document"]["sourceRuntimeId"], "python-node")
self.assertEqual(put["shardId"], "mesh")
self.assertEqual(put["shardEpoch"], 1)
delete = node.delete_raw_message(
document,
"item:emit",
message_id="delete-emit",
shard_id="mesh",
shard_epoch=1,
).to_wire()
self.assertIsNone(node.get(document, "item:emit"))
self.assertEqual(delete["schemaVersion"], "cultnet.document_delete.v0")
self.assertEqual(delete["messageId"], "delete-emit")
self.assertEqual(delete["schemaId"], document.catalog_entry().schema_id)
self.assertEqual(delete["recordKey"], "item:emit")
self.assertEqual(delete["shardId"], "mesh")
self.assertEqual(delete["shardEpoch"], 1)
def test_cultnet_document_delete_helper_matches_schema_v0_shape(self) -> None:
message = document_delete(
message_id="delete-1",

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

@ -306,6 +306,11 @@ test("CultNet TS/Rust/C#/Python peers discover each other and exchange raw state
assert.equal(pythonCultMeshNodeSync.synced[0].recordKey, "note:python-peer");
assert.equal(pythonCultMeshNodeSync.note.authorRuntimeId, "python-peer");
const pythonCultMeshNodeEmit = await runPythonCultMeshNodeEmit(pythonPort);
assert.equal(pythonCultMeshNodeEmit.synced[0].recordKey, "note:python-node-emit");
assert.equal(pythonCultMeshNodeEmit.note.authorRuntimeId, "python-node-emit");
assert.equal(pythonCultMeshNodeEmit.note.title, "Python node emitted raw put");
const pythonChange = await requestPythonDatabaseChange(pythonPort, {
schemaVersion: "cultnet.database_subscribe.v0",
messageId: "python-subscribe",
@ -802,6 +807,71 @@ async function runPythonCultMeshNodeSync(port: number): Promise<any> {
return JSON.parse(stdout);
}
async function runPythonCultMeshNodeEmit(port: number): Promise<any> {
const schemaPathJson = JSON.stringify(interopSchemaPath);
const script = [
"import json",
"import socket",
"import time",
"from pathlib import Path",
"from cultcache_py import define_database_entry_type",
"from cultmesh_py import create_node",
"from cultnet_py import CultNetRawClient, write_frame",
`schema_json = Path(${schemaPathJson}).read_text(encoding='utf-8')`,
"note_doc = define_database_entry_type(",
" 'cultnet.interop-note',",
" [('schemaVersion', 0), ('documentId', 1), ('authorRuntimeId', 2), ('title', 3), ('body', 4), ('tags', 5, [])],",
` schema_id='${interopNoteSchemaId}',`,
" schema_version='cultnet.interop_note.v0',",
" canonical_schema_json=schema_json,",
")",
"writer = create_node(runtime_id='python-node-emit')",
"writer.register_document(note_doc)",
"message = writer.put_raw_message(",
" note_doc,",
" 'note:python-node-emit',",
" {",
" 'schemaVersion': 'cultnet.interop_note.v0',",
" 'documentId': 'note:python-node-emit',",
" 'authorRuntimeId': 'python-node-emit',",
" 'title': 'Python node emitted raw put',",
" 'body': 'node-level Python emit reached the peer',",
" 'tags': ['python', 'node', 'emit'],",
" },",
" message_id='python-node-emit-put',",
" shard_id='interop',",
" shard_epoch=1,",
")",
`with socket.create_connection(('127.0.0.1', ${port}), timeout=4.0) as connection:`,
" stream = connection.makefile('rwb')",
" write_frame(stream, message.to_bytes())",
" stream.flush()",
"reader = create_node(runtime_id='python-node-emit-reader')",
"reader.register_document(note_doc)",
`client = CultNetRawClient('127.0.0.1', ${port}, timeout_seconds=4.0)`,
"changes = []",
"for _ in range(20):",
` changes = reader.sync_snapshot(client, schema_ids=['${interopNoteSchemaId}'], record_keys=['note:python-node-emit'])`,
" if changes:",
" break",
" time.sleep(0.1)",
"note = reader.get_required(note_doc, 'note:python-node-emit')",
"print(json.dumps({",
" 'synced': [{'schemaId': change.schema_id, 'recordKey': change.record_key} for change in changes],",
" 'note': note,",
"}))",
].join("\n");
const { stdout } = await execFileAsync(pythonCommand, ["-c", script], {
cwd: cultcachePyRoot,
env: {
...process.env,
PYTHONPATH: cultcachePySrc,
},
timeout: 8000,
});
return JSON.parse(stdout);
}
async function requestPythonDatabaseChange(port: number, subscribe: Record<string, unknown>, put: Record<string, unknown>): Promise<any> {
const socket = connectTcp(port, "127.0.0.1");
const framer = new LengthPrefixedMessageFramer();