Subconscious: persistent agent state, store activity queries

- Agent state (outputs) persists across runs in subconscious-state.json,
  loaded on startup, saved after each run completes
- Merge semantics: each run's outputs accumulate into persistent_state
  rather than replacing
- Walked keys restored from surface agent state on load
- Store::recent_by_provenance() queries nodes by agent provenance for
  the store activity view
- Switch outputs from HashMap to BTreeMap for stable display ordering

Co-Authored-By: Proof of Concept <poc@bcachefs.org>
This commit is contained in:
Kent Overstreet 2026-04-07 19:03:05 -04:00
parent cf1c64f936
commit edfa1c37f5
3 changed files with 91 additions and 6 deletions

View file

@ -55,7 +55,7 @@ pub struct AutoAgent {
priority: i32, priority: i32,
/// Named outputs from the agent's output() tool calls. /// Named outputs from the agent's output() tool calls.
/// Collected per-run, read by Mind after completion. /// Collected per-run, read by Mind after completion.
pub outputs: std::collections::HashMap<String, String>, pub outputs: std::collections::BTreeMap<String, String>,
// Observable status // Observable status
pub current_phase: String, pub current_phase: String,
pub turn: usize, pub turn: usize,
@ -152,7 +152,7 @@ impl AutoAgent {
temperature, top_p: 0.95, top_k: 20, temperature, top_p: 0.95, top_k: 20,
}, },
priority, priority,
outputs: std::collections::HashMap::new(), outputs: std::collections::BTreeMap::new(),
current_phase: String::new(), current_phase: String::new(),
turn: 0, turn: 0,
} }

View file

@ -38,6 +38,17 @@ impl Store {
Ok(()) Ok(())
} }
/// Recent nodes by provenance, sorted newest-first. Returns (key, timestamp).
pub fn recent_by_provenance(&self, provenance: &str, limit: usize) -> Vec<(String, i64)> {
let mut nodes: Vec<_> = self.nodes.values()
.filter(|n| !n.deleted && n.provenance == provenance)
.map(|n| (n.key.clone(), n.timestamp))
.collect();
nodes.sort_by(|a, b| b.1.cmp(&a.1));
nodes.truncate(limit);
nodes
}
/// Upsert a node: update if exists (and content changed), create if not. /// Upsert a node: update if exists (and content changed), create if not.
/// Returns: "created", "updated", or "unchanged". /// Returns: "created", "updated", or "unchanged".
/// ///

View file

