release: v0.12.0 — network sync, conflict strategies, MCP tracker, WS backpressure

Closes the 4 remaining embedded Core gaps from #73:

1. Configurable per-entity-type conflict resolution (#76)
   - MergeStrategy enum: AppendOnly, LastWriteWins, FirstWriteWins
   - Config::merge_strategy("prefix", strategy) with longest-prefix matching
   - CrdtResolver::with_strategies() wired through EmbeddedCore::open()

2. Network sync transport (#75)
   - SyncClient for HTTP pull/push against remote Core
   - POST /api/v1/sync/pull and /api/v1/sync/push server endpoints
   - EmbeddedCore helpers: version_vector(), events_for_sync(), receive_sync_push()
   - embedded-sync feature now includes reqwest for HTTP transport

3. MCP tool event emission helper
   - McpToolTracker: emit_tool_result(), emit_tool_error(), track() with auto-timing
   - Events feed directly into ToolCallAuditProjection
   - Feature-gated on embedded-projections

4. WebSocket backpressure for token streams
   - WebSocketConfig: capacity, batch_interval_ms, max_batch_size
   - Batched send loop with tokio::time::interval + early flush on max_batch_size
   - RecvError::Lagged → {"type":"lagged","missed":N} client notification
   - Fully backward compatible: WebSocketManager::new() unchanged

20 new tests, 1489 lib tests pass, 0 regressions.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: decebal

apps/control-plane/main.go

@@ -30,7 +30,7 @@ var openAPIYAML []byte
 // Control plane configuration constants.
 const (
 	// Version is the current version of the control plane.
-	Version = "0.11.0"
+	Version = "0.12.0"
 	// DefaultPort is the default port the control plane listens on.
 	DefaultPort = "3901"
 	// CoreServiceURL is the URL of the core event store service.

apps/control-plane/tracing.go

@@ -19,7 +19,7 @@ import (
 
 const (
 	serviceName    = "allsource-control-plane"
-	serviceVersion = "0.11.0"
+	serviceVersion = "0.12.0"
 )
 
 // TracingConfig holds OpenTelemetry configuration

apps/core/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "allsource-core"
-version = "0.11.0"
+version = "0.12.0"
 edition = "2024"
 rust-version = "1.92"
 authors = ["AllSource Team"]
@@ -58,7 +58,7 @@ vector-search = ["fastembed", "instant-distance"]
 keyword-search = ["tantivy"]
 embedded = []
 embedded-phase2 = ["embedded"]
-embedded-sync = ["embedded"]
+embedded-sync = ["embedded", "dep:reqwest"]
 embedded-replicant = ["embedded"]
 embedded-streaming = ["embedded"]
 embedded-projections = ["embedded"]
@@ -156,6 +156,11 @@ name = "bidirectional_sync"
 path = "tests/bidirectional_sync.rs"
 required-features = ["embedded-sync"]
 
+[[test]]
+name = "http_sync_transport"
+path = "tests/http_sync_transport.rs"
+required-features = ["embedded-sync"]
+
 [[test]]
 name = "replicant_protocol"
 path = "tests/replicant_protocol.rs"
@@ -171,6 +176,11 @@ name = "ai_projection_templates"
 path = "tests/ai_projection_templates.rs"
 required-features = ["embedded-projections"]
 
+[[test]]
+name = "mcp_tool_emission"
+path = "tests/mcp_tool_emission.rs"
+required-features = ["embedded-projections"]
+
 [[test]]
 name = "toon_format"
 path = "tests/toon_format.rs"

apps/core/src/embedded/config.rs

@@ -1,3 +1,4 @@
+use crate::infrastructure::cluster::crdt::MergeStrategy;
 use std::path::{Path, PathBuf};
 
 /// Configuration for [`EmbeddedCore`](super::EmbeddedCore).
@@ -11,6 +12,7 @@ pub struct EmbeddedConfig {
     parquet_flush_interval_secs: u64,
     single_tenant: bool,
     node_id: Option<u32>,
+    merge_strategies: Vec<(String, MergeStrategy)>,
 }
 
 impl EmbeddedConfig {
@@ -47,6 +49,11 @@ impl EmbeddedConfig {
         self.node_id
     }
 
+    /// Per-event-type merge strategies for conflict resolution.
+    pub fn merge_strategies(&self) -> &[(String, MergeStrategy)] {
+        &self.merge_strategies
+    }
+
     pub(crate) fn parquet_flush_interval_secs(&self) -> u64 {
         self.parquet_flush_interval_secs
     }
@@ -60,6 +67,7 @@ pub struct ConfigBuilder {
     parquet_flush_interval_secs: u64,
     single_tenant: bool,
     node_id: Option<u32>,
+    merge_strategies: Vec<(String, MergeStrategy)>,
 }
 
 impl Default for ConfigBuilder {
@@ -71,6 +79,7 @@ impl Default for ConfigBuilder {
             parquet_flush_interval_secs: 300,
             single_tenant: true,
             node_id: None,
+            merge_strategies: Vec::new(),
         }
     }
 }
@@ -125,6 +134,16 @@ impl ConfigBuilder {
         self
     }
 
+    /// Register a per-event-type merge strategy for conflict resolution.
+    ///
+    /// The `prefix` is matched against event types: `"config."` matches
+    /// `"config.updated"`, `"config.deleted"`, etc. The longest matching
+    /// prefix wins. Unmatched types default to `AppendOnly`.
+    pub fn merge_strategy(mut self, prefix: impl Into<String>, strategy: MergeStrategy) -> Self {
+        self.merge_strategies.push((prefix.into(), strategy));
+        self
+    }
+
     /// Build the configuration. Returns `Result` for forward compatibility.
     pub fn build(self) -> crate::error::Result<EmbeddedConfig> {
         Ok(EmbeddedConfig {
@@ -134,6 +153,7 @@ impl ConfigBuilder {
             parquet_flush_interval_secs: self.parquet_flush_interval_secs,
             single_tenant: self.single_tenant,
             node_id: self.node_id,
+            merge_strategies: self.merge_strategies,
         })
     }
 }

apps/core/src/embedded/core.rs

@@ -104,7 +104,13 @@ impl EmbeddedCore {
         let (hlc, resolver) = match config.node_id() {
             Some(node_id) => {
                 let hlc = Arc::new(HybridLogicalClock::new(node_id));
-                let resolver = Arc::new(CrdtResolver::new());
+                let resolver = if config.merge_strategies().is_empty() {
+                    Arc::new(CrdtResolver::new())
+                } else {
+                    Arc::new(CrdtResolver::with_strategies(
+                        config.merge_strategies().to_vec(),
+                    ))
+                };
                 (Some(hlc), Some(resolver))
             }
             None => (None, None),
@@ -472,6 +478,175 @@ impl EmbeddedCore {
         self.store.stats()
     }
 
+    /// Get the merged version vector across all known regions.
+    ///
+    /// Returns a map of `region_id → latest HLC timestamp`. Used by the
+    /// sync transport to compute deltas.
+    pub fn version_vector(
+        &self,
+    ) -> std::collections::BTreeMap<String, crate::infrastructure::cluster::hlc::HlcTimestamp> {
+        match &self.resolver {
+            Some(resolver) => {
+                let all_vv = resolver.all_version_vectors();
+                let mut merged = crate::infrastructure::cluster::crdt::VersionVector::new();
+                for (_region, vv) in &all_vv {
+                    merged.merge(vv);
+                }
+                merged.entries().clone()
+            }
+            None => std::collections::BTreeMap::new(),
+        }
+    }
+
+    /// Get the node's region identifier (e.g., "node-1").
+    ///
+    /// Returns `None` if sync is not configured (no `node_id`).
+    pub fn region_id(&self) -> Option<String> {
+        self.hlc.as_ref().map(|hlc| format!("node-{}", hlc.node_id()))
+    }
+
+    /// Receive events from a remote sync push.
+    ///
+    /// Applies CRDT conflict resolution to each event and ingests
+    /// accepted events. Returns `(accepted, skipped)` counts.
+    pub async fn receive_sync_push(
+        &self,
+        events: Vec<ReplicatedEvent>,
+    ) -> Result<(usize, usize)> {
+        let (Some(_hlc), Some(resolver)) = (&self.hlc, &self.resolver) else {
+            return Err(crate::error::AllSourceError::InvalidInput(
+                "sync requires node_id to be configured".to_string(),
+            ));
+        };
+
+        let mut accepted = Vec::new();
+        let mut skipped = 0usize;
+
+        for event in &events {
+            let resolution = resolver.resolve(event);
+            if resolution == ConflictResolution::Accept {
+                resolver.accept(event);
+                accepted.push(event.clone());
+            } else {
+                skipped += 1;
+            }
+        }
+
+        let accepted_count = accepted.len();
+
+        if !accepted.is_empty() {
+            let store = Arc::clone(&self.store);
+            tokio::task::spawn_blocking(move || {
+                for rep_event in accepted {
+                    // Convert ReplicatedEvent back to domain Event
+                    let event_data = &rep_event.event_data;
+                    let event_type = event_data
+                        .get("event_type")
+                        .and_then(|v| v.as_str())
+                        .unwrap_or("unknown")
+                        .to_string();
+                    let entity_id = event_data
+                        .get("entity_id")
+                        .and_then(|v| v.as_str())
+                        .unwrap_or("unknown")
+                        .to_string();
+                    let tenant_id = event_data
+                        .get("tenant_id")
+                        .and_then(|v| v.as_str())
+                        .unwrap_or("default")
+                        .to_string();
+                    let payload = event_data
+                        .get("payload")
+                        .cloned()
+                        .unwrap_or(serde_json::json!({}));
+                    let metadata = event_data.get("metadata").cloned();
+
+                    let domain_event = Event::from_strings(
+                        event_type,
+                        entity_id,
+                        tenant_id,
+                        payload,
+                        metadata,
+                    )?;
+                    store.ingest(domain_event)?;
+                }
+                Ok::<(), crate::error::AllSourceError>(())
+            })
+            .await
+            .map_err(|e| {
+                crate::error::AllSourceError::InvalidInput(format!("spawn_blocking failed: {e}"))
+            })??;
+        }
+
+        Ok((accepted_count, skipped))
+    }
+
+    /// Get all events as `ReplicatedEvent`s for sync push.
+    ///
+    /// Optionally filtered by a version vector threshold (only events
+    /// newer than the threshold are returned).
+    pub async fn events_for_sync(
+        &self,
+        since_vv: &std::collections::BTreeMap<String, crate::infrastructure::cluster::hlc::HlcTimestamp>,
+    ) -> Result<Vec<ReplicatedEvent>> {
+        let Some(hlc) = &self.hlc else {
+            return Err(crate::error::AllSourceError::InvalidInput(
+                "sync requires node_id to be configured".to_string(),
+            ));
+        };
+
+        let self_region = format!("node-{}", hlc.node_id());
+        let since = since_vv
+            .get(&self_region)
+            .map(|ts| {
+                chrono::DateTime::from_timestamp_millis(ts.physical_ms as i64).unwrap_or_default()
+            });
+
+        let store = Arc::clone(&self.store);
+        let all_events = tokio::task::spawn_blocking(move || {
+            store.query(QueryEventsRequest {
+                entity_id: None,
+                event_type: None,
+                tenant_id: None,
+                as_of: None,
+                since,
+                until: None,
+                limit: None,
+                event_type_prefix: None,
+                payload_filter: None,
+            })
+        })
+        .await
+        .map_err(|e| {
+            crate::error::AllSourceError::InvalidInput(format!("spawn_blocking failed: {e}"))
+        })??;
+
+        let node_id = hlc.node_id();
+        let mut replicated = Vec::with_capacity(all_events.len());
+        let mut last_ms = 0u64;
+        let mut logical = 0u32;
+
+        for event in &all_events {
+            let event_ms = event.timestamp().timestamp_millis() as u64;
+            if event_ms == last_ms {
+                logical += 1;
+            } else {
+                last_ms = event_ms;
+                logical = 0;
+            }
+            let ts = HlcTimestamp::new(event_ms, logical, node_id);
+
+            replicated.push(ReplicatedEvent {
+                event_id: event.id().to_string(),
+                hlc_timestamp: ts,
+                origin_region: self_region.clone(),
+                event_data: serde_json::to_value(EventView::from(event)).unwrap_or_default(),
+            });
+        }
+
+        Ok(replicated)
+    }
+
     /// Get a reference to the underlying `EventStore`.
     ///
     /// Escape hatch for advanced use cases that need direct access.