From fd9e4047dbd5bafed086bd48e35102d2798d0da4 Mon Sep 17 00:00:00 2001 From: Travis Gilbert <1travisgilbert@gmail.com> Date: Mon, 22 Jun 2026 21:39:28 -0400 Subject: [PATCH 1/3] feat(morphology): add advisory message passing operation --- crates/rustyred-core/src/algorithm_ops.rs | 274 ++++++++++++++++- crates/rustyred-core/src/lib.rs | 7 + crates/rustyred-core/src/morphology.rs | 349 ++++++++++++++++++++++ 3 files changed, 629 insertions(+), 1 deletion(-) create mode 100644 crates/rustyred-core/src/morphology.rs diff --git a/crates/rustyred-core/src/algorithm_ops.rs b/crates/rustyred-core/src/algorithm_ops.rs index 4adc1a0..769fb0b 100644 --- a/crates/rustyred-core/src/algorithm_ops.rs +++ b/crates/rustyred-core/src/algorithm_ops.rs @@ -8,7 +8,7 @@ //! surfaces it through `execute_request_json`, the MCP tool generation, and the //! HTTP route generation with no per-surface code. -use std::collections::{BTreeSet, HashMap}; +use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::sync::Arc; use serde_json::{json, Value}; @@ -24,6 +24,10 @@ use crate::graph::{ pagerank, paths_shortest, paths_shortest_weighted, personalized_pagerank, EdgeTuple, }; use crate::graph_store::EdgeRecord; +use crate::morphology::{ + default_relation_weights, is_morphological_relation, morphological_edges_from_records, + morphology_stats, MorphologicalEdge, +}; use crate::operation::{ arg_bool, arg_f64, arg_str, arg_u64, arg_usize, estimate_from_coefficients, require_str, AlgorithmGraph, AlgorithmOperation, GraphCounts, MemoryEstimate, OperationError, OperationMode, @@ -51,6 +55,7 @@ pub fn algorithm_operations() -> Vec> { Arc::new(SccOp), Arc::new(NodeSimilarityOp), Arc::new(LinkPredictionOp), + Arc::new(MorphologicalMessagePassingOp), Arc::new(PersonalizedPageRankOp), Arc::new(ConnectedComponentsOp), Arc::new(LabelPropagationOp), @@ -889,6 +894,120 @@ impl AlgorithmOperation for LinkPredictionOp { } } +// ===== Morphological graph message-passing scaffold ===== + +#[derive(Clone, Copy, Debug)] +pub struct MorphologicalMessagePassingOp; + +impl AlgorithmOperation for MorphologicalMessagePassingOp { + fn command(&self) -> &'static str { + "rustyred.algorithm.morphological_message_passing" + } + fn name(&self) -> &'static str { + "morphological_message_passing" + } + fn summary(&self) -> &'static str { + "Advisory message passing over city2graph-style touched_to / connected_to / faced_to edges." + } + fn modes(&self) -> &'static [OperationMode] { + STREAM_STATS_MUTATE + } + fn input_schema(&self) -> Value { + json!({ + "type": "object", + "properties": { + "mode": { "type": "string", "enum": ["stream", "stats", "mutate", "estimate"], "default": "stream" }, + "feature_property": { "type": "string", "default": "features" }, + "mutate_property": { "type": "string", "default": "morphological_embedding" }, + "iterations": { "type": "integer", "default": 1, "minimum": 1 }, + "top_k": { "type": "integer" }, + "relation_weights": { + "type": "object", + "additionalProperties": { "type": "number" }, + "default": { "touched_to": 1.0, "connected_to": 0.8, "faced_to": 0.6 } + } + } + }) + } + fn estimate(&self, counts: GraphCounts, _args: &Value) -> MemoryEstimate { + // Feature table + accumulator table, bounded by the existing graph size. + estimate_from_coefficients(counts, 8192, 96, 192, 48, "Morphological message passing") + } + fn run( + &self, + graph: &mut dyn AlgorithmGraph, + mode: OperationMode, + args: &Value, + ) -> Result { + let edges = graph.list_edges()?; + let morphological_edges = morphological_edges_from_records(&edges); + let stats = morphology_stats(&morphological_edges); + if mode == OperationMode::Stats { + return Ok(json!({ + "operation": self.command(), + "mode": "stats", + "stats": stats, + })); + } + if morphological_edges.is_empty() { + return Ok(json!({ + "operation": self.command(), + "mode": mode.as_str(), + "node_count": 0, + "edge_count": 0, + "rows": [], + })); + } + + let feature_property = arg_str(args, "feature_property").unwrap_or("features"); + let features = morphological_features(graph, &morphological_edges, feature_property)?; + if features.is_empty() { + return Err(OperationError::invalid_params(format!( + "no nodes incident to morphological edges carry a numeric array property {feature_property:?}" + ))); + } + let iterations = arg_usize(args, "iterations", 1).max(1); + let weights = relation_weights_from_args(args); + let passed = + crate::morphology::message_pass(&features, &morphological_edges, iterations, &weights) + .map_err(|error| OperationError::invalid_params(error.to_string()))?; + let rows = morphological_rows(passed, args.get("top_k").and_then(Value::as_u64)); + + match mode { + OperationMode::Stream => Ok(json!({ + "operation": self.command(), + "mode": "stream", + "iterations": iterations, + "feature_property": feature_property, + "relation_weights": weights, + "edge_count": stats.edge_count, + "node_count": rows.len(), + "rows": rows, + })), + OperationMode::Mutate => { + let property = + arg_str(args, "mutate_property").unwrap_or("morphological_embedding"); + for row in &rows { + let node_id = row["node_id"].as_str().ok_or_else(|| { + OperationError::invalid_params("internal row missing node_id") + })?; + graph.write_node_property(node_id, property, row["embedding"].clone())?; + } + Ok(json!({ + "operation": self.command(), + "mode": "mutate", + "iterations": iterations, + "feature_property": feature_property, + "mutate_property": property, + "edge_count": stats.edge_count, + "nodes_written": rows.len(), + })) + } + OperationMode::Stats | OperationMode::Estimate => unreachable!("handled earlier"), + } + } +} + // ===== Conform the pre-existing algorithms to the operation contract ===== #[derive(Clone, Copy, Debug)] @@ -1325,6 +1444,88 @@ fn edge_tuples(edges: &[EdgeRecord]) -> Vec { .collect() } +fn morphological_features( + graph: &dyn AlgorithmGraph, + edges: &[MorphologicalEdge], + feature_property: &str, +) -> Result>, OperationError> { + let mut node_ids = BTreeSet::new(); + for edge in edges { + node_ids.insert(edge.source_id.as_str()); + node_ids.insert(edge.target_id.as_str()); + } + + let mut features = BTreeMap::new(); + for node_id in node_ids { + let Some(node) = graph.get_node(node_id)? else { + continue; + }; + if let Some(vector) = read_f64_vector(&node.properties, feature_property) { + features.insert(node_id.to_string(), vector); + } + } + Ok(features) +} + +fn read_f64_vector(properties: &Value, key: &str) -> Option> { + properties + .get(key)? + .as_array()? + .iter() + .map(Value::as_f64) + .collect() +} + +fn relation_weights_from_args(args: &Value) -> BTreeMap { + let mut weights = default_relation_weights(); + let Some(object) = args.get("relation_weights").and_then(Value::as_object) else { + return weights; + }; + for (relation, value) in object { + if is_morphological_relation(relation) { + if let Some(weight) = value.as_f64() { + weights.insert(relation.trim().to_ascii_lowercase(), weight); + } + } + } + weights +} + +fn morphological_rows(passed: BTreeMap>, top_k: Option) -> Vec { + let mut rows: Vec = passed + .into_iter() + .map(|(node_id, embedding)| { + let norm = embedding + .iter() + .map(|value| value * value) + .sum::() + .sqrt(); + json!({ + "node_id": node_id, + "embedding": embedding, + "norm": norm, + }) + }) + .collect(); + rows.sort_by(|left, right| { + let left_norm = left["norm"].as_f64().unwrap_or(0.0); + let right_norm = right["norm"].as_f64().unwrap_or(0.0); + right_norm + .partial_cmp(&left_norm) + .unwrap_or(std::cmp::Ordering::Equal) + .then_with(|| { + left["node_id"] + .as_str() + .unwrap_or_default() + .cmp(right["node_id"].as_str().unwrap_or_default()) + }) + }); + if let Some(top_k) = top_k { + rows.truncate(top_k as usize); + } + rows +} + #[cfg(test)] mod tests { use super::*; @@ -1461,6 +1662,77 @@ mod tests { assert!(scored.contains("b") || scored.contains("c")); } + #[test] + fn morphological_message_passing_stream_stats_and_mutate() { + let mut store = InMemoryGraphStore::new(); + for (id, features) in [ + ("place:a", vec![1.0, 0.0]), + ("place:b", vec![0.0, 1.0]), + ("movement:main", vec![0.25, 0.25]), + ] { + store + .upsert_node(NodeRecord::new( + id, + ["Morphology"], + json!({ "features": features }), + )) + .unwrap(); + } + store + .upsert_edge(EdgeRecord::new( + "a-touch-b", + "place:a", + "touched_to", + "place:b", + json!({}), + )) + .unwrap(); + store + .upsert_edge(EdgeRecord::new( + "a-face-main", + "place:a", + "faced_to", + "movement:main", + json!({}), + )) + .unwrap(); + + let op = MorphologicalMessagePassingOp; + let stream = dispatch_operation( + &op, + &mut store, + &json!({ + "mode": "stream", + "feature_property": "features", + "relation_weights": { "touched_to": 1.0, "faced_to": 1.0 } + }), + ) + .unwrap(); + assert_eq!(stream["mode"], "stream"); + assert_eq!(stream["edge_count"], 2); + let place_b = stream["rows"] + .as_array() + .unwrap() + .iter() + .find(|row| row["node_id"] == "place:b") + .expect("place:b row"); + assert_eq!(place_b["embedding"], json!([0.5, 0.5])); + + let stats = dispatch_operation(&op, &mut store, &json!({ "mode": "stats" })).unwrap(); + assert_eq!(stats["stats"]["touched_to_count"], 1); + assert_eq!(stats["stats"]["faced_to_count"], 1); + + let mutate = dispatch_operation( + &op, + &mut store, + &json!({ "mode": "mutate", "mutate_property": "morph" }), + ) + .unwrap(); + assert_eq!(mutate["nodes_written"], 3); + let node = GraphStore::get_node(&store, "place:b").unwrap(); + assert_eq!(node.properties["morph"], json!([0.5, 0.5])); + } + #[test] fn scc_stream_reports_cycle_and_condensation() { let mut store = store_with_triangles(); diff --git a/crates/rustyred-core/src/lib.rs b/crates/rustyred-core/src/lib.rs index 825f5b8..b65ed6c 100644 --- a/crates/rustyred-core/src/lib.rs +++ b/crates/rustyred-core/src/lib.rs @@ -23,6 +23,7 @@ pub mod geometry; pub mod graph; pub mod graph_store; pub mod instant_kg; +pub mod morphology; pub mod operation; pub mod plugin; pub mod spatial; @@ -80,6 +81,12 @@ pub use instant_kg::{ InstantKgStatus, PprResult, SearchResult, SessionDelta, INSTANT_KG_DEFAULT_ENCODER_VERSION, INSTANT_KG_DEFAULT_INGEST_VERSION, INSTANT_KG_PROTOCOL_VERSION, }; +pub use morphology::{ + default_relation_weights, dual_graph_edges, is_morphological_relation, + message_pass as morphological_message_pass, morphological_edges_from_records, morphology_stats, + relation_weights_from_map, MorphologicalEdge, MorphologicalNodeKind, MorphologyError, + MorphologyStats, StreetSegmentTopology, CONNECTED_TO, FACED_TO, TOUCHED_TO, +}; pub use operation::{ dispatch_operation, AlgorithmGraph, AlgorithmOperation, GraphCounts, MemoryEstimate, OperationError, OperationMode, diff --git a/crates/rustyred-core/src/morphology.rs b/crates/rustyred-core/src/morphology.rs new file mode 100644 index 0000000..5d35c8f --- /dev/null +++ b/crates/rustyred-core/src/morphology.rs @@ -0,0 +1,349 @@ +//! Morphological graph primitives for the city2graph parity lane. +//! +//! The Python city2graph bridge remains the oracle. This module is the small, +//! advisory Rust-side spine: typed relation records, deterministic topology +//! helpers, and backend-neutral message passing that can be lowered to Burn once +//! the geometry parity layer is in place. + +use std::collections::{BTreeMap, BTreeSet, HashMap}; + +use serde::{Deserialize, Serialize}; + +use crate::graph_store::EdgeRecord; + +pub const TOUCHED_TO: &str = "touched_to"; +pub const CONNECTED_TO: &str = "connected_to"; +pub const FACED_TO: &str = "faced_to"; + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum MorphologicalNodeKind { + Place, + Movement, +} + +impl MorphologicalNodeKind { + pub fn as_str(self) -> &'static str { + match self { + Self::Place => "place", + Self::Movement => "movement", + } + } +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +pub struct MorphologicalEdge { + pub edge_id: String, + pub source_id: String, + pub source_kind: MorphologicalNodeKind, + pub relation: String, + pub target_id: String, + pub target_kind: MorphologicalNodeKind, + pub confidence: f64, +} + +impl MorphologicalEdge { + pub fn new( + source_id: impl Into, + source_kind: MorphologicalNodeKind, + relation: impl Into, + target_id: impl Into, + target_kind: MorphologicalNodeKind, + ) -> Self { + let source_id = source_id.into(); + let relation = normalize_relation(&relation.into()); + let target_id = target_id.into(); + let edge_id = format!("morphological:{source_id}:{relation}:{target_id}"); + Self { + edge_id, + source_id, + source_kind, + relation, + target_id, + target_kind, + confidence: 1.0, + } + } + + pub fn with_confidence(mut self, confidence: f64) -> Self { + self.confidence = confidence.clamp(0.0, 1.0); + self + } + + pub fn from_edge_record(edge: &EdgeRecord) -> Option { + let relation = normalize_relation(&edge.edge_type); + let (source_kind, target_kind) = relation_kinds(&relation)?; + Some(Self { + edge_id: edge.id.clone(), + source_id: edge.from_id.clone(), + source_kind, + relation, + target_id: edge.to_id.clone(), + target_kind, + confidence: edge.effective_confidence(), + }) + } +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct StreetSegmentTopology { + pub segment_id: String, + pub start_node_id: String, + pub end_node_id: String, +} + +impl StreetSegmentTopology { + pub fn new( + segment_id: impl Into, + start_node_id: impl Into, + end_node_id: impl Into, + ) -> Self { + Self { + segment_id: segment_id.into(), + start_node_id: start_node_id.into(), + end_node_id: end_node_id.into(), + } + } + + fn endpoints(&self) -> [&str; 2] { + [&self.start_node_id, &self.end_node_id] + } +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct MorphologyStats { + pub edge_count: usize, + pub touched_to_count: usize, + pub connected_to_count: usize, + pub faced_to_count: usize, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct MorphologyError { + pub message: String, +} + +impl MorphologyError { + fn new(message: impl Into) -> Self { + Self { + message: message.into(), + } + } +} + +impl std::fmt::Display for MorphologyError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.message) + } +} + +impl std::error::Error for MorphologyError {} + +pub fn is_morphological_relation(relation: &str) -> bool { + relation_kinds(&normalize_relation(relation)).is_some() +} + +pub fn morphological_edges_from_records(edges: &[EdgeRecord]) -> Vec { + edges + .iter() + .filter(|edge| !edge.tombstone) + .filter_map(MorphologicalEdge::from_edge_record) + .collect() +} + +pub fn dual_graph_edges(segments: &[StreetSegmentTopology]) -> Vec { + let mut out = Vec::new(); + for (index, left) in segments.iter().enumerate() { + for right in segments.iter().skip(index + 1) { + if share_endpoint(left, right) { + out.push(MorphologicalEdge::new( + &left.segment_id, + MorphologicalNodeKind::Movement, + CONNECTED_TO, + &right.segment_id, + MorphologicalNodeKind::Movement, + )); + out.push(MorphologicalEdge::new( + &right.segment_id, + MorphologicalNodeKind::Movement, + CONNECTED_TO, + &left.segment_id, + MorphologicalNodeKind::Movement, + )); + } + } + } + out +} + +pub fn morphology_stats(edges: &[MorphologicalEdge]) -> MorphologyStats { + let mut stats = MorphologyStats { + edge_count: edges.len(), + touched_to_count: 0, + connected_to_count: 0, + faced_to_count: 0, + }; + for edge in edges { + match edge.relation.as_str() { + TOUCHED_TO => stats.touched_to_count += 1, + CONNECTED_TO => stats.connected_to_count += 1, + FACED_TO => stats.faced_to_count += 1, + _ => {} + } + } + stats +} + +pub fn default_relation_weights() -> BTreeMap { + BTreeMap::from([ + (TOUCHED_TO.to_string(), 1.0), + (CONNECTED_TO.to_string(), 0.8), + (FACED_TO.to_string(), 0.6), + ]) +} + +pub fn message_pass( + features: &BTreeMap>, + edges: &[MorphologicalEdge], + iterations: usize, + relation_weights: &BTreeMap, +) -> Result>, MorphologyError> { + let dimension = feature_dimension(features)?; + let mut state = features.clone(); + for _ in 0..iterations { + let mut sums: BTreeMap> = BTreeMap::new(); + let mut weights: BTreeMap = BTreeMap::new(); + for edge in edges { + let Some(source) = state.get(&edge.source_id) else { + continue; + }; + let weight = + relation_weights.get(&edge.relation).copied().unwrap_or(1.0) * edge.confidence; + if weight == 0.0 { + continue; + } + let entry = sums + .entry(edge.target_id.clone()) + .or_insert_with(|| vec![0.0; dimension]); + for (slot, value) in entry.iter_mut().zip(source.iter()) { + *slot += *value * weight; + } + *weights.entry(edge.target_id.clone()).or_insert(0.0) += weight.abs(); + } + + let mut next = state.clone(); + for (node_id, sum) in sums { + let denom = weights.get(&node_id).copied().unwrap_or(1.0).max(1.0); + let incoming = sum.into_iter().map(|value| value / denom); + match next.get_mut(&node_id) { + Some(existing) => { + for (slot, value) in existing.iter_mut().zip(incoming) { + *slot = (*slot + value) / 2.0; + } + } + None => { + next.insert(node_id, incoming.collect()); + } + } + } + state = next; + } + Ok(state) +} + +fn feature_dimension(features: &BTreeMap>) -> Result { + let mut dimensions = BTreeSet::new(); + for values in features.values() { + if values.is_empty() { + return Err(MorphologyError::new("feature vectors must not be empty")); + } + dimensions.insert(values.len()); + } + match dimensions.len() { + 0 => Err(MorphologyError::new( + "at least one feature vector is required", + )), + 1 => Ok(*dimensions.iter().next().expect("one dimension")), + _ => Err(MorphologyError::new( + "all feature vectors must have the same dimension", + )), + } +} + +fn share_endpoint(left: &StreetSegmentTopology, right: &StreetSegmentTopology) -> bool { + let left_endpoints = left.endpoints(); + let right_endpoints = right.endpoints(); + left_endpoints + .iter() + .any(|endpoint| right_endpoints.contains(endpoint)) +} + +fn relation_kinds(relation: &str) -> Option<(MorphologicalNodeKind, MorphologicalNodeKind)> { + match relation { + TOUCHED_TO => Some((MorphologicalNodeKind::Place, MorphologicalNodeKind::Place)), + CONNECTED_TO => Some(( + MorphologicalNodeKind::Movement, + MorphologicalNodeKind::Movement, + )), + FACED_TO => Some(( + MorphologicalNodeKind::Place, + MorphologicalNodeKind::Movement, + )), + _ => None, + } +} + +fn normalize_relation(raw: &str) -> String { + raw.trim().to_ascii_lowercase() +} + +pub fn relation_weights_from_map(raw: &HashMap) -> BTreeMap { + let mut weights = default_relation_weights(); + for (relation, weight) in raw { + let relation = normalize_relation(relation); + if is_morphological_relation(&relation) { + weights.insert(relation, *weight); + } + } + weights +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn dual_graph_connects_segments_that_share_endpoints() { + let edges = dual_graph_edges(&[ + StreetSegmentTopology::new("s1", "a", "b"), + StreetSegmentTopology::new("s2", "b", "c"), + StreetSegmentTopology::new("s3", "d", "e"), + ]); + + assert_eq!(edges.len(), 2); + assert!(edges + .iter() + .any(|edge| edge.source_id == "s1" && edge.target_id == "s2")); + assert!(edges.iter().all(|edge| edge.relation == CONNECTED_TO)); + } + + #[test] + fn message_passing_uses_typed_relation_weights() { + let features = BTreeMap::from([ + ("place:a".to_string(), vec![1.0, 0.0]), + ("place:b".to_string(), vec![0.0, 1.0]), + ]); + let edges = vec![MorphologicalEdge::new( + "place:a", + MorphologicalNodeKind::Place, + TOUCHED_TO, + "place:b", + MorphologicalNodeKind::Place, + )]; + let weights = BTreeMap::from([(TOUCHED_TO.to_string(), 1.0)]); + + let out = message_pass(&features, &edges, 1, &weights).unwrap(); + assert_eq!(out["place:a"], vec![1.0, 0.0]); + assert_eq!(out["place:b"], vec![0.5, 0.5]); + } +} From 8629e708225a58ebdd64e0b9b911fa0fac68a3e0 Mon Sep 17 00:00:00 2001 From: Travis Gilbert <1travisgilbert@gmail.com> Date: Mon, 22 Jun 2026 21:45:41 -0400 Subject: [PATCH 2/3] chore(fmt): satisfy workspace rustfmt --- crates/rustyred-server/src/graph_sync.rs | 12 ++++++++++-- crates/rustyred-server/src/yjs_sync.rs | 3 ++- 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/crates/rustyred-server/src/graph_sync.rs b/crates/rustyred-server/src/graph_sync.rs index 6d983af..baf18d9 100644 --- a/crates/rustyred-server/src/graph_sync.rs +++ b/crates/rustyred-server/src/graph_sync.rs @@ -448,7 +448,11 @@ mod tests { let decoded: VersionVector = serde_json::from_slice(payload).unwrap(); let diff = diff_since(&a, &decoded); - assert_eq!(diff.mutations.len(), 1, "diff is proportional, not the graph"); + assert_eq!( + diff.mutations.len(), + 1, + "diff is proportional, not the graph" + ); match &diff.mutations[0].mutation { GraphMutation::NodeUpsert(node) => assert_eq!(node.id, "n:2"), _ => panic!("expected the missing node n:2"), @@ -532,7 +536,11 @@ mod tests { &mut store, StampedBatch::new([StampedMutation::new( GraphMutation::EdgeUpsert(EdgeRecord::new( - "e:new", "n:0", "LINK", "n:49", json!({}), + "e:new", + "n:0", + "LINK", + "n:49", + json!({}), )), Hlc::new(99, 0, ActorId::from_label("codex")), )]), diff --git a/crates/rustyred-server/src/yjs_sync.rs b/crates/rustyred-server/src/yjs_sync.rs index 2abb7eb..b8cad9b 100644 --- a/crates/rustyred-server/src/yjs_sync.rs +++ b/crates/rustyred-server/src/yjs_sync.rs @@ -592,7 +592,8 @@ mod tests { let before = server.transact().state_vector(); { let mut txn = server.transact_mut(); - txn.apply_update(Update::decode_v1(&update).unwrap()).unwrap(); + txn.apply_update(Update::decode_v1(&update).unwrap()) + .unwrap(); } let after = server.transact().state_vector(); sidecar.observe(&before, &after, actor, doc_client); From 11e740ba502b511f5f693b6d39ec1552ad704a11 Mon Sep 17 00:00:00 2001 From: Travis Gilbert <1travisgilbert@gmail.com> Date: Mon, 22 Jun 2026 21:49:46 -0400 Subject: [PATCH 3/3] chore(ci): satisfy workspace clippy --- crates/rustyred-core/src/graph_store.rs | 2 +- crates/rustyred-server/src/graph_sync.rs | 5 +++-- crates/rustyred-server/src/memory.rs | 15 +++++++-------- crates/rustyred-server/src/state.rs | 2 +- crates/rustyred-server/src/yjs_sync.rs | 9 +++++---- 5 files changed, 17 insertions(+), 16 deletions(-) diff --git a/crates/rustyred-core/src/graph_store.rs b/crates/rustyred-core/src/graph_store.rs index 0fea689..df881fe 100644 --- a/crates/rustyred-core/src/graph_store.rs +++ b/crates/rustyred-core/src/graph_store.rs @@ -2125,7 +2125,7 @@ impl RedCoreGraphStore { fn should_snapshot_for(&self, txn_id: u64) -> bool { let interval = self.options.snapshot_interval_writes; - interval > 0 && txn_id > self.snapshot_txn_id && txn_id % interval == 0 + interval > 0 && txn_id > self.snapshot_txn_id && txn_id.is_multiple_of(interval) } fn write_snapshot(&mut self) -> GraphStoreResult<()> { diff --git a/crates/rustyred-server/src/graph_sync.rs b/crates/rustyred-server/src/graph_sync.rs index baf18d9..4c2aa27 100644 --- a/crates/rustyred-server/src/graph_sync.rs +++ b/crates/rustyred-server/src/graph_sync.rs @@ -151,8 +151,9 @@ async fn handle_socket(state: AppState, tenant_id: String, room_id: String, sock } } Message::Ping(payload) => { - if sink.send(Message::Pong(payload)).await.is_err() { - break; + match sink.send(Message::Pong(payload)).await { + Ok(()) => {} + Err(_) => break, } } Message::Close(_) => break, diff --git a/crates/rustyred-server/src/memory.rs b/crates/rustyred-server/src/memory.rs index 3d22436..34c3be7 100644 --- a/crates/rustyred-server/src/memory.rs +++ b/crates/rustyred-server/src/memory.rs @@ -144,8 +144,10 @@ pub struct RecallBody { #[serde(default)] pub include_low_fitness: bool, #[serde(default)] + #[allow(dead_code)] pub include_consolidation_sources: bool, #[serde(default)] + #[allow(dead_code)] pub consume_handoffs: bool, } @@ -207,6 +209,7 @@ pub struct SelfArchiveBody { #[serde(default)] pub reason: Option, #[serde(default)] + #[allow(dead_code)] pub actor: Option, } @@ -347,9 +350,7 @@ fn strip_code(input: &str) -> String { } j += 1; } - for _ in i..j { - out.push(b' '); - } + out.resize(out.len() + (j - i), b' '); i = j; continue; } @@ -362,9 +363,7 @@ fn strip_code(input: &str) -> String { if j < bytes.len() && bytes[j] == b'`' { j += 1; } - for _ in i..j { - out.push(b' '); - } + out.resize(out.len() + (j - i), b' '); i = j; continue; } @@ -765,7 +764,7 @@ pub async fn memory_mentions_wait( let actor = body.actor.trim().to_string(); let limit = body.limit.unwrap_or(20).min(200); let timeout = body.timeout_seconds.unwrap_or(30).min(120); - let interval_ms = ((body.interval_seconds.unwrap_or(1.0)).max(0.1).min(5.0) * 1000.0) as u64; + let interval_ms = ((body.interval_seconds.unwrap_or(1.0)).clamp(0.1, 5.0) * 1000.0) as u64; let started = std::time::Instant::now(); let deadline = std::time::Duration::from_secs(timeout); @@ -1207,7 +1206,7 @@ pub async fn memory_self_revise( let properties = Value::Object(props); // Use prior labels (preserves verb kind) plus marker. - let mut labels: Vec = prior.labels.iter().cloned().collect(); + let mut labels: Vec = prior.labels.to_vec(); if !labels.iter().any(|l| l == "MemoryAtom") { labels.insert(0, "MemoryAtom".to_string()); } diff --git a/crates/rustyred-server/src/state.rs b/crates/rustyred-server/src/state.rs index 9d38aa1..88b225e 100644 --- a/crates/rustyred-server/src/state.rs +++ b/crates/rustyred-server/src/state.rs @@ -273,7 +273,7 @@ impl AppState { if let (Some(lat), Some(lon)) = (lat, lon) { let _ = SpatialBackend::upsert(index.as_mut(), &node.id, lat, lon); } else { - let _ = SpatialBackend::remove(index.as_mut(), &node.id); + SpatialBackend::remove(index.as_mut(), &node.id); } } } diff --git a/crates/rustyred-server/src/yjs_sync.rs b/crates/rustyred-server/src/yjs_sync.rs index b8cad9b..a63d137 100644 --- a/crates/rustyred-server/src/yjs_sync.rs +++ b/crates/rustyred-server/src/yjs_sync.rs @@ -277,8 +277,9 @@ async fn handle_socket( } } Message::Ping(payload) => { - if sink.send(Message::Pong(payload)).await.is_err() { - break; + match sink.send(Message::Pong(payload)).await { + Ok(()) => {} + Err(_) => break, } } Message::Close(_) => break, @@ -457,6 +458,7 @@ impl ProvenanceSidecar { } /// Blame: which actor authored the item carrying this yrs `client_id`. + #[cfg(test)] pub(crate) fn writer_of(&self, client_id: u64) -> Option<&str> { self.clients.get(&client_id).map(|a| a.actor.as_str()) } @@ -482,8 +484,7 @@ mod tests { fn doc_text(doc: &Doc) -> String { let text = doc.get_or_insert_text("t"); let txn = doc.transact(); - let content = text.get_string(&txn); - content + text.get_string(&txn) } #[test]