diff --git a/packages/cultcache-py/src/cultnet_py/__init__.py b/packages/cultcache-py/src/cultnet_py/__init__.py index 831fcad9..7110a410 100644 --- a/packages/cultcache-py/src/cultnet_py/__init__.py +++ b/packages/cultcache-py/src/cultnet_py/__init__.py @@ -1,6 +1,7 @@ from .framing import decode_frame, encode_frame, read_frame, write_frame from .messages import ( CultNetMessage, + compute_simulation_claim_hash, database_subscribe, database_unsubscribe, document_delete, @@ -10,11 +11,13 @@ from .messages import ( schema_catalog_request, shard_catalog_request, shard_log_request, + simulation_observation, snapshot_request, ) __all__ = [ "CultNetMessage", + "compute_simulation_claim_hash", "database_subscribe", "database_unsubscribe", "decode_frame", @@ -27,6 +30,7 @@ __all__ = [ "schema_catalog_request", "shard_catalog_request", "shard_log_request", + "simulation_observation", "snapshot_request", "write_frame", ] diff --git a/packages/cultcache-py/src/cultnet_py/interop_peer.py b/packages/cultcache-py/src/cultnet_py/interop_peer.py index a04cdcc0..ca146eb7 100644 --- a/packages/cultcache-py/src/cultnet_py/interop_peer.py +++ b/packages/cultcache-py/src/cultnet_py/interop_peer.py @@ -1,6 +1,7 @@ from __future__ import annotations import argparse +from collections import defaultdict import json import selectors import signal @@ -81,6 +82,8 @@ class PeerState: shard_endpoint: str shard_log: list[dict[str, Any]] shard_log_lock: threading.Lock + observations: dict[tuple[str, int, str, str], dict[str, dict[str, Any]]] + observations_lock: threading.Lock @dataclass @@ -229,6 +232,8 @@ def handle_server_message(state: PeerState, message: dict[str, Any], subscriptio return [shard_catalog_response(state, message)] if schema_version == "cultnet.shard_log_request.v0": return [shard_log_response(state, message)] + if schema_version == "cultnet.simulation_observation.v0": + return simulation_candidate_responses(state, message) if schema_version == "cultnet.database_subscribe.v0": return handle_database_subscribe(state, message, subscriptions) if schema_version == "cultnet.database_unsubscribe.v0": @@ -391,6 +396,66 @@ def append_shard_log_put(state: PeerState, message: dict[str, Any], document: di }) +def simulation_candidate_responses(state: PeerState, message: dict[str, Any]) -> list[dict[str, Any]]: + observation = message.get("observation") + if not isinstance(observation, dict): + return [] + group = ( + str(observation.get("shardId") or ""), + int(observation.get("frame") or 0), + str(observation.get("subjectId") or ""), + str(observation.get("claimKind") or ""), + ) + witness_id = str(observation.get("witnessRuntimeId") or "") + if not all(group) or not witness_id: + return [] + with state.observations_lock: + witnesses = state.observations.setdefault(group, {}) + witnesses[witness_id] = observation + candidates = build_simulation_candidates(message.get("messageId", ""), list(witnesses.values())) + return candidates + + +def build_simulation_candidates(message_id: str, observations: list[dict[str, Any]]) -> list[dict[str, Any]]: + by_claim: dict[str, list[dict[str, Any]]] = defaultdict(list) + total_weight = sum(observation_weight(observation) for observation in observations) + for observation in observations: + by_claim[str(observation.get("claimHash") or "")].append(observation) + + candidates = [] + for claim_hash in sorted(by_claim): + claim_observations = by_claim[claim_hash] + sample = claim_observations[0] + support_weight = sum(observation_weight(observation) for observation in claim_observations) + witness_count = len({str(observation.get("witnessRuntimeId") or "") for observation in claim_observations}) + confidence = support_weight / total_weight if total_weight > 0 else 0.0 + candidates.append({ + "schemaVersion": "cultnet.simulation_consensus_candidate.v0", + "messageId": message_id, + "shardId": str(sample.get("shardId") or ""), + "shardEpoch": int(sample.get("shardEpoch") or 0), + "frame": int(sample.get("frame") or 0), + "subjectId": str(sample.get("subjectId") or ""), + "claimKind": str(sample.get("claimKind") or ""), + "claimHash": claim_hash, + "claimSummary": sample.get("claimSummary"), + "witnessCount": witness_count, + "supportWeight": support_weight, + "totalWeight": total_weight, + "hasQuorum": witness_count >= 1 and support_weight >= 1.0 and confidence >= 0.5, + "confidence": confidence, + }) + candidates.sort(key=lambda candidate: (-candidate["supportWeight"], candidate["claimHash"])) + return candidates + + +def observation_weight(observation: dict[str, Any]) -> float: + weight = observation.get("weight") + if isinstance(weight, (int, float)): + return float(weight) + return 1.0 + + def database_change_notifications(message: dict[str, Any], document: dict[str, Any], subscriptions: dict[str, DatabaseSubscription]) -> list[dict[str, Any]]: notifications = [] for subscription in subscriptions.values(): @@ -562,6 +627,8 @@ def build_state( shard_endpoint=shard_endpoint or f"cultnet://{runtime_id}", shard_log=[], shard_log_lock=threading.Lock(), + observations={}, + observations_lock=threading.Lock(), ) diff --git a/packages/cultcache-py/src/cultnet_py/messages.py b/packages/cultcache-py/src/cultnet_py/messages.py index 71227ee9..1ba64a49 100644 --- a/packages/cultcache-py/src/cultnet_py/messages.py +++ b/packages/cultcache-py/src/cultnet_py/messages.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hashlib from dataclasses import dataclass from typing import Any @@ -147,3 +148,41 @@ def shard_log_request( if limit is not None: body["limit"] = limit return CultNetMessage("cultnet.shard_log_request.v0", body) + + +def compute_simulation_claim_hash(*parts: str) -> str: + canonical = "\x1f".join(part or "" for part in parts) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +def simulation_observation( + *, + message_id: str, + witness_runtime_id: str, + shard_id: str, + shard_epoch: int, + frame: int, + subject_id: str, + claim_kind: str, + claim_hash: str, + claim_summary: str | None = None, + weight: float = 1.0, + observed_at: str = "", +) -> CultNetMessage: + observation: dict[str, Any] = { + "witnessRuntimeId": witness_runtime_id, + "shardId": shard_id, + "shardEpoch": shard_epoch, + "frame": frame, + "subjectId": subject_id, + "claimKind": claim_kind, + "claimHash": claim_hash, + "weight": weight, + "observedAt": observed_at, + } + if claim_summary is not None: + observation["claimSummary"] = claim_summary + return CultNetMessage( + "cultnet.simulation_observation.v0", + {"messageId": message_id, "observation": observation}, + ) diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 645e6731..b0d4ca73 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -14,7 +14,7 @@ from cultcache_py import ( define_document_type, ) from cultcache_py.interop import read_note, write_note -from cultnet_py import database_subscribe, database_unsubscribe, decode_frame, document_put_raw, encode_frame, hello, parse_message, shard_catalog_request, shard_log_request +from cultnet_py import compute_simulation_claim_hash, database_subscribe, database_unsubscribe, decode_frame, document_put_raw, encode_frame, hello, parse_message, shard_catalog_request, shard_log_request, simulation_observation from cultmesh_py import create_node from cultmesh_py import ( CultMeshPeerCard, @@ -232,6 +232,27 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(log["afterSequence"], 3) self.assertEqual(log["limit"], 2) + def test_cultnet_simulation_observation_helper_matches_schema_v0_shape(self) -> None: + claim_hash = compute_simulation_claim_hash("frame:42", "subject:player-1", "hit") + self.assertEqual(len(claim_hash), 64) + message = simulation_observation( + message_id="obs-1", + witness_runtime_id="python-test", + shard_id="interop", + shard_epoch=1, + frame=42, + subject_id="player-1", + claim_kind="hit", + claim_hash=claim_hash, + claim_summary="player-1 hit target-a", + observed_at="2026-06-13T00:00:02Z", + ).to_wire() + self.assertEqual(message["schemaVersion"], "cultnet.simulation_observation.v0") + self.assertEqual(message["messageId"], "obs-1") + self.assertEqual(message["observation"]["witnessRuntimeId"], "python-test") + self.assertEqual(message["observation"]["claimHash"], claim_hash) + self.assertEqual(message["observation"]["weight"], 1.0) + def test_cultmesh_node_uses_cultcache_store(self) -> None: with tempfile.TemporaryDirectory() as tmp: document = define_database_entry_type("mesh.note", [("body", 0)]) diff --git a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts index 3cd1d9f1..5f8be91d 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts @@ -1,5 +1,6 @@ import assert from "node:assert/strict"; import { execFile, spawn, type ChildProcessByStdio } from "node:child_process"; +import { createHash } from "node:crypto"; import { once } from "node:events"; import { existsSync, rmSync } from "node:fs"; import { connect as connectTcp, createServer } from "node:net"; @@ -341,6 +342,37 @@ test("CultNet TS/Rust/C#/Python peers discover each other and exchange raw state && entry.put?.document?.recordKey === "note:ts-python-log" )); + const claimHash = computeSimulationClaimHash("frame:42", "subject:python-target", "hit"); + const pythonCandidate = await requestPythonSimulationCandidate(pythonPort, { + schemaVersion: "cultnet.simulation_observation.v0", + messageId: "python-observation", + observation: { + witnessRuntimeId: "ts-witness", + shardId: "interop", + shardEpoch: 1, + frame: 42, + subjectId: "python-target", + claimKind: "hit", + claimHash, + claimSummary: "python-target was hit by the TypeScript witness", + weight: 1, + observedAt: "2026-06-13T00:00:02Z", + }, + }); + assert.equal(pythonCandidate.schemaVersion, "cultnet.simulation_consensus_candidate.v0"); + assert.equal(pythonCandidate.messageId, "python-observation"); + assert.equal(pythonCandidate.shardId, "interop"); + assert.equal(pythonCandidate.shardEpoch, 1); + assert.equal(pythonCandidate.frame, 42); + assert.equal(pythonCandidate.subjectId, "python-target"); + assert.equal(pythonCandidate.claimKind, "hit"); + assert.equal(pythonCandidate.claimHash, claimHash); + assert.equal(pythonCandidate.witnessCount, 1); + assert.equal(pythonCandidate.supportWeight, 1); + assert.equal(pythonCandidate.totalWeight, 1); + assert.equal(pythonCandidate.hasQuorum, true); + assert.equal(pythonCandidate.confidence, 1); + const expectedPeers = ["csharp-peer", "python-peer", "rust-peer", "ts-peer"]; await expectProbePeers("ts-probe", process.execPath, [ @@ -624,6 +656,39 @@ async function requestPythonShardCatalogAndLog( }); } +async function requestPythonSimulationCandidate(port: number, observation: Record): Promise { + const socket = connectTcp(port, "127.0.0.1"); + const framer = new LengthPrefixedMessageFramer(); + await once(socket, "connect"); + + return await new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + socket.destroy(); + reject(new Error("Timed out waiting for Python simulation consensus candidate.")); + }, 5000); + + socket.on("data", (chunk) => { + for (const frame of framer.push(chunk)) { + const decoded = decode(frame) as any; + if (decoded?.schemaVersion === "cultnet.simulation_consensus_candidate.v0") { + clearTimeout(timeout); + socket.end(); + resolve(decoded); + } + } + }); + socket.once("error", (error) => { + clearTimeout(timeout); + reject(error); + }); + socket.write(encodeFrame(encode(observation))); + }); +} + +function computeSimulationClaimHash(...parts: string[]): string { + return createHash("sha256").update(parts.join("\x1f"), "utf8").digest("hex"); +} + async function stopProcess(processState: RunningServeProcess): Promise { if (processState.child.killed || processState.child.exitCode !== null) { return;