Forward Python shard writes over CultNet

packages/cultcache-py/PARITY.md

@@ -52,7 +52,8 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti.
   unknown shards or stale epoch cursors; `create_shard_log_response(...)`
   remains the raw wire-dict compatibility projection.
 - Python CultNet shard replication:
-  Provides schema-v0 shard-log and shard-snapshot fetchers plus a pull-once
+  Provides schema-v0 shard-log and shard-snapshot fetchers, a schema-v0
+  write-forwarder for raw put/delete messages, plus a pull-once
   `CultNetShardReplicator` that tracks replica cursors, applies log responses
   through the existing database owner, and recovers from compacted-history
   resync pressure by fetching a shard snapshot. Replica cursor storage can be
@@ -180,6 +181,6 @@ node --test packages\cultnet-ts\dist-test\test\interop\cultnet-interop.test.js
 - Python does not claim the full C# shard mutation-log service surface. It can
   create/apply raw snapshots, create/apply package-local raw shard-log
   responses, consume remote shard-log responses, and run pull-once replica
-  catch-up with snapshot recovery and a file-backed replica cursor store. It
-  does not yet claim the full C# background replicator, write-forwarder, or
-  file-backed authoritative mutation-log store.
+  catch-up with snapshot recovery, a file-backed replica cursor store, and
+  schema-v0 write forwarding. It does not yet claim the full C# background
+  replicator or file-backed authoritative mutation-log store.

packages/cultcache-py/src/cultcache_py/verify.py

