Add Python simulation consensus interop

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",
 ]

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(),
     )
 
 

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},
+    )

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)])

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<string, unknown>): Promise<any> {
+  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<void> {
   if (processState.child.killed || processState.child.exitCode !== null) {
     return;