gitops/optimclaw
1
0
Fork 0
mirror of https://github.com/outbackdingo/optimclaw.git synced 2026-08-25 14:53:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: implement FullJob routine mode with scheduler dispatch (#288)

Browse Source 
* feat: implement FullJob routine mode with scheduler dispatch

FullJob routines previously fell back to lightweight mode (single LLM call,
no tools) with a warning. This wires them to the existing Scheduler/Worker
infrastructure so they dispatch real jobs with full tool access.

Fire-and-forget model: the routine creates a job via ContextManager, schedules
it, links the routine_run to the job_id, and completes immediately. The job
runs independently with full tool access.

- Add RoutineError::JobDispatchFailed variant
- Add RoutineStore::link_routine_run_to_job (PostgreSQL + libSQL)
- Add execute_full_job() in routine_engine with context_manager/scheduler
- Wire context_manager + scheduler into RoutineEngine from agent_loop
- Fix pre-existing clippy warnings in tests/html_to_markdown.rs

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: persist job to DB before scheduling in execute_full_job

The worker emits job_actions and llm_calls rows that reference agent_jobs
via foreign key. Without persisting the job first, those inserts can fail.
Match the pattern from commands.rs: fetch JobContext, save_job(), then schedule.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* refactor: consolidate job dispatch into Scheduler::dispatch_job and wire max_iterations

Move the create + persist + schedule sequence into a single
Scheduler::dispatch_job() method so callers (commands.rs, routine_engine.rs)
don't duplicate the logic. FullJob routines now pass max_iterations via job
metadata, and the worker reads it (defaulting to 50 if unset).

Also removes the context_manager field from RoutineEngine since dispatch_job
handles everything internally.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: clamp max_iterations to 500 and log category update failures

Address PR review feedback:
- worker.rs: clamp max_iterations from metadata to MAX_WORKER_ITERATIONS (500)
  to prevent unbounded LLM token usage from malicious/buggy configs
- commands.rs: log warning on category update failure instead of silently
  discarding the error

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* feat: persist worker events to DB and fix activity tab rendering

In-process Worker (used by Scheduler::dispatch_job) now persists events
via save_job_event at key execution points: plan creation, LLM
responses, tool_use, tool_result, and job completion/failure/stuck.
Event data shapes match the container worker format so the gateway
activity tab renders them correctly.

Frontend: tool_result errors now show a red X icon with danger styling
instead of a silent empty output. The result event falls back to the
error field when message is absent.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

---------

