1use 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";
32const EXPIRY_GRACE_HOURS: i64 = 48;
35const DISTILL_MAX_RECORDS_PER_RUN: usize = 8;
37const 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
68pub 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
90pub 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 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 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 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 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 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 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 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, 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 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
429pub 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 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 let mut event = record(RecordKind::Event, RecordStatus::Open);
506 event.time.starts_at = Some(past);
507 assert!(should_expire(&event, now));
508
509 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 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 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 let mut todo = record(RecordKind::Todo, RecordStatus::Open);
526 todo.time.due_at = Some(past);
527 assert!(!should_expire(&todo, now));
528
529 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 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 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 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 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 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}