use super::*;
impl LocalJsonMemoryStoreAdapter {
pub(super) fn from_config(
config: &LocalJsonMemoryBackendConfig,
) -> Result<Self, LocalJsonMemoryStoreError> {
let health_check_endpoint = match config.health_check_endpoint.as_deref() {
None => None,
Some(raw) => {
let endpoint = raw.trim();
if endpoint.is_empty() {
return Err(LocalJsonMemoryStoreError::InvalidConfig(
"memory backend health_check_endpoint must not be empty when provided"
.to_string(),
));
}
Some(endpoint.to_string())
}
};
let state_path = config.state_path.trim();
if state_path.is_empty() {
return Err(LocalJsonMemoryStoreError::InvalidConfig(
"memory backend state_path must not be empty".to_string(),
));
}
Ok(Self {
health_check_endpoint,
state_path: PathBuf::from(state_path),
})
}
fn ensure_remote_health(&self) -> Result<(), LocalJsonMemoryStoreError> {
let Some(endpoint) = self.health_check_endpoint.as_deref() else {
return Ok(());
};
let health_url = format!("{}/v1/health", endpoint.trim_end_matches('/'));
let parsed = parse_http_url(&health_url)?;
let authority = format!("{}:{}", parsed.host, parsed.port);
let mut addrs = authority.to_socket_addrs().map_err(|err| {
LocalJsonMemoryStoreError::ExternalCallFailed(format!(
"healthcheck resolve failed for '{health_url}': {err}"
))
})?;
let addr = addrs.next().ok_or_else(|| {
LocalJsonMemoryStoreError::ExternalCallFailed(format!(
"healthcheck resolve failed for '{health_url}': no socket addresses"
))
})?;
let mut stream = TcpStream::connect_timeout(&addr, MEMORY_LEDGER_HEALTHCHECK_TIMEOUT)
.map_err(|err| {
LocalJsonMemoryStoreError::ExternalCallFailed(format!(
"healthcheck connect failed for '{health_url}': {err}"
))
})?;
stream
.set_read_timeout(Some(MEMORY_LEDGER_HEALTHCHECK_TIMEOUT))
.map_err(|err| {
LocalJsonMemoryStoreError::ExternalCallFailed(format!(
"healthcheck timeout setup failed for '{health_url}': {err}"
))
})?;
stream
.set_write_timeout(Some(MEMORY_LEDGER_HEALTHCHECK_TIMEOUT))
.map_err(|err| {
LocalJsonMemoryStoreError::ExternalCallFailed(format!(
"healthcheck timeout setup failed for '{health_url}': {err}"
))
})?;
let request = format!(
"GET {} HTTP/1.1\r\nHost: {}\r\nConnection: close\r\n\r\n",
parsed.path, parsed.host
);
stream.write_all(request.as_bytes()).map_err(|err| {
LocalJsonMemoryStoreError::ExternalCallFailed(format!(
"healthcheck write failed for '{health_url}': {err}"
))
})?;
let mut reader = BufReader::new(stream);
let mut status_line = String::new();
let bytes_read = reader.read_line(&mut status_line).map_err(|err| {
LocalJsonMemoryStoreError::ExternalCallFailed(format!(
"healthcheck read failed for '{health_url}': {err}"
))
})?;
if bytes_read == 0 {
return Err(LocalJsonMemoryStoreError::ExternalCallFailed(format!(
"healthcheck read failed for '{health_url}': empty response"
)));
}
let status_code = status_line
.split_whitespace()
.nth(1)
.and_then(|value| value.parse::<u16>().ok())
.ok_or_else(|| {
LocalJsonMemoryStoreError::ExternalCallFailed(format!(
"healthcheck parse failed for '{health_url}': invalid status line '{}'",
status_line.trim()
))
})?;
if (200..300).contains(&status_code) {
Ok(())
} else {
Err(LocalJsonMemoryStoreError::ExternalCallFailed(format!(
"healthcheck status failed for '{health_url}': HTTP {status_code}"
)))
}
}
pub(super) fn read_state(&self) -> Result<PersistedMemoryState, LocalJsonMemoryStoreError> {
self.ensure_remote_health()?;
if !self.state_path.exists() {
return Ok(PersistedMemoryState::default());
}
let bytes = fs::read(&self.state_path)
.map_err(|err| LocalJsonMemoryStoreError::Io(err.to_string()))?;
serde_json::from_slice::<PersistedMemoryState>(&bytes)
.map_err(|err| LocalJsonMemoryStoreError::InvalidStoreData(err.to_string()))
}
fn write_state(&self, state: &PersistedMemoryState) -> Result<(), LocalJsonMemoryStoreError> {
self.ensure_remote_health()?;
if let Some(parent) = self.state_path.parent() {
fs::create_dir_all(parent)
.map_err(|err| LocalJsonMemoryStoreError::Io(err.to_string()))?;
}
let tmp_path = self.state_path.with_extension("tmp");
let json = serde_json::to_vec_pretty(state)
.map_err(|err| LocalJsonMemoryStoreError::Serialize(err.to_string()))?;
fs::write(&tmp_path, json).map_err(|err| LocalJsonMemoryStoreError::Io(err.to_string()))?;
fs::rename(&tmp_path, &self.state_path)
.map_err(|err| LocalJsonMemoryStoreError::Io(err.to_string()))?;
Ok(())
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct ParsedHttpUrl {
host: String,
port: u16,
path: String,
}
fn parse_http_url(url: &str) -> Result<ParsedHttpUrl, LocalJsonMemoryStoreError> {
let trimmed = url.trim();
let without_scheme = trimmed.strip_prefix("http://").ok_or_else(|| {
LocalJsonMemoryStoreError::InvalidConfig(format!(
"memory backend endpoint must start with http:// (got '{trimmed}')"
))
})?;
if without_scheme.is_empty() {
return Err(LocalJsonMemoryStoreError::InvalidConfig(
"memory backend endpoint host must not be empty".to_string(),
));
}
let (authority, path_suffix) = without_scheme
.split_once('/')
.map(|(left, right)| (left, format!("/{right}")))
.unwrap_or((without_scheme, "/".to_string()));
if authority.is_empty() {
return Err(LocalJsonMemoryStoreError::InvalidConfig(
"memory backend endpoint host must not be empty".to_string(),
));
}
let (host, port) = match authority.rsplit_once(':') {
Some((host, raw_port))
if !host.is_empty() && raw_port.chars().all(|c| c.is_ascii_digit()) =>
{
let parsed = raw_port.parse::<u16>().map_err(|_| {
LocalJsonMemoryStoreError::InvalidConfig(format!(
"memory backend endpoint port is invalid in '{trimmed}'"
))
})?;
(host.to_string(), parsed)
}
_ => (authority.to_string(), 80_u16),
};
if host.is_empty() {
return Err(LocalJsonMemoryStoreError::InvalidConfig(
"memory backend endpoint host must not be empty".to_string(),
));
}
Ok(ParsedHttpUrl {
host,
port,
path: path_suffix,
})
}
impl MobkitRuntimeHandle {
fn next_memory_sequence(&mut self) -> u64 {
Self::next_sequence(&mut self.memory_sequence)
}
pub(super) fn canonical_memory_token(raw: &str) -> Option<String> {
let token = raw.trim().to_ascii_lowercase();
if token.is_empty() { None } else { Some(token) }
}
pub(super) fn canonical_memory_store(raw: &str) -> Option<String> {
let store = Self::canonical_memory_token(raw)?;
if MEMORY_SUPPORTED_STORES.contains(&store.as_str()) {
Some(store)
} else {
None
}
}
fn default_memory_store() -> String {
"knowledge_graph".to_string()
}
pub(super) fn memory_conflict_for_reference(
&self,
entity: Option<&str>,
topic: Option<&str>,
) -> Option<MemoryConflictSignal> {
let canonical_entity = entity.and_then(Self::canonical_memory_token);
let canonical_topic = topic.and_then(Self::canonical_memory_token);
match (canonical_entity, canonical_topic) {
(Some(entity), Some(topic)) => self
.memory_conflicts
.values()
.find(|signal| signal.entity == entity && signal.topic == topic)
.cloned(),
(Some(entity), None) => self
.memory_conflicts
.values()
.find(|signal| signal.entity == entity)
.cloned(),
(None, Some(topic)) => self
.memory_conflicts
.values()
.find(|signal| signal.topic == topic)
.cloned(),
(None, None) => None,
}
}
pub fn memory_stores(&self) -> Vec<MemoryStoreInfo> {
MEMORY_SUPPORTED_STORES
.iter()
.map(|store| MemoryStoreInfo {
store: (*store).to_string(),
record_count: self
.memory_assertions
.iter()
.filter(|assertion| assertion.store == *store)
.count()
+ self
.memory_conflicts
.values()
.filter(|signal| signal.store == *store)
.count(),
})
.collect()
}
fn persist_memory_state(&self) -> Result<(), MemoryIndexError> {
let Some(backend) = self.memory_backend.as_ref() else {
return Ok(());
};
let state = PersistedMemoryState {
assertions: self.memory_assertions.clone(),
conflicts: self.memory_conflicts.values().cloned().collect::<Vec<_>>(),
};
backend
.write_state(&state)
.map_err(MemoryIndexError::BackendPersistFailed)
}
pub fn memory_index(
&mut self,
request: MemoryIndexRequest,
) -> Result<MemoryIndexResult, MemoryIndexError> {
let entity = Self::canonical_memory_token(&request.entity)
.ok_or(MemoryIndexError::EntityRequired)?;
let topic =
Self::canonical_memory_token(&request.topic).ok_or(MemoryIndexError::TopicRequired)?;
let store = match request.store.as_deref() {
None => Self::default_memory_store(),
Some(raw_store) => Self::canonical_memory_store(raw_store)
.ok_or_else(|| MemoryIndexError::UnsupportedStore(raw_store.trim().to_string()))?,
};
let fact = request
.fact
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToString::to_string);
let conflict = request.conflict.unwrap_or(false);
if fact.is_none() && !conflict {
return Err(MemoryIndexError::FactRequiredWhenConflictUnset);
}
let previous_memory_assertions = self.memory_assertions.clone();
let previous_memory_conflicts = self.memory_conflicts.clone();
let previous_memory_sequence = self.memory_sequence;
let mut assertion_id = None;
if let Some(fact) = fact {
let assertion_sequence = self.next_memory_sequence();
let assertion = MemoryAssertion {
assertion_id: format!("memory-assert-{assertion_sequence:06}"),
entity: entity.clone(),
topic: topic.clone(),
store: store.clone(),
fact,
metadata: request.metadata.clone(),
indexed_at_ms: current_time_ms(),
};
assertion_id = Some(assertion.assertion_id.clone());
self.memory_assertions.push(assertion);
while self.memory_assertions.len() > MEMORY_ASSERTIONS_MAX_RETAINED {
self.memory_assertions.remove(0);
}
}
if conflict {
let conflict_key = MemoryConflictKey {
entity: entity.clone(),
topic: topic.clone(),
store: store.clone(),
};
self.memory_conflicts.insert(
conflict_key,
MemoryConflictSignal {
entity: entity.clone(),
topic: topic.clone(),
store: store.clone(),
reason: request
.conflict_reason
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToString::to_string),
updated_at_ms: current_time_ms(),
},
);
}
if let Err(error) = self.persist_memory_state() {
self.memory_assertions = previous_memory_assertions;
self.memory_conflicts = previous_memory_conflicts;
self.memory_sequence = previous_memory_sequence;
return Err(error);
}
let conflict_active = self
.memory_conflict_for_reference(Some(entity.as_str()), Some(topic.as_str()))
.is_some();
Ok(MemoryIndexResult {
entity,
topic,
store,
assertion_id,
conflict_active,
})
}
pub fn memory_query(&self, request: MemoryQueryRequest) -> MemoryQueryResult {
let entity = request
.entity
.as_deref()
.and_then(Self::canonical_memory_token);
let topic = request
.topic
.as_deref()
.and_then(Self::canonical_memory_token);
let store = request
.store
.as_deref()
.and_then(Self::canonical_memory_store);
let query = request
.query
.as_deref()
.and_then(Self::canonical_memory_token);
let assertion_matches_query = |assertion: &MemoryAssertion| {
query.as_ref().is_none_or(|needle| {
assertion.entity.contains(needle.as_str())
|| assertion.topic.contains(needle.as_str())
|| assertion
.fact
.to_ascii_lowercase()
.contains(needle.as_str())
})
};
let conflict_matches_query = |signal: &MemoryConflictSignal| {
query.as_ref().is_none_or(|needle| {
signal.entity.contains(needle.as_str())
|| signal.topic.contains(needle.as_str())
|| signal
.reason
.as_deref()
.is_some_and(|reason| reason.to_ascii_lowercase().contains(needle.as_str()))
})
};
let assertions = self
.memory_assertions
.iter()
.filter(|assertion| {
entity
.as_ref()
.is_none_or(|value| assertion.entity.as_str() == value.as_str())
})
.filter(|assertion| {
topic
.as_ref()
.is_none_or(|value| assertion.topic.as_str() == value.as_str())
})
.filter(|assertion| {
store
.as_ref()
.is_none_or(|value| assertion.store.as_str() == value.as_str())
})
.filter(|assertion| assertion_matches_query(assertion))
.cloned()
.collect::<Vec<_>>();
let conflicts = self
.memory_conflicts
.values()
.filter(|signal| {
entity
.as_ref()
.is_none_or(|value| signal.entity.as_str() == value.as_str())
})
.filter(|signal| {
topic
.as_ref()
.is_none_or(|value| signal.topic.as_str() == value.as_str())
})
.filter(|signal| {
store
.as_ref()
.is_none_or(|value| signal.store.as_str() == value.as_str())
})
.filter(|signal| conflict_matches_query(signal))
.cloned()
.collect::<Vec<_>>();
MemoryQueryResult {
assertions,
conflicts,
}
}
}
#[cfg(test)]
#[allow(clippy::expect_used)]
mod tests {
use super::*;
#[test]
fn local_json_adapter_roundtrips_state_without_health_endpoint() {
let temp = tempfile::tempdir().expect("temp dir");
let state_path = temp.path().join("memory-ledger-state.json");
let adapter = LocalJsonMemoryStoreAdapter::from_config(&LocalJsonMemoryBackendConfig {
state_path: state_path.to_string_lossy().to_string(),
health_check_endpoint: None,
})
.expect("adapter from config");
let state = PersistedMemoryState {
assertions: vec![MemoryAssertion {
assertion_id: "memory-assert-000001".to_string(),
entity: "delivery".to_string(),
topic: "email_send".to_string(),
store: "todo".to_string(),
fact: "double-check recipient consent".to_string(),
metadata: None,
indexed_at_ms: 1,
}],
conflicts: vec![],
};
adapter.write_state(&state).expect("write state");
let reloaded = adapter.read_state().expect("read state");
assert_eq!(reloaded, state);
}
#[test]
fn local_json_adapter_rejects_empty_health_endpoint() {
let error = LocalJsonMemoryStoreAdapter::from_config(&LocalJsonMemoryBackendConfig {
state_path: "/tmp/state.json".to_string(),
health_check_endpoint: Some(" ".to_string()),
})
.expect_err("empty health endpoint must be rejected");
assert!(matches!(error, LocalJsonMemoryStoreError::InvalidConfig(_)));
}
#[test]
fn legacy_elephant_config_maps_endpoint_to_health_check() {
let legacy = ElephantMemoryBackendConfig {
endpoint: "http://localhost:3000".to_string(),
state_path: "/tmp/state.json".to_string(),
};
let config = LocalJsonMemoryBackendConfig::from(legacy);
assert_eq!(
config.health_check_endpoint.as_deref(),
Some("http://localhost:3000")
);
assert_eq!(config.state_path, "/tmp/state.json");
}
#[test]
fn memory_backend_config_accepts_legacy_elephant_wire_shape() {
let parsed: MemoryBackendConfig = serde_json::from_value(serde_json::json!({
"kind": "elephant",
"endpoint": "http://localhost:3000",
"state_path": "/tmp/state.json"
}))
.expect("legacy wire shape deserializes");
assert_eq!(
parsed,
MemoryBackendConfig::Elephant(ElephantMemoryBackendConfig {
endpoint: "http://localhost:3000".to_string(),
state_path: "/tmp/state.json".to_string(),
})
);
let parsed: MemoryBackendConfig = serde_json::from_value(serde_json::json!({
"kind": "local_json",
"state_path": "/tmp/state.json"
}))
.expect("local_json wire shape deserializes");
assert_eq!(
parsed,
MemoryBackendConfig::LocalJson(LocalJsonMemoryBackendConfig {
state_path: "/tmp/state.json".to_string(),
health_check_endpoint: None,
})
);
}
}