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

WIP: Agent/AgentState split — struct defined, 80+ errors remaining

Browse source 
Split Agent into immutable Agent (behind Arc) and mutable AgentState
(behind its own Mutex). ContextState has its own Mutex on Agent.
Activities moved to AgentState. new() and fork() rewritten.

All callers need mechanical updates: agent.lock().await.field →
agent.state.lock().await.field or agent.context.lock().await.method.

Co-Authored-By: Proof of Concept <poc@bcachefs.org>
This commit is contained in:
Kent Overstreet 2026-04-08 15:36:08 -04:00
parent e587431f9a
commit 7fe4584ba0
1 changed files with 82 additions and 98 deletions
Download patch file Download diff file Expand all files Collapse all files

180
src/agent/mod.rs
View file

@ -38,9 +38,8 @@ pub struct ActivityEntry {
pub expires_at: std::time::Instant,
}
/// RAII guard — marks the activity "(complete)" on drop, starts expiry timer.
pub struct ActivityGuard {
agent: Arc<tokio::sync::Mutex<Agent>>,
agent: Arc<Agent>,
id: u64,
}
@ -48,8 +47,8 @@ const ACTIVITY_LINGER: std::time::Duration = std::time::Duration::from_secs(5);
impl Drop for ActivityGuard {
fn drop(&mut self) {
if let Ok(mut ag) = self.agent.try_lock() {
if let Some(entry) = ag.activities.iter_mut().find(|a| a.id == self.id) {
if let Ok(mut st) = self.agent.state.try_lock() {
if let Some(entry) = st.activities.iter_mut().find(|a| a.id == self.id) {
entry.label.push_str(" (complete)");
entry.expires_at = std::time::Instant::now() + ACTIVITY_LINGER;
}
@ -57,8 +56,7 @@ impl Drop for ActivityGuard {
}
}
impl Agent {
/// Register an activity, returns its ID. Caller creates the guard.
impl AgentState {
pub fn push_activity(&mut self, label: impl Into<String>) -> u64 {
self.expire_activities();
let id = self.next_activity_id;
@ -72,7 +70,6 @@ impl Agent {
id
}
/// Push a notification — auto-expires after 5 seconds.
pub fn notify(&mut self, label: impl Into<String>) {
self.expire_activities();
let id = self.next_activity_id;
@ -85,21 +82,14 @@ impl Agent {
self.changed.notify_one();
}
/// Remove expired activities.
pub fn expire_activities(&mut self) {
let now = std::time::Instant::now();
self.activities.retain(|a| a.expires_at > now);
}
}
/// Create an activity guard from outside the lock.
pub fn activity_guard(agent: &Arc<tokio::sync::Mutex<Agent>>, id: u64) -> ActivityGuard {
ActivityGuard { agent: agent.clone(), id }
}
/// Convenience: lock, push activity, unlock, return guard.
pub async fn start_activity(agent: &Arc<tokio::sync::Mutex<Agent>>, label: impl Into<String>) -> ActivityGuard {
let id = agent.lock().await.push_activity(label);
pub async fn start_activity(agent: &Arc<Agent>, label: impl Into<String>) -> ActivityGuard {
let id = agent.state.lock().await.push_activity(label);
ActivityGuard { agent: agent.clone(), id }
}
@ -138,46 +128,39 @@ impl DispatchState {
}
}
/// Immutable agent config — shared via Arc, no mutex needed.
pub struct Agent {
client: ApiClient,
tools: Vec<tools::Tool>,
/// Last known prompt token count from the API (tracks context size).
last_prompt_tokens: u32,
/// Current reasoning effort level ("none", "low", "high").
pub client: ApiClient,
pub app_config: crate::config::AppConfig,
pub prompt_file: String,
pub session_id: String,
pub context: tokio::sync::Mutex<ContextState>,
pub state: tokio::sync::Mutex<AgentState>,
}
/// Mutable agent state — behind its own mutex.
pub struct AgentState {
pub tools: Vec<tools::Tool>,
pub last_prompt_tokens: u32,
pub reasoning_effort: String,
/// Sampling parameters — adjustable at runtime from the thalamus screen.
pub temperature: f32,
pub top_p: f32,
pub top_k: u32,
/// Active activities — RAII guards auto-remove on drop.
pub activities: Vec<ActivityEntry>,
next_activity_id: u64,
/// Control tool flags — set by tool handlers, consumed by turn loop.
pub pending_yield: bool,
pub pending_model_switch: Option<String>,
pub pending_dmn_pause: bool,
/// Provenance tag for memory operations — identifies who made the change.
pub provenance: String,
/// Persistent conversation log — append-only record of all messages.
pub conversation_log: Option<ConversationLog>,
pub context: ContextState,
/// App config — used to reload identity on compaction and model switching.
pub app_config: crate::config::AppConfig,
pub prompt_file: String,
/// Stable session ID for memory-search dedup across turns.
pub session_id: String,
/// Incremented on compaction — UI uses this to detect resets.
pub generation: u64,
/// Whether incremental memory scoring is currently running.
pub memory_scoring_in_flight: bool,
/// Shared active tools — Agent writes, TUI reads.
pub active_tools: tools::SharedActiveTools,
/// Fires when agent state changes — UI wakes on this instead of polling.
pub changed: Arc<tokio::sync::Notify>,
}
impl Agent {
pub fn new(
pub async fn new(
client: ApiClient,
system_prompt: String,
personality: Vec<(String, String)>,
@ -185,7 +168,7 @@ impl Agent {
prompt_file: String,
conversation_log: Option<ConversationLog>,
active_tools: tools::SharedActiveTools,
) -> Self {
) -> Arc<Self> {
let mut context = ContextState::new();
context.push(Section::System, AstNode::system_msg(&system_prompt));
@ -207,97 +190,98 @@ impl Agent {
for (name, content) in &personality {
context.push(Section::Identity, AstNode::memory(name, content));
}
let session_id = format!("consciousness-{}", chrono::Utc::now().format("%Y%m%d-%H%M%S"));
let mut agent = Self {
let agent = Arc::new(Self {
client,
tools: tools::tools(),
last_prompt_tokens: 0,
reasoning_effort: "none".to_string(),
temperature: 0.6,
top_p: 0.95,
top_k: 20,
activities: Vec::new(),
next_activity_id: 0,
pending_yield: false,
pending_model_switch: None,
pending_dmn_pause: false,
provenance: "manual".to_string(),
conversation_log,
context,
app_config,
prompt_file,
session_id,
generation: 0,
memory_scoring_in_flight: false,
active_tools,
changed: Arc::new(tokio::sync::Notify::new()),
};
context: tokio::sync::Mutex::new(context),
state: tokio::sync::Mutex::new(AgentState {
tools: tools::tools(),
last_prompt_tokens: 0,
reasoning_effort: "none".to_string(),
temperature: 0.6,
top_p: 0.95,
top_k: 20,
activities: Vec::new(),
next_activity_id: 0,
pending_yield: false,
pending_model_switch: None,
pending_dmn_pause: false,
provenance: "manual".to_string(),
conversation_log,
generation: 0,
memory_scoring_in_flight: false,
active_tools,
changed: Arc::new(tokio::sync::Notify::new()),
}),
});
agent.load_startup_journal();
agent.load_startup_journal().await;
agent
}
/// Create a lightweight agent forked from this one's context.
///
/// The forked agent shares the same conversation prefix (system prompt,
/// personality, journal, entries) for KV cache sharing. The caller
/// appends the subconscious prompt as a user message and runs the turn.
pub fn fork(&self, tools: Vec<tools::Tool>) -> Self {
Self {
/// Fork: clones context for KV cache prefix sharing.
pub async fn fork(self: &Arc<Self>, tools: Vec<tools::Tool>) -> Arc<Self> {
let ctx = self.context.lock().await.clone();
let st = self.state.lock().await;
Arc::new(Self {
client: self.client.clone(),
tools,
last_prompt_tokens: 0,
reasoning_effort: "none".to_string(),
temperature: self.temperature,
top_p: self.top_p,
top_k: self.top_k,
activities: Vec::new(),
next_activity_id: 0,
pending_yield: false,
pending_model_switch: None,
pending_dmn_pause: false,
provenance: self.provenance.clone(),
conversation_log: None,
context: self.context.clone(),
app_config: self.app_config.clone(),
prompt_file: self.prompt_file.clone(),
session_id: self.session_id.clone(),
generation: 0,
memory_scoring_in_flight: false,
active_tools: tools::shared_active_tools(),
changed: Arc::new(tokio::sync::Notify::new()),
}
context: tokio::sync::Mutex::new(ctx),
state: tokio::sync::Mutex::new(AgentState {
tools,
last_prompt_tokens: 0,
reasoning_effort: "none".to_string(),
temperature: st.temperature,
top_p: st.top_p,
top_k: st.top_k,
activities: Vec::new(),
next_activity_id: 0,
pending_yield: false,
pending_model_switch: None,
pending_dmn_pause: false,
provenance: st.provenance.clone(),
conversation_log: None,
generation: 0,
memory_scoring_in_flight: false,
active_tools: tools::shared_active_tools(),
changed: Arc::new(tokio::sync::Notify::new()),
}),
})
}
/// Assemble the full prompt as token IDs.
/// Context sections + assistant prompt suffix.
pub fn assemble_prompt_tokens(&self) -> Vec<u32> {
let mut tokens = self.context.token_ids();
pub async fn assemble_prompt_tokens(&self) -> Vec<u32> {
let ctx = self.context.lock().await;
let mut tokens = ctx.token_ids();
tokens.push(tokenizer::IM_START);
tokens.extend(tokenizer::encode("assistant\n"));
tokens
}
/// Push a node into the conversation and log it.
pub fn push_node(&mut self, node: AstNode) {
if let Some(ref log) = self.conversation_log {
pub async fn push_node(&self, node: AstNode) {
let st = self.state.lock().await;
if let Some(ref log) = st.conversation_log {
if let Err(e) = log.append_node(&node) {
eprintln!("warning: failed to log entry: {:#}", e);
}
}
self.context.push(Section::Conversation, node);
self.changed.notify_one();
drop(st);
self.context.lock().await.push(Section::Conversation, node);
self.state.lock().await.changed.notify_one();
}
/// Run the agent turn loop: assemble prompt, stream response,
/// parse into AST, dispatch tool calls, repeat until text response.
pub async fn turn(
agent: Arc<tokio::sync::Mutex<Agent>>,
agent: Arc<Agent>,
) -> Result<TurnResult> {
let active_tools = {
let me = agent.lock().await;
me.active_tools.clone()
agent.state.lock().await.active_tools.clone()
};
// Collect finished background tools