kent/consciousness
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Agent/AgentState split complete — separate context and state locks

Browse source 
Agent is now Arc<Agent> (immutable config). ContextState and AgentState
have separate tokio::sync::Mutex locks. The parser locks only context,
tool dispatch locks only state. No contention between the two.

All callers migrated: mind/, user/, tools/, oneshot, dmn, learn.
28 tests pass, zero errors.

Co-Authored-By: Proof of Concept <poc@bcachefs.org>
This commit is contained in:
Kent Overstreet 2026-04-08 15:47:21 -04:00
parent 1d61b091b0
commit 0b9813431a
8 changed files with 156 additions and 159 deletions
Download patch file Download diff file Expand all files Collapse all files

86
src/mind/mod.rs
View file

@ -260,7 +260,7 @@ pub struct Mind {
}
impl Mind {
pub fn new(
pub async fn new(
config: SessionConfig,
turn_tx: mpsc::Sender<(Result<TurnResult>, StreamTarget)>,
) -> Self {
@ -271,7 +271,7 @@ impl Mind {
config.session_dir.join("conversation.jsonl"),
).ok();
let ag = Agent::new(
let agent = Agent::new(
client,
config.system_prompt.clone(),
config.context_parts.clone(),
@ -279,8 +279,7 @@ impl Mind {
config.prompt_file.clone(),
conversation_log,
shared_active_tools,
);
let agent = Arc::new(tokio::sync::Mutex::new(ag));
).await;
let shared = Arc::new(std::sync::Mutex::new(MindState::new(config.app.dmn.max_turns)));
let (turn_watch, _) = tokio::sync::watch::channel(false);
@ -314,15 +313,13 @@ impl Mind {
pub async fn init(&self) {
// Restore conversation
let mut ag = self.agent.lock().await;
ag.restore_from_log();
self.agent.restore_from_log().await;
// Restore persisted memory scores
let scores_path = self.config.session_dir.join("memory-scores.json");
load_memory_scores(&mut ag.context, &scores_path);
load_memory_scores(&mut *self.agent.context.lock().await, &scores_path);
ag.changed.notify_one();
drop(ag);
self.agent.state.lock().await.changed.notify_one();
// Load persistent subconscious state
let state_path = self.config.session_dir.join("subconscious-state.json");
@ -340,10 +337,9 @@ impl Mind {
MindCommand::None => {}
MindCommand::Compact => {
let threshold = compaction_threshold(&self.config.app) as usize;
let mut ag = self.agent.lock().await;
if ag.context.tokens() > threshold {
ag.compact();
ag.notify("compacted");
if self.agent.context.lock().await.tokens() > threshold {
self.agent.compact().await;
self.agent.state.lock().await.notify("compacted");
}
}
MindCommand::Score => {
@ -356,10 +352,10 @@ impl Mind {
}
MindCommand::Interrupt => {
self.shared.lock().unwrap().interrupt();
let ag = self.agent.lock().await;
let mut tools = ag.active_tools.lock().unwrap();
let active_tools = self.agent.state.lock().await.active_tools.clone();
let mut tools = active_tools.lock().unwrap();
for entry in tools.drain(..) { entry.handle.abort(); }
drop(tools); drop(ag);
drop(tools);
if let Some(h) = self.shared.lock().unwrap().turn_handle.take() { h.abort(); }
self.shared.lock().unwrap().turn_active = false;
let _ = self.turn_watch.send(false);
@ -373,14 +369,17 @@ impl Mind {
let new_log = log::ConversationLog::new(
self.config.session_dir.join("conversation.jsonl"),
).ok();
let mut ag = self.agent.lock().await;
let shared_tools = ag.active_tools.clone();
*ag = Agent::new(
ApiClient::new(&self.config.api_base, &self.config.api_key, &self.config.model),
self.config.system_prompt.clone(), self.config.context_parts.clone(),
self.config.app.clone(), self.config.prompt_file.clone(),
new_log, shared_tools,
);
{
let mut ctx = self.agent.context.lock().await;
ctx.clear(Section::Conversation);
}
{
let mut st = self.agent.state.lock().await;
st.conversation_log = new_log;
st.generation += 1;
st.last_prompt_tokens = 0;
}
self.agent.compact().await;
}
}
}
@ -395,10 +394,12 @@ impl Mind {
let response_window = cfg.scoring_response_window;
tokio::spawn(async move {
let (context, client) = {
let mut ag = agent.lock().await;
if ag.memory_scoring_in_flight { return; }
ag.memory_scoring_in_flight = true;
(ag.context.clone(), ag.client_clone())
let mut st = agent.state.lock().await;
if st.memory_scoring_in_flight { return; }
st.memory_scoring_in_flight = true;
drop(st);
let ctx = agent.context.lock().await.clone();
(ctx, agent.client.clone())
};
let _result = learn::score_memories_incremental(
&context, max_age as i64, response_window, &client, &agent,
@ -407,27 +408,27 @@ impl Mind {
let path = scores_path.clone();
async move {
let scores_snapshot = {
let mut ag = agent.lock().await;
for i in 0..ag.context.conversation().len() {
if let AstNode::Leaf(leaf) = &ag.context.conversation()[i] {
let mut ctx = agent.context.lock().await;
for i in 0..ctx.conversation().len() {
if let AstNode::Leaf(leaf) = &ctx.conversation()[i] {
if let NodeBody::Memory { key: k, .. } = leaf.body() {
if *k == key {
ag.context.set_score(Section::Conversation, i, Some(score));
ctx.set_score(Section::Conversation, i, Some(score));
}
}
}
}
ag.changed.notify_one();
collect_memory_scores(&ag.context)
let snapshot = collect_memory_scores(&ctx);
drop(ctx);
agent.state.lock().await.changed.notify_one();
snapshot
};
// Write to disk after releasing the lock
save_memory_scores(&scores_snapshot, &path);
}
},
).await;
{
let mut ag = agent.lock().await;
ag.memory_scoring_in_flight = false;
agent.state.lock().await.memory_scoring_in_flight = false;
}
let _ = bg_tx.send(BgEvent::ScoringDone);
});
@ -435,21 +436,20 @@ impl Mind {
async fn start_turn(&self, text: &str, target: StreamTarget) {
{
let mut ag = self.agent.lock().await;
match target {
StreamTarget::Conversation => {
ag.push_node(AstNode::user_msg(text));
self.agent.push_node(AstNode::user_msg(text)).await;
}
StreamTarget::Autonomous => {
ag.push_node(AstNode::dmn(text));
self.agent.push_node(AstNode::dmn(text)).await;
}
}
// Compact if over budget before sending
let threshold = compaction_threshold(&self.config.app) as usize;
if ag.context.tokens() > threshold {
ag.compact();
ag.notify("compacted");
if self.agent.context.lock().await.tokens() > threshold {
self.agent.compact().await;
self.agent.state.lock().await.notify("compacted");
}
}
self.shared.lock().unwrap().turn_active = true;