use std::fs::{File, OpenOptions};
use std::io::{BufWriter, Read, Write};
use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
use crate::core::entangled::{hash_u64, EntangledHVec};
const DEFAULT_DECAY_NUM: i64 = 7;
const DEFAULT_DECAY_DEN: i64 = 8;
const DENSITY_DENOM: usize = 256;
const W_STORE: i64 = 1;
const W_STRENGTHEN: i64 = 2;
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum Event {
Store(String, String, String),
Strengthen(String, String, String),
Decay,
}
#[derive(Clone, Copy, Debug)]
pub struct GraphConfig {
pub strengthen_on_read: bool,
pub decay_interval: usize,
pub decay_num: i64,
pub decay_den: i64,
}
impl Default for GraphConfig {
fn default() -> Self {
Self {
strengthen_on_read: true,
decay_interval: 16,
decay_num: DEFAULT_DECAY_NUM,
decay_den: DEFAULT_DECAY_DEN,
}
}
}
pub struct ConnectionGraph {
dim: usize,
field: Vec<i64>,
events: Vec<Event>,
chain: u64,
cfg: GraphConfig,
queries_since_decay: usize,
persist: Option<BufWriter<File>>,
}
impl ConnectionGraph {
pub fn new(dim: usize) -> Self {
Self::with_config(dim, GraphConfig::default())
}
pub fn with_config(dim: usize, cfg: GraphConfig) -> Self {
assert!(dim > 0, "connection graph dim must be non-zero");
Self {
dim,
field: vec![0i64; dim],
events: Vec::new(),
chain: 0,
cfg,
queries_since_decay: 0,
persist: None,
}
}
pub fn open<P: AsRef<Path>>(dim: usize, cfg: GraphConfig, path: P) -> std::io::Result<Self> {
assert!(dim > 0, "connection graph dim must be non-zero");
let path: PathBuf = path.as_ref().to_path_buf();
let mut g = Self::with_config(dim, cfg);
if path.exists() {
let events = read_events(&path)?;
for ev in &events {
g.apply(ev);
g.chain = chain_step(g.chain, ev);
}
g.events = events;
}
let file = OpenOptions::new().create(true).append(true).open(&path)?;
g.persist = Some(BufWriter::new(file));
Ok(g)
}
fn entity_vec(&self, id: &str) -> EntangledHVec {
EntangledHVec::new_with_density(self.dim, DENSITY_DENOM, str_seed(id))
}
fn masks(&self, relation: &str) -> (u32, u32) {
let s = str_seed(relation);
let subj = (hash_u64(s, 0x5B_1EC7) % self.dim as u64) as u32 | 1;
let obj = (hash_u64(s, 0x0B_1EC7) % self.dim as u64) as u32 | 1;
(subj, obj)
}
fn edge_indices(&self, subject: &str, relation: &str, object: &str) -> Vec<u32> {
let (subj_mask, obj_mask) = self.masks(relation);
let s = self.entity_vec(subject);
let o = self.entity_vec(object);
let mut v: Vec<u32> = s.indices().iter().map(|&i| i ^ subj_mask).collect();
v.extend(o.indices().iter().map(|&i| i ^ obj_mask));
v.sort_unstable();
v.dedup();
v
}
fn fold_edge(&mut self, indices: &[u32], w: i64) {
for &i in indices {
self.field[i as usize] += w;
}
}
fn apply_decay(&mut self) {
for v in self.field.iter_mut() {
*v = *v * self.cfg.decay_num / self.cfg.decay_den;
}
}
fn apply(&mut self, ev: &Event) {
match ev {
Event::Store(s, r, o) => {
let idx = self.edge_indices(s, r, o);
self.fold_edge(&idx, W_STORE);
}
Event::Strengthen(s, r, o) => {
let idx = self.edge_indices(s, r, o);
self.fold_edge(&idx, W_STRENGTHEN);
}
Event::Decay => self.apply_decay(),
}
}
fn record(&mut self, ev: Event) {
self.chain = chain_step(self.chain, &ev);
if let Some(writer) = self.persist.as_mut() {
if let Err(e) = append_event(writer, &ev) {
tracing::warn!("connection graph persist append failed: {e}");
}
}
self.events.push(ev);
}
pub fn store(&mut self, subject: &str, relation: &str, object: &str) {
let idx = self.edge_indices(subject, relation, object);
self.fold_edge(&idx, W_STORE);
self.record(Event::Store(
subject.to_string(),
relation.to_string(),
object.to_string(),
));
}
pub fn score(&self, subject: &str, relation: &str, object: &str) -> f64 {
let idx = self.edge_indices(subject, relation, object);
self.score_indices(&idx)
}
fn score_indices(&self, idx: &[u32]) -> f64 {
if idx.is_empty() {
return 0.0;
}
let sum: i64 = idx.iter().map(|&i| self.field[i as usize]).sum();
(sum as f64 / idx.len() as f64) - self.global_mean()
}
fn global_mean(&self) -> f64 {
self.field.iter().sum::<i64>() as f64 / self.field.len() as f64
}
fn is_present(&self, idx: &[u32]) -> bool {
if idx.is_empty() {
return false;
}
let edge_sum = idx.iter().map(|&i| self.field[i as usize]).sum::<i64>() as i128;
let field_total = self.field.iter().sum::<i64>() as i128;
let edge_len = idx.len() as i128;
let field_len = self.field.len() as i128;
2 * (edge_sum * field_len - field_total * edge_len)
>= (W_STORE as i128) * edge_len * field_len
}
pub fn query(&mut self, subject: &str, relation: &str, object: &str) -> f64 {
let idx = self.edge_indices(subject, relation, object);
let s = self.score_indices(&idx);
if self.cfg.strengthen_on_read && self.is_present(&idx) {
self.fold_edge(&idx, W_STRENGTHEN);
self.record(Event::Strengthen(
subject.to_string(),
relation.to_string(),
object.to_string(),
));
if self.cfg.decay_interval > 0 {
self.queries_since_decay += 1;
if self.queries_since_decay >= self.cfg.decay_interval {
self.apply_decay();
self.record(Event::Decay);
self.queries_since_decay = 0;
}
}
}
s
}
pub fn chain_digest(&self) -> u64 {
self.chain
}
pub fn events(&self) -> &[Event] {
&self.events
}
pub fn replay(dim: usize, cfg: GraphConfig, events: &[Event]) -> (Vec<i64>, u64) {
let mut g = Self::with_config(dim, cfg);
for ev in events {
g.apply(ev);
g.chain = chain_step(g.chain, ev);
}
(g.field, g.chain)
}
pub fn verify(&self) -> bool {
let (field, chain) = Self::replay(self.dim, self.cfg, &self.events);
field == self.field && chain == self.chain
}
}
fn str_seed(s: &str) -> u64 {
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
for &b in s.as_bytes() {
h ^= b as u64;
h = h.wrapping_mul(0x0000_0100_0000_01B3);
}
hash_u64(h, s.len() as u64)
}
fn chain_step(prev: u64, ev: &Event) -> u64 {
let (tag, payload) = match ev {
Event::Store(s, r, o) => (0x5701u64, str_seed(&format!("{s}\u{1}{r}\u{1}{o}"))),
Event::Strengthen(s, r, o) => (0x5731u64, str_seed(&format!("{s}\u{1}{r}\u{1}{o}"))),
Event::Decay => (0xDECAu64, 0),
};
hash_u64(prev ^ tag, payload)
}
fn append_event(writer: &mut BufWriter<File>, ev: &Event) -> std::io::Result<()> {
let bytes = bincode::serde::encode_to_vec(ev, bincode::config::standard())
.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
writer.write_all(&(bytes.len() as u32).to_le_bytes())?;
writer.write_all(&bytes)?;
writer.flush()
}
fn read_events(path: &Path) -> std::io::Result<Vec<Event>> {
let mut buf = Vec::new();
File::open(path)?.read_to_end(&mut buf)?;
let mut events = Vec::new();
let mut off = 0;
while off + 4 <= buf.len() {
let len = u32::from_le_bytes([buf[off], buf[off + 1], buf[off + 2], buf[off + 3]]) as usize;
off += 4;
if off + len > buf.len() {
break; }
match bincode::serde::decode_from_slice::<Event, _>(
&buf[off..off + len],
bincode::config::standard(),
) {
Ok((ev, _)) => events.push(ev),
Err(_) => break, }
off += len;
}
Ok(events)
}
#[cfg(test)]
mod tests {
use super::*;
fn no_plasticity() -> GraphConfig {
GraphConfig {
strengthen_on_read: false,
decay_interval: 0,
..GraphConfig::default()
}
}
#[test]
fn stored_scores_above_absent() {
let mut g = ConnectionGraph::with_config(4096, no_plasticity());
g.store("paris", "capital_of", "france");
let present = g.score("paris", "capital_of", "france");
let absent = g.score("berlin", "capital_of", "spain");
assert!(present > absent, "present {present} !> absent {absent}");
}
#[test]
fn persists_and_reloads_across_restart() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("cg.log");
{
let mut g = ConnectionGraph::open(4096, no_plasticity(), &path).unwrap();
g.store("paris", "capital_of", "france");
g.store("berlin", "capital_of", "germany");
}
let g = ConnectionGraph::open(4096, no_plasticity(), &path).unwrap();
assert!(
g.verify(),
"reopened graph must equal the replay of its log"
);
let present = g.score("paris", "capital_of", "france");
let absent = g.score("tokyo", "capital_of", "spain");
assert!(
present > absent,
"reloaded relation ({present}) must outscore an absent one ({absent})"
);
}
#[test]
fn replay_is_bit_exact_and_tamper_evident() {
let mut g = ConnectionGraph::new(4096);
g.store("a", "r", "b");
g.store("c", "r", "d");
g.query("a", "r", "b");
g.query("a", "r", "b");
assert!(g.verify(), "replay of own log must reproduce state exactly");
let mut tampered = g.events().to_vec();
tampered.remove(0);
let (_, chain) = ConnectionGraph::replay(4096, GraphConfig::default(), &tampered);
assert_ne!(
chain,
g.chain_digest(),
"tamper must change the chain digest"
);
}
#[test]
fn plasticity_holds_discrimination_under_load() {
let dim = 8192;
let load = 400; let hot = ("A", "knows", "B");
let absent = ("A", "knows", "Z");
let mut stat = ConnectionGraph::with_config(dim, no_plasticity());
let mut plas = ConnectionGraph::new(dim);
stat.store(hot.0, hot.1, hot.2);
plas.store(hot.0, hot.1, hot.2);
for i in 0..load {
let s = format!("s{i}");
let o = format!("o{i}");
stat.store(&s, "knows", &o);
plas.store(&s, "knows", &o);
}
for _ in 0..80 {
plas.query(hot.0, hot.1, hot.2);
}
let stat_margin =
stat.score(hot.0, hot.1, hot.2) - stat.score(absent.0, absent.1, absent.2);
let plas_margin =
plas.score(hot.0, hot.1, hot.2) - plas.score(absent.0, absent.1, absent.2);
assert!(
plas_margin > stat_margin,
"plastic margin {plas_margin} must exceed static {stat_margin} under load"
);
}
#[test]
fn generalizes_through_shared_substructure() {
let dim = 8192;
let mut g = ConnectionGraph::new(dim);
g.store("A", "knows", "B");
for _ in 0..40 {
g.query("A", "knows", "B");
}
let shared = g.score("A", "knows", "NEVER_STORED"); let unrelated = g.score("X", "likes", "Y"); assert!(
shared > unrelated,
"shared-substructure {shared} must exceed unrelated {unrelated} (else it is a cache)"
);
}
#[test]
fn forgetting_is_real() {
let dim = 8192;
let mut g = ConnectionGraph::new(dim);
g.store("cold", "rel", "thing");
let fresh = g.score("cold", "rel", "thing");
for i in 0..20 {
g.store(&format!("hot{i}"), "rel", "x");
}
for _ in 0..40 {
for i in 0..20 {
g.query(&format!("hot{i}"), "rel", "x");
}
}
let after = g.score("cold", "rel", "thing");
assert!(
after < fresh,
"un-queried relation must decay: after {after} !< fresh {fresh}"
);
}
#[test]
fn absent_query_does_not_confabulate() {
let dim = 8192;
let mut g = ConnectionGraph::new(dim);
g.store("paris", "capital_of", "france");
let events_before = g.events().len();
for _ in 0..50 {
g.query("zzz", "unrelated_rel", "qqq");
}
assert_eq!(
g.events().len(),
events_before,
"absent queries must not mutate the store"
);
assert!(
g.score("zzz", "unrelated_rel", "qqq") < g.score("paris", "capital_of", "france"),
"the absent relation must stay absent after being queried"
);
g.query("paris", "capital_of", "france");
assert!(
g.events().len() > events_before,
"present query must reinforce"
);
}
}