mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-25 14:53:34 +00:00
* fix(ci): secrets can't be used in step if conditions [skip-regression-check] (#787) GitHub Actions step-level `if:` doesn't have access to `secrets` context. Replace `if: secrets.X != ''` with `continue-on-error: true` and let the Set token step handle the fallback. Co-authored-by: Claude Sonnet 4.6 <[email protected]> * fix(ci): clean up staging pipeline — remove hacks, skip redundant checks [skip-regression-check] (#794) - Remove continue-on-error from staging-ci.yml app token steps (secrets are configured) - Skip test.yml and code_style.yml on PRs targeting staging (staging-ci.yml already runs tests before promoting, promotion PR gets full CI on main) - Allow ironclaw-ci[bot] in Claude Code review for bot-created promotion PRs Co-authored-by: Claude Opus 4.6 <[email protected]> * fix(ci): run fmt + clippy on staging PRs, skip Windows clippy [skip-regression-check] (#802) - Remove branches:[main] filter from code_style.yml so it runs on all PRs - Gate clippy-windows with `if: github.base_ref == 'main'` (skip on staging PRs) - Update rollup job to allow skipped clippy-windows - Simplify claude-review.yml to only trigger on labeled event (avoids duplicate runs) Co-authored-by: Claude Opus 4.6 <[email protected]> * feat: persist user_id in save_job and expose job_id on routine runs (#709) * 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]> * feat: wire RoutineEngine into gateway for direct manual trigger firing Replace the message-channel hack in routines_trigger_handler with a direct call to RoutineEngine::fire_manual(), ensuring FullJob routines dispatch correctly when triggered from the web UI. Inject the engine into GatewayState from Agent::run after construction. Also persists user_id in save_job for both PG and libSQL backends, removes the source='sandbox' filter so all jobs are visible, and exposes job_id on RoutineRunInfo for the frontend job link. Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: remove stale gateway_state argument from Agent::new test call sites The gateway_state parameter was removed from Agent::new during rebase (replaced by post-construction set_routine_engine_slot), but three test call sites still passed the extra None argument. Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address PR review — restore sandbox source filter, remove blank lines - Revert removal of `source = 'sandbox'` filter in all SandboxStore queries (8 sites across PG and libSQL). Sandbox-specific APIs should stay scoped to sandbox jobs; unified job listing for the Jobs tab should use a separate query path. - Remove extra blank lines in agent_loop.rs and worker.rs that caused formatting CI failure. [skip-regression-check] Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address review — regenerate Cargo.lock, add user_id regression test - Regenerate Cargo.lock from main's lockfile to eliminate dependency version downgrades (anyhow, syn, etc.) that were churn from rebase. - Add regression test verifying user_id round-trips through save_job and get_job in the libSQL backend. Co-Authored-By: Claude Opus 4.6 <[email protected]> * style: remove trailing blank line in libsql jobs.rs [skip-regression-check] Co-Authored-By: Claude Opus 4.6 <[email protected]> * test: add Postgres-side regression test for user_id persistence in save_job Mirrors the existing libSQL test (test_save_job_persists_user_id) for the Postgres backend. Gated behind #[cfg(feature = "postgres")] + #[ignore] since it requires a running PostgreSQL instance (integration tier). Co-Authored-By: Claude Opus 4.6 <[email protected]> --------- Co-authored-by: Claude Opus 4.6 <[email protected]> * refactor: unify three agentic loops into single AgenticLoop engine (#654) Replace three independent copy-pasted agentic loops (dispatcher, worker, container runtime) with a single shared engine in `agentic_loop.rs` that all consumers customize via the `LoopDelegate` trait. Phase 1 — Shared engine (`src/agent/agentic_loop.rs`, 205 lines): - `run_agentic_loop()` owns the core LLM → tool exec → repeat cycle - `LoopDelegate` trait (Send + Sync, &dyn dispatch) with 6 hook points - Tool intent nudge logic consolidated (was duplicated in 3 files) - Iteration limit + force-text behavior preserved Phase 2 — Three delegate implementations: - `ChatDelegate` (dispatcher.rs): 3-phase approval flow, hooks, cost guard, context compaction, skill attenuation, interruption - `JobDelegate` (worker/job.rs): planning pre-loop phase, parallel JoinSet exec, mark_completed/stuck/failed, SSE streaming, self-repair - `ContainerDelegate` (worker/container.rs): sequential tool exec, HTTP-proxied LLM, container-safe tools, credential injection Phase 3 — File moves and cleanup: - Delete `src/agent/worker.rs` — job logic moved to `src/worker/job.rs` - Rename `src/worker/runtime.rs` → `src/worker/container.rs` - Re-export `Worker`/`WorkerDeps` from `crate::worker` in `agent/mod.rs` - Update `scheduler.rs` imports to new worker location Shared helpers (`src/tools/execute.rs`): - `execute_tool_with_safety()` replaces 4 copies of validate → timeout → execute → serialize - `process_tool_result()` replaces 3 copies of sanitize → wrap → ChatMessage (also used by thread_ops.rs approval resume paths) Net result: -2,408 lines, zero duplicated loop logic, single code path for tool intent nudge and completion detection. Closes #654 Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address review feedback from Copilot 1. scheduler.rs: Replace `unwrap_or` fallback with proper error propagation when parsing tool output JSON — surfaces bugs instead of silently changing the output type. 2. worker/job.rs: Drop MutexGuard before the cancellation `.await` in `check_signals()` to avoid holding a lock across an async I/O call (prevents `await_holding_lock` lint). 3. worker/job.rs: Restore consecutive rate-limit counter (MAX_CONSECUTIVE_RATE_LIMITS = 10) so sustained rate limiting marks the job stuck with "Persistent rate limiting" instead of silently burning through max_iterations. Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: incorporate staging changes — token budget tracking + mark_failed Merge staging's changes into the refactored JobDelegate: - Add token budget tracking in call_llm (update_context/add_tokens) - mark_stuck → mark_failed for iteration cap and rate-limit exhaustion (aligns with staging's #788 fix) Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address zmanian's PR review — eliminate type erasure, clean up Address all 6 review points from zmanian on PR #800: 1. Replace LoopOutcome::Custom(Box<dyn Any>) with typed LoopOutcome::NeedApproval(Box<PendingApproval>) — eliminates type erasure and downcast, resolves clippy large_enum_variant. 2. Remove dead max_tool_iterations field from ChatDelegate struct. 3. Add on_tool_intent_nudge() hook to LoopDelegate trait with implementations in Job and Container delegates for observability. 4. Fix SSE events in job worker to emit raw sanitized content instead of XML-wrapped <tool_output> tags. 5. Remove 4 duplicate completion tests from job.rs that were already covered by the shared util module. 6. Avoid logging full tool results — use result_size_bytes in debug logs (execute.rs, job.rs). Also updates path references in CLAUDE.md, COVERAGE_PLAN.md, and add-sse-event.md command. Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat(doctor): expand diagnostics from 7 to 16 health checks * test: add unit tests for agentic_loop and execute shared modules Add 16 tests covering the two new critical shared modules: agentic_loop.rs (10 tests): - Text response exits loop immediately - Tool call → text response continuation - LoopSignal::Stop exits before LLM call - LoopSignal::InjectMessage adds user message to context - Max iterations terminates with LoopOutcome::MaxIterations - Tool intent nudge fires twice then caps - before_llm_call early exit bypasses LLM - truncate_for_preview: short string, long string, multibyte safety execute.rs (6 tests): - execute_tool_with_safety success path - Missing tool returns ToolError::NotFound - Tool execution failure propagates - Per-tool timeout enforcement (50ms) - process_tool_result XML wrapping on success - process_tool_result error formatting All 2,777 unit tests pass, 0 clippy warnings. Co-Authored-By: Claude Opus 4.6 <[email protected]> * style: cargo fmt Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address code review — 9 issues across agentic loop, job worker, container CRITICAL fixes: - Rate-limit exhaustion now returns Err(LlmError::RateLimited) instead of Ok(Text("")), stopping the loop immediately with no ghost iteration. Below-threshold retries still use Text("") with an explicit empty-string guard in handle_text_response to skip injection. - check_signals drains the entire message channel before returning, prioritizing Stop over UserMessage. Previously returned early on first UserMessage, silently dropping any queued Stop or additional messages. - check_signals now detects all non-progressing job states (Cancelled, Failed, Stuck, Completed, Submitted, Accepted) instead of only Cancelled and Failed. HIGH fixes: - Error path in process_tool_result_job applies truncate_for_preview to bound error strings in SSE/DB events (was unbounded). - Document Send+Sync lifetime constraint on LoopDelegate trait. - Test mock before_llm_call refactored from double-lock to single lock acquisition, eliminating deadlock risk on refactor. MEDIUM fixes: - CompletionReport includes actual iteration count via shared Arc<Mutex<u32>> tracker (was hardcoded 0). - process_tool_result_job return type changed from Result<bool> to Result<()> — the bool was always false (dead API). - Deduplicate truncate in container.rs; now uses truncate_for_preview from agentic_loop. Verified: 0 clippy warnings, 2781 tests pass, cargo fmt clean. Co-Authored-By: Claude Opus 4.6 <[email protected]> --------- Co-authored-by: Henry Park <[email protected]> Co-authored-by: Claude Sonnet 4.6 <[email protected]> Co-authored-by: Illia Polosukhin <[email protected]> Co-authored-by: Umesh Kumar Singh <[email protected]> Co-authored-by: reidliu41 <[email protected]>
This commit is contained in:
Zaki Manian
co-authored by
Henry Park
Claude Sonnet 4.6
Illia Polosukhin
Umesh Kumar Singh
reidliu41
Henry Park
Claude Sonnet 4.6
Illia Polosukhin
Umesh Kumar Singh
reidliu41
parent
ebb22094a5
commit
88f4894a18
@@ -64,7 +64,7 @@ If the event needs custom UI (cards, badges, etc.), add styles. Follow the exist
|
||||
|
||||
Identify where in the backend this event should be triggered. Common locations:
|
||||
- `src/agent/agent_loop.rs` - During message processing or tool execution
|
||||
- `src/agent/worker.rs` - During job execution
|
||||
- `src/worker/job.rs` - During job execution
|
||||
- `src/agent/heartbeat.rs` - During periodic execution
|
||||
|
||||
Use the existing pattern:
|
||||
|
||||
@@ -99,7 +99,8 @@ src/
|
||||
│ └── job_manager.rs # Container lifecycle (create, stop, cleanup)
|
||||
│
|
||||
├── worker/ # Runs inside Docker containers
|
||||
│ ├── runtime.rs # Worker execution loop (tool calls, LLM)
|
||||
│ ├── container.rs # Container worker runtime (ContainerDelegate + shared agentic loop)
|
||||
│ ├── job.rs # Background job worker (JobDelegate + shared agentic loop)
|
||||
│ ├── claude_bridge.rs # Claude Code bridge (spawns claude CLI)
|
||||
│ └── proxy_llm.rs # LlmProvider that proxies through orchestrator
|
||||
│
|
||||
|
||||
+4
-4
@@ -63,12 +63,12 @@ These files account for the vast majority of the coverage gap:
|
||||
| `src/main.rs` | 740 | 522 | 29.4% | 485 |
|
||||
| `src/channels/web/handlers/jobs.rs` | 513 | 456 | 11.1% | 430 |
|
||||
| `src/tools/builder/core.rs` | 524 | 456 | 13.0% | 429 |
|
||||
| `src/agent/worker.rs` | 1,078 | 467 | 56.7% | 413 |
|
||||
| `src/worker/job.rs` | 1,078 | 467 | 56.7% | 413 |
|
||||
| `src/channels/web/handlers/chat.rs` | 564 | 417 | 26.1% | 388 |
|
||||
| `src/tools/wasm/wrapper.rs` | 1,005 | 436 | 56.6% | 385 |
|
||||
| `src/channels/signal.rs` | 1,814 | 472 | 74.0% | 381 |
|
||||
| `src/tools/mcp/auth.rs` | 472 | 378 | 19.9% | 354 |
|
||||
| `src/worker/runtime.rs` | 350 | 330 | 5.7% | 312 |
|
||||
| `src/worker/container.rs` | 350 | 330 | 5.7% | 312 |
|
||||
| `src/tools/builtin/job.rs` | 1,014 | 359 | 64.6% | 308 |
|
||||
| `src/cli/mcp.rs` | 322 | 319 | 0.9% | 302 |
|
||||
| `src/cli/oauth_defaults.rs` | 730 | 335 | 54.1% | 298 |
|
||||
@@ -346,7 +346,7 @@ Test slash commands through the agent loop.
|
||||
|
||||
### Trace: Worker Multi-Turn Execution
|
||||
|
||||
**Covers:** `agent/worker.rs` (+413 lines), `agent/agent_loop.rs` (+207 lines)
|
||||
**Covers:** `worker/job.rs` (+413 lines), `agent/agent_loop.rs` (+207 lines)
|
||||
|
||||
Test multi-turn tool calling, error recovery, and completion flows.
|
||||
|
||||
@@ -769,7 +769,7 @@ HTTP proxy for container network access.
|
||||
- `test_proxy_connect_tunnel` -- HTTPS CONNECT method handling
|
||||
- `test_proxy_logging` -- request/response logging
|
||||
|
||||
### `src/worker/runtime.rs` -- 5.7% -> 95% (+312 lines)
|
||||
### `src/worker/container.rs` -- 5.7% -> 95% (+312 lines)
|
||||
|
||||
Worker execution loop (runs inside containers).
|
||||
|
||||
|
||||
+2
-2
@@ -46,7 +46,7 @@ This document tracks feature parity between IronClaw (Rust implementation) and O
|
||||
| Bonjour/mDNS discovery | ✅ | ❌ | |
|
||||
| Tailscale integration | ✅ | ❌ | |
|
||||
| Health check endpoints | ✅ | ✅ | /api/health + /api/gateway/status + /healthz + /readyz, with channel-backed readiness probes |
|
||||
| `doctor` diagnostics | ✅ | ❌ | |
|
||||
| `doctor` diagnostics | ✅ | 🚧 | 16 checks: settings, LLM, DB, embeddings, routines, gateway, MCP, skills, secrets, service, Docker daemon, tunnel binaries |
|
||||
| Agent event broadcast | ✅ | 🚧 | SSE broadcast manager exists (SseManager) but tool/job-state events not fully wired |
|
||||
| Channel health monitor | ✅ | ❌ | Auto-restart with configurable interval |
|
||||
| Presence system | ✅ | ❌ | Beacons on connect, system presence for agents |
|
||||
@@ -175,7 +175,7 @@ This document tracks feature parity between IronClaw (Rust implementation) and O
|
||||
| `message send` | ✅ | ❌ | P2 | Send to channels |
|
||||
| `browser` | ✅ | ❌ | P3 | Browser automation |
|
||||
| `sandbox` | ✅ | ✅ | - | WASM sandbox |
|
||||
| `doctor` | ✅ | ❌ | P2 | Diagnostics |
|
||||
| `doctor` | ✅ | 🚧 | P2 | 16 subsystem checks |
|
||||
| `logs` | ✅ | ❌ | P3 | Query logs |
|
||||
| `update` | ✅ | ❌ | P3 | Self-update |
|
||||
| `completion` | ✅ | ✅ | - | Shell completion |
|
||||
|
||||
+19
-16
@@ -14,7 +14,8 @@ Core agent logic. This is the most complex subsystem — read this before workin
|
||||
| `session_manager.rs` | Lifecycle: create/lookup sessions, map external thread IDs to internal UUIDs, prune stale sessions, manage undo managers. |
|
||||
| `router.rs` | Routes explicit `/commands` to `MessageIntent`. Natural language bypasses the router entirely. |
|
||||
| `scheduler.rs` | Parallel job scheduling. Maintains `jobs` map (full LLM-driven) and `subtasks` map (tool-exec/background). |
|
||||
| `worker.rs` | Per-job execution for background scheduler jobs: calls LLM, runs tools, handles the reasoning loop. Distinct from `dispatcher.rs`. |
|
||||
| *(moved to `src/worker/job.rs`)* | Per-job execution now lives in `src/worker/job.rs` as `JobDelegate`, using the shared `run_agentic_loop()` engine. |
|
||||
| `agentic_loop.rs` | Shared agentic loop engine: `run_agentic_loop()`, `LoopDelegate` trait, `LoopOutcome`, `LoopSignal`, `TextAction`. All three execution paths (chat, job, container) delegate to this. |
|
||||
| `compaction.rs` | Context window management: summarize old turns, write to workspace daily log, trim context. Three strategies. |
|
||||
| `context_monitor.rs` | Detects memory pressure. Suggests `CompactionStrategy` based on usage level. |
|
||||
| `self_repair.rs` | Detects stuck jobs and broken tools, attempts recovery. |
|
||||
@@ -49,26 +50,28 @@ Session (per user)
|
||||
|
||||
## Agentic Loop (dispatcher.rs)
|
||||
|
||||
The `dispatcher.rs` module handles **direct conversational turns** (user messages processed inline by the main agent). Background scheduler jobs use `worker.rs` instead — these are two separate execution paths.
|
||||
All three execution paths (chat, job, container) now use the shared `run_agentic_loop()` engine in `agentic_loop.rs`, each providing their own `LoopDelegate` implementation:
|
||||
|
||||
- **`ChatDelegate`** (`dispatcher.rs`) — conversational turns, tool approval, skill context injection
|
||||
- **`JobDelegate`** (`src/worker/job.rs`) — background scheduler jobs, planning support, completion detection
|
||||
- **`ContainerDelegate`** (`src/worker/container.rs`) — Docker container worker, sequential tool exec, HTTP event streaming
|
||||
|
||||
```
|
||||
run_agentic_loop() [dispatcher.rs — conversational turns]
|
||||
1. Load workspace system prompt (identity files: AGENTS.md, SOUL.md, etc.)
|
||||
2. Detect group chat from metadata; exclude MEMORY.md if group chat
|
||||
3. Select active skills (keyword/pattern scoring against message content)
|
||||
4. Build skill context block (injected before user message)
|
||||
5. LLM call → text response OR tool calls
|
||||
6. If tool calls:
|
||||
a. Check tool approval (session auto-approvals, pending approval queue)
|
||||
b. Execute tools (parallel via JoinSet)
|
||||
c. Sanitize results through SafetyLayer
|
||||
d. Feed results back → goto 5
|
||||
7. Return AgenticLoopResult::Response or NeedApproval
|
||||
run_agentic_loop(delegate, reasoning, reason_ctx, config)
|
||||
1. Check signals (stop/cancel) via delegate.check_signals()
|
||||
2. Pre-LLM hook via delegate.before_llm_call()
|
||||
3. LLM call via delegate.call_llm()
|
||||
4. If text response → delegate.handle_text_response() → Continue or Return
|
||||
5. If tool calls → delegate.execute_tool_calls() → Continue or Return
|
||||
6. Post-iteration hook via delegate.after_iteration()
|
||||
7. Repeat until LoopOutcome returned or max_iterations reached
|
||||
```
|
||||
|
||||
**Tool approval:** Tools flagged `requires_approval` pause the loop and return `NeedApproval`. The web gateway stores the `PendingApproval` in session state and sends an `approval_needed` SSE event. The user's approval/deny resumes the loop.
|
||||
**Tool approval:** Tools flagged `requires_approval` pause the loop — `ChatDelegate` returns `LoopOutcome::NeedApproval(pending)`. The web gateway stores the `PendingApproval` in session state and sends an `approval_needed` SSE event. The user's approval/deny resumes the loop.
|
||||
|
||||
**worker.rs vs dispatcher.rs:** `dispatcher.rs` runs the agentic loop for user-initiated conversational turns (holds session lock, tracks turns). `worker.rs` is spawned by the `Scheduler` for background jobs created via `CreateJob` / `/job` — it runs independently of the session and has its own LLM reasoning loop with planning support (`use_planning` flag).
|
||||
**Shared tool execution:** `tools/execute.rs` provides `execute_tool_with_safety()` (validate → timeout → execute → serialize) and `process_tool_result()` (sanitize → wrap → ChatMessage), used by all three delegates.
|
||||
|
||||
**ChatDelegate vs JobDelegate:** `ChatDelegate` runs for user-initiated conversational turns (holds session lock, tracks turns). `JobDelegate` is spawned by the `Scheduler` for background jobs created via `CreateJob` / `/job` — it runs independently of the session and has planning support (`use_planning` flag).
|
||||
|
||||
## Command Routing (router.rs)
|
||||
|
||||
|
||||
@@ -0,0 +1,587 @@
|
||||
//! Unified agentic loop engine.
|
||||
//!
|
||||
//! Provides a single implementation of the core LLM call → tool execution →
|
||||
//! result processing → context update → repeat cycle. Three consumers
|
||||
//! (chat dispatcher, job worker, container runtime) customize behavior
|
||||
//! via the `LoopDelegate` trait.
|
||||
|
||||
use async_trait::async_trait;
|
||||
|
||||
use crate::agent::session::PendingApproval;
|
||||
use crate::error::Error;
|
||||
use crate::llm::{ChatMessage, Reasoning, ReasoningContext, RespondResult};
|
||||
|
||||
/// Signal from the delegate indicating how the loop should proceed.
|
||||
pub enum LoopSignal {
|
||||
/// Continue normally.
|
||||
Continue,
|
||||
/// Stop the loop gracefully.
|
||||
Stop,
|
||||
/// Inject a user message into context and continue.
|
||||
InjectMessage(String),
|
||||
}
|
||||
|
||||
/// Outcome of a text response from the LLM.
|
||||
pub enum TextAction {
|
||||
/// Return this as the final loop result.
|
||||
Return(LoopOutcome),
|
||||
/// Continue the loop (text was handled but loop should proceed).
|
||||
Continue,
|
||||
}
|
||||
|
||||
/// Final outcome of the agentic loop.
|
||||
pub enum LoopOutcome {
|
||||
/// Completed with a text response.
|
||||
Response(String),
|
||||
/// Loop was stopped by a signal.
|
||||
Stopped,
|
||||
/// Max iterations exceeded.
|
||||
MaxIterations,
|
||||
/// A tool requires user approval before continuing (chat delegate only).
|
||||
NeedApproval(Box<PendingApproval>),
|
||||
}
|
||||
|
||||
/// Configuration for the agentic loop.
|
||||
pub struct AgenticLoopConfig {
|
||||
pub max_iterations: usize,
|
||||
pub enable_tool_intent_nudge: bool,
|
||||
pub max_tool_intent_nudges: u32,
|
||||
}
|
||||
|
||||
impl Default for AgenticLoopConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
max_iterations: 50,
|
||||
enable_tool_intent_nudge: true,
|
||||
max_tool_intent_nudges: 2,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Strategy trait — each consumer implements this to customize I/O and lifecycle.
|
||||
///
|
||||
/// The shared loop calls these methods at well-defined points. Consumers
|
||||
/// implement only the behavior that differs between chat, job, and container
|
||||
/// contexts. The loop itself handles the common logic: tool intent nudge,
|
||||
/// iteration counting, tool definition refresh, and the respond → execute → process cycle.
|
||||
///
|
||||
/// # `Send + Sync` requirement
|
||||
///
|
||||
/// This trait requires `Send + Sync` because the loop accepts `&dyn LoopDelegate`.
|
||||
/// Delegates using borrowed references (e.g. `ChatDelegate<'a>`) must ensure all
|
||||
/// borrowed fields are `Send + Sync`. This is a load-bearing constraint: if a
|
||||
/// delegate needs to be spawned into a detached task, it must use `Arc`-based
|
||||
/// ownership instead of borrows (as `JobDelegate` and `ContainerDelegate` do).
|
||||
#[async_trait]
|
||||
pub trait LoopDelegate: Send + Sync {
|
||||
/// Called at the start of each iteration. Check for external signals
|
||||
/// (cancellation, user messages, stop requests).
|
||||
async fn check_signals(&self) -> LoopSignal;
|
||||
|
||||
/// Called before the LLM call. Allows the delegate to refresh tool
|
||||
/// definitions, enforce cost guards, or inject messages.
|
||||
/// Return `Some(outcome)` to break the loop early.
|
||||
async fn before_llm_call(
|
||||
&self,
|
||||
reason_ctx: &mut ReasoningContext,
|
||||
iteration: usize,
|
||||
) -> Option<LoopOutcome>;
|
||||
|
||||
/// Call the LLM and return the result. Delegates own the LLM call
|
||||
/// to handle consumer-specific concerns (rate limiting, auto-compaction,
|
||||
/// cost tracking, force_text mode).
|
||||
async fn call_llm(
|
||||
&self,
|
||||
reasoning: &Reasoning,
|
||||
reason_ctx: &mut ReasoningContext,
|
||||
iteration: usize,
|
||||
) -> Result<crate::llm::RespondOutput, Error>;
|
||||
|
||||
/// Handle a text-only response from the LLM.
|
||||
/// Return `TextAction::Return` to exit the loop, `TextAction::Continue` to proceed.
|
||||
async fn handle_text_response(
|
||||
&self,
|
||||
text: &str,
|
||||
reason_ctx: &mut ReasoningContext,
|
||||
) -> TextAction;
|
||||
|
||||
/// Execute tool calls and add results to context.
|
||||
/// Return `Some(outcome)` to break the loop (e.g. approval needed).
|
||||
async fn execute_tool_calls(
|
||||
&self,
|
||||
tool_calls: Vec<crate::llm::ToolCall>,
|
||||
content: Option<String>,
|
||||
reason_ctx: &mut ReasoningContext,
|
||||
) -> Result<Option<LoopOutcome>, Error>;
|
||||
|
||||
/// Called when the LLM expresses tool intent without actually calling a tool.
|
||||
/// Delegates can use this to emit events or log the nudge for observability.
|
||||
async fn on_tool_intent_nudge(&self, _text: &str, _reason_ctx: &mut ReasoningContext) {}
|
||||
|
||||
/// Called after each successful iteration (no error, no early return).
|
||||
async fn after_iteration(&self, _iteration: usize) {}
|
||||
}
|
||||
|
||||
/// Run the unified agentic loop.
|
||||
///
|
||||
/// This is the single implementation used by all three consumers (chat, job, container).
|
||||
/// The `delegate` provides consumer-specific behavior via the `LoopDelegate` trait.
|
||||
pub async fn run_agentic_loop(
|
||||
delegate: &dyn LoopDelegate,
|
||||
reasoning: &Reasoning,
|
||||
reason_ctx: &mut ReasoningContext,
|
||||
config: &AgenticLoopConfig,
|
||||
) -> Result<LoopOutcome, Error> {
|
||||
let mut consecutive_tool_intent_nudges: u32 = 0;
|
||||
|
||||
for iteration in 1..=config.max_iterations {
|
||||
// Check for external signals (stop, cancellation, user messages)
|
||||
match delegate.check_signals().await {
|
||||
LoopSignal::Continue => {}
|
||||
LoopSignal::Stop => return Ok(LoopOutcome::Stopped),
|
||||
LoopSignal::InjectMessage(msg) => {
|
||||
reason_ctx.messages.push(ChatMessage::user(&msg));
|
||||
}
|
||||
}
|
||||
|
||||
// Pre-LLM call hook (cost guard, tool refresh, iteration limit nudge)
|
||||
if let Some(outcome) = delegate.before_llm_call(reason_ctx, iteration).await {
|
||||
return Ok(outcome);
|
||||
}
|
||||
|
||||
// Call LLM
|
||||
let output = delegate.call_llm(reasoning, reason_ctx, iteration).await?;
|
||||
|
||||
match output.result {
|
||||
RespondResult::Text(text) => {
|
||||
// Tool intent nudge: if the LLM says "let me search..." without
|
||||
// actually calling a tool, inject a nudge message.
|
||||
if config.enable_tool_intent_nudge
|
||||
&& !reason_ctx.available_tools.is_empty()
|
||||
&& !reason_ctx.force_text
|
||||
&& consecutive_tool_intent_nudges < config.max_tool_intent_nudges
|
||||
&& crate::llm::llm_signals_tool_intent(&text)
|
||||
{
|
||||
consecutive_tool_intent_nudges += 1;
|
||||
tracing::info!(
|
||||
iteration,
|
||||
"LLM expressed tool intent without calling a tool, nudging"
|
||||
);
|
||||
delegate.on_tool_intent_nudge(&text, reason_ctx).await;
|
||||
reason_ctx.messages.push(ChatMessage::assistant(&text));
|
||||
reason_ctx
|
||||
.messages
|
||||
.push(ChatMessage::user(crate::llm::TOOL_INTENT_NUDGE));
|
||||
delegate.after_iteration(iteration).await;
|
||||
continue;
|
||||
}
|
||||
|
||||
// Reset nudge counter since we got a non-intent text response
|
||||
if !crate::llm::llm_signals_tool_intent(&text) {
|
||||
consecutive_tool_intent_nudges = 0;
|
||||
}
|
||||
|
||||
match delegate.handle_text_response(&text, reason_ctx).await {
|
||||
TextAction::Return(outcome) => return Ok(outcome),
|
||||
TextAction::Continue => {}
|
||||
}
|
||||
}
|
||||
RespondResult::ToolCalls {
|
||||
tool_calls,
|
||||
content,
|
||||
} => {
|
||||
consecutive_tool_intent_nudges = 0;
|
||||
|
||||
if let Some(outcome) = delegate
|
||||
.execute_tool_calls(tool_calls, content, reason_ctx)
|
||||
.await?
|
||||
{
|
||||
return Ok(outcome);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
delegate.after_iteration(iteration).await;
|
||||
}
|
||||
|
||||
Ok(LoopOutcome::MaxIterations)
|
||||
}
|
||||
|
||||
/// Truncate a string for log/status previews.
|
||||
///
|
||||
/// `max` is a byte budget. The result is truncated at the last valid char
|
||||
/// boundary at or before `max` bytes, so it is always valid UTF-8.
|
||||
pub fn truncate_for_preview(s: &str, max: usize) -> String {
|
||||
if s.len() <= max {
|
||||
s.to_string()
|
||||
} else {
|
||||
let end = crate::util::floor_char_boundary(s, max);
|
||||
format!("{}...", &s[..end])
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::llm::{RespondOutput, TokenUsage, ToolCall};
|
||||
use crate::testing::StubLlm;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
fn stub_reasoning() -> Reasoning {
|
||||
Reasoning::new(Arc::new(StubLlm::default()))
|
||||
}
|
||||
|
||||
fn zero_usage() -> TokenUsage {
|
||||
TokenUsage {
|
||||
input_tokens: 0,
|
||||
output_tokens: 0,
|
||||
cache_read_input_tokens: 0,
|
||||
cache_creation_input_tokens: 0,
|
||||
}
|
||||
}
|
||||
|
||||
fn text_output(text: &str) -> RespondOutput {
|
||||
RespondOutput {
|
||||
result: RespondResult::Text(text.to_string()),
|
||||
usage: zero_usage(),
|
||||
}
|
||||
}
|
||||
|
||||
fn tool_calls_output(calls: Vec<ToolCall>) -> RespondOutput {
|
||||
RespondOutput {
|
||||
result: RespondResult::ToolCalls {
|
||||
tool_calls: calls,
|
||||
content: None,
|
||||
},
|
||||
usage: zero_usage(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Configurable mock delegate for testing run_agentic_loop.
|
||||
struct MockDelegate {
|
||||
signal: Mutex<LoopSignal>,
|
||||
llm_responses: Mutex<Vec<RespondOutput>>,
|
||||
tool_exec_count: AtomicUsize,
|
||||
tool_exec_outcome: Mutex<Option<LoopOutcome>>,
|
||||
iterations_seen: Mutex<Vec<usize>>,
|
||||
early_exit: Mutex<Option<(usize, LoopOutcome)>>,
|
||||
nudge_count: AtomicUsize,
|
||||
}
|
||||
|
||||
impl MockDelegate {
|
||||
fn new(responses: Vec<RespondOutput>) -> Self {
|
||||
Self {
|
||||
signal: Mutex::new(LoopSignal::Continue),
|
||||
llm_responses: Mutex::new(responses),
|
||||
tool_exec_count: AtomicUsize::new(0),
|
||||
tool_exec_outcome: Mutex::new(None),
|
||||
iterations_seen: Mutex::new(Vec::new()),
|
||||
early_exit: Mutex::new(None),
|
||||
nudge_count: AtomicUsize::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
fn with_signal(mut self, signal: LoopSignal) -> Self {
|
||||
self.signal = Mutex::new(signal);
|
||||
self
|
||||
}
|
||||
|
||||
fn with_early_exit(mut self, iteration: usize, outcome: LoopOutcome) -> Self {
|
||||
self.early_exit = Mutex::new(Some((iteration, outcome)));
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl LoopDelegate for MockDelegate {
|
||||
async fn check_signals(&self) -> LoopSignal {
|
||||
let mut sig = self.signal.lock().await;
|
||||
std::mem::replace(&mut *sig, LoopSignal::Continue)
|
||||
}
|
||||
|
||||
async fn before_llm_call(
|
||||
&self,
|
||||
_reason_ctx: &mut ReasoningContext,
|
||||
iteration: usize,
|
||||
) -> Option<LoopOutcome> {
|
||||
let mut guard = self.early_exit.lock().await;
|
||||
let should_take = guard
|
||||
.as_ref()
|
||||
.is_some_and(|(target, _)| *target == iteration);
|
||||
if should_take {
|
||||
guard.take().map(|(_, o)| o)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
async fn call_llm(
|
||||
&self,
|
||||
_reasoning: &Reasoning,
|
||||
_reason_ctx: &mut ReasoningContext,
|
||||
_iteration: usize,
|
||||
) -> Result<crate::llm::RespondOutput, crate::error::Error> {
|
||||
let mut responses = self.llm_responses.lock().await;
|
||||
if responses.is_empty() {
|
||||
panic!("MockDelegate: no more LLM responses queued");
|
||||
}
|
||||
Ok(responses.remove(0))
|
||||
}
|
||||
|
||||
async fn handle_text_response(
|
||||
&self,
|
||||
text: &str,
|
||||
_reason_ctx: &mut ReasoningContext,
|
||||
) -> TextAction {
|
||||
TextAction::Return(LoopOutcome::Response(text.to_string()))
|
||||
}
|
||||
|
||||
async fn execute_tool_calls(
|
||||
&self,
|
||||
_tool_calls: Vec<ToolCall>,
|
||||
_content: Option<String>,
|
||||
reason_ctx: &mut ReasoningContext,
|
||||
) -> Result<Option<LoopOutcome>, crate::error::Error> {
|
||||
self.tool_exec_count.fetch_add(1, Ordering::SeqCst);
|
||||
reason_ctx
|
||||
.messages
|
||||
.push(ChatMessage::user("tool result stub"));
|
||||
let outcome = self.tool_exec_outcome.lock().await.take();
|
||||
Ok(outcome)
|
||||
}
|
||||
|
||||
async fn on_tool_intent_nudge(&self, _text: &str, _reason_ctx: &mut ReasoningContext) {
|
||||
self.nudge_count.fetch_add(1, Ordering::SeqCst);
|
||||
}
|
||||
|
||||
async fn after_iteration(&self, iteration: usize) {
|
||||
self.iterations_seen.lock().await.push(iteration);
|
||||
}
|
||||
}
|
||||
|
||||
// --- Tests ---
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_text_response_returns_immediately() {
|
||||
let delegate = MockDelegate::new(vec![text_output("Hello, world!")]);
|
||||
let reasoning = stub_reasoning();
|
||||
let mut ctx = ReasoningContext::new();
|
||||
let config = AgenticLoopConfig::default();
|
||||
|
||||
let outcome = run_agentic_loop(&delegate, &reasoning, &mut ctx, &config)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
match outcome {
|
||||
LoopOutcome::Response(text) => assert_eq!(text, "Hello, world!"),
|
||||
_ => panic!("Expected LoopOutcome::Response"),
|
||||
}
|
||||
// after_iteration is NOT called when handle_text_response returns Return
|
||||
// (the loop exits before reaching after_iteration).
|
||||
assert!(delegate.iterations_seen.lock().await.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_tool_call_then_text_response() {
|
||||
let tool_call = ToolCall {
|
||||
id: "call_1".to_string(),
|
||||
name: "echo".to_string(),
|
||||
arguments: serde_json::json!({}),
|
||||
};
|
||||
let delegate = MockDelegate::new(vec![
|
||||
tool_calls_output(vec![tool_call]),
|
||||
text_output("Done!"),
|
||||
]);
|
||||
let reasoning = stub_reasoning();
|
||||
let mut ctx = ReasoningContext::new();
|
||||
let config = AgenticLoopConfig::default();
|
||||
|
||||
let outcome = run_agentic_loop(&delegate, &reasoning, &mut ctx, &config)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
match outcome {
|
||||
LoopOutcome::Response(text) => assert_eq!(text, "Done!"),
|
||||
_ => panic!("Expected LoopOutcome::Response"),
|
||||
}
|
||||
assert_eq!(delegate.tool_exec_count.load(Ordering::SeqCst), 1);
|
||||
// after_iteration called for iteration 1 (tool call), but not 2
|
||||
// (text response exits before after_iteration).
|
||||
assert_eq!(*delegate.iterations_seen.lock().await, vec![1]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_stop_signal_exits_immediately() {
|
||||
let delegate =
|
||||
MockDelegate::new(vec![text_output("unreachable")]).with_signal(LoopSignal::Stop);
|
||||
let reasoning = stub_reasoning();
|
||||
let mut ctx = ReasoningContext::new();
|
||||
let config = AgenticLoopConfig::default();
|
||||
|
||||
let outcome = run_agentic_loop(&delegate, &reasoning, &mut ctx, &config)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(matches!(outcome, LoopOutcome::Stopped));
|
||||
assert!(delegate.iterations_seen.lock().await.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_inject_message_adds_user_message() {
|
||||
let delegate = MockDelegate::new(vec![text_output("Got it")])
|
||||
.with_signal(LoopSignal::InjectMessage("injected prompt".to_string()));
|
||||
let reasoning = stub_reasoning();
|
||||
let mut ctx = ReasoningContext::new();
|
||||
let config = AgenticLoopConfig::default();
|
||||
|
||||
let outcome = run_agentic_loop(&delegate, &reasoning, &mut ctx, &config)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(matches!(outcome, LoopOutcome::Response(_)));
|
||||
assert!(
|
||||
ctx.messages
|
||||
.iter()
|
||||
.any(|m| m.role == crate::llm::Role::User && m.content.contains("injected prompt")),
|
||||
"Injected message should appear in context"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_max_iterations_reached() {
|
||||
struct ContinueDelegate;
|
||||
|
||||
#[async_trait]
|
||||
impl LoopDelegate for ContinueDelegate {
|
||||
async fn check_signals(&self) -> LoopSignal {
|
||||
LoopSignal::Continue
|
||||
}
|
||||
async fn before_llm_call(
|
||||
&self,
|
||||
_: &mut ReasoningContext,
|
||||
_: usize,
|
||||
) -> Option<LoopOutcome> {
|
||||
None
|
||||
}
|
||||
async fn call_llm(
|
||||
&self,
|
||||
_: &Reasoning,
|
||||
_: &mut ReasoningContext,
|
||||
_: usize,
|
||||
) -> Result<crate::llm::RespondOutput, crate::error::Error> {
|
||||
Ok(text_output("still working"))
|
||||
}
|
||||
async fn handle_text_response(
|
||||
&self,
|
||||
_: &str,
|
||||
ctx: &mut ReasoningContext,
|
||||
) -> TextAction {
|
||||
ctx.messages.push(ChatMessage::assistant("still working"));
|
||||
TextAction::Continue
|
||||
}
|
||||
async fn execute_tool_calls(
|
||||
&self,
|
||||
_: Vec<ToolCall>,
|
||||
_: Option<String>,
|
||||
_: &mut ReasoningContext,
|
||||
) -> Result<Option<LoopOutcome>, crate::error::Error> {
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
|
||||
let delegate = ContinueDelegate;
|
||||
let reasoning = stub_reasoning();
|
||||
let mut ctx = ReasoningContext::new();
|
||||
let config = AgenticLoopConfig {
|
||||
max_iterations: 3,
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let outcome = run_agentic_loop(&delegate, &reasoning, &mut ctx, &config)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(matches!(outcome, LoopOutcome::MaxIterations));
|
||||
let assistant_count = ctx
|
||||
.messages
|
||||
.iter()
|
||||
.filter(|m| m.role == crate::llm::Role::Assistant)
|
||||
.count();
|
||||
assert_eq!(assistant_count, 3);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_tool_intent_nudge_fires_and_caps() {
|
||||
let delegate = MockDelegate::new(vec![
|
||||
text_output("Let me search for that file"),
|
||||
text_output("Let me search for that file"),
|
||||
text_output("Let me search for that file"),
|
||||
]);
|
||||
let reasoning = stub_reasoning();
|
||||
let mut ctx = ReasoningContext::new();
|
||||
ctx.available_tools.push(crate::llm::ToolDefinition {
|
||||
name: "search".to_string(),
|
||||
description: "Search files".to_string(),
|
||||
parameters: serde_json::json!({"type": "object"}),
|
||||
});
|
||||
let config = AgenticLoopConfig {
|
||||
max_iterations: 10,
|
||||
enable_tool_intent_nudge: true,
|
||||
max_tool_intent_nudges: 2,
|
||||
};
|
||||
|
||||
let outcome = run_agentic_loop(&delegate, &reasoning, &mut ctx, &config)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(matches!(outcome, LoopOutcome::Response(_)));
|
||||
assert_eq!(delegate.nudge_count.load(Ordering::SeqCst), 2);
|
||||
let nudge_messages = ctx
|
||||
.messages
|
||||
.iter()
|
||||
.filter(|m| {
|
||||
m.role == crate::llm::Role::User
|
||||
&& m.content.contains("you did not include any tool calls")
|
||||
})
|
||||
.count();
|
||||
assert_eq!(
|
||||
nudge_messages, 2,
|
||||
"Should have exactly 2 nudge messages in context"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_before_llm_call_early_exit() {
|
||||
let delegate = MockDelegate::new(vec![text_output("unreachable")])
|
||||
.with_early_exit(1, LoopOutcome::Stopped);
|
||||
let reasoning = stub_reasoning();
|
||||
let mut ctx = ReasoningContext::new();
|
||||
let config = AgenticLoopConfig::default();
|
||||
|
||||
let outcome = run_agentic_loop(&delegate, &reasoning, &mut ctx, &config)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(matches!(outcome, LoopOutcome::Stopped));
|
||||
assert!(delegate.iterations_seen.lock().await.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_truncate_short_string_unchanged() {
|
||||
assert_eq!(truncate_for_preview("hello", 10), "hello");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_truncate_long_string_adds_ellipsis() {
|
||||
let result = truncate_for_preview("hello world", 5);
|
||||
assert_eq!(result, "hello...");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_truncate_multibyte_safe() {
|
||||
let result = truncate_for_preview("café", 4);
|
||||
assert_eq!(result, "caf...");
|
||||
}
|
||||
}
|
||||
+710
-803
File diff suppressed because it is too large
Load Diff
+3
-3
@@ -11,6 +11,7 @@
|
||||
//! - Context compaction for long conversations
|
||||
|
||||
mod agent_loop;
|
||||
pub mod agentic_loop;
|
||||
mod attachments;
|
||||
mod commands;
|
||||
pub mod compaction;
|
||||
@@ -22,7 +23,7 @@ pub mod job_monitor;
|
||||
mod router;
|
||||
pub mod routine;
|
||||
pub mod routine_engine;
|
||||
mod scheduler;
|
||||
pub(crate) mod scheduler;
|
||||
mod self_repair;
|
||||
pub mod session;
|
||||
mod session_manager;
|
||||
@@ -30,8 +31,8 @@ pub mod submission;
|
||||
pub mod task;
|
||||
mod thread_ops;
|
||||
pub mod undo;
|
||||
pub mod worker;
|
||||
|
||||
pub use crate::worker::{Worker, WorkerDeps};
|
||||
pub(crate) use agent_loop::truncate_for_preview;
|
||||
pub use agent_loop::{Agent, AgentDeps};
|
||||
pub use compaction::{CompactionResult, ContextCompactor};
|
||||
@@ -47,4 +48,3 @@ pub use session_manager::SessionManager;
|
||||
pub use submission::{Submission, SubmissionParser, SubmissionResult};
|
||||
pub use task::{Task, TaskContext, TaskHandler, TaskOutput};
|
||||
pub use undo::{Checkpoint, UndoManager};
|
||||
pub use worker::{Worker, WorkerDeps};
|
||||
|
||||
+21
-35
@@ -9,7 +9,6 @@ use tokio::task::JoinHandle;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::agent::task::{Task, TaskContext, TaskOutput};
|
||||
use crate::agent::worker::{Worker, WorkerDeps};
|
||||
use crate::channels::web::types::SseEvent;
|
||||
use crate::config::AgentConfig;
|
||||
use crate::context::{ContextManager, JobContext, JobState};
|
||||
@@ -19,6 +18,7 @@ use crate::hooks::HookRegistry;
|
||||
use crate::llm::LlmProvider;
|
||||
use crate::safety::SafetyLayer;
|
||||
use crate::tools::{ApprovalContext, ToolRegistry};
|
||||
use crate::worker::job::{Worker, WorkerDeps};
|
||||
|
||||
/// Message to send to a worker.
|
||||
#[derive(Debug)]
|
||||
@@ -462,6 +462,9 @@ impl Scheduler {
|
||||
}
|
||||
|
||||
/// Execute a single tool as a subtask.
|
||||
///
|
||||
/// Performs scheduler-specific checks (approval, cancellation) then
|
||||
/// delegates to the shared `execute_tool_with_safety` pipeline.
|
||||
async fn execute_tool_task(
|
||||
tools: Arc<ToolRegistry>,
|
||||
context_manager: Arc<ContextManager>,
|
||||
@@ -473,7 +476,7 @@ impl Scheduler {
|
||||
) -> Result<TaskOutput, Error> {
|
||||
let start = std::time::Instant::now();
|
||||
|
||||
// Get the tool
|
||||
// Get the tool for approval check
|
||||
let tool = tools.get(tool_name).await.ok_or_else(|| {
|
||||
Error::Tool(crate::error::ToolError::NotFound {
|
||||
name: tool_name.to_string(),
|
||||
@@ -490,6 +493,7 @@ impl Scheduler {
|
||||
.into());
|
||||
}
|
||||
|
||||
// Scheduler-specific approval check
|
||||
let requirement = tool.requires_approval(¶ms);
|
||||
let blocked =
|
||||
ApprovalContext::is_blocked_or_default(&approval_context, tool_name, requirement);
|
||||
@@ -500,41 +504,23 @@ impl Scheduler {
|
||||
.into());
|
||||
}
|
||||
|
||||
// Validate tool parameters
|
||||
let validation = safety.validator().validate_tool_params(¶ms);
|
||||
if !validation.is_valid {
|
||||
let details = validation
|
||||
.errors
|
||||
.iter()
|
||||
.map(|e| format!("{}: {}", e.field, e.message))
|
||||
.collect::<Vec<_>>()
|
||||
.join("; ");
|
||||
return Err(crate::error::ToolError::InvalidParameters {
|
||||
// Delegate to shared tool execution pipeline
|
||||
let output_str = crate::tools::execute::execute_tool_with_safety(
|
||||
&tools, &safety, tool_name, ¶ms, &job_ctx,
|
||||
)
|
||||
.await?;
|
||||
|
||||
// Parse back to Value for TaskOutput; this should be infallible given
|
||||
// `execute_tool_with_safety` uses `serde_json::to_string_pretty`, but if it
|
||||
// ever fails we surface a clear error instead of silently changing types.
|
||||
let result_value: serde_json::Value = serde_json::from_str(&output_str).map_err(|e| {
|
||||
Error::Tool(crate::error::ToolError::ExecutionFailed {
|
||||
name: tool_name.to_string(),
|
||||
reason: format!("Invalid tool parameters: {}", details),
|
||||
}
|
||||
.into());
|
||||
}
|
||||
reason: format!("Failed to parse tool output as JSON: {}", e),
|
||||
})
|
||||
})?;
|
||||
|
||||
// Execute with per-tool timeout
|
||||
let tool_timeout = tool.execution_timeout();
|
||||
let result =
|
||||
tokio::time::timeout(tool_timeout, async { tool.execute(params, &job_ctx).await })
|
||||
.await
|
||||
.map_err(|_| {
|
||||
Error::Tool(crate::error::ToolError::Timeout {
|
||||
name: tool_name.to_string(),
|
||||
timeout: tool_timeout,
|
||||
})
|
||||
})?
|
||||
.map_err(|e| {
|
||||
Error::Tool(crate::error::ToolError::ExecutionFailed {
|
||||
name: tool_name.to_string(),
|
||||
reason: e.to_string(),
|
||||
})
|
||||
})?;
|
||||
|
||||
Ok(TaskOutput::new(result.result, start.elapsed()))
|
||||
Ok(TaskOutput::new(result_value, start.elapsed()))
|
||||
}
|
||||
|
||||
/// Stop a running job.
|
||||
|
||||
+12
-24
@@ -852,19 +852,12 @@ impl Agent {
|
||||
// Sanitize tool result, then record the cleaned version in the
|
||||
// thread. Must happen before auth intercept check which may return early.
|
||||
let is_tool_error = tool_result.is_err();
|
||||
let result_content = match &tool_result {
|
||||
Ok(output) => {
|
||||
let sanitized = self
|
||||
.safety()
|
||||
.sanitize_tool_output(&pending.tool_name, output);
|
||||
self.safety().wrap_for_llm(
|
||||
&pending.tool_name,
|
||||
&sanitized.content,
|
||||
sanitized.was_modified,
|
||||
)
|
||||
}
|
||||
Err(e) => format!("Error: {}", e),
|
||||
};
|
||||
let (result_content, _) = crate::tools::execute::process_tool_result(
|
||||
self.safety(),
|
||||
&pending.tool_name,
|
||||
&pending.tool_call_id,
|
||||
&tool_result,
|
||||
);
|
||||
|
||||
// Record sanitized result in thread
|
||||
{
|
||||
@@ -1104,17 +1097,12 @@ impl Agent {
|
||||
// Sanitize first, then record the cleaned version in thread.
|
||||
// Must happen before auth detection which may set deferred_auth.
|
||||
let is_deferred_error = deferred_result.is_err();
|
||||
let deferred_content = match &deferred_result {
|
||||
Ok(output) => {
|
||||
let sanitized = self.safety().sanitize_tool_output(&tc.name, output);
|
||||
self.safety().wrap_for_llm(
|
||||
&tc.name,
|
||||
&sanitized.content,
|
||||
sanitized.was_modified,
|
||||
)
|
||||
}
|
||||
Err(e) => format!("Error: {}", e),
|
||||
};
|
||||
let (deferred_content, _) = crate::tools::execute::process_tool_result(
|
||||
self.safety(),
|
||||
&tc.name,
|
||||
&tc.id,
|
||||
&deferred_result,
|
||||
);
|
||||
|
||||
// Record sanitized result in thread
|
||||
{
|
||||
|
||||
+541
-5
@@ -7,6 +7,7 @@
|
||||
use std::path::PathBuf;
|
||||
|
||||
use crate::bootstrap::ironclaw_base_dir;
|
||||
use crate::settings::Settings;
|
||||
|
||||
/// Run all diagnostic checks and print results.
|
||||
pub async fn run_doctor_command() -> anyhow::Result<()> {
|
||||
@@ -15,14 +16,35 @@ pub async fn run_doctor_command() -> anyhow::Result<()> {
|
||||
|
||||
let mut passed = 0u32;
|
||||
let mut failed = 0u32;
|
||||
let mut skipped = 0u32;
|
||||
|
||||
// ── Configuration checks ──────────────────────────────────
|
||||
// Load settings once for checks that need them.
|
||||
let settings = Settings::load();
|
||||
|
||||
// ── Settings & core config ─────────────────────────────────
|
||||
|
||||
check(
|
||||
"Settings file",
|
||||
check_settings_file(),
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
check(
|
||||
"NEAR AI session",
|
||||
check_nearai_session().await,
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
check(
|
||||
"LLM configuration",
|
||||
check_llm_config(&settings),
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
check(
|
||||
@@ -30,6 +52,7 @@ pub async fn run_doctor_command() -> anyhow::Result<()> {
|
||||
check_database().await,
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
check(
|
||||
@@ -37,15 +60,75 @@ pub async fn run_doctor_command() -> anyhow::Result<()> {
|
||||
check_workspace_dir(),
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
// ── Subsystem configuration checks ─────────────────────────
|
||||
|
||||
check(
|
||||
"Embeddings",
|
||||
check_embeddings(&settings),
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
check(
|
||||
"Routines config",
|
||||
check_routines_config(),
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
check(
|
||||
"Gateway config",
|
||||
check_gateway_config(&settings),
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
check(
|
||||
"MCP servers",
|
||||
check_mcp_config().await,
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
check(
|
||||
"Skills",
|
||||
check_skills().await,
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
check(
|
||||
"Secrets",
|
||||
check_secrets(&settings),
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
check(
|
||||
"Service",
|
||||
check_service_installed(),
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
// ── External binary checks ────────────────────────────────
|
||||
|
||||
check(
|
||||
"Docker",
|
||||
check_binary("docker", &["--version"]),
|
||||
"Docker daemon",
|
||||
check_docker_daemon().await,
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
check(
|
||||
@@ -53,6 +136,7 @@ pub async fn run_doctor_command() -> anyhow::Result<()> {
|
||||
check_binary("cloudflared", &["--version"]),
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
check(
|
||||
@@ -60,6 +144,7 @@ pub async fn run_doctor_command() -> anyhow::Result<()> {
|
||||
check_binary("ngrok", &["version"]),
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
check(
|
||||
@@ -67,12 +152,13 @@ pub async fn run_doctor_command() -> anyhow::Result<()> {
|
||||
check_binary("tailscale", &["version"]),
|
||||
&mut passed,
|
||||
&mut failed,
|
||||
&mut skipped,
|
||||
);
|
||||
|
||||
// ── Summary ───────────────────────────────────────────────
|
||||
|
||||
println!();
|
||||
println!(" {passed} passed, {failed} failed");
|
||||
println!(" {passed} passed, {failed} failed, {skipped} skipped");
|
||||
|
||||
if failed > 0 {
|
||||
println!("\n Some checks failed. This is normal if you don't use those features.");
|
||||
@@ -83,7 +169,7 @@ pub async fn run_doctor_command() -> anyhow::Result<()> {
|
||||
|
||||
// ── Individual checks ───────────────────────────────────────
|
||||
|
||||
fn check(name: &str, result: CheckResult, passed: &mut u32, failed: &mut u32) {
|
||||
fn check(name: &str, result: CheckResult, passed: &mut u32, failed: &mut u32, skipped: &mut u32) {
|
||||
match result {
|
||||
CheckResult::Pass(detail) => {
|
||||
*passed += 1;
|
||||
@@ -94,6 +180,7 @@ fn check(name: &str, result: CheckResult, passed: &mut u32, failed: &mut u32) {
|
||||
println!(" [FAIL] {name}: {detail}");
|
||||
}
|
||||
CheckResult::Skip(reason) => {
|
||||
*skipped += 1;
|
||||
println!(" [skip] {name}: {reason}");
|
||||
}
|
||||
}
|
||||
@@ -105,6 +192,29 @@ enum CheckResult {
|
||||
Skip(String),
|
||||
}
|
||||
|
||||
// ── Settings file ───────────────────────────────────────────
|
||||
|
||||
fn check_settings_file() -> CheckResult {
|
||||
let path = Settings::default_path();
|
||||
if !path.exists() {
|
||||
return CheckResult::Pass("no settings file (defaults will be used)".into());
|
||||
}
|
||||
|
||||
match std::fs::read_to_string(&path) {
|
||||
Ok(data) => match serde_json::from_str::<serde_json::Value>(&data) {
|
||||
Ok(_) => CheckResult::Pass(format!("valid ({})", path.display())),
|
||||
Err(e) => CheckResult::Fail(format!(
|
||||
"settings.json is malformed: {}. Fix or delete {}",
|
||||
e,
|
||||
path.display()
|
||||
)),
|
||||
},
|
||||
Err(e) => CheckResult::Fail(format!("cannot read {}: {}", path.display(), e)),
|
||||
}
|
||||
}
|
||||
|
||||
// ── NEAR AI session ─────────────────────────────────────────
|
||||
|
||||
async fn check_nearai_session() -> CheckResult {
|
||||
// Check if session file exists
|
||||
let session_path = crate::config::llm::default_session_path();
|
||||
@@ -129,6 +239,27 @@ async fn check_nearai_session() -> CheckResult {
|
||||
}
|
||||
}
|
||||
|
||||
// ── LLM configuration ──────────────────────────────────────
|
||||
|
||||
fn check_llm_config(settings: &Settings) -> CheckResult {
|
||||
match crate::llm::LlmConfig::resolve(settings) {
|
||||
Ok(config) => {
|
||||
// Show the model for the active backend, not always nearai.model.
|
||||
let model = if let Some(ref bedrock) = config.bedrock {
|
||||
&bedrock.model
|
||||
} else if let Some(ref provider) = config.provider {
|
||||
&provider.model
|
||||
} else {
|
||||
&config.nearai.model
|
||||
};
|
||||
CheckResult::Pass(format!("backend={}, model={}", config.backend, model))
|
||||
}
|
||||
Err(e) => CheckResult::Fail(format!("LLM config error: {e}")),
|
||||
}
|
||||
}
|
||||
|
||||
// ── Database ────────────────────────────────────────────────
|
||||
|
||||
async fn check_database() -> CheckResult {
|
||||
let backend = std::env::var("DATABASE_BACKEND")
|
||||
.ok()
|
||||
@@ -192,6 +323,8 @@ async fn try_pg_connect() -> Result<(), String> {
|
||||
Err("postgres feature not compiled in".into())
|
||||
}
|
||||
|
||||
// ── Workspace directory ─────────────────────────────────────
|
||||
|
||||
fn check_workspace_dir() -> CheckResult {
|
||||
let dir = ironclaw_base_dir();
|
||||
|
||||
@@ -206,6 +339,222 @@ fn check_workspace_dir() -> CheckResult {
|
||||
}
|
||||
}
|
||||
|
||||
// ── Embeddings ──────────────────────────────────────────────
|
||||
|
||||
fn check_embeddings(settings: &Settings) -> CheckResult {
|
||||
match crate::config::EmbeddingsConfig::resolve(settings) {
|
||||
Ok(config) => {
|
||||
if !config.enabled {
|
||||
return CheckResult::Skip("disabled (set EMBEDDING_ENABLED=true)".into());
|
||||
}
|
||||
let has_creds = match config.provider.as_str() {
|
||||
"openai" => config.openai_api_key().is_some(),
|
||||
"nearai" => {
|
||||
// NearAiEmbeddings uses SessionManager::get_token() which
|
||||
// only returns session tokens, NOT NEARAI_API_KEY
|
||||
// (src/workspace/embeddings.rs:309, src/llm/session.rs:132).
|
||||
let session_path = crate::config::llm::default_session_path();
|
||||
session_path.exists()
|
||||
&& std::fs::read_to_string(&session_path)
|
||||
.map(|s| !s.trim().is_empty())
|
||||
.unwrap_or(false)
|
||||
}
|
||||
"ollama" => true, // local, no creds needed
|
||||
_ => config.openai_api_key().is_some(),
|
||||
};
|
||||
if has_creds {
|
||||
CheckResult::Pass(format!(
|
||||
"provider={}, model={}",
|
||||
config.provider, config.model
|
||||
))
|
||||
} else {
|
||||
let hint = match config.provider.as_str() {
|
||||
"nearai" => "run `ironclaw onboard` to create a session",
|
||||
_ => "set OPENAI_API_KEY",
|
||||
};
|
||||
CheckResult::Fail(format!(
|
||||
"provider={} but credentials missing ({})",
|
||||
config.provider, hint
|
||||
))
|
||||
}
|
||||
}
|
||||
Err(e) => CheckResult::Fail(format!("config error: {e}")),
|
||||
}
|
||||
}
|
||||
|
||||
// ── Routines config ─────────────────────────────────────────
|
||||
|
||||
fn check_routines_config() -> CheckResult {
|
||||
match crate::config::RoutineConfig::resolve() {
|
||||
Ok(config) => {
|
||||
if config.enabled {
|
||||
CheckResult::Pass(format!(
|
||||
"enabled (interval={}s, max_concurrent={})",
|
||||
config.cron_check_interval_secs, config.max_concurrent_routines
|
||||
))
|
||||
} else {
|
||||
CheckResult::Skip("disabled".into())
|
||||
}
|
||||
}
|
||||
Err(e) => CheckResult::Fail(format!("config error: {e}")),
|
||||
}
|
||||
}
|
||||
|
||||
// ── Gateway config ──────────────────────────────────────────
|
||||
|
||||
fn check_gateway_config(settings: &Settings) -> CheckResult {
|
||||
// Use the same resolve() path as runtime so invalid env values
|
||||
// (e.g. GATEWAY_PORT=abc) are caught here too.
|
||||
match crate::config::ChannelsConfig::resolve(settings) {
|
||||
Ok(channels) => match channels.gateway {
|
||||
Some(gw) => {
|
||||
if gw.auth_token.is_some() {
|
||||
CheckResult::Pass(format!(
|
||||
"enabled at {}:{} (auth token set)",
|
||||
gw.host, gw.port
|
||||
))
|
||||
} else {
|
||||
CheckResult::Pass(format!(
|
||||
"enabled at {}:{} (no auth token — random token will be generated)",
|
||||
gw.host, gw.port
|
||||
))
|
||||
}
|
||||
}
|
||||
None => CheckResult::Skip("disabled (GATEWAY_ENABLED=false)".into()),
|
||||
},
|
||||
Err(e) => CheckResult::Fail(format!("config error: {e}")),
|
||||
}
|
||||
}
|
||||
|
||||
// ── MCP servers ─────────────────────────────────────────────
|
||||
|
||||
async fn check_mcp_config() -> CheckResult {
|
||||
match crate::tools::mcp::config::load_mcp_servers().await {
|
||||
Ok(file) => {
|
||||
let servers: Vec<_> = file.enabled_servers().collect();
|
||||
if servers.is_empty() {
|
||||
return CheckResult::Skip("no MCP servers configured".into());
|
||||
}
|
||||
|
||||
let mut invalid = Vec::new();
|
||||
for server in &servers {
|
||||
if let Err(e) = server.validate() {
|
||||
invalid.push(format!("{}: {}", server.name, e));
|
||||
}
|
||||
}
|
||||
|
||||
if invalid.is_empty() {
|
||||
CheckResult::Pass(format!("{} server(s) configured, all valid", servers.len()))
|
||||
} else {
|
||||
CheckResult::Fail(format!(
|
||||
"{} server(s), {} invalid: {}",
|
||||
servers.len(),
|
||||
invalid.len(),
|
||||
invalid.join("; ")
|
||||
))
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
// Distinguish no config from corrupted config
|
||||
let msg = e.to_string();
|
||||
if msg.contains("not found") || msg.contains("No such file") {
|
||||
CheckResult::Skip("no MCP config file".into())
|
||||
} else {
|
||||
CheckResult::Fail(format!("config error: {e}"))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Skills ──────────────────────────────────────────────────
|
||||
|
||||
async fn check_skills() -> CheckResult {
|
||||
let user_dir = ironclaw_base_dir().join("skills");
|
||||
let installed_dir = ironclaw_base_dir().join("installed_skills");
|
||||
|
||||
let mut registry = crate::skills::SkillRegistry::new(user_dir.clone());
|
||||
registry = registry.with_installed_dir(installed_dir);
|
||||
|
||||
// discover_all() returns loaded skill names (not warnings).
|
||||
let _loaded_names = registry.discover_all().await;
|
||||
|
||||
let count = registry.count();
|
||||
if count == 0 {
|
||||
return CheckResult::Skip("no skills discovered".into());
|
||||
}
|
||||
|
||||
CheckResult::Pass(format!("{count} skill(s) loaded"))
|
||||
}
|
||||
|
||||
// ── Secrets ─────────────────────────────────────────────────
|
||||
|
||||
fn check_secrets(settings: &Settings) -> CheckResult {
|
||||
match settings.secrets_master_key_source {
|
||||
crate::settings::KeySource::Keychain => {
|
||||
CheckResult::Pass("master key source: OS keychain".into())
|
||||
}
|
||||
crate::settings::KeySource::Env => {
|
||||
if std::env::var("SECRETS_MASTER_KEY").is_ok() {
|
||||
CheckResult::Pass("master key source: env var (set)".into())
|
||||
} else {
|
||||
CheckResult::Fail(
|
||||
"master key source: env var but SECRETS_MASTER_KEY not set".into(),
|
||||
)
|
||||
}
|
||||
}
|
||||
crate::settings::KeySource::None => {
|
||||
CheckResult::Skip("secrets not configured (run `ironclaw onboard`)".into())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Service ─────────────────────────────────────────────────
|
||||
|
||||
fn check_service_installed() -> CheckResult {
|
||||
if cfg!(target_os = "macos") {
|
||||
let plist =
|
||||
dirs::home_dir().map(|h| h.join("Library/LaunchAgents/com.ironclaw.daemon.plist"));
|
||||
match plist {
|
||||
Some(path) if path.exists() => {
|
||||
CheckResult::Pass(format!("launchd plist installed ({})", path.display()))
|
||||
}
|
||||
Some(_) => CheckResult::Skip("not installed (run `ironclaw service install`)".into()),
|
||||
None => CheckResult::Skip("cannot determine home directory".into()),
|
||||
}
|
||||
} else if cfg!(target_os = "linux") {
|
||||
let unit = dirs::home_dir().map(|h| h.join(".config/systemd/user/ironclaw.service"));
|
||||
match unit {
|
||||
Some(path) if path.exists() => {
|
||||
CheckResult::Pass(format!("systemd unit installed ({})", path.display()))
|
||||
}
|
||||
Some(_) => CheckResult::Skip("not installed (run `ironclaw service install`)".into()),
|
||||
None => CheckResult::Skip("cannot determine home directory".into()),
|
||||
}
|
||||
} else {
|
||||
CheckResult::Skip("service management not supported on this platform".into())
|
||||
}
|
||||
}
|
||||
|
||||
// ── Docker daemon ───────────────────────────────────────────
|
||||
|
||||
async fn check_docker_daemon() -> CheckResult {
|
||||
let detection = crate::sandbox::check_docker().await;
|
||||
match detection.status {
|
||||
crate::sandbox::DockerStatus::Available => CheckResult::Pass("running".into()),
|
||||
crate::sandbox::DockerStatus::NotInstalled => CheckResult::Skip(format!(
|
||||
"not installed. {}",
|
||||
detection.platform.install_hint()
|
||||
)),
|
||||
crate::sandbox::DockerStatus::NotRunning => CheckResult::Fail(format!(
|
||||
"installed but not running. {}",
|
||||
detection.platform.start_hint()
|
||||
)),
|
||||
crate::sandbox::DockerStatus::Disabled => CheckResult::Skip("sandbox disabled".into()),
|
||||
}
|
||||
}
|
||||
|
||||
// ── External binary ─────────────────────────────────────────
|
||||
|
||||
fn check_binary(name: &str, args: &[&str]) -> CheckResult {
|
||||
match std::process::Command::new(name)
|
||||
.args(args)
|
||||
@@ -273,6 +622,193 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_settings_file_handles_missing() {
|
||||
// Settings::default_path() might or might not exist, but must not panic
|
||||
let result = check_settings_file();
|
||||
match result {
|
||||
CheckResult::Pass(_) | CheckResult::Fail(_) | CheckResult::Skip(_) => {}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_llm_config_does_not_panic() {
|
||||
let settings = Settings::default();
|
||||
let result = check_llm_config(&settings);
|
||||
match result {
|
||||
CheckResult::Pass(_) | CheckResult::Fail(_) | CheckResult::Skip(_) => {}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_routines_config_does_not_panic() {
|
||||
let result = check_routines_config();
|
||||
match result {
|
||||
CheckResult::Pass(_) | CheckResult::Fail(_) | CheckResult::Skip(_) => {}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_gateway_config_does_not_panic() {
|
||||
let settings = Settings::default();
|
||||
let result = check_gateway_config(&settings);
|
||||
match result {
|
||||
CheckResult::Pass(_) | CheckResult::Fail(_) | CheckResult::Skip(_) => {}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_embeddings_does_not_panic() {
|
||||
let settings = Settings::default();
|
||||
let result = check_embeddings(&settings);
|
||||
match result {
|
||||
CheckResult::Pass(_) | CheckResult::Fail(_) | CheckResult::Skip(_) => {}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_secrets_none_returns_skip() {
|
||||
let settings = Settings::default();
|
||||
match check_secrets(&settings) {
|
||||
CheckResult::Skip(msg) => {
|
||||
assert!(
|
||||
msg.contains("not configured"),
|
||||
"expected 'not configured' in skip message, got: {msg}"
|
||||
);
|
||||
}
|
||||
other => panic!(
|
||||
"expected Skip for default settings, got: {}",
|
||||
format_result(&other)
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_service_installed_does_not_panic() {
|
||||
let result = check_service_installed();
|
||||
match result {
|
||||
CheckResult::Pass(_) | CheckResult::Fail(_) | CheckResult::Skip(_) => {}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn check_docker_daemon_does_not_panic() {
|
||||
let result = check_docker_daemon().await;
|
||||
match result {
|
||||
CheckResult::Pass(_) | CheckResult::Fail(_) | CheckResult::Skip(_) => {}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn check_mcp_config_does_not_panic() {
|
||||
let result = check_mcp_config().await;
|
||||
match result {
|
||||
CheckResult::Pass(_) | CheckResult::Fail(_) | CheckResult::Skip(_) => {}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn check_skills_does_not_panic() {
|
||||
let result = check_skills().await;
|
||||
match result {
|
||||
CheckResult::Pass(_) | CheckResult::Fail(_) | CheckResult::Skip(_) => {}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_llm_config_shows_nearai_model_for_nearai_backend() {
|
||||
let _guard = crate::config::helpers::ENV_MUTEX.lock().expect("env mutex");
|
||||
// SAFETY: Under ENV_MUTEX, no concurrent env access.
|
||||
unsafe {
|
||||
std::env::remove_var("LLM_BACKEND");
|
||||
}
|
||||
let settings = Settings::default();
|
||||
match check_llm_config(&settings) {
|
||||
CheckResult::Pass(msg) => {
|
||||
assert!(
|
||||
msg.contains("backend=nearai"),
|
||||
"expected nearai backend, got: {msg}"
|
||||
);
|
||||
// Must NOT show a bedrock or registry model when backend is nearai
|
||||
assert!(
|
||||
!msg.contains("anthropic.claude"),
|
||||
"should not show bedrock model for nearai backend: {msg}"
|
||||
);
|
||||
}
|
||||
other => panic!(
|
||||
"expected Pass for default LLM config, got: {}",
|
||||
format_result(&other)
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_embeddings_disabled_by_default_returns_skip() {
|
||||
let _guard = crate::config::helpers::ENV_MUTEX.lock().expect("env mutex");
|
||||
// SAFETY: Under ENV_MUTEX.
|
||||
unsafe {
|
||||
std::env::remove_var("EMBEDDING_ENABLED");
|
||||
}
|
||||
let settings = Settings::default();
|
||||
match check_embeddings(&settings) {
|
||||
CheckResult::Skip(msg) => {
|
||||
assert!(
|
||||
msg.contains("disabled"),
|
||||
"expected 'disabled' in skip message, got: {msg}"
|
||||
);
|
||||
}
|
||||
other => panic!(
|
||||
"expected Skip for disabled embeddings, got: {}",
|
||||
format_result(&other)
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_routines_enabled_by_default() {
|
||||
let _guard = crate::config::helpers::ENV_MUTEX.lock().expect("env mutex");
|
||||
// SAFETY: Under ENV_MUTEX.
|
||||
unsafe {
|
||||
std::env::remove_var("ROUTINES_ENABLED");
|
||||
}
|
||||
match check_routines_config() {
|
||||
CheckResult::Pass(msg) => {
|
||||
assert!(
|
||||
msg.contains("enabled"),
|
||||
"routines should be enabled by default, got: {msg}"
|
||||
);
|
||||
}
|
||||
other => panic!(
|
||||
"expected Pass for default routines, got: {}",
|
||||
format_result(&other)
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_secrets_env_without_var_returns_fail() {
|
||||
let settings = Settings {
|
||||
secrets_master_key_source: crate::settings::KeySource::Env,
|
||||
..Default::default()
|
||||
};
|
||||
match check_secrets(&settings) {
|
||||
CheckResult::Fail(msg) => {
|
||||
assert!(
|
||||
msg.contains("SECRETS_MASTER_KEY not set"),
|
||||
"expected mention of missing env var, got: {msg}"
|
||||
);
|
||||
}
|
||||
CheckResult::Pass(_) => {
|
||||
// If SECRETS_MASTER_KEY happens to be set in the environment,
|
||||
// Pass is correct — don't fail the test.
|
||||
}
|
||||
other => panic!(
|
||||
"expected Fail or Pass for env key source, got: {}",
|
||||
format_result(&other)
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
fn format_result(r: &CheckResult) -> String {
|
||||
match r {
|
||||
CheckResult::Pass(s) => format!("Pass({s})"),
|
||||
|
||||
@@ -0,0 +1,391 @@
|
||||
//! Shared tool execution pipeline.
|
||||
//!
|
||||
//! Provides a single implementation of the validate → timeout → execute → serialize
|
||||
//! pipeline used by all agentic loop consumers (chat, job, container) and the
|
||||
//! scheduler's subtask execution.
|
||||
|
||||
use crate::context::JobContext;
|
||||
use crate::error::Error;
|
||||
use crate::llm::ChatMessage;
|
||||
use crate::safety::SafetyLayer;
|
||||
use crate::tools::{ToolRegistry, redact_params};
|
||||
|
||||
/// Execute a tool with safety checks: lookup → validate → timeout → execute → serialize.
|
||||
///
|
||||
/// This is the single canonical implementation of tool execution. All consumers
|
||||
/// (chat dispatcher, job worker, container runtime, scheduler subtasks) use this
|
||||
/// function instead of maintaining their own copies.
|
||||
pub async fn execute_tool_with_safety(
|
||||
tools: &ToolRegistry,
|
||||
safety: &SafetyLayer,
|
||||
tool_name: &str,
|
||||
params: &serde_json::Value,
|
||||
job_ctx: &JobContext,
|
||||
) -> Result<String, Error> {
|
||||
let tool = tools
|
||||
.get(tool_name)
|
||||
.await
|
||||
.ok_or_else(|| crate::error::ToolError::NotFound {
|
||||
name: tool_name.to_string(),
|
||||
})?;
|
||||
|
||||
// Validate tool parameters
|
||||
let validation = safety.validator().validate_tool_params(params);
|
||||
if !validation.is_valid {
|
||||
let details = validation
|
||||
.errors
|
||||
.iter()
|
||||
.map(|e| format!("{}: {}", e.field, e.message))
|
||||
.collect::<Vec<_>>()
|
||||
.join("; ");
|
||||
return Err(crate::error::ToolError::InvalidParameters {
|
||||
name: tool_name.to_string(),
|
||||
reason: format!("Invalid tool parameters: {}", details),
|
||||
}
|
||||
.into());
|
||||
}
|
||||
|
||||
let safe_params = redact_params(params, tool.sensitive_params());
|
||||
tracing::debug!(
|
||||
tool = %tool_name,
|
||||
params = %safe_params,
|
||||
"Tool call started"
|
||||
);
|
||||
|
||||
// Execute with per-tool timeout
|
||||
let timeout = tool.execution_timeout();
|
||||
let start = std::time::Instant::now();
|
||||
let result = tokio::time::timeout(timeout, async {
|
||||
tool.execute(params.clone(), job_ctx).await
|
||||
})
|
||||
.await;
|
||||
let elapsed = start.elapsed();
|
||||
|
||||
match &result {
|
||||
Ok(Ok(output)) => {
|
||||
let result_size = serde_json::to_string(&output.result)
|
||||
.map(|s| s.len())
|
||||
.unwrap_or(0);
|
||||
tracing::debug!(
|
||||
tool = %tool_name,
|
||||
elapsed_ms = elapsed.as_millis() as u64,
|
||||
result_size_bytes = result_size,
|
||||
"Tool call succeeded"
|
||||
);
|
||||
}
|
||||
Ok(Err(e)) => {
|
||||
tracing::debug!(
|
||||
tool = %tool_name,
|
||||
elapsed_ms = elapsed.as_millis() as u64,
|
||||
error = %e,
|
||||
"Tool call failed"
|
||||
);
|
||||
}
|
||||
Err(_) => {
|
||||
tracing::debug!(
|
||||
tool = %tool_name,
|
||||
elapsed_ms = elapsed.as_millis() as u64,
|
||||
timeout_secs = timeout.as_secs(),
|
||||
"Tool call timed out"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let result = result
|
||||
.map_err(|_| crate::error::ToolError::Timeout {
|
||||
name: tool_name.to_string(),
|
||||
timeout,
|
||||
})?
|
||||
.map_err(|e| crate::error::ToolError::ExecutionFailed {
|
||||
name: tool_name.to_string(),
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
|
||||
serde_json::to_string_pretty(&result.result).map_err(|e| {
|
||||
crate::error::ToolError::ExecutionFailed {
|
||||
name: tool_name.to_string(),
|
||||
reason: format!("Failed to serialize result: {}", e),
|
||||
}
|
||||
.into()
|
||||
})
|
||||
}
|
||||
|
||||
/// Process a tool result into a `ChatMessage::tool_result` with safety sanitization.
|
||||
///
|
||||
/// On success: sanitize → wrap → ChatMessage::tool_result.
|
||||
/// On error: format error → ChatMessage::tool_result.
|
||||
///
|
||||
/// Returns the content string and the ChatMessage.
|
||||
pub fn process_tool_result(
|
||||
safety: &SafetyLayer,
|
||||
tool_name: &str,
|
||||
tool_call_id: &str,
|
||||
result: &Result<String, impl std::fmt::Display>,
|
||||
) -> (String, ChatMessage) {
|
||||
let content = match result {
|
||||
Ok(output) => {
|
||||
let sanitized = safety.sanitize_tool_output(tool_name, output);
|
||||
safety.wrap_for_llm(tool_name, &sanitized.content, sanitized.was_modified)
|
||||
}
|
||||
Err(e) => format!("Error: {}", e),
|
||||
};
|
||||
let message = ChatMessage::tool_result(tool_call_id, tool_name, content.clone());
|
||||
(content, message)
|
||||
}
|
||||
|
||||
/// Execute a tool with safety checks, returning a string error (for container runtime).
|
||||
///
|
||||
/// This is a thin wrapper around `execute_tool_with_safety` that converts
|
||||
/// `Error` to `String` for the container runtime's simpler error model.
|
||||
pub async fn execute_tool_simple(
|
||||
tools: &ToolRegistry,
|
||||
safety: &SafetyLayer,
|
||||
tool_name: &str,
|
||||
params: &serde_json::Value,
|
||||
job_ctx: &JobContext,
|
||||
) -> Result<String, String> {
|
||||
execute_tool_with_safety(tools, safety, tool_name, params, job_ctx)
|
||||
.await
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::tools::tool::{Tool, ToolError, ToolOutput};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
struct EchoTool;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Tool for EchoTool {
|
||||
fn name(&self) -> &str {
|
||||
"echo"
|
||||
}
|
||||
fn description(&self) -> &str {
|
||||
"Echoes input"
|
||||
}
|
||||
fn parameters_schema(&self) -> serde_json::Value {
|
||||
serde_json::json!({"type": "object", "properties": {}})
|
||||
}
|
||||
async fn execute(
|
||||
&self,
|
||||
params: serde_json::Value,
|
||||
_ctx: &JobContext,
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
Ok(ToolOutput::success(params, Duration::default()))
|
||||
}
|
||||
fn requires_sanitization(&self) -> bool {
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
struct FailTool;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Tool for FailTool {
|
||||
fn name(&self) -> &str {
|
||||
"fail_tool"
|
||||
}
|
||||
fn description(&self) -> &str {
|
||||
"Always fails"
|
||||
}
|
||||
fn parameters_schema(&self) -> serde_json::Value {
|
||||
serde_json::json!({"type": "object", "properties": {}})
|
||||
}
|
||||
async fn execute(
|
||||
&self,
|
||||
_: serde_json::Value,
|
||||
_: &JobContext,
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
Err(ToolError::ExecutionFailed(
|
||||
"intentional failure".to_string(),
|
||||
))
|
||||
}
|
||||
fn requires_sanitization(&self) -> bool {
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
struct SlowTool;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Tool for SlowTool {
|
||||
fn name(&self) -> &str {
|
||||
"slow_tool"
|
||||
}
|
||||
fn description(&self) -> &str {
|
||||
"Sleeps forever"
|
||||
}
|
||||
fn parameters_schema(&self) -> serde_json::Value {
|
||||
serde_json::json!({"type": "object", "properties": {}})
|
||||
}
|
||||
async fn execute(
|
||||
&self,
|
||||
_: serde_json::Value,
|
||||
_: &JobContext,
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
tokio::time::sleep(Duration::from_secs(60)).await;
|
||||
unreachable!()
|
||||
}
|
||||
fn execution_timeout(&self) -> Duration {
|
||||
Duration::from_millis(50)
|
||||
}
|
||||
fn requires_sanitization(&self) -> bool {
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
fn test_safety() -> SafetyLayer {
|
||||
SafetyLayer::new(&crate::config::SafetyConfig {
|
||||
max_output_length: 100_000,
|
||||
injection_check_enabled: false,
|
||||
})
|
||||
}
|
||||
|
||||
fn test_job_ctx() -> JobContext {
|
||||
JobContext::default()
|
||||
}
|
||||
|
||||
async fn registry_with(tools: Vec<Arc<dyn Tool>>) -> ToolRegistry {
|
||||
let registry = ToolRegistry::new();
|
||||
for tool in tools {
|
||||
registry.register(tool).await;
|
||||
}
|
||||
registry
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_execute_success() {
|
||||
let registry = registry_with(vec![Arc::new(EchoTool)]).await;
|
||||
let safety = test_safety();
|
||||
let params = serde_json::json!({"message": "hello"});
|
||||
|
||||
let result =
|
||||
execute_tool_with_safety(®istry, &safety, "echo", ¶ms, &test_job_ctx()).await;
|
||||
|
||||
assert!(result.is_ok(), "Echo tool should succeed");
|
||||
let output = result.unwrap();
|
||||
assert!(
|
||||
output.contains("hello"),
|
||||
"Output should contain the echoed input"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_execute_missing_tool() {
|
||||
let registry = registry_with(vec![]).await;
|
||||
let safety = test_safety();
|
||||
|
||||
let result = execute_tool_with_safety(
|
||||
®istry,
|
||||
&safety,
|
||||
"nonexistent",
|
||||
&serde_json::json!({}),
|
||||
&test_job_ctx(),
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(result.is_err(), "Missing tool should return error");
|
||||
let err = result.unwrap_err().to_string();
|
||||
assert!(
|
||||
err.contains("nonexistent") || err.contains("not found"),
|
||||
"Error should mention the tool: {}",
|
||||
err
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_execute_tool_failure() {
|
||||
let registry = registry_with(vec![Arc::new(FailTool)]).await;
|
||||
let safety = test_safety();
|
||||
|
||||
let result = execute_tool_with_safety(
|
||||
®istry,
|
||||
&safety,
|
||||
"fail_tool",
|
||||
&serde_json::json!({}),
|
||||
&test_job_ctx(),
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(result.is_err(), "FailTool should return error");
|
||||
let err = result.unwrap_err().to_string();
|
||||
assert!(
|
||||
err.contains("intentional failure"),
|
||||
"Error should contain the failure reason: {}",
|
||||
err
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_execute_tool_timeout() {
|
||||
let registry = registry_with(vec![Arc::new(SlowTool)]).await;
|
||||
let safety = test_safety();
|
||||
|
||||
let start = std::time::Instant::now();
|
||||
let result = execute_tool_with_safety(
|
||||
®istry,
|
||||
&safety,
|
||||
"slow_tool",
|
||||
&serde_json::json!({}),
|
||||
&test_job_ctx(),
|
||||
)
|
||||
.await;
|
||||
let elapsed = start.elapsed();
|
||||
|
||||
assert!(result.is_err(), "SlowTool should timeout");
|
||||
let err = result.unwrap_err().to_string();
|
||||
assert!(
|
||||
err.to_lowercase().contains("timeout") || err.to_lowercase().contains("timed out"),
|
||||
"Error should mention timeout: {}",
|
||||
err
|
||||
);
|
||||
assert!(
|
||||
elapsed < Duration::from_secs(1),
|
||||
"Should timeout quickly, not wait 60s"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_process_tool_result_success() {
|
||||
let safety = test_safety();
|
||||
let result: Result<String, String> = Ok("tool output data".to_string());
|
||||
|
||||
let (content, message) = process_tool_result(&safety, "echo", "call_1", &result);
|
||||
|
||||
assert!(
|
||||
content.contains("tool_output"),
|
||||
"Content should be XML-wrapped: {}",
|
||||
content
|
||||
);
|
||||
assert!(
|
||||
content.contains("tool output data"),
|
||||
"Content should contain the output: {}",
|
||||
content
|
||||
);
|
||||
assert_eq!(message.role, crate::llm::Role::Tool);
|
||||
assert_eq!(message.name.as_deref(), Some("echo"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_process_tool_result_error() {
|
||||
let safety = test_safety();
|
||||
let result: Result<String, String> = Err("something went wrong".to_string());
|
||||
|
||||
let (content, message) = process_tool_result(&safety, "echo", "call_1", &result);
|
||||
|
||||
assert!(
|
||||
content.contains("Error:"),
|
||||
"Error content should start with 'Error:': {}",
|
||||
content
|
||||
);
|
||||
assert!(
|
||||
content.contains("something went wrong"),
|
||||
"Error content should contain the message: {}",
|
||||
content
|
||||
);
|
||||
assert_eq!(message.role, crate::llm::Role::Tool);
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,7 @@
|
||||
|
||||
pub mod builder;
|
||||
pub mod builtin;
|
||||
pub mod execute;
|
||||
pub mod mcp;
|
||||
pub mod rate_limiter;
|
||||
pub mod schema_validator;
|
||||
|
||||
+1
-1
@@ -24,7 +24,7 @@ pub fn floor_char_boundary(s: &str, pos: usize) -> usize {
|
||||
pub fn llm_signals_completion(response: &str) -> bool {
|
||||
let lower = response.to_lowercase();
|
||||
|
||||
// Superset of phrases from agent/worker.rs and worker/runtime.rs.
|
||||
// Superset of phrases from worker/job.rs and worker/container.rs.
|
||||
let positive_phrases = [
|
||||
"job is complete",
|
||||
"job is done",
|
||||
|
||||
@@ -0,0 +1,539 @@
|
||||
//! Worker runtime: the main execution loop inside a container.
|
||||
//!
|
||||
//! Reuses the existing `Reasoning` and `SafetyLayer` infrastructure but
|
||||
//! connects to the orchestrator for LLM calls instead of calling APIs directly.
|
||||
//! Streams real-time events (message, tool_use, tool_result, result) through
|
||||
//! the orchestrator's job event pipeline for UI visibility.
|
||||
//!
|
||||
//! Uses the shared `AgenticLoop` engine via `ContainerDelegate`.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use tokio::sync::Mutex;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::agent::agentic_loop::{
|
||||
AgenticLoopConfig, LoopDelegate, LoopOutcome, LoopSignal, TextAction, truncate_for_preview,
|
||||
};
|
||||
use crate::config::SafetyConfig;
|
||||
use crate::context::JobContext;
|
||||
use crate::error::WorkerError;
|
||||
use crate::llm::{ChatMessage, LlmProvider, Reasoning, ReasoningContext};
|
||||
use crate::safety::SafetyLayer;
|
||||
use crate::tools::ToolRegistry;
|
||||
use crate::tools::execute::{execute_tool_simple, process_tool_result};
|
||||
use crate::worker::api::{CompletionReport, JobEventPayload, StatusUpdate, WorkerHttpClient};
|
||||
use crate::worker::proxy_llm::ProxyLlmProvider;
|
||||
|
||||
/// Configuration for the worker runtime.
|
||||
pub struct WorkerConfig {
|
||||
pub job_id: Uuid,
|
||||
pub orchestrator_url: String,
|
||||
pub max_iterations: u32,
|
||||
pub timeout: Duration,
|
||||
}
|
||||
|
||||
impl Default for WorkerConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
job_id: Uuid::nil(),
|
||||
orchestrator_url: String::new(),
|
||||
max_iterations: 50,
|
||||
timeout: Duration::from_secs(600),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The worker runtime runs inside a Docker container.
|
||||
///
|
||||
/// It connects to the orchestrator over HTTP, fetches its job description,
|
||||
/// then runs a tool execution loop until the job is complete. Events are
|
||||
/// streamed to the orchestrator so the UI can show real-time progress.
|
||||
pub struct WorkerRuntime {
|
||||
config: WorkerConfig,
|
||||
client: Arc<WorkerHttpClient>,
|
||||
llm: Arc<dyn LlmProvider>,
|
||||
safety: Arc<SafetyLayer>,
|
||||
tools: Arc<ToolRegistry>,
|
||||
/// Credentials fetched from the orchestrator, injected into child processes
|
||||
/// via `Command::envs()` rather than mutating the global process environment.
|
||||
///
|
||||
/// Wrapped in `Arc` to avoid deep-cloning the map on every tool invocation.
|
||||
extra_env: Arc<HashMap<String, String>>,
|
||||
}
|
||||
|
||||
impl WorkerRuntime {
|
||||
/// Create a new worker runtime.
|
||||
///
|
||||
/// Reads `IRONCLAW_WORKER_TOKEN` from the environment for auth.
|
||||
pub fn new(config: WorkerConfig) -> Result<Self, WorkerError> {
|
||||
let client = Arc::new(WorkerHttpClient::from_env(
|
||||
config.orchestrator_url.clone(),
|
||||
config.job_id,
|
||||
)?);
|
||||
|
||||
let llm: Arc<dyn LlmProvider> = Arc::new(ProxyLlmProvider::new(
|
||||
Arc::clone(&client),
|
||||
"proxied".to_string(),
|
||||
));
|
||||
|
||||
let safety = Arc::new(SafetyLayer::new(&SafetyConfig {
|
||||
max_output_length: 100_000,
|
||||
injection_check_enabled: true,
|
||||
}));
|
||||
|
||||
let tools = Arc::new(ToolRegistry::new());
|
||||
// Register only container-safe tools
|
||||
tools.register_container_tools();
|
||||
|
||||
Ok(Self {
|
||||
config,
|
||||
client,
|
||||
llm,
|
||||
safety,
|
||||
tools,
|
||||
extra_env: Arc::new(HashMap::new()),
|
||||
})
|
||||
}
|
||||
|
||||
/// Run the worker until the job is complete or an error occurs.
|
||||
pub async fn run(mut self) -> Result<(), WorkerError> {
|
||||
tracing::info!("Worker starting for job {}", self.config.job_id);
|
||||
|
||||
// Fetch job description from orchestrator
|
||||
let job = self.client.get_job().await?;
|
||||
|
||||
tracing::info!(
|
||||
"Received job: {} - {}",
|
||||
job.title,
|
||||
truncate_for_preview(&job.description, 100)
|
||||
);
|
||||
|
||||
// Fetch credentials and store them for injection into child processes
|
||||
// via Command::envs() (avoids unsafe std::env::set_var in multi-threaded runtime).
|
||||
let credentials = self.client.fetch_credentials().await?;
|
||||
{
|
||||
let mut env_map = HashMap::new();
|
||||
for cred in &credentials {
|
||||
env_map.insert(cred.env_var.clone(), cred.value.clone());
|
||||
}
|
||||
self.extra_env = Arc::new(env_map);
|
||||
}
|
||||
if !credentials.is_empty() {
|
||||
tracing::info!(
|
||||
"Fetched {} credential(s) for child process injection",
|
||||
credentials.len()
|
||||
);
|
||||
}
|
||||
|
||||
// Report that we're starting
|
||||
self.client
|
||||
.report_status(&StatusUpdate {
|
||||
state: "in_progress".to_string(),
|
||||
message: Some("Worker started, beginning execution".to_string()),
|
||||
iteration: 0,
|
||||
})
|
||||
.await?;
|
||||
|
||||
// Create reasoning engine
|
||||
let reasoning = Reasoning::new(self.llm.clone());
|
||||
|
||||
// Build initial context
|
||||
let mut reason_ctx = ReasoningContext::new().with_job(&job.description);
|
||||
|
||||
reason_ctx.messages.push(ChatMessage::system(format!(
|
||||
r#"You are an autonomous agent running inside a Docker container.
|
||||
|
||||
Job: {}
|
||||
Description: {}
|
||||
|
||||
You have tools for shell commands, file operations, and code editing.
|
||||
Work independently to complete this job. Report when done."#,
|
||||
job.title, job.description
|
||||
)));
|
||||
|
||||
// Load tool definitions
|
||||
reason_ctx.available_tools = self.tools.tool_definitions().await;
|
||||
|
||||
// Shared iteration tracker — read after the loop to report accurate counts.
|
||||
let iteration_tracker = Arc::new(Mutex::new(0u32));
|
||||
|
||||
// Run with timeout using the shared agentic loop
|
||||
let result = tokio::time::timeout(self.config.timeout, async {
|
||||
let delegate = ContainerDelegate {
|
||||
client: self.client.clone(),
|
||||
safety: self.safety.clone(),
|
||||
tools: self.tools.clone(),
|
||||
extra_env: self.extra_env.clone(),
|
||||
last_output: Mutex::new(String::new()),
|
||||
iteration_tracker: iteration_tracker.clone(),
|
||||
};
|
||||
|
||||
let config = AgenticLoopConfig {
|
||||
max_iterations: self.config.max_iterations as usize,
|
||||
enable_tool_intent_nudge: true,
|
||||
max_tool_intent_nudges: 2,
|
||||
};
|
||||
|
||||
crate::agent::agentic_loop::run_agentic_loop(
|
||||
&delegate,
|
||||
&reasoning,
|
||||
&mut reason_ctx,
|
||||
&config,
|
||||
)
|
||||
.await
|
||||
})
|
||||
.await;
|
||||
|
||||
let iterations = *iteration_tracker.lock().await;
|
||||
|
||||
match result {
|
||||
Ok(Ok(LoopOutcome::Response(output))) => {
|
||||
tracing::info!("Worker completed job {} successfully", self.config.job_id);
|
||||
self.post_event(
|
||||
"result",
|
||||
serde_json::json!({
|
||||
"success": true,
|
||||
"message": truncate_for_preview(&output, 2000),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
self.client
|
||||
.report_complete(&CompletionReport {
|
||||
success: true,
|
||||
message: Some(output),
|
||||
iterations,
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
Ok(Ok(LoopOutcome::MaxIterations)) => {
|
||||
let msg = format!("max iterations ({}) exceeded", self.config.max_iterations);
|
||||
tracing::warn!("Worker failed for job {}: {}", self.config.job_id, msg);
|
||||
self.post_event(
|
||||
"result",
|
||||
serde_json::json!({
|
||||
"success": false,
|
||||
"message": format!("Execution failed: {}", msg),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
self.client
|
||||
.report_complete(&CompletionReport {
|
||||
success: false,
|
||||
message: Some(format!("Execution failed: {}", msg)),
|
||||
iterations,
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
Ok(Ok(LoopOutcome::Stopped | LoopOutcome::NeedApproval(_))) => {
|
||||
tracing::info!("Worker for job {} stopped", self.config.job_id);
|
||||
self.client
|
||||
.report_complete(&CompletionReport {
|
||||
success: false,
|
||||
message: Some("Execution stopped".to_string()),
|
||||
iterations,
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
Ok(Err(e)) => {
|
||||
tracing::error!("Worker failed for job {}: {}", self.config.job_id, e);
|
||||
self.post_event(
|
||||
"result",
|
||||
serde_json::json!({
|
||||
"success": false,
|
||||
"message": format!("Execution failed: {}", e),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
self.client
|
||||
.report_complete(&CompletionReport {
|
||||
success: false,
|
||||
message: Some(format!("Execution failed: {}", e)),
|
||||
iterations,
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
Err(_) => {
|
||||
tracing::warn!("Worker timed out for job {}", self.config.job_id);
|
||||
self.post_event(
|
||||
"result",
|
||||
serde_json::json!({
|
||||
"success": false,
|
||||
"message": "Execution timed out",
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
self.client
|
||||
.report_complete(&CompletionReport {
|
||||
success: false,
|
||||
message: Some("Execution timed out".to_string()),
|
||||
iterations,
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Post a job event to the orchestrator (fire-and-forget).
|
||||
async fn post_event(&self, event_type: &str, data: serde_json::Value) {
|
||||
self.client
|
||||
.post_event(&JobEventPayload {
|
||||
event_type: event_type.to_string(),
|
||||
data,
|
||||
})
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Container delegate: implements `LoopDelegate` for the Docker container context.
|
||||
///
|
||||
/// Tools execute sequentially. Events are posted to the orchestrator via HTTP.
|
||||
/// Completion is detected via `llm_signals_completion()`.
|
||||
struct ContainerDelegate {
|
||||
client: Arc<WorkerHttpClient>,
|
||||
safety: Arc<SafetyLayer>,
|
||||
tools: Arc<ToolRegistry>,
|
||||
extra_env: Arc<HashMap<String, String>>,
|
||||
/// Tracks the last successful tool output for the final response.
|
||||
last_output: Mutex<String>,
|
||||
/// Tracks the current iteration — shared with the outer `run` method so
|
||||
/// `CompletionReport` can include accurate iteration counts.
|
||||
iteration_tracker: Arc<Mutex<u32>>,
|
||||
}
|
||||
|
||||
impl ContainerDelegate {
|
||||
async fn post_event(&self, event_type: &str, data: serde_json::Value) {
|
||||
self.client
|
||||
.post_event(&JobEventPayload {
|
||||
event_type: event_type.to_string(),
|
||||
data,
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
/// Poll the orchestrator for a follow-up prompt. If one is available,
|
||||
/// inject it as a user message into the reasoning context.
|
||||
async fn poll_and_inject_prompt(&self, reason_ctx: &mut ReasoningContext) {
|
||||
match self.client.poll_prompt().await {
|
||||
Ok(Some(prompt)) => {
|
||||
tracing::info!(
|
||||
"Received follow-up prompt: {}",
|
||||
truncate_for_preview(&prompt.content, 100)
|
||||
);
|
||||
self.post_event(
|
||||
"message",
|
||||
serde_json::json!({
|
||||
"role": "user",
|
||||
"content": truncate_for_preview(&prompt.content, 2000),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
reason_ctx.messages.push(ChatMessage::user(&prompt.content));
|
||||
}
|
||||
Ok(None) => {}
|
||||
Err(e) => {
|
||||
tracing::debug!("Failed to poll for prompt: {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl LoopDelegate for ContainerDelegate {
|
||||
async fn check_signals(&self) -> LoopSignal {
|
||||
// Container runtime has no stop signals — the orchestrator manages lifecycle.
|
||||
LoopSignal::Continue
|
||||
}
|
||||
|
||||
async fn before_llm_call(
|
||||
&self,
|
||||
reason_ctx: &mut ReasoningContext,
|
||||
iteration: usize,
|
||||
) -> Option<LoopOutcome> {
|
||||
let iteration = iteration as u32;
|
||||
*self.iteration_tracker.lock().await = iteration;
|
||||
|
||||
// Report progress every 5 iterations
|
||||
if iteration % 5 == 1 {
|
||||
let _ = self
|
||||
.client
|
||||
.report_status(&StatusUpdate {
|
||||
state: "in_progress".to_string(),
|
||||
message: Some(format!("Iteration {}", iteration)),
|
||||
iteration,
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
// Poll for follow-up prompts from the user
|
||||
self.poll_and_inject_prompt(reason_ctx).await;
|
||||
|
||||
// Refresh tools (in case WASM tools were built)
|
||||
reason_ctx.available_tools = self.tools.tool_definitions().await;
|
||||
|
||||
None
|
||||
}
|
||||
|
||||
async fn call_llm(
|
||||
&self,
|
||||
reasoning: &Reasoning,
|
||||
reason_ctx: &mut ReasoningContext,
|
||||
_iteration: usize,
|
||||
) -> Result<crate::llm::RespondOutput, crate::error::Error> {
|
||||
// Container uses respond_with_tools (which may return either text or tool calls)
|
||||
reasoning
|
||||
.respond_with_tools(reason_ctx)
|
||||
.await
|
||||
.map_err(Into::into)
|
||||
}
|
||||
|
||||
async fn handle_text_response(
|
||||
&self,
|
||||
text: &str,
|
||||
reason_ctx: &mut ReasoningContext,
|
||||
) -> TextAction {
|
||||
self.post_event(
|
||||
"message",
|
||||
serde_json::json!({
|
||||
"role": "assistant",
|
||||
"content": truncate_for_preview(text, 2000),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
// Check for completion
|
||||
if crate::util::llm_signals_completion(text) {
|
||||
let last = self.last_output.lock().await;
|
||||
let output = if last.is_empty() {
|
||||
text.to_string()
|
||||
} else {
|
||||
last.clone()
|
||||
};
|
||||
return TextAction::Return(LoopOutcome::Response(output));
|
||||
}
|
||||
|
||||
reason_ctx.messages.push(ChatMessage::assistant(text));
|
||||
TextAction::Continue
|
||||
}
|
||||
|
||||
async fn execute_tool_calls(
|
||||
&self,
|
||||
tool_calls: Vec<crate::llm::ToolCall>,
|
||||
content: Option<String>,
|
||||
reason_ctx: &mut ReasoningContext,
|
||||
) -> Result<Option<LoopOutcome>, crate::error::Error> {
|
||||
if let Some(ref text) = content {
|
||||
self.post_event(
|
||||
"message",
|
||||
serde_json::json!({
|
||||
"role": "assistant",
|
||||
"content": truncate_for_preview(text, 2000),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
// Add assistant message with tool_calls (OpenAI protocol)
|
||||
reason_ctx
|
||||
.messages
|
||||
.push(ChatMessage::assistant_with_tool_calls(
|
||||
content,
|
||||
tool_calls.clone(),
|
||||
));
|
||||
|
||||
// Execute tools sequentially (container context — no parallel execution)
|
||||
for tc in tool_calls {
|
||||
self.post_event(
|
||||
"tool_use",
|
||||
serde_json::json!({
|
||||
"tool_name": tc.name,
|
||||
"input": truncate_for_preview(&tc.arguments.to_string(), 500),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
let job_ctx = JobContext {
|
||||
extra_env: self.extra_env.clone(),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let result =
|
||||
execute_tool_simple(&self.tools, &self.safety, &tc.name, &tc.arguments, &job_ctx)
|
||||
.await;
|
||||
|
||||
self.post_event(
|
||||
"tool_result",
|
||||
serde_json::json!({
|
||||
"tool_name": tc.name,
|
||||
"output": match &result {
|
||||
Ok(output) => truncate_for_preview(output, 2000),
|
||||
Err(e) => format!("Error: {}", truncate_for_preview(e, 500)),
|
||||
},
|
||||
"success": result.is_ok(),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
if let Ok(ref output) = result {
|
||||
*self.last_output.lock().await = output.clone();
|
||||
}
|
||||
|
||||
// Use shared result processing
|
||||
let (_, message) = process_tool_result(&self.safety, &tc.name, &tc.id, &result);
|
||||
reason_ctx.messages.push(message);
|
||||
}
|
||||
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
async fn on_tool_intent_nudge(&self, text: &str, _reason_ctx: &mut ReasoningContext) {
|
||||
self.post_event(
|
||||
"message",
|
||||
serde_json::json!({
|
||||
"role": "assistant",
|
||||
"content": truncate_for_preview(text, 2000),
|
||||
"nudge": true,
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
async fn after_iteration(&self, _iteration: usize) {
|
||||
// Brief pause between iterations
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::agent::agentic_loop::truncate_for_preview;
|
||||
|
||||
#[test]
|
||||
fn test_truncate_within_limit() {
|
||||
assert_eq!(truncate_for_preview("hello", 10), "hello");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_truncate_at_limit() {
|
||||
assert_eq!(truncate_for_preview("hello", 5), "hello");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_truncate_beyond_limit() {
|
||||
let result = truncate_for_preview("hello world", 5);
|
||||
assert_eq!(result, "hello...");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_truncate_multibyte_safe() {
|
||||
// "é" is 2 bytes in UTF-8; slicing at byte 1 would panic without safety
|
||||
let result = truncate_for_preview("é is fancy", 1);
|
||||
// Should truncate to 0 chars (can't fit "é" in 1 byte)
|
||||
assert_eq!(result, "...");
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
+5
-3
@@ -26,13 +26,15 @@
|
||||
|
||||
pub mod api;
|
||||
pub mod claude_bridge;
|
||||
pub mod container;
|
||||
pub mod job;
|
||||
pub mod proxy_llm;
|
||||
pub mod runtime;
|
||||
|
||||
pub use api::WorkerHttpClient;
|
||||
pub use claude_bridge::ClaudeBridgeRuntime;
|
||||
pub use container::WorkerRuntime;
|
||||
pub use job::{Worker, WorkerDeps};
|
||||
pub use proxy_llm::ProxyLlmProvider;
|
||||
pub use runtime::WorkerRuntime;
|
||||
|
||||
/// Run the Worker subcommand (inside Docker containers).
|
||||
pub async fn run_worker(
|
||||
@@ -46,7 +48,7 @@ pub async fn run_worker(
|
||||
orchestrator_url
|
||||
);
|
||||
|
||||
let config = runtime::WorkerConfig {
|
||||
let config = container::WorkerConfig {
|
||||
job_id,
|
||||
orchestrator_url: orchestrator_url.to_string(),
|
||||
max_iterations,
|
||||
|
||||
@@ -1,570 +0,0 @@
|
||||
//! Worker runtime: the main execution loop inside a container.
|
||||
//!
|
||||
//! Reuses the existing `Reasoning` and `SafetyLayer` infrastructure but
|
||||
//! connects to the orchestrator for LLM calls instead of calling APIs directly.
|
||||
//! Streams real-time events (message, tool_use, tool_result, result) through
|
||||
//! the orchestrator's job event pipeline for UI visibility.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::config::SafetyConfig;
|
||||
use crate::context::JobContext;
|
||||
use crate::error::WorkerError;
|
||||
use crate::llm::{
|
||||
ChatMessage, LlmProvider, Reasoning, ReasoningContext, RespondResult, ToolSelection,
|
||||
};
|
||||
use crate::safety::SafetyLayer;
|
||||
use crate::tools::ToolRegistry;
|
||||
use crate::worker::api::{CompletionReport, JobEventPayload, StatusUpdate, WorkerHttpClient};
|
||||
use crate::worker::proxy_llm::ProxyLlmProvider;
|
||||
|
||||
/// Configuration for the worker runtime.
|
||||
pub struct WorkerConfig {
|
||||
pub job_id: Uuid,
|
||||
pub orchestrator_url: String,
|
||||
pub max_iterations: u32,
|
||||
pub timeout: Duration,
|
||||
}
|
||||
|
||||
impl Default for WorkerConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
job_id: Uuid::nil(),
|
||||
orchestrator_url: String::new(),
|
||||
max_iterations: 50,
|
||||
timeout: Duration::from_secs(600),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The worker runtime runs inside a Docker container.
|
||||
///
|
||||
/// It connects to the orchestrator over HTTP, fetches its job description,
|
||||
/// then runs a tool execution loop until the job is complete. Events are
|
||||
/// streamed to the orchestrator so the UI can show real-time progress.
|
||||
pub struct WorkerRuntime {
|
||||
config: WorkerConfig,
|
||||
client: Arc<WorkerHttpClient>,
|
||||
llm: Arc<dyn LlmProvider>,
|
||||
safety: Arc<SafetyLayer>,
|
||||
tools: Arc<ToolRegistry>,
|
||||
/// Credentials fetched from the orchestrator, injected into child processes
|
||||
/// via `Command::envs()` rather than mutating the global process environment.
|
||||
///
|
||||
/// Wrapped in `Arc` to avoid deep-cloning the map on every tool invocation.
|
||||
extra_env: Arc<HashMap<String, String>>,
|
||||
}
|
||||
|
||||
impl WorkerRuntime {
|
||||
/// Create a new worker runtime.
|
||||
///
|
||||
/// Reads `IRONCLAW_WORKER_TOKEN` from the environment for auth.
|
||||
pub fn new(config: WorkerConfig) -> Result<Self, WorkerError> {
|
||||
let client = Arc::new(WorkerHttpClient::from_env(
|
||||
config.orchestrator_url.clone(),
|
||||
config.job_id,
|
||||
)?);
|
||||
|
||||
let llm: Arc<dyn LlmProvider> = Arc::new(ProxyLlmProvider::new(
|
||||
Arc::clone(&client),
|
||||
"proxied".to_string(),
|
||||
));
|
||||
|
||||
let safety = Arc::new(SafetyLayer::new(&SafetyConfig {
|
||||
max_output_length: 100_000,
|
||||
injection_check_enabled: true,
|
||||
}));
|
||||
|
||||
let tools = Arc::new(ToolRegistry::new());
|
||||
// Register only container-safe tools
|
||||
tools.register_container_tools();
|
||||
|
||||
Ok(Self {
|
||||
config,
|
||||
client,
|
||||
llm,
|
||||
safety,
|
||||
tools,
|
||||
extra_env: Arc::new(HashMap::new()),
|
||||
})
|
||||
}
|
||||
|
||||
/// Run the worker until the job is complete or an error occurs.
|
||||
pub async fn run(mut self) -> Result<(), WorkerError> {
|
||||
tracing::info!("Worker starting for job {}", self.config.job_id);
|
||||
|
||||
// Fetch job description from orchestrator
|
||||
let job = self.client.get_job().await?;
|
||||
|
||||
tracing::info!(
|
||||
"Received job: {} - {}",
|
||||
job.title,
|
||||
truncate(&job.description, 100)
|
||||
);
|
||||
|
||||
// Fetch credentials and store them for injection into child processes
|
||||
// via Command::envs() (avoids unsafe std::env::set_var in multi-threaded runtime).
|
||||
let credentials = self.client.fetch_credentials().await?;
|
||||
{
|
||||
let mut env_map = HashMap::new();
|
||||
for cred in &credentials {
|
||||
env_map.insert(cred.env_var.clone(), cred.value.clone());
|
||||
}
|
||||
self.extra_env = Arc::new(env_map);
|
||||
}
|
||||
if !credentials.is_empty() {
|
||||
tracing::info!(
|
||||
"Fetched {} credential(s) for child process injection",
|
||||
credentials.len()
|
||||
);
|
||||
}
|
||||
|
||||
// Report that we're starting
|
||||
self.client
|
||||
.report_status(&StatusUpdate {
|
||||
state: "in_progress".to_string(),
|
||||
message: Some("Worker started, beginning execution".to_string()),
|
||||
iteration: 0,
|
||||
})
|
||||
.await?;
|
||||
|
||||
// Create reasoning engine
|
||||
let reasoning = Reasoning::new(self.llm.clone())
|
||||
.with_model_name(self.llm.active_model_name());
|
||||
|
||||
// Build initial context
|
||||
let mut reason_ctx = ReasoningContext::new().with_job(&job.description);
|
||||
|
||||
reason_ctx.messages.push(ChatMessage::system(format!(
|
||||
r#"You are an autonomous agent running inside a Docker container.
|
||||
|
||||
Job: {}
|
||||
Description: {}
|
||||
|
||||
You have tools for shell commands, file operations, and code editing.
|
||||
Work independently to complete this job. Report when done."#,
|
||||
job.title, job.description
|
||||
)));
|
||||
|
||||
// Run with timeout
|
||||
let result = tokio::time::timeout(self.config.timeout, async {
|
||||
self.execution_loop(&reasoning, &mut reason_ctx).await
|
||||
})
|
||||
.await;
|
||||
|
||||
match result {
|
||||
Ok(Ok(output)) => {
|
||||
tracing::info!("Worker completed job {} successfully", self.config.job_id);
|
||||
self.post_event(
|
||||
"result",
|
||||
serde_json::json!({
|
||||
"success": true,
|
||||
"message": truncate(&output, 2000),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
self.client
|
||||
.report_complete(&CompletionReport {
|
||||
success: true,
|
||||
message: Some(output),
|
||||
iterations: 0,
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
Ok(Err(e)) => {
|
||||
tracing::error!("Worker failed for job {}: {}", self.config.job_id, e);
|
||||
self.post_event(
|
||||
"result",
|
||||
serde_json::json!({
|
||||
"success": false,
|
||||
"message": format!("Execution failed: {}", e),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
self.client
|
||||
.report_complete(&CompletionReport {
|
||||
success: false,
|
||||
message: Some(format!("Execution failed: {}", e)),
|
||||
iterations: 0,
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
Err(_) => {
|
||||
tracing::warn!("Worker timed out for job {}", self.config.job_id);
|
||||
self.post_event(
|
||||
"result",
|
||||
serde_json::json!({
|
||||
"success": false,
|
||||
"message": "Execution timed out",
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
self.client
|
||||
.report_complete(&CompletionReport {
|
||||
success: false,
|
||||
message: Some("Execution timed out".to_string()),
|
||||
iterations: 0,
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn execution_loop(
|
||||
&self,
|
||||
reasoning: &Reasoning,
|
||||
reason_ctx: &mut ReasoningContext,
|
||||
) -> Result<String, WorkerError> {
|
||||
let max_iterations = self.config.max_iterations;
|
||||
let mut last_output = String::new();
|
||||
const MAX_TOOL_INTENT_NUDGES: u32 = 2;
|
||||
let mut consecutive_tool_intent_nudges: u32 = 0;
|
||||
|
||||
// Load tool definitions
|
||||
reason_ctx.available_tools = self.tools.tool_definitions().await;
|
||||
|
||||
for iteration in 1..=max_iterations {
|
||||
// Report progress
|
||||
if iteration % 5 == 1 {
|
||||
let _ = self
|
||||
.client
|
||||
.report_status(&StatusUpdate {
|
||||
state: "in_progress".to_string(),
|
||||
message: Some(format!("Iteration {}", iteration)),
|
||||
iteration,
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
// Poll for follow-up prompts from the user
|
||||
self.poll_and_inject_prompt(reason_ctx).await;
|
||||
|
||||
// Refresh tools (in case WASM tools were built)
|
||||
reason_ctx.available_tools = self.tools.tool_definitions().await;
|
||||
|
||||
// Ask the LLM what to do next
|
||||
let selections = reasoning.select_tools(reason_ctx).await.map_err(|e| {
|
||||
WorkerError::ExecutionFailed {
|
||||
reason: format!("tool selection failed: {}", e),
|
||||
}
|
||||
})?;
|
||||
|
||||
if selections.is_empty() {
|
||||
// No tools selected, try direct response
|
||||
let respond_result =
|
||||
reasoning
|
||||
.respond_with_tools(reason_ctx)
|
||||
.await
|
||||
.map_err(|e| WorkerError::ExecutionFailed {
|
||||
reason: format!("respond_with_tools failed: {}", e),
|
||||
})?;
|
||||
|
||||
match respond_result.result {
|
||||
RespondResult::Text(response) => {
|
||||
self.post_event(
|
||||
"message",
|
||||
serde_json::json!({
|
||||
"role": "assistant",
|
||||
"content": truncate(&response, 2000),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
if crate::util::llm_signals_completion(&response) {
|
||||
if last_output.is_empty() {
|
||||
last_output = response.clone();
|
||||
}
|
||||
return Ok(last_output);
|
||||
}
|
||||
reason_ctx.messages.push(ChatMessage::assistant(&response));
|
||||
|
||||
// Nudge the LLM if it expressed tool intent without calling tools
|
||||
let signals_intent = !reason_ctx.available_tools.is_empty()
|
||||
&& crate::llm::llm_signals_tool_intent(&response);
|
||||
if signals_intent && consecutive_tool_intent_nudges < MAX_TOOL_INTENT_NUDGES
|
||||
{
|
||||
consecutive_tool_intent_nudges += 1;
|
||||
tracing::info!(
|
||||
"LLM expressed tool intent without calling a tool, nudging"
|
||||
);
|
||||
reason_ctx
|
||||
.messages
|
||||
.push(ChatMessage::user(crate::llm::TOOL_INTENT_NUDGE));
|
||||
} else if !signals_intent {
|
||||
consecutive_tool_intent_nudges = 0;
|
||||
}
|
||||
}
|
||||
RespondResult::ToolCalls {
|
||||
tool_calls,
|
||||
content,
|
||||
} => {
|
||||
consecutive_tool_intent_nudges = 0;
|
||||
if let Some(ref text) = content {
|
||||
self.post_event(
|
||||
"message",
|
||||
serde_json::json!({
|
||||
"role": "assistant",
|
||||
"content": truncate(text, 2000),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
// Add assistant message with tool_calls (OpenAI protocol)
|
||||
reason_ctx
|
||||
.messages
|
||||
.push(ChatMessage::assistant_with_tool_calls(
|
||||
content,
|
||||
tool_calls.clone(),
|
||||
));
|
||||
|
||||
for tc in tool_calls {
|
||||
self.post_event(
|
||||
"tool_use",
|
||||
serde_json::json!({
|
||||
"tool_name": tc.name,
|
||||
"input": truncate(&tc.arguments.to_string(), 500),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
let result = self.execute_tool(&tc.name, &tc.arguments).await;
|
||||
|
||||
self.post_event(
|
||||
"tool_result",
|
||||
serde_json::json!({
|
||||
"tool_name": tc.name,
|
||||
"output": match &result {
|
||||
Ok(output) => truncate(output, 2000),
|
||||
Err(e) => format!("Error: {}", truncate(e, 500)),
|
||||
},
|
||||
"success": result.is_ok(),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
if let Ok(ref output) = result {
|
||||
last_output = output.clone();
|
||||
}
|
||||
let selection = ToolSelection {
|
||||
tool_name: tc.name.clone(),
|
||||
parameters: tc.arguments.clone(),
|
||||
reasoning: String::new(),
|
||||
alternatives: vec![],
|
||||
tool_call_id: tc.id.clone(),
|
||||
};
|
||||
self.process_result(reason_ctx, &selection, result);
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
consecutive_tool_intent_nudges = 0;
|
||||
// Execute selected tools
|
||||
for selection in &selections {
|
||||
self.post_event(
|
||||
"tool_use",
|
||||
serde_json::json!({
|
||||
"tool_name": selection.tool_name,
|
||||
"input": truncate(&selection.parameters.to_string(), 500),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
let result = self
|
||||
.execute_tool(&selection.tool_name, &selection.parameters)
|
||||
.await;
|
||||
|
||||
self.post_event(
|
||||
"tool_result",
|
||||
serde_json::json!({
|
||||
"tool_name": selection.tool_name,
|
||||
"output": match &result {
|
||||
Ok(output) => truncate(output, 2000),
|
||||
Err(e) => format!("Error: {}", truncate(e, 500)),
|
||||
},
|
||||
"success": result.is_ok(),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
if let Ok(ref output) = result {
|
||||
last_output = output.clone();
|
||||
}
|
||||
|
||||
let completed = self.process_result(reason_ctx, selection, result);
|
||||
if completed {
|
||||
return Ok(last_output);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Brief pause between iterations
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
|
||||
Err(WorkerError::ExecutionFailed {
|
||||
reason: format!("max iterations ({}) exceeded", max_iterations),
|
||||
})
|
||||
}
|
||||
|
||||
async fn execute_tool(
|
||||
&self,
|
||||
tool_name: &str,
|
||||
params: &serde_json::Value,
|
||||
) -> Result<String, String> {
|
||||
let tool = match self.tools.get(tool_name).await {
|
||||
Some(t) => t,
|
||||
None => return Err(format!("tool '{}' not found", tool_name)),
|
||||
};
|
||||
|
||||
let ctx = JobContext {
|
||||
extra_env: self.extra_env.clone(),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
// Validate params
|
||||
let validation = self.safety.validator().validate_tool_params(params);
|
||||
if !validation.is_valid {
|
||||
let details = validation
|
||||
.errors
|
||||
.iter()
|
||||
.map(|e| format!("{}: {}", e.field, e.message))
|
||||
.collect::<Vec<_>>()
|
||||
.join("; ");
|
||||
return Err(format!("invalid parameters: {}", details));
|
||||
}
|
||||
|
||||
// Execute with per-tool timeout
|
||||
let tool_timeout = tool.execution_timeout();
|
||||
let result = tokio::time::timeout(tool_timeout, tool.execute(params.clone(), &ctx)).await;
|
||||
|
||||
match result {
|
||||
Ok(Ok(output)) => serde_json::to_string_pretty(&output.result)
|
||||
.map_err(|e| format!("serialization error: {}", e)),
|
||||
Ok(Err(e)) => Err(e.to_string()),
|
||||
Err(_) => Err("tool execution timed out".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Process a tool result into the reasoning context. Returns true if the job is complete.
|
||||
fn process_result(
|
||||
&self,
|
||||
reason_ctx: &mut ReasoningContext,
|
||||
selection: &ToolSelection,
|
||||
result: Result<String, String>,
|
||||
) -> bool {
|
||||
match result {
|
||||
Ok(output) => {
|
||||
let sanitized = self
|
||||
.safety
|
||||
.sanitize_tool_output(&selection.tool_name, &output);
|
||||
let wrapped = self.safety.wrap_for_llm(
|
||||
&selection.tool_name,
|
||||
&sanitized.content,
|
||||
sanitized.was_modified,
|
||||
);
|
||||
|
||||
reason_ctx.messages.push(ChatMessage::tool_result(
|
||||
&selection.tool_call_id,
|
||||
&selection.tool_name,
|
||||
wrapped,
|
||||
));
|
||||
|
||||
// Tool output should never signal job completion. Only the LLM's
|
||||
// natural language response should decide when a job is done. A
|
||||
// tool could return text containing "TASK_COMPLETE" in its output
|
||||
// (e.g. from file contents) and trigger a false positive.
|
||||
false
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("Tool {} failed: {}", selection.tool_name, e);
|
||||
reason_ctx.messages.push(ChatMessage::tool_result(
|
||||
&selection.tool_call_id,
|
||||
&selection.tool_name,
|
||||
format!("Error: {}", e),
|
||||
));
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Post a job event to the orchestrator (fire-and-forget).
|
||||
async fn post_event(&self, event_type: &str, data: serde_json::Value) {
|
||||
self.client
|
||||
.post_event(&JobEventPayload {
|
||||
event_type: event_type.to_string(),
|
||||
data,
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
/// Poll the orchestrator for a follow-up prompt. If one is available,
|
||||
/// inject it as a user message into the reasoning context.
|
||||
async fn poll_and_inject_prompt(&self, reason_ctx: &mut ReasoningContext) {
|
||||
match self.client.poll_prompt().await {
|
||||
Ok(Some(prompt)) => {
|
||||
tracing::info!(
|
||||
"Received follow-up prompt: {}",
|
||||
truncate(&prompt.content, 100)
|
||||
);
|
||||
self.post_event(
|
||||
"message",
|
||||
serde_json::json!({
|
||||
"role": "user",
|
||||
"content": truncate(&prompt.content, 2000),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
reason_ctx.messages.push(ChatMessage::user(&prompt.content));
|
||||
}
|
||||
Ok(None) => {}
|
||||
Err(e) => {
|
||||
tracing::debug!("Failed to poll for prompt: {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn truncate(s: &str, max: usize) -> String {
|
||||
if s.len() <= max {
|
||||
s.to_string()
|
||||
} else {
|
||||
let end = crate::util::floor_char_boundary(s, max);
|
||||
format!("{}...", &s[..end])
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::worker::runtime::truncate;
|
||||
|
||||
#[test]
|
||||
fn test_truncate_within_limit() {
|
||||
assert_eq!(truncate("hello", 10), "hello");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_truncate_at_limit() {
|
||||
assert_eq!(truncate("hello", 5), "hello");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_truncate_beyond_limit() {
|
||||
let result = truncate("hello world", 5);
|
||||
assert_eq!(result, "hello...");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_truncate_multibyte_safe() {
|
||||
// "é" is 2 bytes in UTF-8; slicing at byte 1 would panic without safety
|
||||
let result = truncate("é is fancy", 1);
|
||||
// Should truncate to 0 chars (can't fit "é" in 1 byte)
|
||||
assert_eq!(result, "...");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user