Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350//! 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()); } if request.session != info["session"].as_str().unwrap_or_default() { return Err("plugin session mismatch".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<()>);
#[derive(Deserialize)]#[serde(deny_unknown_fields)]struct ConfiguredPlugin { enabled: Option<bool>, path: Option<PathBuf>, #[serde(default = "empty_config")] config: Value,}
fn empty_config() -> Value { json!({})}
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( session.clone(), 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(); 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(value) = settings.plugins.as_ref() { let entries = value.as_object().context("plugins must be an object")?; for (name, entry) in entries { if crate::plugins::validate_plugin_name(name).is_err() {continue;} if !entry.is_object() {wanted.insert(name.clone());continue;} let enabled = match entry.get("enabled") { Some(Value::Bool(enabled)) => *enabled, None => desired.contains_key(name), // An invalid enabled setting is a replacement error, not // an implicit request to retire the last valid source. Some(_) => true, }; 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(value) = settings.plugins.as_ref() { let entries = value.as_object().context("plugins must be an object")?; for (name, entry) in entries { crate::plugins::validate_plugin_name(name)?; let entry: ConfiguredPlugin = serde_json::from_value(entry.clone()) .with_context(|| format!("invalid plugin configuration: {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 { Some(path) => { // Relative explicit paths resolve from cwd. No ambient discovery. let path = manager.cwd.join(path); std::fs::read_to_string(&path).with_context(|| { format!("read plugin {name} at {}", path.display()) })? } 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_and_keep_request_identity() { 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 { session: "context-test".into(), plugin: "identity".into(), generation: uuid::Uuid::new_v4(), 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.cwd.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.cwd.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); }}