Skip to main content

wm_memory/
cold_storage.rs

1//! Non-Destructive Phagic Digestive Cold-Storage Subsystem.
2//!
3//! # Sacred Rule (the Law of Non-Destructive Phagocytosis)
4//! Phagic systems must NEVER delete anything. Irrelevant or distant "outer rim" memories
5//! in each galaxy are gently migrated into compressed cold storage, preserving the complete
6//! lineage, provenance, vector clocks, and retrievability.
7//!
8//! # Architecture
9//! - **Outer-Rim Metric**: Calculates distance $D(m) \in [0.0, 1.0]$ from the galactic core as:
10//!   $D(m) = f(\text{age}, \text{access\_frequency}, \text{harmonic\_resonance}, \text{emotional\_valence}, \text{importance\_weight})$
11//! - **Hot Tier**: Active memories in `MemoryStore` and Tantivy `SearchEngine`.
12//! - **Cold Tier**: Compressed archive (MessagePack + Gzip/Deflate), preserving all metadata,
13//!   vector clocks, content, embeddings, and association graph links.
14//! - **Digestion & Condensation**: Synthesizes clusters of outer-rim memories into distilled
15//!   thematic nodes (`Tier::Semantic`, `MemoryType::Symbolic`) in the active hot tier,
16//!   referencing the cold-stored records, while full original memories rest safely in cold storage.
17//! - **Thawing (Warm Restoration)**: Zero data loss — any cold-stored memory can be queried
18//!   directly or thawed back into the hot active tier with full fidelity.
19
20use chrono::{DateTime, Utc};
21use serde::{Deserialize, Serialize};
22use std::collections::HashMap;
23use uuid::Uuid;
24use wm_core::{CoreError, Galaxy, Result};
25
26use crate::memory::{Memory, MemoryId, MemoryType, Tier};
27use crate::search::SearchEngine;
28use crate::store::MemoryStore;
29
30// ── Compression Codec ───────────────────────────────────────────────────
31
32/// Compression algorithm used for cold storage payloads.
33#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Default)]
34#[serde(rename_all = "snake_case")]
35pub enum CompressionCodec {
36    /// Gzip compression (balanced ratio and high compatibility)
37    #[default]
38    Gzip,
39    /// Raw deflate compression
40    Deflate,
41    /// Uncompressed MessagePack binary representation
42    Msgpack,
43}
44
45impl CompressionCodec {
46    /// String label for display/JSON.
47    #[must_use]
48    pub const fn as_str(self) -> &'static str {
49        match self {
50            Self::Gzip => "gzip",
51            Self::Deflate => "deflate",
52            Self::Msgpack => "msgpack",
53        }
54    }
55}
56
57// ── Outer-Rim Distance Metric ───────────────────────────────────────────
58
59/// Breakdown of the factors contributing to a memory's outer-rim distance.
60#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
61pub struct OuterRimFactors {
62    /// Age component in [0.0, 1.0]: 1.0 = ancient / past half-life.
63    pub age_factor: f32,
64    /// Access frequency component in [0.0, 1.0]: 1.0 = unaccessed, 0.0 = frequent.
65    pub access_factor: f32,
66    /// Harmonic resonance component in [0.0, 1.0]: 1.0 = zero resonance, 0.0 = peak resonance.
67    pub resonance_factor: f32,
68    /// Emotional valence & salience in [0.0, 1.0]: 1.0 = emotionally flat, 0.0 = charged.
69    pub emotional_factor: f32,
70    /// Importance component in [0.0, 1.0]: 1.0 = low importance, 0.0 = max importance.
71    pub importance_factor: f32,
72    /// Composite distance in [0.0, 1.0]: 0.0 = core, 1.0 = deep outer rim.
73    pub distance: f32,
74}
75
76/// Configuration parameters for the outer-rim metric and digestion process.
77#[derive(Debug, Clone, Serialize, Deserialize)]
78pub struct PhagicConfig {
79    /// Weight for age factor (default 0.25).
80    pub weight_age: f32,
81    /// Weight for access frequency factor (default 0.25).
82    pub weight_access: f32,
83    /// Weight for harmonic resonance factor (default 0.15).
84    pub weight_resonance: f32,
85    /// Weight for emotional valence/weight factor (default 0.15).
86    pub weight_emotional: f32,
87    /// Weight for importance factor (default 0.20).
88    pub weight_importance: f32,
89    /// Distance threshold above which memories qualify for the outer rim (default 0.70).
90    pub outer_rim_threshold: f32,
91    /// Maximum memories digested in a single pass per galaxy (default 50).
92    pub max_digest_batch: usize,
93    /// Minimum memories in a cluster to form a distilled thematic node (default 2).
94    pub min_cluster_size: usize,
95    /// Preferred compression codec (default Gzip).
96    pub compression_codec: CompressionCodec,
97}
98
99impl Default for PhagicConfig {
100    fn default() -> Self {
101        Self {
102            weight_age: 0.25,
103            weight_access: 0.25,
104            weight_resonance: 0.15,
105            weight_emotional: 0.15,
106            weight_importance: 0.20,
107            outer_rim_threshold: 0.70,
108            max_digest_batch: 50,
109            min_cluster_size: 2,
110            compression_codec: CompressionCodec::Gzip,
111        }
112    }
113}
114
115/// Calculate the outer-rim distance $D(m) \in [0.0, 1.0]$ for a memory.
116///
117/// Protected memories (`is_protected == true`) are anchored at the galactic core
118/// and always receive a distance of 0.0.
119#[must_use]
120pub fn calculate_outer_rim_distance(
121    mem: &Memory,
122    config: &PhagicConfig,
123    now: DateTime<Utc>,
124) -> OuterRimFactors {
125    if mem.metadata.is_protected {
126        return OuterRimFactors {
127            age_factor: 0.0,
128            access_factor: 0.0,
129            resonance_factor: 0.0,
130            emotional_factor: 0.0,
131            importance_factor: 0.0,
132            distance: 0.0,
133        };
134    }
135
136    // 1. Age factor: days elapsed relative to configurable half_life_days
137    let seconds_since_access = (now - mem.metadata.accessed_at).num_seconds().max(0);
138    let days_since = seconds_since_access as f32 / 86400.0;
139    let half_life = mem.metadata.half_life_days.max(1.0);
140    let age_factor = (1.0 - 0.5f32.powf(days_since / half_life)).clamp(0.0, 1.0);
141
142    // 2. Access frequency: hyperbolic decay over combined access + recall reads
143    let reads = (mem.metadata.access_count + mem.metadata.recall_count) as f32;
144    let access_factor = (1.0 / 0.5f32.mul_add(reads, 1.0)).clamp(0.0, 1.0);
145
146    // 3. Harmonic resonance: inverse of neuro_score
147    let resonance_factor = (1.0 - mem.metadata.neuro_score.clamp(0.0, 1.0)).clamp(0.0, 1.0);
148
149    // 4. Emotional valence / salience: |valence| * weight
150    let salience =
151        (mem.metadata.emotional_valence.abs() * mem.metadata.emotional_weight).clamp(0.0, 1.0);
152    let emotional_factor = (1.0 - salience).clamp(0.0, 1.0);
153
154    // 5. Importance: inverse of semantic importance
155    let importance_factor = (1.0 - mem.metadata.importance.clamp(0.0, 1.0)).clamp(0.0, 1.0);
156
157    // Weighted composite
158    let sum_weights = config.weight_age
159        + config.weight_access
160        + config.weight_resonance
161        + config.weight_emotional
162        + config.weight_importance;
163
164    let distance = if sum_weights > 0.0 {
165        let raw = config.weight_importance.mul_add(
166            importance_factor,
167            config.weight_emotional.mul_add(
168                emotional_factor,
169                config.weight_resonance.mul_add(
170                    resonance_factor,
171                    config
172                        .weight_access
173                        .mul_add(access_factor, config.weight_age * age_factor),
174                ),
175            ),
176        );
177        (raw / sum_weights).clamp(0.0, 1.0)
178    } else {
179        0.5
180    };
181
182    OuterRimFactors {
183        age_factor,
184        access_factor,
185        resonance_factor,
186        emotional_factor,
187        importance_factor,
188        distance,
189    }
190}
191
192// ── Compression & Decompression Helpers ─────────────────────────────────
193
194/// Compress a memory into a binary payload using the specified codec.
195pub fn compress_memory(mem: &Memory, codec: CompressionCodec) -> Result<(Vec<u8>, usize)> {
196    let uncompressed = rmp_serde::to_vec_named(mem)
197        .map_err(|e| CoreError::Memory(format!("Cold storage serialization error: {e}")))?;
198    let uncompressed_size = uncompressed.len();
199
200    let compressed = match codec {
201        CompressionCodec::Gzip => {
202            use flate2::Compression;
203            use flate2::write::GzEncoder;
204            use std::io::Write;
205            let mut encoder = GzEncoder::new(Vec::new(), Compression::best());
206            encoder
207                .write_all(&uncompressed)
208                .map_err(|e| CoreError::Memory(format!("Gzip compression error: {e}")))?;
209            encoder
210                .finish()
211                .map_err(|e| CoreError::Memory(format!("Gzip finish error: {e}")))?
212        }
213        CompressionCodec::Deflate => {
214            use flate2::Compression;
215            use flate2::write::DeflateEncoder;
216            use std::io::Write;
217            let mut encoder = DeflateEncoder::new(Vec::new(), Compression::best());
218            encoder
219                .write_all(&uncompressed)
220                .map_err(|e| CoreError::Memory(format!("Deflate compression error: {e}")))?;
221            encoder
222                .finish()
223                .map_err(|e| CoreError::Memory(format!("Deflate finish error: {e}")))?
224        }
225        CompressionCodec::Msgpack => uncompressed,
226    };
227
228    Ok((compressed, uncompressed_size))
229}
230
231/// Decompress a binary payload back into the original memory.
232pub fn decompress_memory(payload: &[u8], codec: CompressionCodec) -> Result<Memory> {
233    let uncompressed = match codec {
234        CompressionCodec::Gzip => {
235            use flate2::read::GzDecoder;
236            use std::io::Read;
237            let mut decoder = GzDecoder::new(payload);
238            let mut buf = Vec::new();
239            decoder
240                .read_to_end(&mut buf)
241                .map_err(|e| CoreError::Memory(format!("Gzip decompression error: {e}")))?;
242            buf
243        }
244        CompressionCodec::Deflate => {
245            use flate2::read::DeflateDecoder;
246            use std::io::Read;
247            let mut decoder = DeflateDecoder::new(payload);
248            let mut buf = Vec::new();
249            decoder
250                .read_to_end(&mut buf)
251                .map_err(|e| CoreError::Memory(format!("Deflate decompression error: {e}")))?;
252            buf
253        }
254        CompressionCodec::Msgpack => payload.to_vec(),
255    };
256
257    crate::codec::decode(&uncompressed)
258        .map_err(|e| CoreError::Memory(format!("Cold storage deserialization error: {e}")))
259}
260
261fn create_snippet(content: &str) -> String {
262    let trimmed = content.trim();
263    if trimmed.chars().count() <= 120 {
264        trimmed.to_string()
265    } else {
266        let mut s: String = trimmed.chars().take(117).collect();
267        s.push_str("...");
268        s
269    }
270}
271
272/// A public digest is itself a public/model-visible derived record. Restricted
273/// sources must not contribute snippets, tags, lineage IDs, or aggregate shape
274/// to it. Keep this at the builder boundary as well as the outer-rim scan:
275/// direct `digest_cluster` callers must not bypass the policy.
276const fn eligible_for_public_digest(mem: &Memory) -> bool {
277    !mem.metadata.is_protected && !mem.metadata.is_private && !mem.metadata.model_exclude
278}
279
280// ── Cold Storage Records & Summaries ────────────────────────────────────
281
282/// A persistent record stored in the cold archive.
283#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
284pub struct ColdRecord {
285    /// Memory UUID.
286    pub id: MemoryId,
287    /// Original galaxy this memory belonged to.
288    pub galaxy: Galaxy,
289    /// Content hash (SHA-256) for integrity verification.
290    pub content_hash: String,
291    /// Timestamp when this memory was migrated into cold storage.
292    pub cold_stored_at: DateTime<Utc>,
293    /// Outer-rim distance calculated at freezing time.
294    pub outer_rim_distance: f32,
295    /// Factor breakdown at freezing time.
296    pub distance_factors: OuterRimFactors,
297    /// ID of the distilled thematic node (if synthesized during digestion).
298    pub digest_id: Option<MemoryId>,
299    /// Lineage & provenance notes.
300    pub lineage_notes: Option<String>,
301    /// Compression codec used.
302    pub codec: CompressionCodec,
303    /// Uncompressed payload size in bytes.
304    pub uncompressed_size: usize,
305    /// Compressed binary payload containing the full original Memory.
306    pub compressed_payload: Vec<u8>,
307    /// Vector clock / version at freezing time.
308    pub version: u64,
309    /// Fast-search tags preserved in cold header.
310    pub tags: Vec<String>,
311    /// Title preserved in cold header.
312    pub title: Option<String>,
313    /// Text snippet for rapid search without decompression.
314    pub snippet: String,
315}
316
317impl ColdRecord {
318    /// Create a new cold record by compressing a hot memory.
319    pub fn new(
320        mem: &Memory,
321        outer_rim_distance: f32,
322        distance_factors: OuterRimFactors,
323        digest_id: Option<MemoryId>,
324        lineage_notes: Option<String>,
325        codec: CompressionCodec,
326    ) -> Result<Self> {
327        let (compressed_payload, uncompressed_size) = compress_memory(mem, codec)?;
328        let snippet = create_snippet(&mem.content);
329
330        Ok(Self {
331            id: mem.metadata.id,
332            galaxy: mem.metadata.galaxy,
333            content_hash: mem.metadata.content_hash.clone(),
334            cold_stored_at: Utc::now(),
335            outer_rim_distance,
336            distance_factors,
337            digest_id,
338            lineage_notes,
339            codec,
340            uncompressed_size,
341            compressed_payload,
342            version: mem.metadata.version,
343            tags: mem.metadata.tags.clone(),
344            title: mem.metadata.title.clone(),
345            snippet,
346        })
347    }
348
349    /// Extract a lightweight summary without decompressing the full memory payload.
350    #[must_use]
351    pub fn summary(&self) -> ColdRecordSummary {
352        ColdRecordSummary {
353            id: self.id,
354            galaxy: self.galaxy,
355            content_hash: self.content_hash.clone(),
356            cold_stored_at: self.cold_stored_at,
357            outer_rim_distance: self.outer_rim_distance,
358            digest_id: self.digest_id,
359            uncompressed_size: self.uncompressed_size,
360            compressed_size: self.compressed_payload.len(),
361            codec: self.codec,
362            tags: self.tags.clone(),
363            title: self.title.clone(),
364            snippet: self.snippet.clone(),
365        }
366    }
367
368    /// Decompress and restore the complete original `Memory`.
369    pub fn decompress(&self) -> Result<Memory> {
370        decompress_memory(&self.compressed_payload, self.codec)
371    }
372}
373
374/// Lightweight summary of a cold-stored memory for listings and queries.
375#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
376pub struct ColdRecordSummary {
377    /// Memory UUID.
378    pub id: MemoryId,
379    /// Original Galaxy.
380    pub galaxy: Galaxy,
381    /// Content hash.
382    pub content_hash: String,
383    /// Cold storage timestamp.
384    pub cold_stored_at: DateTime<Utc>,
385    /// Outer-rim distance at freeze time.
386    pub outer_rim_distance: f32,
387    /// Associated digest node ID (if any).
388    pub digest_id: Option<MemoryId>,
389    /// Uncompressed bytes.
390    pub uncompressed_size: usize,
391    /// Compressed bytes.
392    pub compressed_size: usize,
393    /// Codec used.
394    pub codec: CompressionCodec,
395    /// Tags preserved in cold header.
396    pub tags: Vec<String>,
397    /// Title preserved in cold header.
398    pub title: Option<String>,
399    /// Text snippet.
400    pub snippet: String,
401}
402
403// ── Cold Query Filter ───────────────────────────────────────────────────
404
405/// Filter for querying cold-stored memories.
406#[derive(Debug, Clone, Default)]
407pub struct ColdQuery {
408    /// Optional galaxy filter.
409    pub galaxy: Option<Galaxy>,
410    /// Must contain all specified tags.
411    pub tags: Vec<String>,
412    /// Minimum outer-rim distance filter.
413    pub min_distance: Option<f32>,
414    /// Stored after this timestamp.
415    pub stored_after: Option<DateTime<Utc>>,
416    /// Stored before this timestamp.
417    pub stored_before: Option<DateTime<Utc>>,
418    /// Substring match over title or snippet.
419    pub content_substring: Option<String>,
420    /// Maximum results to return.
421    pub limit: usize,
422}
423
424/// Why a bounded scan stopped. Exhaustion is established only by iterator end.
425#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize)]
426#[serde(rename_all = "snake_case")]
427pub enum ColdDiscoveryStop {
428    #[default]
429    NotStarted,
430    Exhausted,
431    ScanLimit,
432    ResultLimit,
433}
434
435/// Outcome of a bounded cold-storage discovery scan.
436///
437/// Discovery is opt-in and bounded: candidates are hydrated and authorized
438/// from the cold payload (decompress + integrity + visibility) rather than
439/// trusting stale index entries. No record is thawed or mutated.
440#[derive(Debug, Default, Clone)]
441pub struct ColdDiscoveryOutcome {
442    /// Actual termination boundary, not inferred from visit counters.
443    pub stop_reason: ColdDiscoveryStop,
444    /// Cold records visited by the scan.
445    pub scanned: usize,
446    /// Records in the requested galaxy.
447    pub candidates: usize,
448    /// Records that text-matched, verified, and passed visibility.
449    pub matched: usize,
450    /// Records refused because decompression or hash verification failed.
451    pub integrity_rejected: usize,
452    /// Records skipped because they are private (never surface over MCP).
453    pub private_skipped: usize,
454    /// Records skipped because their validity is not current (superseded).
455    pub non_current_skipped: usize,
456    /// Visible text matches rejected by the caller's eligibility predicate.
457    pub eligibility_skipped: usize,
458    /// Verified, visible matching records (caller decompresses for output).
459    pub records: Vec<ColdRecord>,
460}
461
462impl ColdQuery {
463    /// Create a new query matching all cold records (limit 100).
464    #[must_use]
465    pub fn new() -> Self {
466        Self {
467            limit: 100,
468            ..Default::default()
469        }
470    }
471
472    /// Filter by galaxy.
473    #[must_use]
474    pub const fn with_galaxy(mut self, galaxy: Galaxy) -> Self {
475        self.galaxy = Some(galaxy);
476        self
477    }
478
479    /// Filter by tags.
480    #[must_use]
481    pub fn with_tags(mut self, tags: Vec<String>) -> Self {
482        self.tags = tags;
483        self
484    }
485
486    /// Filter by minimum distance.
487    #[must_use]
488    pub const fn with_min_distance(mut self, min_dist: f32) -> Self {
489        self.min_distance = Some(min_dist);
490        self
491    }
492
493    /// Filter by result limit.
494    #[must_use]
495    pub const fn with_limit(mut self, limit: usize) -> Self {
496        self.limit = limit;
497        self
498    }
499
500    /// Filter by substring.
501    #[must_use]
502    pub fn with_content_substring(mut self, substring: impl Into<String>) -> Self {
503        self.content_substring = Some(substring.into().to_lowercase());
504        self
505    }
506
507    /// Check if a cold record summary satisfies this query.
508    #[must_use]
509    pub fn matches(&self, summary: &ColdRecordSummary) -> bool {
510        if let Some(g) = self.galaxy {
511            if summary.galaxy != g {
512                return false;
513            }
514        }
515        if let Some(min_d) = self.min_distance {
516            if summary.outer_rim_distance < min_d {
517                return false;
518            }
519        }
520        if let Some(after) = self.stored_after {
521            if summary.cold_stored_at < after {
522                return false;
523            }
524        }
525        if let Some(before) = self.stored_before {
526            if summary.cold_stored_at > before {
527                return false;
528            }
529        }
530        for tag in &self.tags {
531            if !summary.tags.iter().any(|t| t == tag) {
532                return false;
533            }
534        }
535        if let Some(sub) = &self.content_substring {
536            let in_title = summary
537                .title
538                .as_ref()
539                .is_some_and(|t| t.to_lowercase().contains(sub));
540            let in_snippet = summary.snippet.to_lowercase().contains(sub);
541            if !in_title && !in_snippet {
542                return false;
543            }
544        }
545        true
546    }
547}
548
549// ── Phagic Digest Result Report ─────────────────────────────────────────
550
551/// Telemetry report of a phagic digestion pass on a galaxy.
552#[derive(Debug, Clone, Serialize, Deserialize)]
553pub struct PhagicDigestReport {
554    /// Galaxy processed.
555    pub galaxy: Galaxy,
556    /// Total memories scanned in the galaxy.
557    pub memories_examined: usize,
558    /// Outer-rim memories identified above the distance threshold.
559    pub outer_rim_candidates: usize,
560    /// Memories migrated into cold storage.
561    pub memories_digested: usize,
562    /// ID of the synthesized thematic digest node placed in the active tier.
563    pub digest_memory_id: Option<MemoryId>,
564    /// Uncompressed raw bytes across all digested memories.
565    pub total_raw_bytes: usize,
566    /// Compressed bytes stored in cold archive.
567    pub total_cold_bytes: usize,
568    /// Space compression ratio: `total_cold_bytes / total_raw_bytes`.
569    pub compression_ratio: f32,
570    /// Average outer-rim distance of the digested memories.
571    pub average_outer_rim_distance: f32,
572}
573
574// ── Phagic Digester Engine ──────────────────────────────────────────────
575
576/// Non-destructive digestive processor.
577pub struct PhagicDigester {
578    config: PhagicConfig,
579}
580
581impl PhagicDigester {
582    /// Create a new digester with the specified configuration.
583    #[must_use]
584    pub const fn new(config: PhagicConfig) -> Self {
585        Self { config }
586    }
587
588    /// Create with default configuration.
589    #[must_use]
590    pub fn default_config() -> Self {
591        Self::new(PhagicConfig::default())
592    }
593
594    /// Access configuration.
595    #[must_use]
596    pub const fn config(&self) -> &PhagicConfig {
597        &self.config
598    }
599
600    /// Calculate distance factors for a memory.
601    #[must_use]
602    pub fn calculate_distance(&self, mem: &Memory, now: DateTime<Utc>) -> OuterRimFactors {
603        calculate_outer_rim_distance(mem, &self.config, now)
604    }
605
606    /// Scan a galaxy and identify outer-rim memories sorted by distance descending.
607    pub fn scan_outer_rim(
608        &self,
609        store: &MemoryStore,
610        galaxy: Galaxy,
611    ) -> Result<Vec<(Memory, OuterRimFactors)>> {
612        let memories = store.scan(galaxy, 10_000)?;
613        let now = Utc::now();
614        let mut candidates = Vec::new();
615
616        for mem in memories {
617            // A public digest must not synthesize restricted source material.
618            // The same check is repeated by digest_cluster for direct callers.
619            if !eligible_for_public_digest(&mem) {
620                continue;
621            }
622
623            let factors = self.calculate_distance(&mem, now);
624            if factors.distance >= self.config.outer_rim_threshold {
625                candidates.push((mem, factors));
626            }
627        }
628
629        // Sort descending by distance (furthest memories first)
630        candidates.sort_by(|a, b| {
631            b.1.distance
632                .partial_cmp(&a.1.distance)
633                .unwrap_or(std::cmp::Ordering::Equal)
634        });
635
636        Ok(candidates)
637    }
638
639    /// Synthesize a cluster of outer-rim memories into a distilled thematic node in the hot tier,
640    /// and gently migrate the original memories into the cold storage tier.
641    pub fn digest_cluster(
642        &self,
643        store: &MemoryStore,
644        search: Option<&SearchEngine>,
645        galaxy: Galaxy,
646        cluster: &[(Memory, OuterRimFactors)],
647    ) -> Result<PhagicDigestReport> {
648        // A caller may invoke this public builder directly instead of using
649        // scan_outer_rim. Filter again before collecting any tag, time,
650        // snippet, ID, or count into a public derived record. Mixed clusters
651        // retain restricted originals untouched and digest eligible sources
652        // only; a restricted-only cluster is a truthful no-op.
653        let eligible: Vec<&(Memory, OuterRimFactors)> = cluster
654            .iter()
655            .filter(|(mem, _)| eligible_for_public_digest(mem))
656            .collect();
657        if eligible.is_empty() {
658            return Ok(PhagicDigestReport {
659                galaxy,
660                memories_examined: 0,
661                outer_rim_candidates: 0,
662                memories_digested: 0,
663                digest_memory_id: None,
664                total_raw_bytes: 0,
665                total_cold_bytes: 0,
666                compression_ratio: 0.0,
667                average_outer_rim_distance: 0.0,
668            });
669        }
670
671        let cluster_id = Uuid::new_v4();
672        let now = Utc::now();
673
674        // Compute tag frequency and shared topics
675        let mut tag_counts: HashMap<String, usize> = HashMap::new();
676        let mut total_distance = 0.0;
677        let mut min_time = now;
678        let mut max_time = DateTime::<Utc>::MIN_UTC;
679
680        for entry in &eligible {
681            let (mem, factors) = *entry;
682            total_distance += factors.distance;
683            if mem.metadata.created_at < min_time {
684                min_time = mem.metadata.created_at;
685            }
686            if mem.metadata.created_at > max_time {
687                max_time = mem.metadata.created_at;
688            }
689            for tag in &mem.metadata.tags {
690                if !tag.starts_with("phagic:") && !tag.starts_with("cold_source:") {
691                    *tag_counts.entry(tag.clone()).or_insert(0) += 1;
692                }
693            }
694        }
695
696        let avg_distance = total_distance / eligible.len() as f32;
697
698        let mut top_tags: Vec<(String, usize)> = tag_counts.into_iter().collect();
699        top_tags.sort_by_key(|a| std::cmp::Reverse(a.1));
700        let prominent_tags: Vec<String> =
701            top_tags.into_iter().take(6).map(|(tag, _)| tag).collect();
702
703        // 1. Synthesize thematic digest memory for active hot tier
704        let mut digest_content = format!(
705            "# Phagic Thematic Digest: {} Outer-Rim Condensation\n\n\
706             **Cluster ID**: `{cluster_id}`\n\
707             **Galaxy**: `{}`\n\
708             **Digested Memories**: {} items safely migrated to Cold Storage\n\
709             **Temporal Span**: {} to {}\n\
710             **Average Outer-Rim Distance**: {:.3}\n\
711             **Prominent Themes**: {}\n\n\
712             ## Distilled Semantic Abstract\n\
713             This thematic node preserves the collective semantic essence of {} outer-rim memories\n\
714             from galaxy {}. The original memories have been gently transitioned into compressed\n\
715             cold storage without data loss, retaining complete lineage and full thawability.\n\n\
716             ## Lineage Pointers (Thawable Cold Records)\n",
717            galaxy.db_name(),
718            galaxy.db_name(),
719            eligible.len(),
720            min_time.format("%Y-%m-%d %H:%M:%S UTC"),
721            max_time.format("%Y-%m-%d %H:%M:%S UTC"),
722            avg_distance,
723            if prominent_tags.is_empty() {
724                "none".to_string()
725            } else {
726                prominent_tags.join(", ")
727            },
728            eligible.len(),
729            galaxy.db_name(),
730        );
731
732        use std::fmt::Write as _;
733        for entry in &eligible {
734            let (mem, factors) = *entry;
735            let snippet = create_snippet(&mem.content);
736            let _ = writeln!(
737                digest_content,
738                "- **[Memory `{}`]** (Distance: {:.2}): {snippet}",
739                mem.metadata.id, factors.distance
740            );
741        }
742
743        let mut digest_tags = prominent_tags;
744        digest_tags.push("phagic:digest".to_string());
745        digest_tags.push(format!("phagic:cluster:{cluster_id}"));
746
747        let digest_title = format!(
748            "Phagic Digest: {} ({} items)",
749            galaxy.db_name(),
750            cluster.len()
751        );
752        let mut digest_mem = Memory::new(galaxy, digest_content)
753            .with_tags(digest_tags)
754            .with_importance(0.65)
755            .with_memory_type(MemoryType::Symbolic)
756            .with_source("phagic:digestion".to_string(), 0.9);
757        digest_mem.metadata.tier = Tier::Semantic;
758        digest_mem.metadata.title = Some(digest_title);
759        digest_mem.metadata.topic = Some(format!("{}:phagic_digest", galaxy.db_name()));
760
761        // Store thematic digest memory in the hot store
762        store.put(galaxy, &digest_mem)?;
763
764        // Index the thematic digest node in Tantivy search if available
765        if let Some(engine) = search {
766            if let Ok(mut writer_guard) = engine.writer() {
767                let _ = engine.add_document(
768                    &mut writer_guard,
769                    &digest_mem.metadata.id.to_string(),
770                    galaxy.db_name(),
771                    &digest_mem.content,
772                    &digest_mem.metadata.tags,
773                    digest_mem.metadata.created_at.timestamp(),
774                );
775                let _ = engine.commit(&mut writer_guard);
776            }
777        }
778
779        let digest_id = digest_mem.metadata.id;
780
781        // 2. Freeze each original memory into cold storage
782        let mut total_raw_bytes = 0;
783        let mut total_cold_bytes = 0;
784
785        for entry in &eligible {
786            let (mem, factors) = *entry;
787            let notes = format!("Digested in phagic cluster {cluster_id} (digest_id: {digest_id})");
788            let cold_rec = store.freeze_to_cold(
789                search,
790                mem.metadata.id,
791                factors.distance,
792                factors.clone(),
793                Some(digest_id),
794                Some(notes),
795                self.config.compression_codec,
796            )?;
797            total_raw_bytes += cold_rec.uncompressed_size;
798            total_cold_bytes += cold_rec.compressed_payload.len();
799        }
800
801        let compression_ratio = if total_raw_bytes > 0 {
802            total_cold_bytes as f32 / total_raw_bytes as f32
803        } else {
804            0.0
805        };
806
807        Ok(PhagicDigestReport {
808            galaxy,
809            memories_examined: eligible.len(),
810            outer_rim_candidates: eligible.len(),
811            memories_digested: eligible.len(),
812            digest_memory_id: Some(digest_id),
813            total_raw_bytes,
814            total_cold_bytes,
815            compression_ratio,
816            average_outer_rim_distance: avg_distance,
817        })
818    }
819
820    /// Perform a full non-destructive digestive pass on a galaxy.
821    pub fn digest_galaxy(
822        &self,
823        store: &MemoryStore,
824        search: Option<&SearchEngine>,
825        galaxy: Galaxy,
826    ) -> Result<PhagicDigestReport> {
827        let candidates = self.scan_outer_rim(store, galaxy)?;
828        let count = store.count(galaxy).unwrap_or(0);
829
830        if candidates.is_empty() {
831            return Ok(PhagicDigestReport {
832                galaxy,
833                memories_examined: count,
834                outer_rim_candidates: 0,
835                memories_digested: 0,
836                digest_memory_id: None,
837                total_raw_bytes: 0,
838                total_cold_bytes: 0,
839                compression_ratio: 0.0,
840                average_outer_rim_distance: 0.0,
841            });
842        }
843
844        // Bound to max_digest_batch
845        let batch_size = candidates.len().min(self.config.max_digest_batch);
846        let batch = &candidates[..batch_size];
847
848        let mut report = self.digest_cluster(store, search, galaxy, batch)?;
849        report.memories_examined = count;
850        report.outer_rim_candidates = candidates.len();
851        Ok(report)
852    }
853
854    /// Perform phagic digestion across all memory galaxies.
855    pub fn digest_all(
856        &self,
857        store: &MemoryStore,
858        search: Option<&SearchEngine>,
859    ) -> Result<Vec<PhagicDigestReport>> {
860        let mut reports = Vec::new();
861        for galaxy in Galaxy::all() {
862            match galaxy {
863                Galaxy::Substrate
864                | Galaxy::Dharma
865                | Galaxy::Karma
866                | Galaxy::Embeddings
867                | Galaxy::Associations => continue,
868                _ => {}
869            }
870            if store.count(galaxy).unwrap_or(0) == 0 {
871                continue;
872            }
873            reports.push(self.digest_galaxy(store, search, galaxy)?);
874        }
875        Ok(reports)
876    }
877
878    /// Thaw a cold-stored memory back into the active hot tier.
879    pub fn thaw(
880        &self,
881        store: &MemoryStore,
882        search: Option<&SearchEngine>,
883        id: MemoryId,
884    ) -> Result<Memory> {
885        store.thaw_from_cold(search, id)
886    }
887}
888
889#[cfg(test)]
890mod tests {
891    use super::*;
892    use tempfile::tempdir;
893
894    fn test_store() -> MemoryStore {
895        let tmp = tempdir().unwrap();
896        MemoryStore::open_default(tmp.path()).unwrap()
897    }
898
899    #[test]
900    fn outer_rim_distance_calculation_monotonic() {
901        let config = PhagicConfig::default();
902        let now = Utc::now();
903
904        // 1. Core memory: recent, high importance, high resonance, high emotion, accessed
905        let mut core_mem = Memory::new(Galaxy::Codex, "Vital core memory".into())
906            .with_importance(0.95)
907            .with_neuro_score(0.9)
908            .with_emotional_valence(0.8, 0.9);
909        core_mem.metadata.access_count = 10;
910        core_mem.metadata.accessed_at = now;
911
912        let core_factors = calculate_outer_rim_distance(&core_mem, &config, now);
913        assert!(
914            core_factors.distance < 0.25,
915            "Core memory distance should be low: {}",
916            core_factors.distance
917        );
918
919        // 2. Outer-rim memory: ancient, 0 accesses, low importance, low resonance, neutral emotion
920        let mut rim_mem = Memory::new(Galaxy::Codex, "Faded distant note".into())
921            .with_importance(0.05)
922            .with_neuro_score(0.1)
923            .with_emotional_valence(0.0, 0.0);
924        rim_mem.metadata.access_count = 0;
925        rim_mem.metadata.recall_count = 0;
926        rim_mem.metadata.accessed_at = now - chrono::Duration::days(120);
927
928        let rim_factors = calculate_outer_rim_distance(&rim_mem, &config, now);
929        assert!(
930            rim_factors.distance >= config.outer_rim_threshold,
931            "Rim memory distance should exceed threshold: {}",
932            rim_factors.distance
933        );
934        assert!(
935            rim_factors.distance > core_factors.distance,
936            "Rim memory must be strictly further out than core memory"
937        );
938    }
939
940    #[test]
941    fn protected_memory_is_anchored_at_core() {
942        let config = PhagicConfig::default();
943        let now = Utc::now();
944
945        let mut protected_mem = Memory::new(Galaxy::Codex, "Protected sacred memory".into())
946            .with_protection(true)
947            .with_importance(0.01); // low importance, but protected
948        protected_mem.metadata.accessed_at = now - chrono::Duration::days(365);
949
950        let factors = calculate_outer_rim_distance(&protected_mem, &config, now);
951        assert_eq!(
952            factors.distance, 0.0,
953            "Protected memory must always have distance 0.0"
954        );
955    }
956
957    #[test]
958    fn cold_record_compression_roundtrip() {
959        let mem = Memory::new(
960            Galaxy::Research,
961            "Extensive research logs on cosmic non-destructive phagocytosis and entropy reversal"
962                .into(),
963        )
964        .with_tags(vec!["physics".into(), "entropy".into(), "phagic".into()])
965        .with_importance(0.3);
966
967        for codec in [
968            CompressionCodec::Gzip,
969            CompressionCodec::Deflate,
970            CompressionCodec::Msgpack,
971        ] {
972            let factors = OuterRimFactors {
973                age_factor: 0.8,
974                access_factor: 0.9,
975                resonance_factor: 0.7,
976                emotional_factor: 0.9,
977                importance_factor: 0.7,
978                distance: 0.81,
979            };
980
981            let cold = ColdRecord::new(&mem, 0.81, factors, None, Some("test notes".into()), codec)
982                .unwrap();
983            assert_eq!(cold.id, mem.metadata.id);
984            assert_eq!(cold.galaxy, Galaxy::Research);
985
986            let decompressed = cold.decompress().unwrap();
987            assert_eq!(decompressed.metadata.id, mem.metadata.id);
988            assert_eq!(decompressed.content, mem.content);
989            assert_eq!(decompressed.metadata.tags, mem.metadata.tags);
990            assert_eq!(decompressed.metadata.importance, mem.metadata.importance);
991        }
992    }
993
994    #[test]
995    fn freeze_and_thaw_zero_data_loss() {
996        let store = test_store();
997        let galaxy = Galaxy::Codex;
998
999        let content = "Detailed architectural blueprint for WhiteMagic non-destructive storage";
1000        let mem = Memory::new(galaxy, content.into())
1001            .with_tags(vec!["architecture".into(), "v9".into()])
1002            .with_importance(0.2);
1003        let id = mem.metadata.id;
1004
1005        store.put(galaxy, &mem).unwrap();
1006        assert!(store.get(galaxy, id).unwrap().is_some());
1007
1008        // Freeze to cold storage
1009        let digester = PhagicDigester::default_config();
1010        let factors = digester.calculate_distance(&mem, Utc::now());
1011        let cold_record = store
1012            .freeze_to_cold(
1013                None,
1014                id,
1015                factors.distance,
1016                factors,
1017                None,
1018                Some("unit test freeze".into()),
1019                CompressionCodec::Gzip,
1020            )
1021            .unwrap();
1022
1023        assert_eq!(cold_record.id, id);
1024
1025        // Hot store should no longer contain the memory
1026        assert!(store.get(galaxy, id).unwrap().is_none());
1027
1028        // But cold store DOES contain it
1029        let cold_found = store.get_cold_record(id).unwrap();
1030        assert!(cold_found.is_some());
1031        let cold_rec = cold_found.unwrap();
1032        assert_eq!(cold_rec.content_hash, mem.metadata.content_hash);
1033
1034        // find_anywhere should find it with is_cold = true
1035        let (found_galaxy, found_mem, is_cold) = store.find_anywhere(id).unwrap().unwrap();
1036        assert_eq!(found_galaxy, galaxy);
1037        assert_eq!(found_mem.content, content);
1038        assert!(is_cold);
1039
1040        // Thaw it back into the hot store
1041        let thawed = store.thaw_from_cold(None, id).unwrap();
1042        assert_eq!(thawed.metadata.id, id);
1043        assert_eq!(thawed.content, content);
1044        assert_eq!(
1045            thawed.metadata.tags,
1046            vec!["architecture", "v9", "thawed:phagic"]
1047        );
1048        assert_eq!(thawed.metadata.tier, Tier::Episodic);
1049
1050        // Hot store now contains it again
1051        assert!(store.get(galaxy, id).unwrap().is_some());
1052
1053        // Cold store no longer contains it
1054        assert!(store.get_cold_record(id).unwrap().is_none());
1055
1056        // find_anywhere now returns is_cold = false
1057        let (_, _, is_cold_after) = store.find_anywhere(id).unwrap().unwrap();
1058        assert!(!is_cold_after);
1059    }
1060
1061    #[test]
1062    fn phagic_digest_cluster_synthesizes_thematic_node() {
1063        let store = test_store();
1064        let galaxy = Galaxy::Journals;
1065        let now = Utc::now();
1066
1067        // Populate with 3 distant outer-rim memories
1068        let mut ids = Vec::new();
1069        for i in 1..=3 {
1070            let mut mem = Memory::new(
1071                galaxy,
1072                format!("Ancient journal reflection #{i} about distant journeys"),
1073            )
1074            .with_tags(vec!["travel".into(), "reflection".into()])
1075            .with_importance(0.05);
1076            mem.metadata.accessed_at = now - chrono::Duration::days(150);
1077            mem.metadata.access_count = 0;
1078            store.put(galaxy, &mem).unwrap();
1079            ids.push(mem.metadata.id);
1080        }
1081
1082        let digester = PhagicDigester::default_config();
1083        let report = digester.digest_galaxy(&store, None, galaxy).unwrap();
1084
1085        assert_eq!(report.memories_digested, 3);
1086        assert!(report.digest_memory_id.is_some());
1087        assert!(report.total_cold_bytes > 0);
1088        assert!(report.total_raw_bytes > report.total_cold_bytes);
1089
1090        // All 3 original memories are in cold storage, removed from hot galaxy
1091        for id in &ids {
1092            assert!(store.get(galaxy, *id).unwrap().is_none());
1093            assert!(store.get_cold_record(*id).unwrap().is_some());
1094        }
1095
1096        // Thematic digest memory is present in hot galaxy
1097        let digest_id = report.digest_memory_id.unwrap();
1098        let digest_mem = store.get(galaxy, digest_id).unwrap().unwrap();
1099        assert_eq!(digest_mem.metadata.tier, Tier::Semantic);
1100        assert!(digest_mem.content.contains("Phagic Thematic Digest"));
1101        assert!(
1102            digest_mem
1103                .metadata
1104                .tags
1105                .contains(&"phagic:digest".to_string())
1106        );
1107
1108        // Thaw one of the cold memories back to hot
1109        let thawed = digester.thaw(&store, None, ids[0]).unwrap();
1110        assert_eq!(thawed.metadata.id, ids[0]);
1111        assert!(store.get(galaxy, ids[0]).unwrap().is_some());
1112        assert!(store.get_cold_record(ids[0]).unwrap().is_none());
1113    }
1114
1115    #[test]
1116    fn direct_digest_cluster_does_not_launder_restricted_source_material() {
1117        let store = test_store();
1118        let galaxy = Galaxy::Codex;
1119        let factors = OuterRimFactors {
1120            age_factor: 1.0,
1121            access_factor: 1.0,
1122            resonance_factor: 1.0,
1123            emotional_factor: 1.0,
1124            importance_factor: 1.0,
1125            distance: 1.0,
1126        };
1127
1128        let public =
1129            Memory::new(galaxy, "public digest source".into()).with_tags(vec!["public-tag".into()]);
1130        let mut private = Memory::new(galaxy, "PRIVATE-SOURCE-CONTENT".into())
1131            .with_tags(vec!["private-source-tag".into()]);
1132        private.metadata.is_private = true;
1133        let mut model_excluded = Memory::new(galaxy, "MODEL-EXCLUDED-SOURCE-CONTENT".into())
1134            .with_tags(vec!["model-excluded-source-tag".into()]);
1135        model_excluded.metadata.model_exclude = true;
1136        let protected = Memory::new(galaxy, "PROTECTED-SOURCE-CONTENT".into())
1137            .with_tags(vec!["protected-source-tag".into()])
1138            .with_protection(true);
1139
1140        for memory in [&public, &private, &model_excluded, &protected] {
1141            store.put(galaxy, memory).unwrap();
1142        }
1143        let restricted = [private.clone(), model_excluded.clone(), protected.clone()];
1144        let cluster = vec![
1145            (public.clone(), factors.clone()),
1146            (private, factors.clone()),
1147            (model_excluded, factors.clone()),
1148            (protected, factors),
1149        ];
1150
1151        // Direct builder invocation is the discriminating boundary: relying
1152        // only on scan_outer_rim would leave this path able to synthesize a
1153        // public digest from restricted input.
1154        let report = PhagicDigester::default_config()
1155            .digest_cluster(&store, None, galaxy, &cluster)
1156            .unwrap();
1157        assert_eq!(report.memories_digested, 1);
1158        let digest = store
1159            .get(galaxy, report.digest_memory_id.unwrap())
1160            .unwrap()
1161            .unwrap();
1162        assert!(!digest.metadata.is_private);
1163        assert!(!digest.metadata.model_exclude);
1164        assert!(digest.content.contains("public digest source"));
1165        for source in &restricted {
1166            assert!(!digest.content.contains(&source.content));
1167            assert!(!digest.content.contains(&source.metadata.id.to_string()));
1168            for tag in &source.metadata.tags {
1169                assert!(!digest.metadata.tags.contains(tag));
1170                assert!(!digest.content.contains(tag));
1171            }
1172            let retained = store.get(galaxy, source.metadata.id).unwrap().unwrap();
1173            assert_eq!(
1174                serde_json::to_value(retained).unwrap(),
1175                serde_json::to_value(source).unwrap()
1176            );
1177            assert!(store.get_cold_record(source.metadata.id).unwrap().is_none());
1178        }
1179
1180        assert!(store.get(galaxy, public.metadata.id).unwrap().is_none());
1181        assert!(store.get_cold_record(public.metadata.id).unwrap().is_some());
1182    }
1183}