mod backend;
use std::collections::{BTreeMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex, OnceLock};
use rpi_ai::Provider;
use rpi_harness::agent_harness::AgentHarness;
use rpi_harness::context_files::{format_project_context, load_project_context_files};
use rpi_harness::session::memory::{InMemorySessionStorage, SystemClock};
use rpi_harness::session::session::DefaultIdGenerator;
use rpi_harness::session::types::SessionMetadata;
use rpi_harness::session::Session;
use rpi_harness::system_prompt::compose_system_prompt;
use rpi_harness::types::{
AgentHarnessOptions, AgentHarnessResources, AgentHarnessStreamOptions, DrivingMode,
HarnessTool, HarnessToolExecution, ToolReplay,
};
#[cfg(windows)]
use rpi_tools::create_powershell_tool;
use rpi_tools::{
create_bash_tool, create_edit_tool, create_read_tool, create_write_tool, ExecutionToolContext,
MutationQueueRegistry, OsExecutionEnv,
};
use crate::args::Args;
use crate::docs_tool::create_docs_tool;
use crate::provider::ResolvedModel;
use crate::resource_dirs::{
discover_append_system_prompt_file, discover_system_prompt_file, extension_dirs,
global_extension_dirs, global_prompt_template_dirs, global_skill_dirs,
load_prompt_templates_with_precedence, load_skills_with_precedence, project_extension_dirs,
project_prompt_template_dirs, project_skill_dirs, prompt_template_dirs, skill_dirs,
};
use crate::startup_diagnostics::{Phase, StartupSpan};
use rpi_extensions::{
emit_resources_discover, ExtensionEmitter, ExtensionSession, NullDiagnostics,
PluginDiagnostics, PluginToolAdapter, TeeEmitter, ToolPromptMetadata,
};
type FileIOAttachment = (
std::sync::Weak<dyn rpi_harness::session::types::SessionStorage>,
std::sync::Weak<rpi_tools::file_io::FileIORouter>,
);
fn file_io_attachments() -> &'static Mutex<Vec<FileIOAttachment>> {
static ATTACHMENTS: std::sync::OnceLock<Mutex<Vec<FileIOAttachment>>> =
std::sync::OnceLock::new();
ATTACHMENTS.get_or_init(|| Mutex::new(Vec::new()))
}
pub(crate) fn register_file_io_router(
harness: &AgentHarness,
router: &Arc<rpi_tools::file_io::FileIORouter>,
) {
let storage = harness.session().storage();
let mut attachments = file_io_attachments()
.lock()
.expect("file I/O attachments poisoned");
attachments.retain(|(s, r)| {
s.strong_count() > 0
&& r.strong_count() > 0
&& !s
.upgrade()
.is_some_and(|known| Arc::ptr_eq(&known, &storage))
});
attachments.push((Arc::downgrade(&storage), Arc::downgrade(router)));
}
pub(crate) fn file_io_router(
harness: &AgentHarness,
) -> Option<Arc<rpi_tools::file_io::FileIORouter>> {
let storage = harness.session().storage();
let mut attachments = file_io_attachments()
.lock()
.expect("file I/O attachments poisoned");
attachments.retain(|(s, r)| s.strong_count() > 0 && r.strong_count() > 0);
attachments.iter().find_map(|(s, r)| {
s.upgrade()
.filter(|known| Arc::ptr_eq(known, &storage))
.and_then(|_| r.upgrade())
})
}
pub const BUILTIN_TOOL_NAMES: &[&str] = &["read", "bash", "edit", "write", "docs"];
pub fn default_system_prompt(cwd: &str, prior_reasoning_replayed: bool) -> String {
let tools = BUILTIN_TOOL_NAMES
.iter()
.map(|name| (*name).to_owned())
.collect::<Vec<_>>();
default_system_prompt_with_tools(cwd, prior_reasoning_replayed, &tools)
}
pub fn default_system_prompt_with_tools(
cwd: &str,
prior_reasoning_replayed: bool,
tools: &[String],
) -> String {
default_system_prompt_with_tool_prompts(cwd, prior_reasoning_replayed, tools, &BTreeMap::new())
}
pub fn default_system_prompt_with_tool_prompts(
cwd: &str,
prior_reasoning_replayed: bool,
tools: &[String],
overrides: &BTreeMap<String, ToolPromptMetadata>,
) -> String {
let metadata: BTreeMap<_, _> = tools
.iter()
.map(|name| {
let prompt = overrides
.get(name)
.cloned()
.unwrap_or_else(|| ToolPromptMetadata {
prompt_snippet: builtin_tool_snippet(name).map(str::to_owned),
prompt_guidelines: Vec::new(),
})
.normalized();
(name.as_str(), prompt)
})
.collect();
let working_state_rule = if prior_reasoning_replayed {
"- Your reasoning is carried back to you on the next turn: do not re-derive it or restate
the plan. Continue from the last unfinished step, and when a step lands, say which one.
Re-stating the same plan without acting on it is a bug, not progress."
} else {
"- Your reasoning is not carried into your next turn; only your visible replies and tool
output come back. Before each batch of tool calls, write one short line naming the step you
are on and what remains, and say which step is done when it lands.
Re-stating the same plan without acting on it is a bug, not progress."
};
let tool_lines = tools
.iter()
.filter_map(|name| {
metadata
.get(name.as_str())?
.prompt_snippet
.as_ref()
.map(|snippet| format!("- {name:<5} — {snippet}"))
})
.collect::<Vec<_>>()
.join("\n");
let tool_lines = if tool_lines.is_empty() {
"(none)".to_string()
} else {
tool_lines
};
let guidelines = [
"- Reply in the same language as the user's latest message (the conversation's language when
that message is too short to tell), unless the user asks for another.",
"- Be concise in your responses",
"- Show file paths clearly when working with files",
"- Prefer the smallest change that solves the problem",
working_state_rule,
"- Read only what you need. If a file or search result was already shown earlier in this
conversation, use it instead of fetching it again.",
];
let mut seen = HashSet::new();
let tool_rules = tools
.iter()
.flat_map(|name| &metadata[name.as_str()].prompt_guidelines)
.map(String::as_str);
let core_rules = guidelines
.iter()
.map(|rule| rule.strip_prefix("- ").unwrap_or(rule));
let guidelines = tool_rules
.chain(core_rules)
.map(str::trim)
.filter(|rule| !rule.is_empty() && seen.insert(*rule))
.map(|rule| format!("- {rule}"))
.collect::<Vec<_>>()
.join("\n");
format!(
"You are an expert coding assistant operating inside rpi, a coding agent harness. \
You help users by reading files, executing commands, editing code, and writing new files.
Available tools:
{tool_lines}
In addition to the tools above, you may have access to other custom tools depending on the project.
Guidelines:
{guidelines}
Current working directory: {cwd}"
)
}
fn extension_tool_prompt_metadata(
session: &ExtensionSession,
) -> BTreeMap<String, ToolPromptMetadata> {
session
.snapshot()
.map(|snapshot| {
snapshot
.tools()
.iter()
.map(|tool| (tool.tool.name.clone(), tool.prompt.clone()))
.collect()
})
.unwrap_or_default()
}
fn builtin_tool_snippet(name: &str) -> Option<&'static str> {
Some(match name {
"read" => "Read file contents",
"bash" => "Execute Bash commands (Bash/POSIX syntax, also on Windows)",
"edit" => "Find/replace edits to existing files",
"write" => "Create or overwrite files",
"docs" => "Look up rpi commands, extensions, Pi package compatibility and .rpi configuration instead of guessing",
"grep" => "Search file contents for a pattern",
"find" => "Find files by glob pattern",
"ls" => "List directory contents",
"powershell" => "Execute PowerShell commands on Windows",
_ => return None,
})
}
#[derive(Debug, Clone)]
pub enum SessionSelection {
Ephemeral,
New { dir: PathBuf, name: Option<String> },
ExperimentalNative { dir: PathBuf },
Latest { dir: PathBuf },
ById { id: String, dir: PathBuf },
ByExactId { id: String, dir: PathBuf },
Fork { source: String, dir: PathBuf },
}
pub fn select_session(args: &Args, cwd: &Path) -> SessionSelection {
select_session_in_dir(args, configured_session_dir(args, cwd))
}
pub(crate) fn configured_session_dir(args: &Args, cwd: &Path) -> PathBuf {
if let Some(root) = &args.experimental_native_store {
return if root.is_absolute() {
root.clone()
} else {
cwd.join(root)
};
}
let dir = args
.session_dir
.clone()
.or_else(|| {
crate::settings::load_settings()
.ok()
.and_then(|settings| settings.session_dir)
})
.unwrap_or_else(|| default_session_dir(cwd));
if dir.is_relative() && !dir.has_root() {
cwd.join(dir)
} else {
dir
}
}
pub(crate) fn select_session_in_dir(args: &Args, dir: PathBuf) -> SessionSelection {
if args.experimental_native_store.is_some() {
return SessionSelection::ExperimentalNative { dir };
}
if args.no_session {
return SessionSelection::Ephemeral;
}
if args.continue_session || args.resume {
return SessionSelection::Latest { dir };
}
if let Some(s) = &args.fork {
return SessionSelection::Fork {
source: s.clone(),
dir,
};
}
if let Some(s) = &args.session_id {
return SessionSelection::ByExactId { id: s.clone(), dir };
}
if let Some(s) = &args.session {
return SessionSelection::ById { id: s.clone(), dir };
}
SessionSelection::New {
dir,
name: args.name.clone(),
}
}
pub fn default_session_dir(cwd: &Path) -> PathBuf {
std::env::var("RPI_CODING_AGENT_SESSION_DIR")
.ok()
.map(PathBuf::from)
.unwrap_or_else(|| cwd.join(".rpi").join("sessions"))
}
pub struct PreparedExtensions {
resource_leases: Arc<Vec<crate::resource_management::Lease>>,
resource_selection: Option<crate::resource_selection::ResolvedSelection>,
session_selection: SessionSelection,
transition_session: Option<Session>,
pub extension_session: ExtensionSession,
native_store: Option<PathBuf>,
session_dir: PathBuf,
runtime: tokio::runtime::Handle,
tool_context: rpi_extensions::ToolCallContext,
action_host: Arc<crate::extensions_actions::HarnessActionHost>,
harness_cell: Arc<std::sync::OnceLock<rpi_harness::agent_harness::WeakAgentHarness>>,
reload_mailbox: rpi_extensions::ReloadMailbox,
ui_dialog_mailbox: rpi_extensions::UiDialogMailbox,
ext_status_mailbox: rpi_extensions::ExtensionStatusMailbox,
editor_text_mailbox: rpi_extensions::EditorTextMailbox,
action_bridge: Arc<rpi_extensions::ActionBridge>,
}
pub(crate) fn preflight_extension_startup(
args: &Args,
cwd: &Path,
trusted: bool,
) -> Result<
(
Option<crate::resource_selection::ResolvedSelection>,
SessionSelection,
),
BuildError,
> {
let mut resources = crate::resource_selection::ResolvedSelection::load(args, cwd, trusted)
.map_err(BuildError::HarnessCreate)?;
let selection = select_session(args, cwd);
let pinned = std::thread::scope(|scope| {
scope
.spawn(|| {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| BuildError::SessionOpen(e.to_string()))?
.block_on(async {
if let Some(selected) = &mut resources {
let env: Arc<dyn rpi_tools::ExecutionEnv> =
Arc::new(OsExecutionEnv::with_cwd(cwd.to_path_buf()));
selected
.validate_skills(&env)
.await
.map_err(BuildError::HarnessCreate)?;
}
let cwd_str = cwd.to_string_lossy();
let target = match &selection {
SessionSelection::Fork { source, dir } => {
match session_metadata_at(Path::new(source)) {
Some(meta) => Some(meta),
None => find_session_metadata(source, &cwd_str, dir).await?,
}
}
_ => resolve_restore_target(&selection, &cwd_str).await?,
};
let Some(meta) = target else {
if resources.as_ref().is_some_and(|r| r.transition.is_some()) {
return Err(BuildError::SessionOpen("resource transition requires an existing native session".into()));
}
return Ok(selection.clone());
};
if resources.as_ref().is_some_and(|r| r.transition.is_some()) && matches!(selection, SessionSelection::Fork { .. }) {
return Err(BuildError::SessionOpen("resource transition cannot fork".into()));
}
use rpi_harness::session::jsonl::{
JsonlSessionRepo, JsonlSessionRepoOptions,
};
let dir = selection_dir(&selection);
let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
fs: Arc::new(OsExecutionEnv::with_cwd(cwd.to_path_buf())),
sessions_root: dir.to_string_lossy().into_owned(),
clock: Arc::new(SystemClock),
ids: Arc::new(DefaultIdGenerator::new()),
});
let binding_facts = StartupSpan::start(Phase::BindingFacts);
let facts = if resources.as_ref().is_some_and(|r| r.transition.is_some()) {
rpi_harness::session::jsonl::JsonlSessionStorage::read_resource_transition_facts(
Arc::new(OsExecutionEnv::with_cwd(cwd.to_path_buf())), &meta.path,
Arc::new(SystemClock), Arc::new(DefaultIdGenerator::new()),
).await
} else { repo.read_facts_by_jsonl_metadata(&meta).await }
.map_err(|e| BuildError::SessionOpen(e.to_string()))?;
crate::resource_selection::validate_session_binding(
&facts.values,
resources.as_ref(),
)
.map_err(BuildError::SessionOpen)?;
binding_facts.finish();
Ok(match selection {
SessionSelection::Fork { .. } => SessionSelection::Fork {
source: meta.path,
dir,
},
_ => SessionSelection::ByExactId { id: meta.id, dir },
})
})
})
.join()
.map_err(|_| BuildError::SessionOpen("resource preflight reader panicked".into()))?
})?;
Ok((resources, pinned))
}
pub fn prepare_extensions(
args: &Args,
cwd: &Path,
project_trusted: bool,
) -> Result<PreparedExtensions, BuildError> {
let native_store =
crate::native_store::root(args, cwd).map_err(BuildError::NativeStorePolicy)?;
let resource_leases =
crate::resource_management::host_leases(cwd).map_err(BuildError::HarnessCreate)?;
let (resource_selection, session_selection) =
preflight_extension_startup(args, cwd, project_trusted)?;
let transition_session = if resource_selection
.as_ref()
.is_some_and(|r| r.transition.is_some())
{
Some(std::thread::scope(|scope| {
scope.spawn(|| {
tokio::runtime::Builder::new_current_thread().enable_all().build()
.map_err(|e| BuildError::SessionOpen(e.to_string()))?.block_on(async {
let meta = resolve_restore_target(&session_selection, &cwd.to_string_lossy()).await?
.ok_or_else(|| BuildError::SessionOpen("transition session disappeared".into()))?;
let path = std::fs::canonicalize(&meta.path).map_err(|e| BuildError::SessionOpen(e.to_string()))?;
claim_session_lock_inner(&path.to_string_lossy(), false).map_err(BuildError::SessionOpen)?;
let storage = rpi_harness::session::jsonl::JsonlSessionStorage::load_for_resource_transition(
Arc::new(OsExecutionEnv::with_cwd(cwd.to_path_buf())), &path.to_string_lossy(),
Arc::new(SystemClock), Arc::new(DefaultIdGenerator::new()),
).await.map_err(|e| BuildError::SessionOpen(e.to_string()))?;
let session = Session::new(Arc::new(storage), None);
crate::resource_selection::bind_session(&session, resource_selection.as_ref()).await.map_err(BuildError::SessionOpen)?;
Ok::<_, BuildError>(session)
})
}).join().map_err(|_| BuildError::SessionOpen("transition writer preparation panicked".into()))?
})?)
} else {
None
};
let session_dir = if matches!(session_selection, SessionSelection::Ephemeral) {
configured_session_dir(args, cwd)
} else {
selection_dir(&session_selection)
};
let cwd_str = cwd.to_string_lossy().to_string();
let tool_context = rpi_extensions::ToolCallContext::new(cwd_str.clone());
if !project_trusted && args.verbose {
eprintln!(
"warning: current-project settings, resources, and discovered extensions are explicitly disabled (use --approve or /trust yes to re-enable)"
);
}
let runtime = tokio::runtime::Handle::try_current().map_err(|e| {
BuildError::HarnessCreate(format!("no tokio runtime for action bridge: {e}"))
})?;
let catalog = Vec::new();
let (action_host, harness_cell) = crate::extensions_actions::HarnessActionHost::new_empty(
catalog.clone(),
cwd.to_path_buf(),
runtime.clone(),
args.unknown_flags.clone(),
tool_context.clone(),
);
let action_host = Arc::new(
action_host
.with_session_dir(session_dir.clone())
.with_fresh_session_only(native_store.is_some()),
);
let host_arc: Arc<dyn rpi_extensions::RuntimeActionHost> = action_host.clone();
let reload_mailbox = rpi_extensions::ReloadMailbox::new();
let ui_dialog_mailbox = rpi_extensions::UiDialogMailbox::new();
let ext_status_mailbox = rpi_extensions::ExtensionStatusMailbox::new();
let editor_text_mailbox = rpi_extensions::EditorTextMailbox::new();
let action_bridge = rpi_extensions::ActionBridge::with_reload_and_ui(
runtime.clone(),
host_arc,
rpi_extensions::reload_callback_from_mailbox(reload_mailbox.clone()),
ui_dialog_mailbox.clone(),
ext_status_mailbox.clone(),
editor_text_mailbox.clone(),
);
let extension_register = StartupSpan::start(Phase::ExtensionRegister);
let extension_session = if let Some(selected) = &resource_selection {
let session = rpi_extensions::load_session_mixed_guarded(
&[],
&selected.extensions,
Arc::new(NullDiagnostics),
Some(Arc::clone(&action_bridge)),
Some(resource_leases.clone()),
);
if session.loaded_paths() != selected.extensions.as_slice() {
return Err(BuildError::HarnessCreate(
"selected native extension failed to load/register".into(),
));
}
if session.owns_stdio() {
return Err(BuildError::HarnessCreate(
"stdio-owning extensions are unsupported with --resource-config".into(),
));
}
session
} else if args.no_extensions {
ExtensionSession::none()
} else {
load_extensions(args, cwd, project_trusted, Some(Arc::clone(&action_bridge)))
};
extension_register.finish();
if args.verbose {
if let Some(s) = extension_session.summary() {
eprintln!("extensions: {s}");
}
report_deferred_renderers(&extension_session);
}
crate::native_store::validate_stdio(native_store.is_some(), extension_session.owns_stdio())
.map_err(BuildError::NativeStorePolicy)?;
action_host.publish_registry_impl(extension_session.snapshot_arc());
Ok(PreparedExtensions {
resource_leases,
resource_selection,
session_selection,
transition_session,
native_store,
session_dir,
runtime,
tool_context,
action_host,
harness_cell,
reload_mailbox,
ui_dialog_mailbox,
ext_status_mailbox,
editor_text_mailbox,
action_bridge,
extension_session,
})
}
pub async fn build(
resolved: &ResolvedModel,
args: &Args,
cwd: &Path,
project_trusted: bool,
) -> Result<
(
AgentHarness,
tokio::sync::broadcast::Receiver<rpi_agent::AgentEvent>,
ReloadContext,
),
BuildError,
> {
let prepared = prepare_extensions(args, cwd, project_trusted)?;
build_prepared(resolved, args, cwd, project_trusted, prepared).await
}
pub async fn build_prepared(
resolved: &ResolvedModel,
args: &Args,
cwd: &Path,
project_trusted: bool,
prepared: PreparedExtensions,
) -> Result<
(
AgentHarness,
tokio::sync::broadcast::Receiver<rpi_agent::AgentEvent>,
ReloadContext,
),
BuildError,
> {
let native_store =
crate::native_store::root(args, cwd).map_err(BuildError::NativeStorePolicy)?;
if native_store != prepared.native_store {
return Err(BuildError::NativeStorePolicy(
"experimental session selection changed after extension preparation".into(),
));
}
crate::native_store::validate_stdio(
native_store.is_some(),
prepared.extension_session.owns_stdio(),
)
.map_err(BuildError::NativeStorePolicy)?;
let fresh_session_only = native_store.is_some();
let cwd_str = cwd.to_string_lossy().to_string();
let PreparedExtensions {
resource_leases,
resource_selection,
session_selection: selection,
transition_session,
native_store: _,
session_dir,
runtime,
tool_context,
action_host,
harness_cell,
reload_mailbox,
ui_dialog_mailbox,
ext_status_mailbox,
editor_text_mailbox,
action_bridge,
extension_session,
} = prepared;
let catalog = crate::provider::authenticated_catalog(resolved);
action_host.set_catalog(catalog.clone());
let env = Arc::new(OsExecutionEnv::with_cwd(cwd.to_path_buf()));
let env_dyn: Arc<dyn rpi_tools::ExecutionEnv> = env.clone();
let mut_env: Arc<dyn rpi_tools::MutatingEnv> = env.clone();
let _registry = Arc::new(MutationQueueRegistry::new());
let file_io = Arc::new(rpi_tools::file_io::FileIORouter::new());
let ctx =
ExecutionToolContext::new(env_dyn.clone(), Some(mut_env)).with_file_io(file_io.clone());
let mut tools = build_tools(&ctx, args);
merge_extension_tools(&mut tools, &extension_session, args, &tool_context);
let active = if let Some(selected) = &resource_selection {
selected
.active_tools(
&tools
.iter()
.map(|t| t.tool.schema().name.clone())
.collect::<Vec<_>>(),
)
.map_err(BuildError::HarnessCreate)?
} else {
filter_active_tool_names(active_tool_names(&tools, args), args)
};
let session = match transition_session {
Some(session) => session,
None => build_session(&selection, &cwd_str, resource_selection.as_ref()).await?,
};
crate::resource_selection::bind_session(&session, resource_selection.as_ref())
.await
.map_err(BuildError::SessionOpen)?;
if let Ok(metadata) = session.get_metadata().await {
publish_session_identity(&tool_context, &metadata.id);
}
if let Some(requested) = args.name.as_deref() {
let name = requested.trim();
let name = if name.is_empty() { None } else { Some(name) };
if let Err(error) = session.set_name(name).await {
if fresh_session_only {
return Err(BuildError::SessionOpen(format!(
"native session name persistence failed: {error}"
)));
}
if args.verbose {
eprintln!("warning: could not set session name: {error}");
}
}
}
let prior_reasoning_replayed = rpi_ai::model::prior_reasoning_is_replayed(&resolved.model);
let prompt_tools = active.clone();
let prompt_metadata = extension_tool_prompt_metadata(&extension_session);
let default_prompt = || {
default_system_prompt_with_tool_prompts(
&cwd_str,
prior_reasoning_replayed,
&prompt_tools,
&prompt_metadata,
)
};
let mut context_paths = Vec::new();
let base_prompt = match args.system_prompt.as_deref() {
Some(explicit) => explicit.to_string(),
None if project_trusted && resource_selection.is_none() => {
match discover_system_prompt_file(cwd) {
Some(path) => {
read_prompt_source(&path, &mut context_paths).unwrap_or_else(default_prompt)
}
None => default_prompt(),
}
}
None => default_prompt(),
};
let mut append_texts: Vec<String> = Vec::new();
for extra in &args.append_system_prompt {
let text = read_append_target(extra, &mut context_paths).unwrap_or_else(|| extra.clone());
append_texts.push(text);
}
if let Some(selected) = &resource_selection {
append_texts.extend(
selected
.instructions
.iter()
.map(|(_, _, text)| text.clone()),
);
}
if args.append_system_prompt.is_empty() && resource_selection.is_none() {
if let Some(path) = project_trusted
.then(|| discover_append_system_prompt_file(cwd))
.flatten()
{
if let Some(text) = read_prompt_source(&path, &mut context_paths) {
append_texts.push(text);
}
}
}
let append_join = if append_texts.is_empty() {
None
} else {
Some(append_texts.join("\n\n"))
};
let agent_dir = crate::config::agent_dir().ok();
let discovered = extension_session
.snapshot_arc()
.filter(|_| resource_selection.is_none())
.map(|snap| emit_resources_discover(&cwd_str, "startup", &snap))
.unwrap_or_default();
let mut skills: Vec<rpi_harness::types::Skill> = Vec::new();
let mut skill_diags: Vec<rpi_harness::skills::SkillDiagnostic> = Vec::new();
let mut resource_diags: Vec<rpi_harness::diagnostics::ResourceDiagnostic> = Vec::new();
if let Some(selected) = &resource_selection {
skills = selected
.validated_skills()
.map_err(BuildError::HarnessCreate)?
.skills
.clone();
} else if !args.no_skills {
let mut dirs = if args.dev_local_only || args.project_only {
project_skill_dirs(cwd)
} else if project_trusted {
skill_dirs(cwd)
} else {
global_skill_dirs()
};
dirs.extend(args.skill.iter().cloned());
dirs.extend(discovered.skill_paths.iter().map(PathBuf::from));
let result = load_skills_with_precedence(&env_dyn, &dirs).await;
skills = result.skills;
skill_diags = result.diagnostics;
resource_diags.extend(result.resource_diagnostics);
}
let mut prompt_templates: Vec<rpi_harness::types::PromptTemplate> = Vec::new();
let mut prompt_diags: Vec<rpi_harness::prompt_templates::PromptTemplateDiagnostic> = Vec::new();
if !args.no_prompt_templates && resource_selection.is_none() {
let mut dirs = if args.dev_local_only || args.project_only {
project_prompt_template_dirs(cwd)
} else if project_trusted {
prompt_template_dirs(cwd)
} else {
global_prompt_template_dirs()
};
dirs.extend(args.prompt_template.iter().cloned());
dirs.extend(discovered.prompt_paths.iter().map(PathBuf::from));
let result = load_prompt_templates_with_precedence(&env_dyn, &dirs).await;
prompt_templates = result.prompt_templates;
prompt_diags = result.diagnostics;
resource_diags.extend(result.resource_diagnostics);
}
let context_block = if args.no_context_files || !project_trusted || resource_selection.is_some()
{
String::new()
} else {
let agent_dir_path = agent_dir.clone().unwrap_or_else(|| cwd.to_path_buf());
let files = load_project_context_files(&env_dyn, cwd, &agent_dir_path).await;
context_paths.extend(
files
.iter()
.map(|file| file.path.to_string_lossy().into_owned()),
);
format_project_context(&files)
};
if args.verbose {
for d in &skill_diags {
eprintln!(
"warning: skill {} ({}): {}",
d.path,
d.code.as_str(),
d.message
);
}
for d in &prompt_diags {
eprintln!(
"warning: prompt template {} ({}): {}",
d.path,
d.code.as_str(),
d.message
);
}
for d in &resource_diags {
if let rpi_harness::diagnostics::ResourceDiagnostic::Collision(c) = d {
eprintln!(
"warning: {} name \"{}\" collision: keeping {} (shadowed {})",
c.resource_type.as_str(),
c.name,
c.winner_path,
c.loser_path
);
}
}
}
let system_prompt = compose_system_prompt(
Some(&base_prompt),
&[], if context_block.is_empty() {
None
} else {
Some(&context_block)
},
append_join.as_deref(),
);
if args.debug_system_prompt {
eprintln!("=== --debug-system-prompt ===");
let base_src = if args.system_prompt.is_some() {
"--system-prompt"
} else if discover_system_prompt_file(cwd).is_some() {
"SYSTEM.md"
} else {
"default"
};
eprintln!("[base source: {base_src}]");
eprintln!("--- base ---\n{base_prompt}");
if let Some(append) = append_join.as_deref() {
eprintln!("--- append ---\n{append}");
} else {
eprintln!("--- append: (none) ---");
}
if context_block.is_empty() {
eprintln!("--- context: (none) ---");
} else {
eprintln!("--- context ---{context_block}");
}
let visible_skills = skills
.iter()
.filter(|s| s.disable_model_invocation != Some(true))
.count();
eprintln!(
"--- skills: {} loaded ({} model-visible, {} hidden) ---",
skills.len(),
visible_skills,
skills.len() - visible_skills
);
for s in &skills {
let hidden = if s.disable_model_invocation == Some(true) {
" [hidden]"
} else {
""
};
eprintln!(" {}{hidden} — {}", s.name, s.description);
}
eprintln!("--- prompt templates: {} ---", prompt_templates.len());
for t in &prompt_templates {
eprintln!(" /{}", t.name);
}
eprintln!(
"--- discovered via resources_discover: {} skill(s), {} prompt(s), {} theme(s) ---",
discovered.skill_paths.len(),
discovered.prompt_paths.len(),
discovered.theme_paths.len(),
);
for p in &discovered.skill_paths {
eprintln!(" skill: {p}");
}
for p in &discovered.prompt_paths {
eprintln!(" prompt: {p}");
}
eprintln!(
"--- final composed base+append+context (skills listing added by harness) ---\n{system_prompt}"
);
eprintln!("=== end --debug-system-prompt ===");
}
let (broadcast, event_rx) = rpi_agent::events::BroadcastEmitter::new(256);
let broadcast_emitter: Arc<dyn rpi_agent::AgentEmitter> = Arc::new(broadcast);
let broadcast_for_context: Arc<dyn rpi_agent::AgentEmitter> = Arc::clone(&broadcast_emitter);
let emitter: Arc<dyn rpi_agent::AgentEmitter> = match extension_session.snapshot_arc() {
Some(snapshot) => {
let ext = ExtensionEmitter::new(snapshot, extension_session.keepalive());
Arc::new(TeeEmitter::new(vec![broadcast_emitter, Arc::new(ext)]))
}
None => broadcast_emitter,
};
let saved_settings = crate::settings::load_settings().unwrap_or_default();
if let Some(proxy) = &saved_settings.http_proxy {
for key in ["HTTP_PROXY", "HTTPS_PROXY"] {
if std::env::var_os(key).is_none() {
std::env::set_var(key, proxy);
}
}
}
let extension_session_cell = Arc::new(Mutex::new(extension_session.clone()));
let before_tool_call = extension_session
.snapshot()
.filter(|snapshot| {
!snapshot
.handlers_for(rpi_plugin_sdk::EventTag::ToolCall)
.is_empty()
})
.map(|_| {
extension_tool_call_hook(
Arc::clone(&extension_session_cell),
ui_dialog_mailbox.clone(),
)
});
let options = AgentHarnessOptions {
model: resolved.model.clone(),
thinking_level: resolved.thinking_level,
active_tool_names: active.clone(),
tools,
system_prompt: Some(system_prompt),
resources: AgentHarnessResources {
context_files: context_paths,
skills: if skills.is_empty() {
None
} else {
Some(skills)
},
prompt_templates: if prompt_templates.is_empty() {
None
} else {
Some(prompt_templates)
},
},
allow_existing_session: matches!(
selection,
SessionSelection::Latest { .. }
| SessionSelection::ById { .. }
| SessionSelection::ByExactId { .. }
| SessionSelection::Fork { .. }
),
stream_options: AgentHarnessStreamOptions {
timeout: args.timeout,
stream_idle_timeout: args.stream_idle_timeout.or_else(|| {
saved_settings
.stream_idle_timeout
.map(std::time::Duration::from_secs)
}),
thinking_budgets: saved_settings.thinking_budgets,
..Default::default()
},
retry: saved_settings.retry_policy(),
compaction: saved_settings.compaction_settings(),
steering_mode: saved_settings.steering_mode(),
follow_up_mode: saved_settings.follow_up_mode(),
tool_execution: HarnessToolExecution::default(),
drive: DrivingMode::default(),
session,
models: build_models_with_extensions(resolved, &extension_session, runtime.clone()),
to_provider_messages: None,
entry_projectors: Default::default(),
agent_emitter: Some(emitter),
before_tool_call,
after_tool_call: None,
transform_context: None,
entry_transforms: Vec::new(),
provider_hooks: rpi_extensions::ExtensionProviderHooks::from_session(&extension_session)
.map(|h| Arc::new(h) as Arc<dyn rpi_ai::ProviderHooks>),
};
let harness = AgentHarness::create(options)
.await
.map_err(|e| BuildError::HarnessCreate(e.to_string()))?;
harness.set_skill_execution_env(env_dyn.clone());
let harness_leases = resource_leases.clone();
harness
.events()
.on_lifetime::<rpi_harness::events::RunEndEvent, _>(move |_| {
let _ = &harness_leases;
});
install_extension_settlement_bridge(harness.events(), extension_session_cell.clone());
if let Some(selected) = &resource_selection {
harness
.seal_resources(active)
.map_err(|e| BuildError::HarnessCreate(e.to_string()))?;
crate::resource_selection::attach(
&harness,
serde_json::json!({
"version": 1,
"hostId": uuid::Uuid::new_v4().to_string(),
"host": {"name": crate::APP_NAME, "version": crate::VERSION, "pid": std::process::id()},
"selection": selected.selection,
"package": selected.package,
"selectedSkills": selected.skills,
"skillDiagnostics": [],
"loadedExtensions": extension_session.loaded_paths().iter().map(|p| serde_json::json!({"identity": p.file_stem().and_then(|s| s.to_str()), "path": p})).collect::<Vec<_>>(),
"instructions": selected.provenance(),
"baseSystemPrompt": "native-default",
"projectContext": "disabled",
"resourceDiscovery": "disabled",
"changes": "restart-required"
}),
);
let resource_observation = StartupSpan::start(Phase::ResourceObservation);
crate::resource_selection::get_resource_state(&harness)
.await
.map_err(BuildError::HarnessCreate)?;
resource_observation.finish();
if selected.transition.is_some() {
let meta = resolve_restore_target(&selection, &cwd_str)
.await?
.ok_or_else(|| {
BuildError::SessionOpen("transition session disappeared before commit".into())
})?;
let path = std::fs::canonicalize(&meta.path)
.map_err(|e| BuildError::SessionOpen(e.to_string()))?;
let lock_path = PathBuf::from(format!("{}.lock", path.display()));
if !SESSION_LOCK_PATHS
.get_or_init(|| Mutex::new(HashSet::new()))
.lock()
.unwrap()
.contains(&lock_path)
{
return Err(BuildError::SessionOpen(
"resource transition lost canonical writer ownership".into(),
));
}
let facts =
rpi_harness::session::jsonl::JsonlSessionStorage::read_resource_transition_facts(
Arc::new(OsExecutionEnv::with_cwd(cwd.to_path_buf())),
&path.to_string_lossy(),
Arc::new(SystemClock),
Arc::new(DefaultIdGenerator::new()),
)
.await
.map_err(|e| BuildError::SessionOpen(e.to_string()))?;
crate::resource_selection::validate_session_binding(&facts.values, Some(selected))
.map_err(BuildError::SessionOpen)?;
let mirror = rpi_harness::session::SessionValues::load(harness.session())
.await
.map_err(|e| BuildError::SessionOpen(e.to_string()))?;
if crate::resource_selection::binding_state(&mirror)
!= crate::resource_selection::binding_state(&facts.values)
{
return Err(BuildError::SessionOpen(
"resource binding changed before commit".into(),
));
}
}
let inventory_bind = StartupSpan::start(Phase::InventoryBind);
crate::resource_selection::bind_inventory(&harness, selected)
.await
.map_err(BuildError::SessionOpen)?;
inventory_bind.finish();
}
crate::extensions_actions::HarnessActionHost::set_harness(
&harness_cell,
Arc::new(harness.clone()),
);
register_file_io_router(&harness, &file_io);
{
let runtime = rpi_harness::runtime::SessionRuntime::new(harness.session().clone());
match rpi_harness::runtime::recover_all_lanes(&runtime).await {
Ok(reports) => {
for (lane, report) in reports {
if report.is_clean() {
continue;
}
if report.is_corrupt() {
eprintln!(
"warning: session lane '{lane}' is corrupt ({} open operations); \
refusing to auto-repair — inspect the session file",
report.open_operations.len()
);
continue;
}
for decision in runtime.reconcile(&report) {
if let rpi_harness::runtime::RecoveryDecision::Abort { run_id } = decision {
if args.verbose {
eprintln!(
"recovered interrupted run {run_id} on lane '{lane}' (aborted)"
);
}
}
}
runtime.emit_recovery_events(harness.events(), &report);
}
}
Err(e) => {
if args.verbose {
eprintln!("warning: session recovery sweep failed: {e}");
}
}
}
}
let mut reload_args = args.clone();
reload_args.session_dir = Some(session_dir);
let reload_context = ReloadContext {
resource_leases,
fresh_session_only,
extension_session: extension_session_cell,
action_bridge: Arc::new(Mutex::new(Some(Arc::clone(&action_bridge)))),
tool_context,
file_io,
catalog: Arc::new(std::sync::RwLock::new(catalog)),
providers: Arc::new(std::sync::RwLock::new(resolved.providers.clone())),
runtime: runtime.clone(),
cwd: cwd.to_path_buf(),
project_trusted,
discovered_theme_paths: Arc::new(std::sync::Mutex::new(discovered.theme_paths.clone())),
args: reload_args,
resolved_model: resolved.model.clone(),
broadcast: broadcast_for_context,
mailbox: reload_mailbox,
ui_dialog: ui_dialog_mailbox,
ext_status: ext_status_mailbox,
editor_text: editor_text_mailbox,
dev_extension: None,
startup_notices: Vec::new(),
};
Ok((harness, event_rx, reload_context))
}
fn install_extension_settlement_bridge(
events: &rpi_harness::events::HarnessEventBus,
sessions: ExtensionSessionCell,
) {
events.on_lifetime::<rpi_harness::events::RunEndEvent, _>(move |event| {
if event.lane != "main" {
return;
}
let session = sessions.lock().unwrap().clone();
if let Some(snapshot) = session.snapshot_arc() {
rpi_extensions::dispatch_empty_event(&snapshot, rpi_plugin_sdk::EventTag::AgentSettled);
}
});
}
fn extension_tool_call_hook(
sessions: ExtensionSessionCell,
ui_dialog: rpi_extensions::UiDialogMailbox,
) -> rpi_agent::BeforeToolCall {
Arc::new(move |ctx: rpi_agent::BeforeToolCallContext<'_>, signal| {
let session = sessions
.lock()
.map(|session| session.clone())
.map_err(|_| ());
let ui_dialog = ui_dialog.clone();
let tool_call_id = ctx.tool_call.id.clone();
let tool_name = ctx.tool_call.name.clone();
let params = ctx.args.clone();
Box::pin(async move {
let reason = match session {
Ok(session) => {
rpi_extensions::dispatch_tool_call_event(
&session,
&tool_call_id,
&tool_name,
¶ms,
&signal,
&ui_dialog,
)
.await
}
Err(_) => Some("extension approval session is unavailable".to_string()),
};
reason.map(|reason| rpi_agent::BeforeToolCallResult {
block: true,
reason: Some(reason),
terminate: false,
args: None,
})
})
})
}
pub type ExtensionSessionCell = Arc<Mutex<ExtensionSession>>;
pub type ActionBridgeCell = Arc<Mutex<Option<Arc<rpi_extensions::ActionBridge>>>>;
#[derive(Clone)]
pub struct ReloadContext {
resource_leases: Arc<Vec<crate::resource_management::Lease>>,
fresh_session_only: bool,
pub file_io: Arc<rpi_tools::file_io::FileIORouter>,
pub extension_session: ExtensionSessionCell,
pub tool_context: rpi_extensions::ToolCallContext,
pub action_bridge: ActionBridgeCell,
pub ext_status: rpi_extensions::ExtensionStatusMailbox,
pub editor_text: rpi_extensions::EditorTextMailbox,
pub catalog: Arc<std::sync::RwLock<Vec<rpi_ai::Model>>>,
pub providers: Arc<std::sync::RwLock<Vec<Arc<dyn Provider>>>>,
pub runtime: tokio::runtime::Handle,
pub cwd: PathBuf,
pub project_trusted: bool,
pub discovered_theme_paths: Arc<std::sync::Mutex<Vec<String>>>,
pub args: Args,
pub resolved_model: rpi_ai::Model,
pub broadcast: Arc<dyn rpi_agent::AgentEmitter>,
pub mailbox: rpi_extensions::ReloadMailbox,
pub ui_dialog: rpi_extensions::UiDialogMailbox,
pub dev_extension: Option<Arc<crate::dev_extension::DevExtension>>,
pub startup_notices: Vec<String>,
}
impl ReloadContext {
pub(crate) fn session_dir(&self) -> PathBuf {
self.args
.session_dir
.clone()
.unwrap_or_else(|| configured_session_dir(&self.args, &self.cwd))
}
}
pub async fn dispatch_session_event_async(
reload_context: &ReloadContext,
tag: rpi_plugin_sdk::EventTag,
) -> Option<String> {
let (snapshot, keepalive) = {
let session = reload_context.extension_session.lock().unwrap();
(session.snapshot_arc(), session.keepalive())
};
match snapshot {
Some(snapshot) => rpi_extensions::dispatch_lifecycle_event(&snapshot, tag, keepalive).await,
None => None,
}
}
async fn shutdown_extensions_for_reload(
snapshot: Option<Arc<rpi_extensions::RegistrySnapshot>>,
keepalive: Arc<rpi_extensions::PluginKeepalive>,
) -> Result<(), String> {
if let Some(snapshot) = snapshot {
tokio::task::spawn_blocking(move || {
let _keepalive = keepalive;
rpi_extensions::dispatch_empty_event(
&snapshot,
rpi_plugin_sdk::EventTag::SessionShutdown,
);
})
.await
.map_err(|error| error.to_string())?;
}
Ok(())
}
pub struct ReloadOutcome {
pub summary: String,
pub had_warnings: bool,
pub details: Vec<String>,
pub had_errors: bool,
pub project_trusted: Option<bool>,
}
struct PreparedReloadInputs {
extension_dirs: Vec<PathBuf>,
skill_base_dirs: Vec<PathBuf>,
prompt_base_dirs: Vec<PathBuf>,
}
fn append_reload_resource_paths(
mut paths: Vec<PathBuf>,
explicit: &[PathBuf],
discovered: &[String],
js_paths: &[PathBuf],
package_paths: &[PathBuf],
) -> Vec<PathBuf> {
paths.extend(explicit.iter().cloned());
paths.extend(discovered.iter().map(PathBuf::from));
paths.extend(js_paths.iter().cloned());
paths.extend(package_paths.iter().cloned());
paths
}
pub async fn reload_extension_resources(
harness: &AgentHarness,
ctx: &ReloadContext,
) -> ReloadOutcome {
reload_extension_resources_inner(harness, ctx, || {}).await
}
async fn reload_extension_resources_inner<F>(
harness: &AgentHarness,
ctx: &ReloadContext,
after_prepare: F,
) -> ReloadOutcome
where
F: FnOnce() + Send,
{
if harness.has_exact_resources()
|| ctx.args.resource_config.is_some()
|| ctx.args.resource_transition.is_some()
{
return ReloadOutcome {
summary:
"Exact resource selection is restart-required; reload/dev hot apply unsupported"
.into(),
had_warnings: true,
details: Vec::new(),
had_errors: true,
project_trusted: None,
};
}
let mut effective_args = ctx.args.clone();
if let Some(dev) = &ctx.dev_extension {
if let Err(error) = dev
.rebuild()
.and_then(|_| dev.apply_to_args(&mut effective_args))
{
return ReloadOutcome {
summary: format!(
"Extension build failed for {}: {error}. Keeping the currently loaded version.",
dev.package_name()
),
had_warnings: true,
details: Vec::new(),
had_errors: true,
project_trusted: None,
};
}
}
let cwd_str = ctx.cwd.to_string_lossy().to_string();
let project_trusted = resolve_project_trust(&effective_args, &ctx.cwd);
let prepared = match prepare_reload_inputs(&effective_args, &ctx.cwd, project_trusted) {
Ok(prepared) => prepared,
Err(error) => {
return ReloadOutcome {
summary: format!(
"Settings reload failed: {error}. Keeping the currently loaded resources."
),
had_warnings: true,
details: Vec::new(),
had_errors: true,
project_trusted: None,
};
}
};
after_prepare();
let PreparedReloadInputs {
extension_dirs,
skill_base_dirs,
prompt_base_dirs,
} = prepared;
let mut warnings = false;
let (shutdown_snapshot, shutdown_keepalive) = {
let session = ctx.extension_session.lock().unwrap();
(session.snapshot_arc(), session.keepalive())
};
if let Err(reason) = shutdown_extensions_for_reload(shutdown_snapshot, shutdown_keepalive).await
{
return ReloadOutcome {
summary: "Reload failed during extension shutdown".into(),
had_warnings: true,
details: vec![reason],
had_errors: true,
project_trusted: None,
};
}
let old_bridge = ctx.action_bridge.lock().unwrap().clone();
let host: Arc<dyn rpi_extensions::RuntimeActionHost> = match &old_bridge {
Some(b) => b.clone_host(),
None => {
let (action_host, _cell) = crate::extensions_actions::HarnessActionHost::new_empty(
ctx.catalog
.read()
.unwrap_or_else(|error| error.into_inner())
.clone(),
ctx.cwd.clone(),
ctx.runtime.clone(),
ctx.args.unknown_flags.clone(),
ctx.tool_context.clone(),
);
crate::extensions_actions::HarnessActionHost::set_harness(
&_cell,
Arc::new(harness.clone()),
);
Arc::new(
action_host
.with_session_dir(ctx.session_dir())
.with_fresh_session_only(ctx.fresh_session_only),
)
}
};
let reload_cb = rpi_extensions::reload_callback_from_mailbox(ctx.mailbox.clone());
let publish_host = Arc::clone(&host);
let fresh_bridge = rpi_extensions::ActionBridge::with_reload_and_ui(
ctx.runtime.clone(),
host,
reload_cb,
ctx.ui_dialog.clone(),
ctx.ext_status.clone(),
ctx.editor_text.clone(),
);
let extension_session = if effective_args.no_extensions {
rpi_extensions::ExtensionSession::none()
} else {
load_extensions_from_dirs(
&effective_args,
&extension_dirs,
Some(Arc::clone(&fresh_bridge)),
Some(ctx.resource_leases.clone()),
)
};
if let Err(error) =
crate::native_store::validate_stdio(ctx.fresh_session_only, extension_session.owns_stdio())
{
return ReloadOutcome {
summary: "Reload refused by experimental native session policy".into(),
had_warnings: true,
details: vec![error],
had_errors: true,
project_trusted: None,
};
}
if extension_session.is_empty() && !effective_args.no_extensions {
}
if effective_args.verbose {
if let Some(s) = extension_session.summary() {
eprintln!("reload: {s}");
}
report_deferred_renderers(&extension_session);
}
let has_package_providers = |session: &ExtensionSession| {
session.snapshot().is_some_and(|snapshot| {
snapshot
.providers()
.iter()
.any(|provider| snapshot.provider_manifest(&provider.provider_id).is_some())
})
};
let packaged = has_package_providers(&extension_session)
|| has_package_providers(&ctx.extension_session.lock().unwrap());
let refreshed = if packaged {
let mut args = effective_args.clone();
if let Ok(current) = harness.lane("main").get_model().await {
let was_package = ctx
.extension_session
.lock()
.unwrap()
.snapshot()
.is_some_and(|snapshot| snapshot.provider_manifest(¤t.provider).is_some());
let has_replacement = extension_session.snapshot().is_some_and(|snapshot| {
snapshot
.providers()
.iter()
.any(|provider| provider.provider_id == current.provider)
});
if was_package && !has_replacement {
return ReloadOutcome {
summary: "Reload failed while resolving provider packages".into(),
had_warnings: true,
details: vec![format!(
"active provider package '{}' is no longer loaded",
current.provider
)],
had_errors: true,
project_trusted: None,
};
}
args.provider = Some(current.provider.clone());
args.model = Some(current.id);
if current.provider != ctx.resolved_model.provider {
args.api_key = None;
args.base_url = None;
}
}
match crate::provider::resolve_with_extensions(
&args,
&ctx.cwd,
ctx.project_trusted,
&extension_session,
) {
Ok(resolved) => Some(resolved),
Err(error) => {
return ReloadOutcome {
summary: "Reload failed while resolving provider packages".into(),
had_warnings: true,
details: vec![error.to_string()],
had_errors: true,
project_trusted: None,
}
}
}
} else {
None
};
let old_session = {
let mut session_guard = ctx.extension_session.lock().unwrap();
let old_session = std::mem::replace(&mut *session_guard, extension_session.clone());
if let Some(old_snap) = old_session.snapshot_arc() {
old_snap.active_flag().store(false, Ordering::SeqCst);
}
old_session
};
if let Some(old_b) = old_bridge {
old_b.invalidate();
}
old_session.begin_shutdown();
*ctx.action_bridge.lock().unwrap() = Some(Arc::clone(&fresh_bridge));
publish_host.publish_registry_opt(extension_session.snapshot_arc());
let discovered = extension_session
.snapshot_arc()
.map(|snap| rpi_extensions::emit_resources_discover(&cwd_str, "reload", &snap))
.unwrap_or_default();
if let Ok(mut guard) = ctx.discovered_theme_paths.lock() {
*guard = discovered.theme_paths.clone();
}
let env = Arc::new(rpi_tools::OsExecutionEnv::with_cwd(ctx.cwd.clone()));
let env_dyn: Arc<dyn rpi_tools::ExecutionEnv> = env.clone();
let mut skills: Vec<rpi_harness::types::Skill> = Vec::new();
let mut skill_diags: Vec<rpi_harness::skills::SkillDiagnostic> = Vec::new();
let mut resource_diags: Vec<rpi_harness::diagnostics::ResourceDiagnostic> = Vec::new();
if !effective_args.no_skills {
let js_paths: &[PathBuf] = &[];
let package_paths: &[PathBuf] = &[];
let dirs = append_reload_resource_paths(
skill_base_dirs,
&effective_args.skill,
&discovered.skill_paths,
js_paths,
package_paths,
);
let result = load_skills_with_precedence(&env_dyn, &dirs).await;
skills = result.skills;
skill_diags = result.diagnostics;
resource_diags.extend(result.resource_diagnostics);
}
let mut prompt_templates: Vec<rpi_harness::types::PromptTemplate> = Vec::new();
let mut prompt_diags: Vec<rpi_harness::prompt_templates::PromptTemplateDiagnostic> = Vec::new();
if !effective_args.no_prompt_templates {
let js_paths: &[PathBuf] = &[];
let package_paths: &[PathBuf] = &[];
let dirs = append_reload_resource_paths(
prompt_base_dirs,
&effective_args.prompt_template,
&discovered.prompt_paths,
js_paths,
package_paths,
);
let result = load_prompt_templates_with_precedence(&env_dyn, &dirs).await;
prompt_templates = result.prompt_templates;
prompt_diags = result.diagnostics;
resource_diags.extend(result.resource_diagnostics);
}
let context_files = if effective_args.no_context_files || !project_trusted {
Vec::new()
} else {
let agent_dir = crate::config::agent_dir().ok();
let agent_dir_path = agent_dir.unwrap_or_else(|| ctx.cwd.clone());
load_project_context_files(&env_dyn, &ctx.cwd, &agent_dir_path).await
};
let context_block = format_project_context(&context_files);
let mut details: Vec<String> = Vec::new();
if let Some(resolved) = &refreshed {
warnings |= !resolved.compat_warnings.is_empty();
details.extend(
resolved
.compat_warnings
.iter()
.map(|warning| format!("warning: models.json: {warning}")),
);
}
if !skill_diags.is_empty() || !prompt_diags.is_empty() || !resource_diags.is_empty() {
warnings = true;
for d in &skill_diags {
details.push(format!(
"warning: skill {} ({}): {}",
d.path,
d.code.as_str(),
d.message
));
}
for d in &prompt_diags {
details.push(format!(
"warning: prompt template {} ({}): {}",
d.path,
d.code.as_str(),
d.message
));
}
for d in &resource_diags {
if let rpi_harness::diagnostics::ResourceDiagnostic::Collision(c) = d {
details.push(format!(
"warning: {} name \"{}\" collision: keeping {} (shadowed {})",
c.resource_type.as_str(),
c.name,
c.winner_path,
c.loser_path
));
}
}
}
let mut_env: Arc<dyn rpi_tools::MutatingEnv> = env.clone();
let tool_ctx = rpi_tools::ExecutionToolContext::new(env_dyn.clone(), Some(mut_env))
.with_file_io(ctx.file_io.clone());
let mut tools = build_tools(&tool_ctx, &effective_args);
merge_extension_tools(
&mut tools,
&extension_session,
&effective_args,
&ctx.tool_context,
);
let mut active = active_tool_names(&tools, &effective_args);
active = filter_active_tool_names(active, &effective_args);
let prompt_tools = active.clone();
let prompt_metadata = extension_tool_prompt_metadata(&extension_session);
let prior_reasoning_replayed = rpi_ai::model::prior_reasoning_is_replayed(&ctx.resolved_model);
let default_prompt = || {
default_system_prompt_with_tool_prompts(
&cwd_str,
prior_reasoning_replayed,
&prompt_tools,
&prompt_metadata,
)
};
let mut context_paths = Vec::new();
let base_prompt = match effective_args.system_prompt.as_deref() {
Some(explicit) => explicit.to_string(),
None if project_trusted => match discover_system_prompt_file(&ctx.cwd) {
Some(path) => {
read_prompt_source(&path, &mut context_paths).unwrap_or_else(default_prompt)
}
None => default_prompt(),
},
None => default_prompt(),
};
let mut append_texts: Vec<String> = Vec::new();
for extra in &effective_args.append_system_prompt {
let text = read_append_target(extra, &mut context_paths).unwrap_or_else(|| extra.clone());
append_texts.push(text);
}
if effective_args.append_system_prompt.is_empty() {
if let Some(path) = project_trusted
.then(|| discover_append_system_prompt_file(&ctx.cwd))
.flatten()
{
if let Some(text) = read_prompt_source(&path, &mut context_paths) {
append_texts.push(text);
}
}
}
context_paths.extend(
context_files
.iter()
.map(|file| file.path.to_string_lossy().into_owned()),
);
let append_join = if append_texts.is_empty() {
None
} else {
Some(append_texts.join("\n\n"))
};
let system_prompt = compose_system_prompt(
Some(&base_prompt),
&[],
if context_block.is_empty() {
None
} else {
Some(&context_block)
},
append_join.as_deref(),
);
let emitter: Arc<dyn rpi_agent::AgentEmitter> = match extension_session.snapshot_arc() {
Some(snapshot) => {
let ext = ExtensionEmitter::new(snapshot, extension_session.keepalive());
Arc::new(TeeEmitter::new(vec![ctx.broadcast.clone(), Arc::new(ext)]))
}
None => ctx.broadcast.clone(),
};
let resources = AgentHarnessResources {
context_files: context_paths,
skills: if skills.is_empty() {
None
} else {
Some(skills.clone())
},
prompt_templates: if prompt_templates.is_empty() {
None
} else {
Some(prompt_templates.clone())
},
};
let _ = harness.set_system_prompt(Some(system_prompt)).await;
let _ = harness.set_resources(resources).await;
let _ = harness.set_agent_emitter(Some(emitter)).await;
let _ = harness
.set_models(build_models_with_extensions_for_reload(
&refreshed
.as_ref()
.map(|resolved| resolved.providers.clone())
.unwrap_or_else(|| {
ctx.providers
.read()
.unwrap_or_else(|error| error.into_inner())
.clone()
}),
&extension_session,
ctx.runtime.clone(),
))
.await;
let _ = harness
.set_provider_hooks(
rpi_extensions::ExtensionProviderHooks::from_session(&extension_session)
.map(|h| Arc::new(h) as Arc<dyn rpi_ai::ProviderHooks>),
)
.await;
if let Some(resolved) = &refreshed {
*ctx.providers
.write()
.unwrap_or_else(|error| error.into_inner()) = resolved.providers.clone();
let catalog = crate::provider::authenticated_catalog(resolved);
fresh_bridge
.clone_host()
.replace_model_catalog(catalog.clone());
*ctx.catalog
.write()
.unwrap_or_else(|error| error.into_inner()) = catalog;
let _ = harness.lane("main").set_model(resolved.model.clone()).await;
}
let _ = harness.set_tools(tools, Some(active)).await;
let summary = format!(
"Reloaded {} plugin(s), {} skill(s), {} prompt(s).",
extension_session.loaded_paths().len(),
skills.len(),
prompt_templates.len(),
);
ReloadOutcome {
summary,
had_warnings: warnings,
details,
had_errors: false,
project_trusted: Some(project_trusted),
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct ReloadSettingsFields {
skill_dirs: Option<Vec<String>>,
prompt_dirs: Option<Vec<String>>,
extension_dirs: Option<Vec<String>>,
}
impl From<crate::settings::Settings> for ReloadSettingsFields {
fn from(settings: crate::settings::Settings) -> Self {
Self {
skill_dirs: settings.skill_dirs,
prompt_dirs: settings.prompt_dirs,
extension_dirs: settings.extension_dirs,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct ReloadSettingsSnapshot {
global: ReloadSettingsFields,
project: Option<ReloadSettingsFields>,
}
fn reload_reads_settings(args: &Args) -> bool {
!args.dev_local_only && (!args.no_extensions || !args.no_skills || !args.no_prompt_templates)
}
fn load_reload_settings_snapshot(
args: &Args,
cwd: &Path,
project_trusted: bool,
) -> Result<Option<ReloadSettingsSnapshot>, String> {
if !reload_reads_settings(args) {
return Ok(None);
}
let global = crate::settings::load_settings()
.map_err(|error| format!("could not load global settings: {error}"))?
.into();
let project = if project_trusted {
crate::settings::load_active_project_settings(cwd)
.map_err(|error| format!("could not load project settings: {error}"))?
.map(|(_, settings)| settings.into())
} else {
None
};
Ok(Some(ReloadSettingsSnapshot { global, project }))
}
#[cfg(test)]
fn validate_settings_for_reload(
args: &Args,
cwd: &Path,
project_trusted: bool,
) -> Result<(), String> {
load_reload_settings_snapshot(args, cwd, project_trusted).map(|_| ())
}
fn prepare_reload_inputs(
args: &Args,
cwd: &Path,
project_trusted: bool,
) -> Result<PreparedReloadInputs, String> {
prepare_reload_inputs_inner(args, cwd, project_trusted, || {})
}
fn prepare_reload_inputs_inner<F>(
args: &Args,
cwd: &Path,
project_trusted: bool,
before_verify: F,
) -> Result<PreparedReloadInputs, String>
where
F: FnOnce(),
{
let settings_before = load_reload_settings_snapshot(args, cwd, project_trusted)?;
let mut extension_dirs = if args.no_extensions || args.dev_local_only {
Vec::new()
} else if args.project_only {
project_extension_dirs(cwd)
} else if project_trusted {
extension_dirs(cwd)
} else {
global_extension_dirs()
};
if !args.no_extensions {
extension_dirs.extend(args.extensions_dir.iter().cloned());
}
let skill_base_dirs = if args.no_skills {
Vec::new()
} else if args.dev_local_only || args.project_only {
project_skill_dirs(cwd)
} else if project_trusted {
skill_dirs(cwd)
} else {
global_skill_dirs()
};
let prompt_base_dirs = if args.no_prompt_templates {
Vec::new()
} else if args.dev_local_only || args.project_only {
project_prompt_template_dirs(cwd)
} else if project_trusted {
prompt_template_dirs(cwd)
} else {
global_prompt_template_dirs()
};
before_verify();
let settings_after = load_reload_settings_snapshot(args, cwd, project_trusted)?;
if settings_before != settings_after {
return Err(
"settings changed while reload inputs were being prepared; retry /reload".to_string(),
);
}
Ok(PreparedReloadInputs {
extension_dirs,
skill_base_dirs,
prompt_base_dirs,
})
}
fn build_models_with_extensions_for_reload(
providers: &[Arc<dyn Provider>],
extension_session: &ExtensionSession,
runtime: tokio::runtime::Handle,
) -> Vec<Arc<dyn Provider>> {
let mut models: Vec<Arc<dyn Provider>> = providers.to_vec();
let pluggable = rpi_extensions::PluggableProvider::from_session(extension_session, runtime);
for provider in pluggable {
if extension_session
.snapshot()
.is_some_and(|snapshot| snapshot.provider_manifest(provider.id()).is_some())
{
continue;
}
if !models.iter().any(|known| known.id() == provider.id()) {
models.push(provider);
}
}
models
}
fn report_deferred_renderers(session: &ExtensionSession) {
let Some(snap) = session.snapshot_arc() else {
return;
};
let all = snap.renderers();
let markdown = all
.iter()
.filter(|r| r.kind == rpi_extensions::RegisteredRendererKind::Markdown)
.count();
let message = all
.iter()
.filter(|r| r.kind == rpi_extensions::RegisteredRendererKind::Message)
.count();
let entry = all
.iter()
.filter(|r| r.kind == rpi_extensions::RegisteredRendererKind::Entry)
.count();
if markdown + message + entry == 0 {
return;
}
eprintln!(
"renderers: {} markdown-transform, {} message-render, {} entry-render (active)",
markdown, message, entry
);
}
#[derive(Debug, thiserror::Error)]
pub enum BuildError {
#[error("Experimental native session policy: {0}")]
NativeStorePolicy(String),
#[error("Could not create the session directory: {0}")]
SessionDir(String),
#[error("Could not open the saved session: {0}")]
SessionOpen(String),
#[error("No session found for {requested} in {dir}. Start a fresh session instead (drop --continue/--resume/--session).")]
SessionNotFound { requested: String, dir: String },
#[error("Could not build the harness: {0}")]
HarnessCreate(String),
}
fn build_models_with_extensions(
resolved: &ResolvedModel,
extension_session: &ExtensionSession,
runtime: tokio::runtime::Handle,
) -> Vec<Arc<dyn Provider>> {
build_models_with_extensions_for_reload(&resolved.providers, extension_session, runtime)
}
pub(crate) fn resolve_project_trust(args: &Args, cwd: &Path) -> bool {
if let Some(override_value) = args.trust_override {
return override_value;
}
crate::config::project_trust_decision(cwd)
.ok()
.flatten()
.unwrap_or(true)
}
pub(crate) fn load_extensions(
args: &Args,
cwd: &Path,
project_trusted: bool,
action_bridge: Option<Arc<rpi_extensions::ActionBridge>>,
) -> ExtensionSession {
if args.no_extensions {
return ExtensionSession::none();
}
let mut dirs = if args.dev_local_only {
Vec::new()
} else if args.project_only {
project_extension_dirs(cwd)
} else if project_trusted {
extension_dirs(cwd)
} else {
global_extension_dirs()
};
dirs.extend(args.extensions_dir.iter().cloned());
let leases = match crate::resource_management::host_leases(cwd) {
Ok(leases) => leases,
Err(error) => {
eprintln!("warning: refusing extension loading: {error}");
return ExtensionSession::none();
}
};
load_extensions_from_dirs(args, &dirs, action_bridge, Some(leases))
}
fn load_extensions_from_dirs(
args: &Args,
dirs: &[PathBuf],
action_bridge: Option<Arc<rpi_extensions::ActionBridge>>,
host_guard: Option<Arc<dyn Send + Sync>>,
) -> ExtensionSession {
let diagnostics: Arc<dyn PluginDiagnostics> = Arc::new(NullDiagnostics);
rpi_extensions::load_session_mixed_guarded(
dirs,
&args.extension,
diagnostics,
action_bridge,
host_guard,
)
}
fn merge_extension_tools(
tools: &mut Vec<HarnessTool>,
session: &ExtensionSession,
args: &Args,
tool_context: &rpi_extensions::ToolCallContext,
) {
let Some(snapshot) = session.snapshot() else {
return;
};
for et in snapshot.tools() {
let name = &et.tool.name;
if !tool_name_allowed(name, args) {
continue;
}
let adapter = PluginToolAdapter::new(et.tool.clone(), et.handle(), session.keepalive())
.with_context(tool_context.clone());
let harness_tool = HarnessTool::new(Arc::new(adapter));
match tools.iter_mut().find(|t| t.tool.schema().name == *name) {
Some(slot) => *slot = harness_tool,
None => tools.push(harness_tool),
}
}
}
pub fn bash_options() -> rpi_tools::tools::bash::BashToolOptions {
use rpi_tools::tools::bash::BashToolOptions;
let default = std::env::var("RPI_BASH_TIMEOUT")
.ok()
.and_then(|v| v.parse::<f64>().ok())
.unwrap_or(rpi_tools::tools::bash::DEFAULT_BASH_TIMEOUT_SECONDS);
let settings = crate::settings::load_settings().unwrap_or_default();
BashToolOptions {
command_prefix: settings.shell_command_prefix.clone(),
default_timeout: Some(default),
prepare: None,
}
}
fn build_tools(ctx: &ExecutionToolContext, args: &Args) -> Vec<HarnessTool> {
if args.no_tools {
return Vec::new();
}
let mut all: Vec<(&'static str, HarnessTool)> = vec![
(
"read",
HarnessTool::new(create_read_tool(
ctx,
Some(rpi_tools::tools::read::ReadToolOptions {
image_processor: Some(Arc::new(
rpi_tools::image_processing::InlineImageProcessor,
)),
..Default::default()
}),
))
.with_replay(ToolReplay::Safe),
),
(
"bash",
HarnessTool::new(create_bash_tool(ctx, Some(bash_options()))),
),
("edit", HarnessTool::new(create_edit_tool(ctx))),
("write", HarnessTool::new(create_write_tool(ctx))),
(
"docs",
HarnessTool::new(create_docs_tool()).with_replay(ToolReplay::Safe),
),
];
#[cfg(target_os = "windows")]
all.push(("powershell", HarnessTool::new(create_powershell_tool(ctx))));
if args.tools.is_none() && args.resource_config.is_none() && args.resource_transition.is_none()
{
if let Ok(settings) = crate::settings::load_settings() {
if let Some(default_tools) = &settings.default_tools {
all.retain(|(name, _)| default_tools.iter().any(|t| t == name));
}
}
}
if args.no_builtin_tools {
all.clear();
}
if let Some(allow) = &args.tools {
all.retain(|(name, _)| allow.iter().any(|a| a == name));
}
if let Some(deny) = &args.exclude_tools {
all.retain(|(name, _)| !deny.iter().any(|d| d == name));
}
all.into_iter().map(|(_, t)| t).collect()
}
fn active_tool_names(tools: &[HarnessTool], args: &Args) -> Vec<String> {
filter_active_tool_names(
tools.iter().map(|tool| tool.tool.schema().name.clone()),
args,
)
}
pub(crate) fn tool_name_allowed(name: &str, args: &Args) -> bool {
if args.no_tools {
return false;
}
if args
.tools
.as_ref()
.is_some_and(|allow| !allow.iter().any(|value| value == name))
{
return false;
}
if args
.exclude_tools
.as_ref()
.is_some_and(|deny| deny.iter().any(|value| value == name))
{
return false;
}
true
}
pub(crate) fn filter_active_tool_names<I>(names: I, args: &Args) -> Vec<String>
where
I: IntoIterator<Item = String>,
{
names
.into_iter()
.filter(|name| tool_name_allowed(name, args))
.collect()
}
async fn build_session(
selection: &SessionSelection,
cwd: &str,
selected: Option<&crate::resource_selection::ResolvedSelection>,
) -> Result<Session, BuildError> {
match selection {
SessionSelection::Ephemeral => Ok(ephemeral_session()),
SessionSelection::ExperimentalNative { dir } => Ok(Session::new(
backend::fresh_experimental_native(dir).await?,
None,
)),
SessionSelection::New { dir, .. } => {
std::fs::create_dir_all(dir)
.map_err(|e| BuildError::SessionDir(format!("{}: {e}", dir.display())))?;
let session = create_jsonl_session(dir, cwd)
.await
.map_err(|e| BuildError::SessionDir(format!("{}: {e}", dir.display())))?;
Ok(session)
}
SessionSelection::Latest { .. }
| SessionSelection::ById { .. }
| SessionSelection::ByExactId { .. } => restore_session(selection, cwd).await,
SessionSelection::Fork { source, dir } => {
fork_session_at_launch(source, dir, cwd, selected).await
}
}
}
async fn restore_session(selection: &SessionSelection, cwd: &str) -> Result<Session, BuildError> {
let Some(meta) = resolve_restore_target(selection, cwd).await? else {
return match selection {
SessionSelection::ByExactId { id, dir } => {
std::fs::create_dir_all(dir)
.map_err(|e| BuildError::SessionDir(format!("{}: {e}", dir.display())))?;
create_jsonl_session_with_id(dir, cwd, Some(id.clone()))
.await
.map_err(|e| BuildError::SessionDir(format!("{}: {e}", dir.display())))
}
SessionSelection::Latest { dir } => Err(BuildError::SessionNotFound {
requested: "the most recent session".to_string(),
dir: dir.display().to_string(),
}),
SessionSelection::ById { id, dir } => Err(BuildError::SessionNotFound {
requested: format!("session {id}"),
dir: dir.display().to_string(),
}),
_ => unreachable!("resolve_restore_target only returns None for restore selections"),
};
};
let sessions_root = selection_dir(selection);
let writer_restore = StartupSpan::start(Phase::WriterRestore);
let result = open_session(&meta, cwd, &sessions_root)
.await
.map(|restored| restored.session);
if result.is_ok() {
writer_restore.finish();
}
result
}
fn selection_dir(selection: &SessionSelection) -> PathBuf {
match selection {
SessionSelection::Ephemeral => PathBuf::new(),
SessionSelection::New { dir, .. }
| SessionSelection::ExperimentalNative { dir }
| SessionSelection::Latest { dir }
| SessionSelection::ById { dir, .. }
| SessionSelection::ByExactId { dir, .. }
| SessionSelection::Fork { dir, .. } => dir.clone(),
}
}
async fn resolve_restore_target(
selection: &SessionSelection,
cwd: &str,
) -> Result<Option<rpi_harness::session::jsonl::JsonlSessionMetadata>, BuildError> {
match selection {
SessionSelection::Latest { dir } => {
let metas = list_session_metadata_in_dir(cwd, dir).await?;
Ok(first_resumable_session(&metas).cloned())
}
SessionSelection::ById { id, dir } => find_session_metadata(id, cwd, dir).await,
SessionSelection::ByExactId { id, dir } => {
let metas = list_session_metadata_in_dir(cwd, dir).await?;
Ok(metas.into_iter().find(|m| m.id == *id))
}
SessionSelection::Fork { .. }
| SessionSelection::ExperimentalNative { .. }
| SessionSelection::New { .. }
| SessionSelection::Ephemeral => Ok(None),
}
}
async fn find_session_metadata(
id: &str,
cwd: &str,
dir: &Path,
) -> Result<Option<rpi_harness::session::jsonl::JsonlSessionMetadata>, BuildError> {
let metas = list_session_metadata_in_dir(cwd, dir).await?;
Ok(metas
.into_iter()
.find(|m| m.id == id || m.path.contains(id) || id.contains(&m.id)))
}
pub(crate) struct RestoredSessionCwd {
pub id: String,
pub recorded: PathBuf,
pub adopted: Option<PathBuf>,
}
pub(crate) async fn peek_restored_session_cwd(
selection: &SessionSelection,
launch_cwd: &Path,
) -> Option<RestoredSessionCwd> {
let cwd_str = launch_cwd.to_string_lossy().into_owned();
let meta = resolve_restore_target(selection, &cwd_str).await.ok()??;
let recorded = PathBuf::from(&meta.cwd);
if recorded.as_os_str().is_empty() {
return None;
}
Some(RestoredSessionCwd {
id: meta.id,
adopted: crate::session_cwd::resolve_session_cwd(&recorded),
recorded,
})
}
pub(crate) fn pin_restored_selection(args: &mut Args, restored: &RestoredSessionCwd) {
args.continue_session = false;
args.resume = false;
args.session = Some(restored.id.clone());
}
async fn fork_session_at_launch(
source: &str,
dir: &Path,
cwd: &str,
selected: Option<&crate::resource_selection::ResolvedSelection>,
) -> Result<Session, BuildError> {
use rpi_harness::session::jsonl::{
JsonlSessionCreateOptions, JsonlSessionRepo, JsonlSessionRepoOptions,
};
use rpi_harness::session::types::{ForkOptions, SessionStorage};
use rpi_tools::FileSystem;
let dir = dir.to_path_buf();
std::fs::create_dir_all(&dir)
.map_err(|e| BuildError::SessionDir(format!("{}: {e}", dir.display())))?;
let env = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(cwd)));
let fs: Arc<dyn FileSystem> = env.clone();
let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
fs: fs.clone(),
sessions_root: dir.to_string_lossy().into_owned(),
clock: Arc::new(SystemClock),
ids: Arc::new(DefaultIdGenerator::new()),
});
let source_meta = match session_metadata_at(Path::new(source)) {
Some(meta) => meta,
None => {
let metas = repo
.list_typed(&rpi_harness::session::jsonl::JsonlSessionListOptions::default())
.await
.map_err(|e| BuildError::SessionDir(format!("list sessions: {e}")))?;
metas
.into_iter()
.find(|m| m.id == *source || m.path.contains(source) || source.contains(&m.id))
.ok_or_else(|| BuildError::SessionNotFound {
requested: format!("--fork {source}"),
dir: dir.display().to_string(),
})?
}
};
let fork_storage = repo
.fork_typed_with_value_initializer(
&source_meta,
&JsonlSessionCreateOptions {
id: None,
parent_session_id: Some(source_meta.id.clone()),
cwd: cwd.to_string(),
metadata: None,
},
&ForkOptions::default(),
|values| crate::resource_selection::fork_initial_values_for_selection(values, selected),
)
.await
.map_err(|e| BuildError::SessionDir(format!("fork {}: {e}", source_meta.path)))?;
claim_session_lock(&fork_storage.jsonl_metadata().path).map_err(BuildError::SessionOpen)?;
let storage_arc: Arc<dyn SessionStorage> = Arc::new(fork_storage);
Ok(Session::new(storage_arc, None))
}
fn session_metadata_at(path: &Path) -> Option<rpi_harness::session::jsonl::JsonlSessionMetadata> {
use rpi_harness::session::jsonl::codec::{metadata_from_header, parse_header};
use std::io::BufRead;
let file = std::fs::File::open(path).ok()?;
let modified_at = file
.metadata()
.ok()?
.modified()
.ok()?
.duration_since(std::time::UNIX_EPOCH)
.ok()?
.as_millis() as i64;
let mut first_line = String::new();
std::io::BufReader::new(file)
.read_line(&mut first_line)
.ok()?;
let header = parse_header(first_line.trim_end()).ok()?;
let path = std::fs::canonicalize(path).ok()?;
Some(metadata_from_header(
&header,
&path.to_string_lossy(),
modified_at,
))
}
pub enum OpenError {
NotFound { requested: String },
Other(String),
}
impl std::fmt::Display for OpenError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
OpenError::NotFound { requested } => write!(f, "no session matches {requested}"),
OpenError::Other(msg) => write!(f, "Could not open the saved session: {msg}"),
}
}
}
pub async fn list_session_metadata(
cwd: &str,
) -> Result<Vec<rpi_harness::session::jsonl::JsonlSessionMetadata>, BuildError> {
let dir = default_session_dir(Path::new(cwd));
list_session_metadata_in_dir(cwd, &dir).await
}
pub(crate) async fn list_session_metadata_in_dir(
cwd: &str,
dir: &Path,
) -> Result<Vec<rpi_harness::session::jsonl::JsonlSessionMetadata>, BuildError> {
use rpi_harness::session::jsonl::{
JsonlSessionListOptions, JsonlSessionRepo, JsonlSessionRepoOptions,
};
use rpi_tools::FileSystem;
let env = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(cwd)));
let fs: Arc<dyn FileSystem> = env.clone();
let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
fs: fs.clone(),
sessions_root: dir.to_string_lossy().into_owned(),
clock: Arc::new(SystemClock),
ids: Arc::new(DefaultIdGenerator::new()),
});
let metadata_lookup = StartupSpan::start(Phase::MetadataLookup);
let result = repo
.list_typed(&JsonlSessionListOptions::default())
.await
.map_err(|e| BuildError::SessionDir(format!("list sessions: {e}")));
if result.is_ok() {
metadata_lookup.finish();
}
result
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct SessionSummary {
pub name: Option<String>,
pub preview: Option<String>,
pub empty: bool,
pub bytes: u64,
}
const SUMMARY_HEAD_BYTES: u64 = 256 * 1024;
const SUMMARY_TAIL_BYTES: u64 = 8 * 1024;
const SUMMARY_PREVIEW_CHARS: usize = 80;
pub fn summarize_session_file(path: &Path) -> SessionSummary {
use std::io::{Read, Seek, SeekFrom};
let Ok(mut file) = std::fs::File::open(path) else {
return SessionSummary::default();
};
let Ok(size) = file.metadata().map(|m| m.len()) else {
return SessionSummary::default();
};
if size == 0 {
return SessionSummary {
name: None,
preview: None,
empty: true,
bytes: 0,
};
}
let mut head = vec![0u8; SUMMARY_HEAD_BYTES.min(size) as usize];
let head_len = file.read(&mut head).unwrap_or(0);
head.truncate(head_len);
let head = String::from_utf8_lossy(&head);
let mut lines: Vec<&str> = head.lines().collect();
let header_len = lines.first().map(|line| line.len()).unwrap_or(0);
let empty = (size as usize) <= header_len + 1;
if (head_len as u64) < size && !lines.is_empty() {
lines.pop();
}
let mut summary = SessionSummary {
name: None,
preview: None,
empty,
bytes: size,
};
for line in lines {
apply_summary_line(line, &mut summary);
}
if size > SUMMARY_HEAD_BYTES
&& file
.seek(SeekFrom::End(-(SUMMARY_TAIL_BYTES as i64)))
.is_ok()
{
let mut tail = Vec::new();
if file.read_to_end(&mut tail).is_ok() {
let tail = String::from_utf8_lossy(&tail);
let mut tail_lines = tail.lines();
tail_lines.next();
for line in tail_lines {
if let Some(name) = name_from_line(line) {
summary.name = name;
}
}
}
}
summary
}
fn name_from_line(line: &str) -> Option<Option<String>> {
let value: serde_json::Value = serde_json::from_str(line).ok()?;
if value.get("kind").and_then(|v| v.as_str()) != Some("fact") {
return None;
}
if value.get("fact").and_then(|v| v.as_str()) != Some("name") {
return None;
}
Some(match value.get("name") {
Some(serde_json::Value::String(name)) => Some(name.clone()),
_ => None,
})
}
fn apply_summary_line(line: &str, summary: &mut SessionSummary) {
if let Some(name) = name_from_line(line) {
summary.name = name;
return;
}
if summary.preview.is_some() {
return;
}
let Ok(value) = serde_json::from_str::<serde_json::Value>(line) else {
return;
};
if value.get("kind").and_then(|v| v.as_str()) != Some("entry") {
return;
}
if value.get("type").and_then(|v| v.as_str()) != Some("message") {
return;
}
let Some(message) = value.get("message") else {
return;
};
if message.get("role").and_then(|v| v.as_str()) != Some("user") {
return;
}
let text = match message.get("content") {
Some(serde_json::Value::String(text)) => Some(text.as_str()),
Some(serde_json::Value::Array(blocks)) => blocks.iter().find_map(|block| {
(block.get("type").and_then(|v| v.as_str()) == Some("text"))
.then(|| block.get("text").and_then(|v| v.as_str()))
.flatten()
}),
_ => None,
};
let Some(text) = text else { return };
let collapsed = text
.lines()
.map(str::trim)
.filter(|line| !line.is_empty())
.collect::<Vec<_>>()
.join(" ");
if collapsed.is_empty() {
return;
}
summary.preview = Some(collapsed.chars().take(SUMMARY_PREVIEW_CHARS).collect());
}
pub fn session_display_label(summary: &SessionSummary) -> String {
if let Some(name) = summary.name.as_deref().map(str::trim) {
if !name.is_empty() {
return name.to_string();
}
}
match summary.preview.as_deref().map(str::trim) {
Some(preview) if !preview.is_empty() => preview.to_string(),
_ => "(no messages)".to_string(),
}
}
pub fn format_session_bytes(bytes: u64) -> String {
if bytes >= 1_048_576 {
format!("{:.1}MB", bytes as f64 / 1_048_576.0)
} else if bytes >= 1024 {
format!("{}KB", bytes / 1024)
} else {
format!("{bytes}B")
}
}
pub fn short_session_id(id: &str) -> &str {
&id[..id.len().min(8)]
}
pub fn list_session_files_sync(dir: &Path) -> Vec<PathBuf> {
let mut files: Vec<(PathBuf, std::time::SystemTime)> = Vec::new();
let mut visit = |path: PathBuf| {
if path.extension().and_then(|e| e.to_str()) != Some("jsonl")
|| path.to_string_lossy().ends_with(".progress.jsonl")
{
return;
}
let modified = std::fs::metadata(&path)
.and_then(|m| m.modified())
.unwrap_or(std::time::UNIX_EPOCH);
files.push((path, modified));
};
if let Ok(entries) = std::fs::read_dir(dir) {
for entry in entries.flatten() {
let path = entry.path();
if path.is_dir() {
if let Ok(inner) = std::fs::read_dir(&path) {
for child in inner.flatten() {
visit(child.path());
}
}
} else {
visit(path);
}
}
}
files.sort_by(|a, b| b.1.cmp(&a.1));
files.into_iter().map(|(path, _)| path).collect()
}
fn first_resumable_session(
metas: &[rpi_harness::session::jsonl::JsonlSessionMetadata],
) -> Option<&rpi_harness::session::jsonl::JsonlSessionMetadata> {
metas
.iter()
.find(|meta| !summarize_session_file(Path::new(&meta.path)).empty)
.or_else(|| metas.first())
}
pub async fn open_session_by_id(id: &str, cwd: &str) -> Result<Session, OpenError> {
let dir = default_session_dir(Path::new(cwd));
open_session_by_id_in_dir(id, cwd, &dir)
.await
.map(|restored| restored.session)
}
pub(crate) async fn open_session_by_id_in_dir(
id: &str,
cwd: &str,
dir: &Path,
) -> Result<RestoredSession, OpenError> {
let meta = find_session_metadata(id, cwd, dir)
.await
.map_err(|e| OpenError::Other(e.to_string()))?;
let Some(meta) = meta else {
return Err(OpenError::NotFound {
requested: format!("session {id}"),
});
};
open_session(&meta, cwd, dir).await.map_err(|e| {
OpenError::Other(match e {
BuildError::SessionOpen(reason) => reason,
other => other.to_string(),
})
})
}
fn publish_session_identity(tool_context: &rpi_extensions::ToolCallContext, id: &str) {
std::env::set_var(rpi_plugin_sdk::SESSION_ID_ENV, id);
tool_context.set_session_id(id);
}
pub(crate) async fn set_active_session(
harness: &AgentHarness,
session: Session,
tool_context: Option<&rpi_extensions::ToolCallContext>,
) -> Result<(), String> {
let metadata = session.get_metadata().await.ok();
harness
.set_session(session)
.await
.map_err(|e| e.to_string())?;
if let Some(metadata) = metadata {
std::env::set_var(rpi_plugin_sdk::SESSION_ID_ENV, &metadata.id);
if let Some(context) = tool_context {
context.set_session_id(metadata.id);
}
}
Ok(())
}
pub(crate) async fn fork_session_storage(
harness: &AgentHarness,
cwd: &str,
dir: &Path,
) -> Result<Session, String> {
use rpi_harness::session::jsonl::{JsonlSessionRepo, JsonlSessionRepoOptions};
use rpi_tools::FileSystem;
let env = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(cwd)));
let fs: Arc<dyn FileSystem> = env.clone();
let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
fs,
sessions_root: dir.to_string_lossy().into_owned(),
clock: Arc::new(SystemClock),
ids: Arc::new(DefaultIdGenerator::new()),
});
let id = harness.session().storage().metadata().id.clone();
let metas = list_session_metadata_in_dir(cwd, dir)
.await
.map_err(|e| e.to_string())?;
let Some(source) = metas.iter().find(|m| m.id == id) else {
return Err(format!("current session {id} not found on disk"));
};
let fork_storage = repo
.fork_typed_with_value_initializer(
source,
&rpi_harness::session::jsonl::JsonlSessionCreateOptions {
id: None,
parent_session_id: Some(source.id.clone()),
cwd: cwd.to_string(),
metadata: None,
},
&rpi_harness::session::types::ForkOptions::default(),
crate::resource_selection::fork_initial_values,
)
.await
.map_err(|e| e.to_string())?;
claim_session_lock(&fork_storage.jsonl_metadata().path)?;
Ok(Session::new(Arc::new(fork_storage), None))
}
pub(crate) async fn import_session_file(
source: &Path,
cwd: &str,
dir: &Path,
) -> Result<Session, String> {
use rpi_harness::session::jsonl::{
JsonlSessionCreateOptions, JsonlSessionRepo, JsonlSessionRepoOptions, JsonlSessionStorage,
};
use rpi_harness::session::types::SessionStorage;
use rpi_tools::FileSystem;
let dir = if dir.is_absolute() {
dir.to_path_buf()
} else {
Path::new(cwd).join(dir)
};
std::fs::create_dir_all(&dir).map_err(|e| e.to_string())?;
let source_progress = PathBuf::from(format!("{}.progress.jsonl", source.display()));
let source_lock_path = PathBuf::from(format!("{}.lock", source.display()));
let canonical_lock = std::fs::canonicalize(&source_lock_path).ok();
let owned_here = SESSION_LOCK_PATHS.get().is_some_and(|paths| {
paths.lock().unwrap().iter().any(|path| {
path == &source_lock_path
|| canonical_lock.as_ref().is_some_and(|canonical| {
std::fs::canonicalize(path).ok().as_ref() == Some(canonical)
})
})
});
let _source_lock = if !owned_here && source_progress.try_exists().map_err(|e| e.to_string())? {
let lock = std::fs::OpenOptions::new()
.create(true)
.write(true)
.open(format!("{}.lock", source.display()))
.map_err(|e| format!("lock import source: {e}"))?;
fs2::FileExt::try_lock_exclusive(&lock).map_err(|_| {
"close the source session before importing its recovery pair".to_owned()
})?;
Some(lock)
} else {
None
};
let staging = tempfile::tempdir_in(&dir).map_err(|e| e.to_string())?;
let staged = staging.path().join("history.stage");
let staged_progress = PathBuf::from(format!("{}.progress.jsonl", staged.display()));
let mut attempts = 0;
let has_progress = loop {
std::fs::copy(source, &staged).map_err(|e| format!("copy import: {e}"))?;
let copied_history = std::fs::read(&staged).map_err(|e| e.to_string())?;
let has_progress = match std::fs::copy(&source_progress, &staged_progress) {
Ok(_) => true,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
let _ = std::fs::remove_file(&staged_progress);
false
}
Err(error) => return Err(format!("copy progress import: {error}")),
};
if copied_history == std::fs::read(source).map_err(|e| e.to_string())? {
break has_progress;
}
attempts += 1;
if attempts >= 4 {
return Err(
"source history changed during import; retry when the session is idle".into(),
);
}
tokio::task::yield_now().await;
};
let fs: Arc<dyn FileSystem> = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(cwd)));
let validated = JsonlSessionStorage::load(
fs.clone(),
&staged.to_string_lossy(),
Arc::new(SystemClock),
Arc::new(DefaultIdGenerator::new()),
)
.await
.map_err(|e| format!("invalid session import: {e}"))?;
validated
.read_progress("")
.await
.map_err(|e| format!("invalid recovery import: {e}"))?;
let mut metadata = validated.jsonl_metadata().clone();
drop(validated);
let text = std::fs::read_to_string(&staged).map_err(|e| e.to_string())?;
let (first, rest) = text
.split_once('\n')
.ok_or_else(|| "import has no complete header".to_owned())?;
let mut header =
rpi_harness::session::jsonl::codec::parse_header(first).map_err(|e| e.to_string())?;
if has_progress && header.progress_format.is_none() {
return Err("history-only artifact has an undeclared recovery companion".into());
}
header.progress_format = Some("assistant-progress-v1".into());
std::fs::write(
&staged,
format!(
"{}{}",
rpi_harness::session::jsonl::codec::encode_header(&header),
rest
),
)
.map_err(|e| e.to_string())?;
if list_session_metadata_in_dir(cwd, &dir)
.await
.map_err(|e| e.to_string())?
.iter()
.any(|saved| saved.id == metadata.id)
{
return Err(format!(
"session {} already exists in {}",
metadata.id,
dir.display()
));
}
let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
fs,
sessions_root: dir.to_string_lossy().into_owned(),
clock: Arc::new(SystemClock),
ids: Arc::new(DefaultIdGenerator::new()),
});
let destination = repo
.create_typed(&JsonlSessionCreateOptions {
id: Some(metadata.id.clone()),
parent_session_id: metadata.parent_session_id.clone(),
cwd: cwd.to_string(),
metadata: metadata.metadata.clone(),
})
.await
.map_err(|e| e.to_string())?;
let path = destination.jsonl_metadata().path.clone();
drop(destination);
if let Err(error) = claim_session_lock(&path) {
return Err(error);
}
let destination_progress = format!("{path}.progress.jsonl");
let publish = (|| -> std::io::Result<()> {
if has_progress {
std::fs::rename(&staged_progress, &destination_progress)?;
}
std::fs::rename(&staged, &path)
})();
if let Err(error) = publish {
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(&destination_progress);
return Err(format!("publish import: {error}"));
}
metadata.path = path;
let storage = repo
.open_by_jsonl_metadata(&metadata)
.await
.map_err(|e| e.to_string())?;
Ok(Session::new(Arc::new(storage), None))
}
static SESSION_LOCK_PATHS: OnceLock<Mutex<HashSet<PathBuf>>> = OnceLock::new();
static SESSION_LOCKS: OnceLock<Mutex<Vec<std::fs::File>>> = OnceLock::new();
fn claim_session_lock(session_path: &str) -> Result<(), String> {
claim_session_lock_inner(session_path, true)
}
fn claim_session_lock_inner(session_path: &str, allow_reentry: bool) -> Result<(), String> {
let canonical =
std::fs::canonicalize(session_path).unwrap_or_else(|_| PathBuf::from(session_path));
let lock_path = PathBuf::from(format!("{}.lock", canonical.display()));
let mut paths = SESSION_LOCK_PATHS
.get_or_init(|| Mutex::new(HashSet::new()))
.lock()
.unwrap();
if paths.contains(&lock_path) {
return if allow_reentry {
Ok(())
} else {
Err("transition session already has a writer in this process".into())
};
}
let file = std::fs::OpenOptions::new()
.create(true)
.write(true)
.open(&lock_path)
.map_err(|e| format!("open session lock {}: {e}", lock_path.display()))?;
if fs2::FileExt::try_lock_exclusive(&file).is_err() {
return Err(format!(
"session {session_path} is already open in another rpi process"
));
}
SESSION_LOCKS
.get_or_init(|| Mutex::new(Vec::new()))
.lock()
.unwrap()
.push(file);
paths.insert(lock_path);
Ok(())
}
pub(crate) struct RestoredSession {
pub session: Session,
pub recorded_cwd: Option<PathBuf>,
}
async fn open_session(
meta: &rpi_harness::session::jsonl::JsonlSessionMetadata,
cwd: &str,
sessions_root: &Path,
) -> Result<RestoredSession, BuildError> {
use rpi_harness::session::jsonl::{JsonlSessionRepo, JsonlSessionRepoOptions};
use rpi_harness::session::types::SessionStorage;
use rpi_tools::FileSystem;
let dir = sessions_root;
let env = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(cwd)));
let fs: Arc<dyn FileSystem> = env.clone();
let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
fs: fs.clone(),
sessions_root: dir.to_string_lossy().into_owned(),
clock: Arc::new(SystemClock),
ids: Arc::new(DefaultIdGenerator::new()),
});
claim_session_lock(&meta.path).map_err(BuildError::SessionOpen)?;
let storage = repo
.open_by_jsonl_metadata(meta)
.await
.map_err(|e| BuildError::SessionOpen(format!("open {}: {e}", meta.path)))?;
let storage = storage
.enable_progress_under_exclusive_lock()
.await
.map_err(|e| BuildError::SessionOpen(format!("upgrade {}: {e}", meta.path)))?;
let storage_arc: Arc<dyn SessionStorage> = Arc::new(storage);
Ok(RestoredSession {
session: Session::new(storage_arc, None),
recorded_cwd: crate::session_cwd::resolve_session_cwd(Path::new(&meta.cwd)),
})
}
pub(crate) fn ephemeral_session() -> Session {
let storage = Arc::new(InMemorySessionStorage::new(
SessionMetadata {
id: "ephemeral".into(),
created_at: 0,
parent_session_id: None,
},
Arc::new(SystemClock),
Arc::new(DefaultIdGenerator::new()),
));
Session::new(storage, None).with_volatile_progress()
}
pub(crate) async fn create_jsonl_session(dir: &Path, cwd: &str) -> Result<Session, String> {
create_jsonl_session_with_id(dir, cwd, None).await
}
pub(crate) async fn create_jsonl_session_with_id(
dir: &Path,
cwd: &str,
id: Option<String>,
) -> Result<Session, String> {
use rpi_harness::session::jsonl::{
JsonlSessionCreateOptions, JsonlSessionRepo, JsonlSessionRepoOptions,
};
use rpi_tools::FileSystem;
let env = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(cwd)));
let fs: Arc<dyn FileSystem> = env.clone();
let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
fs: fs.clone(),
sessions_root: dir.to_string_lossy().into_owned(),
clock: Arc::new(SystemClock),
ids: Arc::new(DefaultIdGenerator::new()),
});
let opts = JsonlSessionCreateOptions {
id, parent_session_id: None,
cwd: cwd.to_string(),
metadata: None,
};
let storage = repo
.create_typed(&opts)
.await
.map_err(|e| format!("create session: {e}"))?;
claim_session_lock(&storage.jsonl_metadata().path)
.map_err(|e| format!("create session: {e}"))?;
let storage_arc: Arc<dyn rpi_harness::session::types::SessionStorage> = Arc::new(storage);
Ok(Session::new(storage_arc, None))
}
fn read_append_target(target: &str, sources: &mut Vec<String>) -> Option<String> {
let path = Path::new(target);
if path.is_file() {
read_prompt_source(path, sources)
} else {
None
}
}
fn read_prompt_source(path: &Path, sources: &mut Vec<String>) -> Option<String> {
let text = std::fs::read_to_string(path).ok()?;
let absolute = if path.is_absolute() {
path.to_path_buf()
} else {
std::env::current_dir()
.map(|cwd| cwd.join(path))
.unwrap_or_else(|_| path.to_path_buf())
};
let source = absolute.to_string_lossy().into_owned();
if !sources.contains(&source) {
sources.push(source);
}
Some(text)
}
#[cfg(test)]
mod goal_outbox_smoke_tests;
#[cfg(all(test, feature = "experimental-native-store"))]
mod native_roundtrip_tests;
#[cfg(all(test, feature = "experimental-native-store"))]
mod native_store_tests;
#[cfg(test)]
mod runtime_session_tests;
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn settlement_bridge_follows_reload_and_releases_its_session_with_the_bus() {
use rpi_harness::events::{HarnessEvent, HarnessEventBus, RunEndEvent, RunEndOutcome};
use std::sync::atomic::{AtomicUsize, Ordering};
static OLD: AtomicUsize = AtomicUsize::new(0);
static NEW: AtomicUsize = AtomicUsize::new(0);
extern "C" fn old(_: rpi_plugin_sdk::StablePluginEvent, _: *mut std::ffi::c_void) -> i32 {
OLD.fetch_add(1, Ordering::SeqCst);
0
}
extern "C" fn new(_: rpi_plugin_sdk::StablePluginEvent, _: *mut std::ffi::c_void) -> i32 {
NEW.fetch_add(1, Ordering::SeqCst);
0
}
let make = |handler: rpi_plugin_sdk::EventHandlerFn| {
let mut registry = rpi_extensions::ExtensionRegistry::new();
registry.register_event_handler(
"settlement".into(),
rpi_plugin_sdk::EventTag::AgentSettled,
handler,
std::ptr::null_mut(),
);
ExtensionSession::from_parts(
Arc::new(registry.snapshot()),
rpi_extensions::PluginKeepalive::empty(),
vec![],
None,
)
};
let sessions = Arc::new(Mutex::new(make(old)));
let weak = Arc::downgrade(&sessions);
let bus = HarnessEventBus::new();
install_extension_settlement_bridge(&bus, sessions.clone());
let ended = |outcome| {
HarnessEvent::RunEnd(RunEndEvent {
lane: "main".into(),
run_id: "run".into(),
outcome,
leaf_id: "leaf".into(),
})
};
bus.emit(&ended(RunEndOutcome::Completed));
assert_eq!(OLD.load(Ordering::SeqCst), 1);
*sessions.lock().unwrap() = make(new);
for outcome in [RunEndOutcome::Aborted, RunEndOutcome::Failed] {
bus.emit(&ended(outcome));
}
assert_eq!(OLD.load(Ordering::SeqCst), 1);
assert_eq!(NEW.load(Ordering::SeqCst), 2);
drop(sessions);
assert!(weak.upgrade().is_some());
drop(bus);
assert!(
weak.upgrade().is_none(),
"settlement observer must not create an ownership cycle"
);
}
#[tokio::test]
async fn unauthenticated_packages_are_not_reintroduced_as_legacy_adapters() {
extern "C" fn request(
_: rpi_plugin_sdk::StbStringRef,
_: *mut rpi_plugin_sdk::StbString,
_: *mut std::ffi::c_void,
) -> i32 {
1
}
let mut registry = rpi_extensions::ExtensionRegistry::new();
for id in ["package", "legacy"] {
registry.register_provider(rpi_extensions::RegisteredProvider {
provider_id: id.into(),
base_url: "http://localhost".into(),
api_style: "openai-completions".into(),
request_fn: request,
stream_fn: None,
plugin_free_string: rpi_plugin_sdk::callbacks::free_string,
user_data: std::ptr::null_mut(),
});
}
assert!(registry.apply_declaration_json(&serde_json::json!({"providers":[{
"id":"package", "requiresAuth":true,
"models":[rpi_ai::Model::new("model", "Model", rpi_ai::Api::OpenaiCompletions, "package", "http://localhost")]
}]}).to_string()));
let session = ExtensionSession::from_parts(
Arc::new(registry.snapshot()),
rpi_extensions::PluginKeepalive::empty(),
vec![],
None,
);
let providers = build_models_with_extensions_for_reload(
&[],
&session,
tokio::runtime::Handle::current(),
);
assert_eq!(
providers
.iter()
.map(|provider| provider.id())
.collect::<Vec<_>>(),
["legacy"]
);
}
use crate::args::Args;
#[test]
fn loaded_context_paths_track_precedence_and_reload() {
const CHILD: &str = "RPI_CONTEXT_PATHS_REGRESSION_CHILD";
if std::env::var_os(CHILD).is_none() {
let dir = tempfile::tempdir().unwrap();
let output = std::process::Command::new(std::env::current_exe().unwrap())
.args([
"--exact",
"session::tests::loaded_context_paths_track_precedence_and_reload",
"--nocapture",
])
.env(CHILD, "1")
.env(crate::config::CONFIG_DIR_ENV, dir.path())
.output()
.unwrap();
assert!(
output.status.success(),
"{}\n{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
return;
}
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap()
.block_on(async {
let agent = crate::config::agent_dir().unwrap();
let cwd = agent.join("project");
let project_config = cwd.join(".rpi");
std::fs::create_dir_all(&project_config).unwrap();
for (path, content) in [
(agent.join("SYSTEM.md"), "GLOBAL_BASE"),
(agent.join("APPEND_SYSTEM.md"), "GLOBAL_APPEND"),
(agent.join("AGENTS.md"), "GLOBAL_INSTRUCTIONS"),
(project_config.join("SYSTEM.md"), "PROJECT_BASE"),
(project_config.join("APPEND_SYSTEM.md"), "PROJECT_APPEND"),
(cwd.join("AGENTS.md"), "PROJECT_INSTRUCTIONS"),
(cwd.join("CLAUDE.md"), "SHADOWED_CLAUDE"),
] {
std::fs::write(path, content).unwrap();
}
let resolved = crate::provider::resolve(
Some("anthropic"),
Some(crate::provider::DEFAULT_MODEL_ID),
None,
Some("test-key"),
None,
)
.unwrap();
let args = Args {
no_session: true,
no_extensions: true,
no_skills: true,
no_prompt_templates: true,
trust_override: Some(true),
..Default::default()
};
let (harness, _events, mut context) =
build(&resolved, &args, &cwd, true).await.unwrap();
let resources = harness.get_resources().await.unwrap();
let path = |p: PathBuf| p.to_string_lossy().into_owned();
assert_eq!(
&resources.context_files[..2],
&[
path(project_config.join("SYSTEM.md")),
path(project_config.join("APPEND_SYSTEM.md")),
]
);
assert!(resources
.context_files
.contains(&path(agent.join("AGENTS.md"))));
assert!(resources
.context_files
.contains(&path(cwd.join("AGENTS.md"))));
assert!(!resources
.context_files
.contains(&path(cwd.join("CLAUDE.md"))));
assert!(!resources
.context_files
.contains(&path(agent.join("SYSTEM.md"))));
let prompt = harness.get_system_prompt().await.unwrap().unwrap();
assert!(prompt.contains("PROJECT_BASE") && prompt.contains("PROJECT_APPEND"));
assert!(!prompt.contains("GLOBAL_BASE") && !prompt.contains("SHADOWED_CLAUDE"));
std::fs::remove_file(project_config.join("SYSTEM.md")).unwrap();
std::fs::remove_file(project_config.join("APPEND_SYSTEM.md")).unwrap();
std::fs::remove_file(cwd.join("AGENTS.md")).unwrap();
assert!(
!reload_extension_resources(&harness, &context)
.await
.had_errors
);
let resources = harness.get_resources().await.unwrap();
assert_eq!(
&resources.context_files[..2],
&[
path(agent.join("SYSTEM.md")),
path(agent.join("APPEND_SYSTEM.md")),
]
);
assert!(resources
.context_files
.contains(&path(cwd.join("CLAUDE.md"))));
assert!(!resources
.context_files
.contains(&path(project_config.join("SYSTEM.md"))));
let explicit = agent.join("custom-append.md");
std::fs::write(&explicit, "EXPLICIT_APPEND").unwrap();
context.args.system_prompt = Some("EXPLICIT_BASE".into());
context.args.append_system_prompt =
vec![path(explicit.clone()), "literal append".into()];
context.args.no_context_files = true;
assert!(
!reload_extension_resources(&harness, &context)
.await
.had_errors
);
assert_eq!(
harness.get_resources().await.unwrap().context_files,
vec![path(explicit)]
);
context.args.system_prompt = None;
context.args.append_system_prompt.clear();
context.args.no_context_files = false;
context.args.trust_override = Some(false);
assert!(
!reload_extension_resources(&harness, &context)
.await
.had_errors
);
assert!(harness
.get_resources()
.await
.unwrap()
.context_files
.is_empty());
let prompt = harness.get_system_prompt().await.unwrap().unwrap();
assert!(
!prompt.contains("SHADOWED_CLAUDE") && !prompt.contains("GLOBAL_INSTRUCTIONS")
);
});
}
#[test]
fn prompt_source_read_failures_and_literals_do_not_create_file_provenance() {
let dir = tempfile::tempdir().unwrap();
let file = dir.path().join("SYSTEM.md");
let mut sources = Vec::new();
assert!(read_prompt_source(&file, &mut sources).is_none());
std::fs::write(&file, [0xff, 0xfe]).unwrap();
assert!(read_prompt_source(&file, &mut sources).is_none());
assert!(read_append_target("literal instructions", &mut sources).is_none());
assert!(sources.is_empty());
std::fs::write(&file, "loaded instructions").unwrap();
assert_eq!(
read_prompt_source(&file, &mut sources).as_deref(),
Some("loaded instructions")
);
assert!(read_prompt_source(&file, &mut sources).is_some());
assert_eq!(sources, vec![file.to_string_lossy().into_owned()]);
}
extern "C" fn deny_test_tool_call(
_event: rpi_plugin_sdk::StablePluginEvent,
_data: *mut std::ffi::c_void,
) -> i32 {
rpi_plugin_sdk::EVENT_HANDLER_ABORT
}
async fn invoke_test_tool_hook(
hook: &rpi_agent::BeforeToolCall,
) -> Option<rpi_agent::BeforeToolCallResult> {
let message = rpi_ai::providers::faux::faux_assistant_message(
"approval",
rpi_ai::types::StopReason::ToolUse,
);
let rpi_ai::types::Content::ToolCall(call) =
rpi_ai::types::Content::tool_call("call", "read", serde_json::json!({"path":"x"}))
else {
unreachable!()
};
hook(
rpi_agent::BeforeToolCallContext {
assistant_message: &message,
tool_call: &call,
args: &call.arguments,
context: &rpi_agent::AgentContext::default(),
},
tokio_util::sync::CancellationToken::new(),
)
.await
}
#[tokio::test]
async fn tool_call_hook_uses_reloaded_session_and_newly_discovered_gate() {
let sessions = Arc::new(Mutex::new(ExtensionSession::none()));
let hook =
extension_tool_call_hook(sessions.clone(), rpi_extensions::UiDialogMailbox::new());
assert!(invoke_test_tool_hook(&hook).await.is_none());
for name in ["first-gate", "reloaded-gate"] {
let mut registry = rpi_extensions::ExtensionRegistry::new();
registry.register_event_handler(
name.into(),
rpi_plugin_sdk::EventTag::ToolCall,
deny_test_tool_call,
std::ptr::null_mut(),
);
let fresh = ExtensionSession::from_parts(
Arc::new(registry.snapshot()),
rpi_extensions::PluginKeepalive::empty(),
Vec::new(),
None,
);
{
let mut current = sessions.lock().unwrap();
if let Some(snapshot) = current.snapshot() {
snapshot.active_flag().store(false, Ordering::SeqCst);
}
*current = fresh;
}
let result = invoke_test_tool_hook(&hook).await.unwrap();
assert!(result.block);
assert!(result.reason.unwrap().contains(name));
}
*sessions.lock().unwrap() = ExtensionSession::none();
assert!(invoke_test_tool_hook(&hook).await.is_none());
}
fn scratch(name: &str) -> PathBuf {
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
let path = std::env::temp_dir().join(format!(
"rpi-summary-{}-{nanos}-{name}.jsonl",
std::process::id()
));
path
}
#[test]
fn summary_marks_a_header_only_session_empty() {
let path = scratch("empty");
std::fs::write(
&path,
"{\"kind\":\"header\",\"version\":4,\"id\":\"abc\",\"createdAt\":1,\"cwd\":\"x\"}\n",
)
.unwrap();
let summary = summarize_session_file(&path);
assert!(summary.empty, "a header-only file has nothing to resume");
assert!(summary.preview.is_none());
assert_eq!(summary.bytes, std::fs::metadata(&path).unwrap().len());
std::fs::remove_file(&path).ok();
}
#[test]
fn summary_reads_the_first_user_prompt_and_name() {
let path = scratch("content");
let body = concat!(
"{\"kind\":\"header\",\"version\":4,\"id\":\"abc\",\"createdAt\":1,\"cwd\":\"x\"}\n",
"{\"kind\":\"fact\",\"seq\":1,\"fact\":\"name\",\"name\":\"fix resume picker\"}\n",
"{\"kind\":\"record\",\"type\":\"operation_started\",\"id\":\"r1\"}\n",
"{\"kind\":\"entry\",\"lane\":\"main\",\"type\":\"message\",\"id\":\"e1\",\"message\":{\"kind\":\"user\",\"role\":\"user\",\"content\":\"read the rpi -r picker\"}}\n",
"{\"kind\":\"entry\",\"lane\":\"main\",\"type\":\"message\",\"id\":\"e2\",\"message\":{\"kind\":\"assistant\",\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"second\"}]}}\n",
);
std::fs::write(&path, body).unwrap();
let summary = summarize_session_file(&path);
assert!(!summary.empty);
assert_eq!(summary.name.as_deref(), Some("fix resume picker"));
assert_eq!(summary.preview.as_deref(), Some("read the rpi -r picker"));
std::fs::remove_file(&path).ok();
}
#[test]
fn summary_extracts_a_block_style_user_prompt() {
let path = scratch("blocks");
let body = concat!(
"{\"kind\":\"header\",\"version\":4,\"id\":\"abc\",\"createdAt\":1,\"cwd\":\"x\"}\n",
"{\"kind\":\"entry\",\"type\":\"message\",\"id\":\"e1\",\"message\":{\"role\":\"user\",\"content\":[{\"type\":\"text\",\"text\":\"line one\\nline two\"}]}}\n",
);
std::fs::write(&path, body).unwrap();
let summary = summarize_session_file(&path);
assert_eq!(summary.preview.as_deref(), Some("line one line two"));
assert_eq!(session_display_label(&summary), "line one line two");
std::fs::remove_file(&path).ok();
}
fn meta_for(path: &Path, id: &str) -> rpi_harness::session::jsonl::JsonlSessionMetadata {
rpi_harness::session::jsonl::JsonlSessionMetadata {
id: id.to_string(),
created_at: 0,
cwd: ".".to_string(),
path: path.to_string_lossy().into_owned(),
modified_at: 0,
source_format: rpi_harness::session::jsonl::JsonlSourceFormat::V4,
parent_session_id: None,
legacy_parent_session_path: None,
metadata: None,
}
}
#[test]
fn continue_skips_newer_empty_sessions() {
let empty = scratch("newest-empty");
std::fs::write(
&empty,
"{\"kind\":\"header\",\"version\":4,\"id\":\"empty\",\"createdAt\":1,\"cwd\":\"x\"}\n",
)
.unwrap();
let content = scratch("older-content");
std::fs::write(
&content,
concat!(
"{\"kind\":\"header\",\"version\":4,\"id\":\"old\",\"createdAt\":1,\"cwd\":\"x\"}\n",
"{\"kind\":\"entry\",\"type\":\"message\",\"id\":\"e1\",\"message\":{\"role\":\"user\",\"content\":\"hello\"}}\n",
),
)
.unwrap();
let metas = vec![meta_for(&empty, "empty"), meta_for(&content, "old")];
let picked = first_resumable_session(&metas).expect("a resumable session");
assert_eq!(picked.id, "old");
let all_empty = vec![meta_for(&empty, "empty")];
assert_eq!(first_resumable_session(&all_empty).unwrap().id, "empty");
std::fs::remove_file(&empty).ok();
std::fs::remove_file(&content).ok();
}
#[tokio::test]
async fn import_preserves_validates_and_locks_the_recovery_pair() {
use rpi_harness::session::jsonl::{JsonlSessionStorage, JsonlV4Header};
use rpi_harness::session::types::{
AssistantFrameOp, AssistantFrameRecord, RecordBase, SessionStorage,
};
let tmp = tempfile::tempdir().unwrap();
let source = tmp.path().join("source.jsonl");
let progress = tmp.path().join("source.jsonl.progress.jsonl");
let fs: Arc<dyn rpi_tools::FileSystem> =
Arc::new(OsExecutionEnv::with_cwd(tmp.path().into()));
let storage = JsonlSessionStorage::create(
fs,
&source.to_string_lossy(),
JsonlV4Header::new("import-pair".into(), 1, "/".into(), None, None, None),
Arc::new(SystemClock),
Arc::new(DefaultIdGenerator::new()),
)
.await
.unwrap();
storage
.append_progress(vec![AssistantFrameRecord {
base: RecordBase {
id: "accepted-prefix".into(),
seq: 0,
lane: "main".into(),
timestamp: 0,
},
run_id: "run".into(),
stream_index: 0,
op: AssistantFrameOp::Append,
frame: Some(serde_json::json!({"payload":"keep"})),
}])
.await
.unwrap();
drop(storage);
let original = std::fs::read(&progress).unwrap();
let destination = tmp.path().join("imported");
let imported = import_session_file(&source, &tmp.path().to_string_lossy(), &destination)
.await
.unwrap();
assert_eq!(
imported.read_progress("run").await.unwrap()[0].base.id,
"accepted-prefix"
);
assert_eq!(
std::fs::read(&progress).unwrap(),
original,
"import must not repair or mutate its source"
);
let lock = std::fs::OpenOptions::new()
.create(true)
.write(true)
.open(tmp.path().join("source.jsonl.lock"))
.unwrap();
fs2::FileExt::try_lock_exclusive(&lock).unwrap();
assert!(import_session_file(
&source,
&tmp.path().to_string_lossy(),
&tmp.path().join("busy")
)
.await
.is_err());
drop(lock);
let mut corrupt = String::from_utf8(original.clone()).unwrap();
corrupt = corrupt.replace(
"\"session_id\":\"import-pair\"",
"\"session_id\":\"foreign\"",
);
assert_ne!(corrupt.as_bytes(), original.as_slice());
std::fs::write(&progress, corrupt).unwrap();
assert!(import_session_file(
&source,
&tmp.path().to_string_lossy(),
&tmp.path().join("foreign")
)
.await
.is_err());
std::fs::remove_file(&progress).unwrap();
assert!(import_session_file(
&source,
&tmp.path().to_string_lossy(),
&tmp.path().join("missing")
)
.await
.is_err());
}
#[test]
fn list_session_files_sync_walks_the_cwd_subdirectory() {
let root = std::env::temp_dir().join(format!("rpi-summary-{}-dir", std::process::id()));
let nested = root.join("--proj--");
std::fs::create_dir_all(&nested).unwrap();
let file = nested.join("2026-01-01_a.jsonl");
std::fs::write(&file, "{}\n").unwrap();
let sidecar = nested.join("2026-01-01_a.jsonl.progress.jsonl");
std::fs::write(&sidecar, "{}\n").unwrap();
let found = list_session_files_sync(&root);
assert!(found.iter().any(|p| p == &file));
assert!(!found.contains(&sidecar));
std::fs::remove_dir_all(&root).ok();
}
fn default_prompt_tools() -> Vec<String> {
BUILTIN_TOOL_NAMES
.iter()
.copied()
.chain(["todo"])
.map(str::to_owned)
.collect()
}
#[test]
fn default_prompt_lists_only_the_active_tools() {
let tools = ["read", "grep", "find", "ls"].map(str::to_owned);
let p = default_system_prompt_with_tools("/tmp/proj", true, &tools);
assert!(p.contains("- read — Read file contents"));
assert!(p.contains("- grep — Search file contents"));
for absent in ["- edit", "- write", "- bash", "- docs", "todo", "Windows"] {
assert!(
!p.contains(absent),
"{absent} must not appear for a read-only run: {p}"
);
}
assert!(p.contains("Re-stating the same plan without acting on it is a bug"));
let none = default_system_prompt_with_tools("/tmp/proj", false, &[]);
assert!(none.contains("Available tools:\n(none)"));
}
#[test]
fn default_prompt_mentions_cwd_and_tools() {
let p = default_system_prompt_with_tools("/tmp/proj", false, &default_prompt_tools());
assert!(p.contains("/tmp/proj"));
assert!(p.contains("read"));
assert!(p.contains("bash"));
assert!(p.contains("edit"));
assert!(p.contains("write"));
assert!(p.contains("docs"));
assert!(!p.contains("- grep"));
assert!(!p.contains("- find"));
assert!(!p.contains("- ls"));
assert!(!p.contains("- powershell"));
assert!(p.contains("Bash/POSIX syntax, also on Windows"));
}
#[test]
fn default_prompt_instructs_language_matching() {
let p = default_system_prompt_with_tools("/tmp/proj", false, &default_prompt_tools());
assert!(
p.contains("same language as the user's latest message"),
"the prompt must tell the model to mirror the user's language: {p}"
);
}
#[test]
fn default_prompt_keeps_working_state_in_the_visible_channel() {
let p = default_system_prompt_with_tools("/tmp/proj", false, &default_prompt_tools());
let lower = p.to_lowercase();
assert!(
lower.contains("reasoning is not") && lower.contains("carried into your next turn"),
"the prompt must explain that reasoning does not survive the turn: {p}"
);
assert!(
lower.contains("visible") && !lower.contains("todo tool"),
"the prompt must keep state visible without inventing extension rules: {p}"
);
assert!(
lower.contains("re-stating the same plan"),
"the prompt must call out plan re-statement as a bug: {p}"
);
assert!(
lower.contains("already shown earlier"),
"the prompt must discourage re-fetching known content: {p}"
);
}
#[test]
fn default_prompt_does_not_claim_reasoning_is_lost_when_it_is_replayed() {
let p = default_system_prompt_with_tools("/tmp/proj", true, &default_prompt_tools());
let lower = p.to_lowercase();
assert!(
lower.contains("carried back"),
"a replaying model must be told its reasoning survives: {p}"
);
assert!(
!lower.contains("reasoning is not"),
"the replayed variant must not claim reasoning is dropped: {p}"
);
assert!(
lower.contains("do not re-derive"),
"the replayed variant must forbid re-deriving the plan: {p}"
);
assert!(
lower.contains("do not re-derive") && lower.contains("re-stating the same plan"),
"the anti-repetition rule must survive the flip: {p}"
);
assert!(
lower.contains("already shown earlier"),
"the re-read rule must survive the flip: {p}"
);
}
#[test]
fn default_prompt_states_each_rule_once() {
for replayed in [false, true] {
let p =
default_system_prompt_with_tools("/tmp/proj", replayed, &default_prompt_tools());
for phrase in ["the same plan", "Windows", "guessing"] {
assert_eq!(p.matches(phrase).count(), 1, "{phrase:?} repeated: {p}");
}
assert!(
!p.contains("todo tool"),
"extension rules require explicit metadata"
);
}
}
#[test]
fn only_read_only_tools_are_replayable() {
let env = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(".")));
let env_dyn: Arc<dyn rpi_tools::ExecutionEnv> = env.clone();
let mut_env: Arc<dyn rpi_tools::MutatingEnv> = env.clone();
let context = ExecutionToolContext::new(env_dyn, Some(mut_env));
let tools = build_tools(&context, &Args::default());
let replay_of = |name: &str| {
tools
.iter()
.find(|tool| tool.tool.schema().name == name)
.map(|tool| tool.replay)
};
for read_only in ["read", "docs"] {
assert_eq!(
replay_of(read_only),
Some(ToolReplay::Safe),
"{read_only} is side-effect free, so recovery may re-run it"
);
}
for mutating in ["bash", "edit", "write", "powershell"] {
if let Some(replay) = replay_of(mutating) {
assert_eq!(
replay,
ToolReplay::Never,
"{mutating} must never be replayed: re-running it would duplicate its effect"
);
}
}
}
#[test]
fn default_tools_include_docs_lookup() {
let env = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(".")));
let env_dyn: Arc<dyn rpi_tools::ExecutionEnv> = env.clone();
let mut_env: Arc<dyn rpi_tools::MutatingEnv> = env.clone();
let context = ExecutionToolContext::new(env_dyn, Some(mut_env));
let names: Vec<String> = build_tools(&context, &Args::default())
.iter()
.map(|tool| tool.tool.schema().name.clone())
.collect();
assert!(names.contains(&"docs".to_string()));
for legacy in ["grep", "find", "ls"] {
assert!(
!names.contains(&legacy.to_string()),
"legacy tool {legacy} should not be registered by default"
);
}
}
#[test]
fn reload_settings_preflight_respects_trust() {
struct RestoreConfigDir(Option<std::ffi::OsString>);
impl Drop for RestoreConfigDir {
fn drop(&mut self) {
match self.0.take() {
Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
}
}
}
let _guard = crate::config::test_support::env_lock().lock().unwrap();
let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
let tmp = tempfile::tempdir().unwrap();
let agent = tmp.path().join("agent");
let cwd = tmp.path().join("project");
std::fs::create_dir_all(&agent).unwrap();
std::fs::create_dir_all(cwd.join(".rpi")).unwrap();
std::fs::write(agent.join("settings.json"), "{}").unwrap();
std::fs::write(cwd.join(".rpi/settings.json"), "{ malformed").unwrap();
std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);
let args = Args::default();
let trusted = validate_settings_for_reload(&args, &cwd, true);
let untrusted = validate_settings_for_reload(&args, &cwd, false);
assert!(trusted
.unwrap_err()
.contains("could not load project settings"));
assert!(untrusted.is_ok());
std::fs::write(agent.join("settings.json"), "{ malformed").unwrap();
assert!(validate_settings_for_reload(&args, &cwd, false)
.unwrap_err()
.contains("could not load global settings"));
let local_only = Args {
dev_local_only: true,
..Args::default()
};
assert!(validate_settings_for_reload(&local_only, &cwd, true).is_ok());
}
#[tokio::test(flavor = "current_thread")]
async fn reload_preserves_live_resources_when_settings_break() {
struct RestoreConfigDir(Option<std::ffi::OsString>);
impl Drop for RestoreConfigDir {
fn drop(&mut self) {
match self.0.take() {
Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
}
}
}
let _guard = crate::config::test_support::env_lock().lock().unwrap();
let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
let tmp = tempfile::tempdir().unwrap();
let agent = tmp.path().join("agent");
let cwd = tmp.path().join("project");
let skill_dir = tmp.path().join("configured-skills");
std::fs::create_dir_all(&agent).unwrap();
std::fs::create_dir_all(&cwd).unwrap();
std::fs::create_dir_all(&skill_dir).unwrap();
std::fs::write(
skill_dir.join("SKILL.md"),
"---\nname: keep-me\ndescription: Reload sentinel\n---\nKeep this skill loaded.",
)
.unwrap();
std::fs::write(
agent.join("settings.json"),
serde_json::json!({"skillDirs": [skill_dir]}).to_string(),
)
.unwrap();
std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);
let resolved = crate::provider::resolve(
Some("anthropic"),
Some(crate::provider::DEFAULT_MODEL_ID),
None,
Some("test-key"),
None,
)
.unwrap();
let args = Args {
trust_override: Some(false),
no_session: true,
no_extensions: true,
no_prompt_templates: true,
no_context_files: true,
system_prompt: Some("stable system prompt".into()),
..Args::default()
};
let (harness, _events, context) = build(&resolved, &args, &cwd, false).await.unwrap();
let before_resources = harness.get_resources().await.unwrap();
assert_eq!(before_resources.skills.as_ref().unwrap().len(), 1);
assert_eq!(before_resources.skills.as_ref().unwrap()[0].name, "keep-me");
let before_prompt = harness.get_system_prompt().await.unwrap();
let before_tools: Vec<String> = harness
.get_tools()
.await
.unwrap()
.iter()
.map(|tool| tool.tool.schema().name.clone())
.collect();
let before_bridge = context
.action_bridge
.lock()
.unwrap()
.as_ref()
.unwrap()
.clone();
std::fs::write(agent.join("settings.json"), "{ malformed").unwrap();
let outcome = reload_extension_resources(&harness, &context).await;
assert!(outcome.had_warnings);
assert!(outcome.summary.contains("Settings reload failed"));
assert!(outcome.summary.contains("could not load global settings"));
assert_eq!(outcome.project_trusted, None);
assert_eq!(harness.get_resources().await.unwrap(), before_resources);
assert_eq!(harness.get_system_prompt().await.unwrap(), before_prompt);
let after_tools: Vec<String> = harness
.get_tools()
.await
.unwrap()
.iter()
.map(|tool| tool.tool.schema().name.clone())
.collect();
assert_eq!(after_tools, before_tools);
let after_bridge = context
.action_bridge
.lock()
.unwrap()
.as_ref()
.unwrap()
.clone();
assert!(Arc::ptr_eq(&after_bridge, &before_bridge));
}
#[tokio::test(flavor = "current_thread")]
async fn reload_outcome_carries_the_current_project_trust() {
struct RestoreConfigDir(Option<std::ffi::OsString>);
impl Drop for RestoreConfigDir {
fn drop(&mut self) {
match self.0.take() {
Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
}
}
}
let _guard = crate::config::test_support::env_lock().lock().unwrap();
let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
let tmp = tempfile::tempdir().unwrap();
let agent = tmp.path().join("agent");
let cwd = tmp.path().join("project");
std::fs::create_dir_all(&agent).unwrap();
std::fs::create_dir_all(&cwd).unwrap();
std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);
crate::config::set_project_trust(&cwd, Some(true)).unwrap();
let resolved = crate::provider::resolve(
Some("anthropic"),
Some(crate::provider::DEFAULT_MODEL_ID),
None,
Some("test-key"),
None,
)
.unwrap();
let args = Args {
no_session: true,
no_extensions: true,
no_prompt_templates: true,
no_context_files: true,
system_prompt: Some("stable system prompt".into()),
..Args::default()
};
let (harness, _events, context) = build(&resolved, &args, &cwd, true).await.unwrap();
assert!(context.project_trusted, "startup snapshot is trusted");
crate::config::set_project_trust(&cwd, Some(false)).unwrap();
let outcome = reload_extension_resources(&harness, &context).await;
assert!(!outcome.had_errors, "{}", outcome.summary);
assert_eq!(outcome.project_trusted, Some(false));
crate::config::set_project_trust(&cwd, Some(true)).unwrap();
let outcome = reload_extension_resources(&harness, &context).await;
assert!(!outcome.had_errors, "{}", outcome.summary);
assert_eq!(outcome.project_trusted, Some(true));
}
#[test]
fn reload_preparation_rejects_settings_changed_during_derivation() {
struct RestoreConfigDir(Option<std::ffi::OsString>);
impl Drop for RestoreConfigDir {
fn drop(&mut self) {
match self.0.take() {
Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
}
}
}
let _guard = crate::config::test_support::env_lock().lock().unwrap();
let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
let tmp = tempfile::tempdir().unwrap();
let agent = tmp.path().join("agent");
let cwd = tmp.path().join("project");
let first_skill_dir = tmp.path().join("first-skills");
let second_skill_dir = tmp.path().join("second-skills");
std::fs::create_dir_all(&agent).unwrap();
std::fs::create_dir_all(&cwd).unwrap();
std::fs::write(
agent.join("settings.json"),
serde_json::json!({"skillDirs": [first_skill_dir]}).to_string(),
)
.unwrap();
std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);
let args = Args {
trust_override: Some(false),
no_extensions: true,
no_prompt_templates: true,
..Args::default()
};
let result = prepare_reload_inputs_inner(&args, &cwd, false, || {
std::fs::write(
agent.join("settings.json"),
serde_json::json!({"skillDirs": [second_skill_dir]}).to_string(),
)
.unwrap();
});
let error = result
.err()
.expect("settings mutation must fail preparation");
assert!(error.contains("settings changed while reload inputs were being prepared"));
}
#[tokio::test(flavor = "current_thread")]
async fn reload_uses_frozen_settings_inputs_after_preparation() {
struct RestoreConfigDir(Option<std::ffi::OsString>);
impl Drop for RestoreConfigDir {
fn drop(&mut self) {
match self.0.take() {
Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
}
}
}
let _guard = crate::config::test_support::env_lock().lock().unwrap();
let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
let tmp = tempfile::tempdir().unwrap();
let agent = tmp.path().join("agent");
let cwd = tmp.path().join("project");
let skill_dir = tmp.path().join("configured-skills");
std::fs::create_dir_all(&agent).unwrap();
std::fs::create_dir_all(&cwd).unwrap();
std::fs::create_dir_all(&skill_dir).unwrap();
std::fs::write(
skill_dir.join("SKILL.md"),
"---\nname: frozen-skill\ndescription: Reload sentinel\n---\nFrozen input.",
)
.unwrap();
std::fs::write(
agent.join("settings.json"),
serde_json::json!({"skillDirs": [skill_dir]}).to_string(),
)
.unwrap();
std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);
let resolved = crate::provider::resolve(
Some("anthropic"),
Some(crate::provider::DEFAULT_MODEL_ID),
None,
Some("test-key"),
None,
)
.unwrap();
let args = Args {
trust_override: Some(false),
no_session: true,
no_extensions: true,
no_prompt_templates: true,
no_context_files: true,
system_prompt: Some("stable system prompt".into()),
..Args::default()
};
let (harness, _events, context) = build(&resolved, &args, &cwd, false).await.unwrap();
let outcome = reload_extension_resources_inner(&harness, &context, || {
std::fs::write(agent.join("settings.json"), "{ malformed").unwrap();
})
.await;
assert!(!outcome.summary.contains("Settings reload failed"));
assert_eq!(
outcome.summary,
"Reloaded 0 plugin(s), 1 skill(s), 0 prompt(s)."
);
let resources = harness.get_resources().await.unwrap();
let skills = resources.skills.as_ref().unwrap();
assert_eq!(skills.len(), 1);
assert_eq!(skills[0].name, "frozen-skill");
}
#[tokio::test]
async fn reload_resource_paths_include_explicit_skill_and_prompt_files() {
let tmp = tempfile::tempdir().unwrap();
let skill_path = tmp.path().join("explicit-skill.md");
std::fs::write(
&skill_path,
"---\nname: explicit-skill\ndescription: Explicit skill\n---\nSkill body",
)
.unwrap();
let prompt_path = tmp.path().join("explicit-prompt.md");
std::fs::write(
&prompt_path,
"---\ndescription: Explicit prompt\n---\nPrompt body",
)
.unwrap();
let args = Args {
skill: vec![skill_path.clone()],
prompt_template: vec![prompt_path.clone()],
..Args::default()
};
let skill_paths = append_reload_resource_paths(Vec::new(), &args.skill, &[], &[], &[]);
let prompt_paths =
append_reload_resource_paths(Vec::new(), &args.prompt_template, &[], &[], &[]);
let env = Arc::new(OsExecutionEnv::with_cwd(tmp.path().to_path_buf()));
let env_dyn: Arc<dyn rpi_tools::ExecutionEnv> = env;
let skills = load_skills_with_precedence(&env_dyn, &skill_paths).await;
assert_eq!(skills.skills.len(), 1, "{:?}", skills.diagnostics);
assert_eq!(skills.skills[0].name, "explicit-skill");
let prompts = load_prompt_templates_with_precedence(&env_dyn, &prompt_paths).await;
assert_eq!(
prompts.prompt_templates.len(),
1,
"{:?}",
prompts.diagnostics
);
assert_eq!(prompts.prompt_templates[0].name, "explicit-prompt");
}
#[test]
fn tool_policy_applies_to_rust_and_js_active_names() {
let names = vec![
"read".to_string(),
"ask_user_question".to_string(),
"write".to_string(),
];
let args = Args {
tools: Some(vec!["read".into(), "ask_user_question".into()]),
..Args::default()
};
assert_eq!(
filter_active_tool_names(names.clone(), &args),
vec!["read", "ask_user_question"]
);
let args = Args {
exclude_tools: Some(vec!["ask_user_question".into()]),
..Args::default()
};
assert_eq!(
filter_active_tool_names(names.clone(), &args),
vec!["read", "write"]
);
let args = Args {
no_tools: true,
..Args::default()
};
assert!(filter_active_tool_names(names, &args).is_empty());
}
#[test]
fn select_ephemeral_when_no_session() {
let args = Args {
no_session: true,
..Args::default()
};
let cwd = Path::new("/tmp");
assert!(matches!(
select_session(&args, cwd),
SessionSelection::Ephemeral
));
}
#[test]
fn project_resources_load_by_default_and_allow_explicit_opt_out() {
let defaults = Args::default();
assert!(resolve_project_trust(
&defaults,
Path::new("C:/definitely-not-a-project")
));
let denied = Args {
trust_override: Some(false),
..Args::default()
};
assert!(!resolve_project_trust(
&denied,
Path::new("C:/definitely-not-a-project")
));
}
#[tokio::test(flavor = "current_thread")]
async fn build_preserves_the_supplied_startup_project_snapshot() {
struct RestoreConfigDir(Option<std::ffi::OsString>);
impl Drop for RestoreConfigDir {
fn drop(&mut self) {
match self.0.take() {
Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
}
}
}
let _guard = crate::config::test_support::env_lock().lock().unwrap();
let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
let tmp = tempfile::tempdir().unwrap();
let agent = tmp.path().join("agent");
let cwd = tmp.path().join("project");
std::fs::create_dir_all(&agent).unwrap();
std::fs::create_dir_all(&cwd).unwrap();
std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);
let resolved = crate::provider::resolve(
Some("anthropic"),
Some(crate::provider::DEFAULT_MODEL_ID),
None,
Some("test-key"),
None,
)
.unwrap();
let args = Args {
trust_override: Some(false),
no_session: true,
no_tools: true,
no_extensions: true,
no_skills: true,
no_prompt_templates: true,
no_context_files: true,
system_prompt: Some("test prompt".into()),
..Args::default()
};
let (_, _, context) = build(&resolved, &args, &cwd, true).await.unwrap();
assert_eq!(context.cwd, cwd);
assert!(context.project_trusted);
}
#[tokio::test(flavor = "current_thread")]
async fn build_applies_the_name_flag_to_the_session() {
struct RestoreConfigDir(Option<std::ffi::OsString>);
impl Drop for RestoreConfigDir {
fn drop(&mut self) {
match self.0.take() {
Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
}
}
}
let _guard = crate::config::test_support::env_lock().lock().unwrap();
let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
let tmp = tempfile::tempdir().unwrap();
let agent = tmp.path().join("agent");
let cwd = tmp.path().join("project");
std::fs::create_dir_all(&agent).unwrap();
std::fs::create_dir_all(&cwd).unwrap();
std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);
let resolved = crate::provider::resolve(
Some("anthropic"),
Some(crate::provider::DEFAULT_MODEL_ID),
None,
Some("test-key"),
None,
)
.unwrap();
let args = Args {
trust_override: Some(false),
no_tools: true,
no_extensions: true,
no_skills: true,
no_prompt_templates: true,
no_context_files: true,
system_prompt: Some("test prompt".into()),
name: Some("smoke name".into()),
session_dir: Some(cwd.join("sessions")),
..Args::default()
};
let (harness, _, _) = build(&resolved, &args, &cwd, true).await.unwrap();
assert_eq!(
harness.get_name().await.unwrap().as_deref(),
Some("smoke name")
);
}
#[test]
fn select_latest_for_continue_and_resume() {
let _env = crate::config::test_support::SessionDirEnv::new(None);
let args = Args {
continue_session: true,
..Args::default()
};
let cwd = Path::new("/tmp");
assert!(matches!(
select_session(&args, cwd),
SessionSelection::Latest { .. }
));
let args = Args {
resume: true,
..Args::default()
};
assert!(matches!(
select_session(&args, cwd),
SessionSelection::Latest { .. }
));
}
#[test]
fn select_by_id_for_session_flag() {
let _env = crate::config::test_support::SessionDirEnv::new(None);
let args = Args {
session: Some("01a02ece".into()),
..Args::default()
};
let cwd = Path::new("/tmp");
assert!(matches!(
select_session(&args, cwd),
SessionSelection::ById { id, .. } if id == "01a02ece"
));
}
#[test]
fn select_new_with_custom_dir() {
let _env = crate::config::test_support::SessionDirEnv::new(None);
let args = Args {
session_dir: Some(PathBuf::from("/tmp/sess")),
..Args::default()
};
let cwd = Path::new("/tmp");
match select_session(&args, cwd) {
SessionSelection::New { dir, .. } => assert_eq!(dir, PathBuf::from("/tmp/sess")),
other => panic!("expected New, got {other:?}"),
}
}
#[test]
fn select_new_default_dir() {
let _env = crate::config::test_support::SessionDirEnv::new(None);
let args = Args::default();
let cwd = Path::new("/proj");
match select_session(&args, cwd) {
SessionSelection::New { dir, .. } => {
assert_eq!(dir, Path::new("/proj/.rpi/sessions"));
}
other => panic!("expected New, got {other:?}"),
}
}
#[test]
fn default_session_dir_only_uses_rpi() {
let _env = crate::config::test_support::SessionDirEnv::new(None);
let tmp = tempfile::tempdir().unwrap();
let cwd = tmp.path();
std::fs::create_dir_all(cwd.join(".rpi/sessions")).unwrap();
assert_eq!(default_session_dir(cwd), cwd.join(".rpi/sessions"));
std::fs::create_dir_all(cwd.join(".rpi/sessions")).unwrap();
assert_eq!(default_session_dir(cwd), cwd.join(".rpi/sessions"));
}
#[test]
fn default_session_dir_uses_env_var() {
let tmp = tempfile::tempdir().unwrap();
let cwd = tmp.path();
let custom = tmp.path().join("custom_sessions");
let _env = crate::config::test_support::SessionDirEnv::new(Some(&custom));
assert_eq!(default_session_dir(cwd), custom);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn ephemeral_session_builds_roundtrips() {
let s = ephemeral_session();
let leaf = s.get_leaf_id().await;
assert!(leaf.is_ok());
}
#[test]
fn host_private_env_matches_the_declared_contract() {
assert_eq!(rpi_plugin_sdk::SESSION_ID_ENV, "RPI_SESSION_ID");
assert_eq!(
rpi_plugin_sdk::SUBAGENT_PARENT_ENV,
[
"LANGFUSE_PI_PARENT_TRACE_ID",
"LANGFUSE_PI_PARENT_SPAN_ID",
"LANGFUSE_PI_PARENT_SESSION_ID",
"LANGFUSE_PI_PARENT_DEPTH",
],
"these strings are what rpi-langfuse (a different repo) matches; a \
change here is a breaking cross-repo change, not a refactor"
);
for name in rpi_plugin_sdk::SUBAGENT_PARENT_ENV
.iter()
.chain(std::iter::once(&rpi_plugin_sdk::SESSION_ID_ENV))
{
assert!(
rpi_tools::is_host_private_env(name),
"`{name}` must not reach a spawned shell, but rpi-tools excludes only {:?}",
rpi_tools::HOST_PRIVATE_ENV
);
}
for visible in [
crate::args::RPI_OFFLINE_ENV,
crate::config::CONFIG_DIR_ENV,
"PATH",
] {
assert!(
!rpi_tools::is_host_private_env(visible),
"{visible} describes the user's configuration and must still reach the shell"
);
}
}
#[tokio::test]
async fn reload_shutdown_joins_all_handlers_even_after_veto() {
use rpi_plugin_sdk::{EventTag, StablePluginEvent};
use std::sync::atomic::AtomicUsize;
static CONTEXTS: rpi_plugin_sdk::handles::ToolHandles<Arc<AtomicUsize>> =
rpi_plugin_sdk::handles::ToolHandles::new();
struct ContextOwner(usize);
impl Drop for ContextOwner {
fn drop(&mut self) {
drop(CONTEXTS.remove(self.0 as rpi_plugin_sdk::StepHandle));
}
}
extern "C" fn stop(_: StablePluginEvent, data: *mut std::ffi::c_void) -> i32 {
let context = CONTEXTS.get(data).expect("live test context");
let calls = context.lock().clone();
std::thread::sleep(std::time::Duration::from_millis(20));
if calls.fetch_add(1, Ordering::SeqCst) == 0 {
rpi_plugin_sdk::EVENT_HANDLER_ABORT
} else {
0
}
}
let calls = Arc::new(AtomicUsize::new(0));
let context = CONTEXTS.insert(calls.clone());
let _owner = ContextOwner(context as usize);
let mut registry = rpi_extensions::ExtensionRegistry::new();
for name in ["first", "second"] {
registry.register_event_handler(name.into(), EventTag::SessionShutdown, stop, context);
}
super::shutdown_extensions_for_reload(
Some(Arc::new(registry.snapshot())),
rpi_extensions::PluginKeepalive::empty(),
)
.await
.unwrap();
assert_eq!(calls.load(Ordering::SeqCst), 2);
}
#[tokio::test]
async fn a_restored_session_adopts_its_recorded_directory() {
let tmp = tempfile::tempdir().unwrap();
let project = tmp.path().join("project-a");
let launch = tmp.path().join("launch-dir");
let sessions = tmp.path().join("sessions");
std::fs::create_dir_all(&project).unwrap();
std::fs::create_dir_all(&launch).unwrap();
std::fs::create_dir_all(&sessions).unwrap();
create_jsonl_session_with_id(
&sessions,
&project.to_string_lossy(),
Some("cwdadopt01".into()),
)
.await
.unwrap();
let selection = SessionSelection::ById {
id: "cwdadopt01".into(),
dir: sessions.clone(),
};
let peeked = peek_restored_session_cwd(&selection, &launch)
.await
.expect("a resumable session with a recorded cwd");
assert_eq!(peeked.recorded, project);
assert_eq!(peeked.adopted, Some(project));
}
#[tokio::test]
async fn a_missing_recorded_directory_is_reported_not_adopted() {
let tmp = tempfile::tempdir().unwrap();
let gone = tmp.path().join("project-gone");
let launch = tmp.path().join("launch-dir");
let sessions = tmp.path().join("sessions");
std::fs::create_dir_all(&launch).unwrap();
std::fs::create_dir_all(&sessions).unwrap();
create_jsonl_session_with_id(
&sessions,
&gone.to_string_lossy(),
Some("cwdgone001".into()),
)
.await
.unwrap();
let selection = SessionSelection::ById {
id: "cwdgone001".into(),
dir: sessions.clone(),
};
let peeked = peek_restored_session_cwd(&selection, &launch)
.await
.expect("the session itself is resumable");
assert_eq!(peeked.recorded, gone);
assert_eq!(peeked.adopted, None);
let note = crate::session_cwd::missing_session_cwd_note(&peeked.recorded, &launch);
assert!(note.contains("no longer exists"), "{note}");
assert!(note.contains("launch-dir"), "{note}");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn legacy_progress_upgrade_normal_open_runs_and_reopens() {
use rpi_ai::providers::faux::{FauxProvider, FauxScript};
use rpi_harness::agent_harness::{AgentHarness, AgentLane, HarnessRunOutcome};
use rpi_harness::session::jsonl::{
encode_header, JsonlSessionCreateOptions, JsonlSessionRepo, JsonlSessionRepoOptions,
JsonlV4Header,
};
let tmp = tempfile::tempdir().unwrap();
let sessions = tmp.path().join("sessions");
std::fs::create_dir_all(&sessions).unwrap();
let cwd = tmp.path().to_string_lossy().into_owned();
let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
fs: Arc::new(OsExecutionEnv::with_cwd(tmp.path().into())),
sessions_root: sessions.to_string_lossy().into_owned(),
clock: Arc::new(SystemClock),
ids: Arc::new(DefaultIdGenerator::new()),
});
let fixture = repo
.create_typed(&JsonlSessionCreateOptions {
id: Some("legacycli001".into()),
parent_session_id: None,
cwd: cwd.clone(),
metadata: None,
})
.await
.unwrap();
let path = PathBuf::from(fixture.jsonl_metadata().path);
drop(fixture);
let sidecar = PathBuf::from(format!("{}.progress.jsonl", path.display()));
std::fs::remove_file(&sidecar).unwrap();
let header = JsonlV4Header::new("legacycli001".into(), 1, cwd.clone(), None, None, None);
std::fs::write(&path, encode_header(&header)).unwrap();
assert!(!sidecar.exists());
for _ in 0..2 {
let restored = open_session_by_id_in_dir("legacycli001", &cwd, &sessions)
.await
.map_err(|e| e.to_string())
.unwrap();
assert_eq!(restored.session.storage().metadata().id, "legacycli001");
let provider =
FauxProvider::new(FauxScript::new().with_text("continued legacy session"));
let harness = AgentHarness::create(AgentHarnessOptions {
model: provider.default_model().clone(),
models: vec![provider],
session: restored.session,
allow_existing_session: true,
..Default::default()
})
.await
.unwrap();
let run = harness.prompt_text("continue", vec![]).await.unwrap();
assert!(matches!(run.outcome, HarnessRunOutcome::Completed { .. }));
}
let content = std::fs::read_to_string(&path).unwrap();
assert!(content.lines().next().unwrap().contains("progressFormat"));
assert!(sidecar.exists());
}
#[tokio::test]
async fn open_session_reports_the_recorded_directory() {
let tmp = tempfile::tempdir().unwrap();
let project = tmp.path().join("project-open");
let launch = tmp.path().join("launch-dir");
let sessions = tmp.path().join("sessions");
std::fs::create_dir_all(&project).unwrap();
std::fs::create_dir_all(&launch).unwrap();
std::fs::create_dir_all(&sessions).unwrap();
create_jsonl_session_with_id(
&sessions,
&project.to_string_lossy(),
Some("cwdopen001".into()),
)
.await
.unwrap();
let restored =
open_session_by_id_in_dir("cwdopen001", &launch.to_string_lossy(), &sessions)
.await
.map_err(|error| error.to_string())
.unwrap();
assert_eq!(restored.recorded_cwd, Some(project.clone()));
assert_eq!(restored.session.storage().metadata().id, "cwdopen001");
}
#[tokio::test]
async fn the_peeked_session_is_the_one_the_build_opens() {
let tmp = tempfile::tempdir().unwrap();
let first_project = tmp.path().join("project-first");
let second_project = tmp.path().join("project-second");
let launch = tmp.path().join("launch-dir");
let sessions = tmp.path().join("sessions");
for dir in [&first_project, &second_project, &launch, &sessions] {
std::fs::create_dir_all(dir).unwrap();
}
create_jsonl_session_with_id(
&sessions,
&first_project.to_string_lossy(),
Some("cwdrace001".into()),
)
.await
.unwrap();
let mut args = Args {
continue_session: true,
session_dir: Some(sessions.clone()),
..Default::default()
};
let selection = select_session_in_dir(&args, sessions.clone());
let peeked = peek_restored_session_cwd(&selection, &launch)
.await
.expect("a resumable session");
assert_eq!(peeked.adopted, Some(first_project.clone()));
create_jsonl_session_with_id(
&sessions,
&second_project.to_string_lossy(),
Some("cwdrace002".into()),
)
.await
.unwrap();
pin_restored_selection(&mut args, &peeked);
let pinned = select_session_in_dir(&args, sessions.clone());
let reopened = resolve_restore_target(&pinned, &launch.to_string_lossy())
.await
.unwrap()
.expect("the pinned session still exists");
assert_eq!(
reopened.id, peeked.id,
"the build must open the session the peek adopted the directory of"
);
assert_eq!(reopened.cwd, first_project.to_string_lossy());
}
#[test]
fn trust_is_resolved_per_directory() {
struct RestoreConfigDir(Option<std::ffi::OsString>);
impl Drop for RestoreConfigDir {
fn drop(&mut self) {
match self.0.take() {
Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
}
}
}
let _guard = crate::config::test_support::env_lock().lock().unwrap();
let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
let tmp = tempfile::tempdir().unwrap();
let agent = tmp.path().join("agent");
std::fs::create_dir_all(&agent).unwrap();
std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);
let untrusted = tmp.path().join("project-untrusted");
let launch = tmp.path().join("project-launch");
std::fs::create_dir_all(&untrusted).unwrap();
std::fs::create_dir_all(&launch).unwrap();
crate::config::set_project_trust(&untrusted, Some(false)).unwrap();
let args = Args::default();
assert!(!resolve_project_trust(&args, &untrusted));
assert!(resolve_project_trust(&args, &launch));
}
}