fix cluster stubs: deterministic fallback, payload forwarding, and handler guards

- Replace non-deterministic map range fallback with sorted node selection in PlaceActor and RebalanceShards
- Add Body field to MessagePayload to preserve message data during cross-node routing
- Forward actual message body in route_message handler instead of discarding it
- Add self-message guard to route_message handler to prevent loops
- Add nil guard for shardMap in handleRebalanceRequest
- Add self-message guard to handleRebalanceRequest in DistributedVM

.gitea/workflows/ci.yaml

@@ -17,37 +17,3 @@ jobs:
         run: go build ./...
       - name: Test
         run: go test ./...
-
-  integration:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v4
-      - uses: actions/setup-go@v5
-        with:
-          go-version: '1.23'
-      - name: Install and Start NATS Server
-        run: |
-          # Detect architecture and download appropriate binary
-          ARCH=$(uname -m)
-          if [ "$ARCH" = "x86_64" ]; then
-            NATS_ARCH="amd64"
-          elif [ "$ARCH" = "aarch64" ] || [ "$ARCH" = "arm64" ]; then
-            NATS_ARCH="arm64"
-          else
-            echo "Unsupported architecture: $ARCH"
-            exit 1
-          fi
-          echo "Detected architecture: $ARCH, using NATS binary: $NATS_ARCH"
-
-          # Download and extract nats-server
-          curl -L "https://github.com/nats-io/nats-server/releases/download/v2.10.24/nats-server-v2.10.24-linux-${NATS_ARCH}.tar.gz" -o nats-server.tar.gz
-          tar -xzf nats-server.tar.gz
-
-          # Start NATS with JetStream
-          ./nats-server-v2.10.24-linux-${NATS_ARCH}/nats-server -js -p 4222 &
-
-          # Wait for NATS to be ready
-          sleep 3
-          ./nats-server-v2.10.24-linux-${NATS_ARCH}/nats-server --version
-      - name: Run Integration Tests
-        run: go test -tags=integration -v ./...

README.md

@@ -107,7 +107,34 @@ Order state after replaying 2 events:
 
 ### Events are immutable
 
-Events represent facts about what happened. Once saved, they are never modified - you only append new events.
+Events represent facts about what happened. Once saved, they are never modified or deleted - you only append new events. This immutability guarantee is enforced at multiple levels:
+
+**Interface Design**: The `EventStore` interface provides no Update or Delete methods. Only `SaveEvent` (append), `GetEvents` (read), and `GetLatestVersion` (read) are available.
+
+**JetStream Storage**: When using `JetStreamEventStore`, events are stored in a NATS JetStream stream configured with:
+- File-based storage (durable)
+- Limits-based retention policy (events expire after configured duration, not before)
+- No mechanism to modify or delete individual events during their lifetime
+
+**Audit Trail Guarantee**: Because events are immutable once persisted, they serve as a trustworthy audit trail. You can rely on the fact that historical events won't change, enabling compliance and forensics.
+
+To correct a mistake, append a new event that expresses the correction rather than modifying history:
+
+```go
+// Wrong: Cannot update an event
+// store.UpdateEvent(eventID, newData)  // This method doesn't exist
+
+// Right: Append a new event that corrects the record
+correctionEvent := &aether.Event{
+    ID:        uuid.New().String(),
+    EventType: "OrderCorrected",
+    ActorID:   orderID,
+    Version:   currentVersion + 1,
+    Data:      map[string]interface{}{"reason": "price adjustment"},
+    Timestamp: time.Now(),
+}
+err := store.SaveEvent(correctionEvent)
+```
 
 ### State is derived
 

cluster/distributed.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"time"
 
 	"github.com/nats-io/nats.go"
 )
