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

agent: move oneshot execution from subconscious to agent module

Browse source 
Move agent execution machinery (run_one_agent, spawn_agent, PID
tracking) from subconscious/knowledge.rs to agent/oneshot.rs — the
agent module owns execution, subconscious owns scheduling and defs.

Delete subconscious/llm.rs — callers now use api::call_api_with_tools_sync
directly. Audit and compare inline the call; oneshot inlines tool
filtering.

Update all callers: consolidate, daemon, subconscious, cli/agent.

Co-Authored-By: Proof of Concept <poc@bcachefs.org>
This commit is contained in:
ProofOfConcept 2026-04-04 17:25:10 -04:00 • committed by Kent Overstreet
parent 1457a1b50d
commit 0f4ca9e2f2
10 changed files with 43 additions and 106 deletions
Download patch file Download diff file Expand all files Collapse all files

4
src/subconscious/audit.rs
View file

@ -3,7 +3,6 @@
// Each batch of links gets reviewed by Sonnet, which returns per-link actions:
// KEEP, DELETE, RETARGET, WEAKEN, STRENGTHEN. Batches run in parallel via rayon.
use super::llm;
use crate::store::{self, Store, new_relation};
use std::collections::HashSet;
@ -211,7 +210,8 @@ pub fn link_audit(store: &mut Store, apply: bool) -> Result<AuditStats, String>
// Run batches in parallel via rayon
let batch_results: Vec<_> = batch_data.par_iter()
.map(|(batch_idx, batch_infos, prompt)| {
let response = llm::call_simple("audit", prompt);
let response = super::api::call_api_with_tools_sync(
"audit", &[prompt.clone()], &[], None, 10, &[], None, &|_| {});
let completed = done.fetch_add(1, Ordering::Relaxed) + 1;
eprint!("\r Batches: {}/{} done", completed, total_batches);
(*batch_idx, batch_infos, response)

4
src/subconscious/consolidate.rs
View file

@ -9,7 +9,7 @@
// 6. Summary: final metrics comparison
use super::digest;
use super::knowledge;
use crate::agent::oneshot;
use crate::neuro;
use crate::store::{self, Store};
@ -74,7 +74,7 @@ pub fn consolidate_full_with_progress(
*store = Store::load()?;
}
match knowledge::run_and_apply(store, agent_type, *count, "consolidate") {
match oneshot::run_and_apply(store, agent_type, *count, "consolidate") {
Ok(()) => {
let msg = " Done".to_string();
log_line(&mut log_buf, &msg);

6
src/subconscious/daemon.rs
View file

@ -123,7 +123,7 @@ fn job_targeted_agent(
ctx.log_line(msg);
log_event(&job_name, "progress", msg);
};
super::knowledge::run_one_agent_with_keys(
crate::agent::oneshot::run_one_agent_with_keys(
&mut store, &agent, std::slice::from_ref(&key), 5, "daemon", &log,
)?;
ctx.log_line("done");
@ -208,7 +208,7 @@ fn job_consolidation_agent(
};
// Use run_one_agent_with_keys — we already selected seeds above,
// no need to re-run the query.
let result = super::knowledge::run_one_agent_with_keys(
let result = crate::agent::oneshot::run_one_agent_with_keys(
&mut store, &agent, &claimed_keys, batch, "consolidate", &log,
).map(|_| ());
@ -239,7 +239,7 @@ fn job_rename_agent(
ctx.log_line(format!("running rename agent (batch={})", batch));
let log = |msg: &str| ctx.log_line(msg);
let result = super::knowledge::run_one_agent(&mut store, "rename", batch, "consolidate", &log)?;
let result = crate::agent::oneshot::run_one_agent(&mut store, "rename", batch, "consolidate", &log)?;
// Parse RENAME actions from response (rename uses its own format, not WRITE_NODE/LINK/REFINE)
let mut applied = 0;

394
src/subconscious/knowledge.rs
View file

@ -1,394 +0,0 @@
// knowledge.rs — agent execution and conversation fragment selection
//
// Agent prompts live in agents/*.agent files, dispatched via defs.rs.
// This module handles:
// - Agent execution (build prompt → call LLM with tools → log)
// - Conversation fragment selection (for observation agent)
//
// Agents apply changes via tool calls (poc-memory write/link-add/etc)
// during the LLM call — no action parsing needed.
use super::llm;
use crate::store::{self, Store};
use std::fs;
use std::path::PathBuf;
use std::sync::atomic::{AtomicPtr, Ordering};
// Global pid path for signal handler cleanup — stored as a leaked CString
// so the signal handler can unlink it without allocation.
static PID_CPATH: AtomicPtr<libc::c_char> = AtomicPtr::new(std::ptr::null_mut());
/// RAII guard that removes the pid file on drop (normal exit, panic).
struct PidGuard;
impl Drop for PidGuard {
fn drop(&mut self) {
let ptr = PID_CPATH.swap(std::ptr::null_mut(), Ordering::SeqCst);
if !ptr.is_null() {
unsafe { libc::unlink(ptr); }
// Reclaim the leaked CString
unsafe { drop(std::ffi::CString::from_raw(ptr)); }
}
}
}
/// Register signal handlers to clean up pid file on SIGTERM/SIGINT.
fn register_pid_cleanup(pid_path: &std::path::Path) {
let c_path = std::ffi::CString::new(pid_path.to_string_lossy().as_bytes())
.expect("pid path contains null");
// Leak the CString so the signal handler can access it
let old = PID_CPATH.swap(c_path.into_raw(), Ordering::SeqCst);
if !old.is_null() {
unsafe { drop(std::ffi::CString::from_raw(old)); }
}
unsafe {
libc::signal(libc::SIGTERM, pid_cleanup_handler as *const () as libc::sighandler_t);
libc::signal(libc::SIGINT, pid_cleanup_handler as *const () as libc::sighandler_t);
}
}
extern "C" fn pid_cleanup_handler(sig: libc::c_int) {
let ptr = PID_CPATH.swap(std::ptr::null_mut(), Ordering::SeqCst);
if !ptr.is_null() {
unsafe { libc::unlink(ptr); }
// Don't free — we're in a signal handler, just leak it
}
unsafe {
libc::signal(sig, libc::SIG_DFL);
libc::raise(sig);
}
}
// ---------------------------------------------------------------------------
// Agent execution
// ---------------------------------------------------------------------------
/// Result of running a single agent.
pub struct AgentResult {
pub output: String,
pub node_keys: Vec<String>,
/// Directory containing output() files from the agent run.
pub state_dir: std::path::PathBuf,
}
/// Run a single agent and return the result (no action application — tools handle that).
pub fn run_and_apply(
store: &mut Store,
agent_name: &str,
batch_size: usize,
llm_tag: &str,
) -> Result<(), String> {
run_and_apply_with_log(store, agent_name, batch_size, llm_tag, &|_| {})
}
pub fn run_and_apply_with_log(
store: &mut Store,
agent_name: &str,
batch_size: usize,
llm_tag: &str,
log: &(dyn Fn(&str) + Sync),
) -> Result<(), String> {
run_and_apply_excluded(store, agent_name, batch_size, llm_tag, log, &Default::default())
}
/// Like run_and_apply_with_log but with an in-flight exclusion set.
/// Returns the keys that were processed (for the daemon to track).
pub fn run_and_apply_excluded(
store: &mut Store,
agent_name: &str,
batch_size: usize,
llm_tag: &str,
log: &(dyn Fn(&str) + Sync),
exclude: &std::collections::HashSet<String>,
) -> Result<(), String> {
let _result = run_one_agent_excluded(store, agent_name, batch_size, llm_tag, log, exclude)?;
Ok(())
}
/// Run an agent with explicit target keys, bypassing the agent's query.
pub fn run_one_agent_with_keys(
store: &mut Store,
agent_name: &str,
keys: &[String],
count: usize,
llm_tag: &str,
log: &(dyn Fn(&str) + Sync),
) -> Result<AgentResult, String> {
let def = super::defs::get_def(agent_name)
.ok_or_else(|| format!("no .agent file for {}", agent_name))?;
let (state_dir, pid_path, _guard) = setup_agent_state(agent_name, &def)?;
log(&format!("targeting: {}", keys.join(", ")));
let graph = store.build_graph();
let mut resolved_steps = Vec::new();
let mut all_keys: Vec<String> = keys.to_vec();
for step in &def.steps {
let (prompt, extra_keys) = super::defs::resolve_placeholders(
&step.prompt, store, &graph, keys, count,
);
all_keys.extend(extra_keys);
resolved_steps.push(super::prompts::ResolvedStep {
prompt,
phase: step.phase.clone(),
});
}
let agent_batch = super::prompts::AgentBatch { steps: resolved_steps, node_keys: all_keys };
// Record visits eagerly so concurrent agents pick different seeds
if !agent_batch.node_keys.is_empty() {
store.record_agent_visits(&agent_batch.node_keys, agent_name).ok();
}
run_one_agent_inner(store, agent_name, &def, agent_batch, state_dir, pid_path, llm_tag, log)
}
pub fn run_one_agent(
store: &mut Store,
agent_name: &str,
batch_size: usize,
llm_tag: &str,
log: &(dyn Fn(&str) + Sync),
) -> Result<AgentResult, String> {
run_one_agent_excluded(store, agent_name, batch_size, llm_tag, log, &Default::default())
}
/// Like run_one_agent but excludes nodes currently being worked on by other agents.
pub fn run_one_agent_excluded(
store: &mut Store,
agent_name: &str,
batch_size: usize,
llm_tag: &str,
log: &(dyn Fn(&str) + Sync),
exclude: &std::collections::HashSet<String>,
) -> Result<AgentResult, String> {
let def = super::defs::get_def(agent_name)
.ok_or_else(|| format!("no .agent file for {}", agent_name))?;
// Set up output dir and write pid file BEFORE prompt building
let (state_dir, pid_path, _guard) = setup_agent_state(agent_name, &def)?;
log("building prompt");
let effective_count = def.count.unwrap_or(batch_size);
let agent_batch = super::defs::run_agent(store, &def, effective_count, exclude)?;
run_one_agent_inner(store, agent_name, &def, agent_batch, state_dir, pid_path, llm_tag, log)
}
/// Set up agent state dir, write initial pid file, register cleanup handlers.
/// Returns (state_dir, pid_path, guard). The guard removes the pid file on drop.
fn setup_agent_state(
agent_name: &str,
def: &super::defs::AgentDef,
) -> Result<(PathBuf, PathBuf, PidGuard), String> {
let state_dir = std::env::var("POC_AGENT_OUTPUT_DIR")
.map(PathBuf::from)
.unwrap_or_else(|_| store::memory_dir().join("agent-output").join(agent_name));
fs::create_dir_all(&state_dir)
.map_err(|e| format!("create state dir: {}", e))?;
unsafe { std::env::set_var("POC_AGENT_OUTPUT_DIR", &state_dir); }
// Clean up stale pid files from dead processes
scan_pid_files(&state_dir, 0);
let pid = std::process::id();
let pid_path = state_dir.join(format!("pid-{}", pid));
let first_phase = def.steps.first()
.map(|s| s.phase.as_str())
.unwrap_or("step-0");
fs::write(&pid_path, first_phase).ok();
// Register for cleanup on signals and normal exit
register_pid_cleanup(&pid_path);
Ok((state_dir, pid_path, PidGuard))
}
/// Check for live agent processes in a state dir. Returns (phase, pid) pairs.
/// Cleans up stale pid files and kills timed-out processes.
pub fn scan_pid_files(state_dir: &std::path::Path, timeout_secs: u64) -> Vec<(String, u32)> {
let mut live = Vec::new();
let Ok(entries) = fs::read_dir(state_dir) else { return live };
for entry in entries.flatten() {
let name = entry.file_name();
let name_str = name.to_string_lossy();
if !name_str.starts_with("pid-") { continue; }
let pid: u32 = name_str.strip_prefix("pid-")
.and_then(|s| s.parse().ok())
.unwrap_or(0);
if pid == 0 { continue; }
if unsafe { libc::kill(pid as i32, 0) } != 0 {
fs::remove_file(entry.path()).ok();
continue;
}
if timeout_secs > 0 {
if let Ok(meta) = entry.metadata() {
if let Ok(modified) = meta.modified() {
if modified.elapsed().unwrap_or_default().as_secs() > timeout_secs {
unsafe { libc::kill(pid as i32, libc::SIGTERM); }
fs::remove_file(entry.path()).ok();
continue;
}
}
}
}
let phase = fs::read_to_string(entry.path())
.unwrap_or_default()
.trim().to_string();
live.push((phase, pid));
}
live
}
/// Spawn an agent asynchronously. Writes the pid file before returning
/// so the caller immediately sees the agent as running.
/// Spawn result: child process handle and log path.
pub struct SpawnResult {
pub child: std::process::Child,
pub log_path: PathBuf,
}
pub fn spawn_agent(
agent_name: &str,
state_dir: &std::path::Path,
session_id: &str,
) -> Option<SpawnResult> {
let def = super::defs::get_def(agent_name)?;
let first_phase = def.steps.first()
.map(|s| s.phase.as_str())
.unwrap_or("step-0");
let log_dir = dirs::home_dir().unwrap_or_default()
.join(format!(".consciousness/logs/{}", agent_name));
fs::create_dir_all(&log_dir).ok();
let log_path = log_dir.join(format!("{}.log", store::compact_timestamp()));
let agent_log = fs::File::create(&log_path)
.unwrap_or_else(|_| fs::File::create("/dev/null").unwrap());
let child = std::process::Command::new("poc-memory")
.args(["agent", "run", agent_name, "--count", "1", "--local",
"--state-dir", &state_dir.to_string_lossy()])
.env("POC_SESSION_ID", session_id)
.stdout(agent_log.try_clone().unwrap_or_else(|_| fs::File::create("/dev/null").unwrap()))
.stderr(agent_log)
.spawn()
.ok()?;
let pid = child.id();
let pid_path = state_dir.join(format!("pid-{}", pid));
fs::write(&pid_path, first_phase).ok();
Some(SpawnResult { child, log_path })
}
fn run_one_agent_inner(
_store: &mut Store,
agent_name: &str,
def: &super::defs::AgentDef,
agent_batch: super::prompts::AgentBatch,
state_dir: std::path::PathBuf,
pid_path: std::path::PathBuf,
_llm_tag: &str,
log: &(dyn Fn(&str) + Sync),
) -> Result<AgentResult, String> {
let all_tools = crate::agent::tools::memory_and_journal_tools();
let effective_tools: Vec<crate::agent::tools::Tool> = if def.tools.is_empty() {
all_tools
} else {
all_tools.into_iter()
.filter(|t| def.tools.iter().any(|w| w == &t.name))
.collect()
};
let tools_desc = effective_tools.iter().map(|t| t.name).collect::<Vec<_>>().join(", ");
let n_steps = agent_batch.steps.len();
for key in &agent_batch.node_keys {
log(&format!(" node: {}", key));
}
// Guard: reject oversized first prompt (later steps grow via conversation)
let max_prompt_bytes = 800_000;
let first_len = agent_batch.steps[0].prompt.len();
if first_len > max_prompt_bytes {
let prompt_kb = first_len / 1024;
let oversize_dir = store::memory_dir().join("llm-logs").join("oversized");
fs::create_dir_all(&oversize_dir).ok();
let oversize_path = oversize_dir.join(format!("{}-{}.txt",
agent_name, store::compact_timestamp()));
let header = format!("=== OVERSIZED PROMPT ===\nagent: {}\nsize: {}KB (max {}KB)\nnodes: {:?}\n\n",
agent_name, prompt_kb, max_prompt_bytes / 1024, agent_batch.node_keys);
fs::write(&oversize_path, format!("{}{}", header, &agent_batch.steps[0].prompt)).ok();
log(&format!("oversized prompt logged to {}", oversize_path.display()));
return Err(format!(
"prompt too large: {}KB (max {}KB) — seed nodes may be oversized",
prompt_kb, max_prompt_bytes / 1024,
));
}
let write_pid = |phase: &str| {
fs::write(&pid_path, phase).ok();
};
let phases: Vec<&str> = agent_batch.steps.iter().map(|s| s.phase.as_str()).collect();
log(&format!("{} step(s) {:?}, {}KB initial, {}, {} nodes, output={}",
n_steps, phases, first_len / 1024, tools_desc,
agent_batch.node_keys.len(), state_dir.display()));
let prompts: Vec<String> = agent_batch.steps.iter()
.map(|s| s.prompt.clone()).collect();
let step_phases: Vec<String> = agent_batch.steps.iter()
.map(|s| s.phase.clone()).collect();
let step_phases_for_bail = step_phases.clone();
if std::env::var("POC_AGENT_VERBOSE").is_ok() {
for (i, s) in agent_batch.steps.iter().enumerate() {
log(&format!("=== PROMPT {}/{} ({}) ===\n\n{}", i + 1, n_steps, s.phase, s.prompt));
}
}
log("\n=== CALLING LLM ===");
// Bail check: if the agent defines a bail script, run it between steps.
// The script receives the pid file path as $1, cwd = state dir.
let bail_script = def.bail.as_ref().map(|name| {
// Look for the script next to the .agent file
let agents_dir = super::defs::agents_dir();
agents_dir.join(name)
});
let state_dir_for_bail = state_dir.clone();
let pid_path_for_bail = pid_path.clone();
let bail_fn = move |step_idx: usize| -> Result<(), String> {
// Update phase in pid file and provenance tracking
if step_idx < step_phases_for_bail.len() {
write_pid(&step_phases_for_bail[step_idx]);
}
// Run bail script if defined
if let Some(ref script) = bail_script {
let status = std::process::Command::new(script)
.arg(&pid_path_for_bail)
.current_dir(&state_dir_for_bail)
.status()
.map_err(|e| format!("bail script {:?} failed: {}", script, e))?;
if !status.success() {
return Err(format!("bailed at step {}: {:?} exited {}",
step_idx + 1, script.file_name().unwrap_or_default(),
status.code().unwrap_or(-1)));
}
}
Ok(())
};
let output = llm::call_for_def_multi(def, &prompts, &step_phases, Some(&bail_fn), log)?;
Ok(AgentResult {
output,
node_keys: agent_batch.node_keys,
state_dir,
})
}
// ---------------------------------------------------------------------------
// Conversation fragment selection removed — observe/journal agents handle transcripts.

58
src/subconscious/llm.rs
View file

@ -1,58 +0,0 @@
// LLM utilities: model invocation via direct API
use crate::store::Store;
use std::fs;
/// Simple LLM call for non-agent uses (audit, digest, compare).
/// Logs to llm-logs/{caller}/ file.
pub(crate) fn call_simple(caller: &str, prompt: &str) -> Result<String, String> {
let log_dir = dirs::home_dir().unwrap_or_default()
.join(".consciousness/logs/llm").join(caller);
fs::create_dir_all(&log_dir).ok();
let log_path = log_dir.join(format!("{}.txt", crate::store::compact_timestamp()));
use std::io::Write;
let log = move |msg: &str| {
if let Ok(mut f) = fs::OpenOptions::new()
.create(true).append(true).open(&log_path)
{
let _ = writeln!(f, "{}", msg);
}
};
let prompts = vec![prompt.to_string()];
let phases = vec![];
super::api::call_api_with_tools_sync(caller, &prompts, &phases, None, 10, &[], None, &log)
}
/// Call a model using an agent definition's configuration (multi-step).
/// Optional bail_fn is called between steps — return Err to stop the pipeline.
pub(crate) fn call_for_def_multi(
def: &super::defs::AgentDef,
prompts: &[String],
phases: &[String],
bail_fn: Option<&(dyn Fn(usize) -> Result<(), String> + Sync)>,
log: &(dyn Fn(&str) + Sync),
) -> Result<String, String> {
// Filter tools based on AgentDef specification
let all_tools = crate::agent::tools::memory_and_journal_tools();
let effective_tools: Vec<crate::agent::tools::Tool> = if def.tools.is_empty() {
all_tools
} else {
all_tools.into_iter()
.filter(|t| def.tools.iter().any(|w| w == &t.name))
.collect()
};
super::api::call_api_with_tools_sync(&def.agent, prompts, phases, def.temperature, def.priority, &effective_tools, bail_fn, log)
}
/// Get all keys for prompt context.
pub(crate) fn semantic_keys(store: &Store) -> Vec<String> {
let mut keys: Vec<String> = store.nodes.keys()
.cloned()
.collect();
keys.sort();
keys.truncate(200);
keys
}

2
src/subconscious/mod.rs
View file

@ -19,11 +19,9 @@
pub mod subconscious;
pub mod api;
pub mod llm;
pub mod prompts;
pub mod defs;
pub mod audit;
pub mod consolidate;
pub mod knowledge;
pub mod digest;
pub mod daemon;

8
src/subconscious/subconscious.rs
View file

@ -156,7 +156,7 @@ impl AgentCycleState {
}
fn agent_spawned(&mut self, name: &str, phase: &str,
result: crate::agents::knowledge::SpawnResult) {
result: crate::agent::oneshot::SpawnResult) {
if let Some(agent) = self.agents.iter_mut().find(|a| a.name == name) {
agent.pid = Some(result.child.id());
agent.phase = Some(phase.to_string());
@ -279,7 +279,7 @@ impl AgentCycleState {
if transcript.size > 0 {
fs::write(&offset_path, transcript.size.to_string()).ok();
}
if let Some(result) = crate::agents::knowledge::spawn_agent(
if let Some(result) = crate::agent::oneshot::spawn_agent(
"surface-observe", &state_dir, &session.session_id) {
self.log(format_args!("spawned surface-observe pid {}\n", result.child.id()));
self.agent_spawned("surface-observe", "surface", result);
@ -346,7 +346,7 @@ impl AgentCycleState {
}
fs::write(&offset_path, transcript.size.to_string()).ok();
if let Some(result) = crate::agents::knowledge::spawn_agent(
if let Some(result) = crate::agent::oneshot::spawn_agent(
"reflect", &state_dir, &session.session_id) {
self.log(format_args!("reflect: spawned pid {}\n", result.child.id()));
self.agent_spawned("reflect", "step-0", result);
@ -375,7 +375,7 @@ impl AgentCycleState {
}
fs::write(&offset_path, transcript.size.to_string()).ok();
if let Some(result) = crate::agents::knowledge::spawn_agent(
if let Some(result) = crate::agent::oneshot::spawn_agent(
"journal", &state_dir, &session.session_id) {
self.log(format_args!("journal: spawned pid {}\n", result.child.id()));
self.agent_spawned("journal", "step-0", result);