Skip to main content

bamboo_engine/
ledger_gardener.rs

1//! Background ledger gardener: maintenance for prospective-memory records.
2//!
3//! Three ordered passes, mirroring the memory gardener's cost model (each pass
4//! is config-gated, capped, and free when its deterministic prefilter finds
5//! nothing):
6//!
7//! 1. **Expiry** (deterministic, zero LLM): past events and fully-fired
8//!    reminders transition to `Expired` and drop out of the agenda. Todos
9//!    never auto-expire — an overdue todo is still open work.
10//! 2. **Schedule reconciliation** (deterministic, zero LLM): repairs
11//!    record↔schedule drift through the [`LedgerScheduleBridge`] — terminal
12//!    records release leftover schedules; open records with reminder times but
13//!    no managed schedules (a crash between writes) get them re-synced.
14//! 3. **Distillation** (background model): completed records become durable
15//!    memories — the ledger feeding the long-term memory system once
16//!    prospective records resolve into retrospective knowledge.
17
18use std::sync::Arc;
19use std::time::{Duration, Instant};
20
21use chrono::{DateTime, Utc};
22use serde::Deserialize;
23
24use bamboo_domain::ledger::{LedgerRecord, LedgerScope, RecordStatus};
25use bamboo_memory::ledger_store::{LedgerScheduleBridge, LedgerStore, RecordFilter};
26use bamboo_memory::memory_store::{DurableMemoryType, MemoryStore};
27
28use crate::auto_dream::AutoDreamContext;
29use crate::gardener::{collect_model_json, resolve_background_model};
30
31const LEDGER_GARDENER_TRACING_TARGET: &str = "bamboo.ledger_gardener";
32/// Grace period before a past event/reminder is auto-expired, so "the meeting
33/// just ended" records survive long enough for a follow-up conversation.
34const EXPIRY_GRACE_HOURS: i64 = 48;
35/// Max completed records fed to one distillation call.
36const DISTILL_MAX_RECORDS_PER_RUN: usize = 8;
37/// Tag marking a record as already distilled (or considered and skipped).
38const DISTILLED_TAG: &str = "distilled";
39
40const DISTILL_SYSTEM_INSTRUCTION: &str = "You are Bamboo's background ledger gardener. From the user's completed ledger records, extract only durable, long-term facts worth remembering (habits, recurring obligations, stable preferences, notable life events). Return only the specified JSON array. No prose, no markdown fences.";
41
42pub struct LedgerGardenerContext {
43    pub dream: AutoDreamContext,
44    pub schedule_bridge: Option<Arc<dyn LedgerScheduleBridge>>,
45}
46
47#[derive(Debug, Clone, PartialEq, Eq, Default)]
48pub struct LedgerGardenerRunResult {
49    pub scanned: usize,
50    pub expired: usize,
51    pub schedules_released: usize,
52    pub schedules_synced: usize,
53    pub distilled_records: usize,
54    pub memories_written: usize,
55    pub failed: usize,
56}
57
58#[derive(Debug, Clone, Deserialize, PartialEq, Eq)]
59pub struct DistilledMemoryCandidate {
60    pub title: String,
61    pub content: String,
62    #[serde(default)]
63    pub r#type: Option<String>,
64    #[serde(default)]
65    pub tags: Vec<String>,
66}
67
68/// Whether the record should transition to `Expired` at `now`. Pure so the
69/// policy is unit-testable: only records whose *entire* time purpose is in the
70/// past expire — events past their end, and reminder-only records whose every
71/// reminder has fired. Anything with a due date, a recurrence, or no time at
72/// all stays open.
73pub fn should_expire(record: &LedgerRecord, now: DateTime<Utc>) -> bool {
74    if record.status.is_terminal() {
75        return false;
76    }
77    if record.time.due_at.is_some() || record.time.recurrence.is_some() {
78        return false;
79    }
80    let cutoff = now - chrono::Duration::hours(EXPIRY_GRACE_HOURS);
81    if let Some(end) = record.time.ends_at.or(record.time.starts_at) {
82        return end < cutoff;
83    }
84    if !record.time.remind_at.is_empty() {
85        return record.time.remind_at.iter().all(|at| *at < cutoff);
86    }
87    false
88}
89
90/// Lenient parse of the distillation model output: the first `[`…`]` slice is
91/// parsed as the candidate array; anything unparseable yields an empty list
92/// (a malformed model reply must never fail the run).
93pub fn parse_distilled_candidates(raw: &str) -> Vec<DistilledMemoryCandidate> {
94    let Some(start) = raw.find('[') else {
95        return Vec::new();
96    };
97    let Some(end) = raw.rfind(']') else {
98        return Vec::new();
99    };
100    if end < start {
101        return Vec::new();
102    }
103    serde_json::from_str::<Vec<DistilledMemoryCandidate>>(&raw[start..=end])
104        .unwrap_or_default()
105        .into_iter()
106        .filter(|candidate| {
107            !candidate.title.trim().is_empty() && !candidate.content.trim().is_empty()
108        })
109        .collect()
110}
111
112fn build_distillation_prompt(docs: &[(String, String)]) -> String {
113    let mut prompt = String::from(
114        "The user completed the following ledger records (todos/events/reminders).\n\
115         Extract durable long-term memories ONLY where a record reveals a lasting fact:\n\
116         a habit or routine, a recurring obligation, a stable preference, or a notable\n\
117         life event worth recalling months from now. Routine one-off chores yield nothing.\n\n\
118         Records:\n",
119    );
120    for (title, detail) in docs {
121        prompt.push_str(&format!("- {title}\n  {detail}\n"));
122    }
123    prompt.push_str(
124        "\nReturn a JSON array (possibly empty), each item:\n\
125         {\"title\": \"specific descriptive title\", \"content\": \"one atomic fact\", \
126         \"type\": \"user|reference\", \"tags\": [\"...\"]}\n",
127    );
128    prompt
129}
130
131fn all_scopes(project_keys: &[String]) -> Vec<(LedgerScope, Option<String>)> {
132    let mut scopes: Vec<(LedgerScope, Option<String>)> = vec![(LedgerScope::Global, None)];
133    scopes.extend(
134        project_keys
135            .iter()
136            .map(|key| (LedgerScope::Project, Some(key.clone()))),
137    );
138    scopes
139}
140
141pub async fn run_ledger_gardener_once(
142    ctx: &LedgerGardenerContext,
143) -> Result<Option<LedgerGardenerRunResult>, String> {
144    let ledger = LedgerStore::new(ctx.dream.session_store.bamboo_home_dir());
145    let memory = ctx.dream.memory.clone();
146    run_ledger_gardener_once_with_stores(ctx, &ledger, &memory).await
147}
148
149async fn run_ledger_gardener_once_with_stores(
150    ctx: &LedgerGardenerContext,
151    ledger: &LedgerStore,
152    memory: &MemoryStore,
153) -> Result<Option<LedgerGardenerRunResult>, String> {
154    let config_snapshot = ctx.dream.config.read().await.clone();
155    let memory_cfg = config_snapshot.memory().clone().unwrap_or_default();
156    if !memory_cfg.ledger_gardener_enabled {
157        return Ok(None);
158    }
159
160    let now = Utc::now();
161    let mut result = LedgerGardenerRunResult::default();
162    let project_keys = ledger
163        .list_project_keys()
164        .await
165        .map_err(|error| format!("ledger gardener scope scan failed: {error}"))?;
166    let scopes = all_scopes(&project_keys);
167
168    // Pass 1+2 share one record sweep per scope.
169    for (scope, project_key) in &scopes {
170        let docs = ledger
171            .list_records(
172                *scope,
173                project_key.as_deref(),
174                &RecordFilter {
175                    include_terminal: true,
176                    ..RecordFilter::default()
177                },
178            )
179            .await
180            .map_err(|error| format!("ledger gardener list failed: {error}"))?;
181        result.scanned += docs.len();
182
183        for doc in docs {
184            let record = &doc.record;
185
186            // Pass 1: expiry.
187            if should_expire(record, now) {
188                match ledger
189                    .transition_record(
190                        *scope,
191                        project_key.as_deref(),
192                        &record.id,
193                        RecordStatus::Expired,
194                        Some("auto-expired by the ledger gardener"),
195                    )
196                    .await
197                {
198                    Ok(Some(expired_doc)) => {
199                        result.expired += 1;
200                        // Release the expired record's schedules right away
201                        // instead of waiting a full interval for the terminal
202                        // branch below. Rewrite from the freshly transitioned
203                        // document so the recorded transition history survives.
204                        if let (Some(bridge), false) =
205                            (&ctx.schedule_bridge, record.schedule_ids.is_empty())
206                        {
207                            if bridge.release_schedules(&record.schedule_ids).await.is_ok() {
208                                let mut cleared = expired_doc.record.clone();
209                                cleared.schedule_ids.clear();
210                                let _ = ledger
211                                    .write_record(cleared, Some(expired_doc.body.clone()))
212                                    .await;
213                                result.schedules_released += record.schedule_ids.len();
214                            }
215                        }
216                        continue;
217                    }
218                    Ok(None) => {}
219                    Err(error) => {
220                        result.failed += 1;
221                        tracing::warn!(
222                            target: LEDGER_GARDENER_TRACING_TARGET,
223                            record_id = %record.id,
224                            "[ledger-gardener] expiry transition failed: {error}"
225                        );
226                        continue;
227                    }
228                }
229            }
230
231            // Pass 2: schedule reconciliation.
232            let Some(bridge) = &ctx.schedule_bridge else {
233                continue;
234            };
235            if record.status.is_terminal() && !record.schedule_ids.is_empty() {
236                match bridge.release_schedules(&record.schedule_ids).await {
237                    Ok(()) => {
238                        let mut cleared = record.clone();
239                        cleared.schedule_ids.clear();
240                        if ledger
241                            .write_record(cleared, Some(doc.body.clone()))
242                            .await
243                            .is_ok()
244                        {
245                            result.schedules_released += record.schedule_ids.len();
246                        }
247                    }
248                    Err(error) => {
249                        result.failed += 1;
250                        tracing::warn!(
251                            target: LEDGER_GARDENER_TRACING_TARGET,
252                            record_id = %record.id,
253                            "[ledger-gardener] schedule release failed: {error}"
254                        );
255                    }
256                }
257            } else if !record.status.is_terminal()
258                && record.schedule_ids.is_empty()
259                && (record.time.recurrence.is_some()
260                    || record.time.remind_at.iter().any(|at| *at > now))
261            {
262                match bridge.sync_record_schedules(record).await {
263                    Ok(ids) if !ids.is_empty() => {
264                        let mut updated = record.clone();
265                        updated.schedule_ids = ids.clone();
266                        if ledger
267                            .write_record(updated, Some(doc.body.clone()))
268                            .await
269                            .is_ok()
270                        {
271                            result.schedules_synced += ids.len();
272                        }
273                    }
274                    Ok(_) => {}
275                    Err(error) => {
276                        result.failed += 1;
277                        tracing::warn!(
278                            target: LEDGER_GARDENER_TRACING_TARGET,
279                            record_id = %record.id,
280                            "[ledger-gardener] schedule sync failed: {error}"
281                        );
282                    }
283                }
284            }
285        }
286    }
287
288    // Pass 3: distillation (the only LLM arm; free when nothing completed).
289    if memory_cfg.ledger_distillation_enabled {
290        if let Err(error) = run_distillation_pass(ctx, ledger, memory, &scopes, &mut result).await {
291            result.failed += 1;
292            tracing::warn!(
293                target: LEDGER_GARDENER_TRACING_TARGET,
294                "[ledger-gardener] distillation failed: {error}"
295            );
296        }
297    }
298
299    tracing::info!(
300        target: LEDGER_GARDENER_TRACING_TARGET,
301        event = "run_complete",
302        scanned = result.scanned,
303        expired = result.expired,
304        schedules_released = result.schedules_released,
305        schedules_synced = result.schedules_synced,
306        distilled_records = result.distilled_records,
307        memories_written = result.memories_written,
308        failed = result.failed,
309        "[ledger-gardener] run complete"
310    );
311    Ok(Some(result))
312}
313
314async fn run_distillation_pass(
315    ctx: &LedgerGardenerContext,
316    ledger: &LedgerStore,
317    memory: &MemoryStore,
318    scopes: &[(LedgerScope, Option<String>)],
319    result: &mut LedgerGardenerRunResult,
320) -> Result<(), String> {
321    // Deterministic prefilter: Done records not yet distilled.
322    let mut pending = Vec::new();
323    for (scope, project_key) in scopes {
324        let docs = ledger
325            .list_records(
326                *scope,
327                project_key.as_deref(),
328                &RecordFilter {
329                    statuses: Some([RecordStatus::Done].into_iter().collect()),
330                    include_terminal: true,
331                    ..RecordFilter::default()
332                },
333            )
334            .await
335            .map_err(|error| format!("distillation list failed: {error}"))?;
336        pending.extend(
337            docs.into_iter()
338                .filter(|doc| !doc.record.tags.iter().any(|tag| tag == DISTILLED_TAG)),
339        );
340        if pending.len() >= DISTILL_MAX_RECORDS_PER_RUN {
341            break;
342        }
343    }
344    pending.truncate(DISTILL_MAX_RECORDS_PER_RUN);
345    if pending.is_empty() {
346        return Ok(());
347    }
348
349    let config_snapshot = ctx.dream.config.read().await.clone();
350    let Some((provider, model)) = resolve_background_model(&ctx.dream, &config_snapshot) else {
351        // No background model → leave records unmarked so a later configured
352        // model can still distill them.
353        return Ok(());
354    };
355
356    let lines: Vec<(String, String)> = pending
357        .iter()
358        .map(|doc| {
359            let record = &doc.record;
360            let detail = format!(
361                "kind={}, completed={}, notes: {}",
362                record.kind.as_str(),
363                record.updated_at.format("%Y-%m-%d"),
364                doc.body
365                    .chars()
366                    .take(200)
367                    .collect::<String>()
368                    .replace('\n', " "),
369            );
370            (record.title.clone(), detail)
371        })
372        .collect();
373    let raw = collect_model_json(
374        provider,
375        &model,
376        DISTILL_SYSTEM_INSTRUCTION,
377        build_distillation_prompt(&lines),
378    )
379    .await?;
380    let candidates = parse_distilled_candidates(&raw);
381
382    for candidate in &candidates {
383        let r#type = match candidate.r#type.as_deref() {
384            Some("reference") => DurableMemoryType::Reference,
385            _ => DurableMemoryType::User,
386        };
387        match memory
388            .write_memory(
389                bamboo_memory::memory_store::MemoryScope::Global,
390                None,
391                r#type,
392                candidate.title.trim(),
393                candidate.content.trim(),
394                &candidate.tags,
395                None,
396                "ledger-gardener",
397                true, // merge-if-similar: repeated habits reinforce one memory
398                None,
399            )
400            .await
401        {
402            Ok(_) => result.memories_written += 1,
403            Err(error) => {
404                result.failed += 1;
405                tracing::warn!(
406                    target: LEDGER_GARDENER_TRACING_TARGET,
407                    "[ledger-gardener] distilled memory write failed: {error}"
408                );
409            }
410        }
411    }
412
413    // Mark every considered record — including ones yielding no memory — so
414    // the same records are never re-fed to the model.
415    for doc in pending {
416        let mut record = doc.record.clone();
417        record.tags.push(DISTILLED_TAG.to_string());
418        if ledger
419            .write_record(record, Some(doc.body.clone()))
420            .await
421            .is_ok()
422        {
423            result.distilled_records += 1;
424        }
425    }
426    Ok(())
427}
428
429/// Spawn the recurring ledger gardener loop: one run at startup, then on the
430/// configured interval. Disabled config makes each run return immediately.
431pub fn spawn_ledger_gardener_task(ctx: LedgerGardenerContext) {
432    tokio::spawn(async move {
433        let interval_secs = {
434            let guard = ctx.dream.config.read().await;
435            guard
436                .memory()
437                .as_ref()
438                .map(|memory| memory.ledger_gardener_interval_secs)
439                .filter(|secs| *secs > 0)
440                .unwrap_or_else(|| {
441                    bamboo_config::MemoryConfig::default().ledger_gardener_interval_secs
442                })
443        };
444        // Short poll so a config flip or startup work is picked up promptly;
445        // the interval gate below decides whether a poll becomes a run.
446        let poll_secs = 300.min(interval_secs.max(1));
447        let mut ticker = tokio::time::interval(Duration::from_secs(poll_secs));
448        let mut last_run: Option<Instant> = None;
449
450        loop {
451            ticker.tick().await;
452            let due = last_run
453                .map(|at| at.elapsed().as_secs() >= interval_secs)
454                .unwrap_or(true);
455            if !due {
456                continue;
457            }
458            if let Err(error) = run_ledger_gardener_once(&ctx).await {
459                tracing::warn!(
460                    target: LEDGER_GARDENER_TRACING_TARGET,
461                    event = "run_failed",
462                    "[ledger-gardener] run failed: {error}"
463                );
464            }
465            last_run = Some(Instant::now());
466        }
467    });
468}
469
470#[cfg(test)]
471mod tests {
472    use super::*;
473    use std::collections::HashMap;
474    use std::sync::Mutex;
475
476    use async_trait::async_trait;
477    use bamboo_agent_core::Message;
478    use bamboo_domain::ledger::RecordKind;
479    use bamboo_domain::schedule::ScheduleTrigger;
480    use bamboo_llm::{Config, LLMChunk, LLMError, LLMProvider, LLMStream, ProviderRegistry};
481    use bamboo_storage::SessionStoreV2;
482    use chrono::Duration as ChronoDuration;
483    use futures::stream;
484    use tokio::sync::{Mutex as AsyncMutex, RwLock};
485
486    fn config_with_memory(memory: bamboo_config::MemoryConfig) -> Config {
487        let mut config = Config::default();
488        *config.memory_mut() = Some(memory);
489        config
490    }
491
492    fn record(kind: RecordKind, status: RecordStatus) -> LedgerRecord {
493        let mut record = LedgerRecord::new("rec_test", kind, "Test record");
494        record.status = status;
495        record
496    }
497
498    #[test]
499    fn expiry_policy_expires_only_fully_past_events_and_reminders() {
500        let now = Utc::now();
501        let past = now - ChronoDuration::hours(EXPIRY_GRACE_HOURS + 1);
502        let recent_past = now - ChronoDuration::hours(1);
503
504        // Event fully past the grace window → expires.
505        let mut event = record(RecordKind::Event, RecordStatus::Open);
506        event.time.starts_at = Some(past);
507        assert!(should_expire(&event, now));
508
509        // Event within the grace window → survives.
510        let mut recent_event = record(RecordKind::Event, RecordStatus::Open);
511        recent_event.time.starts_at = Some(recent_past);
512        assert!(!should_expire(&recent_event, now));
513
514        // Reminder with every remind_at past → expires.
515        let mut reminder = record(RecordKind::Reminder, RecordStatus::Open);
516        reminder.time.remind_at = vec![past, past - ChronoDuration::hours(2)];
517        assert!(should_expire(&reminder, now));
518
519        // One future reminder keeps it alive.
520        let mut live_reminder = record(RecordKind::Reminder, RecordStatus::Open);
521        live_reminder.time.remind_at = vec![past, now + ChronoDuration::hours(1)];
522        assert!(!should_expire(&live_reminder, now));
523
524        // A due date means it's open work — never auto-expired.
525        let mut todo = record(RecordKind::Todo, RecordStatus::Open);
526        todo.time.due_at = Some(past);
527        assert!(!should_expire(&todo, now));
528
529        // Recurrence keeps a record alive even with past reminders.
530        let mut habit = record(RecordKind::Habit, RecordStatus::Open);
531        habit.time.remind_at = vec![past];
532        habit.time.recurrence = Some(ScheduleTrigger::Daily {
533            hour: 9,
534            minute: 0,
535            second: 0,
536        });
537        assert!(!should_expire(&habit, now));
538
539        // Terminal and undated records are untouched.
540        assert!(!should_expire(
541            &record(RecordKind::Todo, RecordStatus::Done),
542            now
543        ));
544        assert!(!should_expire(
545            &record(RecordKind::Todo, RecordStatus::Open),
546            now
547        ));
548    }
549
550    #[test]
551    fn parse_distilled_candidates_is_lenient() {
552        let wrapped = "Sure! Here you go:\n```json\n[{\"title\": \"Takes medication daily\", \"content\": \"User takes medication every morning at 9.\", \"type\": \"user\", \"tags\": [\"health\"]}]\n```";
553        let parsed = parse_distilled_candidates(wrapped);
554        assert_eq!(parsed.len(), 1);
555        assert_eq!(parsed[0].title, "Takes medication daily");
556
557        assert!(parse_distilled_candidates("no json here").is_empty());
558        assert!(parse_distilled_candidates("[]").is_empty());
559        assert!(parse_distilled_candidates("[{\"title\": \" \", \"content\": \"x\"}]").is_empty());
560        assert!(parse_distilled_candidates("[{broken").is_empty());
561    }
562
563    #[test]
564    fn distillation_prompt_lists_records_and_requests_json() {
565        let prompt = build_distillation_prompt(&[(
566            "Renew passport".to_string(),
567            "kind=todo, completed=2026-07-13".to_string(),
568        )]);
569        assert!(prompt.contains("Renew passport"));
570        assert!(prompt.contains("JSON array"));
571    }
572
573    #[derive(Clone)]
574    struct CannedProvider {
575        responses: Arc<Mutex<Vec<String>>>,
576    }
577
578    #[async_trait]
579    impl LLMProvider for CannedProvider {
580        async fn chat_stream(
581            &self,
582            _messages: &[Message],
583            _tools: &[bamboo_agent_core::tools::ToolSchema],
584            _max_output_tokens: Option<u32>,
585            _model: &str,
586        ) -> Result<LLMStream, LLMError> {
587            let mut responses = self.responses.lock().expect("lock poisoned");
588            let text = if responses.is_empty() {
589                "[]".to_string()
590            } else {
591                responses.remove(0)
592            };
593            Ok(Box::pin(stream::iter(vec![
594                Ok(LLMChunk::Token(text)),
595                Ok(LLMChunk::Done),
596            ])))
597        }
598    }
599
600    #[derive(Default)]
601    struct RecordingBridge {
602        synced: AsyncMutex<Vec<String>>,
603        released: AsyncMutex<Vec<String>>,
604    }
605
606    #[async_trait]
607    impl LedgerScheduleBridge for RecordingBridge {
608        async fn sync_record_schedules(
609            &self,
610            record: &LedgerRecord,
611        ) -> Result<Vec<String>, String> {
612            self.synced.lock().await.push(record.id.clone());
613            Ok(vec![format!("sched_for_{}", record.id)])
614        }
615
616        async fn release_schedules(&self, schedule_ids: &[String]) -> Result<(), String> {
617            self.released
618                .lock()
619                .await
620                .extend(schedule_ids.iter().cloned());
621            Ok(())
622        }
623    }
624
625    #[tokio::test]
626    async fn full_run_expires_reconciles_and_distills() {
627        let temp = tempfile::tempdir().expect("tempdir");
628        let session_store = Arc::new(
629            SessionStoreV2::new(temp.path().to_path_buf())
630                .await
631                .unwrap(),
632        );
633        let ledger = LedgerStore::new(session_store.bamboo_home_dir());
634        let jiandu_root = temp.path().join("jiandu");
635        let memory = MemoryStore::new(&jiandu_root);
636        let now = Utc::now();
637
638        // 1) A long-past event holding a schedule: must expire + release.
639        let mut past_event = LedgerRecord::new("rec_event", RecordKind::Event, "Old meetup");
640        past_event.time.starts_at = Some(now - ChronoDuration::hours(EXPIRY_GRACE_HOURS + 10));
641        past_event.schedule_ids = vec!["sched_old".to_string()];
642        ledger.write_record(past_event, None).await.unwrap();
643
644        // 2) An open reminder with a future time but no managed schedule
645        //    (crash drift): must get schedules re-synced.
646        let mut drifted = LedgerRecord::new("rec_drift", RecordKind::Reminder, "Call the bank");
647        drifted.time.remind_at = vec![now + ChronoDuration::hours(5)];
648        ledger.write_record(drifted, None).await.unwrap();
649
650        // 3) A completed record: must be distilled into a durable memory and
651        //    tagged so it is never re-fed to the model.
652        let mut done = LedgerRecord::new("rec_done", RecordKind::Habit, "Morning run streak");
653        done.status = RecordStatus::Done;
654        ledger.write_record(done, None).await.unwrap();
655
656        let provider: Arc<dyn LLMProvider> = Arc::new(CannedProvider {
657            responses: Arc::new(Mutex::new(vec![
658                "[{\"title\": \"Runs every morning\", \"content\": \"User keeps a morning run habit.\", \"type\": \"user\", \"tags\": [\"health\"]}]".to_string(),
659            ])),
660        });
661        let config = Arc::new(RwLock::new(config_with_memory(
662            bamboo_config::MemoryConfig {
663                background_model: Some("fast-model".to_string()),
664                ..bamboo_config::MemoryConfig::default()
665            },
666        )));
667        let bridge = Arc::new(RecordingBridge::default());
668        let ctx = LedgerGardenerContext {
669            dream: AutoDreamContext {
670                session_store: session_store.clone(),
671                storage: session_store.clone(),
672                memory: memory.clone(),
673                provider,
674                config,
675                provider_registry: Arc::new(ProviderRegistry::new(
676                    HashMap::new(),
677                    "test".to_string(),
678                )),
679            },
680            schedule_bridge: Some(bridge.clone()),
681        };
682
683        let result = run_ledger_gardener_once_with_stores(&ctx, &ledger, &memory)
684            .await
685            .unwrap()
686            .expect("gardener enabled by default");
687
688        assert_eq!(result.expired, 1);
689        assert_eq!(result.schedules_released, 1);
690        assert_eq!(result.schedules_synced, 1);
691        assert_eq!(result.memories_written, 1);
692        assert_eq!(result.distilled_records, 1);
693        assert_eq!(result.failed, 0);
694
695        let expired = ledger
696            .get_record(LedgerScope::Global, None, "rec_event")
697            .await
698            .unwrap()
699            .unwrap();
700        assert_eq!(expired.record.status, RecordStatus::Expired);
701        assert!(expired.record.schedule_ids.is_empty());
702        assert!(
703            !expired.record.transitions.is_empty(),
704            "history must survive"
705        );
706        assert_eq!(*bridge.released.lock().await, vec!["sched_old".to_string()]);
707
708        let drifted = ledger
709            .get_record(LedgerScope::Global, None, "rec_drift")
710            .await
711            .unwrap()
712            .unwrap();
713        assert_eq!(drifted.record.schedule_ids, vec!["sched_for_rec_drift"]);
714
715        let done = ledger
716            .get_record(LedgerScope::Global, None, "rec_done")
717            .await
718            .unwrap()
719            .unwrap();
720        assert!(done.record.tags.iter().any(|tag| tag == DISTILLED_TAG));
721
722        // A second run finds nothing new to do.
723        let second = run_ledger_gardener_once_with_stores(&ctx, &ledger, &memory)
724            .await
725            .unwrap()
726            .unwrap();
727        assert_eq!(second.expired, 0);
728        assert_eq!(second.distilled_records, 0);
729        assert_eq!(second.memories_written, 0);
730        assert!(
731            !jiandu_root.join("ledger").exists(),
732            "Jiandu memory root must never host Bamboo Ledger data"
733        );
734    }
735}