gitops/optimclaw
1
0
Fork 0
mirror of https://github.com/outbackdingo/optimclaw.git synced 2026-09-01 00:59:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/staging' into v2-architecture

Browse Source
This commit is contained in:
[email protected]
2026-03-23 22:25:43 -07:00
parent 3b309ccbc8 5847479fd8
commit ac7c17ee40
120 changed files with 9766 additions and 2011 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agent/agent_loop.rs
+7 -4
View File
@@ -157,8 +157,8 @@ pub struct AgentDeps {
pub hooks: Arc<HookRegistry>,
/// Cost enforcement guardrails (daily budget, hourly rate limits).
pub cost_guard: Arc<crate::agent::cost_guard::CostGuard>,
/// SSE broadcast sender for live job event streaming to the web gateway.
pub sse_tx: Option<tokio::sync::broadcast::Sender<crate::channels::web::types::SseEvent>>,
/// SSE manager for live job event streaming to the web gateway.
pub sse_tx: Option<Arc<crate::channels::web::sse::SseManager>>,
/// HTTP interceptor for trace recording/replay.
pub http_interceptor: Option<Arc<dyn crate::llm::recording::HttpInterceptor>>,
/// Audio transcription middleware for voice messages.
@@ -169,6 +169,9 @@ pub struct AgentDeps {
pub sandbox_readiness: crate::agent::routine_engine::SandboxReadiness,
/// Software builder for self-repair tool rebuilding.
pub builder: Option<Arc<dyn crate::tools::SoftwareBuilder>>,
/// Resolved LLM backend identifier (e.g., "nearai", "openai", "groq").
/// Used by `/model` persistence to determine which env var to update.
pub llm_backend: String,
}
/// The main agent that coordinates all components.
@@ -235,8 +238,8 @@ impl Agent {
hooks: deps.hooks.clone(),
},
);
if let Some(ref tx) = deps.sse_tx {
scheduler.set_sse_sender(tx.clone());
if let Some(ref sse) = deps.sse_tx {
scheduler.set_sse_sender(Arc::clone(sse));
}
if let Some(ref interceptor) = deps.http_interceptor {
scheduler.set_http_interceptor(Arc::clone(interceptor));
src/agent/commands.rs
+49 -3
View File
@@ -841,12 +841,50 @@ impl Agent {
.await
{
tracing::warn!("Failed to persist model to DB: {}", e);
} else {
tracing::debug!("Persisted selected_model to DB: {}", model);
}
} else {
tracing::warn!("No database store available — model choice will not persist to DB");
}
// 2. Update TOML config file if it exists (sync I/O in spawn_blocking).
// 2. Update .env and TOML config file (sync I/O in spawn_blocking).
let model_owned = model.to_string();
let backend = self.deps.llm_backend.clone();
if let Err(e) = tokio::task::spawn_blocking(move || {
// 2a. Update the backend-specific model env var in ~/.ironclaw/.env.
//
// Env vars have the HIGHEST priority in LlmConfig::resolve_model()
// (env var > TOML > DB > default). If the .env file has e.g.
// NEARAI_MODEL=old-model, it shadows everything else. We must
// update this var or the /model change is invisible on restart.
let registry = crate::llm::ProviderRegistry::load();
let model_env = registry.model_env_var(&backend);
let env_var_prefix = format!("{}=", model_env);
// Only update the .env file if the var is actually set there
// (avoid injecting new vars the user never configured).
let env_path = crate::bootstrap::ironclaw_env_path();
let env_has_var = std::fs::read_to_string(&env_path)
.ok()
.is_some_and(|content| {
content.lines().any(|line| {
let trimmed = line.trim_start();
!trimmed.starts_with('#') && trimmed.starts_with(&env_var_prefix)
})
});
if env_has_var {
if let Err(e) = crate::bootstrap::upsert_bootstrap_var(model_env, &model_owned) {
tracing::warn!("Failed to update {} in .env: {}", model_env, e);
} else {
tracing::debug!("Updated {} in .env to {}", model_env, model_owned);
}
}
// 2b. Update (or create) the TOML config file.
//
// The TOML overlay has higher priority than DB settings on
// startup, so it MUST stay in sync with the DB.
let toml_path = crate::settings::Settings::default_toml_path();
match crate::settings::Settings::load_toml(&toml_path) {
Ok(Some(mut settings)) => {
@@ -856,7 +894,15 @@ impl Agent {
}
}
Ok(None) => {
// No config file on disk; nothing to update.
// No config file yet — create one so the model choice
// survives restarts even when the DB is unavailable.
let settings = crate::settings::Settings {
selected_model: Some(model_owned),
..Default::default()
};
if let Err(e) = settings.save_toml(&toml_path) {
tracing::warn!("Failed to create config.toml for model persistence: {}", e);
}
}
Err(e) => {
tracing::warn!("Failed to load config.toml for model persistence: {}", e);
@@ -865,7 +911,7 @@ impl Agent {
})
.await
{
tracing::warn!("Model TOML persistence task failed: {}", e);
tracing::warn!("Model persistence task failed: {}", e);
}
}
}
src/agent/dispatcher.rs
+35 -7
View File
@@ -915,7 +915,14 @@ pub(super) async fn execute_chat_tool_standalone(
params: &serde_json::Value,
job_ctx: &crate::context::JobContext,
) -> Result<String, Error> {
crate::tools::execute::execute_tool_with_safety(tools, safety, tool_name, params, job_ctx).await
crate::tools::execute::execute_tool_with_safety(
tools,
safety,
tool_name,
params.clone(),
job_ctx,
)
.await
}
/// Parsed auth result fields for emitting StatusUpdate::AuthRequired.
@@ -1091,15 +1098,23 @@ pub(crate) fn extract_suggestions(text: &str) -> (String, Vec<String>) {
Regex::new(r"(?s)<suggestions>\s*(.*?)\s*</suggestions>").expect("valid regex") // safety: constant pattern
});
// Find the position of the last closing code fence to avoid matching inside code blocks
let last_code_fence = text.rfind("```").unwrap_or(0);
// Build a sorted list of code fence positions to determine open/close pairing.
// A position is "inside" a fenced block when it falls between an odd-numbered
// fence (opening) and the next even-numbered fence (closing).
let fence_positions: Vec<usize> = text.match_indices("```").map(|(pos, _)| pos).collect();
// Find all matches, take the last one that's after the last code fence
let is_inside_fence = |pos: usize| -> bool {
// Count how many fences appear before `pos`. If odd, we're inside a fence.
let count = fence_positions.iter().take_while(|&&fp| fp <= pos).count();
count % 2 == 1
};
// Find all matches, take the last one that's outside any code fence
let mut best_match: Option<regex::Match<'_>> = None;
let mut best_capture: Option<String> = None;
for caps in RE.captures_iter(text) {
if let (Some(full), Some(inner)) = (caps.get(0), caps.get(1))
&& full.start() >= last_code_fence
&& !is_inside_fence(full.start())
{
best_match = Some(full);
best_capture = Some(inner.as_str().to_string());
@@ -1218,6 +1233,7 @@ mod tests {
document_extraction: None,
sandbox_readiness: crate::agent::routine_engine::SandboxReadiness::DisabledByConfig,
builder: None,
llm_backend: "nearai".to_string(),
};
Agent::new(
@@ -1893,7 +1909,7 @@ mod tests {
Ok(ToolCompletionResponse {
content: None,
tool_calls: vec![ToolCall {
id: format!("call_{}", uuid::Uuid::new_v4()),
id: crate::llm::generate_tool_call_id(0, 0),
name: "echo".to_string(),
arguments: serde_json::json!({"message": "looping"}),
}],
@@ -2046,7 +2062,7 @@ mod tests {
Ok(ToolCompletionResponse {
content: None,
tool_calls: vec![ToolCall {
id: format!("call_{}", uuid::Uuid::new_v4()),
id: crate::llm::generate_tool_call_id(0, 0),
name: "nonexistent_tool".to_string(),
arguments: serde_json::json!({}),
}],
@@ -2085,6 +2101,7 @@ mod tests {
document_extraction: None,
sandbox_readiness: crate::agent::routine_engine::SandboxReadiness::DisabledByConfig,
builder: None,
llm_backend: "nearai".to_string(),
};
Agent::new(
@@ -2205,6 +2222,7 @@ mod tests {
document_extraction: None,
sandbox_readiness: crate::agent::routine_engine::SandboxReadiness::DisabledByConfig,
builder: None,
llm_backend: "nearai".to_string(),
};
Agent::new(
@@ -2338,6 +2356,16 @@ mod tests {
assert!(suggestions.is_empty()); // safety: test
}
#[test]
fn test_extract_suggestions_inside_unclosed_code_fence() {
// Regression: odd number of fences (unclosed fence) must still be
// treated as "inside a code block".
let input = "```\ncode\n<suggestions>[\"bar\"]</suggestions>";
let (text, suggestions) = super::extract_suggestions(input);
assert_eq!(text, input); // safety: test
assert!(suggestions.is_empty()); // safety: test
}
#[test]
fn test_extract_suggestions_after_code_fence() {
let input = "```\ncode\n```\nAnswer.\n<suggestions>[\"foo\"]</suggestions>";
src/agent/job_monitor.rs
+22 -12
View File
@@ -44,7 +44,7 @@ pub struct JobMonitorRoute {
/// the main agent's context window).
pub fn spawn_job_monitor(
job_id: Uuid,
event_rx: broadcast::Receiver<(Uuid, SseEvent)>,
event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>,
inject_tx: mpsc::Sender<IncomingMessage>,
route: JobMonitorRoute,
) -> JoinHandle<()> {
@@ -56,7 +56,7 @@ pub fn spawn_job_monitor(
/// jobs don't stay `InProgress` forever in the `ContextManager`.
pub fn spawn_job_monitor_with_context(
job_id: Uuid,
mut event_rx: broadcast::Receiver<(Uuid, SseEvent)>,
mut event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>,
inject_tx: mpsc::Sender<IncomingMessage>,
route: JobMonitorRoute,
context_manager: Option<Arc<ContextManager>>,
@@ -68,7 +68,7 @@ pub fn spawn_job_monitor_with_context(
loop {
match event_rx.recv().await {
Ok((ev_job_id, event)) => {
Ok((ev_job_id, _user_id, event)) => {
if ev_job_id != job_id {
continue;
}
@@ -162,7 +162,7 @@ pub fn spawn_job_monitor_with_context(
/// inject messages into) but we still need to free the `max_jobs` slot.
pub fn spawn_completion_watcher(
job_id: Uuid,
mut event_rx: broadcast::Receiver<(Uuid, SseEvent)>,
mut event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>,
context_manager: Arc<ContextManager>,
) -> JoinHandle<()> {
let short_id = job_id.to_string()[..8].to_string();
@@ -170,7 +170,9 @@ pub fn spawn_completion_watcher(
tokio::spawn(async move {
loop {
match event_rx.recv().await {
Ok((ev_job_id, SseEvent::JobResult { status, .. })) if ev_job_id == job_id => {
Ok((ev_job_id, _user_id, SseEvent::JobResult { status, .. }))
if ev_job_id == job_id =>
{
let target = if status == "completed" {
JobState::Completed
} else {
@@ -227,7 +229,7 @@ mod tests {
#[tokio::test]
async fn test_monitor_forwards_assistant_messages() {
let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
let job_id = Uuid::new_v4();
@@ -237,6 +239,7 @@ mod tests {
event_tx
.send((
job_id,
"test-user".to_string(),
SseEvent::JobMessage {
job_id: job_id.to_string(),
role: "assistant".to_string(),
@@ -259,7 +262,7 @@ mod tests {
#[tokio::test]
async fn test_monitor_ignores_other_jobs() {
let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
let job_id = Uuid::new_v4();
@@ -270,6 +273,7 @@ mod tests {
event_tx
.send((
other_job_id,
"test-user".to_string(),
SseEvent::JobMessage {
job_id: other_job_id.to_string(),
role: "assistant".to_string(),
@@ -289,7 +293,7 @@ mod tests {
#[tokio::test]
async fn test_monitor_exits_on_job_result() {
let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
let job_id = Uuid::new_v4();
@@ -299,6 +303,7 @@ mod tests {
event_tx
.send((
job_id,
"test-user".to_string(),
SseEvent::JobResult {
job_id: job_id.to_string(),
status: "completed".to_string(),
@@ -324,7 +329,7 @@ mod tests {
#[tokio::test]
async fn test_monitor_skips_tool_events() {
let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
let job_id = Uuid::new_v4();
@@ -334,6 +339,7 @@ mod tests {
event_tx
.send((
job_id,
"test-user".to_string(),
SseEvent::JobToolUse {
job_id: job_id.to_string(),
tool_name: "shell".to_string(),
@@ -346,6 +352,7 @@ mod tests {
event_tx
.send((
job_id,
"test-user".to_string(),
SseEvent::JobMessage {
job_id: job_id.to_string(),
role: "user".to_string(),
@@ -402,7 +409,7 @@ mod tests {
.await
.unwrap();
let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
let handle = spawn_job_monitor_with_context(
@@ -417,6 +424,7 @@ mod tests {
event_tx
.send((
job_id,
"test-user".to_string(),
SseEvent::JobResult {
job_id: job_id.to_string(),
status: "completed".to_string(),
@@ -450,7 +458,7 @@ mod tests {
.await
.unwrap();
let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
let handle = spawn_job_monitor_with_context(
@@ -465,6 +473,7 @@ mod tests {
event_tx
.send((
job_id,
"test-user".to_string(),
SseEvent::JobResult {
job_id: job_id.to_string(),
status: "failed".to_string(),
@@ -498,12 +507,13 @@ mod tests {
.await
.unwrap();
let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
let handle = spawn_completion_watcher(job_id, event_tx.subscribe(), Arc::clone(&cm));
event_tx
.send((
job_id,
"test-user".to_string(),
SseEvent::JobResult {
job_id: job_id.to_string(),
status: "completed".to_string(),
src/agent/routine_engine.rs
+18 -2
View File
@@ -1305,6 +1305,19 @@ async fn execute_lightweight(
}
}
/// Sanitize a user-controlled string before interpolation into an LLM prompt.
/// Strips newlines (which could break prompt structure) and truncates to a
/// reasonable length to limit abuse surface.
fn sanitize_prompt_field(value: &str) -> String {
const MAX_LEN: usize = 128;
value
.chars()
.filter(|&c| c != '\n' && c != '\r')
.take(MAX_LEN)
.map(|c| if c == '`' { '\'' } else { c })
.collect()
}
fn build_lightweight_prompt(
prompt: &str,
context_parts: &[String],
@@ -1323,14 +1336,16 @@ fn build_lightweight_prompt(
);
if let Some(channel) = notify.channel.as_deref() {
let sanitized = sanitize_prompt_field(channel);
full_prompt.push_str(&format!(
"The configured delivery channel for this routine is `{channel}`.\n"
"The configured delivery channel for this routine is `{sanitized}`.\n"
));
}
if let Some(user) = notify.user.as_deref() {
let sanitized = sanitize_prompt_field(user);
full_prompt.push_str(&format!(
"The configured delivery target for this routine is `{user}`.\n"
"The configured delivery target for this routine is `{sanitized}`.\n"
));
}
@@ -1440,6 +1455,7 @@ fn handle_text_response(
/// This is a simplified version of the full dispatcher loop:
/// - Max 3-5 iterations (configurable)
/// - Sequential tool execution (not parallel)
/// - Uses the owner's live autonomous tool scope when lightweight tools are enabled
/// - Auto-approval of non-Always tools
/// - No hooks or approval dialogs
async fn execute_lightweight_with_tools(
src/agent/scheduler.rs
+6 -11
View File
@@ -9,7 +9,6 @@ use tokio::task::JoinHandle;
use uuid::Uuid;
use crate::agent::task::{Task, TaskContext, TaskOutput};
use crate::channels::web::types::SseEvent;
use crate::config::AgentConfig;
use crate::context::{ContextManager, JobContext, JobState};
use crate::db::Database;
@@ -67,8 +66,8 @@ pub struct Scheduler {
extension_manager: Option<Arc<ExtensionManager>>,
store: Option<Arc<dyn Database>>,
hooks: Arc<HookRegistry>,
/// SSE broadcast sender for live job event streaming.
sse_tx: Option<tokio::sync::broadcast::Sender<SseEvent>>,
/// SSE manager for live job event streaming.
sse_tx: Option<Arc<crate::channels::web::sse::SseManager>>,
/// HTTP interceptor for trace recording/replay (propagated to workers).
http_interceptor: Option<Arc<dyn crate::llm::recording::HttpInterceptor>>,
/// Running jobs (main LLM-driven jobs).
@@ -102,9 +101,9 @@ impl Scheduler {
}
}
/// Set the SSE broadcast sender for live job event streaming.
pub fn set_sse_sender(&mut self, tx: tokio::sync::broadcast::Sender<SseEvent>) {
self.sse_tx = Some(tx);
/// Set the SSE manager for live job event streaming.
pub fn set_sse_sender(&mut self, sse: Arc<crate::channels::web::sse::SseManager>) {
self.sse_tx = Some(sse);
}
/// Set the HTTP interceptor for trace recording/replay.
@@ -549,11 +548,7 @@ impl Scheduler {
// Delegate to shared tool execution pipeline
let output_str = crate::tools::execute::execute_tool_with_safety(
&tools,
&safety,
tool_name,
&normalized_params,
&job_ctx,
&tools, &safety, tool_name, params, &job_ctx,
)
.await?;
src/agent/session.rs
+22 -8
View File
@@ -17,7 +17,7 @@ use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::channels::web::util::truncate_preview;
use crate::llm::{ChatMessage, ToolCall};
use crate::llm::{ChatMessage, ToolCall, generate_tool_call_id};
/// A session containing one or more threads.
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -414,7 +414,12 @@ impl Thread {
/// completed actions in subsequent turns.
pub fn messages(&self) -> Vec<ChatMessage> {
let mut messages = Vec::new();
for turn in &self.turns {
// We use the enumeration index (`turn_idx`) rather than `turn.turn_number`
// intentionally: after `truncate_turns()`, the remaining turns are
// re-numbered starting from 0, so the enumeration index and turn_number
// are equivalent. Using the index avoids coupling to the field and keeps
// tool-call ID generation deterministic for the current message window.
for (turn_idx, turn) in self.turns.iter().enumerate() {
if turn.image_content_parts.is_empty() {
messages.push(ChatMessage::user(&turn.user_input));
} else {
@@ -425,13 +430,23 @@ impl Thread {
}
if !turn.tool_calls.is_empty() {
// Build ToolCall objects with synthetic stable IDs
let tool_calls: Vec<ToolCall> = turn
// Assign synthetic call IDs for this turn's tool calls, so that
// declarations and results can be consistently correlated.
let tool_calls_with_ids: Vec<(String, &_)> = turn
.tool_calls
.iter()
.enumerate()
.map(|(i, tc)| ToolCall {
id: format!("turn{}_{}", turn.turn_number, i),
.map(|(tc_idx, tc)| {
// Use provider-compatible tool call IDs derived from turn/tool indices.
(generate_tool_call_id(turn_idx, tc_idx), tc)
})
.collect();
// Build ToolCall objects using the synthetic call IDs.
let tool_calls: Vec<ToolCall> = tool_calls_with_ids
.iter()
.map(|(call_id, tc)| ToolCall {
id: call_id.clone(),
name: tc.name.clone(),
arguments: tc.parameters.clone(),
})
@@ -441,8 +456,7 @@ impl Thread {
messages.push(ChatMessage::assistant_with_tool_calls(None, tool_calls));
// Individual tool result messages, truncated to limit context size.
for (i, tc) in turn.tool_calls.iter().enumerate() {
let call_id = format!("turn{}_{}", turn.turn_number, i);
for (call_id, tc) in tool_calls_with_ids {
let content = if let Some(ref err) = tc.error {
// .error already contains the full error text;
// pass through without wrapping to avoid double-prefix.
src/agent/thread_ops.rs
+1 -1
View File
@@ -1646,7 +1646,7 @@ impl Agent {
};
match ext_mgr
.configure_token(&pending.extension_name, token)
.configure_token(&pending.extension_name, token, &message.user_id)
.await
{
Ok(result) if result.activated => {