@ -298,6 +298,12 @@ pub struct SubconsciousSnapshot {
pub forked_agent: Option<Arc<tokio::sync::Mutex<crate::agent::Agent>>>, pub forked_agent: Option<Arc<tokio::sync::Mutex<crate::agent::Agent>>>,
/// Entry index where the fork diverged. /// Entry index where the fork diverged.
pub fork_point: usize, pub fork_point: usize,
/// Persistent agent state — accumulated across runs.
pub state: std::collections::BTreeMap<String, String>,
/// Persistent walked keys (shared state, relevant for surface).
pub walked: Vec<String>,
/// Recent store activity for this agent: (key, timestamp), newest first.
pub history: Vec<(String, i64)>,
} }
struct SubconsciousAgent { struct SubconsciousAgent {
@ -311,6 +317,8 @@ struct SubconsciousAgent {
/// Entry index where the fork diverged from the conscious agent. /// Entry index where the fork diverged from the conscious agent.
fork_point: usize, fork_point: usize,
handle: Option<tokio::task::JoinHandle<(AutoAgent, Result<String, String>)>>, handle: Option<tokio::task::JoinHandle<(AutoAgent, Result<String, String>)>>,
/// Persistent state — accumulated across runs, serialized to disk.
persistent_state: std::collections::BTreeMap<String, String>,
} }
impl SubconsciousAgent { impl SubconsciousAgent {
@ -340,6 +348,7 @@ impl SubconsciousAgent {
name: name.to_string(), name: name.to_string(),
auto, last_trigger_bytes: 0, last_run: None, auto, last_trigger_bytes: 0, last_run: None,
forked_agent: None, fork_point: 0, handle: None, forked_agent: None, fork_point: 0, handle: None,
persistent_state: std::collections::BTreeMap::new(),
}) })
} }
@ -353,7 +362,14 @@ impl SubconsciousAgent {
conversation_bytes.saturating_sub(self.last_trigger_bytes) >= interval conversation_bytes.saturating_sub(self.last_trigger_bytes) >= interval
} }
fn snapshot(&self) -> SubconsciousSnapshot { fn snapshot(&self, walked: &[String], history: Vec<(String, i64)>) -> SubconsciousSnapshot {
// Merge persistent state with any live outputs from a running agent
let mut state = self.persistent_state.clone();
if self.is_running() {
for (k, v) in &self.auto.outputs {
state.insert(k.clone(), v.clone());
}
}
SubconsciousSnapshot { SubconsciousSnapshot {
name: self.name.clone(), name: self.name.clone(),
running: self.is_running(), running: self.is_running(),
@ -362,6 +378,9 @@ impl SubconsciousAgent {
last_run_secs_ago: self.last_run.map(|t| t.elapsed().as_secs_f64()), last_run_secs_ago: self.last_run.map(|t| t.elapsed().as_secs_f64()),
forked_agent: self.forked_agent.clone(), forked_agent: self.forked_agent.clone(),
fork_point: self.fork_point, fork_point: self.fork_point,
state,
walked: walked.to_vec(),
history,
} }
} }
} }
@ -371,6 +390,7 @@ impl SubconsciousAgent {
pub struct Subconscious { pub struct Subconscious {
agents: Vec<SubconsciousAgent>, agents: Vec<SubconsciousAgent>,
pub walked: Vec<String>, pub walked: Vec<String>,
state_path: Option<std::path::PathBuf>,
} }
impl Subconscious { impl Subconscious {
@ -378,11 +398,57 @@ impl Subconscious {
let agents = AGENTS.iter() let agents = AGENTS.iter()
.filter_map(|(name, _)| SubconsciousAgent::new(name)) .filter_map(|(name, _)| SubconsciousAgent::new(name))
.collect(); .collect();
Self { agents, walked: Vec::new() } Self { agents, walked: Vec::new(), state_path: None }
} }
pub fn snapshots(&self) -> Vec<SubconsciousSnapshot> { /// Set the state file path and load any existing state from disk.
self.agents.iter().map(|s| s.snapshot()).collect() pub fn set_state_path(&mut self, path: std::path::PathBuf) {
if let Ok(data) = std::fs::read_to_string(&path) {
if let Ok(saved) = serde_json::from_str::<
std::collections::BTreeMap<String, std::collections::BTreeMap<String, String>>
>(&data) {
for agent in &mut self.agents {
if let Some(state) = saved.get(&agent.name) {
agent.persistent_state = state.clone();
}
}
// Restore walked from surface agent if present
if let Some(surface) = self.agents.iter().find(|a| a.name == "subconscious-surface") {
if let Some(walked_str) = surface.persistent_state.get("walked") {
self.walked = walked_str.lines()
.map(|l| l.trim().to_string())
.filter(|l| !l.is_empty())
.collect();
}
}
dbglog!("[subconscious] loaded state from {}", path.display());
}
}
self.state_path = Some(path);
}
fn save_state(&self) {
let Some(path) = &self.state_path else { return };
let mut map: std::collections::BTreeMap<String, std::collections::BTreeMap<String, String>> =
std::collections::BTreeMap::new();
for agent in &self.agents {
if !agent.persistent_state.is_empty() {
map.insert(agent.name.clone(), agent.persistent_state.clone());
}
}
if let Ok(json) = serde_json::to_string_pretty(&map) {
let _ = std::fs::write(path, json);
}
}
pub fn snapshots(&self, store: Option<&crate::store::Store>) -> Vec<SubconsciousSnapshot> {
self.agents.iter().map(|s| {
let history = store.map(|st| {
let prov = format!("agent:{}", s.name);
st.recent_by_provenance(&prov, 30)
}).unwrap_or_default();
s.snapshot(&self.walked, history)
}).collect()
} }
/// Collect results from finished agents, inject outputs into the /// Collect results from finished agents, inject outputs into the
@ -397,6 +463,7 @@ impl Subconscious {
None None
} }
}).collect(); }).collect();
let had_finished = !finished.is_empty();
for (idx, handle) in finished { for (idx, handle) in finished {
let (auto_back, result) = handle.await.unwrap_or_else( let (auto_back, result) = handle.await.unwrap_or_else(
@ -408,6 +475,10 @@ impl Subconscious {
Ok(_) => { Ok(_) => {
let name = self.agents[idx].name.clone(); let name = self.agents[idx].name.clone();
let outputs = std::mem::take(&mut self.agents[idx].auto.outputs); let outputs = std::mem::take(&mut self.agents[idx].auto.outputs);
// Merge into persistent state
for (k, v) in &outputs {
self.agents[idx].persistent_state.insert(k.clone(), v.clone());
}
if let Some(walked_str) = outputs.get("walked") { if let Some(walked_str) = outputs.get("walked") {
self.walked = walked_str.lines() self.walked = walked_str.lines()
@ -470,6 +541,9 @@ impl Subconscious {
Err(e) => dbglog!("[subconscious] agent failed: {}", e), Err(e) => dbglog!("[subconscious] agent failed: {}", e),
} }
} }
if had_finished {
self.save_state();
}
} }
/// Trigger subconscious agents that are due to run. /// Trigger subconscious agents that are due to run.