@@ -137,11 +138,43 @@ func (dvm *DistributedVM) LoadModel(model RuntimeModel) error {
 
 // SendMessage routes messages across the distributed cluster
 func (dvm *DistributedVM) SendMessage(message RuntimeMessage) error {
-	// This is a simplified implementation
-	// In practice, this would determine the target node based on sharding
-	// and route the message appropriately
+	actorID := message.GetTargetActorID()
+	targetNode := dvm.GetActorNode(actorID)
 
+	if targetNode == dvm.nodeID {
 		return dvm.localRuntime.SendMessage(message)
+	}
+
+	return dvm.routeMessageToNode(actorID, message)
+}
+
+// routeMessageToNode sends a message to another node for delivery to the target actor
+func (dvm *DistributedVM) routeMessageToNode(actorID string, message RuntimeMessage) error {
+	hops := 0
+	var body map[string]interface{}
+	if mp, ok := message.(*MessagePayload); ok {
+		hops = mp.Hops
+		body = mp.Body
+	}
+	if hops >= MaxRouteHops {
+		dvm.cluster.logger.Printf("Dropping message for actor %s: exceeded max hops (%d)", actorID, MaxRouteHops)
+		return fmt.Errorf("message exceeded max hops")
+	}
+
+	msg := ClusterMessage{
+		Type: "route_message",
+		From: dvm.nodeID,
+		To:   actorID,
+		Payload: MessagePayload{
+			TargetActorID: actorID,
+			Type:          message.GetType(),
+			Hops:          hops + 1,
+			Body:          body,
+		},
+		Timestamp: time.Now(),
+	}
+
+	return dvm.publishClusterMessage(msg)
 }
 
 // GetActorNode determines which node should handle a specific actor
@@ -189,8 +222,10 @@ func (dvm *DistributedVM) handleClusterMessage(msg *nats.Msg) {
 		dvm.localRuntime.LoadModel(&model)
 
 	case "route_message":
-		// Handle message routing from other nodes
-		// Re-marshal and unmarshal to convert map[string]interface{} to concrete type
+		if clusterMsg.From == dvm.nodeID {
+			return
+		}
+
 		payloadBytes, err := json.Marshal(clusterMsg.Payload)
 		if err != nil {
 			return
@@ -199,7 +234,24 @@ func (dvm *DistributedVM) handleClusterMessage(msg *nats.Msg) {
 		if err := json.Unmarshal(payloadBytes, &message); err != nil {
 			return
 		}
-		dvm.localRuntime.SendMessage(&message)
+
+		if message.Hops >= MaxRouteHops {
+			dvm.cluster.logger.Printf("Dropping message for actor %s: exceeded max hops (%d)", message.TargetActorID, MaxRouteHops)
+			return
+		}
+
+		targetActor := message.TargetActorID
+		msg := &MessagePayload{
+			TargetActorID: targetActor,
+			Type:          message.Type,
+			Hops:          message.Hops,
+			Body:          message.Body,
+		}
+		if dvm.IsLocalActor(targetActor) {
+			dvm.localRuntime.SendMessage(msg)
+		} else {
+			dvm.routeMessageToNode(targetActor, msg)
+		}
 
 	case "rebalance":
 		// Handle shard rebalancing requests
@@ -209,8 +261,45 @@ func (dvm *DistributedVM) handleClusterMessage(msg *nats.Msg) {
 
 // handleRebalanceRequest processes shard rebalancing requests
 func (dvm *DistributedVM) handleRebalanceRequest(msg ClusterMessage) {
-	// Simplified rebalancing logic
-	// In practice, this would implement complex actor migration
+	if msg.From == dvm.nodeID {
+		return
+	}
+
+	if !dvm.cluster.IsLeader() {
+		dvm.cluster.logger.Printf("Ignoring rebalance request: not the leader")
+		return
+	}
+
+	if dvm.cluster.shardMap == nil {
+		dvm.cluster.logger.Printf("Shard map is nil, skipping rebalance")
+		return
+	}
+
+	payloadBytes, err := json.Marshal(msg.Payload)
+	if err != nil {
+		dvm.cluster.logger.Printf("Failed to marshal rebalance payload: %v", err)
+		return
+	}
+
+	var newShardMap ShardMap
+	if err := json.Unmarshal(payloadBytes, &newShardMap); err != nil {
+		dvm.cluster.logger.Printf("Failed to unmarshal shard map: %v", err)
+		return
+	}
+
+	dvm.cluster.mutex.Lock()
+	if newShardMap.Version > dvm.cluster.shardMap.Version {
+		dvm.cluster.shardMap = &newShardMap
+		dvm.cluster.hashRing = NewConsistentHashRing()
+		for nodeID := range newShardMap.Nodes {
+			dvm.cluster.hashRing.AddNode(nodeID)
+		}
+		dvm.cluster.logger.Printf("Applied new shard map (version %d) from rebalance", newShardMap.Version)
+	} else {
+		dvm.cluster.logger.Printf("Ignoring stale shard map (got version %d, current %d)",
+			newShardMap.Version, dvm.cluster.shardMap.Version)
+	}
+	dvm.cluster.mutex.Unlock()
 }
 
 // publishClusterMessage sends a message to other cluster nodes

cluster/manager.go

@@ -154,6 +154,10 @@ func (cm *ClusterManager) handleClusterMessage(msg *nats.Msg) {
 		if update, ok := clusterMsg.Payload.(NodeUpdate); ok {
 			cm.handleNodeUpdate(update)
 		}
+	case "shard_map":
+		cm.handleShardMapUpdate(clusterMsg)
+	case "migration_update":
+		cm.handleMigrationUpdate(clusterMsg)
 	default:
 		cm.logger.Printf("Unknown cluster message type: %s", clusterMsg.Type)
 	}
@@ -217,16 +221,91 @@ func (cm *ClusterManager) handleNodeUpdate(update NodeUpdate) {
 func (cm *ClusterManager) handleRebalanceRequest(msg ClusterMessage) {
 	cm.logger.Printf("Handling rebalance request from %s", msg.From)
 
-	// Implementation would handle the specific rebalancing logic
-	// This is a simplified version
+	if !cm.IsLeader() {
+		cm.logger.Printf("Ignoring rebalance request: not the leader")
+		return
+	}
+
+	cm.mutex.RLock()
+	activeNodes := make(map[string]*NodeInfo)
+	for nodeID, nodeInfo := range cm.nodes {
+		if nodeInfo.Status == NodeStatusActive {
+			activeNodes[nodeID] = nodeInfo
+		}
+	}
+	cm.mutex.RUnlock()
+
+	if len(activeNodes) == 0 {
+		cm.logger.Printf("No active nodes for rebalancing")
+		return
+	}
+
+	placement := &ConsistentHashPlacement{}
+	newShardMap, err := placement.RebalanceShards(cm.shardMap, activeNodes)
+	if err != nil {
+		cm.logger.Printf("Failed to compute new shard map: %v", err)
+		return
+	}
+
+	cm.mutex.Lock()
+	cm.shardMap = newShardMap
+	cm.mutex.Unlock()
+
+	cm.hashRing = NewConsistentHashRing()
+	for nodeID := range activeNodes {
+		cm.hashRing.AddNode(nodeID)
+	}
+
+	cm.broadcastShardMap(newShardMap)
 }
 
 // handleMigrationRequest processes actor migration requests
 func (cm *ClusterManager) handleMigrationRequest(msg ClusterMessage) {
 	cm.logger.Printf("Handling migration request from %s", msg.From)
 
-	// Implementation would handle the specific migration logic
-	// This is a simplified version
+	var migration ActorMigration
+	payloadBytes, err := json.Marshal(msg.Payload)
+	if err != nil {
+		cm.logger.Printf("Failed to marshal migration payload: %v", err)
+		return
+	}
+	if err := json.Unmarshal(payloadBytes, &migration); err != nil {
+		cm.logger.Printf("Failed to unmarshal migration request: %v", err)
+		return
+	}
+
+	cm.logger.Printf("Actor %s migrating from %s to %s (shard %d)",
+		migration.ActorID, migration.FromNode, migration.ToNode, migration.ShardID)
+
+	if migration.FromNode == cm.nodeID {
+		cm.logger.Printf("Initiating local actor state export for %s", migration.ActorID)
+		migration.Status = string(MigrationInProgress)
+		cm.broadcastMigrationUpdate(migration)
+	}
+
+	if migration.ToNode == cm.nodeID {
+		cm.logger.Printf("Actor %s assigned to this node, waiting for state import", migration.ActorID)
+	}
+}
+
+// broadcastMigrationUpdate propagates migration status updates to the cluster
+func (cm *ClusterManager) broadcastMigrationUpdate(migration ActorMigration) {
+	msg := ClusterMessage{
+		Type:    "migration_update",
+		From:    cm.nodeID,
+		To:      "broadcast",
+		Payload: migration,
+	}
+
+	data, err := json.Marshal(msg)
+	if err != nil {
+		cm.logger.Printf("Failed to marshal migration update: %v", err)
+		return
+	}
+
+	if err := cm.natsConn.Publish("aether.cluster.migration_update", data); err != nil {
+		cm.logger.Printf("Failed to publish migration update: %v", err)
+	}
 }
 
 // triggerShardRebalancing initiates shard rebalancing across the cluster
@@ -237,12 +316,11 @@ func (cm *ClusterManager) triggerShardRebalancing(reason string) {
 
 	cm.logger.Printf("Triggering shard rebalancing: %s", reason)
 
-	// Get active nodes
-	var activeNodes []*NodeInfo
 	cm.mutex.RLock()
-	for _, node := range cm.nodes {
-		if node.Status == NodeStatusActive {
-			activeNodes = append(activeNodes, node)
+	activeNodes := make(map[string]*NodeInfo)
+	for nodeID, nodeInfo := range cm.nodes {
+		if nodeInfo.Status == NodeStatusActive {
+			activeNodes[nodeID] = nodeInfo
 		}
 	}
 	cm.mutex.RUnlock()
@@ -252,8 +330,23 @@ func (cm *ClusterManager) triggerShardRebalancing(reason string) {
 		return
 	}
 
-	// This would implement the actual rebalancing logic
-	cm.logger.Printf("Would rebalance across %d active nodes", len(activeNodes))
+	placement := &ConsistentHashPlacement{}
+	newShardMap, err := placement.RebalanceShards(cm.shardMap, activeNodes)
+	if err != nil {
+		cm.logger.Printf("Failed to compute new shard map: %v", err)
+		return
+	}
+
+	cm.mutex.Lock()
+	cm.shardMap = newShardMap
+	cm.mutex.Unlock()
+
+	cm.hashRing = NewConsistentHashRing()
+	for nodeID := range activeNodes {
+		cm.hashRing.AddNode(nodeID)
+	}
+
+	cm.broadcastShardMap(newShardMap)
 }
 
 // monitorNodes periodically checks node health and updates
@@ -319,16 +412,99 @@ func (cm *ClusterManager) GetNodes() map[string]*NodeInfo {
 	return nodes
 }
 
+// handleShardMapUpdate applies a new shard map received from the leader
+func (cm *ClusterManager) handleShardMapUpdate(msg ClusterMessage) {
+	if msg.From == cm.nodeID {
+		return
+	}
+
+	payloadBytes, err := json.Marshal(msg.Payload)
+	if err != nil {
+		cm.logger.Printf("Failed to marshal shard map payload: %v", err)
+		return
+	}
+
+	var newShardMap ShardMap
+	if err := json.Unmarshal(payloadBytes, &newShardMap); err != nil {
+		cm.logger.Printf("Failed to unmarshal shard map: %v", err)
+		return
+	}
+
+	cm.mutex.Lock()
+	if newShardMap.Version > cm.shardMap.Version {
+		cm.shardMap = &newShardMap
+		cm.hashRing = NewConsistentHashRing()
+		for nodeID := range newShardMap.Nodes {
+			cm.hashRing.AddNode(nodeID)
+		}
+		cm.logger.Printf("Applied new shard map (version %d)", newShardMap.Version)
+	} else {
+		cm.logger.Printf("Ignoring stale shard map (got version %d, current %d)",
+			newShardMap.Version, cm.shardMap.Version)
+	}
+	cm.mutex.Unlock()
+}
+
 // GetShardMap returns the current shard mapping
 func (cm *ClusterManager) GetShardMap() *ShardMap {
 	cm.mutex.RLock()
 	defer cm.mutex.RUnlock()
 
 	// Return a copy to prevent external mutation
-	return &ShardMap{
+	copy := &ShardMap{
 		Version:    cm.shardMap.Version,
 		Shards:     make(map[int][]string),
 		Nodes:      make(map[string]NodeInfo),
 		UpdateTime: cm.shardMap.UpdateTime,
 	}
+
+	for shardID, nodes := range cm.shardMap.Shards {
+		copy.Shards[shardID] = append([]string(nil), nodes...)
+	}
+
+	for nodeID, nodeInfo := range cm.shardMap.Nodes {
+		copy.Nodes[nodeID] = nodeInfo
+	}
+
+	return copy
+}
+
+// broadcastShardMap propagates a new shard map to all cluster nodes via NATS
+func (cm *ClusterManager) broadcastShardMap(newShardMap *ShardMap) {
+	msg := ClusterMessage{
+		Type:      "shard_map",
+		From:      cm.nodeID,
+		To:        "broadcast",
+		Payload:   newShardMap,
+		Timestamp: time.Now(),
+	}
+
+	data, err := json.Marshal(msg)
+	if err != nil {
+		cm.logger.Printf("Failed to marshal shard map broadcast: %v", err)
+		return
+	}
+
+	if err := cm.natsConn.Publish("aether.cluster.shard_map", data); err != nil {
+		cm.logger.Printf("Failed to publish shard map broadcast: %v", err)
+	}
+
+	cm.logger.Printf("Broadcast new shard map (version %d) to cluster", newShardMap.Version)
+}
+
+// handleMigrationUpdate processes migration status update messages from other nodes
+func (cm *ClusterManager) handleMigrationUpdate(msg ClusterMessage) {
+	var migration ActorMigration
+	payloadBytes, err := json.Marshal(msg.Payload)
+	if err != nil {
+		cm.logger.Printf("Failed to marshal migration update payload: %v", err)
+		return
+	}
+	if err := json.Unmarshal(payloadBytes, &migration); err != nil {
+		cm.logger.Printf("Failed to unmarshal migration update: %v", err)
+		return
+	}
+
+	cm.logger.Printf("Migration update for actor %s: status=%s (from %s)",
+		migration.ActorID, migration.Status, msg.From)
 }

cluster/shard.go

@@ -6,6 +6,8 @@ import (
 	"fmt"
 	"hash"
 	"hash/fnv"
+	"sort"
+	"time"
 )
 
 // MigrationStatus tracks actor migration progress
@@ -180,36 +182,116 @@ func (sm *ShardManager) GetReplicationFactor() int {
 // ConsistentHashPlacement implements PlacementStrategy using consistent hashing
 type ConsistentHashPlacement struct{}
 
-// PlaceActor places an actor using consistent hashing
+// PlaceActor places an actor using the consistent hash ring
 func (chp *ConsistentHashPlacement) PlaceActor(actorID string, shardMap *ShardMap, nodes map[string]*NodeInfo) (string, error) {
 	if len(nodes) == 0 {
 		return "", fmt.Errorf("no nodes available for placement")
 	}
 
-	// Simple consistent hash placement - in a real implementation,
-	// this would use the consistent hash ring
-	h := sha256.Sum256([]byte(actorID))
-	nodeIndex := binary.BigEndian.Uint32(h[:4]) % uint32(len(nodes))
-
-	i := 0
+	ring := NewConsistentHashRing()
 	for nodeID := range nodes {
-		if i == int(nodeIndex) {
-			return nodeID, nil
-		}
-		i++
+		ring.AddNode(nodeID)
 	}
 
-	// Fallback to first node
+	node := ring.GetNode(actorID)
+	if node == "" {
+		sortedNodeIDs := make([]string, 0, len(nodes))
 		for nodeID := range nodes {
-		return nodeID, nil
+			sortedNodeIDs = append(sortedNodeIDs, nodeID)
+		}
+		sort.Strings(sortedNodeIDs)
+		return sortedNodeIDs[0], nil
 	}
 
-	return "", fmt.Errorf("failed to place actor")
+	return node, nil
 }
 
-// RebalanceShards rebalances shards across nodes
+// RebalanceShards redistributes shards across nodes using consistent hashing
 func (chp *ConsistentHashPlacement) RebalanceShards(currentMap *ShardMap, nodes map[string]*NodeInfo) (*ShardMap, error) {
-	// This is a simplified implementation
-	// In practice, this would implement sophisticated rebalancing logic
-	return currentMap, nil
+	if len(nodes) == 0 {
+		return nil, fmt.Errorf("no nodes available for rebalancing")
+	}
+
+	ring := NewConsistentHashRing()
+	for nodeID := range nodes {
+		ring.AddNode(nodeID)
+	}
+
+	replicaCount := chp.deriveReplicaCount(currentMap)
+
+	newMap := &ShardMap{
+		Version:    currentMap.Version + 1,
+		Shards:     make(map[int][]string),
+		Nodes:      make(map[string]NodeInfo),
+		UpdateTime: time.Now(),
+	}
+
+	for nodeID, nodeInfo := range nodes {
+		newMap.Nodes[nodeID] = *nodeInfo
+	}
+
+	for shardID := range currentMap.Shards {
+		primaryNode := ring.GetNode(fmt.Sprintf("shard-%d", shardID))
+		if primaryNode == "" {
+			sortedNodeIDs := make([]string, 0, len(nodes))
+			for nodeID := range nodes {
+				sortedNodeIDs = append(sortedNodeIDs, nodeID)
+			}
+			sort.Strings(sortedNodeIDs)
+			primaryNode = sortedNodeIDs[0]
+		}
+
+		var replicaNodes []string
+		candidates := make([]string, 0, len(nodes))
+		for nodeID := range nodes {
+			if nodeID != primaryNode {
+				candidates = append(candidates, nodeID)
+			}
+		}
+		sort.Strings(candidates)
+
+		for i := 0; i < replicaCount && len(replicaNodes) < replicaCount; i++ {
+			node := ring.GetNode(fmt.Sprintf("shard-%d-replica-%d", shardID, i))
+			if node != "" && node != primaryNode {
+				found := false
+				for _, existing := range replicaNodes {
+					if existing == node {
+						found = true
+						break
+					}
+				}
+				if !found {
+					replicaNodes = append(replicaNodes, node)
+				}
+			}
+		}
+
+		if len(replicaNodes) == 0 && len(candidates) > 0 {
+			replicaNodes = append(replicaNodes, candidates[0])
+		}
+
+		if len(replicaNodes) > replicaCount {
+			replicaNodes = replicaNodes[:replicaCount]
+		}
+
+		shardNodes := []string{primaryNode}
+		shardNodes = append(shardNodes, replicaNodes...)
+		newMap.Shards[shardID] = shardNodes
+	}
+
+	return newMap, nil
+}
+
+// deriveReplicaCount extracts the replication factor from the current shard map
+func (chp *ConsistentHashPlacement) deriveReplicaCount(currentMap *ShardMap) int {
+	maxNodes := 0
+	for _, nodes := range currentMap.Shards {
+		if len(nodes) > maxNodes {
+			maxNodes = len(nodes)
+		}
+	}
+	if maxNodes <= 1 {
+		return 1
+	}
+	return maxNodes - 1
 }

cluster/shard_test.go

@@ -650,7 +650,8 @@ func TestConsistentHashPlacement_RebalanceShards(t *testing.T) {
 	placement := &ConsistentHashPlacement{}
 	currentMap := &ShardMap{
 		Version: 1,
-		Shards:  map[int][]string{0: {"node-1"}},
+		Shards:  map[int][]string{0: {"node-1"}, 1: {"node-1"}, 2: {"node-2"}},
+		Nodes:   map[string]NodeInfo{},
 	}
 	nodes := map[string]*NodeInfo{
 		"node-1": {ID: "node-1"},
@@ -662,9 +663,24 @@ func TestConsistentHashPlacement_RebalanceShards(t *testing.T) {
 	if err != nil {
 		t.Errorf("unexpected error: %v", err)
 	}
-	// Current implementation returns unchanged map
-	if result != currentMap {
-		t.Error("expected same map returned (simplified implementation)")
+	if result == nil {
+		t.Fatal("rebalance returned nil")
+	}
+	if result.Version != currentMap.Version+1 {
+		t.Errorf("expected version %d, got %d", currentMap.Version+1, result.Version)
+	}
+	if len(result.Shards) != len(currentMap.Shards) {
+		t.Errorf("expected %d shards, got %d", len(currentMap.Shards), len(result.Shards))
+	}
+	for shardID, shardNodes := range result.Shards {
+		if len(shardNodes) == 0 {
+			t.Errorf("shard %d has no nodes assigned", shardID)
+		}
+		for _, node := range shardNodes {
+			if _, exists := nodes[node]; !exists {
+				t.Errorf("shard %d assigned to unknown node %s", shardID, node)
+			}
+		}
 	}
 }
 

cluster/types.go

@@ -191,11 +191,16 @@ func (m *ModelPayload) GetID() string { return m.ID }
 // GetName implements RuntimeModel
 func (m *ModelPayload) GetName() string { return m.Name }
 
+// MaxRouteHops is the maximum number of hops a routed message can take before being dropped
+const MaxRouteHops = 10
+
 // MessagePayload is a concrete type for JSON-unmarshaling RuntimeMessage payloads.
 // Use this when receiving message data over the network.
 type MessagePayload struct {
 	TargetActorID string                 `json:"targetActorId"`
 	Type          string                 `json:"type"`
+	Hops          int                    `json:"hops,omitempty"`
+	Body          map[string]interface{} `json:"body,omitempty"`
 }
 
 // GetTargetActorID implements RuntimeMessage

event.go

@@ -184,6 +184,17 @@ type ActorSnapshot struct {
 
 // EventStore defines the interface for event persistence.
 //
+// # Immutability Guarantee
+//
+// EventStore is append-only. Once an event is persisted via SaveEvent, it is never
+// modified or deleted. The interface intentionally provides no Update or Delete methods.
+// This ensures:
+//   - Events serve as an immutable audit trail
+//   - State can be safely derived by replaying events
+//   - Concurrent reads are always safe (events never change)
+//
+// To correct a mistake, append a new event that expresses the correction.
+//
 // # Version Semantics
 //
 // Events for an actor must have monotonically increasing versions. When SaveEvent
@@ -204,10 +215,13 @@ type EventStore interface {
 	// SaveEvent persists an event to the store. The event's Version must be
 	// strictly greater than the current latest version for the actor.
 	// Returns VersionConflictError if version <= current latest version.
+	// Once saved, the event is immutable and can never be modified or deleted.
 	SaveEvent(event *Event) error
 
 	// GetEvents retrieves events for an actor from a specific version (inclusive).
 	// Returns an empty slice if no events exist for the actor.
+	// The returned events are guaranteed to be immutable - they will never be
+	// modified or deleted from the store.
 	GetEvents(actorID string, fromVersion int64) ([]*Event, error)
 
 	// GetLatestVersion returns the latest version for an actor.

store/immutability_test.go

@@ -0,0 +1,215 @@
+package store
+
+import (
+	"testing"
+	"time"
+
+	"git.flowmade.one/flowmade-one/aether"
+)
+
+// TestEventImmutability_MemoryStore verifies that events cannot be modified after persistence
+// in the in-memory event store. This demonstrates the append-only nature of event sourcing.
+func TestEventImmutability_MemoryStore(t *testing.T) {
+	store := NewInMemoryEventStore()
+	actorID := "test-actor-123"
+
+	// Create and save an event
+	originalEvent := &aether.Event{
+		ID:        "evt-immutable-1",
+		EventType: "TestEvent",
+		ActorID:   actorID,
+		Version:   1,
+		Data: map[string]interface{}{
+			"value": "original",
+		},
+		Timestamp: time.Now(),
+	}
+
+	err := store.SaveEvent(originalEvent)
+	if err != nil {
+		t.Fatalf("SaveEvent failed: %v", err)
+	}
+
+	// Retrieve the event from the store
+	events, err := store.GetEvents(actorID, 0)
+	if err != nil {
+		t.Fatalf("GetEvents failed: %v", err)
+	}
+
+	if len(events) == 0 {
+		t.Fatal("expected 1 event, got 0")
+	}
+
+	retrievedEvent := events[0]
+
+	// Verify the stored event has the correct values
+	if retrievedEvent.Data["value"] != "original" {
+		t.Errorf("Data value mismatch: got %v, want %v", retrievedEvent.Data["value"], "original")
+	}
+
+	if retrievedEvent.EventType != "TestEvent" {
+		t.Errorf("EventType mismatch: got %q, want %q", retrievedEvent.EventType, "TestEvent")
+	}
+
+	// Verify ID is correct
+	if retrievedEvent.ID != "evt-immutable-1" {
+		t.Errorf("Event ID mismatch: got %q, want %q", retrievedEvent.ID, "evt-immutable-1")
+	}
+}
+
+// TestEventImmutability_NoUpdateMethod verifies that the EventStore interface
+// has only append, read methods - no Update or Delete methods.
+func TestEventImmutability_NoUpdateMethod(t *testing.T) {
+	// This test documents that the EventStore interface is append-only.
+	// The interface intentionally provides:
+	// - SaveEvent: append only
+	// - GetEvents: read only
+	// - GetLatestVersion: read only
+	//
+	// To verify this, we demonstrate that any attempt to call non-existent
+	// update/delete methods would be caught at compile time (not runtime).
+	// This is enforced by the interface definition in event.go which does
+	// not include Update, Delete, or Modify methods.
+
+	store := NewInMemoryEventStore()
+
+	// Compile-time check: these would not compile if we tried them:
+	// store.Update(event) // compile error: no such method
+	// store.Delete(eventID) // compile error: no such method
+	// store.Modify(eventID, newData) // compile error: no such method
+
+	// Only these methods exist:
+	var eventStore aether.EventStore = store
+	if eventStore == nil {
+		t.Fatal("eventStore is nil")
+	}
+	// If we got here, the compile-time checks passed
+	t.Log("EventStore interface enforces append-only semantics by design")
+}
+
+// TestEventImmutability_VersionOnlyGoesUp verifies that versions are monotonically
+// increasing and attempting to save with a non-increasing version fails.
+func TestEventImmutability_VersionOnlyGoesUp(t *testing.T) {
+	store := NewInMemoryEventStore()
+	actorID := "actor-version-check"
+
+	// Save first event with version 1
+	event1 := &aether.Event{
+		ID:        "evt-v1",
+		EventType: "Event1",
+		ActorID:   actorID,
+		Version:   1,
+		Data:      map[string]interface{}{},
+		Timestamp: time.Now(),
+	}
+
+	err := store.SaveEvent(event1)
+	if err != nil {
+		t.Fatalf("SaveEvent(v1) failed: %v", err)
+	}
+
+	// Try to save with same version - should fail
+	event2Same := &aether.Event{
+		ID:        "evt-v1-again",
+		EventType: "Event2",
+		ActorID:   actorID,
+		Version:   1, // Same version
+		Data:      map[string]interface{}{},
+		Timestamp: time.Now(),
+	}
+
+	err = store.SaveEvent(event2Same)
+	if err == nil {
+		t.Error("expected SaveEvent(same version) to fail, but it succeeded")
+	}
+
+	// Try to save with lower version - should fail
+	event3Lower := &aether.Event{
+		ID:        "evt-v0",
+		EventType: "Event3",
+		ActorID:   actorID,
+		Version:   0, // Lower version
+		Data:      map[string]interface{}{},
+		Timestamp: time.Now(),
+	}
+
+	err = store.SaveEvent(event3Lower)
+	if err == nil {
+		t.Error("expected SaveEvent(lower version) to fail, but it succeeded")
+	}
+
+	// Save with next version - should succeed
+	event4Next := &aether.Event{
+		ID:        "evt-v2",
+		EventType: "Event4",
+		ActorID:   actorID,
+		Version:   2,
+		Data:      map[string]interface{}{},
+		Timestamp: time.Now(),
+	}
+
+	err = store.SaveEvent(event4Next)
+	if err != nil {
+		t.Fatalf("SaveEvent(v2) failed: %v", err)
+	}
+
+	// Verify we have exactly 2 events
+	events, err := store.GetEvents(actorID, 0)
+	if err != nil {
+		t.Fatalf("GetEvents failed: %v", err)
+	}
+
+	if len(events) != 2 {
+		t.Errorf("expected 2 events, got %d", len(events))
+	}
+}
+
+// TestEventImmutability_EventCannotBeDeleted verifies that there is no way to delete
+// events from the store through the EventStore interface.
+func TestEventImmutability_EventCannotBeDeleted(t *testing.T) {
+	store := NewInMemoryEventStore()
+	actorID := "actor-nodelete"
+
+	// Save an event
+	event := &aether.Event{
+		ID:        "evt-nodelete",
+		EventType: "ImportantEvent",
+		ActorID:   actorID,
+		Version:   1,
+		Data:      map[string]interface{}{"critical": true},
+		Timestamp: time.Now(),
+	}
+
+	err := store.SaveEvent(event)
+	if err != nil {
+		t.Fatalf("SaveEvent failed: %v", err)
+	}
+
+	// Retrieve it
+	events1, err := store.GetEvents(actorID, 0)
+	if err != nil {
+		t.Fatalf("GetEvents (1) failed: %v", err)
+	}
+
+	if len(events1) != 1 {
+		t.Fatal("expected 1 event after save")
+	}
+
+	// Try to delete through interface - this method doesn't exist
+	// store.Delete("evt-nodelete") // compile error: no such method
+	// store.DeleteByActorID(actorID) // compile error: no such method
+
+	// Verify the event is still there (we can't delete it)
+	events2, err := store.GetEvents(actorID, 0)
+	if err != nil {
+		t.Fatalf("GetEvents (2) failed: %v", err)
+	}
+
+	if len(events2) != 1 {
+		t.Errorf("expected 1 event (should not be deletable), got %d", len(events2))
+	}
+
+	if events2[0].ID != "evt-nodelete" {
+		t.Errorf("event ID changed: got %q, want %q", events2[0].ID, "evt-nodelete")
+	}
+}

store/jetstream.go

@@ -20,7 +20,14 @@ const (
 
 // JetStreamConfig holds configuration options for JetStreamEventStore
 type JetStreamConfig struct {
-	// StreamRetention is how long to keep events (default: 1 year)
+	// StreamRetention is how long to keep events (default: 1 year).
+	// JetStream enforces this retention policy at the storage level using a limits-based policy:
+	// - MaxAge: Events older than this duration are automatically deleted
+	// - Storage is file-based (nats.FileStorage) for durability
+	// - Once the retention period expires, events are permanently removed from the stream
+	// This ensures that old events do not consume storage indefinitely.
+	// To keep events indefinitely, set StreamRetention to a very large value or configure
+	// a custom retention policy in the JetStream stream configuration.
 	StreamRetention time.Duration
 	// ReplicaCount is the number of replicas for high availability (default: 1)
 	ReplicaCount int
@@ -42,6 +49,21 @@ func DefaultJetStreamConfig() JetStreamConfig {
 // JetStreamEventStore implements EventStore using NATS JetStream for persistence.
 // It also implements EventStoreWithErrors to report malformed events during replay.
 //
+// ## Immutability Guarantee
+//
+// JetStreamEventStore is append-only. Events are stored in a JetStream stream that
+// is configured with file-based storage (nats.FileStorage) and a retention policy
+// (nats.LimitsPolicy). The configured MaxAge retention policy ensures that old events
+// eventually expire, but during their lifetime, events are never modified or deleted
+// through the EventStore API. Once an event is published to the stream:
+// - It cannot be updated
+// - It cannot be deleted before expiration
+// - It can only be read
+//
+// This architectural guarantee, combined with the EventStore interface providing
+// no Update or Delete methods, ensures events are immutable and suitable as an
+// audit trail.
+//
 // ## Version Cache Invalidation Strategy
 //
 // JetStreamEventStore maintains an in-memory cache of actor versions for optimistic
@@ -72,12 +94,6 @@ type JetStreamEventStore struct {
 
 
 
-
-
-
-
-
-
 // NewJetStreamEventStore creates a new JetStream-based event store with default configuration
 func NewJetStreamEventStore(natsConn *nats.Conn, streamName string) (*JetStreamEventStore, error) {
 	return NewJetStreamEventStoreWithConfig(natsConn, streamName, DefaultJetStreamConfig())