Filter Python interop shard snapshots

packages/cultcache-py/PARITY.md

@@ -11,6 +11,8 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti.
 - CultNet schema-v0 live interop:
   `packages/cultnet-ts/test/interop/cultnet-interop.test.ts` runs a live
   TS/Rust/C#/Python peer ring.
+  The Python interop peer filters shard-scoped raw snapshots by its live
+  shard-log membership before stamping shard cursor metadata.
 - Python CultNet raw client:
   Fetches schema catalogs, raw snapshots, shard catalogs, shard logs, and live
   database subscription changes.

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

@@ -861,6 +861,10 @@ def catalog_response(state: PeerState, request: dict[str, Any]) -> dict[str, Any
 def raw_snapshot_response(state: PeerState, request: dict[str, Any]) -> dict[str, Any]:
     schema_ids = set(request.get("schemaIds") or [])
     record_keys = set(request.get("recordKeys") or [])
+    requested_shard_id = request.get("shardId")
+    shard_record_keys: set[tuple[str, str]] | None = None
+    if requested_shard_id == state.shard_id:
+        shard_record_keys = live_shard_record_keys(state)
     documents = []
     for envelope in state.cache.snapshot_envelopes():
         binding = state.bindings_by_document_type.get(envelope.type)
@@ -871,9 +875,11 @@ def raw_snapshot_response(state: PeerState, request: dict[str, Any]) -> dict[str
             continue
         if record_keys and envelope.key not in record_keys:
             continue
+        if shard_record_keys is not None and (schema_id, envelope.key) not in shard_record_keys:
+            continue
         documents.append(raw_record_from_envelope(envelope, schema_id))
     response: dict[str, Any] = {"schemaVersion": "cultnet.snapshot_response_raw.v0", "messageId": request.get("messageId", ""), "documents": documents}
-    if request.get("shardId") == state.shard_id:
+    if requested_shard_id == state.shard_id:
         with state.shard_log_lock:
             response["shardId"] = state.shard_id
             response["shardEpoch"] = state.shard_epoch
@@ -881,6 +887,26 @@ def raw_snapshot_response(state: PeerState, request: dict[str, Any]) -> dict[str
     return response
 
 
+def live_shard_record_keys(state: PeerState) -> set[tuple[str, str]]:
+    records: set[tuple[str, str]] = set()
+    with state.shard_log_lock:
+        for entry in state.shard_log:
+            put = entry.get("put")
+            if isinstance(put, dict) and isinstance(put.get("document"), dict):
+                document = put["document"]
+                schema_id = document.get("schemaId")
+                record_key = document.get("recordKey")
+                if schema_id and record_key:
+                    records.add((str(schema_id), str(record_key)))
+            delete = entry.get("delete")
+            if isinstance(delete, dict):
+                schema_id = delete.get("schemaId")
+                record_key = delete.get("recordKey")
+                if schema_id and record_key:
+                    records.discard((str(schema_id), str(record_key)))
+    return records
+
+
 def raw_document_put(binding: Binding, message_id: str, key: str, value: dict[str, Any], state: PeerState) -> dict[str, Any]:
     payload = binding.document.encode_payload(value)
     return {

packages/cultcache-py/tests/test_cultcache.py

@@ -8,6 +8,7 @@ import unittest
 from dataclasses import asdict, dataclass
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
+from uuid import uuid4
 
 from cultcache_py import (
     CultCache,
@@ -20,6 +21,7 @@ from cultcache_py.benchmark import run_benchmark
 from cultcache_py.compare_csharp import _median_result
 from cultcache_py.interop import read_note, write_note
 from cultcache_py.verify import verify
+from cultnet_py.interop_peer import append_shard_log_put, build_state, raw_snapshot_response
 from cultnet_py import (
     compute_simulation_claim_hash,
     CultNetDatabaseChange,
@@ -1312,6 +1314,59 @@ class CultCacheTests(unittest.TestCase):
         self.assertEqual(response.shard_epoch, 2)
         self.assertEqual([record.record_key for record in response.documents], ["note:notes"])
 
+    def test_python_interop_peer_filters_shard_snapshot_by_logged_membership(self) -> None:
+        runtime_id = f"python-interop-test-{uuid4().hex}"
+        state = build_state(
+            runtime_id=runtime_id,
+            runtime_kind="python",
+            display_name="Python Interop Test",
+            agent_id="python-interop-test-agent",
+            schema_path=str(Path("packages/cultnet-ts/integration/contracts/cultnet.interop-note.schema.json")),
+        )
+        binding = state.bindings["note"]
+        logged_value = {
+            "schemaVersion": "cultnet.interop_note.v0",
+            "documentId": "note:logged",
+            "authorRuntimeId": runtime_id,
+            "title": "Logged",
+            "body": "This record belongs to the shard log.",
+            "tags": ["interop", "logged"],
+        }
+        unlogged_value = {
+            **logged_value,
+            "documentId": "note:unlogged",
+            "title": "Unlogged",
+            "body": "This record is in cache but not in the shard log.",
+        }
+        state.cache.put(binding.document, "note:logged", logged_value)
+        state.cache.put(binding.document, "note:unlogged", unlogged_value)
+        logged_record = {
+            "schemaId": state.note_schema_id,
+            "recordKey": "note:logged",
+            "storedAt": "2026-06-14T00:00:00Z",
+            "payloadEncoding": "messagepack",
+            "payload": binding.document.encode_payload(logged_value),
+        }
+        append_shard_log_put(
+            state,
+            {"messageId": "logged-put"},
+            logged_record,
+        )
+
+        response = raw_snapshot_response(
+            state,
+            {
+                "schemaVersion": "cultnet.snapshot_request.v0",
+                "messageId": "interop-shard-snapshot",
+                "schemaIds": [state.note_schema_id],
+                "shardId": state.shard_id,
+            },
+        )
+
+        self.assertEqual(response["shardId"], state.shard_id)
+        self.assertEqual(response["shardLogSequence"], 1)
+        self.assertEqual([record["recordKey"] for record in response["documents"]], ["note:logged"])
+
     def test_cultmesh_database_creates_shard_log_response_from_raw_mutations(self) -> None:
         document = define_database_entry_type(
             "mesh.log_note",