Co-authored-by: Claude Opus 4.6 <[email protected]>
This commit is contained in:
Illia Polosukhin
2026-02-22 08:18:21 +00:00
committed by GitHub
co-authored by Claude Opus 4.6
parent ea57447649
commit 04d3b005b1
12 changed files with 279 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agent/agent_loop.rs
+1
View File
@@ -397,6 +397,7 @@ impl Agent {
self.llm().clone(),
Arc::clone(workspace),
notify_tx,
Some(self.scheduler.clone()),
));
// Register routine tools
src/agent/commands.rs
+8 -21
View File
@@ -73,36 +73,23 @@ impl Agent {
description: String,
category: Option<String>,
) -> Result<String, Error> {
// Create job context
let job_id = self
.context_manager
.create_job_for_user(user_id, &title, &description)
.scheduler
.dispatch_job(user_id, &title, &description, None)
.await?;
// Update category if provided
if let Some(cat) = category {
self.context_manager
// Set the dedicated category field (not stored in metadata)
if let Some(cat) = category
&& let Err(e) = self
.context_manager
.update_context(job_id, |ctx| {
ctx.category = Some(cat);
})
.await?;
}
// Persist new job to database (fire-and-forget)
if let Some(store) = self.store()
&& let Ok(ctx) = self.context_manager.get_context(job_id).await
.await
{
let store = store.clone();
tokio::spawn(async move {
if let Err(e) = store.save_job(&ctx).await {
tracing::warn!("Failed to persist new job {}: {}", job_id, e);
}
});
tracing::warn!(job_id = %job_id, "Failed to set job category: {}", e);
}
// Schedule for execution
self.scheduler.schedule(job_id).await?;
Ok(format!(
"Created job: {}\nID: {}\n\nThe job has been scheduled and is now running.",
title, job_id
src/agent/routine_engine.rs
+64 -26
View File
@@ -19,6 +19,7 @@ use regex::Regex;
use tokio::sync::{RwLock, mpsc};
use uuid::Uuid;
use crate::agent::Scheduler;
use crate::agent::routine::{
NotifyConfig, Routine, RoutineAction, RoutineRun, RunStatus, Trigger, next_cron_fire,
};
@@ -41,6 +42,8 @@ pub struct RoutineEngine {
running_count: Arc<AtomicUsize>,
/// Compiled event regex cache: routine_id -> compiled regex.
event_cache: Arc<RwLock<Vec<(Uuid, Routine, Regex)>>>,
/// Scheduler for dispatching jobs (FullJob mode).
scheduler: Option<Arc<Scheduler>>,
}
impl RoutineEngine {
@@ -50,6 +53,7 @@ impl RoutineEngine {
llm: Arc<dyn LlmProvider>,
workspace: Arc<Workspace>,
notify_tx: mpsc::Sender<OutgoingResponse>,
scheduler: Option<Arc<Scheduler>>,
) -> Self {
Self {
config,
@@ -59,6 +63,7 @@ impl RoutineEngine {
notify_tx,
running_count: Arc::new(AtomicUsize::new(0)),
event_cache: Arc::new(RwLock::new(Vec::new())),
scheduler,
}
}
@@ -225,7 +230,7 @@ impl RoutineEngine {
workspace: self.workspace.clone(),
notify_tx: self.notify_tx.clone(),
running_count: self.running_count.clone(),
max_lightweight_tokens: self.config.max_lightweight_tokens,
scheduler: self.scheduler.clone(),
};
tokio::spawn(async move {
@@ -257,7 +262,7 @@ impl RoutineEngine {
workspace: self.workspace.clone(),
notify_tx: self.notify_tx.clone(),
running_count: self.running_count.clone(),
max_lightweight_tokens: self.config.max_lightweight_tokens,
scheduler: self.scheduler.clone(),
};
// Record the run in DB, then spawn execution
@@ -304,7 +309,7 @@ struct EngineContext {
workspace: Arc<Workspace>,
notify_tx: mpsc::Sender<OutgoingResponse>,
running_count: Arc<AtomicUsize>,
max_lightweight_tokens: u32,
scheduler: Option<Arc<Scheduler>>,
}
/// Execute a routine run. Handles both lightweight and full_job modes.
@@ -318,29 +323,11 @@ async fn execute_routine(ctx: EngineContext, routine: Routine, run: RoutineRun)
context_paths,
max_tokens,
} => execute_lightweight(&ctx, &routine, prompt, context_paths, *max_tokens).await,
RoutineAction::FullJob { description, .. } => {
// Full job mode: scheduler integration not yet implemented.
// Execute as lightweight and prepend a warning to the summary.
tracing::warn!(
routine = %routine.name,
"FullJob mode not yet implemented; falling back to lightweight execution"
);
match execute_lightweight(&ctx, &routine, description, &[], ctx.max_lightweight_tokens)
.await
{
Ok((status, summary, tokens)) => {
let warning = "[Note: FullJob mode is not yet implemented. This routine ran as \
a single LLM call without tool access. Configure as 'lightweight' \
or wait for full scheduler integration.]";
let summary = match summary {
Some(s) => Some(format!("{warning}\n\n{s}")),
None => Some(warning.to_string()),
};
Ok((status, summary, tokens))
}
Err(e) => Err(e),
}
}
RoutineAction::FullJob {
title,
description,
max_iterations,
} => execute_full_job(&ctx, &routine, &run, title, description, *max_iterations).await,
};
// Decrement running count
@@ -418,6 +405,57 @@ fn sanitize_routine_name(name: &str) -> String {
.collect()
}
/// Execute a full-job routine by dispatching to the scheduler.
///
/// Fire-and-forget: creates a job via `Scheduler::dispatch_job` (which handles
/// creation, metadata, persistence, and scheduling), links the routine run to
/// the job, and returns immediately. The job runs independently via the
/// existing Worker/Scheduler with full tool access.
async fn execute_full_job(
ctx: &EngineContext,
routine: &Routine,
run: &RoutineRun,
title: &str,
description: &str,
max_iterations: u32,
) -> Result<(RunStatus, Option<String>, Option<i32>), RoutineError> {
let scheduler = ctx
.scheduler
.as_ref()
.ok_or_else(|| RoutineError::JobDispatchFailed {
reason: "scheduler not available".to_string(),
})?;
let metadata = serde_json::json!({ "max_iterations": max_iterations });
let job_id = scheduler
.dispatch_job(&routine.user_id, title, description, Some(metadata))
.await
.map_err(|e| RoutineError::JobDispatchFailed {
reason: format!("failed to dispatch job: {e}"),
})?;
// Link the routine run to the dispatched job
if let Err(e) = ctx.store.link_routine_run_to_job(run.id, job_id).await {
tracing::error!(
routine = %routine.name,
"Failed to link run to job: {}", e
);
}
tracing::info!(
routine = %routine.name,
job_id = %job_id,
max_iterations = max_iterations,
"Dispatched full job for routine"
);
let summary = format!(
"Dispatched job {job_id} for full execution with tool access (max_iterations: {max_iterations})"
);
Ok((RunStatus::Ok, Some(summary), None))
}
/// Execute a lightweight routine (single LLM call).
async fn execute_lightweight(
ctx: &EngineContext,
src/agent/scheduler.rs
+44
View File
@@ -81,6 +81,50 @@ impl Scheduler {
}
}
/// Create, persist, and schedule a job in one shot.
///
/// This is the preferred entry point for dispatching new jobs. It:
/// 1. Creates the job context via `ContextManager`
/// 2. Optionally applies metadata (e.g. `max_iterations`)
/// 3. Persists the job to the database (so FK references from
/// `job_actions` / `llm_calls` work immediately)
/// 4. Schedules the job for worker execution
///
/// Returns the new job ID.
pub async fn dispatch_job(
&self,
user_id: &str,
title: &str,
description: &str,
metadata: Option<serde_json::Value>,
) -> Result<Uuid, JobError> {
let job_id = self
.context_manager
.create_job_for_user(user_id, title, description)
.await?;
// Apply metadata if provided
if let Some(meta) = metadata {
self.context_manager
.update_context(job_id, |ctx| {
ctx.metadata = meta;
})
.await?;
}
// Persist to DB before scheduling so the worker's FK references are valid
if let Some(ref store) = self.store {
let ctx = self.context_manager.get_context(job_id).await?;
store.save_job(&ctx).await.map_err(|e| JobError::Failed {
id: job_id,
reason: format!("failed to persist job: {e}"),
})?;
}
self.schedule(job_id).await?;
Ok(job_id)
}
/// Schedule a job for execution.
pub async fn schedule(&self, job_id: Uuid) -> Result<(), JobError> {
// Hold write lock for the entire check-insert sequence to prevent
src/agent/worker.rs
+94 -1
View File
@@ -98,6 +98,20 @@ impl Worker {
}
}
/// Fire-and-forget persistence of a job event.
fn log_event(&self, event_type: &str, data: serde_json::Value) {
if let Some(store) = self.store() {
let store = store.clone();
let job_id = self.job_id;
let event_type = event_type.to_string();
tokio::spawn(async move {
if let Err(e) = store.save_job_event(job_id, &event_type, &data).await {
tracing::warn!("Failed to persist event for job {}: {}", job_id, e);
}
});
}
}
/// Run the worker until the job is complete or stopped.
pub async fn run(self, mut rx: mpsc::Receiver<WorkerMessage>) -> Result<(), Error> {
tracing::info!("Worker starting for job {}", self.job_id);
@@ -164,7 +178,15 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
reasoning: &Reasoning,
reason_ctx: &mut ReasoningContext,
) -> Result<(), Error> {
let max_iterations = 50;
const MAX_WORKER_ITERATIONS: usize = 500;
let max_iterations = self
.context_manager()
.get_context(self.job_id)
.await
.ok()
.and_then(|ctx| ctx.metadata.get("max_iterations").and_then(|v| v.as_u64()))
.unwrap_or(50) as usize;
let max_iterations = max_iterations.min(MAX_WORKER_ITERATIONS);
let mut iteration = 0;
// Initial tool definitions for planning (will be refreshed in loop)
@@ -193,6 +215,14 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
.join("\n")
)));
self.log_event("message", serde_json::json!({
"role": "assistant",
"content": format!("Plan: {}\n\n{}", p.goal,
p.actions.iter().enumerate()
.map(|(i, a)| format!("{}. {} - {}", i + 1, a.tool_name, a.reasoning))
.collect::<Vec<_>>().join("\n"))
}));
Some(p)
}
Err(e) => {
@@ -267,6 +297,14 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
// Add assistant response to context
reason_ctx.messages.push(ChatMessage::assistant(&response));
self.log_event(
"message",
serde_json::json!({
"role": "assistant",
"content": response,
}),
);
// Give it one more chance to select a tool
if iteration > 3 && iteration % 5 == 0 {
reason_ctx.messages.push(ChatMessage::user(
@@ -285,6 +323,16 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
tool_calls.len()
);
if let Some(ref text) = content {
self.log_event(
"message",
serde_json::json!({
"role": "assistant",
"content": text,
}),
);
}
// Add assistant message with tool_calls (OpenAI protocol)
reason_ctx
.messages
@@ -667,6 +715,15 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
selection: &ToolSelection,
result: Result<String, Error>,
) -> Result<bool, Error> {
self.log_event(
"tool_use",
serde_json::json!({
"tool_name": selection.tool_name,
"input": crate::agent::agent_loop::truncate_for_preview(
&selection.parameters.to_string(), 500),
}),
);
match result {
Ok(output) => {
// Sanitize output
@@ -687,6 +744,12 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
wrapped,
));
self.log_event("tool_result", serde_json::json!({
"tool_name": selection.tool_name,
"success": true,
"output": crate::agent::agent_loop::truncate_for_preview(&sanitized.content, 500),
}));
// Tool output never drives job completion. A malicious tool could
// emit "TASK_COMPLETE" to force premature completion. Only the LLM's
// own structured response (in execution_loop) can mark a job done.
@@ -713,6 +776,15 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
});
}
self.log_event(
"tool_result",
serde_json::json!({
"tool_name": selection.tool_name,
"success": false,
"output": format!("Error: {}", e),
}),
);
reason_ctx.messages.push(ChatMessage::tool_result(
&selection.tool_call_id,
&selection.tool_name,
@@ -834,6 +906,13 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
reason: s,
})?;
self.log_event(
"result",
serde_json::json!({
"success": true,
"message": "Job completed successfully",
}),
);
self.persist_status(
JobState::Completed,
Some("Job completed successfully".to_string()),
@@ -852,6 +931,13 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
reason: s,
})?;
self.log_event(
"result",
serde_json::json!({
"success": false,
"message": format!("Execution failed: {}", reason),
}),
);
self.persist_status(JobState::Failed, Some(reason.to_string()));
Ok(())
}
@@ -865,6 +951,13 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
reason: s,
})?;
self.log_event(
"result",
serde_json::json!({
"success": false,
"message": format!("Job stuck: {}", reason),
}),
);
self.persist_status(JobState::Stuck, Some(reason.to_string()));
Ok(())
}
src/channels/web/static/app.js
+14 -6
View File
@@ -2187,14 +2187,20 @@ function appendActivityEvent(terminal, eventType, data) {
+ escapeHtml(typeof data.input === 'string' ? data.input : JSON.stringify(data.input, null, 2))
+ '</pre></details>';
break;
case 'tool_result':
el.innerHTML = '<details class="activity-tool-block activity-tool-result"><summary>'
+ '<span class="activity-tool-icon">&#10003;</span> '
case 'tool_result': {
const trSuccess = data.success !== false;
const trIcon = trSuccess ? '&#10003;' : '&#10007;';
const trOutput = data.output || data.error || '';
const trClass = 'activity-tool-block activity-tool-result'
+ (trSuccess ? '' : ' activity-tool-error');
el.innerHTML = '<details class="' + trClass + '"><summary>'
+ '<span class="activity-tool-icon">' + trIcon + '</span> '
+ escapeHtml(data.tool_name || 'result')
+ '</summary><pre class="activity-tool-output">'
+ escapeHtml(data.output || '')
+ escapeHtml(trOutput)
+ '</pre></details>';
break;
}
case 'status':
el.innerHTML = '<span class="activity-status">' + escapeHtml(data.message || '') + '</span>';
break;
@@ -2202,7 +2208,7 @@ function appendActivityEvent(terminal, eventType, data) {
el.className += ' activity-final';
const success = data.success !== false;
el.innerHTML = '<span class="activity-result-status" data-success="' + success + '">'
+ escapeHtml(data.message || data.status || 'done') + '</span>';
+ escapeHtml(data.message || data.error || data.status || 'done') + '</span>';
if (data.session_id) {
el.innerHTML += ' <span class="activity-session-id">session: ' + escapeHtml(data.session_id) + '</span>';
}
@@ -2387,7 +2393,9 @@ function renderRoutineDetail(routine) {
+ '<td>' + formatDate(run.started_at) + '</td>'
+ '<td>' + formatDate(run.completed_at) + '</td>'
+ '<td><span class="badge ' + runStatusClass + '">' + escapeHtml(run.status) + '</span></td>'
+ '<td>' + escapeHtml(run.result_summary || '-') + '</td>'
+ '<td>' + escapeHtml(run.result_summary || '-')
+ (run.job_id ? ' <a href="#" onclick="event.preventDefault(); switchTab(\'jobs\'); openJobDetail(\'' + run.job_id + '\')">[view job]</a>' : '')
+ '</td>'
+ '<td>' + (run.tokens_used != null ? run.tokens_used : '-') + '</td>'
+ '</tr>';
}
src/channels/web/static/style.css
+8
View File
@@ -2240,6 +2240,14 @@ body {
color: var(--success);
}
.activity-tool-error .activity-tool-icon {
color: var(--danger);
}
.activity-tool-error summary {
color: var(--danger);
}
.activity-tool-input,
.activity-tool-output {
padding: 8px 10px;
src/db/libsql/routines.rs
+15
View File
@@ -387,4 +387,19 @@ impl RoutineStore for LibSqlBackend {
None => Ok(0),
}
}
async fn link_routine_run_to_job(
&self,
run_id: Uuid,
job_id: Uuid,
) -> Result<(), DatabaseError> {
let conn = self.connect().await?;
conn.execute(
"UPDATE routine_runs SET job_id = ?1 WHERE id = ?2",
params![job_id.to_string(), run_id.to_string()],
)
.await
.map_err(|e| DatabaseError::Query(e.to_string()))?;
Ok(())
}
}
src/db/mod.rs
+5
View File
@@ -274,6 +274,11 @@ pub trait RoutineStore: Send + Sync {
limit: i64,
) -> Result<Vec<RoutineRun>, DatabaseError>;
async fn count_running_routine_runs(&self, routine_id: Uuid) -> Result<i64, DatabaseError>;
async fn link_routine_run_to_job(
&self,
run_id: Uuid,
job_id: Uuid,
) -> Result<(), DatabaseError>;
}
#[async_trait]
src/db/postgres.rs
+8
View File
@@ -437,6 +437,14 @@ impl RoutineStore for PgBackend {
async fn count_running_routine_runs(&self, routine_id: Uuid) -> Result<i64, DatabaseError> {
self.store.count_running_routine_runs(routine_id).await
}
async fn link_routine_run_to_job(
&self,
run_id: Uuid,
job_id: Uuid,
) -> Result<(), DatabaseError> {
self.store.link_routine_run_to_job(run_id, job_id).await
}
}
// ==================== ToolFailureStore ====================
src/error.rs
+3
View File
@@ -407,6 +407,9 @@ pub enum RoutineError {
#[error("LLM call failed: {reason}")]
LlmFailed { reason: String },
#[error("Failed to dispatch full job: {reason}")]
JobDispatchFailed { reason: String },
#[error("LLM returned empty content")]
EmptyResponse,
src/history/store.rs
+15
View File
@@ -1167,6 +1167,21 @@ impl Store {
.await?;
Ok(row.get("cnt"))
}
/// Link a routine run to a dispatched job.
pub async fn link_routine_run_to_job(
&self,
run_id: Uuid,
job_id: Uuid,
) -> Result<(), DatabaseError> {
let conn = self.conn().await?;
conn.execute(
"UPDATE routine_runs SET job_id = $1 WHERE id = $2",
&[&job_id, &run_id],
)
.await?;
Ok(())
}
}
#[cfg(feature = "postgres")]