use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Weak};
use anyhow::Context;
use futures::StreamExt;
use futures::future::{AbortHandle, Abortable, BoxFuture, try_join_all};
use roder_api::catalog::{
EDIT_TOOL_EDIT, EDIT_TOOL_PATCH, PROVIDER_GEMINI, REASONING_NONE, REASONING_ULTRA,
built_in_model_profile, built_in_model_profile_for_provider, lookup_model,
model_supports_reasoning_effort,
};
use roder_api::context::PolicyGate;
use roder_api::events::*;
use roder_api::extension::ExtensionRegistry;
use roder_api::inference::{
AgentInferenceRequest, HostedWebSearchConfig, HostedWebSearchMode, InferenceEngine,
InferenceEvent, InferenceTurnContext, InstructionBundle, ModelHarnessProfile,
ModelSchemaPolicy, ModelSelection, OutputConfig, ProviderTurnCleanup, ReasoningConfig,
RuntimeHints, RuntimeProfile, TokenUsage, ToolCallCompleted, ToolSearchConfig,
ToolSearchConfigOverlay, finish_reason_from_stop_reason,
};
use roder_api::inference_routing::{InferenceRoutingOutcome, ModelSelectionMode};
use roder_api::lifecycle::{
LifecycleMetricsSnapshot, TurnCleanupOwnership, TurnCleanupState, TurnLifecycleReason,
TurnLifecycleRecord, TurnLifecycleSnapshot, TurnLifecycleState, turn_lifecycle_snapshot,
};
use roder_api::policy_mode::{PolicyDecision, PolicyMode};
use roder_api::reliability::{
ReliabilityContext, ReliabilityDetails, ReliabilityErrorClass, ReliabilityLimitDecision,
ReliabilityLimitRecorded, ReliabilityRequestPolicy, ReliabilityRetryDecision,
ReliabilityRetryRecorded, provider_retry_delay_ms,
};
use roder_api::remote_runner::{
RemoteRunnerProvider, RemoteRunnerSession, RemoteWorkspace, RunnerDestination,
RunnerSessionState, ThreadRunnerBinding,
};
use roder_api::subagents::SubagentDefinition;
use roder_api::teams::{
TeamId, TeamMailboxMessage, TeamMailboxMessageKind, TeamMemberDescriptor, TeamMemberRole,
TeamMemberStatus,
};
use roder_api::thread::{
ThreadItemEvent, ThreadItemEventKind, ThreadMetadata, ThreadSnapshot, ThreadStore,
ThreadUsageMetadata, is_synthetic_event_thread_id, validate_thread_workspace,
};
use roder_api::tools::{ToolCall, ToolChoice, ToolExecutionContext, ToolRegistry, ToolResult};
use roder_api::transcript::{
AssistantMessage, ErrorRecord, InputImage, ReasoningSummary, ToolCallRecord, ToolResultRecord,
TranscriptItem, UserMessage,
};
use roder_sandbox::ScopedFilesystem;
use roder_sandbox::process::LocalProcessRunner;
use roder_skills::{SkillRegistry, SkillRegistryOptions};
use time::{Duration, OffsetDateTime};
use tokio::sync::{Mutex, Notify, RwLock, oneshot};
mod codex_v2;
use crate::artifacts::{
ContextArtifactStore as FilesystemContextArtifactStore, default_context_artifact_dir,
};
use crate::bus::EventBus;
use crate::dynamic_workflows::{
DynamicWorkflowEffortProfile, RuntimeDynamicWorkflowConfig, WorkflowTriggerDecision,
classify_workflow_trigger, ultracode_reasoning_level_for_model,
};
use crate::fake_provider::FakeInferenceEngine;
use crate::goals::RuntimeGoalController;
use crate::inference_routing::{
InferenceRoutingRequest, RuntimeInferenceRouterConfig, collect_inference_routing_candidates,
route_inference_selection, transcript_failure_count_since,
};
use crate::instructions::{
apply_agent_swarm_mode, apply_codex_multi_agent_mode, apply_model_instruction_overlay,
apply_plan_mode, apply_runtime_profile, apply_task_ledger_required,
apply_thread_developer_instructions, apply_turn_developer_context,
};
use crate::policy_gate::DefaultPolicyGate;
use crate::reliability::{
ReliabilityLimitHit, RuntimeReliabilityConfig, TurnReliabilityState,
provider_stream_retry_cause,
};
pub use crate::speed_policy::RuntimeSpeedPolicyConfig;
use crate::speed_policy::{SpeedPolicyState, reasoning_from_decision};
use crate::subagent_traces::{RuntimeAgentSwarmProgressSink, RuntimeSubagentTraceSink};
use crate::teams::{TeamManager, TeamMemberStartRequest, TeamStartRequest, TeamState};
use crate::thread_item_cache::{ThreadItemCache, ThreadItemCacheEntry};
use crate::verification_gate::VerificationGateState;
const MAX_TOOL_ROUNDS_PER_TURN: usize = 1024;
/// Injected when `continue_on_failure_limit` turns a consecutive-tool-failure
/// limit into a continuation, or as the empty-tool-call persistence nudge, so
/// the model keeps working instead of ending the turn early.
const RELIABILITY_CONTINUATION_PROMPT: &str = "Verify the task is fully complete. If any part remains unfinished or unverified, keep working using the available tools — try an alternative approach if one is failing. Only stop once the solution is complete and verified; if it is already complete, restate your final answer.";
/// Capacity of the runtime event broadcast ring buffer. The TUI drains this on
/// its render loop, which during an active turn only wakes at the ~6 FPS status
/// animation cadence (backend events do not wake the input poll), so up to
/// ~166ms of events buffer between drains. A reasoning-high provider that runs
/// its whole tool loop inside one turn (e.g. claude-code) streams a firehose of
/// `InferenceEventReceived` deltas plus per-step tool/thinking events; the old
/// 1024-slot buffer overflowed in those windows and silently dropped tool and
/// thinking rows from the live view. Sized for generous headroom across bursts
/// and brief render stalls.
const EVENT_BUS_CAPACITY: usize = 16_384;
const FINAL_ANSWER_PHASE: &str = "final_answer";
pub(crate) const TASK_LEDGER_TOOL_NAME: &str = "task_ledger.update";
const TASK_LEDGER_COMPLETION_REMINDER_LIMIT: u8 = 2;
const TASK_LEDGER_SCOREABLE_CHECKPOINT_SECONDS: u64 = 180;
const TASK_LEDGER_SCOREABLE_CHECKPOINT_LIMIT: u8 = 1;
pub(crate) const MIN_CHILD_DEADLINE_SECONDS: u64 = 2;
const MODEL_PROFILE_TRACE_KIND: &str = "model_profile_segment";
const MODEL_SWITCH_SUMMARY_PREFIX: &str = "Model switch summary:";
const PROVIDER_CLEANUP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum InferenceTimeoutAction {
ScoreableCheckpoint,
Finalization,
}
#[derive(Debug, Clone)]
pub struct RuntimeConfig {
pub default_provider: String,
pub default_model: String,
pub reasoning: Option<String>,
pub auto_compact_token_limit: Option<u32>,
pub file_backed_dynamic_context: bool,
pub hosted_web_search: HostedWebSearchConfig,
pub tool_search: ToolSearchConfig,
pub provider_tool_search: HashMap<String, ToolSearchConfigOverlay>,
pub model_tool_search: HashMap<String, ToolSearchConfigOverlay>,
pub model_edit_tools: HashMap<String, String>,
pub model_parallel_tool_calls: HashMap<String, bool>,
pub model_profiles: HashMap<String, ModelHarnessProfile>,
pub tool_allowlist: Vec<String>,
/// Seconds a host-executed external tool call may stay unresolved before it fails with a timeout error.
pub external_tool_timeout_seconds: u64,
pub command_shell: String,
pub workspace: Option<String>,
pub policy_mode: PolicyMode,
/// Whether agent-swarm mode is active (roadmap 104). When on, the runtime
/// injects the swarm reminder into each turn's developer instructions so any
/// client benefits, not just the TUI.
pub agent_swarm_mode: bool,
pub runtime_profile: RuntimeProfile,
pub inference_router: RuntimeInferenceRouterConfig,
pub speed_policy: RuntimeSpeedPolicyConfig,
pub dynamic_workflows: RuntimeDynamicWorkflowConfig,
pub reliability: RuntimeReliabilityConfig,
pub turn_deadline_seconds: Option<u64>,
pub remote_runner_destination: Option<RunnerDestination>,
pub team_data_dir: Option<PathBuf>,
pub roadmap_data_dir: Option<PathBuf>,
pub media_generation: crate::media_generation::RuntimeMediaGenerationConfig,
}
impl Default for RuntimeConfig {
fn default() -> Self {
Self {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "mock".to_string(),
reasoning: None,
auto_compact_token_limit: None,
file_backed_dynamic_context: true,
hosted_web_search: HostedWebSearchConfig::cached(),
tool_search: ToolSearchConfig::default(),
provider_tool_search: HashMap::new(),
model_tool_search: HashMap::new(),
model_edit_tools: HashMap::new(),
model_parallel_tool_calls: HashMap::new(),
model_profiles: HashMap::new(),
tool_allowlist: Vec::new(),
external_tool_timeout_seconds: DEFAULT_EXTERNAL_TOOL_TIMEOUT_SECONDS,
command_shell: roder_api::command_shell::default_command_shell(),
workspace: None,
policy_mode: PolicyMode::Default,
agent_swarm_mode: false,
runtime_profile: RuntimeProfile::Interactive,
inference_router: RuntimeInferenceRouterConfig::default(),
speed_policy: RuntimeSpeedPolicyConfig::default(),
dynamic_workflows: RuntimeDynamicWorkflowConfig::default(),
reliability: RuntimeReliabilityConfig::default(),
turn_deadline_seconds: None,
remote_runner_destination: None,
team_data_dir: None,
roadmap_data_dir: None,
media_generation: crate::media_generation::RuntimeMediaGenerationConfig::default(),
}
}
}
#[derive(Debug, Clone)]
pub struct StartTurnRequest {
pub thread_id: ThreadId,
pub message: String,
pub images: Vec<InputImage>,
pub provider_override: Option<String>,
pub model_override: Option<String>,
pub reasoning_override: Option<String>,
pub workspace: String,
pub instructions: InstructionBundle,
/**
* Per-turn developer-authority context for this turn's InstructionBundle.
* Applies to every inference round of the turn, is never written to
* thread state, and does not carry over to later turns.
*/
pub developer_context: Option<String>,
pub task_ledger_required: bool,
}
#[derive(Debug, Clone)]
pub struct CreateThreadRequest {
pub title: Option<String>,
pub workspace: String,
pub workspace_id: Option<String>,
pub root_id: Option<String>,
pub provider: Option<String>,
pub model: Option<String>,
pub selection_mode: Option<ModelSelectionMode>,
/// Per-thread tool filter applied on top of the runtime allowlist. Empty = no filtering.
pub tool_allowlist: Vec<String>,
/// Host-supplied instructions added to the developer slot of every turn's inference request.
pub developer_instructions: Option<String>,
/// Host-executed tool specs advertised to the model on every turn of this thread.
pub external_tools: Vec<roder_api::tools::ToolSpec>,
/// Explicit remote-runner binding for the thread's native coding tools.
pub runner: Option<ThreadRunnerSelection>,
}
/**
* Thread-level remote-runner selection. The destination config is persisted
* with the thread, so secrets must reach the provider through its
* environment, not this config.
*/
#[derive(Debug, Clone)]
pub struct ThreadRunnerSelection {
pub provider_id: String,
pub config: serde_json::Value,
/// Absolute path on the runner used as the thread's coding-tool workspace root.
pub workspace: String,
/**
* Extra absolute runner paths file reads may resolve under, beyond
* `workspace`. Writes and the working directory stay confined to
* `workspace`.
*/
pub read_roots: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingPlanExit {
pub thread_id: ThreadId,
pub turn_id: TurnId,
pub request_id: String,
pub target_mode: PolicyMode,
pub plan_summary: Option<String>,
pub next_steps: Vec<String>,
pub requested_at: OffsetDateTime,
pub expires_at: Option<OffsetDateTime>,
}
pub(crate) struct PendingToolApproval {
pub(crate) thread_id: ThreadId,
pub(crate) turn_id: TurnId,
pub(crate) tool_id: String,
pub(crate) tool_name: String,
pub(crate) call: roder_api::tools::ToolCall,
pub(crate) tx: oneshot::Sender<bool>,
}
pub(crate) struct PendingUserInput {
pub(crate) thread_id: ThreadId,
pub(crate) turn_id: TurnId,
pub(crate) tx: oneshot::Sender<serde_json::Value>,
}
/// Host answer to an external tool call delivered via `tools/resolve`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExternalToolResolution {
pub output: String,
pub is_error: bool,
}
pub(crate) struct PendingExternalToolCall {
pub(crate) thread_id: ThreadId,
pub(crate) turn_id: TurnId,
pub(crate) tool_id: String,
pub(crate) tool_name: String,
pub(crate) tx: oneshot::Sender<ExternalToolResolution>,
}
#[derive(Clone)]
struct ActiveTurnHandle {
thread_id: ThreadId,
abort: AbortHandle,
steers: Arc<Mutex<Vec<QueuedTurnSteer>>>,
drain: Arc<TurnDrainHandle>,
}
/// Tracks a task after it has been interrupted and removed from the interactive
/// active-turn map. This lets users start a follow-up turn immediately while a
/// bounded runtime shutdown can still await the original task's cleanup.
struct TurnDrainHandle {
thread_id: ThreadId,
interrupt_requested: AtomicBool,
interrupt_reason: Mutex<Option<TurnLifecycleReason>>,
completed: AtomicBool,
completed_notify: Notify,
}
/// Result of a bounded runtime drain. `Clean` proves the runtime's own turn
/// tasks have reached their cleanup paths; provider-specific child-process
/// reaping remains an explicit provider capability.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RuntimeDrainOutcome {
Clean {
interrupted_turn_ids: Vec<TurnId>,
},
DeadlineExceeded {
interrupted_turn_ids: Vec<TurnId>,
remaining_turn_ids: Vec<TurnId>,
},
/// At least one lifecycle transition initiated by this drain could not be
/// durably appended. The runtime still made its best cleanup attempt, but
/// callers must not claim a persisted terminal state as proof of recovery.
PersistenceFailed {
interrupted_turn_ids: Vec<TurnId>,
remaining_turn_ids: Vec<TurnId>,
},
}
#[derive(Clone)]
struct QueuedTurnSteer {
message: UserMessage,
mailbox_ack: Option<MailboxDeliveryAck>,
}
#[derive(Clone)]
struct MailboxDeliveryAck {
team_id: TeamId,
message_ids: Vec<String>,
}
#[derive(Clone)]
struct InheritedTurnContext {
workspace: String,
instructions: InstructionBundle,
developer_context: Option<String>,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ThreadActivity {
pub active_turn_id: Option<TurnId>,
pub active_flags: Vec<String>,
}
/// Per-thread settings persisted at thread creation and applied to every turn.
#[derive(Debug, Clone, Default)]
pub(crate) struct ThreadTurnOverrides {
pub(crate) tool_allowlist: Vec<String>,
pub(crate) developer_instructions: Option<String>,
pub(crate) external_tools: Vec<roder_api::tools::ToolSpec>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum TurnRunOutcome {
Completed,
Stopped,
}
impl PendingPlanExit {
pub fn new(
thread_id: ThreadId,
turn_id: TurnId,
request_id: String,
target_mode: PolicyMode,
plan_summary: Option<String>,
next_steps: Vec<String>,
) -> Self {
let requested_at = OffsetDateTime::now_utc();
Self {
thread_id,
turn_id,
request_id,
target_mode,
plan_summary,
next_steps,
requested_at,
expires_at: Some(requested_at + default_plan_exit_timeout()),
}
}
pub fn is_expired(&self, now: OffsetDateTime) -> bool {
self.expires_at.is_some_and(|expires_at| now >= expires_at)
}
}
pub fn default_plan_exit_timeout() -> Duration {
Duration::minutes(10)
}
pub const DEFAULT_EXTERNAL_TOOL_TIMEOUT_SECONDS: u64 = 300;
fn format_mailbox_messages(team: &TeamState, messages: &[TeamMailboxMessage]) -> String {
messages
.iter()
.map(|message| {
let recipient = team
.members
.iter()
.find(|member| member.id == message.to_member_id)
.map(canonical_team_member_path)
.unwrap_or_else(|| format!("/root/{}", message.to_member_id));
let sender = message
.from_member_id
.as_deref()
.and_then(|id| team.members.iter().find(|member| member.id == id))
.map(canonical_team_member_path)
.unwrap_or_else(|| "/root".to_string());
let message_type = match message.kind {
TeamMailboxMessageKind::Message => "MESSAGE",
TeamMailboxMessageKind::NewTask => "NEW_TASK",
TeamMailboxMessageKind::FinalAnswer => "FINAL_ANSWER",
};
format!(
"Message Type: {message_type}\nTask name: {recipient}\nSender: {sender}\nPayload:\n{}",
message.text
)
})
.collect::<Vec<_>>()
.join("\n\n")
}
fn canonical_team_member_path(member: &TeamMemberDescriptor) -> String {
if let Some(agent_path) = member
.agent_path
.as_deref()
.filter(|path| *path == "/root" || path.starts_with("/root/"))
{
return agent_path.to_string();
}
if member.role == TeamMemberRole::Lead {
return "/root".to_string();
}
member
.task_name
.as_deref()
.filter(|name| !name.trim().is_empty())
.map(|name| format!("/root/{}", name.trim().trim_matches('/')))
.unwrap_or_else(|| format!("/root/{}", member.id))
}
fn is_codex_v2_team(team: &TeamState) -> bool {
team.members.iter().any(|member| {
member
.model
.as_deref()
.is_some_and(|model| model_supports_reasoning_effort(model, REASONING_ULTRA))
})
}
fn runtime_local_team_view(
mut team: TeamState,
active_thread_ids: &std::collections::HashSet<ThreadId>,
) -> TeamState {
for member in &mut team.members {
if member.status == TeamMemberStatus::Running
&& !active_thread_ids.contains(&member.thread_id)
{
member.status = TeamMemberStatus::Interrupted;
member.current_turn_id = None;
}
}
team
}
type RunnerToolExecutionKey = (String, String);
type RunnerToolExecutionMutex = Mutex<()>;
type RunnerToolExecutionLockRegistry =
Mutex<HashMap<RunnerToolExecutionKey, Weak<RunnerToolExecutionMutex>>>;
pub struct Runtime {
pub bus: EventBus,
pub registry: ExtensionRegistry,
config: RwLock<RuntimeConfig>,
pending_plan_exit: RwLock<Option<PendingPlanExit>>,
pub(crate) pending_tool_approvals: Mutex<HashMap<String, PendingToolApproval>>,
pub(crate) pending_user_inputs: Mutex<HashMap<String, PendingUserInput>>,
pub(crate) pending_external_tool_calls: Mutex<HashMap<String, PendingExternalToolCall>>,
/// Serializes turn admission with shutdown quiescing. A drain takes this
/// gate before flipping `accepting_turns` and snapshotting active work so a
/// turn that observed the old flag cannot be inserted after the snapshot.
turn_admission: Mutex<()>,
active_turns: RwLock<HashMap<TurnId, ActiveTurnHandle>>,
turn_drains: RwLock<HashMap<TurnId, Arc<TurnDrainHandle>>>,
/// Provider cleanup handles register synchronously through the inference
/// tool-executor context. They are intentionally separate from active turn
/// admission so an interrupted turn can remain drainable after a follow-up
/// turn starts on the same thread.
provider_turn_cleanups: std::sync::Mutex<HashMap<TurnId, Arc<dyn ProviderTurnCleanup>>>,
accepting_turns: AtomicBool,
/// Monotonic counter used by bounded drains to surface lifecycle-state
/// persistence failures without making an individual provider turn fail.
lifecycle_persistence_failures: AtomicUsize,
lifecycle_shutdown_drains: AtomicUsize,
lifecycle_clean_shutdowns: AtomicUsize,
lifecycle_deadline_exceeded: AtomicUsize,
lifecycle_persistence_failed_drains: AtomicUsize,
lifecycle_restart_reconciliations: AtomicUsize,
lifecycle_shutdown_drain_duration_ms_total: AtomicUsize,
provider_cleanup_confirmed: AtomicUsize,
provider_cleanup_timed_out: AtomicUsize,
provider_cleanup_unknown: AtomicUsize,
active_turns_changed: Notify,
active_turn_selections: RwLock<HashMap<TurnId, ModelSelectionMode>>,
active_turn_contexts: RwLock<HashMap<TurnId, InheritedTurnContext>>,
/// Process-local execution boundary. Local/CLI runtimes default to true;
/// hosted runtime pools set this false so missing runner metadata fails
/// closed instead of exposing the host workspace.
allow_local_workspaces: AtomicBool,
// Spawn-time live authority retained for every reusable turn of a long-lived teammate.
// This stays process-local because developer_context is explicitly volatile and must never
// be persisted to thread state.
team_member_turn_contexts: Mutex<HashMap<ThreadId, InheritedTurnContext>>,
workspace: PathBuf,
teams: TeamManager,
agent_team_spawn_lock: Mutex<()>,
pub(crate) roadmaps: Mutex<roder_roadmap::RoadmapRuntime>,
pub(crate) goals: Arc<RuntimeGoalController>,
context_artifacts: roder_api::artifacts::ContextArtifactStore,
pub(crate) thread_store: Option<Arc<dyn ThreadStore>>,
thread_item_cache: Mutex<ThreadItemCache>,
pub(crate) tool_registry: ToolRegistry,
media_generation: Arc<crate::media_generation::MediaGenerationService>,
pub(crate) skills: RwLock<SkillRegistry>,
/// Lazily-started bounded dispatch of emitted events to registry
/// `EventSink`s (process extensions etc.); see `event_sink_dispatch`.
event_sink_dispatcher: tokio::sync::OnceCell<crate::event_sink_dispatch::EventSinkDispatcher>,
pub(crate) compaction_hysteresis: std::sync::Mutex<HashMap<ThreadId, u32>>,
/// Per-thread agent-swarm mode overrides (roadmap 104). A thread present in
/// this map uses its stored value; absent threads fall back to the
/// runtime-global `RuntimeConfig.agent_swarm_mode` default. This mirrors the
/// team per-member policy-mode override idiom so swarm mode is per-thread
/// like the other `thread/*` operations, without a separate runtime.
agent_swarm_modes: RwLock<HashMap<ThreadId, bool>>,
/**
* Live remote-runner sessions keyed by thread. Each per-thread mutex is
* held across provider initialization, making the first workspace-tool
* access a singleflight while allowing unrelated threads to initialize in
* parallel. Failed initialization is shared with current waiters, then
* evicted so a later tool call can retry.
*/
runner_sessions: Arc<Mutex<HashMap<ThreadId, Arc<RunnerSessionSlot>>>>,
/**
* Outer tool-call locks keyed by the actual remote session, not the
* per-thread destination id. Hosted runtimes are tenant-scoped, so this
* serializes threads that share one sandbox without coupling different
* tenants or independent runner sessions.
*/
runner_tool_execution_locks: RunnerToolExecutionLockRegistry,
}
#[derive(Clone)]
struct CachedRunnerSession {
destination: RunnerDestination,
session: Arc<dyn RemoteRunnerSession>,
}
struct RunnerSessionSlot {
provider_id: String,
destination_id: String,
state: Mutex<RunnerSessionSlotState>,
initialized: Notify,
}
enum RunnerSessionSlotState {
Empty,
Initializing,
Ready(CachedRunnerSession),
Failed(Arc<str>),
}
impl RunnerSessionSlot {
fn empty(destination: &RunnerDestination) -> Self {
Self {
provider_id: destination.provider_id.clone(),
destination_id: destination.id.clone(),
state: Mutex::new(RunnerSessionSlotState::Empty),
initialized: Notify::new(),
}
}
fn ready(session: CachedRunnerSession) -> Self {
Self {
provider_id: session.destination.provider_id.clone(),
destination_id: session.destination.id.clone(),
state: Mutex::new(RunnerSessionSlotState::Ready(session)),
initialized: Notify::new(),
}
}
fn matches(&self, destination: &RunnerDestination) -> bool {
self.provider_id == destination.provider_id && self.destination_id == destination.id
}
}
impl Runtime {
pub fn new(registry: ExtensionRegistry, config: RuntimeConfig) -> anyhow::Result<Self> {
if registry.inference_engines.is_empty() {
anyhow::bail!("at least one inference engine must be registered");
}
validate_runtime_config_reasoning(&config)?;
validate_runtime_inference_router_config(®istry, &config)?;
let bus = EventBus::new(EVENT_BUS_CAPACITY);
let thread_store = registry
.thread_stores
.first()
.map(|factory| factory.create());
let mut tool_registry = ToolRegistry::default();
for contributor in ®istry.tools {
contributor
.contribute(&mut tool_registry)
.with_context(|| format!("tool contributor {} failed", contributor.id()))?;
}
crate::agent_control_tools::contribute_agent_control_tools(&mut tool_registry)?;
let media_generation = Arc::new(crate::media_generation::MediaGenerationService::new(
registry.media_generator_providers.clone(),
config.media_generation.clone(),
));
tool_registry.replace(Arc::new(
crate::media_generation::MediaGenerateImageTool::new(media_generation.clone()),
));
let team_data_dir = config.team_data_dir.clone();
let workspace = config
.workspace
.clone()
.map(PathBuf::from)
.unwrap_or(std::env::current_dir()?);
let roadmap_data_dir = config
.roadmap_data_dir
.clone()
.unwrap_or_else(|| workspace.join(".roder"));
let context_artifacts = thread_store
.as_ref()
.and_then(|store| store.context_artifact_store())
.or_else(|| {
thread_store
.as_ref()
.and_then(|store| store.local_thread_root())
.map(FilesystemContextArtifactStore::shared_thread_scoped)
})
.unwrap_or_else(|| {
FilesystemContextArtifactStore::shared_legacy(default_context_artifact_dir())
});
let goals = Arc::new(RuntimeGoalController::new(
bus.clone(),
thread_store.clone(),
));
let runtime = Self {
bus,
registry,
config: RwLock::new(config),
pending_plan_exit: RwLock::new(None),
pending_tool_approvals: Mutex::new(HashMap::new()),
pending_user_inputs: Mutex::new(HashMap::new()),
pending_external_tool_calls: Mutex::new(HashMap::new()),
turn_admission: Mutex::new(()),
active_turns: RwLock::new(HashMap::new()),
turn_drains: RwLock::new(HashMap::new()),
provider_turn_cleanups: std::sync::Mutex::new(HashMap::new()),
accepting_turns: AtomicBool::new(true),
lifecycle_persistence_failures: AtomicUsize::new(0),
lifecycle_shutdown_drains: AtomicUsize::new(0),
lifecycle_clean_shutdowns: AtomicUsize::new(0),
lifecycle_deadline_exceeded: AtomicUsize::new(0),
lifecycle_persistence_failed_drains: AtomicUsize::new(0),
lifecycle_restart_reconciliations: AtomicUsize::new(0),
lifecycle_shutdown_drain_duration_ms_total: AtomicUsize::new(0),
provider_cleanup_confirmed: AtomicUsize::new(0),
provider_cleanup_timed_out: AtomicUsize::new(0),
provider_cleanup_unknown: AtomicUsize::new(0),
active_turns_changed: Notify::new(),
active_turn_selections: RwLock::new(HashMap::new()),
active_turn_contexts: RwLock::new(HashMap::new()),
allow_local_workspaces: AtomicBool::new(true),
team_member_turn_contexts: Mutex::new(HashMap::new()),
workspace: workspace.clone(),
teams: TeamManager::new(
team_data_dir.unwrap_or_else(crate::teams::default_team_data_dir),
),
agent_team_spawn_lock: Mutex::new(()),
roadmaps: Mutex::new(roder_roadmap::RoadmapRuntime::new(
workspace,
roadmap_data_dir,
)),
goals,
context_artifacts,
thread_store,
thread_item_cache: Mutex::new(ThreadItemCache::default()),
tool_registry,
media_generation,
skills: RwLock::new(SkillRegistry::load(SkillRegistryOptions::new(
PathBuf::new(),
))),
event_sink_dispatcher: tokio::sync::OnceCell::new(),
compaction_hysteresis: crate::compaction_runtime::compaction_hysteresis_state(),
agent_swarm_modes: RwLock::new(HashMap::new()),
runner_sessions: Arc::new(Mutex::new(HashMap::new())),
runner_tool_execution_locks: Mutex::new(HashMap::new()),
};
runtime.bus.emit(RoderEvent::RuntimeStarted(RuntimeStarted {
timestamp: OffsetDateTime::now_utc(),
}));
for manifest in &runtime.registry.manifests {
runtime
.bus
.emit(RoderEvent::ExtensionRegistered(ExtensionRegistered {
extension_id: manifest.id.clone(),
timestamp: OffsetDateTime::now_utc(),
}));
}
Ok(runtime)
}
pub fn from_engine(engine: Arc<dyn InferenceEngine>) -> anyhow::Result<Self> {
let mut builder = roder_api::extension::ExtensionRegistryBuilder::new();
builder.inference_engine(engine);
Self::new(builder.build()?, RuntimeConfig::default())
}
pub fn fake() -> anyhow::Result<Self> {
Self::from_engine(Arc::new(FakeInferenceEngine))
}
pub fn subscribe_events(&self) -> tokio::sync::broadcast::Receiver<EventEnvelope> {
self.bus.subscribe()
}
/// Returns process-local, redacted lifecycle health counters. The snapshot
/// intentionally has fixed fields and contains no provider, command,
/// process, thread, or turn identifiers.
pub fn lifecycle_metrics(&self) -> LifecycleMetricsSnapshot {
LifecycleMetricsSnapshot {
shutdown_drain_count: self.lifecycle_shutdown_drains.load(Ordering::Acquire) as u64,
clean_shutdown_count: self.lifecycle_clean_shutdowns.load(Ordering::Acquire) as u64,
deadline_exceeded_count: self.lifecycle_deadline_exceeded.load(Ordering::Acquire)
as u64,
persistence_failed_count: self
.lifecycle_persistence_failed_drains
.load(Ordering::Acquire) as u64,
restart_reconciliation_count: self
.lifecycle_restart_reconciliations
.load(Ordering::Acquire) as u64,
lifecycle_persistence_failure_count: self
.lifecycle_persistence_failures
.load(Ordering::Acquire) as u64,
shutdown_drain_duration_ms_total: self
.lifecycle_shutdown_drain_duration_ms_total
.load(Ordering::Acquire) as u64,
provider_cleanup_confirmed_count: self
.provider_cleanup_confirmed
.load(Ordering::Acquire) as u64,
provider_cleanup_timed_out_count: self
.provider_cleanup_timed_out
.load(Ordering::Acquire) as u64,
provider_cleanup_unknown_count: self.provider_cleanup_unknown.load(Ordering::Acquire)
as u64,
}
}
async fn persist_turn_lifecycle_record(&self, record: &TurnLifecycleRecord) -> bool {
let Some(store) = &self.thread_store else {
return true;
};
let state = match record.extension_state() {
Ok(state) => state,
Err(_) => {
self.lifecycle_persistence_failures
.fetch_add(1, Ordering::AcqRel);
return false;
}
};
// Lifecycle diagnostics must not turn a provider result into a failed
// turn merely because a third-party store lacks extension-state support.
// JSONL/Postgres stores persist this append-only record atomically at
// their own storage boundary.
if store
.append_extension_state(&record.thread_id, &state)
.await
.is_err()
{
self.lifecycle_persistence_failures
.fetch_add(1, Ordering::AcqRel);
return false;
}
true
}
async fn record_turn_lifecycle(
&self,
thread_id: ThreadId,
turn_id: TurnId,
state: TurnLifecycleState,
cleanup: TurnCleanupState,
reason: Option<TurnLifecycleReason>,
) -> TurnLifecycleRecord {
self.record_turn_lifecycle_with_ownership(
thread_id,
turn_id,
state,
cleanup,
reason,
TurnCleanupOwnership::RuntimeTaskOnly,
)
.await
}
async fn record_turn_lifecycle_with_ownership(
&self,
thread_id: ThreadId,
turn_id: TurnId,
state: TurnLifecycleState,
cleanup: TurnCleanupState,
reason: Option<TurnLifecycleReason>,
ownership: TurnCleanupOwnership,
) -> TurnLifecycleRecord {
let record = TurnLifecycleRecord::new(
thread_id,
turn_id,
state,
cleanup,
reason,
OffsetDateTime::now_utc(),
)
.with_ownership(ownership);
let _ = self.persist_turn_lifecycle_record(&record).await;
self.emit(RoderEvent::TurnLifecycleUpdated(record.clone()))
.await;
record
}
pub(crate) fn register_provider_turn_cleanup(
&self,
turn_id: &TurnId,
cleanup: Arc<dyn ProviderTurnCleanup>,
) {
if let Ok(mut cleanups) = self.provider_turn_cleanups.lock() {
cleanups.insert(turn_id.clone(), cleanup);
}
}
fn provider_cleanup_ownership(&self, turn_id: &TurnId) -> TurnCleanupOwnership {
self.provider_turn_cleanups
.lock()
.ok()
.and_then(|cleanups| cleanups.get(turn_id).cloned())
.map(|cleanup| cleanup.ownership())
.unwrap_or(TurnCleanupOwnership::RuntimeTaskOnly)
}
async fn await_provider_turn_cleanup(
&self,
turn_id: &TurnId,
) -> (TurnCleanupState, TurnCleanupOwnership) {
let cleanup = self
.provider_turn_cleanups
.lock()
.ok()
.and_then(|mut cleanups| cleanups.remove(turn_id));
let Some(cleanup) = cleanup else {
return (
TurnCleanupState::Unknown,
TurnCleanupOwnership::RuntimeTaskOnly,
);
};
let ownership = cleanup.ownership();
match tokio::time::timeout(PROVIDER_CLEANUP_TIMEOUT, cleanup.wait_for_cleanup()).await {
Ok(Ok(())) => {
self.provider_cleanup_confirmed
.fetch_add(1, Ordering::AcqRel);
(
TurnCleanupState::Completed,
TurnCleanupOwnership::ProviderCleanupConfirmed,
)
}
Ok(Err(_)) => {
self.provider_cleanup_unknown.fetch_add(1, Ordering::AcqRel);
(TurnCleanupState::Unknown, ownership)
}
Err(_) => {
self.provider_cleanup_timed_out
.fetch_add(1, Ordering::AcqRel);
(TurnCleanupState::TimedOut, ownership)
}
}
}
fn record_lifecycle_drain_outcome(
&self,
started_at: tokio::time::Instant,
outcome: &RuntimeDrainOutcome,
) {
self.lifecycle_shutdown_drains
.fetch_add(1, Ordering::AcqRel);
self.lifecycle_shutdown_drain_duration_ms_total.fetch_add(
started_at.elapsed().as_millis().min(usize::MAX as u128) as usize,
Ordering::AcqRel,
);
match outcome {
RuntimeDrainOutcome::Clean { .. } => {
self.lifecycle_clean_shutdowns
.fetch_add(1, Ordering::AcqRel);
}
RuntimeDrainOutcome::DeadlineExceeded { .. } => {
self.lifecycle_deadline_exceeded
.fetch_add(1, Ordering::AcqRel);
}
RuntimeDrainOutcome::PersistenceFailed { .. } => {
self.lifecycle_persistence_failed_drains
.fetch_add(1, Ordering::AcqRel);
}
}
}
fn lifecycle_record_for_event(event: &RoderEvent) -> Option<TurnLifecycleRecord> {
match event {
RoderEvent::TurnStarted(event) => Some(TurnLifecycleRecord::new(
event.thread_id.clone(),
event.turn_id.clone(),
TurnLifecycleState::Running,
TurnCleanupState::NotRequested,
None,
event.timestamp,
)),
// `run_turn` persists completed turns after it has awaited a
// registered provider cleanup handle. Do not let the generic event
// projection overwrite that acknowledgement with an unproven state.
RoderEvent::TurnCompleted(_) => None,
// Some failure paths already have an event before the turn wrapper
// reaches its cleanup acknowledgement. Preserve a durable fallback
// reason here; the wrapper appends a later record with the more
// precise cleanup outcome when a provider registered one.
RoderEvent::TurnFailed(event) => Some(TurnLifecycleRecord::new(
event.thread_id.clone(),
event.turn_id.clone(),
TurnLifecycleState::Failed,
TurnCleanupState::Unknown,
Some(if event.error_kind.as_deref() == Some("deadline_timeout") {
TurnLifecycleReason::DeadlineExceeded
} else {
TurnLifecycleReason::ProviderFailure
}),
event.timestamp,
)),
RoderEvent::TurnDeadlineExceeded(_) | RoderEvent::TurnInterrupted(_) => None,
RoderEvent::TurnLifecycleUpdated(_) => None,
_ => None,
}
}
pub fn registry(&self) -> &ExtensionRegistry {
&self.registry
}
pub fn media_generation(&self) -> Arc<crate::media_generation::MediaGenerationService> {
self.media_generation.clone()
}
pub fn context_artifacts(&self) -> roder_api::artifacts::ContextArtifactStore {
self.context_artifacts.clone()
}
pub async fn execute_workflow_tool(
&self,
thread_id: ThreadId,
tool_name: &str,
arguments: serde_json::Value,
) -> anyhow::Result<ToolResult> {
let Some(executor) = self.tool_registry.get(tool_name) else {
anyhow::bail!("tool not found: {tool_name}");
};
let tool_call = ToolCall {
id: format!("slash-{tool_name}"),
name: tool_name.to_string(),
raw_arguments: serde_json::to_string(&arguments)?,
arguments,
thread_id: thread_id.clone(),
turn_id: "slash-command".to_string(),
};
let runtime_config = self.status().await;
let ctx = self.tool_execution_context(
thread_id,
"slash-command".to_string(),
runtime_config.policy_mode,
runtime_config.workspace.as_deref(),
Some(&runtime_config.command_shell),
);
executor.execute(ctx, tool_call).await
}
pub(crate) fn tool_execution_context(
&self,
thread_id: ThreadId,
turn_id: TurnId,
mode: PolicyMode,
workspace: Option<&str>,
command_shell: Option<&str>,
) -> ToolExecutionContext {
let mut ctx = ToolExecutionContext::new(thread_id, turn_id, mode)
.with_command_shell(command_shell.unwrap_or_default())
.with_process_runner(Arc::new(LocalProcessRunner))
.with_context_artifacts(self.context_artifacts.backend())
.with_goal_controller(self.goals.clone())
.with_subagent_trace_sink(Arc::new(RuntimeSubagentTraceSink::new(
self.bus.clone(),
self.thread_store.clone(),
)))
.with_swarm_progress_sink(Arc::new(RuntimeAgentSwarmProgressSink::new(
self.bus.clone(),
self.thread_store.clone(),
)));
if let Some(workspace) = workspace {
ctx = ctx.with_workspace_handle(Arc::new(ScopedFilesystem::new(workspace)));
}
ctx
}
pub async fn status(&self) -> RuntimeConfig {
self.config.read().await.clone()
}
pub async fn set_skills(&self, skills: SkillRegistry) {
*self.skills.write().await = skills;
}
pub async fn skills_snapshot(&self) -> SkillRegistry {
self.skills.read().await.clone()
}
pub fn workspace(&self) -> PathBuf {
self.workspace.clone()
}
/// Controls whether native workspace tools may execute without an
/// explicit remote-runner binding. Hosted runtimes disable this after
/// construction; ordinary local runtimes retain the compatible default.
pub fn set_allow_local_workspaces(&self, allowed: bool) {
self.allow_local_workspaces
.store(allowed, Ordering::Relaxed);
}
pub fn allows_local_workspaces(&self) -> bool {
self.allow_local_workspaces.load(Ordering::Relaxed)
}
pub async fn set_remote_runner_destination(&self, destination: Option<RunnerDestination>) {
let lifecycle = destination.as_ref().map(|destination| RunnerLifecycle {
destination_id: destination.id.clone(),
provider_id: destination.provider_id.clone(),
state: "configured".to_string(),
session_id: None,
timestamp: OffsetDateTime::now_utc(),
});
self.config.write().await.remote_runner_destination = destination;
if let Some(lifecycle) = lifecycle {
self.emit(RoderEvent::RunnerLifecycle(lifecycle)).await;
} else {
self.emit(RoderEvent::RunnerLifecycle(RunnerLifecycle {
destination_id: "local".to_string(),
provider_id: "local".to_string(),
state: "local_fallback".to_string(),
session_id: None,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
}
}
pub async fn set_file_backed_dynamic_context(&self, enabled: bool) -> RuntimeConfig {
let mut cfg = self.config.write().await;
cfg.file_backed_dynamic_context = enabled;
cfg.clone()
}
pub async fn set_command_shell(&self, shell: String) -> RuntimeConfig {
let mut cfg = self.config.write().await;
cfg.command_shell = shell;
cfg.clone()
}
pub async fn pending_plan_exit(&self) -> Option<PendingPlanExit> {
let mut pending = self.pending_plan_exit.write().await;
let current = pending.clone()?;
if !current.is_expired(OffsetDateTime::now_utc()) {
return Some(current);
}
*pending = None;
drop(pending);
self.emit_plan_exit_resolved(¤t, false, self.status().await.policy_mode)
.await;
None
}
pub async fn set_policy_mode(
&self,
mode: PolicyMode,
reason: Option<String>,
) -> anyhow::Result<RuntimeConfig> {
let mut cfg = self.config.write().await;
let previous_mode = cfg.policy_mode;
cfg.policy_mode = mode;
let next = cfg.clone();
drop(cfg);
self.emit(RoderEvent::PolicyModeChanged(PolicyModeChanged {
thread_id: "runtime".to_string(),
turn_id: None,
previous_mode,
new_mode: mode,
reason,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
self.auto_resolve_pending_tool_approvals_for_mode(mode)
.await;
Ok(next)
}
/// Toggle agent-swarm mode for the runtime (roadmap 104). When enabled, the
/// swarm reminder is injected into each turn's developer instructions so
/// every app-server/SDK client gets it, not only the TUI.
pub async fn set_agent_swarm_mode(
&self,
enabled: bool,
trigger: roder_api::subagents::AgentSwarmModeTrigger,
) -> anyhow::Result<RuntimeConfig> {
let mut cfg = self.config.write().await;
cfg.agent_swarm_mode = enabled;
let next = cfg.clone();
drop(cfg);
self.emit(RoderEvent::AgentSwarmModeChanged(
roder_api::subagents::AgentSwarmModeChanged {
thread_id: "runtime".to_string(),
turn_id: None,
enabled,
trigger,
timestamp: OffsetDateTime::now_utc(),
},
))
.await;
Ok(next)
}
/// Toggle agent-swarm mode for a single thread (roadmap 104). The override is
/// stored per thread so toggling swarm mode on one thread does not leak the
/// reminder into other threads sharing the runtime; absent threads fall back
/// to the runtime-global default. The emitted `AgentSwarmModeChanged` event
/// carries the real `thread_id`.
pub async fn set_agent_swarm_mode_for_thread(
&self,
thread_id: &str,
enabled: bool,
trigger: roder_api::subagents::AgentSwarmModeTrigger,
) -> bool {
{
let mut modes = self.agent_swarm_modes.write().await;
modes.insert(thread_id.to_string(), enabled);
}
self.emit(RoderEvent::AgentSwarmModeChanged(
roder_api::subagents::AgentSwarmModeChanged {
thread_id: thread_id.to_string(),
turn_id: None,
enabled,
trigger,
timestamp: OffsetDateTime::now_utc(),
},
))
.await;
enabled
}
/// Resolve whether agent-swarm mode is active for `thread_id`: the per-thread
/// override when present, otherwise the runtime-global default. The turn loop
/// uses this to decide whether to inject the swarm reminder.
pub async fn effective_agent_swarm_mode_for_thread(&self, thread_id: &str) -> bool {
if let Some(enabled) = self.agent_swarm_modes.read().await.get(thread_id).copied() {
return enabled;
}
self.status().await.agent_swarm_mode
}
pub async fn set_hosted_web_search(
&self,
mode: HostedWebSearchMode,
) -> anyhow::Result<RuntimeConfig> {
let mut cfg = self.config.write().await;
cfg.hosted_web_search = HostedWebSearchConfig { mode };
Ok(cfg.clone())
}
async fn auto_resolve_pending_tool_approvals_for_mode(&self, mode: PolicyMode) {
let gate = DefaultPolicyGate::new();
let mut pending = self.pending_tool_approvals.lock().await;
let approval_ids = pending
.iter()
.filter_map(|(approval_id, approval)| {
let ctx = ToolExecutionContext::new(
approval.thread_id.clone(),
approval.turn_id.clone(),
mode,
);
matches!(
gate.decide(&approval.call, mode, &ctx),
PolicyDecision::AutoApproved { .. }
)
.then_some(approval_id.clone())
})
.collect::<Vec<_>>();
let approvals = approval_ids
.into_iter()
.filter_map(|approval_id| {
pending
.remove(&approval_id)
.map(|approval| (approval_id, approval))
})
.collect::<Vec<_>>();
drop(pending);
for (approval_id, approval) in approvals {
let ctx = ToolExecutionContext::new(
approval.thread_id.clone(),
approval.turn_id.clone(),
mode,
);
let decision = gate.decide(&approval.call, mode, &ctx);
self.emit(RoderEvent::PolicyDecisionRecorded(PolicyDecisionRecorded {
thread_id: approval.thread_id.clone(),
turn_id: approval.turn_id.clone(),
tool_id: approval.tool_id.clone(),
tool_name: approval.tool_name.clone(),
mode,
decision,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
if mode == PolicyMode::Bypass {
self.emit(RoderEvent::PolicyBypassActive(PolicyBypassActive {
thread_id: approval.thread_id.clone(),
turn_id: approval.turn_id.clone(),
tool_id: approval.tool_id.clone(),
tool_name: approval.tool_name.clone(),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
}
self.emit(RoderEvent::ApprovalResolved(ApprovalResolved {
thread_id: approval.thread_id,
turn_id: approval.turn_id,
approval_id,
tool_id: approval.tool_id,
tool_name: approval.tool_name,
approved: true,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
let _ = approval.tx.send(true);
}
}
pub async fn record_pending_plan_exit(&self, pending: PendingPlanExit) {
*self.pending_plan_exit.write().await = Some(pending.clone());
self.emit(RoderEvent::PolicyExitPlanRequested(
PolicyExitPlanRequested {
thread_id: pending.thread_id,
turn_id: pending.turn_id,
request_id: pending.request_id,
target_mode: pending.target_mode,
plan_summary: pending.plan_summary,
next_steps: pending.next_steps,
timestamp: OffsetDateTime::now_utc(),
},
))
.await;
}
pub async fn resolve_pending_plan_exit(
&self,
request_id: &str,
approved: bool,
) -> anyhow::Result<Option<PendingPlanExit>> {
let mut pending = self.pending_plan_exit.write().await;
let Some(current) = pending.clone() else {
return Ok(None);
};
if current.request_id != request_id {
anyhow::bail!("pending plan exit request {request_id:?} was not found");
}
*pending = None;
drop(pending);
let approved = approved && !current.is_expired(OffsetDateTime::now_utc());
let resolved_mode = if approved {
let mut cfg = self.config.write().await;
let previous_mode = cfg.policy_mode;
cfg.policy_mode = current.target_mode;
drop(cfg);
self.emit(RoderEvent::PolicyModeChanged(PolicyModeChanged {
thread_id: current.thread_id.clone(),
turn_id: Some(current.turn_id.clone()),
previous_mode,
new_mode: current.target_mode,
reason: Some("approved plan exit".to_string()),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
self.auto_resolve_pending_tool_approvals_for_mode(current.target_mode)
.await;
current.target_mode
} else {
self.status().await.policy_mode
};
self.emit_plan_exit_resolved(¤t, approved, resolved_mode)
.await;
Ok(Some(current))
}
pub async fn resolve_tool_approval(
&self,
approval_id: &str,
approved: bool,
) -> anyhow::Result<bool> {
let pending = self.pending_tool_approvals.lock().await.remove(approval_id);
let Some(pending) = pending else {
return Ok(false);
};
self.emit(RoderEvent::ApprovalResolved(ApprovalResolved {
thread_id: pending.thread_id,
turn_id: pending.turn_id,
approval_id: approval_id.to_string(),
tool_id: pending.tool_id,
tool_name: pending.tool_name,
approved,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
let _ = pending.tx.send(approved);
Ok(true)
}
pub async fn request_app_server_tool_approval(
&self,
call: ToolCall,
reason: Option<String>,
) -> anyhow::Result<bool> {
let approval_id = call.id.clone();
let (tx, rx) = oneshot::channel();
self.pending_tool_approvals.lock().await.insert(
approval_id.clone(),
PendingToolApproval {
thread_id: call.thread_id.clone(),
turn_id: call.turn_id.clone(),
tool_id: call.id.clone(),
tool_name: call.name.clone(),
call: call.clone(),
tx,
},
);
self.emit(RoderEvent::ApprovalRequested(ApprovalRequested {
thread_id: call.thread_id.clone(),
turn_id: call.turn_id.clone(),
approval_id,
tool_id: call.id.clone(),
tool_name: call.name.clone(),
reason,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
Ok(rx.await.unwrap_or(false))
}
/// Completes a pending host-executed tool call (`tools/resolve`). Returns false when the
/// request id is unknown, already resolved, timed out, or cancelled by a turn interrupt.
pub async fn resolve_external_tool_call(
&self,
request_id: &str,
resolution: ExternalToolResolution,
) -> anyhow::Result<bool> {
let pending = self
.pending_external_tool_calls
.lock()
.await
.remove(request_id);
let Some(pending) = pending else {
return Ok(false);
};
self.emit(RoderEvent::ExternalToolCallResolved(
ExternalToolCallResolved {
thread_id: pending.thread_id,
turn_id: pending.turn_id,
request_id: request_id.to_string(),
tool_id: pending.tool_id,
tool_name: pending.tool_name,
outcome: ExternalToolCallOutcome::Resolved,
is_error: resolution.is_error,
timestamp: OffsetDateTime::now_utc(),
},
))
.await;
let _ = pending.tx.send(resolution);
Ok(true)
}
/// Drops every pending external tool call for the turn and reports them as cancelled so an
/// interrupt does not leak requests waiting on `tools/resolve`.
async fn cancel_pending_external_tool_calls_for_turn(&self, turn_id: &TurnId) {
let cancelled = {
let mut pending = self.pending_external_tool_calls.lock().await;
let request_ids = pending
.iter()
.filter(|(_, call)| &call.turn_id == turn_id)
.map(|(request_id, _)| request_id.clone())
.collect::<Vec<_>>();
request_ids
.into_iter()
.filter_map(|request_id| pending.remove(&request_id).map(|call| (request_id, call)))
.collect::<Vec<_>>()
};
for (request_id, call) in cancelled {
self.emit(RoderEvent::ExternalToolCallResolved(
ExternalToolCallResolved {
thread_id: call.thread_id,
turn_id: call.turn_id,
request_id,
tool_id: call.tool_id,
tool_name: call.tool_name,
outcome: ExternalToolCallOutcome::Cancelled,
is_error: true,
timestamp: OffsetDateTime::now_utc(),
},
))
.await;
}
}
pub async fn resolve_user_input(
&self,
request_id: &str,
answers: serde_json::Value,
) -> anyhow::Result<bool> {
let pending = self.pending_user_inputs.lock().await.remove(request_id);
let Some(pending) = pending else {
return Ok(false);
};
self.emit(RoderEvent::UserInputResolved(UserInputResolved {
thread_id: pending.thread_id,
turn_id: pending.turn_id,
request_id: request_id.to_string(),
answers: answers.clone(),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
let _ = pending.tx.send(answers);
Ok(true)
}
async fn emit_plan_exit_resolved(
&self,
current: &PendingPlanExit,
approved: bool,
resolved_mode: PolicyMode,
) {
self.emit(RoderEvent::PolicyExitPlanResolved(PolicyExitPlanResolved {
thread_id: current.thread_id.clone(),
turn_id: current.turn_id.clone(),
request_id: current.request_id.clone(),
approved,
target_mode: current.target_mode,
resolved_mode,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
}
pub async fn select_provider(
&self,
provider: String,
model: Option<String>,
reasoning: Option<String>,
) -> anyhow::Result<RuntimeConfig> {
let next = self
.preview_provider_selection(provider, model, reasoning)
.await?;
let mut cfg = self.config.write().await;
*cfg = next;
Ok(cfg.clone())
}
pub async fn preview_provider_selection(
&self,
provider: String,
model: Option<String>,
reasoning: Option<String>,
) -> anyhow::Result<RuntimeConfig> {
self.engine_for(&provider)?;
let mut cfg = self.config.read().await.clone();
cfg.default_provider = provider;
if let Some(model) = model {
cfg.default_model = model;
}
if let Some(reasoning) = reasoning {
if reasoning == REASONING_NONE
&& !model_supports_reasoning(&cfg.default_model, &reasoning)
{
return Ok(cfg.clone());
}
validate_reasoning_effort(&cfg.default_model, &reasoning)?;
cfg.reasoning = Some(reasoning);
}
Ok(cfg)
}
pub async fn effective_reasoning(&self) -> String {
let cfg = self.config.read().await;
effective_reasoning_for_model(&cfg, &cfg.default_model)
}
pub fn effective_reasoning_for_config(cfg: &RuntimeConfig) -> String {
effective_reasoning_for_model(cfg, &cfg.default_model)
}
pub async fn set_dynamic_workflow_effort(
&self,
effort_profile: DynamicWorkflowEffortProfile,
) -> RuntimeConfig {
let mut cfg = self.config.write().await;
cfg.dynamic_workflows.effort_profile = effort_profile;
cfg.clone()
}
pub async fn dynamic_workflow_trigger_decision(
&self,
message: &str,
) -> WorkflowTriggerDecision {
let cfg = self.config.read().await;
classify_workflow_trigger(message, &cfg.dynamic_workflows)
}
pub async fn create_thread(&self, title: Option<String>) -> anyhow::Result<ThreadMetadata> {
self.create_thread_with(CreateThreadRequest {
title,
workspace: self.workspace.display().to_string(),
workspace_id: None,
root_id: None,
provider: None,
model: None,
selection_mode: None,
tool_allowlist: Vec::new(),
developer_instructions: None,
external_tools: Vec::new(),
runner: None,
})
.await
}
/**
* Resolves an explicit thread-runner selection into a persisted binding.
* Fails fast at thread creation when the provider is missing or rejects
* the destination, instead of surfacing the error on the first tool call.
*/
async fn resolve_thread_runner_binding(
&self,
thread_id: &str,
selection: ThreadRunnerSelection,
) -> anyhow::Result<ThreadRunnerBinding> {
let provider = self
.registry
.remote_runner_providers
.iter()
.find(|provider| provider.id() == selection.provider_id)
.cloned()
.ok_or_else(|| {
anyhow::anyhow!(
"remote runner provider {:?} is not installed",
selection.provider_id
)
})?;
let workspace = selection.workspace.trim();
anyhow::ensure!(
std::path::Path::new(workspace).is_absolute(),
"runner workspace must be an absolute path on the runner: {workspace:?}"
);
let mut read_roots = Vec::with_capacity(selection.read_roots.len());
for read_root in &selection.read_roots {
let trimmed = read_root.trim();
anyhow::ensure!(
std::path::Path::new(trimmed).is_absolute(),
"runner read root must be an absolute path on the runner: {trimmed:?}"
);
read_roots.push(PathBuf::from(trimmed));
}
let destination = RunnerDestination {
id: format!("thread-{thread_id}"),
provider_id: selection.provider_id,
config: selection.config,
default_manifest: roder_api::remote_runner::RunnerManifest::default(),
};
provider.validate_destination(&destination).await?;
Ok(ThreadRunnerBinding {
destination,
workspace: PathBuf::from(workspace),
read_roots,
})
}
/**
* Build a per-thread binding from a runtime-level destination (selected via
* the TUI runner picker or config `default_destination`) when the
* destination's provider advertises a default workspace. Returns `None` when
* there is no destination or the provider opts out (no default workspace),
* preserving the legacy behavior where such threads keep local tools.
*/
async fn synthesize_runtime_runner_binding(
&self,
thread_id: &str,
destination: Option<RunnerDestination>,
) -> anyhow::Result<Option<ThreadRunnerBinding>> {
let Some(destination) = destination else {
return Ok(None);
};
let Some(provider) = self
.registry
.remote_runner_providers
.iter()
.find(|provider| provider.id() == destination.provider_id)
else {
return Ok(None);
};
// An explicit workspace in the destination config wins over the
// provider default; absence of both means the provider opts out.
let workspace = destination
.config
.get("working_dir")
.and_then(serde_json::Value::as_str)
.map(str::to_string)
.or_else(|| provider.default_workspace());
let Some(workspace) = workspace else {
return Ok(None);
};
let selection = ThreadRunnerSelection {
provider_id: destination.provider_id.clone(),
config: destination.config.clone(),
workspace,
read_roots: Vec::new(),
};
Ok(Some(
self.resolve_thread_runner_binding(thread_id, selection)
.await?,
))
}
/// Validates a runner selection without creating a thread; the placeholder destination id only appears in error messages.
pub async fn validate_thread_runner_selection(
&self,
selection: ThreadRunnerSelection,
) -> anyhow::Result<()> {
self.resolve_thread_runner_binding("validate", selection)
.await
.map(|_| ())
}
pub async fn create_thread_with(
&self,
req: CreateThreadRequest,
) -> anyhow::Result<ThreadMetadata> {
let cfg = self.config.read().await.clone();
let now = OffsetDateTime::now_utc();
let workspace = validate_thread_workspace(&req.workspace)?;
let provider = req.provider.unwrap_or(cfg.default_provider);
let model = req.model.unwrap_or(cfg.default_model);
let selection_mode = req
.selection_mode
.unwrap_or_else(|| ModelSelectionMode::manual(provider.clone(), model.clone(), None));
let thread_id = uuid::Uuid::new_v4().to_string();
let runner_binding = match req.runner {
Some(selection) => Some(
self.resolve_thread_runner_binding(&thread_id, selection)
.await?,
),
// No explicit per-thread selection: if a runtime-level destination
// is active and its provider advertises a default workspace, bind
// the new thread so its coding tools route into the runner. This is
// what makes a TUI/config-selected runner (e.g. Blaxel) actually
// execute tools in the sandbox instead of locally. Providers that
// return no default workspace stay unbound (legacy local-tools).
None => {
self.synthesize_runtime_runner_binding(
&thread_id,
cfg.remote_runner_destination.clone(),
)
.await?
}
};
let runner_destination = runner_binding
.as_ref()
.map(|binding| binding.destination.clone())
.or_else(|| cfg.remote_runner_destination.clone());
let metadata = ThreadMetadata {
thread_id,
title: req.title,
workspace,
workspace_id: req.workspace_id,
root_id: req.root_id,
provider: Some(provider),
model: Some(model),
selection_mode: Some(selection_mode),
tool_allowlist: req.tool_allowlist,
developer_instructions: req.developer_instructions,
external_tools: req.external_tools,
runner_destination,
runner_state: None,
runner_binding,
created_at: now,
updated_at: now,
message_count: 0,
usage: None,
parent_thread_id: None,
forked_from_turn_id: None,
workspace_fork: None,
};
let metadata = if let Some(store) = &self.thread_store {
store.create_thread(metadata).await?
} else {
metadata
};
self.emit(RoderEvent::ThreadCreated(ThreadCreated {
thread_id: metadata.thread_id.clone(),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
Ok(metadata)
}
pub async fn list_threads(&self) -> anyhow::Result<Vec<ThreadMetadata>> {
if let Some(store) = &self.thread_store {
return store.list_threads().await;
}
Ok(Vec::new())
}
pub async fn list_threads_page(
&self,
options: roder_api::thread::ThreadListOptions,
) -> anyhow::Result<roder_api::thread::ThreadListPage> {
if let Some(store) = &self.thread_store {
return store.list_threads_page(options).await;
}
Ok(roder_api::thread::ThreadListPage::default())
}
pub async fn load_thread_metadata(
&self,
thread_id: &str,
) -> anyhow::Result<Option<roder_api::thread::ThreadMetadata>> {
if let Some(store) = &self.thread_store {
return store.load_thread_metadata(&thread_id.to_string()).await;
}
Ok(None)
}
pub async fn archive_thread(&self, thread_id: &str) -> anyhow::Result<bool> {
let archived = if let Some(store) = &self.thread_store {
store.archive_thread(&thread_id.to_string()).await?
} else {
false
};
if archived {
self.thread_item_cache
.lock()
.await
.remove_thread(&thread_id.to_string());
self.evict_runner_session(&thread_id.to_string()).await;
}
Ok(archived)
}
pub async fn start_team(&self, req: TeamStartRequest) -> anyhow::Result<TeamState> {
let cfg = self.config.read().await.clone();
let workspace = self.workspace.display().to_string();
let lead_thread_id = match req.lead_thread_id {
Some(thread_id) => thread_id,
None => {
self.create_thread_with(CreateThreadRequest {
title: Some("Team lead".to_string()),
workspace: workspace.clone(),
workspace_id: None,
root_id: None,
provider: None,
model: None,
selection_mode: None,
tool_allowlist: Vec::new(),
developer_instructions: None,
external_tools: Vec::new(),
runner: None,
})
.await?
.thread_id
}
};
let team_id = uuid::Uuid::new_v4().to_string();
let active_lead_turn_id = self.active_turn_for_thread(&lead_thread_id).await;
let lead_selection = match active_lead_turn_id.as_ref() {
Some(turn_id) => self
.active_turn_selections
.read()
.await
.get(turn_id)
.cloned(),
None => self.selection_mode_for_thread(&lead_thread_id).await?,
};
let lead_concrete_selection = lead_selection
.as_ref()
.map(ModelSelectionMode::concrete_selection);
let mut lead = crate::teams::lead_member(
lead_thread_id.clone(),
lead_concrete_selection
.as_ref()
.map(|selection| selection.provider.clone())
.or_else(|| Some(cfg.default_provider.clone())),
lead_concrete_selection
.as_ref()
.map(|selection| selection.model.clone())
.or_else(|| Some(cfg.default_model.clone())),
cfg.policy_mode,
);
if let Some(turn_id) = active_lead_turn_id {
lead.current_turn_id = Some(turn_id);
lead.status = TeamMemberStatus::Running;
}
let mut members = vec![lead];
for (index, member) in req.members.into_iter().enumerate() {
let thread = self
.create_thread_with(CreateThreadRequest {
title: Some(member.name.clone()),
workspace: workspace.clone(),
workspace_id: None,
root_id: None,
provider: member.model_provider.clone(),
model: member.model.clone(),
selection_mode: None,
tool_allowlist: Vec::new(),
developer_instructions: None,
external_tools: Vec::new(),
runner: None,
})
.await?;
let member_id = format!("member-{}", index + 1);
let descriptor = crate::teams::teammate_member(
member_id.clone(),
member.name,
thread.thread_id.clone(),
member.model_provider.or(thread.provider),
member.model.or(thread.model),
cfg.policy_mode,
);
members.push(descriptor);
}
let now = OffsetDateTime::now_utc();
let team = self
.teams
.insert(TeamState {
id: team_id.clone(),
lead_thread_id: lead_thread_id.clone(),
display_mode: req.display_mode,
members,
mailbox: Vec::new(),
tasks: Vec::new(),
created_at: now,
updated_at: now,
})
.await?;
self.emit(RoderEvent::TeamStarted(TeamStarted {
team_id: team_id.clone(),
lead_thread_id,
display_mode: team.display_mode,
members: team.members.clone(),
tasks: team.tasks.clone(),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
for member in team
.members
.iter()
.filter(|member| member.role != roder_api::teams::TeamMemberRole::Lead)
{
self.emit(RoderEvent::TeamMemberStarted(TeamMemberStarted {
team_id: team_id.clone(),
member: member.clone(),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
}
Ok(team)
}
pub async fn list_teams(&self) -> Vec<TeamState> {
let active_thread_ids = self
.active_turns
.read()
.await
.values()
.map(|handle| handle.thread_id.clone())
.collect::<std::collections::HashSet<_>>();
self.teams
.list()
.await
.into_iter()
.map(|team| runtime_local_team_view(team, &active_thread_ids))
.collect()
}
pub async fn read_team(&self, team_id: &str) -> Option<TeamState> {
let active_thread_ids = self
.active_turns
.read()
.await
.values()
.map(|handle| handle.thread_id.clone())
.collect::<std::collections::HashSet<_>>();
self.teams
.get(team_id)
.await
.map(|team| runtime_local_team_view(team, &active_thread_ids))
}
pub async fn start_team_member(
&self,
team_id: &str,
req: TeamMemberStartRequest,
) -> anyhow::Result<TeamState> {
self.start_team_member_with_selection(team_id, req, None)
.await
}
pub async fn message_team_member(
self: &Arc<Self>,
team_id: &str,
member_id: &str,
message: String,
) -> anyhow::Result<TurnId> {
let team = self
.read_team(team_id)
.await
.ok_or_else(|| anyhow::anyhow!("unknown team {team_id:?}"))?;
self.followup_team_member(&team.lead_thread_id, team_id, member_id, message)
.await
}
/// Queue a mailbox message without starting an idle agent. If the target is
/// already running, the message is delivered at the next inference boundary.
pub(crate) async fn queue_team_member_message(
self: &Arc<Self>,
caller_thread_id: &ThreadId,
team_id: &str,
member_id: &str,
message: String,
) -> anyhow::Result<Option<TurnId>> {
let _delivery_guard = self.agent_team_spawn_lock.lock().await;
let team = self
.read_team(team_id)
.await
.ok_or_else(|| anyhow::anyhow!("unknown team {team_id:?}"))?;
let member = team
.members
.iter()
.find(|member| member.id == member_id)
.ok_or_else(|| anyhow::anyhow!("unknown team member {member_id:?}"))?
.clone();
if member.status == TeamMemberStatus::Closed {
anyhow::bail!("subagent {} is closed", member.name);
}
let from_member_id = team
.members
.iter()
.find(|candidate| candidate.thread_id == *caller_thread_id)
.map(|candidate| candidate.id.clone());
self.teams
.append_mailbox_message(
team_id,
from_member_id,
member_id.to_string(),
TeamMailboxMessageKind::Message,
message,
)
.await?;
let Some(turn_id) = self.active_turn_for_thread(&member.thread_id).await else {
return Ok(None);
};
self.deliver_pending_team_mailbox(team_id, member_id, member.thread_id, turn_id.clone())
.await?;
Ok(Some(turn_id))
}
/// Deliver queued messages and start an idle agent, or steer its active turn.
pub(crate) async fn followup_team_member(
self: &Arc<Self>,
caller_thread_id: &ThreadId,
team_id: &str,
member_id: &str,
message: String,
) -> anyhow::Result<TurnId> {
let _spawn_guard = self.agent_team_spawn_lock.lock().await;
self.followup_team_member_locked(caller_thread_id, team_id, member_id, message)
.await
}
async fn followup_team_member_locked(
self: &Arc<Self>,
caller_thread_id: &ThreadId,
team_id: &str,
member_id: &str,
message: String,
) -> anyhow::Result<TurnId> {
let team = self
.read_team(team_id)
.await
.ok_or_else(|| anyhow::anyhow!("unknown team {team_id:?}"))?;
let member = team
.members
.iter()
.find(|member| member.id == member_id)
.ok_or_else(|| anyhow::anyhow!("unknown team member {member_id:?}"))?
.clone();
if member.status == TeamMemberStatus::Closed {
anyhow::bail!("subagent {} is closed", member.name);
}
let from_member_id = team
.members
.iter()
.find(|candidate| candidate.thread_id == *caller_thread_id)
.map(|candidate| candidate.id.clone());
self.teams
.append_mailbox_message(
team_id,
from_member_id,
member_id.to_string(),
TeamMailboxMessageKind::NewTask,
message,
)
.await?;
if let Some(turn_id) = self.active_turn_for_thread(&member.thread_id).await {
self.deliver_pending_team_mailbox(
team_id,
member_id,
member.thread_id,
turn_id.clone(),
)
.await?;
return Ok(turn_id);
}
self.ensure_codex_v2_team_capacity(&team, &member.id, None)
.await?;
let metadata = self.load_thread_metadata(&member.thread_id).await?;
let inherited_context = self
.team_member_turn_contexts
.lock()
.await
.get(&member.thread_id)
.cloned();
let workspace = inherited_context
.as_ref()
.map(|context| context.workspace.clone())
.or_else(|| metadata.as_ref().map(|metadata| metadata.workspace.clone()))
.unwrap_or_else(|| self.workspace.display().to_string());
let member_thread_id = member.thread_id;
let turn_id = self
.start_turn(StartTurnRequest {
thread_id: member_thread_id.clone(),
message: String::new(),
images: Vec::new(),
provider_override: member.model_provider,
model_override: member.model,
reasoning_override: metadata
.as_ref()
.and_then(|metadata| metadata.selection_mode.as_ref())
.and_then(ModelSelectionMode::reasoning)
.map(str::to_string),
workspace,
instructions: inherited_context
.as_ref()
.map(|context| context.instructions.clone())
.unwrap_or_else(crate::default_instructions),
developer_context: inherited_context
.as_ref()
.and_then(|context| context.developer_context.clone()),
task_ledger_required: false,
})
.await?;
Ok(turn_id)
}
pub async fn set_team_member_policy_mode(
&self,
team_id: &str,
member_id: &str,
policy_mode: PolicyMode,
) -> anyhow::Result<TeamState> {
self.teams
.set_member_policy_mode(team_id, member_id, policy_mode)
.await
}
pub async fn interrupt_team_member(
&self,
team_id: &str,
member_id: &str,
) -> anyhow::Result<Option<TurnId>> {
let _interrupt_guard = self.agent_team_spawn_lock.lock().await;
let team = self
.read_team(team_id)
.await
.ok_or_else(|| anyhow::anyhow!("unknown team {team_id:?}"))?;
let member = team
.members
.iter()
.find(|member| member.id == member_id)
.ok_or_else(|| anyhow::anyhow!("unknown team member {member_id:?}"))?
.clone();
if member.status != TeamMemberStatus::Running {
return Ok(None);
}
let Some(turn_id) = member.current_turn_id.clone() else {
return Ok(None);
};
if self
.active_turn_for_thread(&member.thread_id)
.await
.as_ref()
!= Some(&turn_id)
{
return Ok(None);
}
self.interrupt_turn(member.thread_id.clone(), turn_id.clone())
.await?;
// Make queued mailbox work immediately eligible for the replacement turn while the
// lifecycle lock still prevents a follow-up from starting. Delivery acknowledgements
// are turn-owned, so a late acknowledgement from the aborted turn cannot steal a
// reservation acquired by its replacement.
self.teams
.release_mailbox_reservations_for_turn(&turn_id)
.await;
self.teams
.update_member(team_id, member_id, |member| {
member.status = TeamMemberStatus::Interrupted;
member.current_turn_id = None;
member.terminal_error = None;
})
.await?;
self.emit(RoderEvent::TeamMemberCompleted(TeamMemberCompleted {
team_id: team_id.to_string(),
member_id: member_id.to_string(),
member_thread_id: member.thread_id,
turn_id: Some(turn_id.clone()),
status: TeamMemberStatus::Interrupted,
final_message: member.final_message,
error: None,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
Ok(Some(turn_id))
}
pub async fn close_team_member(
&self,
team_id: &str,
member_id: &str,
) -> anyhow::Result<roder_api::teams::TeamMemberDescriptor> {
let team = self
.read_team(team_id)
.await
.ok_or_else(|| anyhow::anyhow!("unknown team {team_id:?}"))?;
let member = team
.members
.iter()
.find(|member| member.id == member_id)
.ok_or_else(|| anyhow::anyhow!("unknown team member {member_id:?}"))?
.clone();
if member.role == roder_api::teams::TeamMemberRole::Lead {
anyhow::bail!("team lead cannot be closed as a subagent");
}
let interrupted_turn_id = if member.status == TeamMemberStatus::Running {
if let Some(turn_id) = member.current_turn_id.clone() {
self.interrupt_turn(member.thread_id.clone(), turn_id.clone())
.await?;
Some(turn_id)
} else {
None
}
} else {
member.current_turn_id.clone()
};
let updated = self
.teams
.update_member(team_id, member_id, |member| {
member.status = TeamMemberStatus::Closed;
member.current_turn_id = None;
})
.await?;
let closed = updated
.members
.iter()
.find(|member| member.id == member_id)
.cloned()
.ok_or_else(|| anyhow::anyhow!("closed team member disappeared"))?;
self.team_member_turn_contexts
.lock()
.await
.remove(&closed.thread_id);
self.emit(RoderEvent::TeamMemberCompleted(TeamMemberCompleted {
team_id: team_id.to_string(),
member_id: closed.id.clone(),
member_thread_id: closed.thread_id.clone(),
turn_id: interrupted_turn_id,
status: TeamMemberStatus::Closed,
final_message: closed.final_message.clone(),
error: closed.terminal_error.clone(),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
Ok(closed)
}
pub async fn cleanup_team(&self, team_id: &str, force: bool) -> anyhow::Result<bool> {
let Some(team) = self.read_team(team_id).await else {
return Ok(false);
};
if !force
&& team
.members
.iter()
.any(|member| member.status == TeamMemberStatus::Running)
{
anyhow::bail!("team {team_id:?} has active teammates; use forced cleanup");
}
if force {
for member in team.members.iter().filter(|member| {
member.role != roder_api::teams::TeamMemberRole::Lead
&& member.status == TeamMemberStatus::Running
}) {
if let Some(turn_id) = member
.current_turn_id
.clone()
.or(self.active_turn_for_thread(&member.thread_id).await)
{
let _ = self.interrupt_turn(member.thread_id.clone(), turn_id).await;
}
}
}
let removed = self.teams.remove(team_id).await?.is_some();
if removed {
let member_thread_ids = team
.members
.iter()
.map(|member| member.thread_id.clone())
.collect::<std::collections::HashSet<_>>();
self.team_member_turn_contexts
.lock()
.await
.retain(|thread_id, _| !member_thread_ids.contains(thread_id));
self.emit(RoderEvent::TeamCleanupCompleted(TeamCleanupCompleted {
team_id: team_id.to_string(),
forced: force,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
}
Ok(removed)
}
pub async fn effective_policy_mode_for_thread(&self, thread_id: &str) -> PolicyMode {
if let Some(mode) = self.teams.policy_mode_for_thread(thread_id).await {
return mode;
}
self.status().await.policy_mode
}
async fn complete_team_member_turn_with_result(
&self,
thread_id: &ThreadId,
turn_id: &TurnId,
status: TeamMemberStatus,
final_message: Option<String>,
terminal_error: Option<String>,
) -> anyhow::Result<()> {
let _delivery_guard = self.agent_team_spawn_lock.lock().await;
let Some((team_id, member)) = self
.teams
.complete_member_turn(
thread_id,
turn_id,
status,
final_message.clone(),
terminal_error.clone(),
)
.await?
else {
return Ok(());
};
if let Some(parent_thread_id) = member.parent_thread_id.as_ref()
&& let Some(team) = self.read_team(&team_id).await
&& let Some(parent) = team
.members
.iter()
.find(|candidate| candidate.thread_id == *parent_thread_id)
{
let identity = member
.agent_path
.as_deref()
.or(member.task_name.as_deref())
.unwrap_or(&member.name);
let mut report = format!("Agent {identity} finished with status {status:?}.");
if let Some(message) = final_message.as_deref()
&& !message.trim().is_empty()
{
report.push_str("\n\nFinal result:\n");
report.push_str(message);
}
if let Some(error) = terminal_error.as_deref()
&& !error.trim().is_empty()
{
report.push_str("\n\nTerminal error:\n");
report.push_str(error);
}
let mailbox_appended = self
.teams
.append_mailbox_message(
&team_id,
Some(member.id.clone()),
parent.id.clone(),
TeamMailboxMessageKind::FinalAnswer,
report,
)
.await
.is_ok();
if mailbox_appended
&& let Some(parent_turn_id) = self.active_turn_for_thread(parent_thread_id).await
{
// The parent may finish between the active-turn lookup and enqueue. Its durable
// mailbox entry remains pending for the next turn, while terminal observers must
// still receive TeamMemberCompleted for this child.
let _ = self
.deliver_pending_team_mailbox(
&team_id,
&parent.id,
parent_thread_id.clone(),
parent_turn_id,
)
.await;
}
}
self.emit(RoderEvent::TeamMemberCompleted(TeamMemberCompleted {
team_id,
member_id: member.id,
member_thread_id: member.thread_id,
turn_id: Some(turn_id.clone()),
status,
final_message,
error: terminal_error,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
Ok(())
}
/// Stops accepting new turns, requests interruption for every active turn,
/// and waits for their runtime tasks to reach terminal cleanup up to
/// `timeout`. Repeated calls are safe and continue draining the same set.
pub async fn drain_active_turns(&self, timeout: std::time::Duration) -> RuntimeDrainOutcome {
let started_at = tokio::time::Instant::now();
let persistence_failures_before =
self.lifecycle_persistence_failures.load(Ordering::Acquire);
let turn_admission = self.turn_admission.lock().await;
self.accepting_turns.store(false, Ordering::Release);
let active = self
.active_turns
.read()
.await
.iter()
.map(|(turn_id, handle)| (turn_id.clone(), handle.thread_id.clone()))
.collect::<Vec<_>>();
let draining = self
.turn_drains
.read()
.await
.iter()
.map(|(turn_id, drain)| (turn_id.clone(), drain.thread_id.clone()))
.collect::<Vec<_>>();
drop(turn_admission);
let mut interrupted_turns = std::collections::BTreeMap::new();
for (turn_id, thread_id) in active.iter().chain(draining.iter()) {
interrupted_turns.insert(turn_id.clone(), thread_id.clone());
}
let interrupted_turn_ids = interrupted_turns.keys().cloned().collect::<Vec<_>>();
for (turn_id, thread_id) in active {
let _ = self
.interrupt_turn_with_reason(thread_id, turn_id, TurnLifecycleReason::Shutdown)
.await;
}
let wait_for_empty = async {
loop {
let notified = self.active_turns_changed.notified();
if self.active_turns.read().await.is_empty()
&& self.turn_drains.read().await.is_empty()
{
return;
}
notified.await;
}
};
let outcome = if tokio::time::timeout(timeout, wait_for_empty).await.is_ok() {
if self.lifecycle_persistence_failures.load(Ordering::Acquire)
> persistence_failures_before
{
RuntimeDrainOutcome::PersistenceFailed {
interrupted_turn_ids,
remaining_turn_ids: Vec::new(),
}
} else {
RuntimeDrainOutcome::Clean {
interrupted_turn_ids,
}
}
} else {
let active_remaining = self
.active_turns
.read()
.await
.iter()
.map(|(turn_id, handle)| (turn_id.clone(), handle.drain.clone()))
.collect::<Vec<_>>();
let draining_remaining = self
.turn_drains
.read()
.await
.iter()
.map(|(turn_id, drain)| (turn_id.clone(), drain.clone()))
.collect::<Vec<_>>();
let remaining = active_remaining
.into_iter()
.chain(draining_remaining)
.collect::<std::collections::BTreeMap<_, _>>();
let remaining_turn_ids = remaining.keys().cloned().collect::<Vec<_>>();
for turn_id in &remaining_turn_ids {
let handle = remaining
.get(turn_id)
.expect("remaining turn ID must have a drain handle");
self.record_turn_lifecycle(
handle.thread_id.clone(),
turn_id.clone(),
TurnLifecycleState::InterruptRequested,
TurnCleanupState::TimedOut,
Some(TurnLifecycleReason::Shutdown),
)
.await;
}
if self.lifecycle_persistence_failures.load(Ordering::Acquire)
> persistence_failures_before
{
RuntimeDrainOutcome::PersistenceFailed {
interrupted_turn_ids,
remaining_turn_ids,
}
} else {
RuntimeDrainOutcome::DeadlineExceeded {
interrupted_turn_ids,
remaining_turn_ids,
}
}
};
self.record_lifecycle_drain_outcome(started_at, &outcome);
outcome
}
/// Enables turn submission after a prior drain attempt. This is intended
/// for embedders that keep a runtime alive after a bounded drain timeout.
pub fn resume_accepting_turns(&self) {
self.accepting_turns.store(true, Ordering::Release);
}
pub async fn turn_lifecycle_snapshot(
&self,
thread_id: &ThreadId,
) -> anyhow::Result<TurnLifecycleSnapshot> {
let Some(store) = &self.thread_store else {
return Ok(TurnLifecycleSnapshot::default());
};
let extension_states = store.load_extension_states(thread_id).await?;
Ok(turn_lifecycle_snapshot(&extension_states))
}
pub async fn load_thread(
&self,
thread_id: &ThreadId,
) -> anyhow::Result<Option<ThreadSnapshot>> {
let loaded = if let Some(store) = &self.thread_store {
store.load_thread(thread_id).await?
} else {
None
};
if let Some(snapshot) = loaded.as_ref() {
let live_turn_ids = self
.active_turns
.read()
.await
.keys()
.cloned()
.chain(self.turn_drains.read().await.keys().cloned())
.collect::<std::collections::HashSet<_>>();
let lifecycle = turn_lifecycle_snapshot(&snapshot.extension_states);
for record in lifecycle.records.into_iter().filter(|record| {
record.state.requires_recovery() && !live_turn_ids.contains(&record.turn_id)
}) {
self.lifecycle_restart_reconciliations
.fetch_add(1, Ordering::AcqRel);
self.record_turn_lifecycle(
record.thread_id,
record.turn_id,
TurnLifecycleState::RecoveryNeeded,
TurnCleanupState::Unknown,
Some(TurnLifecycleReason::RuntimeRestart),
)
.await;
}
self.emit(RoderEvent::ThreadLoaded(ThreadLoaded {
thread_id: thread_id.clone(),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
}
Ok(loaded)
}
pub async fn workspace_for_thread(&self, thread_id: &ThreadId) -> anyhow::Result<String> {
if let Some(store) = &self.thread_store {
let snapshot = store
.load_thread(thread_id)
.await?
.ok_or_else(|| anyhow::anyhow!("thread not found: {thread_id}"));
match snapshot {
Ok(snapshot) => {
if let Some(metadata) = snapshot.metadata {
// Fork-backed threads fail closed before any write
// when their workspace was removed out-of-band.
if let Some(fork) = &metadata.workspace_fork
&& fork.status == roder_api::forks::ForkStatus::Active
&& !std::path::Path::new(&metadata.workspace).is_dir()
{
anyhow::bail!(
"workspace fork {} is missing its workspace at {}; restore it or \
remove the fork before running turns in this thread",
fork.id,
metadata.workspace
);
}
return Ok(metadata.workspace);
}
eprintln!(
"thread metadata missing while resolving workspace for {thread_id}; falling back to runtime workspace"
);
}
Err(err) => {
eprintln!(
"thread missing while resolving workspace for {thread_id}: {err}; falling back to runtime workspace"
);
}
}
}
Ok(self.workspace.display().to_string())
}
async fn selection_mode_for_thread(
&self,
thread_id: &ThreadId,
) -> anyhow::Result<Option<ModelSelectionMode>> {
let Some(store) = &self.thread_store else {
return Ok(None);
};
Ok(store
.load_thread(thread_id)
.await?
.and_then(|snapshot| snapshot.metadata)
.and_then(|metadata| {
metadata
.selection_mode
.or_else(|| match (metadata.provider, metadata.model) {
(Some(provider), Some(model)) => {
Some(ModelSelectionMode::manual(provider, model, None))
}
_ => None,
})
}))
}
/// Per-thread tool allowlist, developer instructions, and external tools persisted at thread creation.
pub(crate) async fn thread_turn_overrides(
&self,
thread_id: &ThreadId,
) -> anyhow::Result<ThreadTurnOverrides> {
let Some(store) = &self.thread_store else {
return Ok(ThreadTurnOverrides::default());
};
Ok(store
.load_thread_metadata(thread_id)
.await?
.map(|metadata| ThreadTurnOverrides {
tool_allowlist: metadata.tool_allowlist,
developer_instructions: metadata.developer_instructions,
external_tools: metadata.external_tools,
})
.unwrap_or_default())
}
pub async fn set_thread_selection_mode(
&self,
thread_id: &ThreadId,
selection_mode: ModelSelectionMode,
) -> anyhow::Result<()> {
let Some(store) = &self.thread_store else {
return Ok(());
};
let Some(snapshot) = store.load_thread(thread_id).await? else {
anyhow::bail!("thread not found: {thread_id}");
};
let Some(mut metadata) = snapshot.metadata else {
return Ok(());
};
let concrete = selection_mode.concrete_selection();
metadata.provider = Some(concrete.provider);
metadata.model = Some(concrete.model);
metadata.selection_mode = Some(selection_mode);
metadata.updated_at = OffsetDateTime::now_utc();
store.update_thread_metadata(metadata).await?;
Ok(())
}
async fn runner_session_for_thread(
&self,
thread_id: &ThreadId,
) -> anyhow::Result<Option<(RunnerDestination, Arc<dyn RemoteRunnerSession>)>> {
let metadata = if let Some(store) = &self.thread_store {
store.load_thread_metadata(thread_id).await?
} else {
None
};
// An explicit per-thread binding wins over the runtime-level destination.
let destination = metadata
.as_ref()
.and_then(|metadata| metadata.runner_binding.as_ref())
.map(|binding| binding.destination.clone())
.or(self.config.read().await.remote_runner_destination.clone());
let Some(destination) = destination else {
self.evict_runner_session(thread_id).await;
return Ok(None);
};
let provider = self
.registry
.remote_runner_providers
.iter()
.find(|provider| provider.id() == destination.provider_id)
.cloned()
.ok_or_else(|| {
anyhow::anyhow!(
"remote runner provider {:?} is not installed",
destination.provider_id
)
})?;
let persisted_state = metadata.and_then(|metadata| metadata.runner_state);
let slot = {
let mut sessions = self.runner_sessions.lock().await;
match sessions.get(thread_id) {
Some(slot) if slot.matches(&destination) => slot.clone(),
_ => {
let slot = Arc::new(RunnerSessionSlot::empty(&destination));
sessions.insert(thread_id.clone(), slot.clone());
slot
}
}
};
loop {
let initialized = slot.initialized.notified();
let mut state = slot.state.lock().await;
match &*state {
RunnerSessionSlotState::Ready(cached) => {
return Ok(Some((cached.destination.clone(), cached.session.clone())));
}
RunnerSessionSlotState::Failed(error) => {
return Err(anyhow::anyhow!(error.to_string()));
}
RunnerSessionSlotState::Initializing => {
drop(state);
initialized.await;
}
RunnerSessionSlotState::Empty => {
*state = RunnerSessionSlotState::Initializing;
drop(state);
self.spawn_runner_session_initialization(
thread_id.clone(),
destination.clone(),
provider.clone(),
persisted_state.clone(),
slot.clone(),
);
}
}
}
}
fn spawn_runner_session_initialization(
&self,
thread_id: ThreadId,
destination: RunnerDestination,
provider: Arc<dyn RemoteRunnerProvider>,
persisted_state: Option<RunnerSessionState>,
slot: Arc<RunnerSessionSlot>,
) {
let thread_store = self.thread_store.clone();
let runner_sessions = self.runner_sessions.clone();
// The task intentionally outlives the turn that first requested the
// workspace. Interrupting that turn must not strand the slot in
// `Initializing` and deadlock every later tool or lifecycle call.
drop(tokio::spawn(async move {
let initialized = async {
let session = if let Some(state) = persisted_state
&& state.provider_id == destination.provider_id
&& state.destination_id == destination.id
{
match provider.resume_session(state).await {
Ok(session) => session,
Err(_) => provider.create_session(destination.clone()).await?,
}
} else {
provider.create_session(destination.clone()).await?
};
let resolved = (destination.clone(), session.clone());
// Persist before releasing the singleflight or dispatching
// the first tool. A later process can rejoin even if the turn
// that requested initialization has already been interrupted.
Self::persist_runner_state_in_store(
thread_store.as_ref(),
&thread_id,
Some(&resolved),
)
.await?;
Ok::<_, anyhow::Error>(CachedRunnerSession {
destination,
session,
})
}
.await;
match initialized {
Ok(cached) => {
*slot.state.lock().await = RunnerSessionSlotState::Ready(cached);
slot.initialized.notify_waiters();
}
Err(error) => {
let message: Arc<str> = Arc::from(format!("{error:#}"));
*slot.state.lock().await = RunnerSessionSlotState::Failed(message);
slot.initialized.notify_waiters();
let mut sessions = runner_sessions.lock().await;
let is_current = sessions
.get(&thread_id)
.is_some_and(|current| Arc::ptr_eq(current, &slot));
if is_current {
sessions.remove(&thread_id);
}
}
}
}));
}
async fn evict_runner_session(&self, thread_id: &ThreadId) {
self.runner_sessions.lock().await.remove(thread_id);
}
async fn cache_runner_session(
&self,
thread_id: &ThreadId,
destination: RunnerDestination,
session: Arc<dyn RemoteRunnerSession>,
) {
let cached = CachedRunnerSession {
destination,
session,
};
self.runner_sessions.lock().await.insert(
thread_id.clone(),
Arc::new(RunnerSessionSlot::ready(cached)),
);
}
/// Read a thread's explicit runner binding without creating or resuming a
/// session. Tool routing uses this before local previews so runner-backed
/// and fail-closed hosted calls never inspect the host workspace.
pub(crate) async fn runner_binding_for_thread(
&self,
thread_id: &ThreadId,
) -> anyhow::Result<Option<ThreadRunnerBinding>> {
let Some(store) = &self.thread_store else {
return Ok(None);
};
Ok(store
.load_thread_metadata(thread_id)
.await?
.and_then(|metadata| metadata.runner_binding))
}
/// Resolve a previously-read binding into a live remote workspace. The
/// session remains lazy: callers invoke this only after policy approval.
pub(crate) async fn remote_workspace_for_binding(
&self,
thread_id: &ThreadId,
binding: ThreadRunnerBinding,
) -> anyhow::Result<Arc<RemoteWorkspace>> {
let session = self
.runner_session_for_thread(thread_id)
.await?
.map(|(_, session)| session)
.ok_or_else(|| {
anyhow::anyhow!("runner-bound thread {thread_id} has no runner session")
})?;
Ok(Arc::new(RemoteWorkspace {
session,
root: binding.workspace,
read_roots: binding.read_roots,
}))
}
pub(crate) async fn runner_tool_execution_lock(
&self,
session: &dyn RemoteRunnerSession,
) -> Arc<RunnerToolExecutionMutex> {
let state = session.state();
let key = (state.provider_id, state.session_id);
let mut locks = self.runner_tool_execution_locks.lock().await;
if let Some(lock) = locks.get(&key).and_then(Weak::upgrade) {
return lock;
}
locks.retain(|_, lock| lock.strong_count() > 0);
let lock = Arc::new(Mutex::new(()));
locks.insert(key, Arc::downgrade(&lock));
lock
}
async fn persist_runner_state(
&self,
thread_id: &ThreadId,
runner: Option<&(RunnerDestination, Arc<dyn RemoteRunnerSession>)>,
) -> anyhow::Result<()> {
Self::persist_runner_state_in_store(self.thread_store.as_ref(), thread_id, runner).await
}
async fn persist_runner_state_in_store(
store: Option<&Arc<dyn ThreadStore>>,
thread_id: &ThreadId,
runner: Option<&(RunnerDestination, Arc<dyn RemoteRunnerSession>)>,
) -> anyhow::Result<()> {
let Some((destination, session)) = runner else {
return Ok(());
};
let Some(store) = store else {
return Ok(());
};
let Some(snapshot) = store.load_thread(thread_id).await? else {
return Ok(());
};
let Some(mut metadata) = snapshot.metadata else {
return Ok(());
};
metadata.runner_destination = Some(destination.clone());
metadata.runner_state = Some(session.state());
metadata.updated_at = OffsetDateTime::now_utc();
store.update_thread_metadata(metadata).await?;
Ok(())
}
fn remote_runner_provider_by_id(
&self,
provider_id: &str,
) -> Option<Arc<dyn RemoteRunnerProvider>> {
self.registry
.remote_runner_providers
.iter()
.find(|provider| provider.id() == provider_id)
.cloned()
}
/// Resolve the live runner session for a thread along with its provider,
/// failing clearly when the thread is not runner-bound.
async fn thread_runner_session(
&self,
thread_id: &ThreadId,
) -> anyhow::Result<(
RunnerDestination,
Arc<dyn RemoteRunnerProvider>,
Arc<dyn RemoteRunnerSession>,
)> {
let Some((destination, session)) = self.runner_session_for_thread(thread_id).await? else {
anyhow::bail!("thread {thread_id} is not bound to a remote runner");
};
let provider = self
.remote_runner_provider_by_id(&destination.provider_id)
.ok_or_else(|| {
anyhow::anyhow!(
"remote runner provider {:?} is not installed",
destination.provider_id
)
})?;
Ok((destination, provider, session))
}
/// Pause a runner-bound thread's session toward standby and persist the
/// post-pause state. Errors if the provider is not pausable.
pub async fn pause_thread_runner(
&self,
thread_id: &ThreadId,
) -> anyhow::Result<RunnerSessionState> {
let (destination, provider, session) = self.thread_runner_session(thread_id).await?;
anyhow::ensure!(
provider.capabilities().pausable,
"remote runner provider {:?} does not support pausing",
destination.provider_id
);
let state = session.pause().await?;
self.persist_runner_state(thread_id, Some(&(destination, session)))
.await?;
Ok(state)
}
/// Resume (wake) a runner-bound thread's paused session and persist state.
pub async fn resume_thread_runner(
&self,
thread_id: &ThreadId,
) -> anyhow::Result<RunnerSessionState> {
let (destination, _provider, session) = self.thread_runner_session(thread_id).await?;
let state = session.resume().await?;
self.persist_runner_state(thread_id, Some(&(destination, session)))
.await?;
Ok(state)
}
/// Detach a runner-bound thread's session: persist the durable, rejoinable
/// state and leave the remote sandbox alive. Errors if not detachable.
pub async fn detach_thread_runner(
&self,
thread_id: &ThreadId,
) -> anyhow::Result<RunnerSessionState> {
let (destination, provider, session) = self.thread_runner_session(thread_id).await?;
anyhow::ensure!(
provider.capabilities().detachable,
"remote runner provider {:?} does not support detaching",
destination.provider_id
);
let state = session.detach().await?;
// Persist the detached state explicitly so a later turn or process
// rejoins the same sandbox instead of provisioning a new one.
if let Some(store) = &self.thread_store
&& let Some(snapshot) = store.load_thread(thread_id).await?
&& let Some(mut metadata) = snapshot.metadata
{
metadata.runner_destination = Some(destination);
metadata.runner_state = Some(state.clone());
metadata.updated_at = OffsetDateTime::now_utc();
store.update_thread_metadata(metadata).await?;
}
self.evict_runner_session(thread_id).await;
Ok(state)
}
/// Rejoin a previously created sandbox from a thread's persisted runner
/// state without provisioning a new one. An optional `sandbox` overrides the
/// persisted sandbox name (recovery by name). Persists refreshed state.
pub async fn rejoin_thread_runner(
&self,
thread_id: &ThreadId,
sandbox: Option<String>,
) -> anyhow::Result<RunnerSessionState> {
let store = self
.thread_store
.as_ref()
.ok_or_else(|| anyhow::anyhow!("thread store is required to rejoin a runner"))?;
let metadata = store
.load_thread_metadata(thread_id)
.await?
.ok_or_else(|| anyhow::anyhow!("thread {thread_id} has no metadata"))?;
let destination = metadata
.runner_binding
.as_ref()
.map(|binding| binding.destination.clone())
.or_else(|| metadata.runner_destination.clone())
.ok_or_else(|| anyhow::anyhow!("thread {thread_id} is not bound to a remote runner"))?;
let mut state = metadata
.runner_state
.clone()
.ok_or_else(|| anyhow::anyhow!("thread {thread_id} has no persisted runner state"))?;
if let Some(sandbox) = sandbox
&& let Some(object) = state.metadata.as_object_mut()
{
object.insert("sandbox_name".to_string(), serde_json::Value::from(sandbox));
}
let provider = self
.remote_runner_provider_by_id(&destination.provider_id)
.ok_or_else(|| {
anyhow::anyhow!(
"remote runner provider {:?} is not installed",
destination.provider_id
)
})?;
let session = provider.rejoin_session(state).await?;
self.persist_runner_state(thread_id, Some(&(destination.clone(), session.clone())))
.await?;
self.cache_runner_session(thread_id, destination, session.clone())
.await;
Ok(session.state())
}
async fn record_thread_usage_metadata(
&self,
thread_id: &ThreadId,
usage: &TokenUsage,
) -> anyhow::Result<()> {
if usage.is_empty() {
return Ok(());
}
let Some(store) = &self.thread_store else {
return Ok(());
};
let Some(snapshot) = store.load_thread(thread_id).await? else {
return Ok(());
};
let Some(mut metadata) = snapshot.metadata else {
return Ok(());
};
metadata
.usage
.get_or_insert_with(ThreadUsageMetadata::default)
.add_token_usage(usage);
metadata.updated_at = OffsetDateTime::now_utc();
store.update_thread_metadata(metadata).await?;
Ok(())
}
pub fn start_turn(
self: &Arc<Self>,
mut req: StartTurnRequest,
) -> BoxFuture<'_, anyhow::Result<TurnId>> {
Box::pin(async move {
let turn_admission = self.turn_admission.lock().await;
anyhow::ensure!(
self.accepting_turns.load(Ordering::Acquire),
"runtime is quiescing and cannot accept new turns"
);
req.workspace = validate_thread_workspace(&req.workspace)?;
let team_member = self.teams.member_for_thread(&req.thread_id).await;
let cfg = self.config.read().await.clone();
let provider = req
.provider_override
.clone()
.unwrap_or_else(|| cfg.default_provider.clone());
self.engine_for(&provider)?;
let turn_id = uuid::Uuid::new_v4().to_string();
let mut initial_mailbox_ack = None;
if let Some((team_id, member)) = &team_member {
let pending = self
.teams
.reserve_pending_mailbox_messages(team_id, &member.id, &turn_id)
.await?;
if !pending.is_empty()
&& let Some(team) = self.read_team(team_id).await
{
let mailbox = format_mailbox_messages(&team, &pending);
req.message = if req.message.trim().is_empty() {
mailbox
} else {
format!("{mailbox}\n\n[Direct task input]\n{}", req.message)
};
initial_mailbox_ack = Some(MailboxDeliveryAck {
team_id: team_id.clone(),
message_ids: pending.iter().map(|message| message.id.clone()).collect(),
});
}
}
let (abort_handle, abort_registration) = AbortHandle::new_pair();
let drain = Arc::new(TurnDrainHandle {
thread_id: req.thread_id.clone(),
interrupt_requested: AtomicBool::new(false),
interrupt_reason: Mutex::new(None),
completed: AtomicBool::new(false),
completed_notify: Notify::new(),
});
let active = ActiveTurnHandle {
thread_id: req.thread_id.clone(),
abort: abort_handle,
steers: Arc::new(Mutex::new(Vec::new())),
drain,
};
self.active_turns
.write()
.await
.insert(turn_id.clone(), active);
self.record_turn_lifecycle(
req.thread_id.clone(),
turn_id.clone(),
TurnLifecycleState::Running,
TurnCleanupState::NotRequested,
None,
)
.await;
self.active_turn_contexts.write().await.insert(
turn_id.clone(),
InheritedTurnContext {
workspace: req.workspace.clone(),
instructions: req.instructions.clone(),
developer_context: req.developer_context.clone(),
},
);
if let Some((team_id, member)) = team_member {
let updated = match self
.teams
.update_member(&team_id, &member.id, |member| {
member.current_turn_id = Some(turn_id.clone());
member.status = TeamMemberStatus::Running;
member.final_message = None;
member.terminal_error = None;
})
.await
{
Ok(updated) => updated,
Err(error) => {
self.active_turns.write().await.remove(&turn_id);
self.active_turn_contexts.write().await.remove(&turn_id);
self.teams
.release_mailbox_reservations_for_turn(&turn_id)
.await;
return Err(error);
}
};
if let Some(member) = updated
.members
.into_iter()
.find(|candidate| candidate.id == member.id)
{
self.emit(RoderEvent::TeamMemberStatusChanged(
TeamMemberStatusChanged {
team_id,
member_id: member.id,
member_thread_id: member.thread_id,
status: TeamMemberStatus::Running,
timestamp: OffsetDateTime::now_utc(),
},
))
.await;
}
}
let runtime = Arc::clone(self);
let turn_req = req;
let thread_id_for_task = turn_req.thread_id.clone();
let turn_id_for_task = turn_id.clone();
tokio::spawn(async move {
let result = Abortable::new(
runtime.run_turn(turn_req, turn_id_for_task.clone(), initial_mailbox_ack),
abort_registration,
)
.await;
/*
* A failed sibling in a parallel tool batch drops in-flight external tool
* futures (`try_join_all` in `route_tool_calls`), stranding their
* `pending_external_tool_calls` entries. Sweep before reporting the turn
* outcome so every `thread/toolExecutionRequested` gets a terminal
* resolution; on clean completion the map holds nothing for this turn.
*/
runtime
.cancel_pending_external_tool_calls_for_turn(&turn_id_for_task)
.await;
let completed = matches!(&result, Ok(Ok(TurnRunOutcome::Completed)));
match &result {
Ok(Err(err)) => {
let (cleanup, ownership) =
runtime.await_provider_turn_cleanup(&turn_id_for_task).await;
// run_turn emits failures after the stream starts; this covers setup/startup errors.
runtime
.emit(RoderEvent::TurnFailed(TurnFailed {
thread_id: thread_id_for_task.clone(),
turn_id: turn_id_for_task.clone(),
error: err.to_string(),
error_kind: None,
usage: None,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
runtime
.record_turn_lifecycle_with_ownership(
thread_id_for_task.clone(),
turn_id_for_task.clone(),
TurnLifecycleState::Failed,
cleanup,
Some(TurnLifecycleReason::ProviderFailure),
ownership,
)
.await;
let _ = runtime
.complete_team_member_turn_with_result(
&thread_id_for_task,
&turn_id_for_task,
TeamMemberStatus::Failed,
None,
Some(err.to_string()),
)
.await;
}
Ok(Ok(TurnRunOutcome::Stopped)) => {
let (cleanup, ownership) =
runtime.await_provider_turn_cleanup(&turn_id_for_task).await;
runtime
.record_turn_lifecycle_with_ownership(
thread_id_for_task.clone(),
turn_id_for_task.clone(),
TurnLifecycleState::Failed,
cleanup,
Some(TurnLifecycleReason::ProviderFailure),
ownership,
)
.await;
let _ = runtime
.complete_team_member_turn_with_result(
&thread_id_for_task,
&turn_id_for_task,
TeamMemberStatus::Failed,
None,
Some("turn stopped before completion".to_string()),
)
.await;
}
Err(_) => {
let reason = if let Some(handle) = runtime
.turn_drains
.read()
.await
.get(&turn_id_for_task)
.cloned()
{
handle
.interrupt_reason
.lock()
.await
.unwrap_or(TurnLifecycleReason::RuntimeFailure)
} else {
TurnLifecycleReason::RuntimeFailure
};
let (cleanup, ownership) =
runtime.await_provider_turn_cleanup(&turn_id_for_task).await;
runtime
.record_turn_lifecycle_with_ownership(
thread_id_for_task.clone(),
turn_id_for_task.clone(),
TurnLifecycleState::Interrupted,
cleanup,
Some(reason),
ownership,
)
.await;
runtime
.emit(RoderEvent::TurnInterrupted(TurnInterrupted {
thread_id: thread_id_for_task.clone(),
turn_id: turn_id_for_task.clone(),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
let _ = runtime
.complete_team_member_turn_with_result(
&thread_id_for_task,
&turn_id_for_task,
TeamMemberStatus::Interrupted,
None,
None,
)
.await;
}
Ok(Ok(TurnRunOutcome::Completed)) => {}
}
runtime
.teams
.release_mailbox_reservations_for_turn(&turn_id_for_task)
.await;
runtime.active_turns.write().await.remove(&turn_id_for_task);
if let Some(drain) = runtime.turn_drains.write().await.remove(&turn_id_for_task) {
drain.completed.store(true, Ordering::Release);
drain.completed_notify.notify_waiters();
}
runtime
.active_turn_selections
.write()
.await
.remove(&turn_id_for_task);
runtime
.active_turn_contexts
.write()
.await
.remove(&turn_id_for_task);
if !completed {
// Completion paths above consume registered provider cleanup
// handles. A setup failure before an engine can stream has no
// such handle; this is a harmless final defensive sweep.
let _ = runtime.provider_turn_cleanups.lock().map(|mut cleanups| {
cleanups.remove(&turn_id_for_task);
});
}
runtime.active_turns_changed.notify_waiters();
if completed {
let _ = runtime
.continue_active_goal_after_turn(thread_id_for_task)
.await;
}
});
drop(turn_admission);
Ok(turn_id)
})
}
pub(crate) async fn has_active_turn_for_thread(&self, thread_id: &ThreadId) -> bool {
self.active_turns
.read()
.await
.values()
.any(|handle| &handle.thread_id == thread_id)
}
async fn ensure_codex_v2_team_capacity(
&self,
team: &TeamState,
resuming_member_id: &str,
candidate_model: Option<&str>,
) -> anyhow::Result<()> {
if !is_codex_v2_team(team)
&& !candidate_model
.is_some_and(|model| model_supports_reasoning_effort(model, REASONING_ULTRA))
{
return Ok(());
}
let active_thread_ids = self
.active_turns
.read()
.await
.values()
.map(|handle| handle.thread_id.clone())
.collect::<std::collections::HashSet<_>>();
let resident_threads = 1 + team
.members
.iter()
.filter(|member| {
member.role != roder_api::teams::TeamMemberRole::Lead
&& member.id != resuming_member_id
&& active_thread_ids.contains(&member.thread_id)
})
.count();
anyhow::ensure!(
resident_threads < codex_v2::CODEX_V2_MAX_RESIDENT_TEAM_THREADS,
"agent thread limit reached for this Codex V2 team: maximum {} resident threads (the lead plus 3 running subagents)",
codex_v2::CODEX_V2_MAX_RESIDENT_TEAM_THREADS
);
Ok(())
}
/// Number of currently running turns across all threads. Hosted runtime
/// pools use this to avoid evicting tenants with active work.
pub async fn active_turn_count(&self) -> usize {
self.active_turns.read().await.len()
}
pub async fn active_turn_for_thread(&self, thread_id: &ThreadId) -> Option<TurnId> {
self.active_turns
.read()
.await
.iter()
.find_map(|(turn_id, handle)| (&handle.thread_id == thread_id).then(|| turn_id.clone()))
}
pub async fn thread_activity(&self, thread_id: &ThreadId) -> ThreadActivity {
let active_turn_id = self.active_turn_for_thread(thread_id).await;
let draining_turn_id = if active_turn_id.is_none() {
self.turn_drains
.read()
.await
.iter()
.find_map(|(turn_id, drain)| {
(&drain.thread_id == thread_id).then(|| turn_id.clone())
})
} else {
None
};
let Some(active_turn_id) = active_turn_id.or(draining_turn_id) else {
return ThreadActivity::default();
};
let mut active_flags = Vec::new();
{
let pending_approvals = self.pending_tool_approvals.lock().await;
if pending_approvals
.values()
.any(|pending| &pending.thread_id == thread_id && pending.turn_id == active_turn_id)
{
active_flags.push("approvalRequired".to_string());
}
}
{
let pending_inputs = self.pending_user_inputs.lock().await;
if pending_inputs
.values()
.any(|pending| &pending.thread_id == thread_id && pending.turn_id == active_turn_id)
{
active_flags.push("userInputRequired".to_string());
}
}
{
let pending_external = self.pending_external_tool_calls.lock().await;
if pending_external
.values()
.any(|pending| &pending.thread_id == thread_id && pending.turn_id == active_turn_id)
{
active_flags.push("externalToolPending".to_string());
}
}
let interrupting = self
.active_turns
.read()
.await
.get(&active_turn_id)
.is_some_and(|handle| handle.drain.interrupt_requested.load(Ordering::Acquire))
|| self
.turn_drains
.read()
.await
.get(&active_turn_id)
.is_some_and(|drain| drain.interrupt_requested.load(Ordering::Acquire));
if interrupting {
active_flags.push("interrupting".to_string());
}
if self.pending_plan_exit().await.is_some_and(|pending| {
&pending.thread_id == thread_id && pending.turn_id == active_turn_id
}) {
active_flags.push("planExitRequired".to_string());
}
ThreadActivity {
active_turn_id: Some(active_turn_id),
active_flags,
}
}
pub async fn interrupt_turn(&self, thread_id: ThreadId, turn_id: TurnId) -> anyhow::Result<()> {
self.interrupt_turn_with_reason(thread_id, turn_id, TurnLifecycleReason::UserInterrupt)
.await
}
async fn interrupt_turn_with_reason(
&self,
thread_id: ThreadId,
turn_id: TurnId,
reason: TurnLifecycleReason,
) -> anyhow::Result<()> {
let handle = self.active_turns.write().await.remove(&turn_id);
if let Some(handle) = handle {
if handle
.drain
.interrupt_requested
.swap(true, Ordering::AcqRel)
{
return Ok(());
}
*handle.drain.interrupt_reason.lock().await = Some(reason);
self.turn_drains
.write()
.await
.insert(turn_id.clone(), handle.drain.clone());
let ownership = self.provider_cleanup_ownership(&turn_id);
self.record_turn_lifecycle_with_ownership(
thread_id.clone(),
turn_id.clone(),
TurnLifecycleState::InterruptRequested,
TurnCleanupState::Requested,
Some(reason),
ownership,
)
.await;
handle.abort.abort();
self.active_turns_changed.notify_waiters();
}
self.active_turn_selections.write().await.remove(&turn_id);
self.cancel_pending_external_tool_calls_for_turn(&turn_id)
.await;
Ok(())
}
pub async fn steer_turn(
&self,
thread_id: ThreadId,
turn_id: TurnId,
message: String,
images: Vec<InputImage>,
) -> anyhow::Result<()> {
self.enqueue_turn_steer(thread_id, turn_id, message, images, None)
.await
}
pub(crate) async fn has_pending_turn_steers(&self, turn_id: &TurnId) -> bool {
let active = self.active_turns.read().await.get(turn_id).cloned();
let Some(active) = active else {
return false;
};
let has_pending = !active.steers.lock().await.is_empty();
has_pending
}
async fn steer_turn_with_mailbox_ack(
&self,
thread_id: ThreadId,
turn_id: TurnId,
message: String,
message_ids: Vec<String>,
team_id: TeamId,
) -> anyhow::Result<()> {
self.enqueue_turn_steer(
thread_id,
turn_id,
message,
Vec::new(),
Some(MailboxDeliveryAck {
team_id,
message_ids,
}),
)
.await
}
async fn deliver_pending_team_mailbox(
&self,
team_id: &str,
member_id: &str,
member_thread_id: ThreadId,
turn_id: TurnId,
) -> anyhow::Result<()> {
let pending = self
.teams
.reserve_pending_mailbox_messages(team_id, member_id, &turn_id)
.await?;
if pending.is_empty() {
return Ok(());
}
let message_ids = pending
.iter()
.map(|message| message.id.clone())
.collect::<Vec<_>>();
let team = self
.read_team(team_id)
.await
.ok_or_else(|| anyhow::anyhow!("unknown team {team_id:?}"))?;
if let Err(error) = self
.steer_turn_with_mailbox_ack(
member_thread_id,
turn_id.clone(),
format_mailbox_messages(&team, &pending),
message_ids.clone(),
team_id.to_string(),
)
.await
{
self.teams
.release_mailbox_reservations(&turn_id, &message_ids)
.await;
return Err(error);
}
Ok(())
}
async fn enqueue_turn_steer(
&self,
thread_id: ThreadId,
turn_id: TurnId,
message: String,
images: Vec<InputImage>,
mailbox_ack: Option<MailboxDeliveryAck>,
) -> anyhow::Result<()> {
let message = message.trim().to_string();
if message.is_empty() && images.is_empty() {
return Ok(());
}
let Some(active) = self.active_turns.read().await.get(&turn_id).cloned() else {
anyhow::bail!("no active turn to steer");
};
active.steers.lock().await.push(QueuedTurnSteer {
message: UserMessage::with_images(message.clone(), images),
mailbox_ack,
});
self.emit(RoderEvent::TurnSteered(TurnSteered {
thread_id,
turn_id,
message,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
Ok(())
}
pub async fn tool_specs(&self) -> Vec<roder_api::tools::ToolSpec> {
let cfg = self.config.read().await;
let model_profile =
model_profile_for_provider_model(&cfg, &cfg.default_provider, &cfg.default_model);
self.filtered_tool_specs(&cfg, &cfg.default_model, model_profile.as_ref(), &[], &[])
}
pub fn subagent_definitions(&self) -> Vec<SubagentDefinition> {
self.registry
.subagent_dispatchers
.iter()
.flat_map(|dispatcher| dispatcher.definitions())
.collect()
}
async fn run_turn(
self: &Arc<Self>,
req: StartTurnRequest,
turn_id: TurnId,
initial_mailbox_ack: Option<MailboxDeliveryAck>,
) -> anyhow::Result<TurnRunOutcome> {
let turn_started_at = OffsetDateTime::now_utc();
self.emit(RoderEvent::TurnStarted(TurnStarted {
thread_id: req.thread_id.clone(),
turn_id: turn_id.clone(),
runtime_profile: self.config.read().await.runtime_profile,
timestamp: turn_started_at,
}))
.await;
self.persist_turn_item(
&req.thread_id,
&turn_id,
&TranscriptItem::UserMessage(UserMessage::with_images(
req.message.clone(),
req.images.clone(),
)),
)
.await?;
if let Some(ack) = initial_mailbox_ack {
self.teams
.mark_mailbox_messages_delivered(&ack.team_id, &turn_id, &ack.message_ids)
.await?;
}
let mut cfg = self.config.read().await.clone();
let runtime_profile = cfg.runtime_profile;
let turn_deadline = turn_deadline_for_config(&cfg);
let deadline_finalization_reserve =
crate::deadline_policy::finalization_reserve_seconds(cfg.turn_deadline_seconds);
let selection_mode = self.selection_mode_for_thread(&req.thread_id).await?;
let concrete_selection = selection_mode
.as_ref()
.map(ModelSelectionMode::concrete_selection);
let default_provider = req
.provider_override
.clone()
.or_else(|| {
concrete_selection
.as_ref()
.map(|selection| selection.provider.clone())
})
.unwrap_or(cfg.default_provider.clone());
let default_model = req
.model_override
.clone()
.or_else(|| {
concrete_selection
.as_ref()
.map(|selection| selection.model.clone())
})
.unwrap_or(cfg.default_model.clone());
if let Some(reasoning) = req.reasoning_override.as_deref().or_else(|| {
selection_mode
.as_ref()
.and_then(ModelSelectionMode::reasoning)
}) {
validate_reasoning_effort(&default_model, reasoning)?;
cfg.reasoning = Some(reasoning.to_string());
}
let turn_has_concrete_model_override =
req.provider_override.is_some() || req.model_override.is_some();
let (turn_inference_router, turn_inference_router_profile) = match &selection_mode {
Some(ModelSelectionMode::Auto {
router_id, profile, ..
}) if !turn_has_concrete_model_override => (
RuntimeInferenceRouterConfig {
enabled: true,
router_id: Some(router_id.clone()),
},
profile.clone(),
),
_ => (RuntimeInferenceRouterConfig::disabled(), None),
};
let mut provider = default_provider.clone();
let mut model = default_model.clone();
let mut model_profile = model_profile_for_provider_model(&cfg, &provider, &model);
let workspace = req.workspace.clone();
let mut transcript = self.transcript_for_turn(&req, &turn_id, &model).await?;
let mut compacted_this_turn = transcript
.iter()
.any(crate::compaction::is_compaction_boundary);
let effective_policy_mode = self.effective_policy_mode_for_thread(&req.thread_id).await;
let agent_swarm_mode_active = self
.effective_agent_swarm_mode_for_thread(&req.thread_id)
.await;
let thread_overrides = self.thread_turn_overrides(&req.thread_id).await?;
let mut final_assistant_text = String::new();
let mut final_phase_messages = Vec::<AssistantMessage>::new();
let mut final_reasoning_text = String::new();
let mut final_provider_metadata = None;
let mut exhausted_tool_rounds = true;
let mut verification_gate =
VerificationGateState::new(req.message.clone(), runtime_profile);
let mut speed_policy = SpeedPolicyState::default();
let mut reliability = TurnReliabilityState::default();
let mut turn_usage = TokenUsage::default();
// Overwritten on every inference step's Completed event so only the
// terminal step's stop reason survives (mid-turn tool_use steps must
// not leak onto turn/completed).
let mut turn_finish_reason: Option<String> = None;
let mut deadline_finalization_requested = false;
let mut deadline_scoreable_completion_requested = false;
let mut task_ledger_completion_reminders = 0_u8;
let mut task_ledger_scoreable_checkpoints = 0_u8;
let mut empty_tool_call_nudges_used = 0_u32;
let mut provider_stream_retry_attempts = 0_u32;
let mut routing_candidates = None;
let routing_transcript_start = transcript.len().saturating_sub(1);
let mut routing_escalations = 0_u32;
let mut model_switch_summary_selection = None::<ModelSelection>;
'tool_rounds: for round_index in 0..MAX_TOOL_ROUNDS_PER_TURN {
if let Some(deadline) = turn_deadline
&& deadline_expired(deadline)
{
self.fail_turn_due_to_deadline(&req.thread_id, &turn_id, deadline, &transcript)
.await?;
return Ok(TurnRunOutcome::Stopped);
}
let steers = self.drain_turn_steers(&turn_id).await;
self.append_steers(&req, &turn_id, &mut transcript, steers)
.await?;
if runtime_profile == RuntimeProfile::Eval
&& let Some(remaining) = crate::deadline_policy::should_start_finalization(
turn_deadline,
deadline_finalization_reserve,
deadline_finalization_requested || deadline_scoreable_completion_requested,
)
{
if req.task_ledger_required
&& task_ledger_completion_reminders < TASK_LEDGER_COMPLETION_REMINDER_LIMIT
&& let Some(prompt) = task_ledger_completion_prompt(&transcript)
{
task_ledger_completion_reminders += 1;
deadline_scoreable_completion_requested = true;
let item = TranscriptItem::UserMessage(UserMessage::text(
task_ledger_deadline_completion_prompt(
remaining,
deadline_finalization_reserve,
&prompt,
),
));
self.persist_turn_item(&req.thread_id, &turn_id, &item)
.await?;
transcript.push(item);
continue 'tool_rounds;
} else {
self.start_deadline_finalization(
&req.thread_id,
&turn_id,
&mut transcript,
remaining,
)
.await?;
deadline_finalization_requested = true;
}
}
if runtime_profile == RuntimeProfile::Eval
&& req.task_ledger_required
&& !deadline_finalization_requested
&& task_ledger_scoreable_checkpoints < TASK_LEDGER_SCOREABLE_CHECKPOINT_LIMIT
&& let Some(remaining) = deadline_remaining_seconds(turn_deadline)
&& remaining <= TASK_LEDGER_SCOREABLE_CHECKPOINT_SECONDS
&& remaining > deadline_finalization_reserve
&& let Some(prompt) = task_ledger_completion_prompt(&transcript)
{
task_ledger_scoreable_checkpoints += 1;
let item = TranscriptItem::UserMessage(UserMessage::text(
task_ledger_scoreable_checkpoint_prompt(remaining, &prompt),
));
self.persist_turn_item(&req.thread_id, &turn_id, &item)
.await?;
transcript.push(item);
continue 'tool_rounds;
}
if turn_inference_router.is_active() && routing_candidates.is_none() {
routing_candidates =
Some(collect_inference_routing_candidates(&self.registry).await);
}
let routing_tools_model = model.clone();
let routing_tools = self.filtered_tool_specs(
&cfg,
&model,
model_profile.as_ref(),
&thread_overrides.tool_allowlist,
&thread_overrides.external_tools,
);
let prior_failures =
transcript_failure_count_since(&transcript, routing_transcript_start)
.max(reliability.tool_failure_count())
.saturating_add(provider_stream_retry_attempts);
let routing_selection = route_inference_selection(
&self.registry,
&turn_inference_router,
InferenceRoutingRequest {
thread_id: &req.thread_id,
turn_id: &turn_id,
round_index: round_index as u32,
runtime_profile,
phase: speed_policy.phase(),
profile: turn_inference_router_profile.as_deref(),
default_selection: ModelSelection {
provider: default_provider.clone(),
model: default_model.clone(),
},
transcript: &transcript,
tools: &routing_tools,
candidates: routing_candidates.as_deref(),
prior_failures,
prior_escalations: routing_escalations,
},
)
.await;
if let Some(decision) = routing_selection.decision.clone() {
if matches!(decision.outcome, InferenceRoutingOutcome::Escalated) {
routing_escalations = routing_escalations.saturating_add(1);
}
self.emit(RoderEvent::InferenceRoutingDecision(
InferenceRoutingDecisionEvent {
thread_id: req.thread_id.clone(),
turn_id: turn_id.clone(),
round_index: round_index as u32,
default_selection: ModelSelection {
provider: default_provider.clone(),
model: default_model.clone(),
},
selected_selection: routing_selection.selection.clone(),
decision,
timestamp: OffsetDateTime::now_utc(),
},
))
.await;
}
provider = routing_selection.selection.provider.clone();
model = routing_selection.selection.model.clone();
let engine = self.engine_for(&provider)?;
let capabilities = engine.capabilities();
model_profile = model_profile_for_provider_model(&cfg, &provider, &model);
let tools = if capabilities.tool_calls {
if model == routing_tools_model {
routing_tools.clone()
} else {
self.filtered_tool_specs(
&cfg,
&model,
model_profile.as_ref(),
&thread_overrides.tool_allowlist,
&thread_overrides.external_tools,
)
}
} else {
Vec::new()
};
let parallel_tool_calls = parallel_tool_calls_for_model(&cfg, &model);
let tool_choice = if tools.is_empty() {
ToolChoice::None
} else {
ToolChoice::Auto
};
let summary_selection = ModelSelection {
provider: provider.clone(),
model: model.clone(),
};
if model_switch_summary_selection.as_ref() != Some(&summary_selection) {
if let Some(summary) = model_switch_summary(
&transcript,
model_profile.as_ref(),
&provider,
&model,
&tools,
) {
let item = TranscriptItem::UserMessage(UserMessage::text(summary));
self.persist_turn_item(&req.thread_id, &turn_id, &item)
.await?;
transcript.push(item);
}
model_switch_summary_selection = Some(summary_selection);
}
if !capabilities.image_input && transcript_has_images(&transcript) {
self.fail_turn_with_error(
&req.thread_id,
&turn_id,
format!("provider {provider} does not support image input"),
)
.await?;
return Ok(TurnRunOutcome::Stopped);
}
transcript = self
.compact_transcript_if_needed(
&req.thread_id,
&turn_id,
&provider,
&model,
transcript,
self.compaction_options_for_turn(&req.thread_id, !compacted_this_turn),
)
.await?;
compacted_this_turn = compacted_this_turn
|| transcript
.iter()
.any(crate::compaction::is_compaction_boundary);
let speed_policy_decision =
speed_policy.decision(runtime_profile, &model, &cfg.speed_policy);
let request_reasoning = reasoning_from_decision(
speed_policy_decision.as_ref(),
routing_selection
.reasoning
.clone()
.unwrap_or_else(|| reasoning_for_model(&cfg, &model)),
);
self.active_turn_selections.write().await.insert(
turn_id.clone(),
ModelSelectionMode::manual(
provider.clone(),
model.clone(),
request_reasoning.level.clone(),
),
);
if let Some(limit) = reliability.record_model_call(
&cfg.reliability,
runtime_profile == RuntimeProfile::Interactive,
) {
self.fail_turn_due_to_reliability_limit(
&req.thread_id,
&turn_id,
&provider,
&model,
limit,
&transcript,
)
.await?;
return Ok(TurnRunOutcome::Stopped);
}
self.emit(RoderEvent::InferenceStarted(InferenceStarted {
thread_id: req.thread_id.clone(),
turn_id: turn_id.clone(),
engine_id: engine.id(),
model: ModelSelection {
provider: provider.clone(),
model: model.clone(),
},
reasoning: request_reasoning.clone(),
speed_policy: speed_policy_decision.clone(),
deadline_remaining_seconds: deadline_remaining_seconds(turn_deadline),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
let mut instructions = req.instructions.clone();
if let Some(extra) = &thread_overrides.developer_instructions {
instructions = apply_thread_developer_instructions(instructions, extra);
}
if let Some(context) = req.developer_context.as_deref() {
instructions = apply_turn_developer_context(instructions, context);
}
let mut instructions = apply_runtime_profile(instructions, runtime_profile);
if let Some(profile) = &model_profile {
instructions = apply_model_instruction_overlay(instructions, profile);
}
if req.task_ledger_required
&& runtime_profile == RuntimeProfile::Eval
&& !transcript_has_task_ledger(&transcript)
{
instructions = apply_task_ledger_required(instructions);
}
if effective_policy_mode == PolicyMode::Plan {
instructions = apply_plan_mode(instructions);
}
if agent_swarm_mode_active {
instructions = apply_agent_swarm_mode(instructions);
}
if model_supports_reasoning_effort(&model, REASONING_ULTRA) {
instructions = apply_codex_multi_agent_mode(
instructions,
request_reasoning.level.as_deref() == Some(REASONING_ULTRA),
);
}
instructions = self
.goals
.apply_goal_instructions(&req.thread_id, instructions)
.await?;
let mut request_metadata = serde_json::json!({});
if let Some(decision) = &speed_policy_decision {
request_metadata["speedPolicy"] = serde_json::json!(decision);
}
if let Some(decision) = routing_selection.decision.as_ref() {
request_metadata["inferenceRouting"] = serde_json::json!(decision);
}
if let Some(remaining) = deadline_remaining_seconds(turn_deadline) {
request_metadata["deadlineRemainingSeconds"] = serde_json::json!(remaining);
}
if let Some(profile) = &model_profile {
request_metadata["modelProfile"] = serde_json::json!({
"model": profile.model,
"providerFamily": profile.provider_family,
"editTool": profile.edit_tool,
"schemaPolicy": profile.schema_policy,
"instructionOverlay": profile.instruction_overlay,
"parallelToolCalls": profile.parallel_tool_calls,
"autoCompactTokenLimit": profile.auto_compact_token_limit,
});
}
let task_ledger_required_this_round = req.task_ledger_required
&& runtime_profile == RuntimeProfile::Eval
&& !deadline_finalization_requested
&& !transcript_has_task_ledger(&transcript);
let task_ledger_tools = (capabilities.tool_calls && task_ledger_required_this_round)
.then(|| self.task_ledger_tool_specs(model_profile.as_ref()))
.filter(|tools| !tools.is_empty());
let request_tools = if deadline_finalization_requested {
Vec::new()
} else if let Some(ledger_tools) = &task_ledger_tools {
ledger_tools.clone()
} else {
tools.clone()
};
let request_tool_choice = if deadline_finalization_requested {
ToolChoice::None
} else if task_ledger_tools.is_some() {
ToolChoice::Specific(TASK_LEDGER_TOOL_NAME.to_string())
} else {
tool_choice.clone()
};
if deadline_finalization_requested {
request_metadata["deadlineFinalization"] = serde_json::json!({
"reserveSeconds": deadline_finalization_reserve,
"remainingSeconds": deadline_remaining_seconds(turn_deadline),
});
} else if deadline_scoreable_completion_requested {
request_metadata["deadlineScoreableCompletion"] = serde_json::json!({
"reserveSeconds": deadline_finalization_reserve,
"remainingSeconds": deadline_remaining_seconds(turn_deadline),
});
}
let request = AgentInferenceRequest {
model: ModelSelection {
provider: provider.clone(),
model: model.clone(),
},
instructions,
transcript: transcript.clone(),
tools: request_tools,
tool_choice: request_tool_choice,
reasoning: request_reasoning,
output: OutputConfig::default(),
runtime: RuntimeHints {
auto_compact_token_limit: server_side_compaction_threshold(&cfg, &model),
profile: runtime_profile,
parallel_tool_calls: Some(parallel_tool_calls),
hosted_web_search: cfg.hosted_web_search.clone(),
tool_search: tool_search_for_provider_model(&cfg, &provider, &model),
speed_policy: speed_policy_decision,
reliability: Some(cfg.reliability.clone().into()),
deadline_remaining_seconds: deadline_remaining_seconds(turn_deadline),
..RuntimeHints::default()
},
metadata: request_metadata,
};
let ctx = InferenceTurnContext {
thread_id: &req.thread_id,
turn_id: &turn_id,
tool_executor: Some(std::sync::Arc::new(
crate::tool_execution::RuntimeTurnToolExecutor {
runtime: Arc::clone(self),
thread_id: req.thread_id.clone(),
turn_id: turn_id.clone(),
workspace: Some(workspace.clone()),
deadline: turn_deadline,
},
)),
};
let stream_future = engine.stream_turn(ctx, request);
let mut stream = if let Some((deadline, timeout_action)) = inference_timeout_deadline(
turn_deadline,
runtime_profile,
req.task_ledger_required,
deadline_finalization_reserve,
deadline_finalization_requested || deadline_scoreable_completion_requested,
task_ledger_scoreable_checkpoints,
&transcript,
) {
match tokio::time::timeout_at(deadline_instant(deadline), stream_future).await {
Ok(stream) => stream?,
Err(_) => {
if runtime_profile == RuntimeProfile::Eval
&& !deadline_finalization_requested
{
let remaining = deadline_remaining_seconds(turn_deadline).unwrap_or(0);
if timeout_action == InferenceTimeoutAction::ScoreableCheckpoint
&& task_ledger_scoreable_checkpoints
< TASK_LEDGER_SCOREABLE_CHECKPOINT_LIMIT
&& let Some(prompt) = task_ledger_completion_prompt(&transcript)
{
task_ledger_scoreable_checkpoints += 1;
let item = TranscriptItem::UserMessage(UserMessage::text(
task_ledger_scoreable_checkpoint_prompt(remaining, &prompt),
));
self.persist_turn_item(&req.thread_id, &turn_id, &item)
.await?;
transcript.push(item);
continue 'tool_rounds;
}
self.start_deadline_finalization(
&req.thread_id,
&turn_id,
&mut transcript,
remaining,
)
.await?;
deadline_finalization_requested = true;
continue 'tool_rounds;
}
self.fail_turn_due_to_deadline(
&req.thread_id,
&turn_id,
deadline,
&transcript,
)
.await?;
return Ok(TurnRunOutcome::Stopped);
}
}
} else {
stream_future.await?
};
let mut assistant_text = String::new();
let mut phase_messages = Vec::<AssistantMessage>::new();
let mut reasoning_text = String::new();
let mut tool_calls = Vec::new();
let mut provider_metadata = None;
// Captured from mid-stream Compaction(completed) events. Codex/OpenAI
// sometimes abort the SSE stream after emitting the compaction item
// ("error decoding response body") before response.completed, so we
// must not rely solely on ProviderMetadata for the boundary.
let mut stream_compaction_item: Option<serde_json::Value> = None;
loop {
let next = if let Some((deadline, timeout_action)) = inference_timeout_deadline(
turn_deadline,
runtime_profile,
req.task_ledger_required,
deadline_finalization_reserve,
deadline_finalization_requested || deadline_scoreable_completion_requested,
task_ledger_scoreable_checkpoints,
&transcript,
) {
match tokio::time::timeout_at(deadline_instant(deadline), stream.next()).await {
Ok(next) => next,
Err(_) => {
if runtime_profile == RuntimeProfile::Eval
&& !deadline_finalization_requested
{
let remaining =
deadline_remaining_seconds(turn_deadline).unwrap_or(0);
if timeout_action == InferenceTimeoutAction::ScoreableCheckpoint
&& task_ledger_scoreable_checkpoints
< TASK_LEDGER_SCOREABLE_CHECKPOINT_LIMIT
&& let Some(prompt) = task_ledger_completion_prompt(&transcript)
{
task_ledger_scoreable_checkpoints += 1;
let item = TranscriptItem::UserMessage(UserMessage::text(
task_ledger_scoreable_checkpoint_prompt(remaining, &prompt),
));
self.persist_turn_item(&req.thread_id, &turn_id, &item)
.await?;
transcript.push(item);
continue 'tool_rounds;
}
self.start_deadline_finalization(
&req.thread_id,
&turn_id,
&mut transcript,
remaining,
)
.await?;
deadline_finalization_requested = true;
continue 'tool_rounds;
}
self.fail_turn_due_to_deadline(
&req.thread_id,
&turn_id,
deadline,
&transcript,
)
.await?;
return Ok(TurnRunOutcome::Stopped);
}
}
} else {
stream.next().await
};
let Some(res) = next else {
break;
};
let event = match res {
Ok(event) => event,
Err(err) => {
let error = err.to_string();
if runtime_profile == RuntimeProfile::Eval
&& !deadline_finalization_requested
&& let Some(cause) = provider_stream_retry_cause(&error)
{
let retry_attempt = provider_stream_retry_attempts.saturating_add(1);
let policy: ReliabilityRequestPolicy = cfg.reliability.clone().into();
if retry_attempt < policy.provider_retry_max_attempts {
provider_stream_retry_attempts = retry_attempt;
let delay_ms = provider_retry_delay_ms(&policy, retry_attempt);
self.emit(RoderEvent::ReliabilityRetryRecorded(
ReliabilityRetryRecorded {
context: ReliabilityContext {
thread_id: req.thread_id.clone(),
turn_id: turn_id.clone(),
provider: Some(provider.clone()),
model: Some(model.clone()),
..ReliabilityContext::default()
},
error_class: ReliabilityErrorClass::ProviderError,
decision: ReliabilityRetryDecision::Retry,
attempt: retry_attempt,
max_attempts: policy.provider_retry_max_attempts,
delay_ms: Some(delay_ms),
details: ReliabilityDetails::redacted(format!(
"{cause}: {error}"
)),
timestamp: OffsetDateTime::now_utc(),
},
))
.await;
if delay_ms > 0 {
tokio::time::sleep(std::time::Duration::from_millis(delay_ms))
.await;
}
continue 'tool_rounds;
}
}
self.emit(RoderEvent::TurnFailed(TurnFailed {
thread_id: req.thread_id.clone(),
turn_id: turn_id.clone(),
error: error.clone(),
error_kind: None,
usage: None,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
self.complete_team_member_turn_with_result(
&req.thread_id,
&turn_id,
TeamMemberStatus::Failed,
(!assistant_text.trim().is_empty()).then(|| assistant_text.clone()),
Some(error),
)
.await?;
return Err(err);
}
};
let inference_timestamp = OffsetDateTime::now_utc();
self.emit(RoderEvent::InferenceEventReceived(InferenceEventReceived {
thread_id: req.thread_id.clone(),
turn_id: turn_id.clone(),
event: event.clone(),
timestamp: inference_timestamp,
}))
.await;
match event {
InferenceEvent::MessageDelta(delta) => {
if let Some((team_id, member)) =
self.teams.member_for_thread(&req.thread_id).await
{
self.emit(RoderEvent::TeamMemberMessageDelta(TeamMemberMessageDelta {
team_id,
member_id: member.id,
member_thread_id: req.thread_id.clone(),
turn_id: turn_id.clone(),
delta: delta.text.clone(),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
}
if is_final_answer_phase(delta.phase.as_deref()) {
assistant_text.push_str(&delta.text);
} else if let Some(last) = phase_messages.last_mut()
&& last.phase == delta.phase
{
last.text.push_str(&delta.text);
} else {
phase_messages.push(AssistantMessage {
text: delta.text,
phase: delta.phase,
});
}
}
InferenceEvent::ReasoningDelta(delta) => reasoning_text.push_str(&delta.text),
InferenceEvent::ToolCallCompleted(call) => tool_calls.push(call),
InferenceEvent::Failed(failure) => {
speed_policy.record_failure();
let error = failure.message;
self.persist_turn_item(
&req.thread_id,
&turn_id,
&TranscriptItem::Error(ErrorRecord {
message: error.clone(),
}),
)
.await?;
self.emit(RoderEvent::TurnFailed(TurnFailed {
thread_id: req.thread_id.clone(),
turn_id: turn_id.clone(),
error: error.clone(),
error_kind: None,
usage: None,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
self.complete_team_member_turn_with_result(
&req.thread_id,
&turn_id,
TeamMemberStatus::Failed,
(!assistant_text.trim().is_empty()).then(|| assistant_text.clone()),
Some(error),
)
.await?;
return Ok(TurnRunOutcome::Stopped);
}
InferenceEvent::Usage(usage) => {
turn_usage.add_assign(&usage);
}
InferenceEvent::Completed(metadata) => {
turn_finish_reason = metadata
.stop_reason
.as_deref()
.map(finish_reason_from_stop_reason);
}
InferenceEvent::Compaction(progress) => {
// Persist the boundary as soon as the provider emits a
// completed compaction item. ChatGPT Codex often aborts
// the SSE stream right after ("error decoding response
// body"), so waiting for ProviderMetadata loses the
// boundary and every subsequent request re-compacts.
if progress.status == "completed"
&& let Some(item) = progress.item
{
let already = stream_compaction_item
.as_ref()
.and_then(|existing| {
existing.get("id").and_then(serde_json::Value::as_str)
})
.zip(item.get("id").and_then(serde_json::Value::as_str))
.is_some_and(|(a, b)| a == b);
if !already {
stream_compaction_item = Some(item.clone());
let boundary = TranscriptItem::ProviderMetadata(
crate::compaction::provider_metadata_from_compaction_item(item),
);
self.persist_turn_item(&req.thread_id, &turn_id, &boundary)
.await?;
transcript.push(boundary);
transcript =
crate::compaction::trim_to_last_compaction_boundary(transcript);
compacted_this_turn = true;
}
}
}
InferenceEvent::HostedToolCallStarted(_)
| InferenceEvent::HostedToolCallCompleted(_)
| InferenceEvent::ToolCallStarted(_)
| InferenceEvent::ToolCallDelta(_) => {}
InferenceEvent::ProviderMetadata(mut metadata) => {
if let Some(compaction_item) = stream_compaction_item.as_ref() {
let _ = crate::compaction::ensure_provider_metadata_compaction(
&mut metadata,
compaction_item,
);
}
provider_metadata = Some(metadata);
}
}
}
speed_policy.record_model_output(
!assistant_text.is_empty() || !phase_messages.is_empty(),
tool_calls.len(),
);
if tool_calls.is_empty() {
let steers = self.drain_turn_steers(&turn_id).await;
if !steers.is_empty() {
for message in phase_messages {
let item = TranscriptItem::AssistantMessage(message);
self.persist_turn_item(&req.thread_id, &turn_id, &item)
.await?;
transcript.push(item);
self.persist_model_profile_segment(
&req.thread_id,
&turn_id,
model_profile.as_ref(),
&provider,
&model,
"assistant",
)
.await?;
}
if !assistant_text.is_empty() {
let assistant = TranscriptItem::AssistantMessage(AssistantMessage {
text: assistant_text,
phase: Some(FINAL_ANSWER_PHASE.to_string()),
});
self.persist_turn_item(&req.thread_id, &turn_id, &assistant)
.await?;
transcript.push(assistant);
self.persist_model_profile_segment(
&req.thread_id,
&turn_id,
model_profile.as_ref(),
&provider,
&model,
"assistant",
)
.await?;
}
if let Some(metadata) = provider_metadata {
let had_provider_compaction =
crate::compaction::provider_metadata_has_compaction(&metadata);
let item = TranscriptItem::ProviderMetadata(metadata);
self.persist_turn_item(&req.thread_id, &turn_id, &item)
.await?;
transcript.push(item);
if had_provider_compaction {
// Server-side compaction replaced the prior window;
// drop pre-boundary items so the next request does
// not re-send (and re-compact) the full history.
transcript =
crate::compaction::trim_to_last_compaction_boundary(transcript);
compacted_this_turn = true;
}
}
self.append_steers(&req, &turn_id, &mut transcript, steers)
.await?;
continue;
}
if !deadline_finalization_requested
&& req.task_ledger_required
&& runtime_profile == RuntimeProfile::Eval
&& task_ledger_completion_reminders < TASK_LEDGER_COMPLETION_REMINDER_LIMIT
&& (!assistant_text.trim().is_empty() || !phase_messages.is_empty())
&& let Some(prompt) = task_ledger_completion_prompt(&transcript)
{
task_ledger_completion_reminders += 1;
let item = TranscriptItem::UserMessage(UserMessage::text(prompt));
self.persist_turn_item(&req.thread_id, &turn_id, &item)
.await?;
transcript.push(item);
continue;
}
if !deadline_finalization_requested
&& let Some(prompt) = verification_gate.blocking_prompt()
{
speed_policy.record_verification_required();
self.emit(RoderEvent::VerificationRequired(VerificationRequired {
thread_id: req.thread_id.clone(),
turn_id: turn_id.clone(),
reason: verification_gate.reason(),
changed_files: verification_gate.changed_files(),
tool_evidence: verification_gate.tool_evidence.clone(),
tests_run: verification_gate.tests_run.clone(),
open_gaps: verification_gate.open_gaps.clone(),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
let item = TranscriptItem::UserMessage(UserMessage::text(prompt));
self.persist_turn_item(&req.thread_id, &turn_id, &item)
.await?;
transcript.push(item);
continue;
}
// Loop persistence nudge: in non-interactive/eval runs, when the
// model returns a final message with no tool calls, gently prompt
// it to keep going (bounded by `empty_tool_call_nudges`) before
// ending the turn. Gated to non-interactive/eval so interactive
// chat completions are never delayed, and only fires when the
// model actually produced a final answer to re-examine.
if !deadline_finalization_requested
&& runtime_profile != RuntimeProfile::Interactive
&& cfg.reliability.empty_tool_call_nudges > 0
&& empty_tool_call_nudges_used < cfg.reliability.empty_tool_call_nudges
&& (!assistant_text.trim().is_empty() || !phase_messages.is_empty())
{
empty_tool_call_nudges_used += 1;
let item = TranscriptItem::UserMessage(UserMessage::text(
RELIABILITY_CONTINUATION_PROMPT.to_string(),
));
self.persist_turn_item(&req.thread_id, &turn_id, &item)
.await?;
transcript.push(item);
continue;
}
if deadline_finalization_requested
&& assistant_text.trim().is_empty()
&& phase_messages.is_empty()
{
assistant_text = format!(
"Deadline finalization completed without model text. {}",
turn_partial_result(&transcript)
);
}
final_phase_messages = phase_messages;
final_assistant_text = assistant_text;
final_reasoning_text = reasoning_text;
final_provider_metadata = provider_metadata;
exhausted_tool_rounds = false;
break;
}
for message in phase_messages {
let item = TranscriptItem::AssistantMessage(message);
self.persist_turn_item(&req.thread_id, &turn_id, &item)
.await?;
transcript.push(item);
self.persist_model_profile_segment(
&req.thread_id,
&turn_id,
model_profile.as_ref(),
&provider,
&model,
"assistant",
)
.await?;
}
if !assistant_text.is_empty() {
transcript.push(TranscriptItem::AssistantMessage(AssistantMessage {
text: assistant_text,
phase: Some(FINAL_ANSWER_PHASE.to_string()),
}));
self.persist_model_profile_segment(
&req.thread_id,
&turn_id,
model_profile.as_ref(),
&provider,
&model,
"assistant",
)
.await?;
}
if let Some(metadata) = provider_metadata {
let had_provider_compaction =
crate::compaction::provider_metadata_has_compaction(&metadata);
let item = TranscriptItem::ProviderMetadata(metadata);
self.persist_turn_item(&req.thread_id, &turn_id, &item)
.await?;
transcript.push(item);
if had_provider_compaction {
// Server-side compaction replaced the prior window; drop
// pre-boundary items so subsequent tool rounds use the
// compact window as the next input.
transcript = crate::compaction::trim_to_last_compaction_boundary(transcript);
compacted_this_turn = true;
}
}
for call in &tool_calls {
let tool_item = TranscriptItem::ToolCall(ToolCallRecord {
id: call.id.clone(),
name: call.name.clone(),
arguments: call.arguments.clone(),
});
self.persist_turn_item(&req.thread_id, &turn_id, &tool_item)
.await?;
transcript.push(tool_item);
self.persist_model_profile_segment(
&req.thread_id,
&turn_id,
model_profile.as_ref(),
&provider,
&model,
"tool_call",
)
.await?;
}
if let Some(deadline) = turn_deadline
&& deadline_expired(deadline)
{
self.fail_turn_due_to_deadline(&req.thread_id, &turn_id, deadline, &transcript)
.await?;
return Ok(TurnRunOutcome::Stopped);
}
let results = self
.route_tool_calls(
&req.thread_id,
&turn_id,
tool_calls,
parallel_tool_calls,
Some(workspace.as_str()),
turn_deadline,
)
.await?;
let reliability_limit = reliability.record_tool_results(
&cfg.reliability,
&results,
runtime_profile == RuntimeProfile::Interactive,
);
for result in results {
verification_gate.record_tool_result(&result);
transcript.push(TranscriptItem::ToolResult(result));
self.persist_model_profile_segment(
&req.thread_id,
&turn_id,
model_profile.as_ref(),
&provider,
&model,
"tool_result",
)
.await?;
}
if let Some(limit) = reliability_limit {
if limit.decision == ReliabilityLimitDecision::RequestContinuation {
// Loop persistence: rather than ending the turn, reset the
// consecutive-failure counter, nudge the model to keep going,
// and continue the round loop. Bounded by the per-turn tool
// failure ceiling, `max_model_calls_per_turn`, and
// `MAX_TOOL_ROUNDS_PER_TURN`.
self.record_reliability_limit_continuation(
&req.thread_id,
&turn_id,
&provider,
&model,
limit,
)
.await;
reliability.reset_consecutive_failures();
let nudge = TranscriptItem::UserMessage(UserMessage::text(
RELIABILITY_CONTINUATION_PROMPT.to_string(),
));
self.persist_turn_item(&req.thread_id, &turn_id, &nudge)
.await?;
transcript.push(nudge);
} else {
self.fail_turn_due_to_reliability_limit(
&req.thread_id,
&turn_id,
&provider,
&model,
limit,
&transcript,
)
.await?;
return Ok(TurnRunOutcome::Stopped);
}
}
transcript = self
.compact_transcript_if_needed(
&req.thread_id,
&turn_id,
&provider,
&model,
transcript,
self.compaction_options_for_turn(&req.thread_id, !compacted_this_turn),
)
.await?;
compacted_this_turn = compacted_this_turn
|| transcript
.iter()
.any(crate::compaction::is_compaction_boundary);
}
if exhausted_tool_rounds {
let message =
format!("tool call limit reached after {MAX_TOOL_ROUNDS_PER_TURN} rounds");
self.persist_turn_item(
&req.thread_id,
&turn_id,
&TranscriptItem::Error(ErrorRecord {
message: message.clone(),
}),
)
.await?;
self.emit(RoderEvent::TurnFailed(TurnFailed {
thread_id: req.thread_id.clone(),
turn_id: turn_id.clone(),
error: message.clone(),
error_kind: None,
usage: None,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
self.complete_team_member_turn_with_result(
&req.thread_id,
&turn_id,
TeamMemberStatus::Failed,
None,
Some(message),
)
.await?;
return Ok(TurnRunOutcome::Stopped);
}
if !final_reasoning_text.is_empty() {
self.persist_turn_item(
&req.thread_id,
&turn_id,
&TranscriptItem::ReasoningSummary(ReasoningSummary {
text: final_reasoning_text,
}),
)
.await?;
}
for message in final_phase_messages {
self.persist_turn_item(
&req.thread_id,
&turn_id,
&TranscriptItem::AssistantMessage(message),
)
.await?;
self.persist_model_profile_segment(
&req.thread_id,
&turn_id,
model_profile.as_ref(),
&provider,
&model,
"assistant",
)
.await?;
}
if !final_assistant_text.is_empty() {
self.persist_turn_item(
&req.thread_id,
&turn_id,
&TranscriptItem::AssistantMessage(AssistantMessage {
text: final_assistant_text.clone(),
phase: Some(FINAL_ANSWER_PHASE.to_string()),
}),
)
.await?;
self.persist_model_profile_segment(
&req.thread_id,
&turn_id,
model_profile.as_ref(),
&provider,
&model,
"assistant",
)
.await?;
}
if let Some(metadata) = final_provider_metadata {
self.persist_turn_item(
&req.thread_id,
&turn_id,
&TranscriptItem::ProviderMetadata(metadata),
)
.await?;
}
let turn_usage_tokens = turn_usage.total_tokens as i64;
let completed_usage = (!turn_usage.is_empty()).then_some(turn_usage.clone());
self.record_thread_usage_metadata(&req.thread_id, &turn_usage)
.await?;
self.goals
.account_turn_usage(
&req.thread_id,
turn_usage_tokens,
OffsetDateTime::now_utc() - turn_started_at,
)
.await?;
let (cleanup, ownership) = self.await_provider_turn_cleanup(&turn_id).await;
self.record_turn_lifecycle_with_ownership(
req.thread_id.clone(),
turn_id.clone(),
TurnLifecycleState::Completed,
cleanup,
None,
ownership,
)
.await;
self.emit(RoderEvent::TurnCompleted(TurnCompleted {
thread_id: req.thread_id.clone(),
turn_id: turn_id.clone(),
usage: completed_usage,
finish_reason: turn_finish_reason,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
self.complete_team_member_turn_with_result(
&req.thread_id,
&turn_id,
TeamMemberStatus::Completed,
(!final_assistant_text.is_empty()).then_some(final_assistant_text),
None,
)
.await?;
Ok(TurnRunOutcome::Completed)
}
async fn drain_turn_steers(&self, turn_id: &TurnId) -> Vec<QueuedTurnSteer> {
let Some(active) = self.active_turns.read().await.get(turn_id).cloned() else {
return Vec::new();
};
let mut steers = active.steers.lock().await;
std::mem::take(&mut *steers)
}
async fn route_tool_calls(
self: &Arc<Self>,
thread_id: &ThreadId,
turn_id: &TurnId,
calls: Vec<ToolCallCompleted>,
parallel: bool,
workspace: Option<&str>,
deadline: Option<OffsetDateTime>,
) -> anyhow::Result<Vec<ToolResultRecord>> {
// Enforce the agent_swarm exclusivity rule (roadmap 104, Task 2):
// `agent_swarm` must be the only tool call in a model response. A mixed
// or multi-swarm batch is denied wholesale with actionable retry text so
// the model re-issues `agent_swarm` by itself, and every tool_call_id
// still gets a response (keeping the chat-completions transcript valid).
if let Some(violation) = roder_api::subagents::agent_swarm_batch_violation(
calls.iter().map(|call| call.name.as_str()),
) {
let message = violation.deny_message();
return Ok(calls
.into_iter()
.map(|call| ToolResultRecord {
id: call.id,
name: Some(call.name),
result: message.clone(),
display_payload: None,
is_error: true,
})
.collect());
}
// Emit swarm lifecycle events on the bus (roadmap 104, Task 1) so any
// app-server/SDK/TUI client can observe a swarm as a whole; per-child
// progress flows through the existing Subagent* trace events.
let swarm_call = (calls.len() == 1
&& calls[0].name == roder_api::subagents::AGENT_SWARM_TOOL_NAME)
.then(|| (calls[0].id.clone(), calls[0].arguments.clone()));
if let Some((tool_id, args)) = &swarm_call {
self.emit(RoderEvent::AgentSwarmStarted(
roder_api::subagents::AgentSwarmStarted {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
tool_id: tool_id.clone(),
child_count: agent_swarm_child_count(args),
timestamp: OffsetDateTime::now_utc(),
},
))
.await;
}
let force_sequential = calls
.iter()
.any(|call| crate::agent_control_tools::is_agent_control_tool(&call.name));
let results =
if parallel && !force_sequential {
try_join_all(calls.into_iter().map(|call| {
self.route_tool_call(thread_id, turn_id, call, workspace, deadline)
}))
.await
} else {
let mut results = Vec::with_capacity(calls.len());
for call in calls {
results.push(
self.route_tool_call(thread_id, turn_id, call, workspace, deadline)
.await?,
);
}
Ok(results)
}?;
if let Some((tool_id, _)) = &swarm_call
&& let Some(result) = results.iter().find(|result| &result.id == tool_id)
&& let Some((completed, failed, aborted)) = parse_swarm_counts(&result.result)
{
self.emit(RoderEvent::AgentSwarmCompleted(
roder_api::subagents::AgentSwarmCompleted {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
tool_id: tool_id.clone(),
completed,
failed,
aborted,
timestamp: OffsetDateTime::now_utc(),
},
))
.await;
}
Ok(results)
}
async fn fail_turn_with_error(
&self,
thread_id: &ThreadId,
turn_id: &TurnId,
message: String,
) -> anyhow::Result<()> {
self.persist_turn_item(
thread_id,
turn_id,
&TranscriptItem::Error(ErrorRecord {
message: message.clone(),
}),
)
.await?;
self.emit(RoderEvent::TurnFailed(TurnFailed {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
error: message.clone(),
error_kind: None,
usage: None,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
self.complete_team_member_turn_with_result(
thread_id,
turn_id,
TeamMemberStatus::Failed,
None,
Some(message),
)
.await?;
Ok(())
}
async fn fail_turn_due_to_deadline(
&self,
thread_id: &ThreadId,
turn_id: &TurnId,
deadline: OffsetDateTime,
transcript: &[TranscriptItem],
) -> anyhow::Result<()> {
let partial_result = turn_partial_result(transcript);
self.emit(RoderEvent::TurnPartialResult(TurnPartialResult {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
summary: partial_result.clone(),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
self.emit(RoderEvent::TurnDeadlineExceeded(TurnDeadlineExceeded {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
deadline,
partial_result: partial_result.clone(),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
let message = "turn deadline expired".to_string();
self.persist_turn_item(
thread_id,
turn_id,
&TranscriptItem::Error(ErrorRecord {
message: format!("{message}: {partial_result}"),
}),
)
.await?;
self.emit(RoderEvent::TurnFailed(TurnFailed {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
error: message.clone(),
error_kind: Some("deadline_timeout".to_string()),
usage: None,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
self.complete_team_member_turn_with_result(
thread_id,
turn_id,
TeamMemberStatus::Failed,
None,
Some(format!("{message}: {partial_result}")),
)
.await?;
Ok(())
}
async fn start_deadline_finalization(
&self,
thread_id: &ThreadId,
turn_id: &TurnId,
transcript: &mut Vec<TranscriptItem>,
remaining_seconds: u64,
) -> anyhow::Result<()> {
let item = TranscriptItem::UserMessage(crate::deadline_policy::finalization_message(
remaining_seconds,
));
self.persist_turn_item(thread_id, turn_id, &item).await?;
transcript.push(item);
self.emit(RoderEvent::TurnPartialResult(TurnPartialResult {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
summary: turn_partial_result(transcript),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
Ok(())
}
/// Emits the reliability-limit event for a continuation (loop persistence)
/// without failing the turn. The caller resets the failure counter and
/// injects a nudge so the round loop keeps going.
async fn record_reliability_limit_continuation(
&self,
thread_id: &ThreadId,
turn_id: &TurnId,
provider: &str,
model: &str,
limit: ReliabilityLimitHit,
) {
self.emit(RoderEvent::ReliabilityLimitRecorded(
ReliabilityLimitRecorded {
context: ReliabilityContext {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
tool_id: None,
tool_name: None,
provider: Some(provider.to_string()),
model: Some(model.to_string()),
},
error_class: limit.error_class,
limit_kind: limit.limit_kind,
decision: limit.decision,
current: limit.current,
limit: limit.limit,
details: ReliabilityDetails::redacted(&limit.message),
timestamp: OffsetDateTime::now_utc(),
},
))
.await;
}
async fn fail_turn_due_to_reliability_limit(
&self,
thread_id: &ThreadId,
turn_id: &TurnId,
provider: &str,
model: &str,
limit: ReliabilityLimitHit,
transcript: &[TranscriptItem],
) -> anyhow::Result<()> {
self.emit(RoderEvent::ReliabilityLimitRecorded(
ReliabilityLimitRecorded {
context: ReliabilityContext {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
tool_id: None,
tool_name: None,
provider: Some(provider.to_string()),
model: Some(model.to_string()),
},
error_class: limit.error_class,
limit_kind: limit.limit_kind,
decision: limit.decision,
current: limit.current,
limit: limit.limit,
details: ReliabilityDetails::redacted(&limit.message),
timestamp: OffsetDateTime::now_utc(),
},
))
.await;
let partial_result = turn_partial_result(transcript);
self.emit(RoderEvent::TurnPartialResult(TurnPartialResult {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
summary: partial_result.clone(),
timestamp: OffsetDateTime::now_utc(),
}))
.await;
let message = format!("reliability limit reached: {}", limit.message);
self.persist_turn_item(
thread_id,
turn_id,
&TranscriptItem::Error(ErrorRecord {
message: format!("{message}: {partial_result}"),
}),
)
.await?;
self.emit(RoderEvent::TurnFailed(TurnFailed {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
error: message.clone(),
error_kind: Some("reliability_limit".to_string()),
usage: None,
timestamp: OffsetDateTime::now_utc(),
}))
.await;
self.complete_team_member_turn_with_result(
thread_id,
turn_id,
TeamMemberStatus::Failed,
None,
Some(format!("{message}: {partial_result}")),
)
.await?;
Ok(())
}
async fn append_steers(
&self,
req: &StartTurnRequest,
turn_id: &TurnId,
transcript: &mut Vec<TranscriptItem>,
steers: Vec<QueuedTurnSteer>,
) -> anyhow::Result<()> {
for queued in steers {
let mut steer = queued.message;
steer.text = steer.text.trim().to_string();
if steer.text.is_empty() && steer.images.is_empty() {
continue;
}
let item = TranscriptItem::UserMessage(steer);
self.persist_turn_item(&req.thread_id, turn_id, &item)
.await?;
transcript.push(item);
if let Some(ack) = queued.mailbox_ack {
self.teams
.mark_mailbox_messages_delivered(&ack.team_id, turn_id, &ack.message_ids)
.await?;
}
}
Ok(())
}
async fn persist_model_profile_segment(
&self,
thread_id: &ThreadId,
turn_id: &TurnId,
profile: Option<&ModelHarnessProfile>,
provider: &str,
model: &str,
segment: &str,
) -> anyhow::Result<()> {
let item = TranscriptItem::ProviderMetadata(model_profile_segment_metadata(
profile, provider, model, segment,
));
self.persist_turn_item(thread_id, turn_id, &item).await
}
/**
* Allowlists apply to built-in tools only; external tools are advertised with their
* host-supplied schemas as given. An external tool shadows a built-in with the same name in
* both advertisement and dispatch (see `route_tool_call`).
*/
fn filtered_tool_specs(
&self,
cfg: &RuntimeConfig,
model: &str,
profile: Option<&ModelHarnessProfile>,
thread_allowlist: &[String],
external_tools: &[roder_api::tools::ToolSpec],
) -> Vec<roder_api::tools::ToolSpec> {
let mut specs = self
.tool_registry
.specs_for_edit_tool_with_schema_policy(
edit_tool_for_model(cfg, model),
schema_policy_for_model(profile),
)
.into_iter()
.filter(|spec| {
allowlist_permits(&cfg.tool_allowlist, &spec.name)
&& allowlist_permits(thread_allowlist, &spec.name)
&& !external_tools.iter().any(|tool| tool.name == spec.name)
})
.collect::<Vec<_>>();
specs.extend(external_tools.iter().cloned());
specs
}
fn task_ledger_tool_specs(
&self,
profile: Option<&ModelHarnessProfile>,
) -> Vec<roder_api::tools::ToolSpec> {
self.tool_registry
.get(TASK_LEDGER_TOOL_NAME)
.map(|tool| {
tool.spec()
.normalized_for_model_profile(schema_policy_for_model(profile))
})
.into_iter()
.collect()
}
pub(crate) fn engine_for(&self, provider: &str) -> anyhow::Result<Arc<dyn InferenceEngine>> {
self.registry
.inference_engine(provider)
.or_else(|| {
self.registry
.default_inference_engine()
.filter(|engine| provider.is_empty() || engine.id() == provider)
})
.ok_or_else(|| anyhow::anyhow!("inference provider {provider:?} is not registered"))
}
pub async fn emit(&self, event: RoderEvent) -> EventEnvelope {
if let Some(record) = Self::lifecycle_record_for_event(&event) {
let _ = self.persist_turn_lifecycle_record(&record).await;
}
let envelope = self.bus.emit(event);
if let (Some(store), Some(thread_id)) = (&self.thread_store, envelope.thread_id.as_ref())
&& should_persist_thread_event(thread_id)
{
let _ = store.append_event(thread_id, &envelope).await;
}
// Registered event sinks (e.g. process extensions) receive the
// persisted envelope through bounded per-sink queues; a slow sink
// never blocks emit or turn progress.
let dispatcher = self
.event_sink_dispatcher
.get_or_init(|| async {
crate::event_sink_dispatch::EventSinkDispatcher::start(
&self.registry.event_sinks,
self.bus.clone(),
)
})
.await;
if !dispatcher.is_empty() {
dispatcher.dispatch(&envelope, &self.bus);
}
envelope
}
/// Records a projected thread item event and persists it through the configured thread store.
///
/// App-server protocol notification bridges currently call this after translating runtime
/// events into item events. Headless runtime consumers that subscribe to `RoderEvent` directly
/// must make the same call if they need the derived item event stream persisted.
pub async fn record_thread_item_event_kind(
&self,
thread_id: &ThreadId,
turn_id: &TurnId,
timestamp: OffsetDateTime,
kind: ThreadItemEventKind,
) -> anyhow::Result<ThreadItemEvent> {
let seq = self.next_thread_item_event_seq(thread_id).await?;
let item_event = ThreadItemEvent {
seq,
event_id: format!("{turn_id}-item-event-{seq}"),
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
timestamp,
event: kind,
};
if let Some(store) = &self.thread_store {
store.append_item_event(thread_id, &item_event).await?;
}
self.remember_thread_item_event(&item_event).await?;
Ok(item_event)
}
async fn next_thread_item_event_seq(&self, thread_id: &ThreadId) -> anyhow::Result<u64> {
self.ensure_thread_item_cache(thread_id).await?;
Ok(self
.thread_item_cache
.lock()
.await
.next_item_event_seq(thread_id))
}
pub async fn thread_item_exists(
&self,
thread_id: &ThreadId,
turn_id: &TurnId,
item_id: &str,
) -> anyhow::Result<bool> {
self.ensure_thread_item_cache(thread_id).await?;
Ok(self
.thread_item_cache
.lock()
.await
.thread_item_exists(thread_id, turn_id, item_id))
}
pub async fn current_reasoning_item_id(
&self,
thread_id: &ThreadId,
turn_id: &TurnId,
) -> anyhow::Result<Option<String>> {
self.ensure_thread_item_cache(thread_id).await?;
Ok(self
.thread_item_cache
.lock()
.await
.current_reasoning_item_id(thread_id, turn_id))
}
async fn remember_thread_item_event(&self, item_event: &ThreadItemEvent) -> anyhow::Result<()> {
self.ensure_thread_item_cache(&item_event.thread_id).await?;
self.thread_item_cache
.lock()
.await
.remember_item_event(item_event);
Ok(())
}
pub async fn latest_transcript_item_index(
&self,
thread_id: &ThreadId,
turn_id: &TurnId,
) -> anyhow::Result<Option<usize>> {
self.ensure_thread_item_cache(thread_id).await?;
Ok(self
.thread_item_cache
.lock()
.await
.latest_transcript_item_index(thread_id, turn_id))
}
async fn next_transcript_item_index(
&self,
thread_id: &ThreadId,
turn_id: &TurnId,
) -> anyhow::Result<usize> {
self.ensure_thread_item_cache(thread_id).await?;
Ok(self
.thread_item_cache
.lock()
.await
.next_transcript_item_index(thread_id, turn_id))
}
async fn remember_transcript_item_index(
&self,
thread_id: &ThreadId,
turn_id: &TurnId,
item_index: usize,
) -> anyhow::Result<()> {
self.ensure_thread_item_cache(thread_id).await?;
self.thread_item_cache
.lock()
.await
.remember_transcript_item_index(thread_id, turn_id, item_index);
Ok(())
}
async fn ensure_thread_item_cache(&self, thread_id: &ThreadId) -> anyhow::Result<()> {
if self
.thread_item_cache
.lock()
.await
.contains_thread(thread_id)
{
return Ok(());
}
let snapshot = if let Some(store) = &self.thread_store {
store.load_thread(thread_id).await?
} else {
None
};
self.thread_item_cache.lock().await.ensure_thread(
thread_id,
ThreadItemCacheEntry::from_snapshot(snapshot.as_ref()),
);
Ok(())
}
pub(crate) async fn persist_turn_item(
&self,
thread_id: &ThreadId,
turn_id: &TurnId,
item: &TranscriptItem,
) -> anyhow::Result<()> {
let item_index = self.next_transcript_item_index(thread_id, turn_id).await?;
let timestamp = OffsetDateTime::now_utc();
self.emit(RoderEvent::TranscriptItemAppended(TranscriptItemAppended {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
item_type: match item {
TranscriptItem::UserMessage(_) => "user_message",
TranscriptItem::AssistantMessage(_) => "assistant_message",
TranscriptItem::ReasoningSummary(_) => "reasoning_summary",
TranscriptItem::ToolCall(_) => "tool_call",
TranscriptItem::ToolResult(_) => "tool_result",
TranscriptItem::FileChange(_) => "file_change",
TranscriptItem::ContextCompaction(_) => "context_compaction",
TranscriptItem::Error(_) => "error",
TranscriptItem::ProviderMetadata(_) => "provider_metadata",
}
.to_string(),
item_index: Some(item_index),
item: Some(item.clone()),
timestamp,
}))
.await;
self.remember_transcript_item_index(thread_id, turn_id, item_index)
.await?;
Ok(())
}
}
fn transcript_has_images(transcript: &[TranscriptItem]) -> bool {
transcript.iter().any(|item| {
matches!(
item,
TranscriptItem::UserMessage(message) if !message.images.is_empty()
)
})
}
fn transcript_has_task_ledger(transcript: &[TranscriptItem]) -> bool {
transcript.iter().any(|item| {
matches!(
item,
TranscriptItem::ToolResult(result)
if result.name.as_deref() == Some(TASK_LEDGER_TOOL_NAME) && !result.is_error
)
})
}
fn task_ledger_completion_prompt(transcript: &[TranscriptItem]) -> Option<String> {
let latest = transcript.iter().rev().find_map(|item| match item {
TranscriptItem::ToolResult(result)
if result.name.as_deref() == Some(TASK_LEDGER_TOOL_NAME) && !result.is_error =>
{
Some(result.result.as_str())
}
_ => None,
})?;
if !task_ledger_has_open_items(latest) {
return None;
}
let mut ledger = latest.chars().take(1500).collect::<String>();
if latest.chars().nth(1500).is_some() {
ledger.push_str("...");
}
Some(format!(
"Task Ledger Completion Required: the latest task ledger still has pending or in-progress items. Do not provide a final answer yet. Use tools to complete the remaining scoreable work, create or update any required output files, then call `{TASK_LEDGER_TOOL_NAME}` with every task completed and evidence before finalizing.\n\nLatest ledger:\n{ledger}"
))
}
fn task_ledger_deadline_completion_prompt(
remaining_seconds: u64,
reserve_seconds: u64,
completion_prompt: &str,
) -> String {
format!(
"Eval deadline scoreable completion: {remaining_seconds} seconds remain in the {reserve_seconds}-second finalization reserve. Do not browse, search, or start slow work. Use the available tools now to create or update the required scoreable output files, run only a quick local check if needed, then update the task ledger to completed before finalizing.\n\n{completion_prompt}"
)
}
fn task_ledger_scoreable_checkpoint_prompt(
remaining_seconds: u64,
completion_prompt: &str,
) -> String {
format!(
"Scoreable Output Checkpoint: {remaining_seconds} seconds remain before the eval deadline. Before any further research, browsing, or long commands, use tools now to ensure the required output file(s) exist with the best evidence-backed answer, even if provisional. If a scoreable file already exists, read it and preserve that candidate unless you have stronger task-specific evidence for a replacement. Do not overwrite a plausible dated, historical, or local-evidence candidate with a current live-page, partial-coverage, or weaker guess merely to refresh the checkpoint. You may continue refining afterward, but do not apologize or finalize until the scoreable file exists and the task ledger is updated.\n\n{completion_prompt}"
)
}
fn task_ledger_has_open_items(ledger: &str) -> bool {
ledger.lines().any(|line| {
let line = line.trim_start();
line.starts_with("- pending:") || line.starts_with("- in_progress:")
})
}
fn turn_deadline_for_config(cfg: &RuntimeConfig) -> Option<OffsetDateTime> {
if !cfg.runtime_profile.is_non_interactive() {
return None;
}
cfg.turn_deadline_seconds
.filter(|seconds| *seconds > 0)
.map(|seconds| OffsetDateTime::now_utc() + Duration::seconds(seconds as i64))
}
fn deadline_expired(deadline: OffsetDateTime) -> bool {
OffsetDateTime::now_utc() >= deadline
}
pub(crate) fn deadline_remaining_seconds(deadline: Option<OffsetDateTime>) -> Option<u64> {
let deadline = deadline?;
if deadline <= OffsetDateTime::now_utc() {
return Some(0);
}
Some(
(deadline - OffsetDateTime::now_utc())
.unsigned_abs()
.as_secs()
.max(1),
)
}
fn deadline_instant(deadline: OffsetDateTime) -> tokio::time::Instant {
let now = OffsetDateTime::now_utc();
if deadline <= now {
return tokio::time::Instant::now();
}
tokio::time::Instant::now() + (deadline - now).unsigned_abs()
}
fn inference_timeout_deadline(
deadline: Option<OffsetDateTime>,
runtime_profile: RuntimeProfile,
task_ledger_required: bool,
reserve_seconds: u64,
finalization_requested: bool,
task_ledger_scoreable_checkpoints: u8,
transcript: &[TranscriptItem],
) -> Option<(OffsetDateTime, InferenceTimeoutAction)> {
let deadline = deadline?;
if runtime_profile == RuntimeProfile::Eval
&& task_ledger_required
&& !finalization_requested
&& task_ledger_scoreable_checkpoints < TASK_LEDGER_SCOREABLE_CHECKPOINT_LIMIT
&& TASK_LEDGER_SCOREABLE_CHECKPOINT_SECONDS > reserve_seconds
&& task_ledger_completion_prompt(transcript).is_some()
{
let checkpoint_deadline =
deadline - Duration::seconds(TASK_LEDGER_SCOREABLE_CHECKPOINT_SECONDS as i64);
if checkpoint_deadline > OffsetDateTime::now_utc() {
return Some((
checkpoint_deadline,
InferenceTimeoutAction::ScoreableCheckpoint,
));
}
}
if runtime_profile == RuntimeProfile::Eval && !finalization_requested {
return Some((
deadline - Duration::seconds(reserve_seconds as i64),
InferenceTimeoutAction::Finalization,
));
}
Some((deadline, InferenceTimeoutAction::Finalization))
}
fn turn_partial_result(transcript: &[TranscriptItem]) -> String {
let tool_results = transcript
.iter()
.filter(|item| matches!(item, TranscriptItem::ToolResult(_)))
.count();
let assistant_messages = transcript
.iter()
.filter(|item| matches!(item, TranscriptItem::AssistantMessage(_)))
.count();
format!(
"partial turn state: {} transcript items, {assistant_messages} assistant messages, {tool_results} tool results",
transcript.len()
)
}
fn reasoning_for_model(cfg: &RuntimeConfig, model: &str) -> ReasoningConfig {
let level = effective_reasoning_for_model(cfg, model);
match level.as_str() {
"" | REASONING_NONE => ReasoningConfig::default(),
level => ReasoningConfig {
enabled: true,
level: Some(level.to_string()),
},
}
}
fn server_side_compaction_threshold(cfg: &RuntimeConfig, model: &str) -> Option<u32> {
let entry = lookup_model(model)?;
if !entry.supports_compaction {
return None;
}
cfg.auto_compact_token_limit
.or_else(|| {
model_profile_for_model(cfg, model).and_then(|profile| profile.auto_compact_token_limit)
})
.or(Some(entry.auto_compact_token_limit))
.filter(|threshold| *threshold > 0)
}
pub(crate) fn tool_search_for_provider_model(
cfg: &RuntimeConfig,
provider: &str,
model: &str,
) -> ToolSearchConfig {
let mut resolved = cfg.tool_search.clone();
if let Some(provider_config) = cfg.provider_tool_search.get(provider) {
provider_config.apply_to(&mut resolved);
}
if let Some(model_config) = cfg.model_tool_search.get(model) {
model_config.apply_to(&mut resolved);
}
resolved
}
fn parallel_tool_calls_for_model(cfg: &RuntimeConfig, model: &str) -> bool {
cfg.model_parallel_tool_calls
.get(model)
.copied()
.or_else(|| {
model_profile_for_model(cfg, model).and_then(|profile| profile.parallel_tool_calls)
})
.unwrap_or(true)
}
/// Count the children an `agent_swarm` call will launch from its arguments
/// (item-based spawns plus resumes). Lenient: returns 0 on malformed input.
fn agent_swarm_child_count(arguments: &str) -> usize {
serde_json::from_str::<roder_api::subagents::AgentSwarmRequest>(arguments)
.map(|request| request.items.len() + request.resume_agent_ids.len())
.unwrap_or(0)
}
/// Parse `completed`/`failed`/`aborted` counts from an `<agent_swarm_result>`
/// summary line, e.g. `<summary>completed: 2, failed: 1</summary>`. Omitted
/// buckets default to 0. Returns `None` when the text is not a swarm result.
fn parse_swarm_counts(text: &str) -> Option<(usize, usize, usize)> {
if !text.contains("<agent_swarm_result>") {
return None;
}
let bucket = |label: &str| -> usize {
let needle = format!("{label}: ");
text.find(&needle)
.map(|start| start + needle.len())
.map(|start| {
text[start..]
.chars()
.take_while(|c| c.is_ascii_digit())
.collect::<String>()
})
.and_then(|digits| digits.parse().ok())
.unwrap_or(0)
};
Some((bucket("completed"), bucket("failed"), bucket("aborted")))
}
fn effective_reasoning_for_model(cfg: &RuntimeConfig, model: &str) -> String {
let base_reasoning = default_effective_reasoning_for_model(cfg, model);
if cfg.dynamic_workflows.effort_profile == DynamicWorkflowEffortProfile::Ultracode {
return ultracode_reasoning_level_for_model(
model,
&cfg.speed_policy.ultracode_reasoning,
&base_reasoning,
);
}
base_reasoning
}
fn default_effective_reasoning_for_model(cfg: &RuntimeConfig, model: &str) -> String {
let Some(entry) = lookup_model(model) else {
return cfg
.reasoning
.clone()
.unwrap_or_else(|| REASONING_NONE.to_string());
};
if entry.supported_reasoning.is_empty() {
return REASONING_NONE.to_string();
}
cfg.reasoning
.as_deref()
.filter(|reasoning| {
entry
.supported_reasoning
.iter()
.any(|option| option.effort == *reasoning)
})
.map(str::to_string)
.or_else(|| {
model_profile_for_model(cfg, model)
.and_then(|profile| profile.reasoning.orientation)
.filter(|reasoning| {
entry
.supported_reasoning
.iter()
.any(|option| option.effort == reasoning)
})
})
.unwrap_or_else(|| entry.default_reasoning.to_string())
}
fn validate_reasoning_effort(model: &str, effort: &str) -> anyhow::Result<()> {
if effort == REASONING_NONE && !model_supports_reasoning(model, effort) {
return Ok(());
}
let Some(entry) = lookup_model(model) else {
return Ok(());
};
if entry
.supported_reasoning
.iter()
.any(|option| option.effort == effort)
{
Ok(())
} else {
anyhow::bail!("model {model} does not support reasoning effort {effort}")
}
}
fn validate_runtime_config_reasoning(cfg: &RuntimeConfig) -> anyhow::Result<()> {
let Some(reasoning) = cfg.reasoning.as_deref() else {
return Ok(());
};
let Some(entry) = lookup_model(&cfg.default_model) else {
return Ok(());
};
if entry.provider != PROVIDER_GEMINI {
return Ok(());
}
validate_reasoning_effort(&cfg.default_model, reasoning)
}
fn validate_runtime_inference_router_config(
registry: &ExtensionRegistry,
cfg: &RuntimeConfig,
) -> anyhow::Result<()> {
if !cfg.inference_router.enabled {
return Ok(());
}
let Some(router_id) = cfg.inference_router.router_id.as_deref() else {
anyhow::bail!("inference_router.enabled requires inference_router.router");
};
if registry.inference_router(router_id).is_some() {
return Ok(());
}
let available = registry
.inference_routers
.iter()
.map(|router| router.id())
.collect::<Vec<_>>()
.join(", ");
if available.is_empty() {
anyhow::bail!("inference router {router_id:?} is not registered");
}
anyhow::bail!(
"inference router {router_id:?} is not registered; available routers: {available}"
);
}
fn model_supports_reasoning(model: &str, effort: &str) -> bool {
lookup_model(model)
.map(|entry| {
entry
.supported_reasoning
.iter()
.any(|option| option.effort == effort)
})
.unwrap_or(false)
}
fn is_final_answer_phase(phase: Option<&str>) -> bool {
phase.is_none_or(|phase| phase.is_empty() || phase == FINAL_ANSWER_PHASE)
}
fn edit_tool_for_model<'a>(cfg: &'a RuntimeConfig, model: &'a str) -> Option<&'a str> {
cfg.model_edit_tools
.get(model)
.map(String::as_str)
.or_else(|| {
cfg.model_profiles
.get(model)
.and_then(|profile| profile.edit_tool.as_deref())
})
.or_else(|| lookup_model(model).and_then(|entry| entry.edit_tool))
.or(Some(EDIT_TOOL_EDIT))
}
fn model_profile_for_model(cfg: &RuntimeConfig, model: &str) -> Option<ModelHarnessProfile> {
cfg.model_profiles
.get(model)
.cloned()
.or_else(|| built_in_model_profile(model))
}
pub(crate) fn allowlist_permits(allowlist: &[String], tool_name: &str) -> bool {
allowlist.is_empty() || allowlist.iter().any(|allowed| allowed == tool_name)
}
/// Provider-aware profile resolution for the active turn.
///
/// Many model ids are shared across providers (for example Cursor proxies
/// `claude-opus-4-8`). Resolving the harness profile by id alone picks the
/// first catalog entry, which assigns cross-provider ids the wrong family and
/// instruction overlay (e.g. a `cursor/claude-opus-4-8` turn would otherwise
/// inherit the Anthropic overlay). Prefer the explicit `(provider, model)`
/// catalog entry, keeping user-configured profile overrides as the top
/// precedence.
fn model_profile_for_provider_model(
cfg: &RuntimeConfig,
provider: &str,
model: &str,
) -> Option<ModelHarnessProfile> {
cfg.model_profiles
.get(model)
.cloned()
.or_else(|| built_in_model_profile_for_provider(provider, model))
}
fn schema_policy_for_model(profile: Option<&ModelHarnessProfile>) -> ModelSchemaPolicy {
profile
.map(|profile| profile.schema_policy)
.unwrap_or_default()
}
fn model_profile_segment_metadata(
profile: Option<&ModelHarnessProfile>,
provider: &str,
model: &str,
segment: &str,
) -> serde_json::Value {
serde_json::json!({
"kind": MODEL_PROFILE_TRACE_KIND,
"segment": segment,
"provider": provider,
"model": model,
"profileModel": profile.map(|profile| profile.model.as_str()).unwrap_or(model),
"providerFamily": profile.map(|profile| profile.provider_family),
"editTool": profile.and_then(|profile| profile.edit_tool.as_deref()),
"schemaPolicy": profile.map(|profile| profile.schema_policy),
"instructionOverlay": profile.map(|profile| profile.instruction_overlay),
"parallelToolCalls": profile.and_then(|profile| profile.parallel_tool_calls),
"autoCompactTokenLimit": profile.and_then(|profile| profile.auto_compact_token_limit),
})
}
fn model_switch_summary(
transcript: &[TranscriptItem],
profile: Option<&ModelHarnessProfile>,
provider: &str,
model: &str,
tools: &[roder_api::tools::ToolSpec],
) -> Option<String> {
let previous = latest_model_profile_segment(transcript)?;
let previous_model = previous
.get("model")
.and_then(serde_json::Value::as_str)
.unwrap_or_default();
let previous_provider = previous
.get("provider")
.and_then(serde_json::Value::as_str)
.unwrap_or_default();
if previous_model == model && previous_provider == provider {
return None;
}
let previous_profile = previous
.get("profileModel")
.and_then(serde_json::Value::as_str)
.unwrap_or(previous_model);
let current_profile = profile
.map(|profile| profile.model.as_str())
.unwrap_or(model);
let previous_edit_tool = previous
.get("editTool")
.and_then(serde_json::Value::as_str)
.unwrap_or("none");
let current_edit_tool = profile
.and_then(|profile| profile.edit_tool.as_deref())
.unwrap_or("none");
let tool_names = tools
.iter()
.map(|tool| tool.name.as_str())
.take(12)
.collect::<Vec<_>>()
.join(", ");
Some(format!(
"{MODEL_SWITCH_SUMMARY_PREFIX} previous profile {previous_provider}/{previous_profile} used edit tool {previous_edit_tool}. Current profile {provider}/{current_profile} uses edit tool {current_edit_tool}. Available tools now: {}.",
if tool_names.is_empty() {
"none"
} else {
&tool_names
}
))
}
fn latest_model_profile_segment(transcript: &[TranscriptItem]) -> Option<&serde_json::Value> {
transcript.iter().rev().find_map(|item| {
let TranscriptItem::ProviderMetadata(value) = item else {
return None;
};
(value.get("kind").and_then(serde_json::Value::as_str) == Some(MODEL_PROFILE_TRACE_KIND))
.then_some(value)
})
}
pub fn validate_edit_tool(value: &str) -> anyhow::Result<()> {
match value.trim() {
EDIT_TOOL_PATCH | EDIT_TOOL_EDIT => Ok(()),
_ => anyhow::bail!(
"unsupported edit_tool {value:?}; allowed values: {EDIT_TOOL_PATCH}, {EDIT_TOOL_EDIT}"
),
}
}
fn should_persist_thread_event(thread_id: &str) -> bool {
!is_synthetic_event_thread_id(thread_id)
}
#[cfg(test)]
#[path = "runtime/codex_v2_tests.rs"]
mod codex_v2_tests;
#[cfg(test)]
#[path = "runtime/codex_v2_lifecycle_tests.rs"]
mod codex_v2_lifecycle_tests;
#[cfg(test)]
mod tests {
use super::*;
use futures::stream;
use roder_api::catalog::{
PROVIDER_MOCK, REASONING_HIGH, REASONING_LOW, REASONING_MEDIUM, REASONING_MINIMAL,
REASONING_NONE, REASONING_XHIGH,
};
use roder_api::extension::ExtensionRegistryBuilder;
use roder_api::inference::{
CompletionMetadata, InferenceCapabilities, InferenceEngine, InferenceEventStream,
InferenceProviderContext, InferenceTurnContext, MessageDelta, ModelDescriptor,
ModelInstructionOverlay, ModelProfileReasoning, ModelSchemaPolicy, ProviderFamily,
ReasoningEffortDescriptor,
};
use roder_api::inference_routing::{
InferenceRouter, InferenceRoutingContext, InferenceRoutingDecision, InferenceRoutingOutcome,
};
use roder_api::thread::ThreadStoreFactory;
use roder_api::tools::{ToolContributor, ToolExecutor, ToolSpec};
use roder_ext_jsonl_thread_store::store::JsonlThreadStoreFactory;
use std::sync::Mutex as StdMutex;
fn test_workspace() -> String {
std::env::current_dir().unwrap().display().to_string()
}
struct MetadataMissingStore;
#[async_trait::async_trait]
impl ThreadStore for MetadataMissingStore {
fn id(&self) -> roder_api::thread::ThreadStoreId {
"metadata-missing-store".to_string()
}
async fn create_thread(&self, metadata: ThreadMetadata) -> anyhow::Result<ThreadMetadata> {
Ok(metadata)
}
async fn list_threads(&self) -> anyhow::Result<Vec<ThreadMetadata>> {
Ok(Vec::new())
}
async fn load_thread(
&self,
_thread_id: &ThreadId,
) -> anyhow::Result<Option<ThreadSnapshot>> {
Ok(Some(ThreadSnapshot {
metadata: None,
..ThreadSnapshot::default()
}))
}
async fn append_event(
&self,
_thread_id: &ThreadId,
_envelope: &EventEnvelope,
) -> anyhow::Result<()> {
Ok(())
}
}
struct MetadataMissingStoreFactory;
impl ThreadStoreFactory for MetadataMissingStoreFactory {
fn id(&self) -> roder_api::thread::ThreadStoreId {
"metadata-missing-store".to_string()
}
fn create(&self) -> Arc<dyn ThreadStore> {
Arc::new(MetadataMissingStore)
}
}
#[test]
fn synthetic_app_server_events_are_not_thread_events() {
for thread_id in ["app-server", "runtime", "thread-workflow"] {
assert!(!should_persist_thread_event(thread_id));
}
assert!(should_persist_thread_event("thread-discovery"));
assert!(should_persist_thread_event("thread-plan"));
assert!(should_persist_thread_event("thread-process"));
assert!(should_persist_thread_event("thread-1"));
}
#[test]
fn server_side_compaction_uses_catalog_ninety_percent_default() {
assert_eq!(
server_side_compaction_threshold(&RuntimeConfig::default(), "gpt-5.5"),
Some(945_000)
);
assert_eq!(
server_side_compaction_threshold(&RuntimeConfig::default(), "gpt-5.3-codex-spark"),
Some(115_200)
);
}
#[test]
fn server_side_compaction_respects_explicit_config_override() {
let cfg = RuntimeConfig {
auto_compact_token_limit: Some(123_456),
..RuntimeConfig::default()
};
assert_eq!(
server_side_compaction_threshold(&cfg, "gpt-5.5"),
Some(123_456)
);
}
#[tokio::test]
async fn pre_request_compaction_runs_when_server_side_model_is_at_context_window() {
let captured = Arc::new(StdMutex::new(None));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(CapturingEngine {
request: captured.clone(),
}));
let thread_root = std::env::temp_dir().join(format!(
"roder-pre-request-compaction-{}",
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 {
default_provider: PROVIDER_MOCK.to_string(),
default_model: "gpt-5.5".to_string(),
file_backed_dynamic_context: true,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let thread_id = runtime
.create_thread(Some("Pre-request compaction".to_string()))
.await
.unwrap()
.thread_id;
let old_turn = "old-turn".to_string();
runtime
.persist_turn_item(
&thread_id,
&old_turn,
&TranscriptItem::UserMessage(UserMessage::text("old context ".repeat(4_300_000))),
)
.await
.unwrap();
let mut events = runtime.subscribe_events();
runtime
.start_turn(StartTurnRequest {
thread_id: thread_id.clone(),
message: "continue".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle::default(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
loop {
let envelope = tokio::time::timeout(std::time::Duration::from_secs(5), events.recv())
.await
.unwrap()
.unwrap();
if envelope.thread_id.as_deref() == Some(&thread_id)
&& matches!(envelope.event, RoderEvent::TurnCompleted(_))
{
break;
}
}
let request = captured.lock().unwrap().clone().unwrap();
assert!(
matches!(
request.transcript.first(),
Some(TranscriptItem::ContextCompaction(_))
),
"provider request should start with a local emergency compaction item"
);
assert!(
request.transcript.len() < 4,
"provider request should not replay the full oversized prior transcript: {:?}",
request.transcript
);
let _ = std::fs::remove_dir_all(thread_root);
}
#[tokio::test]
async fn continue_after_context_window_failure_compacts_before_provider_request() {
let captured = Arc::new(StdMutex::new(None));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(CapturingEngine {
request: captured.clone(),
}));
let thread_root = std::env::temp_dir().join(format!(
"roder-context-failure-continue-{}",
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 {
default_provider: PROVIDER_MOCK.to_string(),
default_model: "gpt-5.5".to_string(),
file_backed_dynamic_context: true,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let thread_id = runtime
.create_thread(Some("Context failure continue".to_string()))
.await
.unwrap()
.thread_id;
let failed_turn = "failed-turn".to_string();
runtime
.persist_turn_item(
&thread_id,
&failed_turn,
&TranscriptItem::UserMessage(UserMessage::text("old work ".repeat(10_000))),
)
.await
.unwrap();
runtime
.persist_turn_item(
&thread_id,
&failed_turn,
&TranscriptItem::Error(ErrorRecord {
message: "Your input exceeds the context window of this model. Please adjust your input and try again."
.to_string(),
}),
)
.await
.unwrap();
let mut events = runtime.subscribe_events();
runtime
.start_turn(StartTurnRequest {
thread_id: thread_id.clone(),
message: "continue".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle::default(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
loop {
let envelope = tokio::time::timeout(std::time::Duration::from_secs(5), events.recv())
.await
.unwrap()
.unwrap();
if envelope.thread_id.as_deref() == Some(&thread_id)
&& matches!(envelope.event, RoderEvent::TurnCompleted(_))
{
break;
}
}
let request = captured.lock().unwrap().clone().unwrap();
assert!(
matches!(
request.transcript.first(),
Some(TranscriptItem::ContextCompaction(_))
),
"provider request after context-window failure should start with local compaction"
);
assert!(
request
.transcript
.iter()
.any(|item| matches!(item, TranscriptItem::UserMessage(message) if message.text == "continue")),
"current continue prompt must be preserved: {:?}",
request.transcript
);
assert!(
!request.transcript.iter().any(
|item| matches!(item, TranscriptItem::Error(error) if error.message.contains("context window"))
),
"raw prior context-window error should be summarized, not replayed: {:?}",
request.transcript
);
let _ = std::fs::remove_dir_all(thread_root);
}
#[tokio::test]
async fn workspace_for_thread_falls_back_when_metadata_is_missing() {
let workspace = test_workspace();
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(FakeInferenceEngine));
builder.thread_store_factory(Arc::new(MetadataMissingStoreFactory));
let runtime = Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
workspace: Some(workspace.clone()),
..RuntimeConfig::default()
},
)
.unwrap();
let resolved = runtime
.workspace_for_thread(&ThreadId::from("thread-workflow"))
.await
.unwrap();
assert_eq!(resolved, workspace);
}
#[tokio::test]
async fn automations_can_create_project_thread_with_model_overrides() {
let runtime = Runtime::fake().unwrap();
let workspace = std::env::temp_dir().join("project");
let metadata = runtime
.create_thread_with(CreateThreadRequest {
title: Some("Automation: nightly status".to_string()),
workspace: workspace.display().to_string(),
workspace_id: None,
root_id: None,
provider: Some("mock".to_string()),
model: Some("mock".to_string()),
selection_mode: None,
tool_allowlist: Vec::new(),
developer_instructions: None,
external_tools: Vec::new(),
runner: None,
})
.await
.unwrap();
assert_eq!(
metadata.title.as_deref(),
Some("Automation: nightly status")
);
assert_eq!(metadata.workspace, workspace.display().to_string());
assert_eq!(metadata.provider.as_deref(), Some("mock"));
assert_eq!(metadata.model.as_deref(), Some("mock"));
}
#[tokio::test]
async fn prior_provider_compaction_boundary_is_used_on_next_turn() {
let captured = Arc::new(StdMutex::new(None));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(CapturingEngine {
request: captured.clone(),
}));
let thread_root = std::env::temp_dir().join(format!(
"roder-provider-compaction-boundary-{}",
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 {
default_provider: PROVIDER_MOCK.to_string(),
default_model: "gpt-5.5".to_string(),
..RuntimeConfig::default()
},
)
.unwrap(),
);
let thread_id = runtime
.create_thread(Some("Provider compaction boundary".to_string()))
.await
.unwrap()
.thread_id;
let old_turn = "old-turn".to_string();
runtime
.persist_turn_item(
&thread_id,
&old_turn,
&TranscriptItem::UserMessage(UserMessage::text(
"old history that must not be replayed after provider compaction",
)),
)
.await
.unwrap();
runtime
.persist_turn_item(
&thread_id,
&old_turn,
&TranscriptItem::AssistantMessage(AssistantMessage {
text: "old answer".to_string(),
phase: None,
}),
)
.await
.unwrap();
runtime
.persist_turn_item(
&thread_id,
&old_turn,
&TranscriptItem::ProviderMetadata(serde_json::json!({
"output": [{
"id": "cmp_1",
"type": "compaction",
"encrypted_content": "opaque-state"
}]
})),
)
.await
.unwrap();
runtime
.persist_turn_item(
&thread_id,
&old_turn,
&TranscriptItem::AssistantMessage(AssistantMessage {
text: "after compact".to_string(),
phase: None,
}),
)
.await
.unwrap();
let mut events = runtime.subscribe_events();
runtime
.start_turn(StartTurnRequest {
thread_id: thread_id.clone(),
message: "continue".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle::default(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
loop {
let envelope = tokio::time::timeout(std::time::Duration::from_secs(5), events.recv())
.await
.unwrap()
.unwrap();
if envelope.thread_id.as_deref() == Some(&thread_id)
&& matches!(envelope.event, RoderEvent::TurnCompleted(_))
{
break;
}
}
let request = captured.lock().unwrap().clone().unwrap();
let compaction_idx = request.transcript.iter().position(|item| {
matches!(
item,
TranscriptItem::ProviderMetadata(metadata)
if crate::compaction::provider_metadata_has_compaction(metadata)
)
});
assert!(
compaction_idx.is_some(),
"next provider request should include the provider compaction item: {:?}",
request.transcript
);
// Skill/context injectors may prepend non-history items, but no prior-turn
// conversation may appear before the latest provider compaction boundary.
let history_before_boundary =
request
.transcript
.iter()
.take(compaction_idx.unwrap())
.any(|item| match item {
TranscriptItem::AssistantMessage(_)
| TranscriptItem::ToolCall(_)
| TranscriptItem::ToolResult(_) => true,
TranscriptItem::UserMessage(message) => {
!message.text.contains("<skills>") && message.text != "continue"
}
_ => false,
});
assert!(
!history_before_boundary,
"pre-provider-compaction conversation must not be replayed: {:?}",
request.transcript
);
assert!(
!request.transcript.iter().any(|item| {
matches!(
item,
TranscriptItem::UserMessage(message)
if message.text.contains("old history that must not be replayed")
)
}),
"pre-provider-compaction history must not be replayed: {:?}",
request.transcript
);
assert!(
request.transcript.iter().any(|item| {
matches!(
item,
TranscriptItem::UserMessage(message) if message.text == "continue"
)
}),
"current continue prompt must be preserved: {:?}",
request.transcript
);
assert_eq!(
request.runtime.auto_compact_token_limit,
Some(945_000),
"gpt-5.5 should keep server-side auto compaction configured"
);
let _ = std::fs::remove_dir_all(thread_root);
}
#[test]
fn server_side_compaction_is_only_enabled_for_supported_models() {
let cfg = RuntimeConfig {
auto_compact_token_limit: Some(123_456),
..RuntimeConfig::default()
};
assert_eq!(server_side_compaction_threshold(&cfg, "mock"), None);
assert_eq!(
server_side_compaction_threshold(&cfg, "codex-auto-review"),
None
);
}
#[test]
fn reasoning_is_disabled_for_models_without_reasoning_support() {
let cfg = RuntimeConfig {
reasoning: Some(REASONING_HIGH.to_string()),
..RuntimeConfig::default()
};
assert_eq!(
effective_reasoning_for_model(&cfg, "claude-haiku-4-5-20251001"),
REASONING_NONE
);
assert_eq!(
reasoning_for_model(&cfg, "claude-haiku-4-5-20251001"),
ReasoningConfig::default()
);
}
#[test]
fn unsupported_configured_reasoning_falls_back_to_model_default() {
let cfg = RuntimeConfig {
reasoning: Some(REASONING_MINIMAL.to_string()),
..RuntimeConfig::default()
};
assert_eq!(
effective_reasoning_for_model(&cfg, "gpt-5.5"),
REASONING_MEDIUM
);
}
#[test]
fn unsupported_configured_gemini_reasoning_is_rejected() {
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(std::sync::Arc::new(FakeInferenceEngine));
let err = match Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_model: "gemini-3.5-flash".to_string(),
reasoning: Some(REASONING_XHIGH.to_string()),
..RuntimeConfig::default()
},
) {
Ok(_) => panic!("expected unsupported Gemini reasoning to be rejected"),
Err(err) => err,
};
assert!(
err.to_string()
.contains("model gemini-3.5-flash does not support reasoning effort xhigh")
);
}
#[tokio::test]
async fn selecting_none_for_non_reasoning_model_preserves_stored_preference() {
let runtime = Runtime::new(
Runtime::fake().unwrap().registry,
RuntimeConfig {
reasoning: Some(REASONING_HIGH.to_string()),
..RuntimeConfig::default()
},
)
.unwrap();
let cfg = runtime
.select_provider(
roder_api::catalog::PROVIDER_MOCK.to_string(),
Some("claude-haiku-4-5-20251001".to_string()),
Some(REASONING_NONE.to_string()),
)
.await
.unwrap();
assert_eq!(cfg.reasoning.as_deref(), Some(REASONING_HIGH));
assert_eq!(runtime.effective_reasoning().await, REASONING_NONE);
}
#[tokio::test]
async fn selecting_none_for_model_that_supports_none_updates_preference() {
let runtime = Runtime::new(
Runtime::fake().unwrap().registry,
RuntimeConfig {
reasoning: Some(REASONING_HIGH.to_string()),
..RuntimeConfig::default()
},
)
.unwrap();
let cfg = runtime
.select_provider(
roder_api::catalog::PROVIDER_MOCK.to_string(),
Some("mock".to_string()),
Some(REASONING_NONE.to_string()),
)
.await
.unwrap();
assert_eq!(cfg.reasoning.as_deref(), Some(REASONING_NONE));
}
#[test]
fn parallel_tool_calls_default_on_with_model_override() {
assert!(parallel_tool_calls_for_model(
&RuntimeConfig::default(),
"custom-model"
));
let cfg = RuntimeConfig {
model_parallel_tool_calls: std::collections::HashMap::from([(
"custom-model".to_string(),
false,
)]),
..RuntimeConfig::default()
};
assert!(!parallel_tool_calls_for_model(&cfg, "custom-model"));
assert!(parallel_tool_calls_for_model(&cfg, "other-model"));
}
#[test]
fn profile_parallel_tool_calls_applies_between_config_and_default() {
let cfg = RuntimeConfig {
model_profiles: std::collections::HashMap::from([(
"gpt-5.5".to_string(),
test_model_profile("gpt-5.5"),
)]),
..RuntimeConfig::default()
};
assert!(!parallel_tool_calls_for_model(&cfg, "gpt-5.5"));
let cfg = RuntimeConfig {
model_parallel_tool_calls: std::collections::HashMap::from([(
"gpt-5.5".to_string(),
true,
)]),
..cfg
};
assert!(parallel_tool_calls_for_model(&cfg, "gpt-5.5"));
}
struct CapturingEngine {
request: Arc<StdMutex<Option<AgentInferenceRequest>>>,
}
#[async_trait::async_trait]
impl InferenceEngine for CapturingEngine {
fn id(&self) -> String {
roder_api::catalog::PROVIDER_MOCK.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<roder_api::inference::ModelDescriptor>> {
Ok(roder_api::catalog::models_for_provider(
roder_api::catalog::PROVIDER_MOCK,
true,
))
}
async fn stream_turn(
&self,
_ctx: InferenceTurnContext<'_>,
request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
*self.request.lock().unwrap() = Some(request);
Ok(Box::pin(stream::iter(vec![
Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "done".to_string(),
phase: None,
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("stop".to_string()),
provider_response_id: None,
})),
])))
}
}
struct RoutingCaptureEngine {
id: &'static str,
models: Vec<ModelDescriptor>,
requests: Arc<StdMutex<Vec<AgentInferenceRequest>>>,
}
#[async_trait::async_trait]
impl InferenceEngine for RoutingCaptureEngine {
fn id(&self) -> String {
self.id.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<ModelDescriptor>> {
Ok(self.models.clone())
}
async fn stream_turn(
&self,
_ctx: InferenceTurnContext<'_>,
request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
self.requests.lock().unwrap().push(request);
Ok(Box::pin(stream::iter(vec![
Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "routed".to_string(),
phase: None,
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("stop".to_string()),
provider_response_id: None,
})),
])))
}
}
struct StaticRouter {
id: &'static str,
decision: InferenceRoutingDecision,
contexts: Arc<StdMutex<Vec<InferenceRoutingContext>>>,
}
#[async_trait::async_trait]
impl InferenceRouter for StaticRouter {
fn id(&self) -> String {
self.id.to_string()
}
async fn route(
&self,
context: InferenceRoutingContext,
) -> anyhow::Result<InferenceRoutingDecision> {
self.contexts.lock().unwrap().push(context);
Ok(self.decision.clone())
}
}
fn routing_test_model(id: &str, supported_reasoning: &[&str]) -> ModelDescriptor {
ModelDescriptor {
id: id.to_string(),
name: id.to_string(),
context_window: Some(128_000),
default_reasoning: supported_reasoning
.first()
.map(|effort| (*effort).to_string()),
supported_reasoning: supported_reasoning
.iter()
.map(|effort| ReasoningEffortDescriptor {
effort: (*effort).to_string(),
description: format!("{effort} reasoning"),
})
.collect(),
}
}
struct TaskLedgerCompletionGateEngine {
calls: StdMutex<u32>,
requests: Arc<StdMutex<Vec<AgentInferenceRequest>>>,
}
#[async_trait::async_trait]
impl InferenceEngine for TaskLedgerCompletionGateEngine {
fn id(&self) -> String {
roder_api::catalog::PROVIDER_MOCK.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<roder_api::inference::ModelDescriptor>> {
Ok(roder_api::catalog::models_for_provider(
roder_api::catalog::PROVIDER_MOCK,
true,
))
}
async fn stream_turn(
&self,
_ctx: InferenceTurnContext<'_>,
request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
self.requests.lock().unwrap().push(request);
let mut calls = self.calls.lock().unwrap();
*calls += 1;
let events = match *calls {
1 => vec![Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "ledger-open".to_string(),
name: TASK_LEDGER_TOOL_NAME.to_string(),
arguments: serde_json::json!({
"tasks": [
{
"id": "inspect",
"content": "Inspect local assets",
"status": "completed",
"evidence": "listed workspace"
},
{
"id": "write",
"content": "Write /app/result.txt",
"status": "pending"
}
],
"requireCompletionEvidence": true
})
.to_string(),
}))],
3 => vec![Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "ledger-complete".to_string(),
name: TASK_LEDGER_TOOL_NAME.to_string(),
arguments: serde_json::json!({
"tasks": [
{
"id": "inspect",
"content": "Inspect local assets",
"status": "completed",
"evidence": "listed workspace"
},
{
"id": "write",
"content": "Write /app/result.txt",
"status": "completed",
"evidence": "wrote answer"
}
],
"requireCompletionEvidence": true
})
.to_string(),
}))],
_ => vec![Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "final".to_string(),
phase: None,
}))],
};
Ok(Box::pin(stream::iter(events.into_iter().chain(
std::iter::once(Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("stop".to_string()),
provider_response_id: None,
}))),
))))
}
}
struct VerificationGateEngine {
calls: StdMutex<u32>,
}
#[async_trait::async_trait]
impl InferenceEngine for VerificationGateEngine {
fn id(&self) -> String {
roder_api::catalog::PROVIDER_MOCK.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<roder_api::inference::ModelDescriptor>> {
Ok(roder_api::catalog::models_for_provider(
roder_api::catalog::PROVIDER_MOCK,
true,
))
}
async fn stream_turn(
&self,
_ctx: InferenceTurnContext<'_>,
request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
let mut calls = self.calls.lock().unwrap();
*calls += 1;
let events = match *calls {
1 => vec![Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "write-1".to_string(),
name: "write_file".to_string(),
arguments: serde_json::json!({
"path": "src/lib.rs",
"content": "pub fn answer() -> u8 { 42 }\n"
})
.to_string(),
}))],
2 => vec![Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "done too early".to_string(),
phase: None,
}))],
3 if request.transcript.iter().any(|item| {
matches!(
item,
TranscriptItem::UserMessage(message)
if message.text.contains("Verification gate blocked final completion")
)
}) =>
{
vec![Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "verify-1".to_string(),
name: crate::verification_gate::VERIFICATION_TOOL_NAME.to_string(),
arguments: serde_json::json!({
"originalTask": "write code",
"changedFiles": ["src/lib.rs"],
"toolEvidence": ["write_file wrote src/lib.rs"],
"testsRun": ["cargo test -p roder-core verification_gate"],
"openGaps": [],
"status": "completed"
})
.to_string(),
}))]
}
_ => vec![Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "verified final".to_string(),
phase: None,
}))],
};
Ok(Box::pin(stream::iter(events.into_iter().chain(
std::iter::once(Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("stop".to_string()),
provider_response_id: None,
}))),
))))
}
}
struct SpeedPolicyEngine {
calls: StdMutex<u32>,
requests: Arc<StdMutex<Vec<AgentInferenceRequest>>>,
}
#[async_trait::async_trait]
impl InferenceEngine for SpeedPolicyEngine {
fn id(&self) -> String {
roder_api::catalog::PROVIDER_MOCK.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<roder_api::inference::ModelDescriptor>> {
Ok(roder_api::catalog::models_for_provider(
roder_api::catalog::PROVIDER_MOCK,
true,
))
}
async fn stream_turn(
&self,
_ctx: InferenceTurnContext<'_>,
request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
self.requests.lock().unwrap().push(request.clone());
let mut calls = self.calls.lock().unwrap();
*calls += 1;
let events = match *calls {
1 => vec![Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "write-1".to_string(),
name: "write_file".to_string(),
arguments: serde_json::json!({
"path": "src/lib.rs",
"content": "pub fn answer() -> u8 { 42 }\n"
})
.to_string(),
}))],
3 if request.transcript.iter().any(|item| {
matches!(
item,
TranscriptItem::UserMessage(message)
if message.text.contains("Verification gate blocked final completion")
)
}) =>
{
vec![Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "verify-1".to_string(),
name: crate::verification_gate::VERIFICATION_TOOL_NAME.to_string(),
arguments: serde_json::json!({
"originalTask": "write code",
"changedFiles": ["src/lib.rs"],
"toolEvidence": ["write_file wrote src/lib.rs"],
"testsRun": ["cargo test -p roder-core speed_policy"],
"openGaps": [],
"status": "completed"
})
.to_string(),
}))]
}
_ => vec![Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "done".to_string(),
phase: None,
}))],
};
Ok(Box::pin(stream::iter(events.into_iter().chain(
std::iter::once(Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("stop".to_string()),
provider_response_id: None,
}))),
))))
}
}
struct SwitchCaptureEngine {
requests: Arc<StdMutex<Vec<AgentInferenceRequest>>>,
}
#[async_trait::async_trait]
impl InferenceEngine for SwitchCaptureEngine {
fn id(&self) -> String {
roder_api::catalog::PROVIDER_MOCK.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<roder_api::inference::ModelDescriptor>> {
Ok(roder_api::catalog::models_for_provider(
roder_api::catalog::PROVIDER_MOCK,
true,
))
}
async fn stream_turn(
&self,
_ctx: InferenceTurnContext<'_>,
request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
self.requests.lock().unwrap().push(request);
Ok(Box::pin(stream::iter(vec![
Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "done".to_string(),
phase: None,
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("stop".to_string()),
provider_response_id: None,
})),
])))
}
}
struct DeadlineEngine;
#[async_trait::async_trait]
impl InferenceEngine for DeadlineEngine {
fn id(&self) -> String {
roder_api::catalog::PROVIDER_MOCK.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<roder_api::inference::ModelDescriptor>> {
Ok(Vec::new())
}
async fn stream_turn(
&self,
_ctx: InferenceTurnContext<'_>,
_request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
Ok(Box::pin(stream::once(async {
tokio::time::sleep(std::time::Duration::from_secs(60)).await;
Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "too late".to_string(),
phase: None,
}))
})))
}
}
struct WriteFileContributor;
impl ToolContributor for WriteFileContributor {
fn id(&self) -> String {
"test-write".to_string()
}
fn contribute(&self, registry: &mut ToolRegistry) -> anyhow::Result<()> {
registry.register(Arc::new(WriteFileTool))
}
}
struct WriteFileTool;
#[async_trait::async_trait]
impl ToolExecutor for WriteFileTool {
fn spec(&self) -> ToolSpec {
ToolSpec {
name: "write_file".to_string(),
description: "Write a test file.".to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"path": { "type": "string" },
"content": { "type": "string" }
},
"required": ["path", "content"],
"additionalProperties": false
}),
}
}
async fn execute(
&self,
_ctx: ToolExecutionContext,
call: ToolCall,
) -> anyhow::Result<ToolResult> {
let path = call
.arguments
.get("path")
.and_then(serde_json::Value::as_str)
.unwrap_or("src/lib.rs");
Ok(ToolResult {
id: call.id,
name: call.name,
text: format!("wrote {path}"),
data: serde_json::json!({ "path": path }),
is_error: false,
})
}
}
struct ProfileToolContributor;
impl ToolContributor for ProfileToolContributor {
fn id(&self) -> String {
"profile-tools".to_string()
}
fn contribute(&self, registry: &mut ToolRegistry) -> anyhow::Result<()> {
for name in ["apply_patch", "edit", "multi_edit", "write_file"] {
registry.register(Arc::new(ProfileTool {
name: name.to_string(),
}))?;
}
Ok(())
}
}
struct ProfileTool {
name: String,
}
#[async_trait::async_trait]
impl ToolExecutor for ProfileTool {
fn spec(&self) -> ToolSpec {
ToolSpec {
name: self.name.clone(),
description: format!("{} test tool", self.name),
parameters: serde_json::json!({
"type": "object",
"properties": {
"path": { "type": "string" },
"content": { "type": "string" }
},
"required": ["path", "content"],
"additionalProperties": false
}),
}
}
async fn execute(
&self,
_ctx: ToolExecutionContext,
call: ToolCall,
) -> anyhow::Result<ToolResult> {
Ok(ToolResult {
id: call.id,
name: call.name,
text: "ok".to_string(),
data: serde_json::json!({}),
is_error: false,
})
}
}
fn test_model_profile(model: &str) -> ModelHarnessProfile {
ModelHarnessProfile {
model: model.to_string(),
provider: roder_api::catalog::PROVIDER_OPENAI.to_string(),
provider_family: ProviderFamily::OpenAi,
edit_tool: Some(EDIT_TOOL_EDIT.to_string()),
schema_policy: ModelSchemaPolicy::StandardRequiredFirst,
instruction_overlay: ModelInstructionOverlay::IntuitiveContext,
reasoning: ModelProfileReasoning {
orientation: Some(REASONING_LOW.to_string()),
execution: Some(REASONING_LOW.to_string()),
verification: Some(REASONING_LOW.to_string()),
recovery: Some(REASONING_LOW.to_string()),
},
parallel_tool_calls: Some(false),
auto_compact_token_limit: Some(123_000),
}
}
async fn captured_profile_request(cfg: RuntimeConfig) -> AgentInferenceRequest {
let captured = Arc::new(StdMutex::new(None));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(CapturingEngine {
request: captured.clone(),
}));
builder.tool_contributor(Arc::new(ProfileToolContributor));
let runtime = Arc::new(Runtime::new(builder.build().unwrap(), cfg).unwrap());
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: "thread-model-profile".to_string(),
message: "use profile knobs".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle {
system: None,
developer: Some("base developer".to_string()),
developer_context: None,
},
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::TurnCompleted(_) => break,
RoderEvent::TurnFailed(event) => panic!("turn failed: {}", event.error),
_ => {}
}
}
})
.await
.unwrap();
captured.lock().unwrap().clone().unwrap()
}
struct ToolThenStopEngine {
calls: StdMutex<u32>,
}
#[async_trait::async_trait]
impl InferenceEngine for ToolThenStopEngine {
fn id(&self) -> String {
roder_api::catalog::PROVIDER_MOCK.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<roder_api::inference::ModelDescriptor>> {
Ok(roder_api::catalog::models_for_provider(
roder_api::catalog::PROVIDER_MOCK,
true,
))
}
async fn stream_turn(
&self,
_ctx: InferenceTurnContext<'_>,
_request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
let mut calls = self.calls.lock().unwrap();
*calls += 1;
let events = match *calls {
1 => vec![
Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
id: "write-1".to_string(),
name: "write_file".to_string(),
arguments: serde_json::json!({
"path": "src/lib.rs",
"content": "pub fn answer() -> u8 { 42 }\n"
})
.to_string(),
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("tool_use".to_string()),
provider_response_id: None,
})),
],
_ => vec![
Ok(InferenceEvent::MessageDelta(MessageDelta {
text: "final".to_string(),
phase: None,
})),
Ok(InferenceEvent::Completed(CompletionMetadata {
stop_reason: Some("end_turn".to_string()),
provider_response_id: None,
})),
],
};
Ok(Box::pin(stream::iter(events)))
}
}
#[tokio::test]
async fn turn_completed_reports_terminal_step_finish_reason() {
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(ToolThenStopEngine {
calls: StdMutex::new(0),
}));
builder.tool_contributor(Arc::new(WriteFileContributor));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
policy_mode: PolicyMode::Bypass,
agent_swarm_mode: false,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: "thread-finish-reason".to_string(),
message: "write then finish".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle {
system: None,
developer: None,
developer_context: None,
},
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
let completed = tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::TurnCompleted(event) => break event,
RoderEvent::TurnFailed(event) => panic!("turn failed: {}", event.error),
_ => {}
}
}
})
.await
.unwrap();
// The mid-turn tool_use stop reason must not leak; the terminal
// end_turn step decides the turn's finish reason.
assert_eq!(completed.finish_reason.as_deref(), Some("stop"));
}
#[tokio::test]
async fn inference_router_selection_changes_request_model_and_records_event() {
let default_requests = Arc::new(StdMutex::new(Vec::<AgentInferenceRequest>::new()));
let routed_requests = Arc::new(StdMutex::new(Vec::<AgentInferenceRequest>::new()));
let contexts = Arc::new(StdMutex::new(Vec::<InferenceRoutingContext>::new()));
let selected = ModelSelection {
provider: "routed-provider".to_string(),
model: "routed-model".to_string(),
};
let default = ModelSelection {
provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
model: "mock".to_string(),
};
let decision = InferenceRoutingDecision {
reasoning: Some(ReasoningConfig {
enabled: true,
level: Some(REASONING_LOW.to_string()),
}),
confidence: Some(0.91),
baseline: Some(default.clone()),
matched_signals: vec![roder_api::inference_routing::InferenceRoutingSignal::new(
"intent", "routine",
)],
..InferenceRoutingDecision::selected("test-router", selected.clone(), "routine request")
};
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(RoutingCaptureEngine {
id: roder_api::catalog::PROVIDER_MOCK,
models: vec![routing_test_model("mock", &[REASONING_LOW])],
requests: default_requests.clone(),
}));
builder.inference_engine(Arc::new(RoutingCaptureEngine {
id: "routed-provider",
models: vec![routing_test_model(
"routed-model",
&[REASONING_LOW, REASONING_MEDIUM],
)],
requests: routed_requests.clone(),
}));
builder.inference_router(Arc::new(StaticRouter {
id: "test-router",
decision,
contexts: contexts.clone(),
}));
let thread_root =
std::env::temp_dir().join(format!("roder-routing-auto-{}", 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 {
default_provider: default.provider.clone(),
default_model: default.model.clone(),
..RuntimeConfig::default()
},
)
.unwrap(),
);
let thread_id = runtime
.create_thread_with(CreateThreadRequest {
title: Some("Routing auto".to_string()),
workspace: test_workspace(),
workspace_id: None,
root_id: None,
provider: Some(default.provider.clone()),
model: Some(default.model.clone()),
tool_allowlist: Vec::new(),
developer_instructions: None,
external_tools: Vec::new(),
selection_mode: Some(ModelSelectionMode::auto(
"test-router:coding",
"test-router",
"Auto: Coding",
default.clone(),
Some("coding".to_string()),
None,
)),
runner: None,
})
.await
.unwrap()
.thread_id;
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: thread_id.clone(),
message: "small cleanup".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle::default(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
let mut routing_event = None;
let mut inference_started = None;
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::InferenceRoutingDecision(event) => {
routing_event = Some(event);
}
RoderEvent::InferenceStarted(event) => {
inference_started = Some(event);
}
RoderEvent::TurnCompleted(_) => break,
RoderEvent::TurnFailed(event) => panic!("turn failed: {}", event.error),
_ => {}
}
}
})
.await
.unwrap();
assert!(default_requests.lock().unwrap().is_empty());
let routed_requests = routed_requests.lock().unwrap();
assert_eq!(routed_requests.len(), 1);
assert_eq!(routed_requests[0].model, selected);
assert_eq!(
routed_requests[0].reasoning.level.as_deref(),
Some(REASONING_LOW)
);
assert_eq!(
routed_requests[0].metadata["inferenceRouting"]["outcome"],
"selected"
);
let routing_event = routing_event.expect("routing decision event");
assert_eq!(routing_event.default_selection, default);
assert_eq!(routing_event.selected_selection, selected);
assert_eq!(
routing_event.decision.outcome,
InferenceRoutingOutcome::Selected
);
assert_eq!(
inference_started.expect("inference started event").model,
selected
);
let contexts = contexts.lock().unwrap();
assert_eq!(contexts.len(), 1);
assert_eq!(contexts[0].default_selection, default);
assert_eq!(contexts[0].candidates.len(), 2);
assert!(
contexts[0]
.signals
.iter()
.any(|signal| signal.key == "profile" && signal.value == "coding")
);
let _ = std::fs::remove_dir_all(thread_root);
}
#[tokio::test]
async fn inference_router_is_bypassed_for_explicit_selection() {
let requests = Arc::new(StdMutex::new(Vec::<AgentInferenceRequest>::new()));
let contexts = Arc::new(StdMutex::new(Vec::<InferenceRoutingContext>::new()));
let selected = ModelSelection {
provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
model: "mock".to_string(),
};
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(RoutingCaptureEngine {
id: roder_api::catalog::PROVIDER_MOCK,
models: vec![routing_test_model("mock", &[REASONING_LOW])],
requests: requests.clone(),
}));
builder.inference_router(Arc::new(StaticRouter {
id: "test-router",
decision: InferenceRoutingDecision::selected(
"test-router",
ModelSelection {
provider: "missing".to_string(),
model: "missing".to_string(),
},
"would route if called",
),
contexts: contexts.clone(),
}));
let thread_root =
std::env::temp_dir().join(format!("roder-routing-explicit-{}", 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 {
default_provider: selected.provider.clone(),
default_model: selected.model.clone(),
inference_router: RuntimeInferenceRouterConfig {
enabled: true,
router_id: Some("test-router".to_string()),
},
..RuntimeConfig::default()
},
)
.unwrap(),
);
let thread_id = runtime
.create_thread_with(CreateThreadRequest {
title: Some("Routing explicit".to_string()),
workspace: test_workspace(),
workspace_id: None,
root_id: None,
provider: Some(selected.provider.clone()),
model: Some(selected.model.clone()),
tool_allowlist: Vec::new(),
developer_instructions: None,
external_tools: Vec::new(),
selection_mode: Some(ModelSelectionMode::auto(
"test-router:default",
"test-router",
"Auto",
selected.clone(),
None,
None,
)),
runner: None,
})
.await
.unwrap()
.thread_id;
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id,
message: "use explicit selection".to_string(),
images: Vec::new(),
provider_override: Some(selected.provider.clone()),
model_override: Some(selected.model.clone()),
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle::default(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
let mut saw_routing_event = false;
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::InferenceRoutingDecision(_) => {
saw_routing_event = true;
}
RoderEvent::TurnCompleted(_) => break,
RoderEvent::TurnFailed(event) => panic!("turn failed: {}", event.error),
_ => {}
}
}
})
.await
.unwrap();
assert!(!saw_routing_event);
assert!(contexts.lock().unwrap().is_empty());
let requests = requests.lock().unwrap();
assert_eq!(requests.len(), 1);
assert_eq!(requests[0].model, selected);
let _ = std::fs::remove_dir_all(thread_root);
}
#[tokio::test]
async fn inference_router_is_bypassed_for_manual_selection_mode() {
let requests = Arc::new(StdMutex::new(Vec::<AgentInferenceRequest>::new()));
let contexts = Arc::new(StdMutex::new(Vec::<InferenceRoutingContext>::new()));
let selected = ModelSelection {
provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
model: "mock".to_string(),
};
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(RoutingCaptureEngine {
id: roder_api::catalog::PROVIDER_MOCK,
models: vec![routing_test_model("mock", &[REASONING_LOW])],
requests: requests.clone(),
}));
builder.inference_router(Arc::new(StaticRouter {
id: "test-router",
decision: InferenceRoutingDecision::selected(
"test-router",
ModelSelection {
provider: "missing".to_string(),
model: "missing".to_string(),
},
"would route if called",
),
contexts: contexts.clone(),
}));
let thread_root =
std::env::temp_dir().join(format!("roder-routing-manual-{}", 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 {
default_provider: selected.provider.clone(),
default_model: selected.model.clone(),
inference_router: RuntimeInferenceRouterConfig {
enabled: true,
router_id: Some("test-router".to_string()),
},
..RuntimeConfig::default()
},
)
.unwrap(),
);
let thread_id = runtime
.create_thread_with(CreateThreadRequest {
title: Some("Routing manual".to_string()),
workspace: test_workspace(),
workspace_id: None,
root_id: None,
provider: Some(selected.provider.clone()),
model: Some(selected.model.clone()),
tool_allowlist: Vec::new(),
developer_instructions: None,
external_tools: Vec::new(),
selection_mode: Some(ModelSelectionMode::manual(
selected.provider.clone(),
selected.model.clone(),
None,
)),
runner: None,
})
.await
.unwrap()
.thread_id;
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id,
message: "use selected manual model".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle::default(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
let mut saw_routing_event = false;
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::InferenceRoutingDecision(_) => {
saw_routing_event = true;
}
RoderEvent::TurnCompleted(_) => break,
RoderEvent::TurnFailed(event) => panic!("turn failed: {}", event.error),
_ => {}
}
}
})
.await
.unwrap();
assert!(!saw_routing_event);
assert!(contexts.lock().unwrap().is_empty());
let requests = requests.lock().unwrap();
assert_eq!(requests.len(), 1);
assert_eq!(requests[0].model, selected);
let _ = std::fs::remove_dir_all(thread_root);
}
#[test]
fn enabled_inference_router_requires_registered_router() {
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(FakeInferenceEngine));
let err = match Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
inference_router: RuntimeInferenceRouterConfig {
enabled: true,
router_id: Some("missing-router".to_string()),
},
..RuntimeConfig::default()
},
) {
Ok(_) => panic!("runtime should reject unknown inference router"),
Err(err) => err,
};
assert!(
err.to_string()
.contains("inference router \"missing-router\" is not registered")
);
}
#[tokio::test]
async fn model_profile_routes_request_knobs_to_next_inference() {
let request = captured_profile_request(RuntimeConfig {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "gpt-5.5".to_string(),
model_profiles: std::collections::HashMap::from([(
"gpt-5.5".to_string(),
test_model_profile("gpt-5.5"),
)]),
..RuntimeConfig::default()
})
.await;
let tool_names = request
.tools
.iter()
.map(|tool| tool.name.as_str())
.collect::<Vec<_>>();
assert!(tool_names.contains(&"apply_patch"));
assert!(tool_names.contains(&"edit"));
assert!(tool_names.contains(&"multi_edit"));
assert!(tool_names.contains(&"write_file"));
assert_eq!(request.reasoning.level.as_deref(), Some(REASONING_LOW));
assert_eq!(request.runtime.parallel_tool_calls, Some(false));
assert_eq!(request.runtime.auto_compact_token_limit, Some(123_000));
assert!(
request
.instructions
.developer
.as_deref()
.unwrap_or_default()
.contains("Use the provided context as the current working set")
);
assert_eq!(
request
.metadata
.pointer("/modelProfile/schemaPolicy")
.and_then(serde_json::Value::as_str),
Some("standard_required_first")
);
}
#[tokio::test]
async fn ultra_reasoning_reaches_transport_state_and_enables_proactive_delegation() {
let request = captured_profile_request(RuntimeConfig {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "gpt-5.6-sol".to_string(),
reasoning: Some(REASONING_ULTRA.to_string()),
..RuntimeConfig::default()
})
.await;
assert_eq!(request.reasoning.level.as_deref(), Some(REASONING_ULTRA));
let developer = request
.instructions
.developer
.as_deref()
.expect("ultra developer instructions");
assert!(developer.contains("Proactive multi-agent delegation is active"));
}
#[tokio::test]
async fn lower_sol_effort_requires_explicit_multi_agent_request() {
let request = captured_profile_request(RuntimeConfig {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "gpt-5.6-sol".to_string(),
reasoning: Some(roder_api::catalog::REASONING_MEDIUM.to_string()),
..RuntimeConfig::default()
})
.await;
let developer = request
.instructions
.developer
.as_deref()
.expect("sol developer instructions");
assert!(developer.contains("Do not spawn sub-agents unless"));
assert!(!developer.contains("Proactive multi-agent delegation is active"));
}
#[tokio::test]
async fn luna_does_not_receive_codex_v2_multi_agent_policy() {
let request = captured_profile_request(RuntimeConfig {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "gpt-5.6-luna".to_string(),
reasoning: Some(roder_api::catalog::REASONING_MAX.to_string()),
..RuntimeConfig::default()
})
.await;
let developer = request
.instructions
.developer
.as_deref()
.unwrap_or_default();
assert!(!developer.contains("Do not spawn sub-agents unless"));
assert!(!developer.contains("Proactive multi-agent delegation is active"));
}
#[tokio::test]
async fn turn_developer_context_reaches_inference_and_does_not_persist() {
let captured = Arc::new(StdMutex::new(None));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(CapturingEngine {
request: captured.clone(),
}));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "gpt-5.5".to_string(),
..RuntimeConfig::default()
},
)
.unwrap(),
);
async fn run_turn(runtime: &Arc<Runtime>, developer_context: Option<String>) {
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: "thread-turn-context".to_string(),
message: "hello".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle::default(),
developer_context,
task_ledger_required: false,
})
.await
.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::TurnCompleted(_) => break,
RoderEvent::TurnFailed(event) => panic!("turn failed: {}", event.error),
_ => {}
}
}
})
.await
.unwrap();
}
run_turn(
&runtime,
Some("Connected accounts: example-service.".to_string()),
)
.await;
let request = captured.lock().unwrap().clone().unwrap();
assert_eq!(
request.instructions.developer_context.as_deref(),
Some("Connected accounts: example-service.")
);
// The context is per-turn only: the next turn on the same thread
// without a developerContext must not see the previous one.
run_turn(&runtime, None).await;
let request = captured.lock().unwrap().clone().unwrap();
assert_eq!(request.instructions.developer_context, None);
}
#[tokio::test]
async fn tool_search_overrides_route_to_next_inference_request() {
let request = captured_profile_request(RuntimeConfig {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "gpt-5.4".to_string(),
tool_search: ToolSearchConfig {
mode: roder_api::inference::ToolSearchMode::Auto,
max_catalog_items: Some(100),
..ToolSearchConfig::default()
},
provider_tool_search: std::collections::HashMap::from([(
roder_api::catalog::PROVIDER_MOCK.to_string(),
roder_api::inference::ToolSearchConfigOverlay {
include_skills: Some(false),
provider_variant: Some(roder_api::inference::ToolSearchProviderVariant::Regex),
..Default::default()
},
)]),
model_tool_search: std::collections::HashMap::from([(
"gpt-5.4".to_string(),
roder_api::inference::ToolSearchConfigOverlay {
mode: Some(roder_api::inference::ToolSearchMode::ProviderNative),
max_catalog_items: Some(25),
provider_variant: Some(roder_api::inference::ToolSearchProviderVariant::Bm25),
..Default::default()
},
)]),
..RuntimeConfig::default()
})
.await;
assert_eq!(
request.runtime.tool_search.mode,
roder_api::inference::ToolSearchMode::ProviderNative
);
assert_eq!(request.runtime.tool_search.max_catalog_items, Some(25));
assert_eq!(
request.runtime.tool_search.provider_variant,
roder_api::inference::ToolSearchProviderVariant::Bm25
);
}
#[tokio::test]
async fn context_entrypoint_hints_use_turn_workspace() {
let process_workspace = runtime_test_workspace("entrypoint-process");
let thread_workspace = runtime_test_workspace("entrypoint-thread");
std::fs::create_dir_all(process_workspace.join("src")).unwrap();
std::fs::create_dir_all(thread_workspace.join("src")).unwrap();
std::fs::write(
process_workspace.join("src/sidebar-thread-groups.ts"),
"export const desktopLeak = true;\n",
)
.unwrap();
std::fs::write(
thread_workspace.join("src/voice-plan-feedback.ts"),
"export const voicePlanFeedback = true;\n",
)
.unwrap();
let captured = Arc::new(StdMutex::new(None));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(CapturingEngine {
request: captured.clone(),
}));
builder.context_planner(Arc::new(roder_context::EntrypointContextPlanner::new(
process_workspace.clone(),
)));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
workspace: Some(process_workspace.display().to_string()),
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: "thread-workspace-entrypoint".to_string(),
message: "investigate voice plan feedback".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: thread_workspace.display().to_string(),
instructions: crate::instructions::default_instructions(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::TurnCompleted(_) => break,
RoderEvent::TurnFailed(event) => panic!("turn failed: {}", event.error),
_ => {}
}
}
})
.await
.unwrap();
let request = captured.lock().unwrap().clone().expect("captured request");
let transcript_text = request
.transcript
.iter()
.map(|item| match item {
TranscriptItem::UserMessage(message) => message.text.as_str(),
_ => "",
})
.collect::<Vec<_>>()
.join("\n");
assert!(transcript_text.contains("src/voice-plan-feedback.ts"));
assert!(!transcript_text.contains("src/sidebar-thread-groups.ts"));
let _ = std::fs::remove_dir_all(process_workspace);
let _ = std::fs::remove_dir_all(thread_workspace);
}
fn runtime_test_workspace(name: &str) -> std::path::PathBuf {
let path =
std::env::temp_dir().join(format!("roder-runtime-{name}-{}", uuid::Uuid::new_v4()));
let _ = std::fs::remove_dir_all(&path);
std::fs::create_dir_all(&path).unwrap();
path
}
#[tokio::test]
async fn model_profile_user_model_knobs_override_profile_defaults() {
let request = captured_profile_request(RuntimeConfig {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "gpt-5.5".to_string(),
reasoning: Some(REASONING_HIGH.to_string()),
auto_compact_token_limit: Some(999),
model_edit_tools: std::collections::HashMap::from([(
"gpt-5.5".to_string(),
EDIT_TOOL_PATCH.to_string(),
)]),
model_parallel_tool_calls: std::collections::HashMap::from([(
"gpt-5.5".to_string(),
true,
)]),
model_profiles: std::collections::HashMap::from([(
"gpt-5.5".to_string(),
test_model_profile("gpt-5.5"),
)]),
..RuntimeConfig::default()
})
.await;
let tool_names = request
.tools
.iter()
.map(|tool| tool.name.as_str())
.collect::<Vec<_>>();
assert!(tool_names.contains(&"apply_patch"));
assert!(!tool_names.contains(&"edit"));
assert_eq!(request.reasoning.level.as_deref(), Some(REASONING_HIGH));
assert_eq!(request.runtime.parallel_tool_calls, Some(true));
assert_eq!(request.runtime.auto_compact_token_limit, Some(999));
}
#[tokio::test]
async fn runtime_tool_allowlist_filters_advertised_tools() {
let request = captured_profile_request(RuntimeConfig {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "gpt-5.5".to_string(),
tool_allowlist: vec!["edit".to_string()],
model_profiles: std::collections::HashMap::from([(
"gpt-5.5".to_string(),
test_model_profile("gpt-5.5"),
)]),
..RuntimeConfig::default()
})
.await;
let tool_names = request
.tools
.iter()
.map(|tool| tool.name.as_str())
.collect::<Vec<_>>();
assert_eq!(tool_names, vec!["edit"]);
}
/// Builds a store-backed runtime turn for a thread created with the given per-thread
/// overrides and returns the captured inference request.
async fn captured_thread_override_request(
runtime: &Arc<Runtime>,
requests: &Arc<StdMutex<Vec<AgentInferenceRequest>>>,
tool_allowlist: Vec<String>,
developer_instructions: Option<String>,
external_tools: Vec<ToolSpec>,
) -> AgentInferenceRequest {
let thread_id = runtime
.create_thread_with(CreateThreadRequest {
title: Some("Thread overrides".to_string()),
workspace: test_workspace(),
workspace_id: None,
root_id: None,
provider: None,
model: None,
selection_mode: None,
tool_allowlist,
developer_instructions,
external_tools,
runner: None,
})
.await
.unwrap()
.thread_id;
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id,
message: "hello".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: crate::default_instructions(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::TurnCompleted(_) => break,
RoderEvent::TurnFailed(event) => panic!("turn failed: {}", event.error),
_ => {}
}
}
})
.await
.unwrap();
requests.lock().unwrap().pop().expect("captured request")
}
#[tokio::test]
async fn thread_tool_allowlist_filters_only_that_thread() {
let requests = Arc::new(StdMutex::new(Vec::new()));
let thread_root =
std::env::temp_dir().join(format!("roder-thread-allowlist-{}", uuid::Uuid::new_v4()));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(SwitchCaptureEngine {
requests: requests.clone(),
}));
builder.thread_store_factory(Arc::new(JsonlThreadStoreFactory {
base_path: thread_root.clone(),
}));
builder.tool_contributor(Arc::new(ProfileToolContributor));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "gpt-5.5".to_string(),
model_profiles: std::collections::HashMap::from([(
"gpt-5.5".to_string(),
test_model_profile("gpt-5.5"),
)]),
..RuntimeConfig::default()
},
)
.unwrap(),
);
let allowlisted = captured_thread_override_request(
&runtime,
&requests,
vec!["edit".to_string()],
None,
Vec::new(),
)
.await;
let unrestricted =
captured_thread_override_request(&runtime, &requests, Vec::new(), None, Vec::new())
.await;
let allowlisted_names = allowlisted
.tools
.iter()
.map(|tool| tool.name.as_str())
.collect::<Vec<_>>();
assert_eq!(allowlisted_names, vec!["edit"]);
let unrestricted_names = unrestricted
.tools
.iter()
.map(|tool| tool.name.as_str())
.collect::<Vec<_>>();
assert!(unrestricted_names.contains(&"edit"));
assert!(unrestricted_names.len() > 1);
let _ = std::fs::remove_dir_all(thread_root);
}
/// Builds a store-backed runtime whose `RuntimeConfig.tool_allowlist` is `["edit"]`.
fn runtime_with_edit_allowlist(
requests: &Arc<StdMutex<Vec<AgentInferenceRequest>>>,
thread_root: &std::path::Path,
) -> Arc<Runtime> {
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(SwitchCaptureEngine {
requests: requests.clone(),
}));
builder.thread_store_factory(Arc::new(JsonlThreadStoreFactory {
base_path: thread_root.to_path_buf(),
}));
builder.tool_contributor(Arc::new(ProfileToolContributor));
Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "gpt-5.5".to_string(),
tool_allowlist: vec!["edit".to_string()],
model_profiles: std::collections::HashMap::from([(
"gpt-5.5".to_string(),
test_model_profile("gpt-5.5"),
)]),
..RuntimeConfig::default()
},
)
.unwrap(),
)
}
#[tokio::test]
async fn runtime_and_thread_allowlists_intersect() {
let requests = Arc::new(StdMutex::new(Vec::new()));
let thread_root = std::env::temp_dir().join(format!(
"roder-allowlist-intersect-{}",
uuid::Uuid::new_v4()
));
let runtime = runtime_with_edit_allowlist(&requests, &thread_root);
let request = captured_thread_override_request(
&runtime,
&requests,
vec!["edit".to_string(), "write_file".to_string()],
None,
Vec::new(),
)
.await;
// The thread allowlist must not re-enable tools the runtime allowlist bans.
let names = request
.tools
.iter()
.map(|tool| tool.name.as_str())
.collect::<Vec<_>>();
assert_eq!(names, vec!["edit"]);
let _ = std::fs::remove_dir_all(thread_root);
}
#[tokio::test]
async fn route_tool_call_denies_tools_outside_allowlists() {
let requests = Arc::new(StdMutex::new(Vec::new()));
let thread_root =
std::env::temp_dir().join(format!("roder-allowlist-dispatch-{}", uuid::Uuid::new_v4()));
let runtime = runtime_with_edit_allowlist(&requests, &thread_root);
let thread_id = runtime
.create_thread_with(CreateThreadRequest {
title: Some("Dispatch allowlist".to_string()),
workspace: test_workspace(),
workspace_id: None,
root_id: None,
provider: None,
model: None,
selection_mode: None,
tool_allowlist: vec!["edit".to_string(), "write_file".to_string()],
developer_instructions: None,
external_tools: Vec::new(),
runner: None,
})
.await
.unwrap()
.thread_id;
// write_file is registered and on the thread allowlist but banned by the runtime allowlist.
let result = runtime
.route_tool_call(
&thread_id,
&"turn-allowlist-dispatch".to_string(),
roder_api::inference::ToolCallCompleted {
id: "call-1".to_string(),
name: "write_file".to_string(),
arguments: r#"{"path":"a.txt","content":"hi"}"#.to_string(),
},
None,
None,
)
.await
.unwrap();
assert!(result.is_error);
assert!(
result
.result
.contains("not permitted by the tool allowlist"),
"unexpected result: {}",
result.result
);
let _ = std::fs::remove_dir_all(thread_root);
}
#[tokio::test]
async fn route_tool_calls_denies_agent_swarm_mixed_with_other_tools() {
let requests = Arc::new(StdMutex::new(Vec::new()));
let thread_root =
std::env::temp_dir().join(format!("roder-swarm-exclusive-{}", uuid::Uuid::new_v4()));
let runtime = runtime_with_edit_allowlist(&requests, &thread_root);
// A response that mixes agent_swarm with another tool is denied wholesale:
// both calls get an error result with retry guidance, and no tool runs.
let results = runtime
.route_tool_calls(
&"thread-swarm".to_string(),
&"turn-swarm".to_string(),
vec![
roder_api::inference::ToolCallCompleted {
id: "swarm-1".to_string(),
name: "agent_swarm".to_string(),
arguments: "{}".to_string(),
},
roder_api::inference::ToolCallCompleted {
id: "read-1".to_string(),
name: "read_file".to_string(),
arguments: "{}".to_string(),
},
],
true,
None,
None,
)
.await
.unwrap();
assert_eq!(results.len(), 2, "every tool_call_id must get a response");
assert_eq!(results[0].id, "swarm-1");
assert_eq!(results[1].id, "read-1");
for result in &results {
assert!(result.is_error);
assert!(
result.result.contains("only tool call"),
"unexpected result: {}",
result.result
);
}
let _ = std::fs::remove_dir_all(thread_root);
}
#[tokio::test]
async fn route_tool_calls_emits_agent_swarm_started_event() {
let requests = Arc::new(StdMutex::new(Vec::new()));
let thread_root =
std::env::temp_dir().join(format!("roder-swarm-started-{}", uuid::Uuid::new_v4()));
let runtime = runtime_with_edit_allowlist(&requests, &thread_root);
let mut events = runtime.subscribe_events();
// A single agent_swarm call (the valid shape) emits AgentSwarmStarted on
// the event bus before dispatch, with the child count parsed from args.
let _ = runtime
.route_tool_calls(
&"thread-swarm".to_string(),
&"turn-swarm".to_string(),
vec![roder_api::inference::ToolCallCompleted {
id: "swarm-1".to_string(),
name: "agent_swarm".to_string(),
arguments: r#"{"description":"x","prompt_template":"Read {{item}}","items":["a.rs","b.rs"]}"#
.to_string(),
}],
true,
None,
None,
)
.await
.unwrap();
let mut started_child_count = None;
for _ in 0..16 {
let envelope = tokio::time::timeout(std::time::Duration::from_secs(2), events.recv())
.await
.unwrap()
.unwrap();
if let RoderEvent::AgentSwarmStarted(event) = envelope.event {
started_child_count = Some(event.child_count);
assert_eq!(event.tool_id, "swarm-1");
break;
}
}
assert_eq!(started_child_count, Some(2));
let _ = std::fs::remove_dir_all(thread_root);
}
#[test]
fn agent_swarm_child_count_sums_items_and_resumes() {
assert_eq!(
agent_swarm_child_count(r#"{"description":"x","items":["a","b","c"]}"#),
3
);
assert_eq!(
agent_swarm_child_count(
r#"{"description":"x","items":["a"],"resume_agent_ids":{"id1":"continue"}}"#
),
2
);
// Malformed input is lenient.
assert_eq!(agent_swarm_child_count("not json"), 0);
}
#[test]
fn parse_swarm_counts_reads_summary_with_omitted_buckets() {
let text = "<agent_swarm_result>\n<summary>completed: 2, failed: 1</summary>\n</agent_swarm_result>";
assert_eq!(parse_swarm_counts(text), Some((2, 1, 0)));
let text = "<agent_swarm_result>\n<summary>completed: 0</summary>\n</agent_swarm_result>";
assert_eq!(parse_swarm_counts(text), Some((0, 0, 0)));
// Not a swarm result.
assert_eq!(parse_swarm_counts("just text"), None);
}
/// Signals when inference starts, then waits for `proceed` and fails the stream.
struct SignalledFailureEngine {
started: tokio::sync::mpsc::UnboundedSender<()>,
proceed: Arc<tokio::sync::Notify>,
}
#[async_trait::async_trait]
impl InferenceEngine for SignalledFailureEngine {
fn id(&self) -> String {
roder_api::catalog::PROVIDER_MOCK.to_string()
}
fn capabilities(&self) -> InferenceCapabilities {
InferenceCapabilities::coding_agent_default()
}
async fn list_models(
&self,
_ctx: InferenceProviderContext<'_>,
) -> anyhow::Result<Vec<roder_api::inference::ModelDescriptor>> {
Ok(roder_api::catalog::models_for_provider(
roder_api::catalog::PROVIDER_MOCK,
true,
))
}
async fn stream_turn(
&self,
_ctx: InferenceTurnContext<'_>,
_request: AgentInferenceRequest,
) -> anyhow::Result<InferenceEventStream> {
let _ = self.started.send(());
self.proceed.notified().await;
anyhow::bail!("engine failed mid-turn")
}
}
#[tokio::test]
async fn failed_turn_sweeps_pending_external_tool_calls() {
let (started_tx, mut started_rx) = tokio::sync::mpsc::unbounded_channel();
let proceed = Arc::new(tokio::sync::Notify::new());
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(SignalledFailureEngine {
started: started_tx,
proceed: proceed.clone(),
}));
let runtime =
Arc::new(Runtime::new(builder.build().unwrap(), RuntimeConfig::default()).unwrap());
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: "thread-sweep".to_string(),
message: "go".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle {
system: None,
developer: None,
developer_context: None,
},
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(5), started_rx.recv())
.await
.unwrap()
.unwrap();
let (tx, _pending_rx) = oneshot::channel();
runtime.pending_external_tool_calls.lock().await.insert(
"exttool-sweep-test".to_string(),
PendingExternalToolCall {
thread_id: "thread-sweep".to_string(),
turn_id: turn_id.clone(),
tool_id: "call-1".to_string(),
tool_name: "acme_lookup".to_string(),
tx,
},
);
proceed.notify_one();
let outcome = tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if let RoderEvent::ExternalToolCallResolved(event) = envelope.event
&& event.request_id == "exttool-sweep-test"
{
break event.outcome;
}
}
})
.await
.expect("turn failure must resolve pending external tool calls");
assert_eq!(outcome, ExternalToolCallOutcome::Cancelled);
assert!(runtime.pending_external_tool_calls.lock().await.is_empty());
}
#[tokio::test]
async fn thread_developer_instructions_layer_under_harness_prompt() {
let requests = Arc::new(StdMutex::new(Vec::new()));
let thread_root = std::env::temp_dir().join(format!(
"roder-thread-instructions-{}",
uuid::Uuid::new_v4()
));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(SwitchCaptureEngine {
requests: requests.clone(),
}));
builder.thread_store_factory(Arc::new(JsonlThreadStoreFactory {
base_path: thread_root.clone(),
}));
builder.tool_contributor(Arc::new(ProfileToolContributor));
let runtime =
Arc::new(Runtime::new(builder.build().unwrap(), RuntimeConfig::default()).unwrap());
let request = captured_thread_override_request(
&runtime,
&requests,
Vec::new(),
Some("You are embedded in a host app.".to_string()),
Vec::new(),
)
.await;
let system = request.instructions.system.expect("system instructions");
assert!(system.starts_with("You are Roder"));
let developer = request
.instructions
.developer
.expect("developer instructions");
assert!(developer.starts_with("You are embedded in a host app."));
let plain =
captured_thread_override_request(&runtime, &requests, Vec::new(), None, Vec::new())
.await;
assert_eq!(plain.instructions.developer, None);
let _ = std::fs::remove_dir_all(thread_root);
}
#[tokio::test]
async fn thread_external_tools_are_advertised_and_shadow_builtins() {
let requests = Arc::new(StdMutex::new(Vec::new()));
let thread_root = std::env::temp_dir().join(format!(
"roder-thread-external-tools-{}",
uuid::Uuid::new_v4()
));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(SwitchCaptureEngine {
requests: requests.clone(),
}));
builder.thread_store_factory(Arc::new(JsonlThreadStoreFactory {
base_path: thread_root.clone(),
}));
builder.tool_contributor(Arc::new(ProfileToolContributor));
let runtime =
Arc::new(Runtime::new(builder.build().unwrap(), RuntimeConfig::default()).unwrap());
let external_tools = vec![
ToolSpec {
name: "acme_lookup".to_string(),
description: "Look up Acme workspace state.".to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": { "query": { "type": "string" } },
"required": ["query"]
}),
},
ToolSpec {
name: "edit".to_string(),
description: "Host-managed edit.".to_string(),
parameters: serde_json::json!({ "type": "object" }),
},
];
let request =
captured_thread_override_request(&runtime, &requests, Vec::new(), None, external_tools)
.await;
let acme = request
.tools
.iter()
.find(|tool| tool.name == "acme_lookup")
.expect("external tool advertised");
assert_eq!(acme.description, "Look up Acme workspace state.");
assert_eq!(acme.parameters["required"][0], "query");
let edits = request
.tools
.iter()
.filter(|tool| tool.name == "edit")
.collect::<Vec<_>>();
assert_eq!(edits.len(), 1, "external edit shadows the builtin");
assert_eq!(edits[0].description, "Host-managed edit.");
let plain =
captured_thread_override_request(&runtime, &requests, Vec::new(), None, Vec::new())
.await;
assert!(plain.tools.iter().all(|tool| tool.name != "acme_lookup"));
let plain_edit = plain
.tools
.iter()
.find(|tool| tool.name == "edit")
.expect("builtin edit advertised on plain thread");
assert_eq!(plain_edit.description, "edit test tool");
let _ = std::fs::remove_dir_all(thread_root);
}
#[tokio::test]
async fn model_switch_injects_summary_and_records_profile_segments() {
let requests = Arc::new(StdMutex::new(Vec::new()));
let thread_root = std::env::temp_dir().join(format!(
"roder-model-switch-thread-{}",
uuid::Uuid::new_v4()
));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(SwitchCaptureEngine {
requests: requests.clone(),
}));
builder.thread_store_factory(Arc::new(JsonlThreadStoreFactory {
base_path: thread_root.clone(),
}));
builder.tool_contributor(Arc::new(ProfileToolContributor));
let mut claude_profile = test_model_profile("claude-haiku-4-5-20251001");
claude_profile.provider_family = ProviderFamily::Anthropic;
claude_profile.edit_tool = Some(EDIT_TOOL_EDIT.to_string());
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "gpt-5.5".to_string(),
model_profiles: std::collections::HashMap::from([
("gpt-5.5".to_string(), test_model_profile("gpt-5.5")),
("claude-haiku-4-5-20251001".to_string(), claude_profile),
]),
..RuntimeConfig::default()
},
)
.unwrap(),
);
let thread_id = runtime
.create_thread_with(CreateThreadRequest {
title: Some("Model switch".to_string()),
workspace: test_workspace(),
workspace_id: None,
root_id: None,
provider: None,
model: None,
selection_mode: None,
tool_allowlist: Vec::new(),
developer_instructions: None,
external_tools: Vec::new(),
runner: None,
})
.await
.unwrap()
.thread_id;
let mut rx = runtime.subscribe_events();
for (message, model_override) in [
("first turn", None),
("second turn", Some("claude-haiku-4-5-20251001".to_string())),
] {
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: thread_id.clone(),
message: message.to_string(),
images: Vec::new(),
provider_override: None,
model_override,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle::default(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::TurnCompleted(_) => break,
RoderEvent::TurnFailed(event) => panic!("turn failed: {}", event.error),
_ => {}
}
}
})
.await
.unwrap();
}
let captured = requests.lock().unwrap().clone();
assert_eq!(captured.len(), 2);
assert!(captured[1].transcript.iter().any(|item| {
matches!(
item,
TranscriptItem::UserMessage(message)
if message.text.starts_with(MODEL_SWITCH_SUMMARY_PREFIX)
&& message.text.contains("previous profile mock/gpt-5.5")
&& message.text.contains("Current profile mock/claude-haiku-4-5-20251001")
&& message.text.contains("Available tools now:")
)
}));
let snapshot = runtime
.thread_store
.as_ref()
.unwrap()
.load_thread(&thread_id)
.await
.unwrap()
.unwrap();
let trace_segments = snapshot
.turns
.iter()
.flat_map(|turn| &turn.items)
.filter(|item| {
matches!(
item,
TranscriptItem::ProviderMetadata(value)
if value.get("kind").and_then(serde_json::Value::as_str)
== Some(MODEL_PROFILE_TRACE_KIND)
&& value.get("segment").and_then(serde_json::Value::as_str)
== Some("assistant")
)
})
.count();
assert!(trace_segments >= 2);
let _ = std::fs::remove_dir_all(thread_root);
}
struct CountingTaskTool {
calls: Arc<StdMutex<u32>>,
}
#[async_trait::async_trait]
impl ToolExecutor for CountingTaskTool {
fn spec(&self) -> ToolSpec {
ToolSpec {
name: "task".to_string(),
description: "Dispatch a test subagent.".to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"description": { "type": "string" },
"prompt": { "type": "string" },
"parent_deadline_seconds": { "type": "integer" }
},
"required": ["description", "prompt"],
"additionalProperties": false
}),
}
}
async fn execute(
&self,
_ctx: ToolExecutionContext,
call: ToolCall,
) -> anyhow::Result<ToolResult> {
*self.calls.lock().unwrap() += 1;
Ok(ToolResult {
id: call.id,
name: call.name,
text: "started child".to_string(),
data: serde_json::json!({}),
is_error: false,
})
}
}
#[tokio::test]
async fn runtime_profile_reaches_inference_request_and_turn_metadata() {
let captured = Arc::new(StdMutex::new(None));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(CapturingEngine {
request: captured.clone(),
}));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
runtime_profile: RuntimeProfile::NonInteractive,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: "thread-profile".to_string(),
message: "work unattended".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle {
system: None,
developer: Some("base developer".to_string()),
developer_context: None,
},
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
let mut observed_profile = None;
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::TurnStarted(event) => {
observed_profile = Some(event.runtime_profile);
}
RoderEvent::TurnCompleted(_) => break,
RoderEvent::TurnFailed(event) => panic!("turn failed: {}", event.error),
_ => {}
}
}
})
.await
.unwrap();
assert_eq!(observed_profile, Some(RuntimeProfile::NonInteractive));
let request = captured.lock().unwrap().clone().unwrap();
assert_eq!(request.runtime.profile, RuntimeProfile::NonInteractive);
let developer = request.instructions.developer.unwrap();
assert!(developer.contains("base developer"));
assert!(developer.contains("non-interactive profile"));
}
#[tokio::test]
async fn global_policy_mode_changes_do_not_create_runtime_thread_directory() {
let workspace = runtime_test_workspace("global-policy-mode");
let thread_root = workspace.join("threads");
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(FakeInferenceEngine));
builder.thread_store_factory(Arc::new(JsonlThreadStoreFactory {
base_path: thread_root.clone(),
}));
let runtime = Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
workspace: Some(workspace.display().to_string()),
..Default::default()
},
)
.unwrap();
runtime
.set_policy_mode(PolicyMode::AcceptAll, Some("test".to_string()))
.await
.unwrap();
assert!(!thread_root.join("runtime").exists());
let _ = std::fs::remove_dir_all(workspace);
}
#[tokio::test]
async fn task_ledger_enforcement_injects_eval_reminder_before_work() {
let captured = Arc::new(StdMutex::new(None));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(CapturingEngine {
request: captured.clone(),
}));
builder.tool_contributor(Arc::new(
roder_ext_task_ledger::TaskLedgerToolContributor::default(),
));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
runtime_profile: RuntimeProfile::Eval,
policy_mode: PolicyMode::Bypass,
agent_swarm_mode: false,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: "thread-ledger".to_string(),
message: "decomposed work".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle::default(),
developer_context: None,
task_ledger_required: true,
})
.await
.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() == Some(&turn_id)
&& matches!(envelope.event, RoderEvent::TurnCompleted(_))
{
break;
}
}
})
.await
.unwrap();
let request = captured.lock().unwrap().clone().unwrap();
let developer = request.instructions.developer.unwrap();
assert!(developer.contains("Task Ledger Required"));
assert!(developer.contains("task_ledger.update"));
let tool_names: Vec<_> = request
.tools
.iter()
.map(|tool| tool.name.as_str())
.collect();
assert!(
tool_names.contains(&TASK_LEDGER_TOOL_NAME),
"tool names: {tool_names:?}"
);
assert_eq!(
request.tool_choice,
ToolChoice::Specific(TASK_LEDGER_TOOL_NAME.to_string())
);
assert_eq!(request.tools.len(), 1);
assert_eq!(request.tools[0].name, TASK_LEDGER_TOOL_NAME);
}
#[tokio::test]
async fn eval_task_ledger_blocks_final_answer_until_open_items_are_completed() {
let requests = Arc::new(StdMutex::new(Vec::new()));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(TaskLedgerCompletionGateEngine {
calls: StdMutex::new(0),
requests: requests.clone(),
}));
builder.tool_contributor(Arc::new(
roder_ext_task_ledger::TaskLedgerToolContributor::default(),
));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
runtime_profile: RuntimeProfile::Eval,
policy_mode: PolicyMode::Bypass,
agent_swarm_mode: false,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: "thread-ledger-completion".to_string(),
message: "write the answer file".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle::default(),
developer_context: None,
task_ledger_required: true,
})
.await
.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::TurnCompleted(_) => break,
RoderEvent::TurnFailed(event) => panic!("turn failed: {}", event.error),
_ => {}
}
}
})
.await
.unwrap();
let requests = requests.lock().unwrap().clone();
assert_eq!(requests.len(), 4);
assert!(requests[2].transcript.iter().any(|item| {
matches!(
item,
TranscriptItem::UserMessage(message)
if message.text.contains("Task Ledger Completion Required")
&& message.text.contains("Write /app/result.txt")
)
}));
assert!(requests[3].transcript.iter().any(|item| {
matches!(
item,
TranscriptItem::ToolResult(result)
if result.name.as_deref() == Some(TASK_LEDGER_TOOL_NAME)
&& result.result.contains("Task ledger: 2/2 completed")
)
}));
}
#[tokio::test]
async fn eval_task_ledger_checkpoint_requests_scoreable_file_before_final_reserve() {
let requests = Arc::new(StdMutex::new(Vec::new()));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(TaskLedgerCompletionGateEngine {
calls: StdMutex::new(0),
requests: requests.clone(),
}));
builder.tool_contributor(Arc::new(
roder_ext_task_ledger::TaskLedgerToolContributor::default(),
));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
runtime_profile: RuntimeProfile::Eval,
policy_mode: PolicyMode::Bypass,
agent_swarm_mode: false,
turn_deadline_seconds: Some(120),
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: "thread-ledger-checkpoint".to_string(),
message: "write the answer file".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle::default(),
developer_context: None,
task_ledger_required: true,
})
.await
.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::TurnCompleted(_) => break,
RoderEvent::TurnFailed(event) => panic!("turn failed: {}", event.error),
_ => {}
}
}
})
.await
.unwrap();
let requests = requests.lock().unwrap().clone();
assert!(requests.len() >= 2);
assert!(requests[1].transcript.iter().any(|item| {
matches!(
item,
TranscriptItem::UserMessage(message)
if message.text.contains("Scoreable Output Checkpoint")
&& message.text.contains("ensure the required output file(s) exist")
&& message.text.contains("Write /app/result.txt")
)
}));
}
#[test]
fn deadline_task_ledger_prompt_preserves_scoreable_work_instruction() {
let prompt = task_ledger_deadline_completion_prompt(
12,
30,
"Task Ledger Completion Required: write /app/result.txt, then call task_ledger.update",
);
assert!(prompt.contains("12 seconds remain"));
assert!(prompt.contains("create or update the required scoreable output files"));
assert!(prompt.contains("write /app/result.txt"));
assert!(prompt.contains(TASK_LEDGER_TOOL_NAME));
}
#[test]
fn scoreable_checkpoint_prompt_preserves_provisional_file_instruction() {
let prompt = task_ledger_scoreable_checkpoint_prompt(
120,
"Task Ledger Completion Required: write /app/result.txt, then call task_ledger.update",
);
assert!(prompt.contains("120 seconds remain"));
assert!(prompt.contains("best evidence-backed answer"));
assert!(prompt.contains("even if provisional"));
assert!(prompt.contains("preserve that candidate"));
assert!(prompt.contains("partial-coverage"));
assert!(prompt.contains("write /app/result.txt"));
assert!(prompt.contains(TASK_LEDGER_TOOL_NAME));
}
#[test]
fn open_task_ledger_moves_inference_timeout_to_scoreable_checkpoint() {
let deadline = Some(OffsetDateTime::now_utc() + Duration::seconds(870));
let transcript = vec![TranscriptItem::ToolResult(ToolResultRecord {
id: "ledger-open".to_string(),
name: Some(TASK_LEDGER_TOOL_NAME.to_string()),
result: "Task ledger: 0/1 completed\n- pending: Write /app/result.txt [write]"
.to_string(),
display_payload: None,
is_error: false,
})];
let (_, action) = inference_timeout_deadline(
deadline,
RuntimeProfile::Eval,
true,
30,
false,
0,
&transcript,
)
.unwrap();
assert_eq!(action, InferenceTimeoutAction::ScoreableCheckpoint);
}
#[tokio::test]
async fn verification_gate_forces_eval_code_changes_through_review() {
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(VerificationGateEngine {
calls: StdMutex::new(0),
}));
builder.tool_contributor(Arc::new(WriteFileContributor));
builder.tool_contributor(Arc::new(
roder_ext_verification::VerificationToolContributor,
));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "mock".to_string(),
runtime_profile: RuntimeProfile::Eval,
policy_mode: PolicyMode::Bypass,
agent_swarm_mode: false,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: "thread-verification".to_string(),
message: "write code".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle::default(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
let mut saw_required = false;
let mut saw_completed = false;
let mut final_text = String::new();
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::VerificationRequired(event) => {
saw_required = true;
assert_eq!(event.changed_files, vec!["src/lib.rs"]);
}
RoderEvent::VerificationCompleted(event) => {
saw_completed = true;
assert!(event.passed);
}
RoderEvent::InferenceEventReceived(event) => {
if let InferenceEvent::MessageDelta(delta) = event.event {
final_text.push_str(&delta.text);
}
}
RoderEvent::TurnCompleted(_) => break,
_ => {}
}
}
})
.await
.unwrap();
assert!(saw_required);
assert!(saw_completed);
assert!(final_text.contains("verified final"));
}
#[tokio::test]
async fn speed_policy_changes_reasoning_across_eval_model_calls_without_model_switch() {
let requests = Arc::new(StdMutex::new(Vec::new()));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(SpeedPolicyEngine {
calls: StdMutex::new(0),
requests: requests.clone(),
}));
builder.tool_contributor(Arc::new(WriteFileContributor));
builder.tool_contributor(Arc::new(
roder_ext_verification::VerificationToolContributor,
));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "gpt-5.5".to_string(),
runtime_profile: RuntimeProfile::Eval,
policy_mode: PolicyMode::Bypass,
agent_swarm_mode: false,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: "thread-speed-policy".to_string(),
message: "write code".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle::default(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
let mut saw_speed_policy_event = false;
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::InferenceStarted(event) => {
if event.speed_policy.is_some() {
saw_speed_policy_event = true;
}
}
RoderEvent::TurnCompleted(_) => break,
RoderEvent::TurnFailed(event) => panic!("turn failed: {}", event.error),
_ => {}
}
}
})
.await
.unwrap();
let requests = requests.lock().unwrap().clone();
assert!(saw_speed_policy_event);
assert!(requests.len() >= 4);
assert!(requests.iter().all(|request| {
request.model.provider == roder_api::catalog::PROVIDER_MOCK
&& request.model.model == "gpt-5.5"
}));
assert_eq!(
requests[0].runtime.speed_policy.as_ref().map(|d| d.phase),
Some(roder_api::inference::SpeedPolicyPhase::Orientation)
);
assert_eq!(requests[0].reasoning.level.as_deref(), Some(REASONING_HIGH));
assert_eq!(
requests[1].runtime.speed_policy.as_ref().map(|d| d.phase),
Some(roder_api::inference::SpeedPolicyPhase::Execution)
);
assert_eq!(requests[1].reasoning.level.as_deref(), Some(REASONING_LOW));
assert_eq!(
requests[2].runtime.speed_policy.as_ref().map(|d| d.phase),
Some(roder_api::inference::SpeedPolicyPhase::Verification)
);
assert_eq!(requests[2].reasoning.level.as_deref(), Some(REASONING_HIGH));
assert_eq!(
requests[2]
.metadata
.pointer("/speedPolicy/phase")
.and_then(serde_json::Value::as_str),
Some("verification")
);
}
#[tokio::test]
async fn deadline_turn_timeout_emits_partial_result_and_clears_active_turn() {
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(DeadlineEngine));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
default_provider: roder_api::catalog::PROVIDER_MOCK.to_string(),
default_model: "mock".to_string(),
runtime_profile: RuntimeProfile::Eval,
turn_deadline_seconds: Some(1),
..RuntimeConfig::default()
},
)
.unwrap(),
);
let mut rx = runtime.subscribe_events();
let turn_id = runtime
.start_turn(StartTurnRequest {
thread_id: "thread-deadline".to_string(),
message: "slow work".to_string(),
images: Vec::new(),
provider_override: None,
model_override: None,
reasoning_override: None,
workspace: test_workspace(),
instructions: InstructionBundle::default(),
developer_context: None,
task_ledger_required: false,
})
.await
.unwrap();
let mut saw_partial = false;
let mut saw_deadline = false;
let mut failed_kind = None;
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let envelope = rx.recv().await.unwrap();
if envelope.turn_id.as_deref() != Some(&turn_id) {
continue;
}
match envelope.event {
RoderEvent::TurnPartialResult(event) => {
saw_partial = event.summary.contains("partial turn state");
}
RoderEvent::TurnDeadlineExceeded(event) => {
saw_deadline = event.partial_result.contains("transcript items");
}
RoderEvent::TurnFailed(event) => {
failed_kind = event.error_kind;
break;
}
_ => {}
}
}
})
.await
.unwrap();
for _ in 0..20 {
if !runtime.active_turns.read().await.contains_key(&turn_id) {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
assert!(saw_partial);
assert!(saw_deadline);
assert_eq!(failed_kind.as_deref(), Some("deadline_timeout"));
assert!(!runtime.active_turns.read().await.contains_key(&turn_id));
}
#[tokio::test]
async fn deadline_skips_subagent_task_when_remaining_budget_is_too_low() {
let calls = Arc::new(StdMutex::new(0));
let mut builder = ExtensionRegistryBuilder::new();
builder.inference_engine(Arc::new(CapturingEngine {
request: Arc::new(StdMutex::new(None)),
}));
let task_tool = Arc::new(CountingTaskTool {
calls: calls.clone(),
});
builder.tool_contributor(Arc::new(TestToolContributor { tool: task_tool }));
let runtime = Arc::new(
Runtime::new(
builder.build().unwrap(),
RuntimeConfig {
policy_mode: PolicyMode::Bypass,
agent_swarm_mode: false,
..RuntimeConfig::default()
},
)
.unwrap(),
);
let result = runtime
.route_tool_call(
&"thread-deadline-task".to_string(),
&"turn-deadline-task".to_string(),
ToolCallCompleted {
id: "task-1".to_string(),
name: "task".to_string(),
arguments: serde_json::json!({
"description": "inspect",
"prompt": "read"
})
.to_string(),
},
None,
Some(OffsetDateTime::now_utc() + Duration::seconds(1)),
)
.await
.unwrap();
assert!(result.is_error);
assert!(result.result.contains("deadline policy skipped"));
assert_eq!(*calls.lock().unwrap(), 0);
}
struct TestToolContributor {
tool: Arc<dyn ToolExecutor>,
}
impl ToolContributor for TestToolContributor {
fn id(&self) -> String {
"test-tool".to_string()
}
fn contribute(&self, registry: &mut ToolRegistry) -> anyhow::Result<()> {
registry.register(self.tool.clone())
}
}
#[tokio::test]
async fn agent_swarm_mode_override_is_per_thread() {
let runtime = Runtime::fake().unwrap();
let trigger = roder_api::subagents::AgentSwarmModeTrigger::Manual;
// Off everywhere by default.
assert!(
!runtime
.effective_agent_swarm_mode_for_thread("thread-a")
.await
);
assert!(
!runtime
.effective_agent_swarm_mode_for_thread("thread-b")
.await
);
// Enabling on thread-a does not leak into thread-b.
assert!(
runtime
.set_agent_swarm_mode_for_thread("thread-a", true, trigger)
.await
);
assert!(
runtime
.effective_agent_swarm_mode_for_thread("thread-a")
.await
);
assert!(
!runtime
.effective_agent_swarm_mode_for_thread("thread-b")
.await
);
// A per-thread `off` override wins over a runtime-global `on` default.
runtime.set_agent_swarm_mode(true, trigger).await.unwrap();
runtime
.set_agent_swarm_mode_for_thread("thread-b", false, trigger)
.await;
assert!(
!runtime
.effective_agent_swarm_mode_for_thread("thread-b")
.await,
"explicit per-thread off overrides the global on default"
);
// A thread with no override still follows the global default.
assert!(
runtime
.effective_agent_swarm_mode_for_thread("thread-c")
.await,
"threads without an override follow the runtime-global default"
);
}
#[tokio::test]
async fn set_agent_swarm_mode_for_thread_emits_event_with_real_thread_id() {
let runtime = Runtime::fake().unwrap();
let mut events = runtime.subscribe_events();
runtime
.set_agent_swarm_mode_for_thread(
"thread-xyz",
true,
roder_api::subagents::AgentSwarmModeTrigger::Task,
)
.await;
let mut saw = false;
for _ in 0..8 {
let envelope = tokio::time::timeout(std::time::Duration::from_secs(2), events.recv())
.await
.unwrap()
.unwrap();
if let RoderEvent::AgentSwarmModeChanged(event) = envelope.event {
assert_eq!(event.thread_id, "thread-xyz");
assert!(event.enabled);
assert_eq!(
event.trigger,
roder_api::subagents::AgentSwarmModeTrigger::Task
);
saw = true;
break;
}
}
assert!(
saw,
"expected an AgentSwarmModeChanged event for the thread"
);
}
}