Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316//! Session-local plugin composition. Kernel callbacks capture only immutable identity//! and the host Arc: they never wait for the engine/session mutex held by a tool turn.
use std::{ collections::BTreeMap, path::PathBuf, sync::{Arc, Mutex, PoisonError},};
use anyhow::{anyhow, bail, Context};use serde::Deserialize;use serde_json::{json, Value};use tokio_util::sync::CancellationToken;
use super::{ engine::SessionEngine, slash_commands::{parse_session_command, SessionSlashCommand}, AgentSession,};use crate::{ kernel::shared::{host_handler, HostRequestHandlers}, plugins::{ CapabilityFuture, CapabilityRequest, Contribution, ContributionKind, ContributionSnapshot, HostCapabilities, InvocationEvent, InvocationMode, InvocationOutcome, PluginHost, PluginSource, }, settings::SettingsManager,};
struct SessionFacts { info: Value, delegates: Mutex<BTreeMap<String, Arc<dyn HostCapabilities>>>,}
impl HostCapabilities for SessionFacts { fn call( &self, request: CapabilityRequest, cancellation: CancellationToken, ) -> CapabilityFuture { let info = self.info.clone(); let delegate = self .delegates .lock() .unwrap_or_else(PoisonError::into_inner) .iter() .filter(|(prefix, _)| request.operation.starts_with(prefix.as_str())) .max_by_key(|(prefix, _)| prefix.len()) .map(|(_, delegate)| Arc::clone(delegate)); Box::pin(async move { if cancellation.is_cancelled() { return Err("cancelled".into()); } match request.operation.as_str() { "session.info" => Ok(info), _ => match delegate { Some(delegate) => delegate.call(request, cancellation).await, None => Err(format!( "unsupported plugin capability: {}", request.operation )), }, } }) }}
/// A core command inventory entry, independent of daemon/TUI discovery metadata.#[derive(Clone, Debug, PartialEq, Eq)]pub struct SessionCommandInfo { pub name: String, pub description: String, pub plugin: Option<Contribution>,}
/// One session's explicit settings paths, active Lua host, and reload diagnostic./// It never owns canonical history, credentials, or mutable global runtime state.pub(crate) struct SessionPlugins { host: Arc<PluginHost>, session: String, cwd: PathBuf, agent_dir: PathBuf, diagnostic: Mutex<Option<String>>, context_selection: Mutex<Option<Contribution>>, // Rotated only while holding context_selection. This identifies one actual // selection registration, including away-then-back to the same generation. context_registration: Mutex<Arc<()>>, disabled_plugins: Mutex<std::collections::BTreeSet<String>>, pinned_context_strategy: Mutex<Option<String>>, optional_builtins: Mutex<std::collections::BTreeMap<String, String>>, capabilities: Arc<SessionFacts>,}
#[derive(Clone)]pub(super) struct ContextRegistration(Arc<()>);
impl SessionPlugins { pub(crate) fn new(session: String, cwd: PathBuf, agent_dir: PathBuf, depth: u32) -> Self { // Command reservations belong to session composition, not the Lua runtime. let registry = super::slash_commands::SlashCommandRegistry::builtin(); let reserved = registry .suggestion_candidates() .into_iter() .map(str::to_string); let capabilities = Arc::new(SessionFacts { info: json!({ "session": session, "cwd": cwd, "role": if depth == 0 { "root" } else { "child" }, }), delegates: Mutex::default(), }); let host = Arc::new(PluginHost::new(capabilities.clone(), reserved)); let snapshot = host .reconcile(crate::plugins::context::builtin_sources()) .expect("compiled context plugins must activate"); let context_selection = crate::plugins::context::resolve( &snapshot, ContributionKind::ContextStrategy, crate::plugins::context::DEFAULT_STRATEGY, ) .expect("compiled default context strategy must register"); Self { host, session, cwd, agent_dir, diagnostic: Mutex::default(), context_selection: Mutex::new(Some(context_selection)), context_registration: Mutex::new(Arc::new(())), disabled_plugins: Mutex::default(), pinned_context_strategy: Mutex::new(None), optional_builtins: Mutex::default(), capabilities, } }
#[tracing::instrument(skip_all)] pub(crate) async fn reload(self: &Arc<Self>) -> anyhow::Result<Arc<ContributionSnapshot>> { let manager = Arc::clone(self); let result = tokio::task::spawn_blocking(move || { let settings = SettingsManager::create(&manager.cwd, &manager.agent_dir); if let Some(error) = settings.errors().first() { bail!("plugin settings read failed: {}", error.message); } let settings = settings.settings(); if settings.extra.contains_key("plugins") { bail!("plugins must be an object of {{enabled?, path?, config?}} entries"); } let pinned = manager .pinned_context_strategy .lock() .unwrap_or_else(PoisonError::into_inner) .clone(); let strategy = match pinned.as_deref() { Some(strategy) => strategy, None => match settings.extra.get("contextStrategy") { None => crate::plugins::context::DEFAULT_STRATEGY, Some(value) => value .as_str() .context("contextStrategy must be plugin/name")?, }, }; let mut desired: std::collections::BTreeMap<_, _> = crate::plugins::context::builtin_sources() .into_iter() .map(|source| (source.name.clone(), source)) .collect(); // Retirement is a mandatory control operation, not an enabled-source // replacement. Resolve intent before reading or evaluating any source. let mut wanted: std::collections::BTreeSet<_> = desired.keys().cloned().collect(); let mut disabled = manager .disabled_plugins .lock() .unwrap_or_else(PoisonError::into_inner) .clone(); if let Some(plugins) = settings.plugins.as_ref() { for (name, entry) in plugins { if crate::plugins::validate_plugin_name(name).is_err() { continue; } let enabled = entry.enabled.unwrap_or(desired.contains_key(name)); if enabled { wanted.insert(name.clone()); } else { wanted.remove(name); disabled.insert(name.clone()); } } } disabled.retain(|name| !wanted.contains(name)); let removed; { let mut selection = manager .context_selection .lock() .unwrap_or_else(PoisonError::into_inner); removed = manager.host.retain_plugins(&wanted)?; if selection .as_ref() .is_some_and(|target| removed.contains(&target.plugin)) { *selection = None; *manager.context_registration.lock().unwrap_or_else(PoisonError::into_inner) = Arc::new(()); } disabled.extend(removed.iter().cloned()); manager.disabled_plugins.lock().unwrap_or_else(PoisonError::into_inner).clone_from(&disabled); } let enabled_result=(||->anyhow::Result<Arc<ContributionSnapshot>> { if let Some(plugins) = settings.plugins.as_ref() { for (name, entry) in plugins { crate::plugins::validate_plugin_name(name)?; let builtin = desired.remove(name); if !entry.enabled.unwrap_or(builtin.is_some()) { disabled.insert(name.clone()); continue; } if !entry.config.is_object() { bail!("plugin {name} config must be an object"); } let source = match entry.path.as_deref() { Some(path) => crate::plugins::source::read(name, path, &manager.agent_dir)? .1, None => builtin .map(|source| source.source) .or_else(|| { manager .optional_builtins .lock() .unwrap_or_else(PoisonError::into_inner) .get(name) .cloned() }) .with_context(|| format!("plugin {name} needs an explicit path"))?, }; let mut plugin_config = json!({}); if let Some(config) = plugin_config.as_object_mut() { config.extend( entry .config .as_object() .expect("validated config object") .clone(), ); } desired.insert( name.clone(), PluginSource { name: name.clone(), source, config: plugin_config, }, ); } } for contribution in &manager.host.snapshot().contributions { if !desired.contains_key(&contribution.plugin) { disabled.insert(contribution.plugin.clone()); } } let selected_plugin = strategy.split_once('/').map_or("", |(plugin, _)| plugin); let selection_disabled = disabled.contains(selected_plugin); // Compiled policy declares a static chain. Custom Lua strategies can // choose dependencies dynamically and must validate them at invocation. let builtin_compactors = desired .get("default_context") .filter(|source| { strategy == crate::plugins::context::DEFAULT_STRATEGY && source.source == crate::plugins::context::DEFAULT_SOURCE }) .map(|source| { source .config .get("compactors") .cloned() .unwrap_or_else(|| json!(["standard_compaction/standard"])) }); // Readers take the same mutex before checking host authority. Selection and // candidate publication therefore become visible as one composition change. let mut selection = manager .context_selection .lock() .unwrap_or_else(PoisonError::into_inner); let snapshot = manager .host .reconcile_validated(desired.into_values().collect(), |candidate| { if !selection_disabled { crate::plugins::context::resolve( candidate, ContributionKind::ContextStrategy, strategy, )?; } if let Some(routes) = &builtin_compactors { let routes = routes.as_array().ok_or_else(|| { crate::plugins::PluginError::Invalid( "default context compactors must be an array".into(), ) })?; for route in routes { let route = route.as_str().ok_or_else(|| { crate::plugins::PluginError::Invalid( "default context compactor routes must be strings".into(), ) })?; let plugin = route.split_once('/').map_or("", |(plugin, _)| plugin); if !disabled.contains(plugin) { crate::plugins::context::resolve( candidate, ContributionKind::Compactor, route, )?; } } } Ok(()) })?; let next_selection = if selection_disabled { None } else { Some(crate::plugins::context::resolve( &snapshot, ContributionKind::ContextStrategy, strategy, )?) }; if *selection != next_selection { *manager.context_registration.lock().unwrap_or_else(PoisonError::into_inner) = Arc::new(()); *selection = next_selection; } *manager .disabled_plugins .lock() .unwrap_or_else(PoisonError::into_inner) = disabled; Ok(snapshot) })(); if removed.is_empty(){enabled_result}else{enabled_result.map_err(|error|{ let detail=format!("disabled/removed plugins applied [{}]; enabled replacements rejected: {error:#}",removed.iter().cloned().collect::<Vec<_>>().join(", ")); error.context(detail) })} }) .await .context("plugin reload task failed")?; *self .diagnostic .lock() .unwrap_or_else(PoisonError::into_inner) = result.as_ref().err().map(ToString::to_string); result }
pub(super) fn host(&self) -> Arc<PluginHost> { Arc::clone(&self.host) }
pub(super) fn set_native_capabilities(&self, delegate: Arc<dyn HostCapabilities>) { self.set_capability_delegate("native.", delegate); }
/// Replace one namespace adapter without changing other registered namespaces. pub(super) fn set_capability_delegate( &self, prefix: &str, delegate: Arc<dyn HostCapabilities>, ) { self.capabilities .delegates .lock() .unwrap_or_else(PoisonError::into_inner) .insert(prefix.to_owned(), delegate); }
/// Register a compiled source without activating it. Only an explicitly /// enabled plugins entry can publish this optional builtin's contributions. // Optional first-party backends register sources at composition time; the // base engine intentionally links none of those optional sources by default. #[allow(dead_code)] pub(super) fn register_optional_builtin(&self, name: &str, source: &str) { self.optional_builtins .lock() .unwrap_or_else(PoisonError::into_inner) .insert(name.to_owned(), source.to_owned()); }
pub(super) fn pin_context_strategy(&self, selection: Option<String>) { *self .pinned_context_strategy .lock() .unwrap_or_else(PoisonError::into_inner) = selection; }
pub(super) fn context_selection(&self) -> anyhow::Result<Contribution> { let selection = self .context_selection .lock() .unwrap_or_else(PoisonError::into_inner); let target=selection.as_ref().ok_or_else(||anyhow::anyhow!("selected context strategy is unavailable; re-enable its plugin, select another strategy, or start a new session"))?; self.host .with_current_contributions(&[target], || target.clone()) .map_err(Into::into) }
pub(super) fn context_contribution_is_current(&self, target: &Contribution) -> bool { self.host .with_current_contributions(&[target], || ()) .is_ok() }
pub(super) fn context_target(&self, route: &str) -> anyhow::Result<Contribution> { crate::plugins::context::resolve( &self.host.snapshot(), ContributionKind::ContextStrategy, route, ) .map_err(Into::into) }
pub(super) fn context_route_disabled(&self, route: &str) -> bool { route.split_once('/').is_some_and(|(plugin, _)| { self.disabled_plugins .lock() .unwrap_or_else(PoisonError::into_inner) .contains(plugin) }) }
pub(super) fn commit_context_selection( &self, target: &Contribution, commit: impl FnOnce() -> anyhow::Result<()>, ) -> anyhow::Result<()> { let mut selection = self .context_selection .lock() .unwrap_or_else(PoisonError::into_inner); self.host.with_current_contributions(&[target], || { commit()?; *self .pinned_context_strategy .lock() .unwrap_or_else(PoisonError::into_inner) = Some(format!("{}/{}", target.plugin, target.name)); *selection = Some(target.clone()); *self .context_registration .lock() .unwrap_or_else(PoisonError::into_inner) = Arc::new(()); Ok(()) })? }
pub(super) fn context_session_id(&self) -> &str { &self.session }
pub(super) async fn invoke_strategy_candidate( &self, target: &Contribution, input: Value, cancellation: &CancellationToken, ) -> anyhow::Result<Value> { if target.kind != ContributionKind::ContextStrategy || target.mode != InvocationMode::Unary { bail!("not a unary context strategy"); } if !self.context_contribution_is_current(target) { return Err(crate::plugins::PluginError::Retired.into()); } let output = invoke(&self.host, target, input, cancellation).await?; if !self.context_contribution_is_current(target) { return Err(crate::plugins::PluginError::Retired.into()); } Ok(output) }
pub(super) fn context_compactor(&self, route: &str) -> anyhow::Result<Contribution> { crate::plugins::context::resolve(&self.host.snapshot(), ContributionKind::Compactor, route) .map_err(Into::into) }
pub(super) fn context_is_current(&self, target: &Contribution) -> bool { let selection = self .context_selection .lock() .unwrap_or_else(PoisonError::into_inner); (target.kind != ContributionKind::ContextStrategy || selection.as_ref() == Some(target)) && self .host .with_current_contributions(&[target], || ()) .is_ok() }
pub(super) fn context_retirement( &self, target: &Contribution, ) -> anyhow::Result<CancellationToken> { self.host.retirement(target).map_err(Into::into) }
pub(super) fn with_strategy_authority<T>( &self, target: &Contribution, commit: impl FnOnce() -> anyhow::Result<T>, ) -> anyhow::Result<T> { let selection = self .context_selection .lock() .unwrap_or_else(PoisonError::into_inner); if selection.as_ref() != Some(target) { return Err(crate::plugins::PluginError::Retired.into()); } self.host.with_current_contributions(&[target], commit)? }
pub(super) fn capture_context_registration( &self, strategy: &Contribution, ) -> anyhow::Result<ContextRegistration> { self.with_strategy_authority(strategy, || { Ok(ContextRegistration(Arc::clone( &self .context_registration .lock() .unwrap_or_else(PoisonError::into_inner), ))) }) }
// A successful prewrite closure admits exactly one already-built mutation. // Publication must pass the same registration guard as the durable append. pub(super) fn with_registered_context_authority<T>( &self, registration: &ContextRegistration, strategy: &Contribution, compactor: &Contribution, commit: impl FnOnce() -> anyhow::Result<T>, ) -> anyhow::Result<T> { let selection = self .context_selection .lock() .unwrap_or_else(PoisonError::into_inner); let current = Arc::ptr_eq( ®istration.0, &self .context_registration .lock() .unwrap_or_else(PoisonError::into_inner), ); if !current || selection.as_ref() != Some(strategy) { return Err(crate::plugins::PluginError::Retired.into()); } if compactor.kind != ContributionKind::Compactor { bail!("not a compactor contribution"); } self.host .with_current_contributions(&[strategy, compactor], commit)? }
pub(super) async fn invoke_context( &self, target: &Contribution, input: Value, cancellation: &CancellationToken, ) -> anyhow::Result<Value> { if !matches!( target.kind, ContributionKind::ContextStrategy | ContributionKind::Compactor ) || target.mode != InvocationMode::Unary { bail!("not a unary context contribution"); } if !self.context_is_current(target) { return Err(crate::plugins::PluginError::Retired.into()); } let output = invoke(&self.host, target, input, cancellation).await?; if !self.context_is_current(target) { return Err(crate::plugins::PluginError::Retired.into()); } Ok(output) }
fn prompt_tail(&self) -> String { let snapshot = self.host.snapshot(); let exports: Vec<_> = snapshot.contributions.iter() .filter(|target| target.kind == ContributionKind::Export && target.mode == InvocationMode::Unary) .map(|target| json!({"call":format!("plugin.{}.{}(**kwargs)",target.plugin,target.name),"description":target.description})).collect(); let commands: Vec<_> = snapshot.contributions.iter() .filter(|target| target.kind == ContributionKind::Command && target.mode == InvocationMode::Unary) .map(|target| json!({"command":format!("/{}",target.name),"description":target.description})).collect(); if exports.is_empty() && commands.is_empty() { return String::new(); } format!("\n\n## Active plugin inventory\nSession-local Python exports (keyword arguments only) and slash commands:\n{}", json!({"exports":exports,"commands":commands})) }
pub(crate) fn handlers(&self) -> HostRequestHandlers { let mut handlers = HostRequestHandlers::new(); let host = Arc::clone(&self.host); let session = self.session.clone(); handlers.register("plugin.catalog", host_handler(move |payload| { let host = Arc::clone(&host); let session = session.clone(); async move { #[derive(Deserialize)] #[serde(deny_unknown_fields)] struct Request { #[serde(rename = "type")] kind: String, session: String, } let request: Request = serde_json::from_value(payload.data)?; if request.kind != "plugin.catalog" || request.session != session { bail!("plugin session or request type mismatch"); } let snapshot = host.snapshot(); let exports: Vec<_> = snapshot.contributions.iter() .filter(|target| target.kind == ContributionKind::Export) .map(|target| json!({ "plugin": target.plugin, "generation": target.generation.to_string(), "name": target.name, "description": target.description, "mode": match target.mode { InvocationMode::Unary => "unary", InvocationMode::Streaming => "streaming" }, })).collect(); Ok(json!({"session": session, "revision": snapshot.revision, "exports": exports})) } })); let host = Arc::clone(&self.host); let session = self.session.clone(); handlers.register( "plugin.invoke", host_handler(move |payload| { let host = Arc::clone(&host); let session = session.clone(); async move { #[derive(Deserialize)] #[serde(deny_unknown_fields)] struct Request { #[serde(rename = "type")] kind: String, session: String, generation: String, plugin: String, export: String, input: serde_json::Map<String, Value>, } let request: Request = serde_json::from_value(payload.data)?; if request.kind != "plugin.invoke" || request.session != session { bail!("plugin session mismatch"); } let generation = uuid::Uuid::parse_str(&request.generation)?; let snapshot = host.snapshot(); let target = snapshot .contributions .iter() .find(|target| { target.kind == ContributionKind::Export && target.plugin == request.plugin && target.generation == generation && target.name == request.export }) .context("plugin export generation is not active")?; if target.mode != InvocationMode::Unary { bail!("Python plugin streaming invocation is unsupported"); } invoke( &host, target, Value::Object(request.input), &CancellationToken::new(), ) .await } }), ); handlers }}
async fn invoke( host: &PluginHost, target: &Contribution, input: Value, cancellation: &CancellationToken,) -> anyhow::Result<Value> { let mut invocation = host.invoke(target, input, cancellation)?; // Observer updates are diagnostic events, not a client UI stream. while let Some(event) = invocation.next().await { match event { InvocationEvent::Update(value) => { tracing::debug!(plugin = %target.plugin, update = %value, "plugin observer update"); } InvocationEvent::Finished(Ok( InvocationOutcome::Unary(value) | InvocationOutcome::Streamed(value), )) => return Ok(value), InvocationEvent::Finished(Err(error)) => return Err(error.into()), } } Err(anyhow!("plugin invocation ended without an outcome"))}
/// Read active inventory only at the model-call seam. Mandatory prompt layers/// keep their stable prefix; reload/disable changes the dynamic tail immediately.pub(crate) fn wrap_stream_fn( plugins: Arc<SessionPlugins>, stream: pa_agent::stream::StreamFn,) -> pa_agent::stream::StreamFn { Arc::new(move |model, mut context, options| { let tail = plugins.prompt_tail(); if !tail.is_empty() { context .system_prompt .get_or_insert_with(String::new) .push_str(&tail); } stream(model, context, options) })}
impl AgentSession { pub(crate) fn parse_core_session_command(&self, text: &str) -> Option<SessionSlashCommand> { if let Some(command) = parse_session_command(&self.slash_commands, text) { return Some(command); } let (name, args) = super::slash_commands::parse_slash_command(text)?; let plugins = self.plugins.as_ref()?; let target = plugins .host .snapshot() .contributions .iter() .find(|target| target.kind == ContributionKind::Command && target.name == name)? .clone(); Some(SessionSlashCommand { name, args, text: text.to_string(), plugin: Some(target), }) }
/// Active core command inventory. Plugin entries disappear immediately on disable. /// This does not claim daemon/TUI command discovery support. pub fn session_command_inventory(&self) -> Vec<SessionCommandInfo> { let mut inventory: Vec<_> = self .slash_commands .all() .iter() .map(|command| SessionCommandInfo { name: command.name.to_string(), description: command.description.to_string(), plugin: None, }) .collect(); if let Some(plugins) = &self.plugins { inventory.extend( plugins .host .snapshot() .contributions .iter() .filter(|target| target.kind == ContributionKind::Command) .map(|target| SessionCommandInfo { name: target.name.clone(), description: target.description.clone(), plugin: Some(target.clone()), }), ); } inventory }}
impl SessionEngine { /// Current assembled prompt, including the active inventory tail used by model /// calls. `system_prompt` remains the cache-stable base captured at assembly. pub fn assembled_system_prompt(&self) -> String { let mut prompt = self.system_prompt.clone(); if let Some(plugins) = &self.session.plugins { prompt.push_str(&plugins.prompt_tail()); } prompt }
/// Exact plugin handlers installed in this session's kernel transport. Captures /// no engine/session lock, so a kernel request during an active tool cannot deadlock. pub fn plugin_host_handlers(&self) -> HostRequestHandlers { self.session .plugins .as_ref() .map_or_else(HostRequestHandlers::new, |plugins| plugins.handlers()) }
/// Reconcile explicit global/project plugin configuration without rebuilding the /// engine, rewriting history, or deleting saved data. Failure retains the old host. /// /// # Errors /// Returns settings, source, activation, or reconciliation errors without retiring /// the previous activation. pub async fn reload_plugins(&self) -> anyhow::Result<Arc<ContributionSnapshot>> { self.session .plugins .as_ref() .context("session plugin host is unavailable")? .reload() .await }
/// Latest activation/reload failure. Ordinary agent startup remains available. pub fn plugin_diagnostic(&self) -> Option<String> { self.session.plugins.as_ref().and_then(|plugins| { plugins .diagnostic .lock() .unwrap_or_else(PoisonError::into_inner) .clone() }) }
/// Invoke the generation-bound target returned by core command admission. /// /// # Errors /// Returns an error for unsupported modes, stale/disabled targets, cancellation, /// or a Lua/capability failure. pub async fn invoke_plugin_command( &self, target: &Contribution, args: &str, ) -> anyhow::Result<Value> { if target.kind != ContributionKind::Command { bail!("not a plugin command"); } if target.mode != InvocationMode::Unary { bail!("streaming plugin commands require a streaming client"); } let plugins = self .session .plugins .as_ref() .context("session plugin host is unavailable")?; invoke( &plugins.host, target, json!({"args": args}), &CancellationToken::new(), ) .await }}
/// Install a real settled-turn observer. Its JSON outcome is diagnostic only; it/// cannot rewrite messages or decide engine control flow. Failures do not fail turns.pub(crate) async fn observe_turns( agent: &Arc<pa_agent::agent::Agent>, plugins: Arc<SessionPlugins>,) { agent.subscribe(move |event, _signal| { let host = Arc::clone(&plugins.host); let session = plugins.session.clone(); Box::pin(async move { if let pa_agent::types::AgentEvent::TurnEnd { message, tool_results } = event { let input = json!({"stage": "turn_settled", "session": session, "message": message, "tool_results": tool_results}); for target in &host.snapshot().contributions { if target.kind != ContributionKind::Hook || target.name != "turn_settled" { continue; } // Observers cannot hold turn completion indefinitely. let outcome = tokio::time::timeout(std::time::Duration::from_secs(1), invoke(&host, target, input.clone(), &CancellationToken::new())).await; match outcome { Ok(Ok(value)) => tracing::debug!(plugin = %target.plugin, outcome = %value, "plugin turn observer settled"), Ok(Err(error)) => tracing::warn!(plugin = %target.plugin, %error, "plugin turn observer failed"), Err(error) => tracing::warn!(plugin = %target.plugin, %error, "plugin turn observer timed out"), } } } Ok(()) }) }).await;}
#[cfg(test)]mod tests;
#[cfg(test)]mod context_selection_tests { use super::*;
struct Fixture { _dir: tempfile::TempDir, manager: Arc<SessionPlugins>, settings: PathBuf, }
impl Fixture { fn new() -> Self { let dir = tempfile::tempdir().unwrap(); let cwd = dir.path().join("project"); let agent = dir.path().join("agent"); std::fs::create_dir_all(&cwd).unwrap(); std::fs::create_dir_all(&agent).unwrap(); let settings = agent.join("settings.json"); let manager = Arc::new(SessionPlugins::new("context-test".into(), cwd, agent, 0)); Self { _dir: dir, manager, settings, } }
fn configure(&self, value: &Value) { std::fs::write(&self.settings, value.to_string()).unwrap(); } }
#[tokio::test] async fn capability_namespaces_route_independently() { struct Capture { name: &'static str, requests: tokio::sync::mpsc::UnboundedSender<(String, CapabilityRequest, CancellationToken)>, } impl HostCapabilities for Capture { fn call( &self, request: CapabilityRequest, cancellation: CancellationToken, ) -> CapabilityFuture { self.requests .send((self.name.into(), request, cancellation)) .unwrap(); let name = self.name; Box::pin(async move { Ok(json!({"delegate":name})) }) } } let f = Fixture::new(); let (sent, mut received) = tokio::sync::mpsc::unbounded_channel(); f.manager.set_native_capabilities(Arc::new(Capture { name: "native", requests: sent.clone(), })); f.manager.set_capability_delegate( "records.", Arc::new(Capture { name: "records", requests: sent.clone(), }), ); f.manager.set_capability_delegate( "native.specific.", Arc::new(Capture { name: "specific", requests: sent.clone(), }), ); let request = CapabilityRequest { invocation: uuid::Uuid::new_v4(), operation: String::new(), input: json!({"retained":true}), }; for (operation, expected) in [ ("native.call", "native"), ("records.read", "records"), ("native.specific.call", "specific"), ] { let request = CapabilityRequest { operation: operation.into(), ..request.clone() }; let cancel = CancellationToken::new(); assert_eq!( f.manager .capabilities .call(request.clone(), cancel.clone()) .await .unwrap(), json!({"delegate":expected}) ); let (name, actual, delegated_cancel) = received.recv().await.unwrap(); assert_eq!((name, actual), (expected.to_string(), request)); cancel.cancel(); assert!(delegated_cancel.is_cancelled()); } // Replacing the native namespace cannot replace the independent records adapter. f.manager.set_native_capabilities(Arc::new(Capture { name: "replacement", requests: sent, })); for (operation, expected) in [("native.call", "replacement"), ("records.read", "records")] { let request = CapabilityRequest { operation: operation.into(), ..request.clone() }; assert_eq!( f.manager .capabilities .call(request, CancellationToken::new()) .await .unwrap(), json!({"delegate":expected}) ); received.recv().await.unwrap(); } for operation in ["unknown.call", "nativeish.call"] { let request = CapabilityRequest { operation: operation.into(), ..request.clone() }; assert!(f .manager .capabilities .call(request, CancellationToken::new()) .await .is_err()); assert!(received.try_recv().is_err()); } let info = CapabilityRequest { operation: "session.info".into(), ..request }; assert_eq!( f.manager .capabilities .call(info, CancellationToken::new()) .await .unwrap(), f.manager.capabilities.info ); assert!(received.try_recv().is_err()); }
#[tokio::test] async fn compiled_default_and_full_chain_are_real_lua_contributions() { let f = Fixture::new(); f.manager.reload().await.unwrap(); let strategy = f.manager.context_selection().unwrap(); let cancel = CancellationToken::new(); let messages = json!([{"role":"user","content":"exact ordered history"}]); assert_eq!( f.manager .invoke_context( &strategy, json!({"stage":"assemble","messages":messages}), &cancel ) .await .unwrap(), json!({"messages":messages}) ); assert_eq!( f.manager .invoke_context(&strategy, json!({"stage":"prepare"}), &cancel) .await .unwrap(), json!({"policy":"rolling","compactors":["server_compaction/server","snapcompact/snapcompact","standard_compaction/standard"]}) ); for (route, expected) in [ ( "standard_compaction/standard", json!({"status":"native","operation":"standard"}), ), ( "server_compaction/server", json!({"status":"native","operation":"server","config":{}}), ), ( "snapcompact/snapcompact", json!({"status":"native","operation":"snapcompact","config":{}}), ), ] { let compactor = f.manager.context_compactor(route).unwrap(); assert_eq!( f.manager .invoke_context(&compactor, json!({}), &cancel) .await .unwrap(), expected ); } f.configure(&json!({"plugins":{"default_context":{"config":{"policy":"full","compactors":["server_compaction/server","snapcompact/snapcompact","standard_compaction/standard"]}}}})); f.manager.reload().await.unwrap(); assert!(!f.manager.context_is_current(&strategy)); let strategy = f.manager.context_selection().unwrap(); assert_eq!( f.manager .invoke_context(&strategy, json!({"stage":"prepare"}), &cancel) .await .unwrap(), json!({"policy":"full","compactors":["server_compaction/server","snapcompact/snapcompact","standard_compaction/standard"]}) ); }
#[tokio::test] async fn archive_restore_validates_schema_without_requiring_creator_generation() { let f = Fixture::new(); let strategy = f.manager.context_selection().unwrap(); let cancel = CancellationToken::new(); for (archive, accepted) in [ ( json!({"version":1,"strategy":"default_context/rolling","creatorGeneration":"old-process"}), true, ), ( json!({"version":2,"strategy":"default_context/rolling"}), false, ), (json!({"version":1,"strategy":"other/strategy"}), false), (Value::Null, false), ] { assert_eq!( f.manager .invoke_context( &strategy, json!({"stage":"restore","archive":archive}), &cancel ) .await .unwrap(), json!({"accepted":accepted}) ); } assert!(f.manager.invoke_context(&strategy,json!({"stage":"assemble","messages":[],"archive":{"version":2,"strategy":"default_context/rolling"}}),&cancel).await.is_err()); }
#[tokio::test] async fn invalid_enabled_replacement_preserves_host_but_explicit_disable_retires_it() { let f = Fixture::new(); let old = f.manager.context_selection().unwrap(); let snapshot = f.manager.host.snapshot(); for settings in [ json!({"contextStrategy":"missing/strategy"}), json!({"contextStrategy":null}), json!({"plugins":{"default_context":{"config":{"policy":"unknown"}}}}), ] { f.configure(&settings); assert!(f.manager.reload().await.is_err()); assert_eq!(f.manager.context_selection().unwrap(), old); assert!(Arc::ptr_eq(&snapshot, &f.manager.host.snapshot())); } let compactor = f .manager .context_compactor("standard_compaction/standard") .unwrap(); let registration = f.manager.capture_context_registration(&old).unwrap(); assert_eq!( f.manager .with_registered_context_authority(®istration, &old, &compactor, || Ok( "durable commit" )) .unwrap(), "durable commit" ); f.configure(&json!({"plugins":{"standard_compaction":{"enabled":false},"default_context":{"config":{"compactors":[]}}}})); f.manager.reload().await.unwrap(); assert!(f .manager .with_registered_context_authority(®istration, &old, &compactor, || Ok(())) .is_err()); assert!(f.manager.context_retirement(&compactor).is_err()); let selected = f.manager.context_selection().unwrap(); f.configure(&json!({"plugins":{"default_context":{"enabled":false}}})); let snapshot = f.manager.reload().await.unwrap(); assert!(!snapshot .contributions .iter() .any(|target| target.plugin == "default_context")); assert!(f .manager .context_selection() .unwrap_err() .to_string() .contains("re-enable")); assert!(!f.manager.context_is_current(&selected)); assert!(f.manager.context_retirement(&selected).is_err()); }
#[tokio::test] async fn invalid_builtin_chain_preserves_host_revision_and_selection() { let f = Fixture::new(); let original = f.manager.host.snapshot(); let strategy = f.manager.context_selection().unwrap(); for settings in [ json!({"plugins":{"default_context":{"config":{"compactors":["missing/compactor"]}}}}), json!({"plugins":{"default_context":{"config":{"compactors":[7]}}}}), ] { f.configure(&settings); assert!(f.manager.reload().await.is_err()); assert_eq!(f.manager.context_selection().unwrap(), strategy); assert!(Arc::ptr_eq(&original, &f.manager.host.snapshot())); } f.configure(&json!({"plugins":{"standard_compaction":{"enabled":false},"default_context":{"config":{"compactors":[]}}}})); f.manager.reload().await.unwrap(); let strategy = f.manager.context_selection().unwrap(); assert_eq!( f.manager .invoke_context( &strategy, json!({"stage":"prepare"}), &CancellationToken::new() ) .await .unwrap(), json!({"policy":"rolling","compactors":[]}) ); }
#[tokio::test] async fn custom_strategy_and_compactors_use_same_typed_invocation_and_retirement() { let f = Fixture::new(); let source = r"return { context_strategies={{name='custom',run=function(input) if input.stage=='assemble' then return {messages=input.messages} end return {compactors={'custom/decline','custom/failure','custom/fatal','custom/success'},policy='full'} end}}, compactors={ {name='decline',run=function() return {status='decline'} end}, {name='failure',run=function() return {status='failure',message='try next'} end}, {name='fatal',run=function() return {status='fatal',message='stop'} end}, {name='success',run=function() return {status='success',summary='summary',first_kept_entry_id='kept'} end} }}"; std::fs::write(f.manager.agent_dir.join("custom.lua"), source).unwrap(); f.configure(&json!({"contextStrategy":"custom/custom","plugins":{"custom":{"enabled":true,"path":"custom.lua"},"default_context":{"enabled":false}}})); f.manager.reload().await.unwrap(); let strategy = f.manager.context_selection().unwrap(); assert_eq!(strategy.plugin, "custom"); let cancel = CancellationToken::new(); for (status, expected) in [ ("decline", json!({"status":"decline"})), ("failure", json!({"status":"failure","message":"try next"})), ("fatal", json!({"status":"fatal","message":"stop"})), ( "success", json!({"status":"success","summary":"summary","first_kept_entry_id":"kept"}), ), ] { let target = f .manager .context_compactor(&format!("custom/{status}")) .unwrap(); assert_eq!( f.manager .invoke_context(&target, json!({}), &cancel) .await .unwrap(), expected ); } let retired = f.manager.context_retirement(&strategy).unwrap(); f.configure(&json!({})); f.manager.reload().await.unwrap(); assert!(retired.is_cancelled()); assert!(!f.manager.context_is_current(&strategy)); assert!(f .manager .invoke_context(&strategy, json!({}), &cancel) .await .is_err()); } #[tokio::test] async fn context_registration_rejects_away_then_back_even_same_generation() { let f = Fixture::new(); std::fs::write(f.manager.agent_dir.join("away.lua"), "return {context_strategies={{name='other',run=function(input) return {messages=input.messages} end}}}").unwrap(); f.configure(&json!({"plugins":{"away":{"enabled":true,"path":"away.lua"}}})); f.manager.reload().await.unwrap(); let strategy = f.manager.context_selection().unwrap(); let compactor = f .manager .context_compactor("standard_compaction/standard") .unwrap(); let registration = f.manager.capture_context_registration(&strategy).unwrap(); f.manager .with_registered_context_authority(®istration, &strategy, &compactor, || Ok(())) .unwrap(); let away = f.manager.context_target("away/other").unwrap(); f.manager .commit_context_selection(&away, || Ok(())) .unwrap(); f.manager .commit_context_selection(&strategy, || Ok(())) .unwrap(); assert_eq!(f.manager.context_selection().unwrap(), strategy); let mut published = false; assert!(f .manager .with_registered_context_authority(®istration, &strategy, &compactor, || { published = true; Ok(()) }) .is_err()); assert!(!published); }
#[tokio::test] async fn context_commit_admission_does_not_publish_after_compactor_retirement() { let f = Fixture::new(); let strategy = f.manager.context_selection().unwrap(); let compactor = f .manager .context_compactor("standard_compaction/standard") .unwrap(); let registration = f.manager.capture_context_registration(&strategy).unwrap(); let admitted = f .manager .with_registered_context_authority(®istration, &strategy, &compactor, || { Ok("one frozen write") }) .unwrap(); f.configure(&json!({"plugins":{"standard_compaction":{"enabled":false},"default_context":{"config":{"compactors":[]}}}})); f.manager.reload().await.unwrap(); // Owner work already admitted must still finish; only publication is refused. assert_eq!(admitted, "one frozen write"); let mut published = false; assert!(f .manager .with_registered_context_authority(®istration, &strategy, &compactor, || { published = true; Ok(()) }) .is_err()); assert!(!published); }}