use std::sync::{Arc, Mutex};
use std::time::Duration;
use futures::stream;
use roder_api::catalog::PROVIDER_MOCK;
use roder_api::extension::{ExtensionRegistryBuilder, InferenceEngineId, ToolProviderId};
use roder_api::inference::*;
use roder_api::tools::{
ToolCall, ToolContributor, ToolExecutionContext, ToolExecutor, ToolResult, ToolSpec,
};
use roder_api::transcript::TranscriptItem;
use roder_core::{Runtime, RuntimeConfig, StartTurnRequest, default_instructions};
use roder_ext_jsonl_thread_store::store::JsonlThreadStoreFactory;
use serde_json::json;
use tokio::sync::Notify;
struct ToolLoopEngine {
requests: Mutex<Vec<AgentInferenceRequest>>,
tool_rounds: usize,
}
struct ParallelToolLoopEngine {
requests: Mutex<Vec<AgentInferenceRequest>>,
}
struct AgentControlEngine {
parent_thread_id: Mutex<Option<String>>,
thread_ids: Mutex<Vec<String>>,
tool_names_by_thread: Mutex<Vec<(String, Vec<String>)>>,
parent_saw_child_result: Mutex<bool>,
child_saw_queued_message: Mutex<bool>,
}
struct EchoContributor;
impl ToolContributor for EchoContributor {
fn id(&self) -> ToolProviderId {
"test-echo".to_string()
}
fn contribute(&self, registry: &mut roder_api::tools::ToolRegistry) -> anyhow::Result<()> {
registry.register(Arc::new(EchoTool))
}
}
struct EchoTool;
#[async_trait::async_trait]
impl ToolExecutor for EchoTool {
fn spec(&self) -> ToolSpec {
ToolSpec {
name: "echo".to_string(),
description: "Echo text".to_string(),
parameters: json!({
"type": "object",
"properties": { "text": { "type": "string" } },
"required": ["text"]
}),
}
}
async fn execute(
&self,
_ctx: ToolExecutionContext,
call: ToolCall,
) -> anyhow::Result<ToolResult> {
let text = call
.arguments
.get("text")
.and_then(serde_json::Value::as_str)
.unwrap_or("")
.to_string();
Ok(ToolResult {
id: call.id,
name: call.name,
text: text.clone(),
data: json!({ "text": text }),
is_error: false,
})
}
}
struct BlockingEchoContributor {
release: Arc<Notify>,
}
impl ToolContributor for BlockingEchoContributor {
fn id(&self) -> ToolProviderId {
"test-blocking-echo".to_string()
}
fn contribute(&self, registry: &mut roder_api::tools::ToolRegistry) -> anyhow::Result<()> {
registry.register(Arc::new(BlockingEchoTool {
release: self.release.clone(),
}))
}
}
struct BlockingEchoTool {
release: Arc<Notify>,
}
#[async_trait::async_trait]
impl ToolExecutor for BlockingEchoTool {
fn spec(&self) -> ToolSpec {
ToolSpec {
name: "echo".to_string(),
description: "Echo text after release".to_string(),
parameters: json!({
"type": "object",
"properties": { "text": { "type": "string" } },
"required": ["text"]
}),
}
}
async fn execute(
&self,
_ctx: ToolExecutionContext,
call: ToolCall,
) -> anyhow::Result<ToolResult> {
self.release.notified().await;
Ok(ToolResult {
id: call.id,
name: call.name,
text: "from blocked tool".to_string(),
data: json!({ "text": "from blocked tool" }),
is_error: false,
})
}
}
struct DuplicateWebSearchContributor(&'static str);
impl ToolContributor for DuplicateWebSearchContributor {
fn id(&self) -> ToolProviderId {
self.0.to_string()
}
fn contribute(&self, registry: &mut roder_api::tools::ToolRegistry) -> anyhow::Result<()> {
registry.register(Arc::new(WebSearchTool))
}
}
struct EditSurfaceContributor;
impl ToolContributor for EditSurfaceContributor {
fn id(&self) -> ToolProviderId {
"test-edit-surface".to_string()
}
fn contribute(&self, registry: &mut roder_api::tools::ToolRegistry) -> anyhow::Result<()> {
for name in [
"apply_patch",
"write_file",
"edit",
"multi_edit",
"read_file",
] {
registry.register(Arc::new(NamedTool(name)))?;
}
Ok(())
}
}
struct NamedTool(&'static str);
#[async_trait::async_trait]
impl ToolExecutor for NamedTool {
fn spec(&self) -> ToolSpec {
ToolSpec {
name: self.0.to_string(),
description: format!("{} tool", self.0),
parameters: json!({ "type": "object" }),
}
}
async fn execute(
&self,
_ctx: ToolExecutionContext,
call: ToolCall,
) -> anyhow::Result<ToolResult> {
Ok(ToolResult {
id: call.id,
name: call.name,
text: String::new(),
data: json!({}),
is_error: false,
})
}
}
struct FailingContributor;
impl ToolContributor for FailingContributor {
fn id(&self) -> ToolProviderId {
"test-failing".to_string()
}
fn contribute(&self, registry: &mut roder_api::tools::ToolRegistry) -> anyhow::Result<()> {
registry.register(Arc::new(FailingTool))
}
}
struct FailingTool;
#[async_trait::async_trait]
impl ToolExecutor for FailingTool {
fn spec(&self) -> ToolSpec {
ToolSpec {
name: "failing_tool".to_string(),
description: "Always fails".to_string(),
parameters: json!({ "type": "object" }),
}
}
async fn execute(
&self,
_ctx: ToolExecutionContext,
_call: ToolCall,
) -> anyhow::Result<ToolResult> {
anyhow::bail!("path does not exist: crates/roder-tui/src/main.rs")
}
}
struct WebSearchTool;
#[async_trait::async_trait]
impl ToolExecutor for WebSearchTool {
fn spec(&self) -> ToolSpec {
ToolSpec {
name: "web_search".to_string(),
description: "Search the web.".to_string(),
parameters: json!({
"type": "object",
"properties": { "query": { "type": "string" } },
"required": ["query"]
}),
}
}
async fn execute(
&self,
_ctx: ToolExecutionContext,
call: ToolCall,
) -> anyhow::Result<ToolResult> {
Ok(ToolResult {
id: call.id,
name: call.name,
text: String::new(),
data: json!({}),
is_error: false,
})
}
}
#[async_trait::async_trait]
impl InferenceEngine for ToolLoopEngine {
fn id(&self) -> InferenceEngineId {
PROVIDER_MOCK.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<ModelDescriptor>> {
Ok(Vec::new())
}
async fn stream_turn(
&self,
_ctx: InferenceTurnContext<'_>,
request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
let mut requests = self.requests.lock().unwrap();
requests.push(request);
let turn = requests.len();
drop(requests);
let events = if turn <= self.tool_rounds {
vec![
Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: format!("call_{turn}"),
name: "echo".to_string(),
arguments: r#"{"text":"from tool"}"#.to_string(),
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("tool_calls".to_string()),
provider_response_id: Some("resp_1".to_string()),
})),
]
} else {
vec![
Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "final after tool".to_string(),
phase: None,
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("stop".to_string()),
provider_response_id: Some("resp_2".to_string()),
})),
]
};
Ok(Box::pin(stream::iter(events)))
}
}
#[async_trait::async_trait]
impl InferenceEngine for ParallelToolLoopEngine {
fn id(&self) -> InferenceEngineId {
PROVIDER_MOCK.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<ModelDescriptor>> {
Ok(Vec::new())
}
async fn stream_turn(
&self,
_ctx: InferenceTurnContext<'_>,
request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
let mut requests = self.requests.lock().unwrap();
requests.push(request);
let turn = requests.len();
drop(requests);
let events = if turn == 1 {
vec![
Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "call_1".to_string(),
name: "echo".to_string(),
arguments: r#"{"text":"one"}"#.to_string(),
})),
Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "call_2".to_string(),
name: "echo".to_string(),
arguments: r#"{"text":"two"}"#.to_string(),
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("tool_calls".to_string()),
provider_response_id: Some("resp_tools".to_string()),
})),
]
} else {
vec![
Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "done".to_string(),
phase: None,
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("stop".to_string()),
provider_response_id: Some("resp_done".to_string()),
})),
]
};
Ok(Box::pin(stream::iter(events)))
}
}
#[async_trait::async_trait]
impl InferenceEngine for AgentControlEngine {
fn id(&self) -> InferenceEngineId {
PROVIDER_MOCK.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<ModelDescriptor>> {
Ok(Vec::new())
}
async fn stream_turn(
&self,
ctx: InferenceTurnContext<'_>,
request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
let is_parent = self.parent_thread_id.lock().unwrap().as_deref() == Some(ctx.thread_id);
if is_parent
&& request.transcript.iter().any(|item| {
matches!(
item,
TranscriptItem::UserMessage(message) if message.text.contains("child done")
)
})
{
*self.parent_saw_child_result.lock().unwrap() = true;
}
if !is_parent
&& request.transcript.iter().any(|item| {
matches!(
item,
TranscriptItem::UserMessage(message)
if message.text.contains("add one more detail")
&& message.text.contains("turn this into a final review")
)
})
{
*self.child_saw_queued_message.lock().unwrap() = true;
}
let has_result = |name: &str| {
request.transcript.iter().any(|item| {
matches!(
item,
TranscriptItem::ToolResult(result)
if result.name.as_deref() == Some(name)
)
})
};
let mut thread_ids = self.thread_ids.lock().unwrap();
thread_ids.push(ctx.thread_id.to_string());
drop(thread_ids);
self.tool_names_by_thread.lock().unwrap().push((
ctx.thread_id.to_string(),
request.tools.into_iter().map(|tool| tool.name).collect(),
));
let events = if is_parent && !has_result("spawn_agent") {
vec![
Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "call_spawn".to_string(),
name: "spawn_agent".to_string(),
arguments: json!({
"task_name": "reviewer",
"message": "review the code"
})
.to_string(),
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("tool_calls".to_string()),
provider_response_id: Some("resp_spawn".to_string()),
})),
]
} else if is_parent && !has_result("list_agents") {
vec![
Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "call_list".to_string(),
name: "list_agents".to_string(),
arguments: "{}".to_string(),
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("tool_calls".to_string()),
provider_response_id: Some("resp_list".to_string()),
})),
]
} else if is_parent && !has_result("send_message") {
vec![
Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "call_send".to_string(),
name: "send_message".to_string(),
arguments: json!({
"target": "reviewer",
"message": "add one more detail"
})
.to_string(),
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("tool_calls".to_string()),
provider_response_id: Some("resp_send".to_string()),
})),
]
} else if is_parent && !has_result("followup_task") {
vec![
Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "call_followup".to_string(),
name: "followup_task".to_string(),
arguments: json!({
"target": "reviewer",
"message": "turn this into a final review"
})
.to_string(),
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("tool_calls".to_string()),
provider_response_id: Some("resp_followup".to_string()),
})),
]
} else if is_parent && !has_result("wait_agent") {
vec![
Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "call_wait".to_string(),
name: "wait_agent".to_string(),
arguments: json!({
"target": "reviewer",
"timeout_ms": 10_000
})
.to_string(),
})),
Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "call_interrupt".to_string(),
name: "interrupt_agent".to_string(),
arguments: json!({
"target": "reviewer"
})
.to_string(),
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("tool_calls".to_string()),
provider_response_id: Some("resp_followup_batch".to_string()),
})),
]
} else if is_parent {
vec![
Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "parent done".to_string(),
phase: None,
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("stop".to_string()),
provider_response_id: Some("resp_parent".to_string()),
})),
]
} else {
vec![
Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "child done".to_string(),
phase: None,
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("stop".to_string()),
provider_response_id: Some("resp_child".to_string()),
})),
]
};
Ok(Box::pin(stream::iter(events)))
}
}
struct ErrorRecoveringEngine {
requests: Mutex<Vec<AgentInferenceRequest>>,
}
struct PhasePreservingEngine {
requests: Mutex<Vec<AgentInferenceRequest>>,
}
struct FailingStreamStartEngine;
#[async_trait::async_trait]
impl InferenceEngine for ErrorRecoveringEngine {
fn id(&self) -> InferenceEngineId {
PROVIDER_MOCK.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<ModelDescriptor>> {
Ok(Vec::new())
}
async fn stream_turn(
&self,
_ctx: InferenceTurnContext<'_>,
request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
let mut requests = self.requests.lock().unwrap();
requests.push(request);
let has_tool_error = requests.last().unwrap().transcript.iter().any(|item| {
matches!(
item,
TranscriptItem::ToolResult(result)
if result.is_error
&& result.result.contains("path does not exist")
)
});
drop(requests);
let events = if has_tool_error {
vec![
Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "recovered".to_string(),
phase: None,
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("stop".to_string()),
provider_response_id: Some("resp_recovered".to_string()),
})),
]
} else {
vec![
Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "call_failed_read".to_string(),
name: "failing_tool".to_string(),
arguments: "{}".to_string(),
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("tool_calls".to_string()),
provider_response_id: Some("resp_tool".to_string()),
})),
]
};
Ok(Box::pin(stream::iter(events)))
}
}
#[async_trait::async_trait]
impl InferenceEngine for PhasePreservingEngine {
fn id(&self) -> InferenceEngineId {
PROVIDER_MOCK.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<ModelDescriptor>> {
Ok(Vec::new())
}
async fn stream_turn(
&self,
_ctx: InferenceTurnContext<'_>,
request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
let mut requests = self.requests.lock().unwrap();
requests.push(request);
let turn = requests.len();
drop(requests);
let events = if turn == 1 {
vec![
Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "I will inspect first.".to_string(),
phase: Some("commentary".to_string()),
})),
Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "call_1".to_string(),
name: "echo".to_string(),
arguments: r#"{"text":"from tool"}"#.to_string(),
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("tool_calls".to_string()),
provider_response_id: None,
})),
]
} else {
vec![
Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "Done.".to_string(),
phase: Some("final_answer".to_string()),
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("stop".to_string()),
provider_response_id: None,
})),
]
};
Ok(Box::pin(stream::iter(events)))
}
}
#[async_trait::async_trait]
impl InferenceEngine for FailingStreamStartEngine {
fn id(&self) -> InferenceEngineId {
PROVIDER_MOCK.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<ModelDescriptor>> {
Ok(Vec::new())
}
async fn stream_turn(
&self,
_ctx: InferenceTurnContext<'_>,
_request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
anyhow::bail!("input exceeds the context window")
}
}
#[tokio::test]
async fn run_turn_continues_after_tool_result() {
let engine = Arc::new(ToolLoopEngine {
requests: Mutex::new(Vec::new()),
tool_rounds: 1,
});
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(engine.clone());
builder.tool_contributor(Arc::new(EchoContributor));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: PROVIDER_MOCK.to_string(),
default_model: "gpt-5.5".to_string(),
reasoning: Some("low".to_string()),
auto_compact_token_limit: Some(10_000),
file_backed_dynamic_context: true,
hosted_web_search: roder_api::inference::HostedWebSearchConfig::disabled(),
model_edit_tools: std::collections::HashMap::new(),
model_parallel_tool_calls: std::collections::HashMap::new(),
model_profiles: std::collections::HashMap::new(),
tool_allowlist: Vec::new(),
command_shell: roder_api::command_shell::default_command_shell(),
workspace: None,
policy_mode: roder_api::policy_mode::PolicyMode::Default,
runtime_profile: roder_api::inference::RuntimeProfile::Interactive,
speed_policy: Default::default(),
dynamic_workflows: Default::default(),
reliability: Default::default(),
turn_deadline_seconds: None,
remote_runner_destination: None,
team_data_dir: None,
roadmap_data_dir: None,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut events = runtime.subscribe_events();
runtime
.start_turn(StartTurnRequest {
thread_id: "thread_1".to_string(),
message: "echo please".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: std::env::current_dir().unwrap().display().to_string(),
instructions: default_instructions(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
loop {
let event = tokio::time::timeout(Duration::from_secs(2), events.recv())
.await
.unwrap()
.unwrap();
if event.kind == "turn.completed" && event.thread_id.as_deref() == Some("thread_1") {
break;
}
}
let requests = engine.requests.lock().unwrap();
assert_eq!(requests.len(), 2);
assert_eq!(requests[0].reasoning.level.as_deref(), Some("low"));
assert!(
requests[0]
.instructions
.system
.as_deref()
.is_some_and(|text| text.contains("You are Roder")
&& text.contains("send a brief preamble message")),
"first request should include the Roder system prompt with intermediary message guidance"
);
assert!(
requests[1]
.transcript
.iter()
.any(|item| matches!(item, TranscriptItem::ToolResult(result) if result.result == "from tool")),
"second request should include the tool result: {:?}",
requests[1].transcript
);
}
#[tokio::test]
async fn run_turn_executes_parallel_tool_call_batch_concurrently() {
let engine = Arc::new(ParallelToolLoopEngine {
requests: Mutex::new(Vec::new()),
});
let release = Arc::new(Notify::new());
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(engine.clone());
builder.tool_contributor(Arc::new(BlockingEchoContributor {
release: release.clone(),
}));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: PROVIDER_MOCK.to_string(),
default_model: "gpt-5.5".to_string(),
reasoning: Some("low".to_string()),
auto_compact_token_limit: None,
file_backed_dynamic_context: true,
hosted_web_search: roder_api::inference::HostedWebSearchConfig::disabled(),
model_edit_tools: std::collections::HashMap::new(),
model_parallel_tool_calls: std::collections::HashMap::new(),
model_profiles: std::collections::HashMap::new(),
tool_allowlist: Vec::new(),
command_shell: roder_api::command_shell::default_command_shell(),
workspace: None,
policy_mode: roder_api::policy_mode::PolicyMode::Default,
runtime_profile: roder_api::inference::RuntimeProfile::Interactive,
speed_policy: Default::default(),
dynamic_workflows: Default::default(),
reliability: Default::default(),
turn_deadline_seconds: None,
remote_runner_destination: None,
team_data_dir: None,
roadmap_data_dir: None,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut events = runtime.subscribe_events();
runtime
.start_turn(StartTurnRequest {
thread_id: "thread_parallel".to_string(),
message: "run tools".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: std::env::current_dir().unwrap().display().to_string(),
instructions: default_instructions(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
let started_two_tools = tokio::time::timeout(Duration::from_millis(300), async {
let mut started = Vec::new();
while started.len() < 2 {
let event = events.recv().await.unwrap();
if let roder_api::events::RoderEvent::ToolCallStarted(tool) = event.event {
started.push(tool.tool_id);
}
}
started
})
.await;
if started_two_tools.is_err() {
release.notify_waiters();
panic!("runtime did not start both tool calls before the first blocked");
}
let started = started_two_tools.unwrap();
assert_eq!(started, vec!["call_1".to_string(), "call_2".to_string()]);
release.notify_waiters();
loop {
let event = tokio::time::timeout(Duration::from_secs(2), events.recv())
.await
.unwrap()
.unwrap();
if event.kind == "turn.completed" && event.thread_id.as_deref() == Some("thread_parallel") {
break;
}
}
let requests = engine.requests.lock().unwrap();
assert_eq!(requests.len(), 2);
assert_eq!(requests[0].runtime.parallel_tool_calls, Some(true));
let result_count = requests[1]
.transcript
.iter()
.filter(|item| matches!(item, TranscriptItem::ToolResult(_)))
.count();
assert_eq!(result_count, 2);
}
#[tokio::test]
async fn provider_start_errors_are_emitted_for_the_active_thread() {
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(FailingStreamStartEngine));
let runtime =
Arc::new(Runtime::new(builder.build().unwrap(), RuntimeConfig::default()).unwrap());
let mut events = runtime.subscribe_events();
runtime
.start_turn(StartTurnRequest {
thread_id: "thread_context_error".to_string(),
message: "short prompt".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: std::env::current_dir().unwrap().display().to_string(),
instructions: default_instructions(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
loop {
let event = tokio::time::timeout(Duration::from_secs(2), events.recv())
.await
.unwrap()
.unwrap();
if let roder_api::events::RoderEvent::TurnFailed(failed) = event.event {
assert_eq!(failed.thread_id, "thread_context_error");
assert!(failed.error.contains("context window"));
break;
}
}
}
#[tokio::test]
async fn commentary_phase_messages_are_preserved_for_next_provider_request() {
let engine = Arc::new(PhasePreservingEngine {
requests: Mutex::new(Vec::new()),
});
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(engine.clone());
builder.tool_contributor(Arc::new(EchoContributor));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: PROVIDER_MOCK.to_string(),
default_model: "mock".to_string(),
reasoning: None,
auto_compact_token_limit: None,
file_backed_dynamic_context: true,
hosted_web_search: roder_api::inference::HostedWebSearchConfig::disabled(),
model_edit_tools: std::collections::HashMap::new(),
model_parallel_tool_calls: std::collections::HashMap::new(),
model_profiles: std::collections::HashMap::new(),
tool_allowlist: Vec::new(),
command_shell: roder_api::command_shell::default_command_shell(),
workspace: None,
policy_mode: roder_api::policy_mode::PolicyMode::Default,
runtime_profile: roder_api::inference::RuntimeProfile::Interactive,
speed_policy: Default::default(),
dynamic_workflows: Default::default(),
reliability: Default::default(),
turn_deadline_seconds: None,
remote_runner_destination: None,
team_data_dir: None,
roadmap_data_dir: None,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut events = runtime.subscribe_events();
runtime
.start_turn(StartTurnRequest {
thread_id: "thread_phase_messages".to_string(),
message: "inspect".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: std::env::current_dir().unwrap().display().to_string(),
instructions: default_instructions(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
wait_for_completed(&mut events, "thread_phase_messages").await;
let requests = engine.requests.lock().unwrap();
assert_eq!(requests.len(), 2);
assert!(
requests[1].transcript.iter().any(|item| matches!(
item,
TranscriptItem::AssistantMessage(message)
if message.text == "I will inspect first."
&& message.phase.as_deref() == Some("commentary")
)),
"second request should preserve commentary assistant message: {:?}",
requests[1].transcript
);
}
#[tokio::test]
async fn steer_turn_is_included_in_next_provider_request() {
let engine = Arc::new(ToolLoopEngine {
requests: Mutex::new(Vec::new()),
tool_rounds: 1,
});
let release_tool = Arc::new(Notify::new());
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(engine.clone());
builder.tool_contributor(Arc::new(BlockingEchoContributor {
release: release_tool.clone(),
}));
let runtime =
Arc::new(Runtime::new(builder.build().unwrap(), RuntimeConfig::default()).unwrap());
let mut events = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: "thread_steer".to_string(),
message: "start".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: std::env::current_dir().unwrap().display().to_string(),
instructions: default_instructions(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
wait_for_kind(&mut events, "thread_steer", "tool.call_requested").await;
runtime
.steer_turn(
"thread_steer".to_string(),
turn_id,
"use the new constraint".to_string(),
Vec::new(),
)
.await
.unwrap();
release_tool.notify_one();
wait_for_completed(&mut events, "thread_steer").await;
let requests = engine.requests.lock().unwrap();
assert_eq!(requests.len(), 2);
assert!(
requests[1].transcript.iter().any(|item| {
matches!(item, TranscriptItem::UserMessage(message) if message.text == "use the new constraint")
}),
"second request should include steer message: {:?}",
requests[1].transcript
);
}
#[tokio::test]
async fn runtime_advertises_apply_patch_for_default_edit_models() {
let engine = Arc::new(ToolLoopEngine {
requests: Mutex::new(Vec::new()),
tool_rounds: 0,
});
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(engine.clone());
builder.tool_contributor(Arc::new(EditSurfaceContributor));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: PROVIDER_MOCK.to_string(),
default_model: "gpt-5.5".to_string(),
reasoning: None,
auto_compact_token_limit: None,
file_backed_dynamic_context: true,
hosted_web_search: roder_api::inference::HostedWebSearchConfig::disabled(),
model_edit_tools: std::collections::HashMap::new(),
model_parallel_tool_calls: std::collections::HashMap::new(),
model_profiles: std::collections::HashMap::new(),
tool_allowlist: Vec::new(),
command_shell: roder_api::command_shell::default_command_shell(),
workspace: None,
policy_mode: roder_api::policy_mode::PolicyMode::Default,
runtime_profile: roder_api::inference::RuntimeProfile::Interactive,
speed_policy: Default::default(),
dynamic_workflows: Default::default(),
reliability: Default::default(),
turn_deadline_seconds: None,
remote_runner_destination: None,
team_data_dir: None,
roadmap_data_dir: None,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut events = runtime.subscribe_events();
runtime
.start_turn(StartTurnRequest {
thread_id: "thread_patch_tools".to_string(),
message: "patch please".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: std::env::current_dir().unwrap().display().to_string(),
instructions: default_instructions(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
wait_for_completed(&mut events, "thread_patch_tools").await;
let requests = engine.requests.lock().unwrap();
let names = request_tool_names(&requests[0]);
assert!(names.contains(&"apply_patch".to_string()), "{names:?}");
for excluded in ["write_file", "edit", "multi_edit"] {
assert!(!names.contains(&excluded.to_string()), "{names:?}");
}
assert!(names.contains(&"read_file".to_string()), "{names:?}");
}
#[tokio::test]
async fn runtime_keeps_apply_patch_for_edit_profile_models() {
let engine = Arc::new(ToolLoopEngine {
requests: Mutex::new(Vec::new()),
tool_rounds: 0,
});
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(engine.clone());
builder.tool_contributor(Arc::new(EditSurfaceContributor));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: PROVIDER_MOCK.to_string(),
default_model: "edit-model".to_string(),
model_edit_tools: std::collections::HashMap::from([(
"edit-model".to_string(),
"edit".to_string(),
)]),
policy_mode: roder_api::policy_mode::PolicyMode::Default,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut events = runtime.subscribe_events();
runtime
.start_turn(StartTurnRequest {
thread_id: "thread_edit_tools".to_string(),
message: "edit please".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: std::env::current_dir().unwrap().display().to_string(),
instructions: default_instructions(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
wait_for_completed(&mut events, "thread_edit_tools").await;
let requests = engine.requests.lock().unwrap();
let names = request_tool_names(&requests[0]);
for included in [
"apply_patch",
"write_file",
"edit",
"multi_edit",
"read_file",
] {
assert!(names.contains(&included.to_string()), "{names:?}");
}
}
#[tokio::test]
async fn runtime_uses_custom_model_edit_tool_override() {
let engine = Arc::new(ToolLoopEngine {
requests: Mutex::new(Vec::new()),
tool_rounds: 0,
});
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(engine.clone());
builder.tool_contributor(Arc::new(EditSurfaceContributor));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: PROVIDER_MOCK.to_string(),
default_model: "custom-model".to_string(),
reasoning: None,
auto_compact_token_limit: None,
file_backed_dynamic_context: true,
hosted_web_search: roder_api::inference::HostedWebSearchConfig::disabled(),
model_edit_tools: std::collections::HashMap::from([(
"custom-model".to_string(),
"patch".to_string(),
)]),
model_parallel_tool_calls: std::collections::HashMap::new(),
model_profiles: std::collections::HashMap::new(),
tool_allowlist: Vec::new(),
command_shell: roder_api::command_shell::default_command_shell(),
workspace: None,
policy_mode: roder_api::policy_mode::PolicyMode::Default,
runtime_profile: roder_api::inference::RuntimeProfile::Interactive,
speed_policy: Default::default(),
dynamic_workflows: Default::default(),
reliability: Default::default(),
turn_deadline_seconds: None,
remote_runner_destination: None,
team_data_dir: None,
roadmap_data_dir: None,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let names = runtime
.tool_specs()
.await
.into_iter()
.map(|tool| tool.name)
.collect::<Vec<_>>();
assert!(names.contains(&"apply_patch".to_string()), "{names:?}");
assert!(!names.contains(&"edit".to_string()), "{names:?}");
assert!(!names.contains(&"multi_edit".to_string()), "{names:?}");
assert!(!names.contains(&"write_file".to_string()), "{names:?}");
}
#[tokio::test]
async fn model_can_spawn_long_lived_subagent_with_agent_control_tool() {
let engine = Arc::new(AgentControlEngine {
parent_thread_id: Mutex::new(None),
thread_ids: Mutex::new(Vec::new()),
tool_names_by_thread: Mutex::new(Vec::new()),
parent_saw_child_result: Mutex::new(false),
child_saw_queued_message: Mutex::new(false),
});
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(engine.clone());
let thread_root = std::env::temp_dir().join(format!(
"roder-agent-control-tool-threads-{}",
uuid::Uuid::new_v4()
));
let team_root = std::env::temp_dir().join(format!(
"roder-agent-control-tool-teams-{}",
uuid::Uuid::new_v4()
));
builder.thread_store_factory(Arc::new(JsonlThreadStoreFactory {
base_path: thread_root.clone(),
}));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
policy_mode: roder_api::policy_mode::PolicyMode::Bypass,
team_data_dir: Some(team_root.clone()),
..RuntimeConfig::default()
},
)
.unwrap(),
);
let parent_thread_id = runtime
.create_thread(Some("Agent control parent".to_string()))
.await
.unwrap()
.thread_id;
*engine.parent_thread_id.lock().unwrap() = Some(parent_thread_id.clone());
let mut rx = runtime.subscribe_events();
let _turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: parent_thread_id.clone(),
message: "delegate this".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: std::env::current_dir().unwrap().display().to_string(),
instructions: default_instructions(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
let mut saw_legacy_one_shot_completion = false;
loop {
let envelope = tokio::time::timeout(Duration::from_secs(2), rx.recv())
.await
.unwrap()
.unwrap();
if matches!(
envelope.event,
roder_api::events::RoderEvent::SubagentCompleted(_)
) {
saw_legacy_one_shot_completion = true;
}
if envelope.kind == "turn.completed"
&& envelope.thread_id.as_deref() == Some(parent_thread_id.as_str())
{
break;
}
}
assert!(
!saw_legacy_one_shot_completion,
"long-lived agent controls must not emit synthetic one-shot SubagentCompleted events"
);
let teams = runtime.list_teams().await;
assert_eq!(teams.len(), 1);
assert_eq!(teams[0].lead_thread_id, parent_thread_id);
assert_eq!(teams[0].members.len(), 2);
assert_eq!(teams[0].members[1].name, "reviewer");
assert_eq!(
teams[0].members[1].status,
roder_api::teams::TeamMemberStatus::Completed
);
for text in ["add one more detail", "turn this into a final review"] {
assert!(
teams[0]
.mailbox
.iter()
.any(|message| message.text == text && message.delivered),
"mailbox entry {text:?} should be acknowledged only after it reaches the child transcript"
);
}
assert_ne!(teams[0].members[1].thread_id, teams[0].lead_thread_id);
let thread_ids = engine.thread_ids.lock().unwrap().clone();
assert!(thread_ids.contains(&teams[0].lead_thread_id));
assert!(thread_ids.iter().any(|id| id != &teams[0].lead_thread_id));
assert_eq!(
thread_ids
.iter()
.filter(|id| *id == &teams[0].members[1].thread_id)
.count(),
2,
"send_message must stay queued while followup_task starts exactly one reusable-agent turn"
);
let tool_names_by_thread = engine.tool_names_by_thread.lock().unwrap().clone();
let parent_tools = tool_names_by_thread
.iter()
.find(|(thread_id, _)| thread_id == &teams[0].lead_thread_id)
.map(|(_, tools)| tools)
.unwrap();
assert!(parent_tools.contains(&"spawn_agent".to_string()));
assert!(parent_tools.contains(&"send_message".to_string()));
assert!(parent_tools.contains(&"list_agents".to_string()));
assert!(parent_tools.contains(&"wait_agent".to_string()));
assert!(parent_tools.contains(&"interrupt_agent".to_string()));
let child_tools = tool_names_by_thread
.iter()
.find(|(thread_id, _)| thread_id != &teams[0].lead_thread_id)
.map(|(_, tools)| tools)
.unwrap();
assert!(child_tools.contains(&"spawn_agent".to_string()));
assert!(child_tools.contains(&"send_message".to_string()));
assert!(child_tools.contains(&"list_agents".to_string()));
assert!(child_tools.contains(&"wait_agent".to_string()));
assert!(child_tools.contains(&"interrupt_agent".to_string()));
assert!(
*engine.parent_saw_child_result.lock().unwrap(),
"the direct parent should receive the child's terminal result in its next inference context"
);
assert!(
*engine.child_saw_queued_message.lock().unwrap(),
"the follow-up turn should receive both the queued send_message and followup_task input"
);
let _ = std::fs::remove_dir_all(thread_root);
let _ = std::fs::remove_dir_all(team_root);
}
#[tokio::test]
async fn tool_execution_errors_are_returned_to_model() {
let engine = Arc::new(ErrorRecoveringEngine {
requests: Mutex::new(Vec::new()),
});
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(engine.clone());
builder.tool_contributor(Arc::new(FailingContributor));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: PROVIDER_MOCK.to_string(),
default_model: "mock".to_string(),
reasoning: None,
auto_compact_token_limit: None,
file_backed_dynamic_context: true,
hosted_web_search: roder_api::inference::HostedWebSearchConfig::disabled(),
model_edit_tools: std::collections::HashMap::new(),
model_parallel_tool_calls: std::collections::HashMap::new(),
model_profiles: std::collections::HashMap::new(),
tool_allowlist: Vec::new(),
command_shell: roder_api::command_shell::default_command_shell(),
workspace: None,
policy_mode: roder_api::policy_mode::PolicyMode::Default,
runtime_profile: roder_api::inference::RuntimeProfile::Interactive,
speed_policy: Default::default(),
dynamic_workflows: Default::default(),
reliability: Default::default(),
turn_deadline_seconds: None,
remote_runner_destination: None,
team_data_dir: None,
roadmap_data_dir: None,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut events = runtime.subscribe_events();
runtime
.start_turn(StartTurnRequest {
thread_id: "thread_tool_error".to_string(),
message: "read missing file".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: std::env::current_dir().unwrap().display().to_string(),
instructions: default_instructions(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
wait_for_completed(&mut events, "thread_tool_error").await;
let requests = engine.requests.lock().unwrap();
assert_eq!(requests.len(), 2);
assert!(
requests[1].transcript.iter().any(|item| matches!(
item,
TranscriptItem::ToolResult(result)
if result.name.as_deref() == Some("failing_tool")
&& result.is_error
&& result.result == "path does not exist: crates/roder-tui/src/main.rs"
)),
"second request should include the tool error result: {:?}",
requests[1].transcript
);
}
#[tokio::test]
async fn run_turn_allows_more_than_eight_tool_rounds() {
let engine = Arc::new(ToolLoopEngine {
requests: Mutex::new(Vec::new()),
tool_rounds: 9,
});
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(engine.clone());
builder.tool_contributor(Arc::new(EchoContributor));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: PROVIDER_MOCK.to_string(),
default_model: "mock".to_string(),
reasoning: Some("low".to_string()),
auto_compact_token_limit: Some(10_000),
file_backed_dynamic_context: true,
hosted_web_search: roder_api::inference::HostedWebSearchConfig::disabled(),
model_edit_tools: std::collections::HashMap::new(),
model_parallel_tool_calls: std::collections::HashMap::new(),
model_profiles: std::collections::HashMap::new(),
tool_allowlist: Vec::new(),
command_shell: roder_api::command_shell::default_command_shell(),
workspace: None,
policy_mode: roder_api::policy_mode::PolicyMode::Default,
runtime_profile: roder_api::inference::RuntimeProfile::Interactive,
speed_policy: Default::default(),
dynamic_workflows: Default::default(),
reliability: Default::default(),
turn_deadline_seconds: None,
remote_runner_destination: None,
team_data_dir: None,
roadmap_data_dir: None,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut events = runtime.subscribe_events();
runtime
.start_turn(StartTurnRequest {
thread_id: "thread_many_tools".to_string(),
message: "keep using tools".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: std::env::current_dir().unwrap().display().to_string(),
instructions: default_instructions(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
loop {
let event = tokio::time::timeout(Duration::from_secs(2), events.recv())
.await
.unwrap()
.unwrap();
if event.kind == "turn.completed" && event.thread_id.as_deref() == Some("thread_many_tools")
{
break;
}
}
let requests = engine.requests.lock().unwrap();
assert_eq!(requests.len(), 10);
}
#[tokio::test]
async fn unknown_tool_completion_is_marked_as_error() {
let engine = Arc::new(ToolLoopEngine {
requests: Mutex::new(Vec::new()),
tool_rounds: 1,
});
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(engine);
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: PROVIDER_MOCK.to_string(),
default_model: "mock".to_string(),
reasoning: Some("low".to_string()),
auto_compact_token_limit: Some(10_000),
file_backed_dynamic_context: true,
hosted_web_search: roder_api::inference::HostedWebSearchConfig::disabled(),
model_edit_tools: std::collections::HashMap::new(),
model_parallel_tool_calls: std::collections::HashMap::new(),
model_profiles: std::collections::HashMap::new(),
tool_allowlist: Vec::new(),
command_shell: roder_api::command_shell::default_command_shell(),
workspace: None,
policy_mode: roder_api::policy_mode::PolicyMode::Default,
runtime_profile: roder_api::inference::RuntimeProfile::Interactive,
speed_policy: Default::default(),
dynamic_workflows: Default::default(),
reliability: Default::default(),
turn_deadline_seconds: None,
remote_runner_destination: None,
team_data_dir: None,
roadmap_data_dir: None,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut events = runtime.subscribe_events();
runtime
.start_turn(StartTurnRequest {
thread_id: "thread_unknown_tool".to_string(),
message: "echo please".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: std::env::current_dir().unwrap().display().to_string(),
instructions: default_instructions(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
loop {
let event = tokio::time::timeout(Duration::from_secs(2), events.recv())
.await
.unwrap()
.unwrap();
if let roder_api::events::RoderEvent::ToolCallCompleted(completed) = event.event {
assert!(completed.is_error);
break;
}
}
}
#[test]
fn duplicate_tool_contributors_fail_with_contributor_context() {
let engine = Arc::new(ToolLoopEngine {
requests: Mutex::new(Vec::new()),
tool_rounds: 1,
});
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(engine);
builder.tool_contributor(Arc::new(DuplicateWebSearchContributor("search-a")));
builder.tool_contributor(Arc::new(DuplicateWebSearchContributor("search-b")));
let err = match builder.build() {
Ok(_) => panic!("duplicate web_search tools should fail registry construction"),
Err(err) => err,
};
let message = format!("{err:#}");
assert!(
message.contains("tool \"web_search\" is already registered"),
"{message}"
);
}
async fn wait_for_completed(
events: &mut tokio::sync::broadcast::Receiver<roder_api::events::EventEnvelope>,
thread_id: &str,
) {
wait_for_kind(events, thread_id, "turn.completed").await;
}
async fn wait_for_kind(
events: &mut tokio::sync::broadcast::Receiver<roder_api::events::EventEnvelope>,
thread_id: &str,
kind: &str,
) {
loop {
let event = tokio::time::timeout(Duration::from_secs(2), events.recv())
.await
.unwrap()
.unwrap();
if event.kind == kind && event.thread_id.as_deref() == Some(thread_id) {
break;
}
}
}
fn request_tool_names(request: &AgentInferenceRequest) -> Vec<String> {
request.tools.iter().map(|tool| tool.name.clone()).collect()
}