From 3731f17db767c5fefbdbc930679bbc8dd91ff287 Mon Sep 17 00:00:00 2001
From: Meta Vens <rwvens@hotmail.com>
Date: Sun, 14 Jun 2026 16:31:32 +0100
Subject: [PATCH] Verify Python daemon write forwarding

---
 packages/cultcache-py/PARITY.md               |  5 +-
 packages/cultcache-py/tests/test_cultcache.py | 92 +++++++++++++++++++
 2 files changed, 96 insertions(+), 1 deletion(-)

diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md
index 510f49fb..1e1b8837 100644
--- a/packages/cultcache-py/PARITY.md
+++ b/packages/cultcache-py/PARITY.md
@@ -65,7 +65,10 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti.
   non-primary shards with advertised primaries, reports loop errors through a
   caller-owned callback, and recovers from compacted-history resync pressure by
   fetching a shard snapshot. Replica cursor storage can be in-memory or
-  restart-safe through a local MessagePack cursor file.
+  restart-safe through a local MessagePack cursor file. The write-forwarder is
+  also exercised against a launched Python daemon endpoint: forwarded raw
+  put/delete messages mutate daemon state and are recorded in the daemon-served
+  shard log.
 - Python CultNet database subscription changes:
   Provides typed `CultNetDatabaseChange` parsing plus
   `CultNetDatabaseSubscription.read_next_change(...)` / `iter_changes(...)` for
diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py
index 4761e733..4da34ef4 100644
--- a/packages/cultcache-py/tests/test_cultcache.py
+++ b/packages/cultcache-py/tests/test_cultcache.py
@@ -3431,6 +3431,98 @@ class CultCacheTests(unittest.TestCase):
             finally:
                 self._terminate_process(process)
 
+    def test_cultmesh_daemon_accepts_schema_write_forwarder_mutations(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            ready_path = Path(temp) / "ready.json"
+            package_src = Path(__file__).resolve().parents[1] / "src"
+            env = dict(os.environ)
+            existing_pythonpath = env.get("PYTHONPATH")
+            env["PYTHONPATH"] = (
+                str(package_src)
+                if not existing_pythonpath
+                else f"{package_src}{os.pathsep}{existing_pythonpath}"
+            )
+            process = subprocess.Popen(
+                [
+                    sys.executable,
+                    "-m",
+                    "cultmesh_py.daemon",
+                    "--runtime-id",
+                    "forwarder-daemon-peer",
+                    "--port",
+                    "0",
+                    "--register-interop-note",
+                    "--ready-file",
+                    str(ready_path),
+                ],
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                env=env,
+            )
+            try:
+                ready = self._wait_for_ready_file(process, ready_path)
+                endpoint = f"cultnet://127.0.0.1:{ready['port']}"
+                shard = CultNetShardDescriptor(
+                    shard_id="forwarded-interop",
+                    owner_runtime_id="forwarder-daemon-peer",
+                    epoch=3,
+                    is_primary=False,
+                    schema_ids=("cultcache.interop-note",),
+                    primary_endpoints=(endpoint,),
+                )
+                forwarder: CultNetShardWriteForwarder = CultNetSchemaWriteForwarder(timeout_seconds=2.0)
+                note = {
+                    "schemaVersion": INTEROP_SCHEMA_VERSION,
+                    "documentId": "note:forwarded-daemon",
+                    "authorRuntimeId": "forwarding-client",
+                    "title": "forwarded daemon note",
+                    "body": "forwarded through the schema write-forwarder",
+                    "tags": ["daemon", "forwarded"],
+                }
+
+                forwarder.forward_put(shard, document_put_raw(
+                    message_id="forwarded-daemon-put",
+                    key="note:forwarded-daemon",
+                    schema_id="cultcache.interop-note",
+                    stored_at="2026-06-14T00:00:00Z",
+                    payload=interop_note_document.encode_payload(note),
+                    source_runtime_id="forwarding-client",
+                ).to_wire())
+
+                client = CultMesh.create_client("127.0.0.1", int(ready["port"]), timeout_seconds=2.0)
+                self._eventually(lambda: bool(client.fetch_snapshot_response(
+                    schema_ids=["cultcache.interop-note"],
+                    record_keys=["note:forwarded-daemon"],
+                ).documents))
+                snapshot = client.fetch_snapshot_response(
+                    schema_ids=["cultcache.interop-note"],
+                    record_keys=["note:forwarded-daemon"],
+                )
+                self.assertEqual(
+                    interop_note_document.decode_payload(snapshot.documents[0].payload),
+                    note,
+                )
+
+                forwarder.forward_delete(shard, document_delete(
+                    message_id="forwarded-daemon-delete",
+                    schema_id="cultcache.interop-note",
+                    record_key="note:forwarded-daemon",
+                ).to_wire())
+                self._eventually(lambda: not client.fetch_snapshot_response(
+                    schema_ids=["cultcache.interop-note"],
+                    record_keys=["note:forwarded-daemon"],
+                ).documents)
+
+                shard_log = client.fetch_shard_log_response(shard_id="forwarded-interop", shard_epoch=3)
+                self.assertEqual([entry.change_kind for entry in shard_log.entries], ["added", "removed"])
+                self.assertEqual(shard_log.entries[0].put["shardId"], "forwarded-interop")
+                self.assertEqual(shard_log.entries[0].put["shardEpoch"], 3)
+                self.assertEqual(shard_log.entries[1].delete["shardId"], "forwarded-interop")
+                self.assertEqual(shard_log.entries[1].delete["shardEpoch"], 3)
+            finally:
+                self._terminate_process(process)
+
     def _wait_for_ready_file(self, process: subprocess.Popen[str], ready_path: Path) -> dict[str, object]:
         deadline = time.monotonic() + 5.0
         while not ready_path.exists() and time.monotonic() < deadline: