mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-25 14:53:34 +00:00
feat(engine): add Python orchestrator module and host functions
Add the orchestrator infrastructure for replacing the Rust execution loop with versioned Python code. This commit adds the module and host functions without switching over — the existing Rust loop is unchanged. New files: - orchestrator/default.py: v0 Python orchestrator (run_loop + helpers) - executor/orchestrator.rs: host function dispatch, orchestrator loading from Store with version selection, OrchestratorResult parsing Host functions exposed to orchestrator Python via Monty suspension: __llm_complete__, __execute_code_step__ (nested Monty VM), __execute_action__, __check_signals__, __emit_event__, __add_message__, __save_checkpoint__, __transition_to__, __retrieve_docs__, __check_budget__, __get_actions__ Also makes json_to_monty, monty_to_json, monty_to_string pub(crate) in scripting.rs for cross-module use. Design doc: docs/plans/2026-03-25-python-orchestrator.md Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
This commit is contained in:
@@ -0,0 +1,289 @@
|
||||
# Engine v2 Orchestrator (default, v0)
|
||||
#
|
||||
# This is the self-modifiable execution loop. It replaces the Rust
|
||||
# ExecutionLoop::run() with Python that can be patched at runtime
|
||||
# by the self-improvement Mission.
|
||||
#
|
||||
# Host functions (provided by Rust via Monty suspension):
|
||||
# __llm_complete__(messages, actions, config) -> response dict
|
||||
# __execute_code_step__(code, state) -> result dict
|
||||
# __execute_action__(name, params) -> result dict
|
||||
# __check_signals__() -> None | "stop" | {"inject": msg}
|
||||
# __emit_event__(kind, **data) -> None
|
||||
# __add_message__(role, content) -> None
|
||||
# __save_checkpoint__(state, counters) -> None
|
||||
# __transition_to__(state, reason) -> None
|
||||
# __retrieve_docs__(goal, max_docs) -> list of doc dicts
|
||||
# __check_budget__() -> budget dict
|
||||
# __get_actions__() -> list of action dicts
|
||||
#
|
||||
# Context variables (injected by Rust before execution):
|
||||
# context - list of prior messages [{role, content}]
|
||||
# goal - thread goal string
|
||||
# actions - list of available action defs
|
||||
# state - persisted state dict from prior steps
|
||||
# config - thread config dict
|
||||
|
||||
|
||||
def run_loop(context, goal, actions, state, config):
|
||||
"""Main execution loop. Returns an outcome dict."""
|
||||
max_iterations = config.get("max_iterations", 30)
|
||||
max_nudges = config.get("max_tool_intent_nudges", 2)
|
||||
nudge_enabled = config.get("enable_tool_intent_nudge", True)
|
||||
max_consecutive_errors = config.get("max_consecutive_errors", 5)
|
||||
nudge_count = 0
|
||||
consecutive_errors = 0
|
||||
step_count = config.get("step_count", 0)
|
||||
|
||||
for step in range(step_count, max_iterations):
|
||||
# 1. Check signals
|
||||
signal = __check_signals__()
|
||||
if signal == "stop":
|
||||
__transition_to__("completed", "stopped by signal")
|
||||
return {"outcome": "stopped"}
|
||||
if signal and isinstance(signal, dict) and "inject" in signal:
|
||||
__add_message__("user", signal["inject"])
|
||||
|
||||
# 2. Check budget
|
||||
budget = __check_budget__()
|
||||
if budget.get("tokens_remaining", 1) <= 0:
|
||||
__transition_to__("completed", "token budget exhausted")
|
||||
return {"outcome": "completed", "response": "Token budget exhausted."}
|
||||
if budget.get("time_remaining_ms", 1) <= 0:
|
||||
__transition_to__("completed", "time budget exhausted")
|
||||
return {"outcome": "completed", "response": "Time budget exhausted."}
|
||||
if budget.get("usd_remaining") is not None and budget["usd_remaining"] <= 0:
|
||||
__transition_to__("completed", "cost budget exhausted")
|
||||
return {"outcome": "completed", "response": "Cost budget exhausted."}
|
||||
|
||||
# 3. Inject prior knowledge on first step
|
||||
if step == 0:
|
||||
docs = __retrieve_docs__(goal, 5)
|
||||
if docs:
|
||||
knowledge = format_docs(docs)
|
||||
__add_message__("system_append", knowledge)
|
||||
|
||||
# 4. Call LLM
|
||||
__emit_event__("step_started", step=step)
|
||||
response = __llm_complete__(None, actions, None)
|
||||
__emit_event__("step_completed", step=step,
|
||||
input_tokens=response.get("usage", {}).get("input_tokens", 0),
|
||||
output_tokens=response.get("usage", {}).get("output_tokens", 0))
|
||||
|
||||
# 5. Handle response based on type
|
||||
resp_type = response.get("type", "text")
|
||||
|
||||
if resp_type == "text":
|
||||
text = response.get("content", "")
|
||||
__add_message__("assistant", text)
|
||||
|
||||
# Check for FINAL()
|
||||
final = extract_final(text)
|
||||
if final is not None:
|
||||
__transition_to__("completed", "FINAL() in text")
|
||||
return {"outcome": "completed", "response": final}
|
||||
|
||||
# Check for tool intent nudge
|
||||
if nudge_enabled and nudge_count < max_nudges and signals_tool_intent(text):
|
||||
nudge_count += 1
|
||||
__add_message__("user",
|
||||
"You described what you'd do but didn't write code. "
|
||||
"Please write a ```repl code block to execute your plan.")
|
||||
continue
|
||||
|
||||
# Plain text response - done
|
||||
__transition_to__("completed", "text response")
|
||||
return {"outcome": "completed", "response": text}
|
||||
|
||||
elif resp_type == "code":
|
||||
code = response.get("code", "")
|
||||
nudge_count = 0
|
||||
__add_message__("assistant", "```repl\n" + code + "\n```")
|
||||
|
||||
# Execute code in nested Monty VM
|
||||
result = __execute_code_step__(code, state)
|
||||
|
||||
# Update persisted state with results
|
||||
if result.get("return_value") is not None:
|
||||
state["step_" + str(step) + "_return"] = result["return_value"]
|
||||
state["last_return"] = result["return_value"]
|
||||
for r in result.get("action_results", []):
|
||||
state[r.get("action_name", "unknown")] = r.get("output")
|
||||
|
||||
# Format output for next LLM context
|
||||
output = format_output(result)
|
||||
__add_message__("user", output)
|
||||
|
||||
# Check for FINAL() in code output
|
||||
if result.get("final_answer") is not None:
|
||||
__transition_to__("completed", "FINAL() in code")
|
||||
return {"outcome": "completed", "response": result["final_answer"]}
|
||||
|
||||
# Check for approval needed
|
||||
if result.get("need_approval") is not None:
|
||||
approval = result["need_approval"]
|
||||
__save_checkpoint__(state, {
|
||||
"nudge_count": nudge_count,
|
||||
"consecutive_errors": consecutive_errors,
|
||||
})
|
||||
return {
|
||||
"outcome": "need_approval",
|
||||
"action_name": approval.get("action_name", ""),
|
||||
"call_id": approval.get("call_id", ""),
|
||||
"parameters": approval.get("parameters", {}),
|
||||
}
|
||||
|
||||
# Track consecutive errors
|
||||
if result.get("had_error"):
|
||||
consecutive_errors += 1
|
||||
if consecutive_errors >= max_consecutive_errors:
|
||||
__transition_to__("failed", "too many consecutive errors")
|
||||
return {"outcome": "failed",
|
||||
"error": str(max_consecutive_errors) + " consecutive code errors"}
|
||||
else:
|
||||
consecutive_errors = 0
|
||||
|
||||
__save_checkpoint__(state, {
|
||||
"nudge_count": nudge_count,
|
||||
"consecutive_errors": consecutive_errors,
|
||||
})
|
||||
|
||||
elif resp_type == "actions":
|
||||
# Tier 0: structured tool calls
|
||||
nudge_count = 0
|
||||
calls = response.get("calls", [])
|
||||
__add_message__("assistant_actions", str(calls))
|
||||
|
||||
for call in calls:
|
||||
name = call.get("name", "")
|
||||
params = call.get("params", {})
|
||||
call_id = call.get("call_id", "")
|
||||
|
||||
r = __execute_action__(name, params)
|
||||
|
||||
__emit_event__("action_executed" if not r.get("is_error") else "action_failed",
|
||||
action_name=name, call_id=call_id)
|
||||
__add_message__("action_result", str(r.get("output", {})))
|
||||
|
||||
if r.get("need_approval"):
|
||||
__save_checkpoint__(state, {
|
||||
"nudge_count": nudge_count,
|
||||
"consecutive_errors": consecutive_errors,
|
||||
})
|
||||
return {
|
||||
"outcome": "need_approval",
|
||||
"action_name": name,
|
||||
"call_id": call_id,
|
||||
"parameters": params,
|
||||
}
|
||||
|
||||
__save_checkpoint__(state, {
|
||||
"nudge_count": nudge_count,
|
||||
"consecutive_errors": consecutive_errors,
|
||||
})
|
||||
|
||||
# Max iterations reached
|
||||
__transition_to__("completed", "max iterations reached")
|
||||
return {"outcome": "max_iterations"}
|
||||
|
||||
|
||||
# ── Helper functions (self-modifiable glue) ──────────────────
|
||||
|
||||
|
||||
def extract_final(text):
|
||||
"""Extract FINAL() content from text. Returns None if not found."""
|
||||
idx = text.find("FINAL(")
|
||||
if idx < 0:
|
||||
return None
|
||||
after = text[idx + 6:]
|
||||
# Handle triple-quoted strings
|
||||
for q in ['"""', "'''"]:
|
||||
if after.startswith(q):
|
||||
end = after.find(q, len(q))
|
||||
if end >= 0:
|
||||
return after[len(q):end]
|
||||
# Handle single/double quoted strings
|
||||
if after and after[0] in ('"', "'"):
|
||||
quote = after[0]
|
||||
end = after.find(quote, 1)
|
||||
if end >= 0:
|
||||
return after[1:end]
|
||||
# Handle balanced parens
|
||||
depth = 1
|
||||
for i, ch in enumerate(after):
|
||||
if ch == "(":
|
||||
depth += 1
|
||||
elif ch == ")":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return after[:i]
|
||||
return None
|
||||
|
||||
|
||||
def signals_tool_intent(text):
|
||||
"""Check if text describes tool usage without actually executing tools."""
|
||||
lower = text.lower()
|
||||
intent_phrases = ["i will", "i'll", "let me", "i would", "i should",
|
||||
"i can", "i need to", "we should", "we can"]
|
||||
tool_phrases = ["search", "fetch", "call", "run", "execute",
|
||||
"use the", "query", "look up"]
|
||||
has_intent = any(p in lower for p in intent_phrases)
|
||||
has_tool = any(p in lower for p in tool_phrases)
|
||||
return has_intent and has_tool
|
||||
|
||||
|
||||
def format_output(result, max_chars=8000):
|
||||
"""Format code execution result for the next LLM context message."""
|
||||
parts = []
|
||||
|
||||
stdout = result.get("stdout", "")
|
||||
if stdout:
|
||||
parts.append("[stdout]\n" + stdout)
|
||||
|
||||
for r in result.get("action_results", []):
|
||||
name = r.get("action_name", "?")
|
||||
output = str(r.get("output", ""))
|
||||
if r.get("is_error"):
|
||||
parts.append("[" + name + " ERROR] " + output)
|
||||
else:
|
||||
preview = output[:500] + "..." if len(output) > 500 else output
|
||||
parts.append("[" + name + "] " + preview)
|
||||
|
||||
ret = result.get("return_value")
|
||||
if ret is not None:
|
||||
parts.append("[return] " + str(ret))
|
||||
|
||||
text = "\n\n".join(parts)
|
||||
|
||||
# Truncate from the front (keep the tail with most recent results)
|
||||
if len(text) > max_chars:
|
||||
text = "... (truncated) ...\n" + text[-max_chars:]
|
||||
|
||||
# Add hint about state dict if there was an error
|
||||
if result.get("had_error") and state:
|
||||
keys = list(state.keys()) if isinstance(state, dict) else []
|
||||
if keys:
|
||||
text += "\n\n[HINT] Variables don't persist between code blocks. " \
|
||||
"Use the `state` dict. Keys: " + str(keys)
|
||||
|
||||
if not text:
|
||||
text = "[code executed, no output]"
|
||||
|
||||
return text
|
||||
|
||||
|
||||
def format_docs(docs):
|
||||
"""Format memory docs for context injection."""
|
||||
parts = ["## Prior Knowledge (from completed threads)\n"]
|
||||
for doc in docs:
|
||||
label = doc.get("type", "NOTE").upper()
|
||||
content = doc.get("content", "")[:500]
|
||||
truncated = "..." if len(doc.get("content", "")) > 500 else ""
|
||||
parts.append("### [" + label + "] " + doc.get("title", "") +
|
||||
"\n" + content + truncated + "\n")
|
||||
return "\n".join(parts)
|
||||
|
||||
|
||||
# Entry point: call run_loop with injected context variables
|
||||
result = run_loop(context, goal, actions, state, config)
|
||||
FINAL(result)
|
||||
@@ -9,6 +9,7 @@ pub mod compaction;
|
||||
pub mod context;
|
||||
pub mod intent;
|
||||
pub mod loop_engine;
|
||||
pub mod orchestrator;
|
||||
pub mod prompt;
|
||||
pub mod scripting;
|
||||
pub mod structured;
|
||||
|
||||
@@ -0,0 +1,984 @@
|
||||
//! Python orchestrator — the self-modifiable execution loop.
|
||||
//!
|
||||
//! Replaces the Rust `ExecutionLoop::run()` with versioned Python code
|
||||
//! executed via Monty. The orchestrator is the "glue layer" between the
|
||||
//! LLM and tools — tool dispatch, output formatting, state management,
|
||||
//! truncation — all in Python, patchable by the self-improvement Mission.
|
||||
//!
|
||||
//! Host functions exposed to the orchestrator Python:
|
||||
//! - `__llm_complete__` — make an LLM call
|
||||
//! - `__execute_code_step__` — run user CodeAct code in a nested Monty VM
|
||||
//! - `__execute_action__` — execute a single tool action
|
||||
//! - `__check_signals__` — poll for stop/inject signals
|
||||
//! - `__emit_event__` — broadcast a ThreadEvent
|
||||
//! - `__add_message__` — append a message to the thread
|
||||
//! - `__save_checkpoint__` — persist thread state
|
||||
//! - `__transition_to__` — change thread state (validated)
|
||||
//! - `__retrieve_docs__` — query memory docs
|
||||
//! - `__check_budget__` — remaining tokens/time/USD
|
||||
//! - `__get_actions__` — available tool definitions
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
use monty::{
|
||||
ExtFunctionResult, LimitedTracker, MontyObject, MontyRun, NameLookupResult, PrintWriter,
|
||||
ResourceLimits, RunProgress,
|
||||
};
|
||||
use tracing::{debug, warn};
|
||||
|
||||
use crate::capability::lease::LeaseManager;
|
||||
use crate::capability::policy::PolicyEngine;
|
||||
use crate::memory::RetrievalEngine;
|
||||
use crate::runtime::messaging::{SignalReceiver, ThreadOutcome, ThreadSignal};
|
||||
use crate::traits::effect::{EffectExecutor, ThreadExecutionContext};
|
||||
use crate::traits::llm::{LlmBackend, LlmCallConfig};
|
||||
use crate::traits::store::Store;
|
||||
use crate::types::error::EngineError;
|
||||
use crate::types::event::{EventKind, ThreadEvent};
|
||||
use crate::types::message::ThreadMessage;
|
||||
use crate::types::project::ProjectId;
|
||||
use crate::types::step::{StepId, TokenUsage};
|
||||
use crate::types::thread::Thread;
|
||||
|
||||
use super::scripting::{execute_code, json_to_monty, monty_to_json, monty_to_string};
|
||||
|
||||
/// The compiled-in default orchestrator (v0).
|
||||
const DEFAULT_ORCHESTRATOR: &str = include_str!("../../orchestrator/default.py");
|
||||
|
||||
/// Well-known title for orchestrator code in the Store.
|
||||
pub const ORCHESTRATOR_TITLE: &str = "orchestrator:main";
|
||||
|
||||
/// Well-known tag for orchestrator code docs.
|
||||
pub const ORCHESTRATOR_TAG: &str = "orchestrator_code";
|
||||
|
||||
/// Result of running the orchestrator.
|
||||
pub struct OrchestratorResult {
|
||||
/// The thread outcome parsed from the orchestrator's return value.
|
||||
pub outcome: ThreadOutcome,
|
||||
/// Total tokens used by LLM calls within the orchestrator.
|
||||
pub tokens_used: TokenUsage,
|
||||
}
|
||||
|
||||
/// Resource limits for the orchestrator VM.
|
||||
fn orchestrator_limits() -> ResourceLimits {
|
||||
ResourceLimits::new()
|
||||
.max_duration(std::time::Duration::from_secs(300)) // 5 min (longer than user code)
|
||||
.max_allocations(5_000_000)
|
||||
.max_memory(128 * 1024 * 1024) // 128 MB
|
||||
}
|
||||
|
||||
/// Load orchestrator code: runtime version from Store, or compiled-in default.
|
||||
pub async fn load_orchestrator(
|
||||
store: Option<&Arc<dyn Store>>,
|
||||
project_id: ProjectId,
|
||||
) -> (String, u64) {
|
||||
if let Some(store) = store
|
||||
&& let Ok(docs) = store.list_memory_docs(project_id).await
|
||||
&& let Some(doc) = docs
|
||||
.iter()
|
||||
.filter(|d| {
|
||||
d.title == ORCHESTRATOR_TITLE
|
||||
&& d.tags.contains(&ORCHESTRATOR_TAG.to_string())
|
||||
})
|
||||
.max_by_key(|d| {
|
||||
d.metadata
|
||||
.get("version")
|
||||
.and_then(|v| v.as_u64())
|
||||
.unwrap_or(0)
|
||||
})
|
||||
{
|
||||
let version = doc
|
||||
.metadata
|
||||
.get("version")
|
||||
.and_then(|v| v.as_u64())
|
||||
.unwrap_or(1);
|
||||
debug!(version, "loaded runtime orchestrator");
|
||||
return (doc.content.clone(), version);
|
||||
}
|
||||
debug!("using compiled-in default orchestrator (v0)");
|
||||
(DEFAULT_ORCHESTRATOR.to_string(), 0)
|
||||
}
|
||||
|
||||
/// Execute the orchestrator Python code with host function dispatch.
|
||||
///
|
||||
/// This is the core function that replaces `ExecutionLoop::run()`'s inner loop.
|
||||
/// The orchestrator Python calls host functions via Monty's suspension mechanism,
|
||||
/// and this function handles each suspension by delegating to the appropriate
|
||||
/// Rust implementation.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub async fn execute_orchestrator(
|
||||
code: &str,
|
||||
thread: &mut Thread,
|
||||
llm: &Arc<dyn LlmBackend>,
|
||||
effects: &Arc<dyn EffectExecutor>,
|
||||
leases: &Arc<LeaseManager>,
|
||||
policy: &Arc<PolicyEngine>,
|
||||
signal_rx: &mut SignalReceiver,
|
||||
event_tx: Option<&tokio::sync::broadcast::Sender<ThreadEvent>>,
|
||||
retrieval: Option<&RetrievalEngine>,
|
||||
_store: Option<&Arc<dyn Store>>,
|
||||
persisted_state: &serde_json::Value,
|
||||
) -> Result<OrchestratorResult, EngineError> {
|
||||
let mut total_tokens = TokenUsage::default();
|
||||
|
||||
// Build context variables for the orchestrator
|
||||
let (input_names, input_values) = build_orchestrator_inputs(thread, persisted_state);
|
||||
|
||||
// Parse and compile
|
||||
let runner = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
|
||||
MontyRun::new(code.to_string(), "orchestrator.py", input_names)
|
||||
})) {
|
||||
Ok(Ok(runner)) => runner,
|
||||
Ok(Err(e)) => {
|
||||
return Err(EngineError::Effect {
|
||||
reason: format!("Orchestrator parse error: {e}"),
|
||||
});
|
||||
}
|
||||
Err(_) => {
|
||||
return Err(EngineError::Effect {
|
||||
reason: "Monty VM panicked during orchestrator parsing".into(),
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
// Start execution
|
||||
let mut stdout = String::new();
|
||||
let tracker = LimitedTracker::new(orchestrator_limits());
|
||||
|
||||
let run_result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
|
||||
runner.start(input_values, tracker, PrintWriter::Collect(&mut stdout))
|
||||
}));
|
||||
|
||||
let mut progress = match run_result {
|
||||
Ok(Ok(p)) => p,
|
||||
Ok(Err(e)) => {
|
||||
return Err(EngineError::Effect {
|
||||
reason: format!("Orchestrator runtime error: {e}"),
|
||||
});
|
||||
}
|
||||
Err(_) => {
|
||||
return Err(EngineError::Effect {
|
||||
reason: "Monty VM panicked during orchestrator start".into(),
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
// Drive the orchestrator dispatch loop
|
||||
let mut final_result: Option<serde_json::Value> = None;
|
||||
|
||||
loop {
|
||||
match progress {
|
||||
RunProgress::Complete(obj) => {
|
||||
// Orchestrator finished without calling FINAL — shouldn't happen
|
||||
// but handle gracefully
|
||||
let result = monty_to_json(&obj);
|
||||
return Ok(OrchestratorResult {
|
||||
outcome: parse_outcome(&result),
|
||||
tokens_used: total_tokens,
|
||||
});
|
||||
}
|
||||
|
||||
RunProgress::FunctionCall(call) => {
|
||||
let action_name = call.function_name.clone();
|
||||
let args = &call.args;
|
||||
let kwargs = &call.kwargs;
|
||||
|
||||
debug!(action = %action_name, "orchestrator: host function call");
|
||||
|
||||
let ext_result = match action_name.as_str() {
|
||||
// FINAL(result) — orchestrator returns its outcome
|
||||
"FINAL" => {
|
||||
let val = args.first().map(monty_to_json).unwrap_or_default();
|
||||
final_result = Some(val);
|
||||
ExtFunctionResult::Return(MontyObject::None)
|
||||
}
|
||||
|
||||
// __llm_complete__(messages, actions, config)
|
||||
"__llm_complete__" => {
|
||||
handle_llm_complete(args, kwargs, thread, llm, effects, leases, &mut total_tokens)
|
||||
.await
|
||||
}
|
||||
|
||||
// __execute_code_step__(code, state)
|
||||
"__execute_code_step__" => {
|
||||
handle_execute_code_step(args, kwargs, thread, llm, effects, leases, policy)
|
||||
.await
|
||||
}
|
||||
|
||||
// __execute_action__(name, params)
|
||||
"__execute_action__" => {
|
||||
handle_execute_action(args, kwargs, thread, effects, leases, policy).await
|
||||
}
|
||||
|
||||
// __check_signals__()
|
||||
"__check_signals__" => handle_check_signals(signal_rx),
|
||||
|
||||
// __emit_event__(kind, **data)
|
||||
"__emit_event__" => handle_emit_event(args, kwargs, thread, event_tx),
|
||||
|
||||
// __add_message__(role, content)
|
||||
"__add_message__" => handle_add_message(args, kwargs, thread),
|
||||
|
||||
// __save_checkpoint__(state, counters)
|
||||
"__save_checkpoint__" => handle_save_checkpoint(args, kwargs, thread),
|
||||
|
||||
// __transition_to__(state, reason)
|
||||
"__transition_to__" => handle_transition_to(args, kwargs, thread),
|
||||
|
||||
// __retrieve_docs__(goal, max_docs)
|
||||
"__retrieve_docs__" => {
|
||||
handle_retrieve_docs(args, kwargs, thread, retrieval).await
|
||||
}
|
||||
|
||||
// __check_budget__()"
|
||||
"__check_budget__" => handle_check_budget(thread),
|
||||
|
||||
// __get_actions__()
|
||||
"__get_actions__" => handle_get_actions(thread, effects, leases).await,
|
||||
|
||||
// Unknown — error
|
||||
other => ExtFunctionResult::Error(monty::MontyException::new(
|
||||
monty::ExcType::NameError,
|
||||
Some(format!("Unknown orchestrator host function: {other}")),
|
||||
)),
|
||||
};
|
||||
|
||||
// Resume the orchestrator VM
|
||||
match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
|
||||
call.resume(ext_result, PrintWriter::Collect(&mut stdout))
|
||||
})) {
|
||||
Ok(Ok(p)) => progress = p,
|
||||
Ok(Err(e)) => {
|
||||
return Err(EngineError::Effect {
|
||||
reason: format!("Orchestrator error after resume: {e}"),
|
||||
});
|
||||
}
|
||||
Err(_) => {
|
||||
return Err(EngineError::Effect {
|
||||
reason: "Monty VM panicked during orchestrator resume".into(),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// If FINAL was called, the VM should complete on next iteration
|
||||
if final_result.is_some() {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
RunProgress::NameLookup(lookup) => {
|
||||
// Undefined variable — resume with NameError
|
||||
let name = lookup.name.clone();
|
||||
debug!(name = %name, "orchestrator: unresolved name");
|
||||
match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
|
||||
lookup.resume(
|
||||
NameLookupResult::Undefined,
|
||||
PrintWriter::Collect(&mut stdout),
|
||||
)
|
||||
})) {
|
||||
Ok(Ok(p)) => progress = p,
|
||||
Ok(Err(e)) => {
|
||||
return Err(EngineError::Effect {
|
||||
reason: format!("Orchestrator NameError '{name}': {e}"),
|
||||
});
|
||||
}
|
||||
Err(_) => {
|
||||
return Err(EngineError::Effect {
|
||||
reason: format!("Monty panic on NameLookup '{name}'"),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
RunProgress::OsCall(_) => {
|
||||
return Err(EngineError::Effect {
|
||||
reason: "Orchestrator attempted OS call (blocked)".into(),
|
||||
});
|
||||
}
|
||||
|
||||
RunProgress::ResolveFutures(_) => {
|
||||
return Err(EngineError::Effect {
|
||||
reason: "Orchestrator attempted async (not supported)".into(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Host function handlers ──────────────────────────────────
|
||||
|
||||
/// Handle `__llm_complete__(messages, actions, config)`.
|
||||
///
|
||||
/// Calls the LLM and returns the response as a dict:
|
||||
/// `{type: "text"|"code"|"actions", content/code/calls: ..., usage: {...}}`
|
||||
async fn handle_llm_complete(
|
||||
_args: &[MontyObject],
|
||||
_kwargs: &[(MontyObject, MontyObject)],
|
||||
thread: &Thread,
|
||||
llm: &Arc<dyn LlmBackend>,
|
||||
effects: &Arc<dyn EffectExecutor>,
|
||||
leases: &Arc<LeaseManager>,
|
||||
total_tokens: &mut TokenUsage,
|
||||
) -> ExtFunctionResult {
|
||||
use crate::types::step::LlmResponse;
|
||||
|
||||
// Build messages from thread (the orchestrator's __add_message__ calls
|
||||
// have already populated thread.messages)
|
||||
let active_leases = leases.active_for_thread(thread.id).await;
|
||||
let actions = effects
|
||||
.available_actions(&active_leases)
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
|
||||
let config = LlmCallConfig {
|
||||
max_tokens: None,
|
||||
temperature: None,
|
||||
force_text: false,
|
||||
depth: thread.config.depth,
|
||||
metadata: HashMap::new(),
|
||||
};
|
||||
|
||||
match llm.complete(&thread.messages, &actions, &config).await {
|
||||
Ok(output) => {
|
||||
total_tokens.input_tokens += output.usage.input_tokens;
|
||||
total_tokens.output_tokens += output.usage.output_tokens;
|
||||
|
||||
let usage = serde_json::json!({
|
||||
"input_tokens": output.usage.input_tokens,
|
||||
"output_tokens": output.usage.output_tokens,
|
||||
});
|
||||
|
||||
let result = match output.response {
|
||||
LlmResponse::Text(text) => {
|
||||
serde_json::json!({"type": "text", "content": text, "usage": usage})
|
||||
}
|
||||
LlmResponse::Code { code, .. } => {
|
||||
serde_json::json!({"type": "code", "code": code, "usage": usage})
|
||||
}
|
||||
LlmResponse::ActionCalls { calls, .. } => {
|
||||
let calls_json: Vec<serde_json::Value> = calls
|
||||
.iter()
|
||||
.map(|c| {
|
||||
serde_json::json!({
|
||||
"name": c.action_name,
|
||||
"call_id": c.id,
|
||||
"params": c.parameters,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
serde_json::json!({"type": "actions", "calls": calls_json, "usage": usage})
|
||||
}
|
||||
};
|
||||
|
||||
ExtFunctionResult::Return(json_to_monty(&result))
|
||||
}
|
||||
Err(e) => ExtFunctionResult::Error(monty::MontyException::new(
|
||||
monty::ExcType::RuntimeError,
|
||||
Some(format!("LLM call failed: {e}")),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle `__execute_code_step__(code, state)`.
|
||||
///
|
||||
/// Runs user CodeAct code in a nested Monty VM with full tool dispatch.
|
||||
/// Returns a dict with stdout, return_value, action_results, etc.
|
||||
async fn handle_execute_code_step(
|
||||
args: &[MontyObject],
|
||||
_kwargs: &[(MontyObject, MontyObject)],
|
||||
thread: &Thread,
|
||||
llm: &Arc<dyn LlmBackend>,
|
||||
effects: &Arc<dyn EffectExecutor>,
|
||||
leases: &Arc<LeaseManager>,
|
||||
policy: &Arc<PolicyEngine>,
|
||||
) -> ExtFunctionResult {
|
||||
let code = match args.first() {
|
||||
Some(obj) => monty_to_string(obj),
|
||||
None => {
|
||||
return ExtFunctionResult::Error(monty::MontyException::new(
|
||||
monty::ExcType::TypeError,
|
||||
Some("__execute_code_step__ requires a code string".into()),
|
||||
))
|
||||
}
|
||||
};
|
||||
|
||||
let state = args
|
||||
.get(1)
|
||||
.map(monty_to_json)
|
||||
.unwrap_or(serde_json::json!({}));
|
||||
|
||||
let exec_ctx = ThreadExecutionContext {
|
||||
thread_id: thread.id,
|
||||
thread_type: thread.thread_type,
|
||||
project_id: thread.project_id,
|
||||
user_id: "orchestrator".into(),
|
||||
step_id: StepId::new(),
|
||||
};
|
||||
|
||||
// Run user code in a nested Monty VM (same pattern as rlm_query)
|
||||
match Box::pin(execute_code(
|
||||
&code, thread, llm, effects, leases, policy, &exec_ctx, &[], &state,
|
||||
))
|
||||
.await
|
||||
{
|
||||
Ok(result) => {
|
||||
let action_results: Vec<serde_json::Value> = result
|
||||
.action_results
|
||||
.iter()
|
||||
.map(|r| {
|
||||
serde_json::json!({
|
||||
"action_name": r.action_name,
|
||||
"output": r.output,
|
||||
"is_error": r.is_error,
|
||||
"duration_ms": r.duration.as_millis(),
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
|
||||
let result_json = serde_json::json!({
|
||||
"return_value": result.return_value,
|
||||
"stdout": result.stdout,
|
||||
"action_results": action_results,
|
||||
"final_answer": result.final_answer,
|
||||
"had_error": result.had_error,
|
||||
"need_approval": result.need_approval.as_ref().map(|na| {
|
||||
match na {
|
||||
ThreadOutcome::NeedApproval { action_name, call_id, parameters } => {
|
||||
serde_json::json!({
|
||||
"action_name": action_name,
|
||||
"call_id": call_id,
|
||||
"parameters": parameters,
|
||||
})
|
||||
}
|
||||
_ => serde_json::Value::Null,
|
||||
}
|
||||
}),
|
||||
});
|
||||
|
||||
ExtFunctionResult::Return(json_to_monty(&result_json))
|
||||
}
|
||||
Err(e) => ExtFunctionResult::Error(monty::MontyException::new(
|
||||
monty::ExcType::RuntimeError,
|
||||
Some(format!("Code execution failed: {e}")),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle `__execute_action__(name, params)`.
|
||||
async fn handle_execute_action(
|
||||
args: &[MontyObject],
|
||||
kwargs: &[(MontyObject, MontyObject)],
|
||||
thread: &Thread,
|
||||
effects: &Arc<dyn EffectExecutor>,
|
||||
leases: &Arc<LeaseManager>,
|
||||
policy: &Arc<PolicyEngine>,
|
||||
) -> ExtFunctionResult {
|
||||
let name = match extract_string_arg(args, kwargs, "name", 0) {
|
||||
Some(n) => n,
|
||||
None => {
|
||||
return ExtFunctionResult::Error(monty::MontyException::new(
|
||||
monty::ExcType::TypeError,
|
||||
Some("__execute_action__ requires a name argument".into()),
|
||||
))
|
||||
}
|
||||
};
|
||||
|
||||
let params = args
|
||||
.get(1)
|
||||
.map(monty_to_json)
|
||||
.unwrap_or(serde_json::json!({}));
|
||||
|
||||
let exec_ctx = ThreadExecutionContext {
|
||||
thread_id: thread.id,
|
||||
thread_type: thread.thread_type,
|
||||
project_id: thread.project_id,
|
||||
user_id: "orchestrator".into(),
|
||||
step_id: StepId::new(),
|
||||
};
|
||||
|
||||
// Find lease for this action
|
||||
let lease = match leases.find_lease_for_action(thread.id, &name).await {
|
||||
Some(l) => l,
|
||||
None => {
|
||||
let result = serde_json::json!({
|
||||
"output": {"error": format!("No lease for action '{name}'")},
|
||||
"is_error": true,
|
||||
});
|
||||
return ExtFunctionResult::Return(json_to_monty(&result));
|
||||
}
|
||||
};
|
||||
|
||||
// Check policy
|
||||
let action_def = effects
|
||||
.available_actions(std::slice::from_ref(&lease))
|
||||
.await
|
||||
.ok()
|
||||
.and_then(|actions| actions.into_iter().find(|a| a.name == name));
|
||||
|
||||
if let Some(ref ad) = action_def {
|
||||
match policy.evaluate(ad, &lease, &[]) {
|
||||
crate::capability::policy::PolicyDecision::Deny { reason } => {
|
||||
let result = serde_json::json!({
|
||||
"output": {"error": format!("Denied: {reason}")},
|
||||
"is_error": true,
|
||||
});
|
||||
return ExtFunctionResult::Return(json_to_monty(&result));
|
||||
}
|
||||
crate::capability::policy::PolicyDecision::RequireApproval { .. } => {
|
||||
let result = serde_json::json!({
|
||||
"need_approval": true,
|
||||
"action_name": name,
|
||||
});
|
||||
return ExtFunctionResult::Return(json_to_monty(&result));
|
||||
}
|
||||
crate::capability::policy::PolicyDecision::Allow => {}
|
||||
}
|
||||
}
|
||||
|
||||
// Execute
|
||||
match effects
|
||||
.execute_action(&name, params, &lease, &exec_ctx)
|
||||
.await
|
||||
{
|
||||
Ok(r) => {
|
||||
let result = serde_json::json!({
|
||||
"action_name": r.action_name,
|
||||
"output": r.output,
|
||||
"is_error": r.is_error,
|
||||
"duration_ms": r.duration.as_millis(),
|
||||
});
|
||||
ExtFunctionResult::Return(json_to_monty(&result))
|
||||
}
|
||||
Err(e) => {
|
||||
let result = serde_json::json!({
|
||||
"output": {"error": e.to_string()},
|
||||
"is_error": true,
|
||||
});
|
||||
ExtFunctionResult::Return(json_to_monty(&result))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle `__check_signals__()`.
|
||||
fn handle_check_signals(signal_rx: &mut SignalReceiver) -> ExtFunctionResult {
|
||||
match signal_rx.try_recv() {
|
||||
Ok(ThreadSignal::Stop) | Ok(ThreadSignal::Suspend) => {
|
||||
ExtFunctionResult::Return(MontyObject::String("stop".into()))
|
||||
}
|
||||
Ok(ThreadSignal::InjectMessage(msg)) => {
|
||||
let result = serde_json::json!({"inject": msg.content});
|
||||
ExtFunctionResult::Return(json_to_monty(&result))
|
||||
}
|
||||
Ok(ThreadSignal::Resume) | Ok(ThreadSignal::ChildCompleted { .. }) => {
|
||||
ExtFunctionResult::Return(MontyObject::None)
|
||||
}
|
||||
Err(_) => ExtFunctionResult::Return(MontyObject::None),
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle `__emit_event__(kind, **data)`.
|
||||
fn handle_emit_event(
|
||||
args: &[MontyObject],
|
||||
kwargs: &[(MontyObject, MontyObject)],
|
||||
thread: &mut Thread,
|
||||
event_tx: Option<&tokio::sync::broadcast::Sender<ThreadEvent>>,
|
||||
) -> ExtFunctionResult {
|
||||
let kind_str = args.first().map(monty_to_string).unwrap_or_default();
|
||||
|
||||
let kind = match kind_str.as_str() {
|
||||
"step_started" => {
|
||||
let _step = extract_u64_kwarg(kwargs, "step").unwrap_or(0);
|
||||
EventKind::StepStarted {
|
||||
step_id: StepId::new(),
|
||||
}
|
||||
}
|
||||
"step_completed" => {
|
||||
let input = extract_u64_kwarg(kwargs, "input_tokens").unwrap_or(0);
|
||||
let output = extract_u64_kwarg(kwargs, "output_tokens").unwrap_or(0);
|
||||
EventKind::StepCompleted {
|
||||
step_id: StepId::new(),
|
||||
tokens: TokenUsage {
|
||||
input_tokens: input,
|
||||
output_tokens: output,
|
||||
..Default::default()
|
||||
},
|
||||
}
|
||||
}
|
||||
"action_executed" => {
|
||||
let action_name = extract_string_kwarg(kwargs, "action_name").unwrap_or_default();
|
||||
let call_id = extract_string_kwarg(kwargs, "call_id").unwrap_or_default();
|
||||
EventKind::ActionExecuted {
|
||||
step_id: StepId::new(),
|
||||
action_name,
|
||||
call_id,
|
||||
duration_ms: 0,
|
||||
}
|
||||
}
|
||||
"action_failed" => {
|
||||
let action_name = extract_string_kwarg(kwargs, "action_name").unwrap_or_default();
|
||||
let call_id = extract_string_kwarg(kwargs, "call_id").unwrap_or_default();
|
||||
let error = extract_string_kwarg(kwargs, "error").unwrap_or_default();
|
||||
EventKind::ActionFailed {
|
||||
step_id: StepId::new(),
|
||||
action_name,
|
||||
call_id,
|
||||
error,
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
debug!(kind = %kind_str, "orchestrator: unknown event kind, skipping");
|
||||
return ExtFunctionResult::Return(MontyObject::None);
|
||||
}
|
||||
};
|
||||
|
||||
let event = ThreadEvent::new(thread.id, kind);
|
||||
if let Some(tx) = event_tx {
|
||||
let _ = tx.send(event.clone());
|
||||
}
|
||||
thread.events.push(event);
|
||||
thread.updated_at = chrono::Utc::now();
|
||||
|
||||
ExtFunctionResult::Return(MontyObject::None)
|
||||
}
|
||||
|
||||
/// Handle `__add_message__(role, content)`.
|
||||
fn handle_add_message(
|
||||
args: &[MontyObject],
|
||||
_kwargs: &[(MontyObject, MontyObject)],
|
||||
thread: &mut Thread,
|
||||
) -> ExtFunctionResult {
|
||||
let role = args.first().map(monty_to_string).unwrap_or_default();
|
||||
let content = args.get(1).map(monty_to_string).unwrap_or_default();
|
||||
|
||||
match role.as_str() {
|
||||
"user" => thread.add_message(ThreadMessage::user(&content)),
|
||||
"assistant" | "assistant_actions" => {
|
||||
thread.add_message(ThreadMessage::assistant(&content))
|
||||
}
|
||||
"system" => thread.add_message(ThreadMessage::system(&content)),
|
||||
"system_append" => {
|
||||
// Append to existing system message (for doc injection)
|
||||
if let Some(msg) = thread
|
||||
.messages
|
||||
.iter_mut()
|
||||
.find(|m| m.role == crate::types::message::MessageRole::System)
|
||||
{
|
||||
msg.content.push_str("\n\n");
|
||||
msg.content.push_str(&content);
|
||||
}
|
||||
}
|
||||
"action_result" => {
|
||||
thread.add_message(ThreadMessage::action_result("", "", &content));
|
||||
}
|
||||
_ => {
|
||||
thread.add_message(ThreadMessage::user(&content));
|
||||
}
|
||||
}
|
||||
|
||||
thread.step_count += 0; // Message addition tracked by thread itself
|
||||
ExtFunctionResult::Return(MontyObject::None)
|
||||
}
|
||||
|
||||
/// Handle `__save_checkpoint__(state, counters)`.
|
||||
fn handle_save_checkpoint(
|
||||
args: &[MontyObject],
|
||||
_kwargs: &[(MontyObject, MontyObject)],
|
||||
thread: &mut Thread,
|
||||
) -> ExtFunctionResult {
|
||||
let state = args
|
||||
.first()
|
||||
.map(monty_to_json)
|
||||
.unwrap_or(serde_json::json!({}));
|
||||
let counters = args
|
||||
.get(1)
|
||||
.map(monty_to_json)
|
||||
.unwrap_or(serde_json::json!({}));
|
||||
|
||||
if let Some(metadata) = thread.metadata.as_object_mut() {
|
||||
metadata.insert(
|
||||
"runtime_checkpoint".into(),
|
||||
serde_json::json!({
|
||||
"persisted_state": state,
|
||||
"nudge_count": counters.get("nudge_count").and_then(|v| v.as_u64()).unwrap_or(0),
|
||||
"consecutive_errors": counters.get("consecutive_errors").and_then(|v| v.as_u64()).unwrap_or(0),
|
||||
"compaction_count": counters.get("compaction_count").and_then(|v| v.as_u64()).unwrap_or(0),
|
||||
}),
|
||||
);
|
||||
}
|
||||
thread.updated_at = chrono::Utc::now();
|
||||
|
||||
ExtFunctionResult::Return(MontyObject::None)
|
||||
}
|
||||
|
||||
/// Handle `__transition_to__(state, reason)`.
|
||||
fn handle_transition_to(
|
||||
args: &[MontyObject],
|
||||
_kwargs: &[(MontyObject, MontyObject)],
|
||||
thread: &mut Thread,
|
||||
) -> ExtFunctionResult {
|
||||
let state_str = args.first().map(monty_to_string).unwrap_or_default();
|
||||
let reason = args.get(1).map(monty_to_string);
|
||||
|
||||
let target = match state_str.as_str() {
|
||||
"running" => crate::types::thread::ThreadState::Running,
|
||||
"completed" => crate::types::thread::ThreadState::Completed,
|
||||
"failed" => crate::types::thread::ThreadState::Failed,
|
||||
"waiting" => crate::types::thread::ThreadState::Waiting,
|
||||
"suspended" => crate::types::thread::ThreadState::Suspended,
|
||||
other => {
|
||||
return ExtFunctionResult::Error(monty::MontyException::new(
|
||||
monty::ExcType::ValueError,
|
||||
Some(format!("Unknown thread state: {other}")),
|
||||
))
|
||||
}
|
||||
};
|
||||
|
||||
match thread.transition_to(target, reason) {
|
||||
Ok(()) => ExtFunctionResult::Return(MontyObject::None),
|
||||
Err(e) => ExtFunctionResult::Error(monty::MontyException::new(
|
||||
monty::ExcType::RuntimeError,
|
||||
Some(format!("State transition failed: {e}")),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle `__retrieve_docs__(goal, max_docs)`.
|
||||
async fn handle_retrieve_docs(
|
||||
args: &[MontyObject],
|
||||
_kwargs: &[(MontyObject, MontyObject)],
|
||||
thread: &Thread,
|
||||
retrieval: Option<&RetrievalEngine>,
|
||||
) -> ExtFunctionResult {
|
||||
let retrieval = match retrieval {
|
||||
Some(r) => r,
|
||||
None => return ExtFunctionResult::Return(json_to_monty(&serde_json::json!([]))),
|
||||
};
|
||||
|
||||
let goal = args.first().map(monty_to_string).unwrap_or_default();
|
||||
let max_docs = args
|
||||
.get(1)
|
||||
.and_then(|v| match v {
|
||||
MontyObject::Int(i) => Some(*i as usize),
|
||||
_ => None,
|
||||
})
|
||||
.unwrap_or(5);
|
||||
|
||||
match retrieval
|
||||
.retrieve_context(thread.project_id, &goal, max_docs)
|
||||
.await
|
||||
{
|
||||
Ok(docs) => {
|
||||
let docs_json: Vec<serde_json::Value> = docs
|
||||
.iter()
|
||||
.map(|d| {
|
||||
serde_json::json!({
|
||||
"type": format!("{:?}", d.doc_type),
|
||||
"title": d.title,
|
||||
"content": d.content,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
ExtFunctionResult::Return(json_to_monty(&serde_json::json!(docs_json)))
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("retrieve_docs failed: {e}");
|
||||
ExtFunctionResult::Return(json_to_monty(&serde_json::json!([])))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle `__check_budget__()`.
|
||||
fn handle_check_budget(thread: &Thread) -> ExtFunctionResult {
|
||||
let tokens_remaining = thread
|
||||
.config
|
||||
.max_tokens_total
|
||||
.map(|max| max.saturating_sub(thread.total_tokens_used))
|
||||
.unwrap_or(u64::MAX);
|
||||
|
||||
let time_remaining_ms = thread
|
||||
.config
|
||||
.max_duration
|
||||
.map(|dur| {
|
||||
let elapsed = chrono::Utc::now()
|
||||
.signed_duration_since(thread.created_at)
|
||||
.num_milliseconds()
|
||||
.max(0) as u64;
|
||||
dur.as_millis() as u64 - elapsed.min(dur.as_millis() as u64)
|
||||
})
|
||||
.unwrap_or(u64::MAX);
|
||||
|
||||
let usd_remaining = thread
|
||||
.config
|
||||
.max_budget_usd
|
||||
.map(|max| (max - thread.total_cost_usd).max(0.0));
|
||||
|
||||
let result = serde_json::json!({
|
||||
"tokens_remaining": tokens_remaining,
|
||||
"time_remaining_ms": time_remaining_ms,
|
||||
"usd_remaining": usd_remaining,
|
||||
});
|
||||
|
||||
ExtFunctionResult::Return(json_to_monty(&result))
|
||||
}
|
||||
|
||||
/// Handle `__get_actions__()`.
|
||||
async fn handle_get_actions(
|
||||
thread: &Thread,
|
||||
effects: &Arc<dyn EffectExecutor>,
|
||||
leases: &Arc<LeaseManager>,
|
||||
) -> ExtFunctionResult {
|
||||
let active_leases = leases.active_for_thread(thread.id).await;
|
||||
match effects.available_actions(&active_leases).await {
|
||||
Ok(actions) => {
|
||||
let actions_json: Vec<serde_json::Value> = actions
|
||||
.iter()
|
||||
.map(|a| {
|
||||
serde_json::json!({
|
||||
"name": a.name,
|
||||
"description": a.description,
|
||||
"params": a.parameters_schema,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
ExtFunctionResult::Return(json_to_monty(&serde_json::json!(actions_json)))
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("get_actions failed: {e}");
|
||||
ExtFunctionResult::Return(json_to_monty(&serde_json::json!([])))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Helpers ─────────────────────────────────────────────────
|
||||
|
||||
/// Build the context variables injected into the orchestrator Python.
|
||||
fn build_orchestrator_inputs(
|
||||
thread: &Thread,
|
||||
persisted_state: &serde_json::Value,
|
||||
) -> (Vec<String>, Vec<MontyObject>) {
|
||||
let names = vec![
|
||||
"context".into(),
|
||||
"goal".into(),
|
||||
"actions".into(),
|
||||
"state".into(),
|
||||
"config".into(),
|
||||
];
|
||||
|
||||
// Build context (message history)
|
||||
let context: Vec<serde_json::Value> = thread
|
||||
.messages
|
||||
.iter()
|
||||
.map(|m| {
|
||||
serde_json::json!({
|
||||
"role": format!("{:?}", m.role),
|
||||
"content": m.content,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
|
||||
// Build config
|
||||
let config = serde_json::json!({
|
||||
"max_iterations": thread.config.max_iterations,
|
||||
"max_tool_intent_nudges": thread.config.max_tool_intent_nudges,
|
||||
"enable_tool_intent_nudge": thread.config.enable_tool_intent_nudge,
|
||||
"max_consecutive_errors": thread.config.max_consecutive_errors,
|
||||
"max_tokens_total": thread.config.max_tokens_total,
|
||||
"max_budget_usd": thread.config.max_budget_usd,
|
||||
"model_context_limit": thread.config.model_context_limit,
|
||||
"enable_compaction": thread.config.enable_compaction,
|
||||
"depth": thread.config.depth,
|
||||
"max_depth": thread.config.max_depth,
|
||||
"step_count": thread.step_count,
|
||||
});
|
||||
|
||||
let values = vec![
|
||||
json_to_monty(&serde_json::json!(context)),
|
||||
MontyObject::String(thread.goal.clone()),
|
||||
json_to_monty(&serde_json::json!([])), // actions loaded dynamically via __get_actions__
|
||||
json_to_monty(persisted_state),
|
||||
json_to_monty(&config),
|
||||
];
|
||||
|
||||
(names, values)
|
||||
}
|
||||
|
||||
/// Parse the orchestrator's return value into a ThreadOutcome.
|
||||
fn parse_outcome(result: &serde_json::Value) -> ThreadOutcome {
|
||||
let outcome = result
|
||||
.get("outcome")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("completed");
|
||||
|
||||
match outcome {
|
||||
"completed" => ThreadOutcome::Completed {
|
||||
response: result.get("response").and_then(|v| v.as_str()).map(String::from),
|
||||
},
|
||||
"stopped" => ThreadOutcome::Stopped,
|
||||
"max_iterations" => ThreadOutcome::MaxIterations,
|
||||
"failed" => ThreadOutcome::Failed {
|
||||
error: result
|
||||
.get("error")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("unknown error")
|
||||
.to_string(),
|
||||
},
|
||||
"need_approval" => ThreadOutcome::NeedApproval {
|
||||
action_name: result
|
||||
.get("action_name")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("")
|
||||
.to_string(),
|
||||
call_id: result
|
||||
.get("call_id")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("")
|
||||
.to_string(),
|
||||
parameters: result
|
||||
.get("parameters")
|
||||
.cloned()
|
||||
.unwrap_or(serde_json::json!({})),
|
||||
},
|
||||
_ => ThreadOutcome::Completed { response: None },
|
||||
}
|
||||
}
|
||||
|
||||
fn extract_string_arg(
|
||||
args: &[MontyObject],
|
||||
kwargs: &[(MontyObject, MontyObject)],
|
||||
name: &str,
|
||||
position: usize,
|
||||
) -> Option<String> {
|
||||
for (k, v) in kwargs {
|
||||
if let MontyObject::String(key) = k
|
||||
&& key == name
|
||||
{
|
||||
return Some(monty_to_string(v));
|
||||
}
|
||||
}
|
||||
args.get(position).map(monty_to_string)
|
||||
}
|
||||
|
||||
fn extract_string_kwarg(kwargs: &[(MontyObject, MontyObject)], name: &str) -> Option<String> {
|
||||
for (k, v) in kwargs {
|
||||
if let MontyObject::String(key) = k
|
||||
&& key == name
|
||||
{
|
||||
return Some(monty_to_string(v));
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
fn extract_u64_kwarg(kwargs: &[(MontyObject, MontyObject)], name: &str) -> Option<u64> {
|
||||
for (k, v) in kwargs {
|
||||
if let MontyObject::String(key) = k
|
||||
&& key == name
|
||||
&& let MontyObject::Int(i) = v
|
||||
{
|
||||
return Some(*i as u64);
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
@@ -859,7 +859,7 @@ fn extract_string_arg(
|
||||
args.get(position).map(monty_to_string)
|
||||
}
|
||||
|
||||
fn monty_to_string(obj: &MontyObject) -> String {
|
||||
pub(crate) fn monty_to_string(obj: &MontyObject) -> String {
|
||||
match obj {
|
||||
MontyObject::String(s) => s.clone(),
|
||||
MontyObject::None => "None".into(),
|
||||
@@ -983,7 +983,7 @@ async fn dispatch_action(
|
||||
|
||||
// ── MontyObject ↔ JSON ──────────────────────────────────────
|
||||
|
||||
fn monty_to_json(obj: &MontyObject) -> serde_json::Value {
|
||||
pub(crate) fn monty_to_json(obj: &MontyObject) -> serde_json::Value {
|
||||
match obj {
|
||||
MontyObject::None => serde_json::Value::Null,
|
||||
MontyObject::Bool(b) => serde_json::Value::Bool(*b),
|
||||
@@ -1017,7 +1017,7 @@ fn monty_to_json(obj: &MontyObject) -> serde_json::Value {
|
||||
}
|
||||
}
|
||||
|
||||
fn json_to_monty(val: &serde_json::Value) -> MontyObject {
|
||||
pub(crate) fn json_to_monty(val: &serde_json::Value) -> MontyObject {
|
||||
match val {
|
||||
serde_json::Value::Null => MontyObject::None,
|
||||
serde_json::Value::Bool(b) => MontyObject::Bool(*b),
|
||||
|
||||
@@ -0,0 +1,511 @@
|
||||
# Python Orchestrator: Move the Engine Loop to CodeAct
|
||||
|
||||
**Date:** 2026-03-25
|
||||
**Status:** Design
|
||||
**Context:** The engine's Rust loop has frequent bugs in the glue layer (tool dispatch, output formatting, state management, truncation). The LLM can't fix Rust at runtime. Moving the loop to Python via CodeAct makes the orchestration layer self-modifiable by the self-improvement Mission.
|
||||
|
||||
---
|
||||
|
||||
## Architecture
|
||||
|
||||
### Before (current)
|
||||
|
||||
```
|
||||
ExecutionLoop::run() [Rust, 900 lines]
|
||||
├── Build system prompt
|
||||
├── for iteration in 0..max:
|
||||
│ ├── Check signals
|
||||
│ ├── Check budgets
|
||||
│ ├── Build context (messages + actions)
|
||||
│ ├── Call LLM
|
||||
│ ├── Match response:
|
||||
│ │ ├── Text → extract FINAL(), check nudge
|
||||
│ │ ├── ActionCalls → execute_action_calls()
|
||||
│ │ └── Code → execute_code() via Monty
|
||||
│ ├── Format output metadata
|
||||
│ ├── Update persisted state
|
||||
│ └── Persist checkpoint
|
||||
└── Return ThreadOutcome
|
||||
```
|
||||
|
||||
### After (proposed)
|
||||
|
||||
```
|
||||
ExecutionLoop::run() [Rust, ~50 lines — bootstrap only]
|
||||
├── Load orchestrator code from Store (versioned MemoryDoc)
|
||||
├── If missing, use compiled-in default
|
||||
├── Set up Monty VM with host functions
|
||||
├── Execute orchestrator Python code
|
||||
└── Return ThreadOutcome from Python's return value
|
||||
|
||||
Host functions [Rust, exposed to Python via Monty suspension]:
|
||||
├── llm_complete(messages, actions, config) → response
|
||||
├── execute_action(name, params) → result (lease + policy + safety)
|
||||
├── check_signals() → signal or None
|
||||
├── save_checkpoint(state) → persist thread/step/events
|
||||
├── emit_event(kind) → broadcast + record
|
||||
├── transition_to(state, reason) → validated state change
|
||||
├── retrieve_docs(goal, max) → memory docs
|
||||
├── get_actions() → available ActionDefs
|
||||
└── check_budget() → remaining tokens/time/usd
|
||||
|
||||
Orchestrator [Python, versioned, self-modifiable]:
|
||||
└── run_loop(context, goal, actions, state, config) → outcome
|
||||
├── Tool dispatch + name resolution
|
||||
├── Output formatting + truncation
|
||||
├── State management (persisted_state dict)
|
||||
├── FINAL() extraction
|
||||
├── Tool intent nudge detection
|
||||
├── Context compaction decisions
|
||||
└── The step loop itself
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Versioned Orchestrator Code
|
||||
|
||||
The orchestrator Python source is stored as a MemoryDoc:
|
||||
|
||||
```
|
||||
DocType: Note
|
||||
Title: "orchestrator:main"
|
||||
Tag: "orchestrator_code"
|
||||
Content: <Python source code>
|
||||
Metadata: {
|
||||
"version": 3,
|
||||
"parent_version": 2,
|
||||
"source_thread_id": "...", // which self-improvement thread created this
|
||||
"created_at": "2026-03-25T10:00:00Z"
|
||||
}
|
||||
```
|
||||
|
||||
### Version lifecycle
|
||||
|
||||
```
|
||||
v0 (compiled-in) → v1 (self-improvement fix) → v2 (another fix) → ...
|
||||
↑
|
||||
auto-rollback if v2 causes
|
||||
3 consecutive thread failures
|
||||
```
|
||||
|
||||
### Operations
|
||||
|
||||
- **Load**: Query Store for `orchestrator:main` docs, pick highest version
|
||||
- **Update**: Self-improvement Mission saves a new version with `parent_version` pointing to current
|
||||
- **Rollback**: On consecutive failures, load the `parent_version` doc instead
|
||||
- **Reset**: Delete all runtime versions, fall back to compiled-in v0
|
||||
|
||||
### Auto-rollback logic
|
||||
|
||||
Tracked per-version in mission metadata or thread config:
|
||||
|
||||
```python
|
||||
# Pseudo-logic in the bootstrap (Rust side)
|
||||
consecutive_failures = count_recent_failures(orchestrator_version)
|
||||
if consecutive_failures >= 3:
|
||||
orchestrator = load_version(parent_version)
|
||||
emit_event(SelfImprovementRollback { from: current, to: parent })
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Host Functions
|
||||
|
||||
These replace direct Rust calls with Python-callable suspension points, using the same mechanism Monty already uses for tool calls.
|
||||
|
||||
### `llm_complete(messages, actions=None, config=None)`
|
||||
|
||||
```python
|
||||
# Python side
|
||||
response = llm_complete(
|
||||
messages=[{"role": "user", "content": "search for AI news"}],
|
||||
actions=get_actions(),
|
||||
config={"force_text": False}
|
||||
)
|
||||
# response = {"type": "text", "content": "..."}
|
||||
# | {"type": "actions", "calls": [...]}
|
||||
# | {"type": "code", "code": "..."}
|
||||
# Also: response["usage"] = {"input_tokens": N, "output_tokens": M}
|
||||
```
|
||||
|
||||
Rust side: calls `LlmBackend::complete()`, converts `LlmOutput` to JSON dict.
|
||||
|
||||
### `execute_action(name, params)`
|
||||
|
||||
```python
|
||||
result = execute_action("web_search", {"query": "AI news", "count": 5})
|
||||
# result = {"output": {...}, "is_error": false, "duration_ms": 123}
|
||||
# Includes: lease check, policy evaluation, safety sanitization, hooks
|
||||
```
|
||||
|
||||
Rust side: full `EffectExecutor::execute_action()` pipeline with all v1 security controls.
|
||||
|
||||
### `check_signals()`
|
||||
|
||||
```python
|
||||
signal = check_signals()
|
||||
# signal = None | "stop" | {"inject": "new message"} | "suspend"
|
||||
```
|
||||
|
||||
Rust side: `signal_rx.try_recv()` on the tokio channel.
|
||||
|
||||
### `save_checkpoint(state, step=None)`
|
||||
|
||||
```python
|
||||
save_checkpoint(state={"last_return": result, "web_search": data})
|
||||
```
|
||||
|
||||
Rust side: serializes to thread metadata, optionally saves Step + events to Store.
|
||||
|
||||
### `emit_event(kind, **kwargs)`
|
||||
|
||||
```python
|
||||
emit_event("action_executed", action_name="web_search", duration_ms=123)
|
||||
emit_event("step_completed", tokens={"input": 500, "output": 200})
|
||||
```
|
||||
|
||||
Rust side: constructs `EventKind` variant, broadcasts + records.
|
||||
|
||||
### `transition_to(state, reason=None)`
|
||||
|
||||
```python
|
||||
transition_to("completed", reason="FINAL() called")
|
||||
# Raises error if transition is invalid (state machine enforcement stays in Rust)
|
||||
```
|
||||
|
||||
### `retrieve_docs(goal, max_docs=5)`
|
||||
|
||||
```python
|
||||
docs = retrieve_docs("search for AI news", max_docs=5)
|
||||
# docs = [{"type": "LESSON", "title": "...", "content": "..."}, ...]
|
||||
```
|
||||
|
||||
### `check_budget()`
|
||||
|
||||
```python
|
||||
budget = check_budget()
|
||||
# budget = {"tokens_remaining": 50000, "time_remaining_ms": 25000, "usd_remaining": 0.45}
|
||||
```
|
||||
|
||||
### `get_actions()`
|
||||
|
||||
```python
|
||||
actions = get_actions()
|
||||
# actions = [{"name": "web_search", "description": "...", "params": {...}}, ...]
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Default Orchestrator (v0)
|
||||
|
||||
The compiled-in Python code that ships with the binary. This is what `include_str!` loads as the seed version. It replicates the current Rust loop logic:
|
||||
|
||||
```python
|
||||
def run_loop(context, goal, actions, state, config):
|
||||
"""Engine v2 orchestrator — the self-modifiable execution loop."""
|
||||
max_iterations = config.get("max_iterations", 30)
|
||||
max_nudges = config.get("max_tool_intent_nudges", 2)
|
||||
nudge_count = 0
|
||||
consecutive_errors = 0
|
||||
|
||||
for step in range(max_iterations):
|
||||
# 1. Check signals
|
||||
signal = check_signals()
|
||||
if signal == "stop":
|
||||
transition_to("completed", "stopped by signal")
|
||||
return {"type": "stopped"}
|
||||
if signal and "inject" in signal:
|
||||
context.append({"role": "user", "content": signal["inject"]})
|
||||
|
||||
# 2. Check budget
|
||||
budget = check_budget()
|
||||
if budget["tokens_remaining"] <= 0:
|
||||
transition_to("completed", "token budget exhausted")
|
||||
return {"type": "completed", "response": "Token budget exhausted."}
|
||||
|
||||
# 3. Build messages for LLM
|
||||
messages = list(context) # copy
|
||||
|
||||
# 4. Inject prior knowledge on first step
|
||||
if step == 0:
|
||||
docs = retrieve_docs(goal)
|
||||
if docs:
|
||||
knowledge = format_docs(docs)
|
||||
if messages and messages[0]["role"] == "system":
|
||||
messages[0]["content"] += "\n\n" + knowledge
|
||||
|
||||
# 5. Call LLM
|
||||
emit_event("step_started")
|
||||
response = llm_complete(messages, actions)
|
||||
emit_event("step_completed", tokens=response["usage"])
|
||||
|
||||
# 6. Handle response
|
||||
if response["type"] == "text":
|
||||
text = response["content"]
|
||||
|
||||
# Check for FINAL()
|
||||
final = extract_final(text)
|
||||
if final is not None:
|
||||
context.append({"role": "assistant", "content": text})
|
||||
transition_to("completed", "FINAL() called")
|
||||
return {"type": "completed", "response": final}
|
||||
|
||||
# Check for tool intent nudge
|
||||
if nudge_count < max_nudges and signals_tool_intent(text):
|
||||
nudge_count += 1
|
||||
context.append({"role": "assistant", "content": text})
|
||||
context.append({"role": "user", "content":
|
||||
"You described what you'd do but didn't write code. "
|
||||
"Please write a ```repl code block to execute your plan."})
|
||||
continue
|
||||
|
||||
# Plain text response — done
|
||||
context.append({"role": "assistant", "content": text})
|
||||
transition_to("completed", "text response")
|
||||
return {"type": "completed", "response": text}
|
||||
|
||||
elif response["type"] == "code":
|
||||
code = response["code"]
|
||||
nudge_count = 0
|
||||
context.append({"role": "assistant", "content": f"```repl\n{code}\n```"})
|
||||
|
||||
# Code is executed by the Monty VM outside this function.
|
||||
# We receive results via state dict after execution.
|
||||
# The host handles code execution and resumes us with results.
|
||||
result = execute_code_step(code, state)
|
||||
|
||||
# Update state with results
|
||||
state[f"step_{step}_return"] = result.get("return_value")
|
||||
state["last_return"] = result.get("return_value")
|
||||
for r in result.get("action_results", []):
|
||||
state[r["action_name"]] = r["output"]
|
||||
|
||||
# Format output for next iteration
|
||||
output = format_output(result)
|
||||
context.append({"role": "user", "content": output})
|
||||
|
||||
# Check for FINAL() in code output
|
||||
if result.get("final_answer") is not None:
|
||||
transition_to("completed", "FINAL() in code")
|
||||
return {"type": "completed", "response": result["final_answer"]}
|
||||
|
||||
# Track errors
|
||||
if result.get("had_error"):
|
||||
consecutive_errors += 1
|
||||
if consecutive_errors >= 5:
|
||||
transition_to("failed", "too many consecutive errors")
|
||||
return {"type": "failed", "error": "5 consecutive code errors"}
|
||||
else:
|
||||
consecutive_errors = 0
|
||||
|
||||
save_checkpoint(state)
|
||||
|
||||
elif response["type"] == "actions":
|
||||
# Tier 0: structured tool calls
|
||||
nudge_count = 0
|
||||
results = []
|
||||
for call in response["calls"]:
|
||||
r = execute_action(call["name"], call.get("params", {}))
|
||||
results.append(r)
|
||||
if r.get("need_approval"):
|
||||
save_checkpoint(state)
|
||||
return {"type": "need_approval",
|
||||
"action_name": call["name"],
|
||||
"call_id": call.get("call_id", ""),
|
||||
"parameters": call.get("params", {})}
|
||||
|
||||
# Add results to context
|
||||
for r in results:
|
||||
context.append({"role": "tool", "content": format_action_result(r)})
|
||||
save_checkpoint(state)
|
||||
|
||||
# Max iterations reached
|
||||
transition_to("completed", "max iterations")
|
||||
return {"type": "max_iterations"}
|
||||
|
||||
|
||||
# ── Helper functions (the self-modifiable glue) ──────────────
|
||||
|
||||
def extract_final(text):
|
||||
"""Extract FINAL() content from text. Returns None if not found."""
|
||||
idx = text.find("FINAL(")
|
||||
if idx < 0:
|
||||
return None
|
||||
after = text[idx + 6:]
|
||||
# Handle triple-quoted strings
|
||||
if after.startswith('"""'):
|
||||
end = after.find('"""', 3)
|
||||
if end >= 0:
|
||||
return after[3:end]
|
||||
# Handle quoted strings
|
||||
if after.startswith('"') or after.startswith("'"):
|
||||
quote = after[0]
|
||||
end = after.find(quote, 1)
|
||||
if end >= 0:
|
||||
return after[1:end]
|
||||
# Handle balanced parens
|
||||
depth = 1
|
||||
for i, ch in enumerate(after):
|
||||
if ch == '(':
|
||||
depth += 1
|
||||
elif ch == ')':
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return after[:i]
|
||||
return None
|
||||
|
||||
|
||||
def signals_tool_intent(text):
|
||||
"""Check if text describes tool usage without actually using tools."""
|
||||
lower = text.lower()
|
||||
intent_phrases = ["i will", "i'll", "let me", "i would", "i should",
|
||||
"i can", "i need to", "we should", "we can"]
|
||||
tool_phrases = ["search", "fetch", "call", "run", "execute", "use the"]
|
||||
has_intent = any(p in lower for p in intent_phrases)
|
||||
has_tool = any(p in lower for p in tool_phrases)
|
||||
return has_intent and has_tool
|
||||
|
||||
|
||||
def format_output(result, max_chars=8000):
|
||||
"""Format code execution result for the next LLM context message."""
|
||||
parts = []
|
||||
|
||||
stdout = result.get("stdout", "")
|
||||
if stdout:
|
||||
parts.append(f"[stdout]\n{stdout}")
|
||||
|
||||
for r in result.get("action_results", []):
|
||||
name = r.get("action_name", "?")
|
||||
output = str(r.get("output", ""))
|
||||
if r.get("is_error"):
|
||||
parts.append(f"[{name} ERROR] {output}")
|
||||
else:
|
||||
preview = output[:500] + "..." if len(output) > 500 else output
|
||||
parts.append(f"[{name}] {preview}")
|
||||
|
||||
ret = result.get("return_value")
|
||||
if ret is not None:
|
||||
parts.append(f"[return] {ret}")
|
||||
|
||||
text = "\n\n".join(parts)
|
||||
|
||||
# Truncate from the front (keep the tail, which has the most recent results)
|
||||
if len(text) > max_chars:
|
||||
text = "... (truncated) ...\n" + text[-max_chars:]
|
||||
|
||||
return text
|
||||
|
||||
|
||||
def format_docs(docs):
|
||||
"""Format memory docs for context injection."""
|
||||
parts = ["## Prior Knowledge (from completed threads)\n"]
|
||||
for doc in docs:
|
||||
label = doc["type"].upper()
|
||||
content = doc["content"][:500]
|
||||
truncated = "..." if len(doc["content"]) > 500 else ""
|
||||
parts.append(f"### [{label}] {doc['title']}\n{content}{truncated}\n")
|
||||
return "\n".join(parts)
|
||||
|
||||
|
||||
def format_action_result(result):
|
||||
"""Format a single action result for the LLM context."""
|
||||
name = result.get("action_name", "unknown")
|
||||
output = result.get("output", {})
|
||||
if result.get("is_error"):
|
||||
return f"Tool '{name}' failed: {output}"
|
||||
return str(output)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Implementation Steps
|
||||
|
||||
### Step 1: Expose host functions in scripting.rs
|
||||
|
||||
Add new `FunctionCall` handlers alongside the existing tool dispatch:
|
||||
|
||||
- `__llm_complete__` → calls `LlmBackend::complete()`
|
||||
- `__check_signals__` → calls `signal_rx.try_recv()`
|
||||
- `__save_checkpoint__` → persists thread state
|
||||
- `__emit_event__` → broadcasts event
|
||||
- `__transition_to__` → validates + transitions thread state
|
||||
- `__retrieve_docs__` → queries RetrievalEngine
|
||||
- `__check_budget__` → reads remaining tokens/time/usd
|
||||
- `__get_actions__` → enumerates available ActionDefs from leases
|
||||
|
||||
These use `__dunder__` names to avoid collision with user tools.
|
||||
|
||||
### Step 2: Create the bootstrap in loop_engine.rs
|
||||
|
||||
Replace `ExecutionLoop::run()` body with:
|
||||
|
||||
1. Load orchestrator code from Store (`orchestrator:main` MemoryDoc, highest version)
|
||||
2. If no runtime version, use `include_str!("../../orchestrator/default.py")`
|
||||
3. Inject context variables: `context`, `goal`, `actions`, `state`, `config`
|
||||
4. Execute via Monty with the orchestrator code
|
||||
5. Parse the return value as `ThreadOutcome`
|
||||
6. Handle auto-rollback if execution fails
|
||||
|
||||
### Step 3: Write the default orchestrator
|
||||
|
||||
Create `crates/ironclaw_engine/orchestrator/default.py` with the v0 code shown above.
|
||||
|
||||
### Step 4: Wire versioning into the self-improvement Mission
|
||||
|
||||
Update the Mission goal prompt to include:
|
||||
- How to read the current orchestrator: `memory_search("orchestrator:main")`
|
||||
- How to update it: `memory_write` with title="orchestrator:main", tag="orchestrator_code", metadata with version++
|
||||
- The constraint: changes must be minimal, one fix at a time
|
||||
|
||||
### Step 5: Add auto-rollback
|
||||
|
||||
In the bootstrap (Step 2), after orchestrator execution fails:
|
||||
- Increment a failure counter in thread metadata
|
||||
- If counter >= 3, load `parent_version` instead
|
||||
- Emit `SelfImprovementRollback` event
|
||||
- Reset failure counter
|
||||
|
||||
### Step 6: Add `execute_code_step` host function
|
||||
|
||||
This is the interesting one — the orchestrator needs to run user Python code (the CodeAct step). Two options:
|
||||
|
||||
**Option A: Nested Monty execution** — The orchestrator Python calls `execute_code_step(code, state)` which suspends to Rust, Rust creates a nested Monty VM for the user code, runs it with tool dispatch, returns results. Clean but complex.
|
||||
|
||||
**Option B: Host-managed code execution** — The orchestrator returns a `{"type": "execute_code", "code": "...", "state": {...}}` action, Rust runs the code in the existing Monty pipeline, then re-enters the orchestrator with results. Simpler but requires the orchestrator to yield/resume.
|
||||
|
||||
Recommend **Option A** for clean separation. The orchestrator is a management layer; user code runs in a sandboxed sub-VM.
|
||||
|
||||
---
|
||||
|
||||
## What This Enables
|
||||
|
||||
1. **Self-improvement Mission fixes glue bugs at runtime** — no Rust rebuild
|
||||
2. **Format_output bug?** Mission patches `format_output()` in the orchestrator
|
||||
3. **Tool name mismatch?** Mission adds an alias in the orchestrator's dispatch
|
||||
4. **State persistence bug?** Mission fixes `save_checkpoint()` call
|
||||
5. **New feature?** Mission adds a new helper function
|
||||
6. **Bad fix?** Auto-rollback to previous version after 3 failures
|
||||
|
||||
The Rust layer becomes an OS kernel — stable, provides capabilities. The Python orchestrator is userspace — where iteration happens fast.
|
||||
|
||||
---
|
||||
|
||||
## Safety
|
||||
|
||||
| Concern | Mitigation |
|
||||
|---------|-----------|
|
||||
| Orchestrator loops forever | Rust-enforced timeout (existing 30s per code step, plus thread-level budget) |
|
||||
| Orchestrator skips safety checks | `execute_action()` enforces lease + policy in Rust regardless |
|
||||
| Orchestrator calls `transition_to("failed")` inappropriately | State machine validation stays in Rust |
|
||||
| Bad version breaks all threads | Auto-rollback after 3 consecutive failures |
|
||||
| Orchestrator tries to escape sandbox | Monty blocks OS calls, network, filesystem |
|
||||
| Self-improvement Mission writes bad code | Versioning allows instant rollback; compiled v0 always available |
|
||||
|
||||
---
|
||||
|
||||
## Migration Path
|
||||
|
||||
1. **Phase 1**: Add host functions, keep Rust loop as-is. Test that Python can call `llm_complete()` etc.
|
||||
2. **Phase 2**: Write default orchestrator in Python. Run it alongside Rust loop, compare outcomes.
|
||||
3. **Phase 3**: Switch to Python orchestrator as primary. Remove Rust loop code.
|
||||
4. **Phase 4**: Wire versioning + self-improvement Mission + auto-rollback.
|
||||
Reference in New Issue
Block a user