use anyhow::{anyhow, Result};
use std::path::{Path, PathBuf};
use std::time::Duration;
use tracing::{debug, error, info, warn};
use crate::cognitive::dream::{
DreamConfig, DreamPhase, DreamResult, DreamState, MemoryEntry, MemoryStore,
};
#[derive(Debug, Clone)]
pub struct AutoDreamConfig {
pub model: String,
pub endpoint: String,
pub api_key: Option<String>,
pub timeout_secs: u64,
pub llm_timeout_secs: u64,
pub prefer_local: bool,
}
impl Default for AutoDreamConfig {
fn default() -> Self {
Self {
model: "qwen3.5-9b".to_string(), endpoint: "http://localhost:8000/v1".to_string(),
api_key: None,
timeout_secs: 300, llm_timeout_secs: 60,
prefer_local: true,
}
}
}
impl AutoDreamConfig {
pub fn new() -> Self {
Self::default()
}
pub fn with_model(mut self, model: impl Into<String>) -> Self {
self.model = model.into();
self
}
pub fn with_endpoint(mut self, endpoint: impl Into<String>) -> Self {
self.endpoint = endpoint.into();
self
}
pub fn with_api_key(mut self, key: impl Into<String>) -> Self {
self.api_key = Some(key.into());
self
}
pub fn from_user_config() -> Self {
match crate::config::Config::load(None) {
Ok(cfg) => Self {
model: cfg.model,
endpoint: cfg.endpoint,
api_key: cfg.api_key.as_ref().map(|k| k.expose().to_string()),
..Self::default()
},
Err(e) => {
warn!(
"autoDream: could not load user config ({}); using built-in defaults",
e
);
Self::default()
}
}
}
}
pub struct AutoDreamHandle {
task: tokio::task::JoinHandle<Result<DreamResult>>,
project_key: String,
start_time: std::time::Instant,
}
impl AutoDreamHandle {
pub async fn is_running(&mut self) -> bool {
!self.task.is_finished()
}
pub async fn wait_with_timeout(&mut self, timeout: Duration) -> Result<DreamResult> {
match tokio::time::timeout(timeout, &mut self.task).await {
Ok(Ok(Ok(result))) => {
let duration = self.start_time.elapsed().as_secs();
if result.success {
info!(
"AutoDream completed successfully for {} in {}s",
self.project_key, duration
);
} else {
warn!(
"AutoDream completed with errors for {}: {:?}",
self.project_key, result.errors
);
}
Ok(result)
}
Ok(Ok(Err(e))) => {
error!("AutoDream failed for {}: {}", self.project_key, e);
Ok(DreamResult::failure(format!("{}", e)))
}
Ok(Err(join_err)) => {
error!(
"AutoDream task join error for {}: {}",
self.project_key, join_err
);
Err(anyhow!("Dream task join error: {}", join_err))
}
Err(_) => {
warn!("AutoDream timed out for {}", self.project_key);
self.task.abort();
Ok(DreamResult::failure("Dream process timed out"))
}
}
}
pub async fn kill(&mut self) -> Result<()> {
info!("Killing autoDream task for {}", self.project_key);
self.task.abort();
Ok(())
}
}
pub async fn spawn_autodream(
project_path: &Path,
project_key: &str,
config: &AutoDreamConfig,
dream_config: &DreamConfig,
) -> Result<AutoDreamHandle> {
info!(
"Spawning autoDream background task for {} at {:?}",
project_key, project_path
);
let project_path = project_path.to_path_buf();
let project_key_owned = project_key.to_string();
let config = config.clone();
let dream_config = dream_config.clone();
let task = tokio::spawn(async move {
run_dream_consolidation(&project_path, &project_key_owned, &config, &dream_config).await
});
Ok(AutoDreamHandle {
task,
project_key: project_key.to_string(),
start_time: std::time::Instant::now(),
})
}
pub async fn run_dream_consolidation(
project_path: &Path,
project_key: &str,
config: &AutoDreamConfig,
dream_config: &DreamConfig,
) -> Result<DreamResult> {
let start_time = std::time::Instant::now();
let mut phases_completed = Vec::new();
let mut errors = Vec::new();
info!(
"Starting in-process dream consolidation for {}",
project_key
);
info!("Dream Phase 1: Orient");
let session_files = match orient_phase(project_path).await {
Ok(files) => {
debug!("Found {} session files to process", files.len());
phases_completed.push(DreamPhase::Orient);
files
}
Err(e) => {
error!("Orient phase failed: {}", e);
errors.push(format!("Orient: {}", e));
Vec::new()
}
};
info!("Dream Phase 2: Gather Recent Signal");
let memories = match gather_phase(&session_files, 5).await {
Ok(mem) => {
debug!("Gathered {} memories", mem.len());
phases_completed.push(DreamPhase::Gather);
mem
}
Err(e) => {
error!("Gather phase failed: {}", e);
errors.push(format!("Gather: {}", e));
Vec::new()
}
};
info!("Dream Phase 3: Consolidate");
let consolidation = match consolidate_phase(&memories, config).await {
Ok(content) => {
debug!("Consolidated {} memories", memories.len());
phases_completed.push(DreamPhase::Consolidate);
Some(content)
}
Err(e) => {
error!("Consolidate phase failed: {}", e);
errors.push(format!("Consolidate: {}", e));
None
}
};
let consolidated_count = consolidation
.as_deref()
.map(count_consolidated_entries)
.unwrap_or(0);
info!("Dream Phase 4: Prune & Index");
let pruned_count =
match prune_and_index_phase(project_key, dream_config, consolidation.as_deref()).await {
Ok(count) => {
debug!("Pruned {} memories", count);
phases_completed.push(DreamPhase::PruneAndIndex);
count
}
Err(e) => {
error!("Prune & Index phase failed: {}", e);
errors.push(format!("Prune & Index: {}", e));
0
}
};
let duration = start_time.elapsed().as_secs();
let consolidate_failed = !memories.is_empty() && consolidation.is_none();
let success = !consolidate_failed && (errors.is_empty() || phases_completed.len() >= 3);
info!(
"Dream consolidation completed for {}: {} phases, {} consolidated, {} pruned, {}s",
project_key,
phases_completed.len(),
consolidated_count,
pruned_count,
duration
);
Ok(DreamResult {
success,
phases_completed,
memories_consolidated: consolidated_count,
memories_pruned: pruned_count,
errors,
duration_secs: duration,
})
}
async fn orient_phase(project_path: &Path) -> Result<Vec<PathBuf>> {
let mut session_files = Vec::new();
let selfware_dir = project_path.join(".selfware");
if selfware_dir.exists() {
let mut entries = tokio::fs::read_dir(&selfware_dir).await?;
while let Some(entry) = entries.next_entry().await? {
let path = entry.path();
if let Some(ext) = path.extension() {
if ext == "json" || ext == "jsonl" {
let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
if name.contains("session") || name.contains("memory") {
session_files.push(path);
}
}
}
}
}
session_files.sort_by(|a, b| {
let meta_a = std::fs::metadata(a).ok();
let meta_b = std::fs::metadata(b).ok();
match (meta_a, meta_b) {
(Some(ma), Some(mb)) => {
let time_a = ma.modified().ok();
let time_b = mb.modified().ok();
time_b.cmp(&time_a) }
_ => std::cmp::Ordering::Equal,
}
});
Ok(session_files)
}
async fn gather_phase(session_files: &[PathBuf], max_sessions: usize) -> Result<Vec<MemoryEntry>> {
use crate::cognitive::dream::MemorySection;
let mut all_memories = Vec::new();
for file in session_files.iter().take(max_sessions) {
debug!("Gathering from session file: {:?}", file);
let content = match tokio::fs::read_to_string(file).await {
Ok(c) => c,
Err(e) => {
warn!("Failed to read session file {:?}: {}", file, e);
continue;
}
};
for line in content.lines() {
if let Some(entry) = MemoryEntry::parse(line, &MemorySection::Facts) {
all_memories.push(entry);
}
}
}
Ok(all_memories)
}
async fn consolidate_phase(memories: &[MemoryEntry], config: &AutoDreamConfig) -> Result<String> {
if memories.is_empty() {
return Ok(String::new());
}
let prompt = crate::cognitive::dream::generate_consolidation_prompt(memories);
let agent = crate::config::AgentConfig {
step_timeout_secs: config.llm_timeout_secs.max(60),
..Default::default()
};
let client_config = crate::config::Config {
endpoint: config.endpoint.clone(),
model: config.model.clone(),
api_key: config
.api_key
.clone()
.filter(|k| !k.is_empty())
.map(crate::config::RedactedString::new),
temperature: 0.3,
max_tokens: 2048,
agent,
..Default::default()
};
let client = crate::api::ApiClient::new(&client_config)?;
let response = client
.chat(
vec![crate::api::Message::user(prompt)],
None,
crate::api::ThinkingMode::Disabled,
)
.await?;
let content = response
.choices
.first()
.map(|c| c.message.content.text().to_string())
.unwrap_or_default();
if content.trim().is_empty() {
return Err(anyhow!(
"consolidation LLM ({}) returned empty content",
config.model
));
}
info!(
"consolidated {} memories via {} ({})",
memories.len(),
config.model,
config.endpoint
);
Ok(content)
}
fn count_consolidated_entries(content: &str) -> usize {
let mut in_section = false;
let mut count = 0usize;
for line in content.lines() {
let trimmed = line.trim();
if trimmed.starts_with("## ") {
in_section = true;
continue;
}
if in_section && trimmed.starts_with("- ") {
count += 1;
}
}
count
}
async fn prune_and_index_phase(
project_key: &str,
dream_config: &DreamConfig,
consolidated: Option<&str>,
) -> Result<usize> {
let memory_path = dream_config.memory_file_path(project_key);
let mut store = if memory_path.exists() {
let content = tokio::fs::read_to_string(&memory_path).await?;
MemoryStore::parse(&content)
} else {
MemoryStore {
entries: Vec::new(),
sections: std::collections::HashMap::new(),
}
};
if let Some(content) = consolidated {
let incoming = MemoryStore::parse(content);
let mut added = 0usize;
for entry in incoming.entries {
let is_dup = store.entries.iter().any(|e| e.content == entry.content);
if !is_dup {
let idx = store.entries.len();
store
.sections
.entry(entry.section.clone())
.or_default()
.push(idx);
store.entries.push(entry);
added += 1;
}
}
debug!("Merged {} consolidated memories into MEMORY.md", added);
}
let pruned_stale = store.prune_stale(dream_config.stale_memory_days);
debug!("Pruned {} stale memories", pruned_stale);
let pruned_size = store.cap_size(dream_config.max_memory_lines, dream_config.max_memory_size);
debug!("Pruned {} memories for size constraints", pruned_size);
let output = store.format();
if let Some(parent) = memory_path.parent() {
tokio::fs::create_dir_all(parent).await?;
}
let temp_path = memory_path.with_extension(format!("tmp.{}", std::process::id()));
tokio::fs::write(&temp_path, output).await?;
tokio::fs::rename(&temp_path, &memory_path).await?;
let stats = store.stats();
info!(
"MEMORY.md updated: {} entries, {} lines, {} bytes",
stats.total_entries, stats.total_lines, stats.total_bytes
);
Ok(pruned_stale + pruned_size)
}
pub async fn check_and_spawn_autodream(
project_path: &Path,
project_key: &str,
auto_config: &AutoDreamConfig,
dream_config: &DreamConfig,
) -> Result<Option<AutoDreamHandle>> {
let mut state = DreamState::load(&dream_config.state_path())?;
state.record_session_end();
let trigger = dream_config.trigger.clone();
if !crate::cognitive::dream::should_run_dream(&mut state, &trigger) {
state.save(&dream_config.state_path())?;
debug!(
"AutoDream gates not passed for {}: {} sessions, last dream {} hours ago",
project_key,
state.sessions_since_last_dream,
(std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
- state.last_dream_timestamp)
/ 3600
);
return Ok(None);
}
let handle = spawn_autodream(project_path, project_key, auto_config, dream_config).await?;
state.save(&dream_config.state_path())?;
info!("AutoDream spawned for {}", project_key);
Ok(Some(handle))
}
pub async fn get_dream_status(dream_config: &DreamConfig) -> crate::cognitive::dream::DreamStatus {
let state = DreamState::load(&dream_config.state_path()).unwrap_or_default();
let trigger = &dream_config.trigger;
crate::cognitive::dream::DreamStatus {
last_dream_timestamp: if state.last_dream_timestamp > 0 {
Some(state.last_dream_timestamp)
} else {
None
},
sessions_since_last_dream: state.sessions_since_last_dream,
dream_count: state.dream_count,
hours_until_next: state.hours_until_next(trigger),
sessions_until_next: state.sessions_until_next(trigger),
is_running: state.consolidation_lock,
}
}
#[cfg(test)]
#[path = "../../tests/unit/cognitive/dream_subprocess/dream_subprocess_test.rs"]
mod tests;