diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index 1c3c35c3..915d8067 100644 --- a/packages/cultcache-py/PARITY.md +++ b/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. diff --git a/packages/cultcache-py/src/cultnet_py/interop_peer.py b/packages/cultcache-py/src/cultnet_py/interop_peer.py index 0012d81b..e9e5e9b2 100644 --- a/packages/cultcache-py/src/cultnet_py/interop_peer.py +++ b/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 { diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index aea4bef3..5bc49a11 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/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",