@@ -44,9 +44,11 @@ EXPECTED_EXPORTS: dict[str, tuple[str, ...]] = {
         "CultNetShardLogResponse",
         "CultNetFileShardReplicaCursorStore",
         "CultNetInMemoryShardReplicaCursorStore",
+        "CultNetSchemaWriteForwarder",
         "CultNetSchemaShardLogFetcher",
         "CultNetSchemaShardSnapshotFetcher",
         "CultNetShardLogFetcher",
+        "CultNetShardWriteForwarder",
         "CultNetShardReplicaCursor",
         "CultNetShardReplicaCursorStore",
         "CultNetShardReplicator",

packages/cultcache-py/src/cultnet_py/__init__.py

@@ -40,9 +40,11 @@ from .shard_log import CultNetShardLogEntry, CultNetShardLogResponse
 from .shard_replication import (
     CultNetFileShardReplicaCursorStore,
     CultNetInMemoryShardReplicaCursorStore,
+    CultNetSchemaWriteForwarder,
     CultNetSchemaShardLogFetcher,
     CultNetSchemaShardSnapshotFetcher,
     CultNetShardLogFetcher,
+    CultNetShardWriteForwarder,
     CultNetShardReplicaCursor,
     CultNetShardReplicaCursorStore,
     CultNetShardReplicator,
@@ -73,9 +75,11 @@ __all__ = [
     "CultNetShardLogResponse",
     "CultNetFileShardReplicaCursorStore",
     "CultNetInMemoryShardReplicaCursorStore",
+    "CultNetSchemaWriteForwarder",
     "CultNetSchemaShardLogFetcher",
     "CultNetSchemaShardSnapshotFetcher",
     "CultNetShardLogFetcher",
+    "CultNetShardWriteForwarder",
     "CultNetShardReplicaCursor",
     "CultNetShardReplicaCursorStore",
     "CultNetShardReplicator",

packages/cultcache-py/src/cultnet_py/client.py

@@ -29,6 +29,14 @@ class CultNetRawClient:
     port: int
     timeout_seconds: float = 4.0
 
+    def send(self, message: CultNetMessage | dict[str, Any]) -> None:
+        wire = message.to_wire() if isinstance(message, CultNetMessage) else message
+        with socket.create_connection((self.host, self.port), timeout=self.timeout_seconds) as connection:
+            connection.settimeout(self.timeout_seconds)
+            stream = connection.makefile("rwb")
+            write_frame(stream, msgpack.packb(wire, use_bin_type=True))
+            stream.flush()
+
     def request(self, message: CultNetMessage | dict[str, Any], *, expected_schema_version: str) -> dict[str, Any]:
         wire = message.to_wire() if isinstance(message, CultNetMessage) else message
         with socket.create_connection((self.host, self.port), timeout=self.timeout_seconds) as connection:

packages/cultcache-py/src/cultnet_py/shard_replication.py

@@ -14,6 +14,14 @@ from .shard_log import CultNetShardLogResponse
 from .snapshot import CultNetRawSnapshotResponse
 
 
+class CultNetShardWriteForwarder(Protocol):
+    def forward_put(self, shard: CultNetShardDescriptor, message: dict[str, Any]) -> None:
+        ...
+
+    def forward_delete(self, shard: CultNetShardDescriptor, message: dict[str, Any]) -> None:
+        ...
+
+
 class CultNetShardLogFetcher(Protocol):
     def fetch(
         self,
@@ -114,6 +122,32 @@ class CultNetFileShardReplicaCursorStore:
         )
 
 
+@dataclass(frozen=True)
+class CultNetSchemaWriteForwarder:
+    timeout_seconds: float = 4.0
+
+    def forward_put(self, shard: CultNetShardDescriptor, message: dict[str, Any]) -> None:
+        if message.get("schemaVersion") != "cultnet.document_put_raw.v0":
+            raise ValueError("forward_put requires a cultnet.document_put_raw.v0 message")
+        wire = dict(message)
+        wire.setdefault("shardId", shard.shard_id)
+        wire.setdefault("shardEpoch", shard.epoch)
+        self._send(shard, wire)
+
+    def forward_delete(self, shard: CultNetShardDescriptor, message: dict[str, Any]) -> None:
+        if message.get("schemaVersion") != "cultnet.document_delete.v0":
+            raise ValueError("forward_delete requires a cultnet.document_delete.v0 message")
+        wire = dict(message)
+        wire.setdefault("shardId", shard.shard_id)
+        wire.setdefault("shardEpoch", shard.epoch)
+        self._send(shard, wire)
+
+    def _send(self, shard: CultNetShardDescriptor, message: dict[str, Any]) -> None:
+        endpoint = _resolve_primary_endpoint(shard)
+        host, port = _parse_endpoint(endpoint)
+        CultNetRawClient(host, port, self.timeout_seconds).send(message)
+
+
 @dataclass(frozen=True)
 class CultNetSchemaShardLogFetcher:
     timeout_seconds: float = 4.0

packages/cultcache-py/tests/test_cultcache.py

@@ -4,10 +4,12 @@ import json
 import socket
 import tempfile
 import threading
+import time
 import unittest
 from dataclasses import asdict, dataclass
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
+from typing import Callable
 from uuid import uuid4
 
 from cultcache_py import (
@@ -37,10 +39,12 @@ from cultnet_py import (
     CultNetShardLogResponse,
     CultNetFileShardReplicaCursorStore,
     CultNetInMemoryShardReplicaCursorStore,
+    CultNetSchemaWriteForwarder,
     CultNetSchemaShardLogFetcher,
     CultNetSchemaShardSnapshotFetcher,
     CultNetShardReplicator,
     CultNetShardReplicatorOptions,
+    CultNetShardWriteForwarder,
     CultNetSimulationConsensusOptions,
     CultNetSimulationConsensusCandidate,
     CultNetSimulationObservation,
@@ -1556,6 +1560,60 @@ class CultCacheTests(unittest.TestCase):
         self.assertEqual(resync_sequence, 2)
         self.assertEqual(resync_target.get_required(document, "note:1")["body"], "one")
 
+    def test_cultnet_schema_write_forwarder_sends_raw_writes_to_primary(self) -> None:
+        document = define_database_entry_type(
+            "mesh.forward_note",
+            [("body", 0)],
+            schema_id="mesh.forward_note.v1",
+        )
+        primary = CultMesh.create_node(runtime_id="forward-primary")
+        primary.database.register_document(document)
+        server = CultMesh.serve_node(primary)
+        try:
+            shard = CultNetShardDescriptor(
+                shard_id="forward",
+                owner_runtime_id="forward-primary",
+                epoch=9,
+                is_primary=False,
+                schema_ids=("mesh.forward_note.v1",),
+                primary_endpoints=(f"cultnet://127.0.0.1:{server.port}",),
+            )
+            forwarder: CultNetShardWriteForwarder = CultNetSchemaWriteForwarder(timeout_seconds=2.0)
+            put = document_put_raw(
+                message_id="forward-put",
+                key="note:forward",
+                schema_id=document.catalog_entry().schema_id,
+                stored_at="2026-06-14T00:00:00Z",
+                payload=document.encode_payload({"body": "forwarded"}),
+            ).to_wire()
+            delete = document_delete(
+                message_id="forward-delete",
+                schema_id=document.catalog_entry().schema_id,
+                record_key="note:forward",
+            ).to_wire()
+
+            forwarder.forward_put(shard, put)
+            self._eventually(lambda: primary.database.get(document, "note:forward") is not None)
+            self.assertEqual(primary.database.get_required(document, "note:forward")["body"], "forwarded")
+            forwarder.forward_delete(shard, delete)
+            self._eventually(lambda: primary.database.get(document, "note:forward") is None)
+
+            log = primary.database.build_shard_log_response(shard_id="forward", shard_epoch=9)
+            self.assertIsNone(primary.database.get(document, "note:forward"))
+            self.assertEqual([entry.change_kind for entry in log.entries], ["added", "removed"])
+            self.assertEqual(log.entries[0].put["shardId"], "forward")
+            self.assertEqual(log.entries[1].delete["shardEpoch"], 9)
+        finally:
+            server.stop()
+
+    def _eventually(self, predicate: Callable[[], bool], *, timeout_seconds: float = 2.0) -> None:
+        deadline = time.monotonic() + timeout_seconds
+        while time.monotonic() < deadline:
+            if predicate():
+                return
+            time.sleep(0.01)
+        self.assertTrue(predicate())
+
     def test_cultmesh_database_watchers_observe_name_and_index_changes(self) -> None:
         document = define_database_entry_type(
             "mesh.named_watch",