use std::sync::Arc;
use pretty_assertions::assert_eq;
use rho_sdk::{
model::{ContentBlock, Message, ModelIdentity, ModelResponse, ModelUsage},
provider::{ModelProvider, ScriptedProvider, ScriptedTurn},
CompactionFuture, CompactionOutput, CompactionRequest, Compactor, PolicyDecision,
ProviderError, ProviderErrorKind, Retryability, RunEvent, RunId, SessionId, SessionOptions,
SystemPrompt, UserInput, Workspace,
};
use super::{
build_runtime, InteractiveRunController, InteractiveRuntime, InteractiveSessionController,
ProviderController, RuntimeBuildOptions,
};
use crate::{
agent::{AgentCapabilities, ToolCapability},
app::{
interactive_state::{state_after_event, InteractiveState},
policy::AppPolicy,
},
compaction::CompactionConfig,
config::Config,
diagnostics::RuntimeDiagnostics,
permission::PermissionMode,
session::Session as StoredSession,
tools::{
agent::BackgroundSubagents,
sdk_registry::{AppToolSet, DelegationConfig, ToolSetOptions},
},
};
#[tokio::test]
async fn configured_token_threshold_installs_sdk_automatic_compaction_policy() {
let provider = ScriptedProvider::new(
ModelIdentity::new("test", "test", "test"),
[
ScriptedTurn::completed(ModelResponse::Assistant(vec![ContentBlock::Text(
"compact summary".into(),
)])),
ScriptedTurn::completed(ModelResponse::Assistant(vec![ContentBlock::Text(
"done".into(),
)])),
],
);
let shared_provider: Arc<dyn ModelProvider> = Arc::new(provider.clone());
let tools = AppToolSet::disabled();
let workspace = Workspace::new(std::env::current_dir().unwrap()).unwrap();
let runtime = build_runtime(RuntimeBuildOptions {
provider: shared_provider,
tools: tools.tools(),
workspace,
workspace_policy: AppPolicy::for_mode(PermissionMode::Auto),
approval_handler: None,
system_prompt: SystemPrompt::None,
reasoning: rho_sdk::ReasoningLevel::Off,
compaction: CompactionConfig {
auto_compact: true,
threshold_percent: 1,
target_percent: 1,
},
context_window: Some(1_000),
usage_purpose: "agent",
usage_parent_session_id: None,
usage_recording: Default::default(),
})
.unwrap();
assert_eq!(runtime.diagnostics().compaction_trigger_tokens(), Some(10));
let session = runtime
.session(SessionOptions::new().history(vec![
rho_sdk::model::Message::user_text("x".repeat(2_000)),
rho_sdk::model::Message::assistant_text("y".repeat(2_000)),
]))
.await
.unwrap();
let mut run = session.start(UserInput::text("continue")).await.unwrap();
let mut events = Vec::new();
while let Some(event) = run.next_event().await {
events.push(event);
}
let outcome = run.outcome().await.unwrap();
assert_eq!(outcome.text(), "done");
assert!(events.iter().any(|event| matches!(
event,
RunEvent::CompactionStarted {
trigger: rho_sdk::CompactionTrigger::Automatic,
..
}
)));
assert!(events.iter().any(|event| matches!(
event,
RunEvent::CompactionCompleted {
trigger: rho_sdk::CompactionTrigger::Automatic,
..
}
)));
assert_eq!(provider.recorded_requests().len(), 2);
}
struct PendingCompactor;
impl Compactor for PendingCompactor {
fn compact<'a>(&'a self, _request: CompactionRequest) -> CompactionFuture<'a> {
Box::pin(std::future::pending::<
Result<CompactionOutput, rho_sdk::Error>,
>())
}
}
#[tokio::test]
async fn set_context_window_installs_automatic_compaction_when_idle() {
let mut interactive = pending_compaction_runtime("done").await;
interactive.compaction = CompactionConfig {
auto_compact: true,
threshold_percent: 1,
target_percent: 1,
};
assert_eq!(
interactive
.sessions
.session()
.diagnostics()
.compaction_trigger_tokens(),
None
);
interactive.set_context_window(Some(1_000));
assert_eq!(
interactive
.sessions
.session()
.diagnostics()
.compaction_trigger_tokens(),
Some(10)
);
}
#[tokio::test]
async fn replace_provider_rebuilds_compactor_with_current_context_window() {
let mut interactive = pending_compaction_runtime("done").await;
interactive.compaction = CompactionConfig {
auto_compact: true,
threshold_percent: 80,
target_percent: 50,
};
interactive.context_window = Some(2_000);
let replacement: Arc<dyn ModelProvider> = Arc::new(ScriptedProvider::new(
ModelIdentity::new("replacement", "test", "model"),
Vec::<ScriptedTurn>::new(),
));
interactive
.replace_provider(Arc::clone(&replacement), rho_sdk::ReasoningLevel::Low)
.unwrap();
assert_eq!(
interactive
.sessions
.session()
.diagnostics()
.compaction_trigger_tokens(),
Some(1_600)
);
assert_eq!(
interactive.sessions.session().diagnostics().provider(),
&ModelIdentity::new("replacement", "test", "model")
);
assert_eq!(
interactive.sessions.session().reasoning_level(),
rho_sdk::ReasoningLevel::Low
);
}
#[tokio::test]
async fn new_sessions_seed_prompt_cache_keys() {
let provider = Arc::new(ScriptedProvider::new(
ModelIdentity::new("test", "test", "test"),
Vec::<ScriptedTurn>::new(),
));
let tools = AppToolSet::disabled();
let workspace = Workspace::new(std::env::current_dir().unwrap()).unwrap();
let runtime = build_runtime(RuntimeBuildOptions {
provider: Arc::clone(&provider) as Arc<dyn ModelProvider>,
tools: tools.tools(),
workspace: workspace.clone(),
workspace_policy: AppPolicy::for_mode(PermissionMode::Auto),
approval_handler: None,
system_prompt: SystemPrompt::None,
reasoning: rho_sdk::ReasoningLevel::Off,
compaction: CompactionConfig::default(),
context_window: None,
usage_purpose: "agent",
usage_parent_session_id: None,
usage_recording: Default::default(),
})
.unwrap();
let id = SessionId::new();
let cache_key = format!("rho:{}", id.as_str());
let session = runtime
.session(
SessionOptions::new()
.id(id.clone())
.prompt_cache_key(cache_key.clone()),
)
.await
.unwrap();
assert_eq!(
session.snapshot().prompt_cache_key(),
Some(cache_key.as_str())
);
}
async fn test_runtime(turns: Vec<ScriptedTurn>) -> InteractiveRuntime {
let provider = Arc::new(ScriptedProvider::new(
ModelIdentity::new("test", "test", "test"),
turns,
));
let shared_provider: Arc<dyn ModelProvider> = provider;
let tools = AppToolSet::disabled();
let workspace = Workspace::new(std::env::current_dir().unwrap()).unwrap();
let runtime = rho_sdk::Rho::builder()
.provider_shared(Arc::clone(&shared_provider))
.compactor(PendingCompactor)
.build()
.unwrap();
let session = runtime.session(SessionOptions::default()).await.unwrap();
InteractiveRuntime {
runtime,
runs: InteractiveRunController::default(),
sessions: InteractiveSessionController::new(session, None),
provider: ProviderController::new(shared_provider, rho_sdk::ReasoningLevel::Off),
tools,
workspace,
system_prompt: SystemPrompt::None,
compaction: CompactionConfig::default(),
context_window: None,
usage_recording: Default::default(),
permission_mode: PermissionMode::Auto,
approval_handler: None,
approval_receiver: None,
agent_id: "default".into(),
agent_fingerprint: "test-fingerprint".into(),
pending_persistence_error: None,
pending_persistence_checkpoint: None,
}
}
async fn pending_compaction_runtime(response: &str) -> InteractiveRuntime {
test_runtime(vec![ScriptedTurn::completed(ModelResponse::Assistant(
vec![ContentBlock::Text(response.into())],
))])
.await
}
async fn permission_mode_runtime() -> InteractiveRuntime {
let mut interactive = pending_compaction_runtime("done").await;
let config = Config::default();
let capabilities = AgentCapabilities::new(
[ToolCapability::Agent, ToolCapability::Agents]
.into_iter()
.collect(),
);
interactive.tools = AppToolSet::new(
&config,
RuntimeDiagnostics::new(&config),
ToolSetOptions::new(capabilities).delegation(DelegationConfig::new(
std::env::current_dir().unwrap(),
std::path::PathBuf::new(),
BackgroundSubagents::Disabled,
)),
);
interactive
}
#[tokio::test]
async fn permission_mode_switch_rebuilds_runtime_and_updates_future_delegated_policy() {
let mut interactive = permission_mode_runtime().await;
interactive
.sessions
.session()
.append_message(Message::user_text("preserved history"))
.unwrap();
let session_id = interactive.sessions.session().id().clone();
let history = interactive.sessions.session().history();
interactive
.set_permission_mode(PermissionMode::Plan)
.await
.unwrap();
assert_eq!(interactive.permission_mode(), PermissionMode::Plan);
assert_eq!(
interactive
.tools
.subagents()
.unwrap()
.launch_permission_mode()
.decision_for(rho_sdk::CapabilityKind::Write),
PolicyDecision::Deny {
reason: "capability is not allowed in plan mode".into()
}
);
assert!(interactive.approval_handler.is_none());
assert!(interactive.approval_receiver().is_none());
assert_eq!(interactive.sessions.session().id(), &session_id);
assert_eq!(interactive.sessions.session().history(), history);
interactive
.set_permission_mode(PermissionMode::Supervised)
.await
.unwrap();
assert_eq!(interactive.permission_mode(), PermissionMode::Supervised);
assert_eq!(
interactive
.tools
.subagents()
.unwrap()
.launch_permission_mode()
.decision_for(rho_sdk::CapabilityKind::Write),
PolicyDecision::RequireApproval {
reason: "host approval is required".into()
}
);
assert!(interactive.approval_handler.is_some());
assert!(interactive.approval_receiver().is_some());
assert_eq!(interactive.sessions.session().id(), &session_id);
assert_eq!(interactive.sessions.session().history(), history);
let supervised_handler = interactive.approval_handler.clone().unwrap();
interactive
.set_permission_mode(PermissionMode::Supervised)
.await
.unwrap();
assert!(Arc::ptr_eq(
interactive.approval_handler.as_ref().unwrap(),
&supervised_handler
));
interactive
.set_permission_mode(PermissionMode::Auto)
.await
.unwrap();
assert_eq!(interactive.permission_mode(), PermissionMode::Auto);
assert!(interactive.approval_handler.is_none());
assert!(interactive.approval_receiver().is_none());
assert_eq!(interactive.sessions.session().id(), &session_id);
assert_eq!(interactive.sessions.session().history(), history);
}
#[tokio::test]
async fn permission_mode_switch_preserves_a_pending_new_session() {
let mut interactive = pending_compaction_runtime("done").await;
let previous_id = interactive.session_id().clone();
interactive.reset().unwrap();
let pending_id = interactive.session_id().clone();
assert_ne!(pending_id, previous_id);
interactive
.set_permission_mode(PermissionMode::Plan)
.await
.unwrap();
interactive
.set_permission_mode(PermissionMode::Auto)
.await
.unwrap();
assert_eq!(interactive.session_id(), &pending_id);
interactive
.start(UserInput::text("first new-session turn"), None)
.await
.unwrap();
while interactive.next_event().await.is_some() {}
interactive.finish_run().await.unwrap();
assert_eq!(interactive.session_id(), &pending_id);
assert_eq!(interactive.sessions.session().id(), &pending_id);
}
#[tokio::test]
async fn permission_mode_switch_rejects_an_active_run_without_mutation() {
let mut interactive = pending_compaction_runtime("done").await;
interactive
.start(UserInput::text("start"), None)
.await
.unwrap();
let error = interactive
.set_permission_mode(PermissionMode::Supervised)
.await
.unwrap_err();
assert!(error.to_string().contains("while a run is active"));
assert_eq!(interactive.permission_mode(), PermissionMode::Auto);
assert!(interactive.approval_receiver().is_none());
}
#[tokio::test]
async fn a_new_run_resets_the_context_usage_baseline() {
let mut interactive = pending_compaction_runtime("done").await;
interactive.context_window = Some(10_000);
interactive.observe_event(&RunEvent::UsageUpdated {
usage: ModelUsage {
input_tokens: Some(50_000),
..ModelUsage::default()
},
});
interactive.observe_event(&RunEvent::Started {
run_id: RunId::new(),
revision: Default::default(),
});
interactive.observe_event(&RunEvent::StepStarted { step: 1 });
interactive.observe_event(&RunEvent::UsageUpdated {
usage: ModelUsage {
input_tokens: Some(300),
cache_read_tokens: Some(700),
..ModelUsage::default()
},
});
assert_eq!(
interactive.take_context_usage(),
Some(rho_sdk::model::ContextUsage::provider_reported(
1_000,
Some(10_000)
))
);
}
#[tokio::test]
async fn finished_run_reports_context_from_committed_history() {
let mut interactive = pending_compaction_runtime("assistant output").await;
interactive.context_window = Some(10_000);
interactive
.start(UserInput::text("user input"), None)
.await
.unwrap();
while interactive.next_event().await.is_some() {}
interactive.finish_run().await.unwrap();
let expected_tokens = rho_sdk::model::context::estimate_context_tokens(
&interactive.history(),
&interactive.tools.specs(),
);
assert_eq!(
interactive.take_context_usage(),
Some(rho_sdk::model::ContextUsage::estimated(
expected_tokens,
Some(10_000)
))
);
}
#[tokio::test]
async fn dropping_manual_compaction_does_not_leave_the_runtime_busy() {
let mut interactive = pending_compaction_runtime("done").await;
let mut compact = Box::pin(interactive.compact());
tokio::select! {
result = &mut compact => panic!("compaction unexpectedly completed: {result:?}"),
() = tokio::task::yield_now() => {}
}
drop(compact);
interactive
.start(UserInput::text("continue"), None)
.await
.unwrap();
}
#[tokio::test]
async fn failed_turn_does_not_duplicate_the_previous_assistant_in_display_history() {
let mut interactive = test_runtime(vec![
ScriptedTurn::completed(ModelResponse::Assistant(vec![ContentBlock::Text(
"previous answer".into(),
)])),
ScriptedTurn::failed(ProviderError::new(
ProviderErrorKind::Unavailable,
"provider unavailable",
Retryability::Permanent,
)),
])
.await;
let root = tempfile::tempdir().unwrap();
let cwd = root.path().join("workspace");
std::fs::create_dir(&cwd).unwrap();
let storage = StoredSession::create_in_root(root.path(), &cwd).unwrap();
let sdk_session = interactive
.runtime
.session(SessionOptions::new().id(SessionId::from_string(storage.id()).unwrap()))
.await
.unwrap();
interactive.sessions.replace_session(sdk_session, None);
interactive.sessions.attach_storage(storage.clone());
interactive
.start(UserInput::text("successful prompt"), None)
.await
.unwrap();
while interactive.next_event().await.is_some() {}
interactive.finish_run().await.unwrap();
interactive
.start(UserInput::text("failed prompt"), None)
.await
.unwrap();
while interactive.next_event().await.is_some() {}
assert!(interactive.finish_run().await.is_err());
let committed_assistant = interactive.history()[1].clone();
let (_, histories) =
StoredSession::open_by_id_with_histories_in_root(root.path(), &cwd, storage.id()).unwrap();
assert_eq!(
histories.display,
vec![
Message::user_text("successful prompt"),
committed_assistant,
Message::user_text("failed prompt"),
]
);
}
#[tokio::test]
async fn failed_resume_preserves_the_current_runtime() {
let mut interactive = pending_compaction_runtime("still available").await;
let root = tempfile::tempdir().unwrap();
let cwd = root.path().join("workspace");
std::fs::create_dir(&cwd).unwrap();
let target = StoredSession::create_in_root(root.path(), &cwd).unwrap();
std::fs::write(
target.path(),
format!(
"{}\n",
serde_json::json!({
"type": "session",
"version": 999,
"id": target.id(),
"timestamp": "1",
"cwd": cwd,
})
),
)
.unwrap();
assert!(interactive.resume(target, Vec::new()).await.is_err());
interactive
.start(UserInput::text("continue"), None)
.await
.unwrap();
}
#[tokio::test]
async fn successful_sdk_completion_reaches_completed_state() {
let provider = ScriptedProvider::new(
ModelIdentity::new("test", "test", "test"),
[ScriptedTurn::completed(ModelResponse::Assistant(vec![
ContentBlock::Text("done".into()),
]))],
);
let runtime = rho_sdk::Rho::builder().provider(provider).build().unwrap();
let session = runtime.session(Default::default()).await.unwrap();
let mut run = session.start(rho_sdk::UserInput::text("go")).await.unwrap();
let mut state = InteractiveState::Idle;
while let Some(event) = run.next_event().await {
state = state_after_event(state, &event);
}
let outcome = run.outcome().await.unwrap();
assert_eq!(outcome.text(), "done");
assert_eq!(state, InteractiveState::Completed);
}