1use std::collections::{BTreeMap, HashMap, HashSet};
16use std::path::{Path, PathBuf};
17
18use chrono::{DateTime, Utc};
19use serde::{Deserialize, Serialize};
20use tokio::fs;
21
22use crate::error::{PluginError, PluginResult};
23use crate::manifest::{
24 EventSinkCapabilityState, EventSinkManifestEntry, ObservationPermissionId, Platform,
25 PluginManifest,
26};
27
28pub static PLUGIN_OPERATION_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
32
33#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
36#[serde(tag = "type", rename_all = "snake_case")]
37pub enum PluginSource {
38 LocalDir { path: PathBuf },
42 LocalArchive { path: PathBuf },
44 Url {
82 url: String,
83 #[serde(default, skip_serializing_if = "Option::is_none")]
84 sha256: Option<String>,
85 #[serde(default, skip_serializing_if = "is_false")]
91 allow_unverified: bool,
92 #[serde(default, skip_serializing_if = "is_false")]
97 allow_untrusted_host: bool,
98 #[serde(default, skip_serializing_if = "is_false")]
102 allow_unsigned: bool,
103 #[serde(default, skip_serializing_if = "Option::is_none")]
108 signed_by: Option<String>,
109 #[serde(default, skip_serializing_if = "is_false")]
115 insecure: bool,
116 },
117}
118
119fn is_false(value: &bool) -> bool {
123 !*value
124}
125
126pub type EventSinkPermissionGrants = BTreeMap<String, Vec<ObservationPermissionId>>;
129
130#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
136pub struct RegisteredCapabilities {
137 #[serde(default, skip_serializing_if = "Vec::is_empty")]
138 pub hooks: Vec<crate::hooks::HookRegistration>,
139 #[serde(default, skip_serializing_if = "Vec::is_empty")]
141 pub mcp_server_ids: Vec<String>,
142 #[serde(default, skip_serializing_if = "Vec::is_empty")]
145 pub skill_dirs: Vec<String>,
146 #[serde(default, skip_serializing_if = "Vec::is_empty")]
148 pub preset_ids: Vec<String>,
149 #[serde(default, skip_serializing_if = "Vec::is_empty")]
153 pub workflow_filenames: Vec<String>,
154 #[serde(default, skip_serializing_if = "Vec::is_empty")]
158 pub service_ids: Vec<String>,
159 #[serde(default, skip_serializing_if = "Vec::is_empty")]
163 pub event_sink_ids: Vec<String>,
164 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
167 pub event_sink_grants: EventSinkPermissionGrants,
168}
169
170impl RegisteredCapabilities {
171 pub fn is_empty(&self) -> bool {
172 self.hooks.is_empty()
173 && self.mcp_server_ids.is_empty()
174 && self.skill_dirs.is_empty()
175 && self.preset_ids.is_empty()
176 && self.workflow_filenames.is_empty()
177 && self.service_ids.is_empty()
178 && self.event_sink_ids.is_empty()
179 }
180
181 pub fn removed_since(&self, old: &RegisteredCapabilities) -> RegisteredCapabilities {
192 RegisteredCapabilities {
193 hooks: old
194 .hooks
195 .iter()
196 .filter(|old| !self.hooks.iter().any(|new| new.config == old.config))
197 .cloned()
198 .collect(),
199 mcp_server_ids: subtract(&old.mcp_server_ids, &self.mcp_server_ids),
200 skill_dirs: subtract(&old.skill_dirs, &self.skill_dirs),
201 preset_ids: subtract(&old.preset_ids, &self.preset_ids),
202 workflow_filenames: subtract(&old.workflow_filenames, &self.workflow_filenames),
203 service_ids: subtract(&old.service_ids, &self.service_ids),
204 event_sink_ids: subtract(&old.event_sink_ids, &self.event_sink_ids),
205 event_sink_grants: BTreeMap::new(),
208 }
209 }
210
211 pub fn removal_order(&self) -> EventSinkRemovalOrder {
215 EventSinkRemovalOrder {
216 event_sink_ids_before_services: self.event_sink_ids.clone(),
217 service_ids_after_sinks: self.service_ids.clone(),
218 }
219 }
220}
221
222#[derive(Debug, Clone, Default, PartialEq, Eq)]
223pub struct EventSinkRemovalOrder {
224 pub event_sink_ids_before_services: Vec<String>,
225 pub service_ids_after_sinks: Vec<String>,
226}
227
228#[derive(Debug, Clone, PartialEq, Eq)]
229pub struct ReconciledEventSink {
230 pub id: String,
231 pub service_id: String,
232 pub state: EventSinkCapabilityState,
233}
234
235#[derive(Debug, Clone, Default, PartialEq, Eq)]
239pub struct EventSinkReconciliation {
240 pub deactivate_before_services: Vec<String>,
242 pub service_dependencies_before_sinks: Vec<String>,
245 pub sinks_after_services: Vec<ReconciledEventSink>,
247}
248
249#[derive(Debug, Clone)]
252pub struct PluginBootCandidate {
253 pub installed: InstalledPlugin,
254 pub manifest: Option<PluginManifest>,
255}
256
257#[derive(Debug, Clone, PartialEq, Eq)]
259pub enum PluginBootIssue {
260 DuplicatePluginId { id: String },
261 ManifestUnavailable,
262 ManifestIdMismatch { manifest_id: String },
263 InvalidManifest { detail: String },
264 InstallIncomplete,
265 UnknownPlatform,
266 PlatformIneligible,
267 DuplicateEventSinkOwner { id: String },
268 DuplicateServiceOwner { id: String },
269}
270
271#[derive(Debug, Clone, Default, PartialEq, Eq)]
275pub struct PluginBootReconciliation {
276 pub plugin_id: String,
277 pub service_ids_to_start: Vec<String>,
278 pub event_sinks: EventSinkReconciliation,
279 pub issues: Vec<PluginBootIssue>,
280}
281
282pub fn reconcile_plugin_boot(
288 candidates: &[PluginBootCandidate],
289 platform: Option<Platform>,
290) -> Vec<PluginBootReconciliation> {
291 let mut plugin_id_counts: HashMap<&str, usize> = HashMap::new();
292 let mut sink_owner_counts: HashMap<&str, usize> = HashMap::new();
293 let mut service_owner_counts: HashMap<&str, usize> = HashMap::new();
294 for candidate in candidates {
295 *plugin_id_counts
296 .entry(candidate.installed.id.as_str())
297 .or_default() += 1;
298 for id in &candidate.installed.registered.event_sink_ids {
299 *sink_owner_counts.entry(id.as_str()).or_default() += 1;
300 }
301 for id in &candidate.installed.registered.service_ids {
302 *service_owner_counts.entry(id.as_str()).or_default() += 1;
303 }
304 }
305
306 candidates
307 .iter()
308 .map(|candidate| {
309 let installed = &candidate.installed;
310 let mut plan = PluginBootReconciliation {
311 plugin_id: installed.id.clone(),
312 event_sinks: EventSinkReconciliation {
313 deactivate_before_services: unique_strings(
314 &installed.registered.event_sink_ids,
315 ),
316 ..Default::default()
317 },
318 ..Default::default()
319 };
320
321 if plugin_id_counts
322 .get(installed.id.as_str())
323 .copied()
324 .unwrap_or_default()
325 > 1
326 {
327 plan.issues.push(PluginBootIssue::DuplicatePluginId {
328 id: installed.id.clone(),
329 });
330 return plan;
331 }
332 if installed.status == PluginInstallStatus::Installing {
333 plan.issues.push(PluginBootIssue::InstallIncomplete);
334 return plan;
335 }
336 let Some(platform) = platform else {
337 plan.issues.push(PluginBootIssue::UnknownPlatform);
338 return plan;
339 };
340 let Some(manifest) = candidate.manifest.as_ref() else {
341 plan.issues.push(PluginBootIssue::ManifestUnavailable);
342 return plan;
343 };
344 if manifest.id != installed.id {
345 plan.issues.push(PluginBootIssue::ManifestIdMismatch {
346 manifest_id: manifest.id.clone(),
347 });
348 return plan;
349 }
350 if let Err(error) = manifest.validate() {
351 plan.issues.push(PluginBootIssue::InvalidManifest {
352 detail: error.to_string(),
353 });
354 return plan;
355 }
356 if !manifest.supports_platform(platform) {
357 plan.issues.push(PluginBootIssue::PlatformIneligible);
358 return plan;
359 }
360
361 let mut sink_plan = reconcile_event_sinks(
362 manifest,
363 &installed.registered,
364 installed.status,
365 Some(platform),
366 )
367 .expect("manifest was validated above");
368 let mut unsafe_sink_ids = HashSet::new();
369 let mut unsafe_backing_services = HashSet::new();
370
371 for sink_id in &installed.registered.event_sink_ids {
372 if sink_owner_counts
373 .get(sink_id.as_str())
374 .copied()
375 .unwrap_or_default()
376 > 1
377 {
378 push_issue_once(
379 &mut plan.issues,
380 PluginBootIssue::DuplicateEventSinkOwner {
381 id: sink_id.clone(),
382 },
383 );
384 unsafe_sink_ids.insert(sink_id.as_str());
385 if let Some(sink) = manifest
386 .provides
387 .event_sinks
388 .iter()
389 .find(|sink| sink.id == *sink_id)
390 {
391 unsafe_backing_services.insert(sink.service_id.as_str());
392 }
393 }
394 }
395
396 for service_id in &installed.registered.service_ids {
397 if service_owner_counts
398 .get(service_id.as_str())
399 .copied()
400 .unwrap_or_default()
401 > 1
402 {
403 push_issue_once(
404 &mut plan.issues,
405 PluginBootIssue::DuplicateServiceOwner {
406 id: service_id.clone(),
407 },
408 );
409 unsafe_backing_services.insert(service_id.as_str());
410 }
411 }
412 for sink in &manifest.provides.event_sinks {
413 if installed
414 .registered
415 .event_sink_ids
416 .iter()
417 .any(|id| id == &sink.id)
418 && service_owner_counts
419 .get(sink.service_id.as_str())
420 .copied()
421 .unwrap_or_default()
422 != 1
423 {
424 unsafe_sink_ids.insert(sink.id.as_str());
425 unsafe_backing_services.insert(sink.service_id.as_str());
426 }
427 }
428
429 sink_plan.sinks_after_services.retain(|sink| {
430 !unsafe_sink_ids.contains(sink.id.as_str())
431 && !unsafe_backing_services.contains(sink.service_id.as_str())
432 });
433 sink_plan
434 .service_dependencies_before_sinks
435 .retain(|service_id| !unsafe_backing_services.contains(service_id.as_str()));
436 for sink_id in unsafe_sink_ids {
437 if !sink_plan
438 .deactivate_before_services
439 .iter()
440 .any(|id| id == sink_id)
441 {
442 sink_plan
443 .deactivate_before_services
444 .push(sink_id.to_string());
445 }
446 }
447
448 let owned_services: HashSet<&str> = installed
449 .registered
450 .service_ids
451 .iter()
452 .map(String::as_str)
453 .collect();
454 plan.service_ids_to_start = manifest
455 .provides
456 .services
457 .iter()
458 .filter(|service| {
459 service.enabled
460 && owned_services.contains(service.id.as_str())
461 && service_owner_counts
462 .get(service.id.as_str())
463 .copied()
464 .unwrap_or_default()
465 == 1
466 && !unsafe_backing_services.contains(service.id.as_str())
467 })
468 .map(|service| service.id.clone())
469 .collect();
470 plan.event_sinks = sink_plan;
471 plan
472 })
473 .collect()
474}
475
476fn unique_strings(values: &[String]) -> Vec<String> {
477 let mut seen = HashSet::new();
478 values
479 .iter()
480 .filter(|value| seen.insert(value.as_str()))
481 .cloned()
482 .collect()
483}
484
485fn push_issue_once(issues: &mut Vec<PluginBootIssue>, issue: PluginBootIssue) {
486 if !issues.contains(&issue) {
487 issues.push(issue);
488 }
489}
490
491pub fn reconcile_event_sinks(
496 manifest: &PluginManifest,
497 registered: &RegisteredCapabilities,
498 install_status: PluginInstallStatus,
499 platform: Option<Platform>,
500) -> PluginResult<EventSinkReconciliation> {
501 manifest.validate()?;
502 let plugin_platform_eligible =
503 platform.is_some_and(|platform| manifest.supports_platform(platform));
504 let owned_sink_ids: HashSet<&str> = registered
505 .event_sink_ids
506 .iter()
507 .map(String::as_str)
508 .collect();
509 let owned_service_ids: HashSet<&str> =
510 registered.service_ids.iter().map(String::as_str).collect();
511
512 let mut plan = EventSinkReconciliation::default();
513 let mut reconciled_ids = HashSet::new();
514 let mut service_dependencies = HashSet::new();
515 for sink in &manifest.provides.event_sinks {
516 if !owned_sink_ids.contains(sink.id.as_str()) {
517 continue;
518 }
519 let Some(service) =
520 same_plugin_owned_service(sink, &manifest.provides.services, &owned_service_ids)
521 else {
522 if reconciled_ids.insert(sink.id.as_str()) {
523 plan.deactivate_before_services.push(sink.id.clone());
524 }
525 continue;
526 };
527 if !reconciled_ids.insert(sink.id.as_str()) {
528 continue;
529 }
530 let state = if install_status == PluginInstallStatus::Installing {
531 EventSinkCapabilityState::Inactive {
532 detail: crate::manifest::EventSinkInactiveReason::InstallIncomplete,
533 }
534 } else if !plugin_platform_eligible {
535 EventSinkCapabilityState::Inactive {
536 detail: crate::manifest::EventSinkInactiveReason::PlatformIneligible,
537 }
538 } else {
539 sink.capability_state(service, platform)
540 };
541 if matches!(state, EventSinkCapabilityState::Eligible)
542 && service_dependencies.insert(service.id.as_str())
543 {
544 plan.service_dependencies_before_sinks
545 .push(service.id.clone());
546 }
547 plan.sinks_after_services.push(ReconciledEventSink {
548 id: sink.id.clone(),
549 service_id: service.id.clone(),
550 state,
551 });
552 }
553
554 for owned_id in ®istered.event_sink_ids {
555 if !reconciled_ids.contains(owned_id.as_str())
556 && !plan.deactivate_before_services.contains(owned_id)
557 {
558 plan.deactivate_before_services.push(owned_id.clone());
559 }
560 }
561 Ok(plan)
562}
563
564fn same_plugin_owned_service<'a>(
565 sink: &EventSinkManifestEntry,
566 services: &'a [crate::manifest::ServiceManifestEntry],
567 owned_service_ids: &HashSet<&str>,
568) -> Option<&'a crate::manifest::ServiceManifestEntry> {
569 if !owned_service_ids.contains(sink.service_id.as_str()) {
570 return None;
571 }
572 services
573 .iter()
574 .find(|service| service.id == sink.service_id)
575}
576
577fn subtract(from: &[String], remove: &[String]) -> Vec<String> {
579 let drop: HashSet<&str> = remove.iter().map(String::as_str).collect();
580 from.iter()
581 .filter(|value| !drop.contains(value.as_str()))
582 .cloned()
583 .collect()
584}
585
586#[derive(Debug, Clone, Copy, PartialEq, Eq)]
591pub enum Ownership {
592 New,
595 OwnedReinstall,
598 ForeignConflict,
601}
602
603pub fn classify_ownership(
606 id: &str,
607 existing: &HashSet<&str>,
608 owned_previously: &HashSet<&str>,
609) -> Ownership {
610 if !existing.contains(id) {
611 Ownership::New
612 } else if owned_previously.contains(id) {
613 Ownership::OwnedReinstall
614 } else {
615 Ownership::ForeignConflict
616 }
617}
618
619#[derive(Debug, Clone, Default, PartialEq, Eq)]
622pub struct ExclusiveReconciliation {
623 pub to_register: Vec<String>,
626 pub foreign_conflicts: Vec<String>,
630}
631
632pub fn reconcile_exclusive(
643 declared: &[String],
644 existing: &[String],
645 owned_previously: &[String],
646) -> ExclusiveReconciliation {
647 let existing_set: HashSet<&str> = existing.iter().map(String::as_str).collect();
648 let owned_set: HashSet<&str> = owned_previously.iter().map(String::as_str).collect();
649
650 let mut result = ExclusiveReconciliation::default();
651 for id in declared {
652 match classify_ownership(id, &existing_set, &owned_set) {
653 Ownership::New | Ownership::OwnedReinstall => result.to_register.push(id.clone()),
654 Ownership::ForeignConflict => result.foreign_conflicts.push(id.clone()),
655 }
656 }
657 result
658}
659
660#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
674#[serde(rename_all = "snake_case")]
675pub enum PluginInstallStatus {
676 Installing,
679 #[default]
683 Installed,
684}
685
686#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
688pub struct InstalledPlugin {
689 pub id: String,
690 pub version: String,
692 pub source: PluginSource,
693 pub plugin_dir: PathBuf,
695 pub installed_at: DateTime<Utc>,
699 #[serde(default)]
703 pub status: PluginInstallStatus,
704 #[serde(default)]
705 pub registered: RegisteredCapabilities,
706}
707
708#[derive(Debug, Clone, Default, Serialize, Deserialize)]
710pub struct InstalledPlugins {
711 #[serde(default)]
712 pub plugins: Vec<InstalledPlugin>,
713}
714
715impl InstalledPlugins {
716 pub async fn load(path: &Path) -> PluginResult<Self> {
720 match fs::try_exists(path).await {
721 Ok(true) => {}
722 Ok(false) => return Ok(Self::default()),
723 Err(error) => return Err(PluginError::Io(error)),
724 }
725
726 let raw = fs::read_to_string(path).await?;
727 if raw.trim().is_empty() {
728 return Ok(Self::default());
729 }
730 let store: Self = serde_json::from_str(&raw)?;
731 Ok(store)
732 }
733
734 pub async fn save(&self, path: &Path) -> PluginResult<()> {
742 if let Some(parent) = path.parent() {
743 fs::create_dir_all(parent).await?;
744 }
745 let serialized = serde_json::to_string_pretty(self)?;
746 let tmp_path = tmp_path_for(path);
747 fs::write(&tmp_path, serialized).await?;
748 fs::rename(&tmp_path, path).await?;
749 Ok(())
750 }
751
752 pub fn get(&self, id: &str) -> Option<&InstalledPlugin> {
754 self.plugins.iter().find(|plugin| plugin.id == id)
755 }
756
757 pub fn get_unique(&self, id: &str) -> PluginResult<Option<&InstalledPlugin>> {
761 let mut matches = self.plugins.iter().filter(|plugin| plugin.id == id);
762 let first = matches.next();
763 if matches.next().is_some() {
764 return Err(PluginError::Registration(format!(
765 "installed plugin registry contains duplicate rows for id '{id}'"
766 )));
767 }
768 Ok(first)
769 }
770
771 pub fn add(&mut self, plugin: InstalledPlugin) {
774 self.remove(&plugin.id);
775 self.plugins.push(plugin);
776 }
777
778 pub fn remove(&mut self, id: &str) -> Option<InstalledPlugin> {
780 let index = self.plugins.iter().position(|plugin| plugin.id == id)?;
781 Some(self.plugins.remove(index))
782 }
783
784 pub fn list(&self) -> &[InstalledPlugin] {
786 &self.plugins
787 }
788}
789
790fn tmp_path_for(path: &Path) -> PathBuf {
795 let mut tmp_name = path.file_name().unwrap_or_default().to_os_string();
796 tmp_name.push(".tmp");
797 path.with_file_name(tmp_name)
798}
799
800#[cfg(test)]
801mod tests {
802 use super::*;
803
804 fn sample_plugin(id: &str) -> InstalledPlugin {
805 InstalledPlugin {
806 id: id.to_string(),
807 version: "0.1.0".to_string(),
808 source: PluginSource::LocalDir {
809 path: PathBuf::from("/tmp/source"),
810 },
811 plugin_dir: PathBuf::from(format!("/home/user/.bamboo/plugins/{id}")),
812 installed_at: DateTime::parse_from_rfc3339("2026-07-12T00:00:00Z")
813 .unwrap()
814 .with_timezone(&Utc),
815 status: PluginInstallStatus::Installed,
816 registered: RegisteredCapabilities {
817 hooks: vec![],
818 mcp_server_ids: vec![],
819 skill_dirs: vec!["hello-world".to_string()],
820 preset_ids: vec!["hello_preset".to_string()],
821 workflow_filenames: vec![],
822 service_ids: vec![],
823 event_sink_ids: vec![],
824 event_sink_grants: BTreeMap::new(),
825 },
826 }
827 }
828
829 fn event_sink_manifest(service_enabled: bool, protocol_version: u16) -> PluginManifest {
830 let json = serde_json::json!({
831 "id": "event-plugin",
832 "name": "Event Plugin",
833 "version": "1.0.0",
834 "provides": {
835 "services": [{
836 "id": "audit-service",
837 "enabled": service_enabled,
838 "command": "${platform_bin}",
839 "input_protocol": "ndjson_v1"
840 }],
841 "event_sinks": [{
842 "id": "audit-events",
843 "service_id": "audit-service",
844 "protocol": {"name": "tool_event", "version": protocol_version},
845 "subscriptions": [{"id": "tool.file_changed.v1"}],
846 "requested_permissions": ["metadata"]
847 }]
848 }
849 });
850 let manifest = PluginManifest::parse_str(&json.to_string()).expect("parse sink manifest");
851 manifest.validate().expect("validate sink manifest");
852 manifest
853 }
854
855 #[tokio::test]
856 async fn load_missing_file_returns_empty_registry() {
857 let dir = tempfile::tempdir().expect("tempdir");
858 let path = dir.path().join("plugins").join("installed.json");
859 let loaded = InstalledPlugins::load(&path).await.expect("load");
860 assert!(loaded.plugins.is_empty());
861 }
862
863 #[tokio::test]
864 async fn save_is_atomic_via_tmp_file_rename() {
865 let dir = tempfile::tempdir().expect("tempdir");
866 let path = dir.path().join("installed.json");
867 let tmp_path = tmp_path_for(&path);
868
869 let mut store = InstalledPlugins::default();
870 store.add(sample_plugin("hello-plugin"));
871 store.save(&path).await.expect("save");
872
873 assert!(path.exists(), "installed.json should exist after save");
874 assert!(
875 !tmp_path.exists(),
876 "the .tmp staging file must be renamed over the target, never left behind"
877 );
878
879 let mut reloaded = InstalledPlugins::load(&path).await.expect("load");
883 reloaded.add(sample_plugin("other-plugin"));
884 reloaded.save(&path).await.expect("save again");
885 assert!(!tmp_path.exists());
886
887 let loaded = InstalledPlugins::load(&path).await.expect("load");
888 assert_eq!(loaded.plugins.len(), 2);
889 }
890
891 #[tokio::test]
892 async fn save_then_load_round_trips() {
893 let dir = tempfile::tempdir().expect("tempdir");
894 let path = dir.path().join("plugins").join("installed.json");
895
896 let mut store = InstalledPlugins::default();
897 store.add(sample_plugin("hello-plugin"));
898 store.add(sample_plugin("other-plugin"));
899 store.save(&path).await.expect("save");
900
901 let loaded = InstalledPlugins::load(&path).await.expect("load");
902 assert_eq!(loaded.plugins.len(), 2);
903 let hello = loaded.get("hello-plugin").expect("hello-plugin present");
904 assert_eq!(hello.version, "0.1.0");
905 assert_eq!(hello.registered.skill_dirs, vec!["hello-world".to_string()]);
906 assert_eq!(
907 hello.registered.preset_ids,
908 vec!["hello_preset".to_string()]
909 );
910 assert_eq!(
911 hello.source,
912 PluginSource::LocalDir {
913 path: PathBuf::from("/tmp/source")
914 }
915 );
916 }
917
918 #[test]
919 fn legacy_provenance_defaults_and_omits_event_sink_ids() {
920 let raw = r#"{
921 "plugins": [{
922 "id": "legacy-plugin",
923 "version": "1.0.0",
924 "source": {"type": "local_dir", "path": "/tmp/legacy"},
925 "plugin_dir": "/tmp/legacy",
926 "installed_at": "2026-07-12T00:00:00Z",
927 "registered": {"service_ids": ["legacy-service"]}
928 }]
929 }"#;
930 let store: InstalledPlugins = serde_json::from_str(raw).expect("load legacy provenance");
931 assert!(store.plugins[0].registered.event_sink_ids.is_empty());
932
933 let serialized = serde_json::to_value(&store).expect("serialize provenance");
934 assert!(serialized["plugins"][0]["registered"]
935 .get("event_sink_ids")
936 .is_none());
937 assert!(!serde_json::to_string(&store)
938 .expect("serialize legacy provenance bytes")
939 .contains("event_sink_ids"));
940 }
941
942 #[tokio::test]
943 async fn add_upserts_by_id() {
944 let dir = tempfile::tempdir().expect("tempdir");
945 let path = dir.path().join("installed.json");
946
947 let mut store = InstalledPlugins::default();
948 store.add(sample_plugin("hello-plugin"));
949
950 let mut upgraded = sample_plugin("hello-plugin");
951 upgraded.version = "0.2.0".to_string();
952 store.add(upgraded);
953
954 assert_eq!(store.plugins.len(), 1);
955 assert_eq!(store.get("hello-plugin").unwrap().version, "0.2.0");
956
957 store.save(&path).await.expect("save");
958 let loaded = InstalledPlugins::load(&path).await.expect("load");
959 assert_eq!(loaded.plugins.len(), 1);
960 assert_eq!(loaded.get("hello-plugin").unwrap().version, "0.2.0");
961 }
962
963 #[test]
964 fn unique_lookup_rejects_duplicate_plugin_rows() {
965 let mut store = InstalledPlugins::default();
966 store.plugins.push(sample_plugin("hello-plugin"));
967 let mut duplicate = sample_plugin("hello-plugin");
968 duplicate.plugin_dir = PathBuf::from("/tmp/duplicate-plugin-dir");
969 store.plugins.push(duplicate);
970
971 let error = store
972 .get_unique("hello-plugin")
973 .expect_err("duplicate identity must be ambiguous");
974 assert!(matches!(error, PluginError::Registration(_)));
975 assert!(error.to_string().contains("duplicate rows"));
976 assert!(store.get_unique("missing-plugin").unwrap().is_none());
977 }
978
979 #[tokio::test]
980 async fn remove_deletes_and_returns_entry() {
981 let mut store = InstalledPlugins::default();
982 store.add(sample_plugin("hello-plugin"));
983
984 let removed = store.remove("hello-plugin").expect("present before remove");
985 assert_eq!(removed.id, "hello-plugin");
986 assert!(store.get("hello-plugin").is_none());
987 assert!(store.remove("hello-plugin").is_none());
988 }
989
990 #[test]
991 fn reconcile_exclusive_fresh_install_splits_new_from_foreign() {
992 let declared = vec!["a".to_string(), "b".to_string()];
995 let existing = vec!["b".to_string(), "user-thing".to_string()];
996 let owned_previously: Vec<String> = vec![];
997
998 let reconciliation = reconcile_exclusive(&declared, &existing, &owned_previously);
999 assert_eq!(reconciliation.to_register, vec!["a".to_string()]);
1000 assert_eq!(reconciliation.foreign_conflicts, vec!["b".to_string()]);
1001 }
1002
1003 #[test]
1004 fn reconcile_exclusive_upgrade_reregisters_own_but_refuses_new_foreign() {
1005 let declared = vec!["a".to_string(), "c".to_string(), "d".to_string()];
1008 let existing = vec!["a".to_string(), "d".to_string()];
1009 let owned_previously = vec!["a".to_string()];
1010
1011 let reconciliation = reconcile_exclusive(&declared, &existing, &owned_previously);
1012 assert_eq!(
1013 reconciliation.to_register,
1014 vec!["a".to_string(), "c".to_string()]
1015 );
1016 assert_eq!(reconciliation.foreign_conflicts, vec!["d".to_string()]);
1017 }
1018
1019 #[test]
1020 fn classify_ownership_three_way() {
1021 let existing: HashSet<&str> = ["x", "y"].into_iter().collect();
1022 let owned: HashSet<&str> = ["y"].into_iter().collect();
1023 assert_eq!(classify_ownership("z", &existing, &owned), Ownership::New);
1024 assert_eq!(
1025 classify_ownership("y", &existing, &owned),
1026 Ownership::OwnedReinstall
1027 );
1028 assert_eq!(
1029 classify_ownership("x", &existing, &owned),
1030 Ownership::ForeignConflict
1031 );
1032 }
1033
1034 #[test]
1035 fn removed_since_computes_dropped_capabilities_per_kind() {
1036 let old = RegisteredCapabilities {
1037 hooks: vec![],
1038 mcp_server_ids: vec!["srv-a".to_string(), "srv-b".to_string()],
1039 skill_dirs: vec!["skill-a".to_string()],
1040 preset_ids: vec!["preset-a".to_string(), "preset-b".to_string()],
1041 workflow_filenames: vec!["wf-a.md".to_string()],
1042 service_ids: vec!["svc-a".to_string(), "svc-b".to_string()],
1043 event_sink_ids: vec!["sink-a".to_string(), "sink-b".to_string()],
1044 event_sink_grants: BTreeMap::from([(
1045 "sink-a".to_string(),
1046 vec![ObservationPermissionId::new("metadata")],
1047 )]),
1048 };
1049 let new = RegisteredCapabilities {
1051 hooks: vec![],
1052 mcp_server_ids: vec!["srv-a".to_string(), "srv-c".to_string()],
1053 skill_dirs: vec!["skill-a".to_string()],
1054 preset_ids: vec!["preset-b".to_string()],
1055 workflow_filenames: vec!["wf-a.md".to_string()],
1056 service_ids: vec!["svc-a".to_string()],
1057 event_sink_ids: vec!["sink-a".to_string()],
1058 event_sink_grants: BTreeMap::from([(
1059 "sink-a".to_string(),
1060 vec![
1061 ObservationPermissionId::new("metadata"),
1062 ObservationPermissionId::new("paths"),
1063 ],
1064 )]),
1065 };
1066
1067 let removed = new.removed_since(&old);
1068 assert_eq!(removed.mcp_server_ids, vec!["srv-b".to_string()]);
1069 assert!(removed.skill_dirs.is_empty());
1070 assert_eq!(removed.preset_ids, vec!["preset-a".to_string()]);
1071 assert!(removed.workflow_filenames.is_empty());
1072 assert_eq!(removed.service_ids, vec!["svc-b".to_string()]);
1073 assert_eq!(removed.event_sink_ids, vec!["sink-b".to_string()]);
1074 assert!(removed.event_sink_grants.is_empty());
1075 assert!(RegisteredCapabilities {
1076 event_sink_grants: BTreeMap::from([(
1077 "sink-a".to_string(),
1078 vec![ObservationPermissionId::new("metadata")],
1079 )]),
1080 ..Default::default()
1081 }
1082 .is_empty());
1083 }
1084
1085 #[test]
1086 fn event_sink_grants_round_trip_and_legacy_absence_defaults_empty() {
1087 let legacy: RegisteredCapabilities = serde_json::from_value(serde_json::json!({
1088 "event_sink_ids": ["audit-events"]
1089 }))
1090 .unwrap();
1091 assert!(legacy.event_sink_grants.is_empty());
1092
1093 let exact = RegisteredCapabilities {
1094 event_sink_ids: vec!["audit-events".to_string()],
1095 event_sink_grants: BTreeMap::from([(
1096 "audit-events".to_string(),
1097 vec![
1098 ObservationPermissionId::new("metadata"),
1099 ObservationPermissionId::new("paths"),
1100 ],
1101 )]),
1102 ..Default::default()
1103 };
1104 let round_trip: RegisteredCapabilities =
1105 serde_json::from_value(serde_json::to_value(&exact).unwrap()).unwrap();
1106 assert_eq!(round_trip, exact);
1107 }
1108
1109 #[test]
1110 fn event_sink_reconciliation_preserves_order_and_same_plugin_ownership() {
1111 let manifest = event_sink_manifest(true, 1);
1112 let registered = RegisteredCapabilities {
1113 service_ids: vec!["audit-service".to_string()],
1114 event_sink_ids: vec!["audit-events".to_string(), "orphaned".to_string()],
1115 ..Default::default()
1116 };
1117
1118 let plan = reconcile_event_sinks(
1119 &manifest,
1120 ®istered,
1121 PluginInstallStatus::Installed,
1122 Some(Platform::Linux),
1123 )
1124 .expect("reconcile owned sink");
1125 assert_eq!(plan.deactivate_before_services, vec!["orphaned"]);
1126 assert_eq!(
1127 plan.service_dependencies_before_sinks,
1128 vec!["audit-service"]
1129 );
1130 assert_eq!(plan.sinks_after_services.len(), 1);
1131 assert_eq!(plan.sinks_after_services[0].id, "audit-events");
1132 assert_eq!(
1133 plan.sinks_after_services[0].state,
1134 EventSinkCapabilityState::Eligible
1135 );
1136
1137 let removal = registered.removal_order();
1138 assert_eq!(
1139 removal.event_sink_ids_before_services,
1140 vec!["audit-events", "orphaned"]
1141 );
1142 assert_eq!(removal.service_ids_after_sinks, vec!["audit-service"]);
1143 }
1144
1145 #[test]
1146 fn event_sink_reconciliation_fails_closed_on_service_ownership_mismatch() {
1147 let manifest = event_sink_manifest(true, 1);
1148 let registered = RegisteredCapabilities {
1149 event_sink_ids: vec!["audit-events".to_string()],
1150 ..Default::default()
1151 };
1152
1153 let plan = reconcile_event_sinks(
1154 &manifest,
1155 ®istered,
1156 PluginInstallStatus::Installed,
1157 Some(Platform::Linux),
1158 )
1159 .expect("reconcile ownership mismatch");
1160 assert_eq!(plan.deactivate_before_services, vec!["audit-events"]);
1161 assert!(plan.service_dependencies_before_sinks.is_empty());
1162 assert!(plan.sinks_after_services.is_empty());
1163
1164 let mut malformed = manifest;
1165 malformed.provides.event_sinks[0].protocol.name = "tool_evnet".to_string();
1166 assert!(reconcile_event_sinks(
1167 &malformed,
1168 ®istered,
1169 PluginInstallStatus::Installed,
1170 Some(Platform::Linux),
1171 )
1172 .is_err());
1173 }
1174
1175 #[test]
1176 fn installing_and_disabled_sinks_never_request_live_service_dependencies() {
1177 let registered = RegisteredCapabilities {
1178 service_ids: vec!["audit-service".to_string()],
1179 event_sink_ids: vec!["audit-events".to_string()],
1180 ..Default::default()
1181 };
1182
1183 let installing = reconcile_event_sinks(
1184 &event_sink_manifest(true, 1),
1185 ®istered,
1186 PluginInstallStatus::Installing,
1187 Some(Platform::Linux),
1188 )
1189 .expect("reconcile installing sink");
1190 assert!(installing.service_dependencies_before_sinks.is_empty());
1191 assert_eq!(
1192 installing.sinks_after_services[0].state,
1193 EventSinkCapabilityState::Inactive {
1194 detail: crate::manifest::EventSinkInactiveReason::InstallIncomplete,
1195 }
1196 );
1197
1198 let disabled = reconcile_event_sinks(
1199 &event_sink_manifest(false, 1),
1200 ®istered,
1201 PluginInstallStatus::Installed,
1202 Some(Platform::Linux),
1203 )
1204 .expect("reconcile disabled sink");
1205 assert!(disabled.service_dependencies_before_sinks.is_empty());
1206 assert_eq!(
1207 disabled.sinks_after_services[0].state,
1208 EventSinkCapabilityState::Inactive {
1209 detail: crate::manifest::EventSinkInactiveReason::ServiceDisabled,
1210 }
1211 );
1212 }
1213
1214 #[test]
1215 fn reconciliation_applies_the_plugin_level_platform_gate() {
1216 let mut manifest = event_sink_manifest(true, 1);
1217 manifest.platforms = Some(vec![Platform::Macos]);
1218 manifest.validate().expect("macOS-only manifest");
1219 let registered = RegisteredCapabilities {
1220 service_ids: vec!["audit-service".to_string()],
1221 event_sink_ids: vec!["audit-events".to_string()],
1222 ..Default::default()
1223 };
1224
1225 let plan = reconcile_event_sinks(
1226 &manifest,
1227 ®istered,
1228 PluginInstallStatus::Installed,
1229 Some(Platform::Linux),
1230 )
1231 .expect("platform-ineligible plan");
1232 assert!(plan.service_dependencies_before_sinks.is_empty());
1233 assert_eq!(
1234 plan.sinks_after_services[0].state,
1235 EventSinkCapabilityState::Inactive {
1236 detail: crate::manifest::EventSinkInactiveReason::PlatformIneligible,
1237 }
1238 );
1239 }
1240
1241 fn boot_candidate(
1242 id: &str,
1243 manifest: Option<PluginManifest>,
1244 service_ids: &[&str],
1245 sink_ids: &[&str],
1246 status: PluginInstallStatus,
1247 ) -> PluginBootCandidate {
1248 let mut installed = sample_plugin(id);
1249 installed.status = status;
1250 installed.registered.service_ids = service_ids.iter().map(|id| (*id).to_string()).collect();
1251 installed.registered.event_sink_ids = sink_ids.iter().map(|id| (*id).to_string()).collect();
1252 PluginBootCandidate {
1253 installed,
1254 manifest,
1255 }
1256 }
1257
1258 #[test]
1259 fn global_boot_audit_blocks_duplicate_sink_owners_and_their_backing_services() {
1260 let first = event_sink_manifest(true, 1);
1261 let mut second = event_sink_manifest(true, 1);
1262 second.id = "other-plugin".to_string();
1263 second.provides.services[0].id = "other-service".to_string();
1264 second.provides.event_sinks[0].service_id = "other-service".to_string();
1265 second.validate().expect("second manifest");
1266 let candidates = vec![
1267 boot_candidate(
1268 "event-plugin",
1269 Some(first),
1270 &["audit-service"],
1271 &["audit-events"],
1272 PluginInstallStatus::Installed,
1273 ),
1274 boot_candidate(
1275 "other-plugin",
1276 Some(second),
1277 &["other-service"],
1278 &["audit-events"],
1279 PluginInstallStatus::Installed,
1280 ),
1281 ];
1282
1283 let plans = reconcile_plugin_boot(&candidates, Some(Platform::Linux));
1284 assert_eq!(plans.len(), 2);
1285 for plan in plans {
1286 assert!(plan.service_ids_to_start.is_empty());
1287 assert_eq!(
1288 plan.event_sinks.deactivate_before_services,
1289 ["audit-events"]
1290 );
1291 assert!(plan.event_sinks.sinks_after_services.is_empty());
1292 assert!(plan
1293 .issues
1294 .contains(&PluginBootIssue::DuplicateEventSinkOwner {
1295 id: "audit-events".to_string(),
1296 }));
1297 }
1298 }
1299
1300 #[test]
1301 fn global_boot_audit_blocks_duplicate_service_owners_and_dependent_sinks() {
1302 let first = event_sink_manifest(true, 1);
1303 let mut second = event_sink_manifest(true, 1);
1304 second.id = "other-plugin".to_string();
1305 second.provides.event_sinks[0].id = "other-events".to_string();
1306 second.validate().expect("second manifest");
1307 let candidates = vec![
1308 boot_candidate(
1309 "event-plugin",
1310 Some(first),
1311 &["audit-service"],
1312 &["audit-events"],
1313 PluginInstallStatus::Installed,
1314 ),
1315 boot_candidate(
1316 "other-plugin",
1317 Some(second),
1318 &["audit-service"],
1319 &["other-events"],
1320 PluginInstallStatus::Installed,
1321 ),
1322 ];
1323
1324 let plans = reconcile_plugin_boot(&candidates, Some(Platform::Linux));
1325 assert_eq!(
1326 plans[0].event_sinks.deactivate_before_services,
1327 ["audit-events"]
1328 );
1329 assert_eq!(
1330 plans[1].event_sinks.deactivate_before_services,
1331 ["other-events"]
1332 );
1333 for plan in plans {
1334 assert!(plan.service_ids_to_start.is_empty());
1335 assert!(plan.event_sinks.sinks_after_services.is_empty());
1336 assert!(plan
1337 .issues
1338 .contains(&PluginBootIssue::DuplicateServiceOwner {
1339 id: "audit-service".to_string(),
1340 }));
1341 }
1342 }
1343
1344 #[test]
1345 fn global_boot_audit_blocks_duplicate_plugin_rows_but_keeps_safe_plugins() {
1346 let first = event_sink_manifest(true, 1);
1347 let mut second = event_sink_manifest(true, 1);
1348 second.provides.services[0].id = "other-service".to_string();
1349 second.provides.event_sinks[0].id = "other-events".to_string();
1350 second.provides.event_sinks[0].service_id = "other-service".to_string();
1351 second.validate().expect("second same-id manifest");
1352 let mut safe = event_sink_manifest(true, 1);
1353 safe.id = "safe-plugin".to_string();
1354 safe.provides.services[0].id = "safe-service".to_string();
1355 safe.provides.event_sinks[0].id = "safe-events".to_string();
1356 safe.provides.event_sinks[0].service_id = "safe-service".to_string();
1357 safe.validate().expect("safe manifest");
1358
1359 let plans = reconcile_plugin_boot(
1360 &[
1361 boot_candidate(
1362 "event-plugin",
1363 Some(first),
1364 &["audit-service"],
1365 &["audit-events"],
1366 PluginInstallStatus::Installed,
1367 ),
1368 boot_candidate(
1369 "event-plugin",
1370 Some(second),
1371 &["other-service"],
1372 &["other-events"],
1373 PluginInstallStatus::Installed,
1374 ),
1375 boot_candidate(
1376 "safe-plugin",
1377 Some(safe),
1378 &["safe-service"],
1379 &["safe-events"],
1380 PluginInstallStatus::Installed,
1381 ),
1382 ],
1383 Some(Platform::Linux),
1384 );
1385
1386 for plan in &plans[..2] {
1387 assert!(plan.service_ids_to_start.is_empty());
1388 assert_eq!(
1389 plan.issues,
1390 [PluginBootIssue::DuplicatePluginId {
1391 id: "event-plugin".to_string(),
1392 }]
1393 );
1394 }
1395 assert_eq!(plans[2].service_ids_to_start, ["safe-service"]);
1396 assert!(plans[2].issues.is_empty());
1397 }
1398
1399 #[test]
1400 fn global_boot_audit_blocks_incomplete_identity_mismatch_and_unknown_platform() {
1401 let manifest = event_sink_manifest(true, 1);
1402 let installing = boot_candidate(
1403 "event-plugin",
1404 Some(manifest.clone()),
1405 &["audit-service"],
1406 &["audit-events"],
1407 PluginInstallStatus::Installing,
1408 );
1409 let mut mismatch_manifest = manifest.clone();
1410 mismatch_manifest.id = "different-plugin".to_string();
1411 let mismatch = boot_candidate(
1412 "event-plugin",
1413 Some(mismatch_manifest),
1414 &["audit-service"],
1415 &["audit-events"],
1416 PluginInstallStatus::Installed,
1417 );
1418
1419 let installing_plan = reconcile_plugin_boot(&[installing], Some(Platform::Linux));
1420 assert_eq!(
1421 installing_plan[0].issues,
1422 [PluginBootIssue::InstallIncomplete]
1423 );
1424 assert!(installing_plan[0].service_ids_to_start.is_empty());
1425
1426 let mismatch_plan = reconcile_plugin_boot(&[mismatch], Some(Platform::Linux));
1427 assert!(matches!(
1428 mismatch_plan[0].issues.as_slice(),
1429 [PluginBootIssue::ManifestIdMismatch { .. }]
1430 ));
1431 assert!(mismatch_plan[0].service_ids_to_start.is_empty());
1432
1433 let unknown = boot_candidate(
1434 "event-plugin",
1435 Some(manifest),
1436 &["audit-service"],
1437 &["audit-events"],
1438 PluginInstallStatus::Installed,
1439 );
1440 let unknown_plan = reconcile_plugin_boot(&[unknown], None);
1441 assert_eq!(unknown_plan[0].issues, [PluginBootIssue::UnknownPlatform]);
1442 assert!(unknown_plan[0].service_ids_to_start.is_empty());
1443 }
1444
1445 #[test]
1446 fn reconcile_exclusive_covers_service_ids_same_as_other_kinds() {
1447 let declared = vec!["svc-a".to_string(), "svc-b".to_string()];
1451 let existing = vec!["svc-b".to_string(), "other-plugins-svc".to_string()];
1452 let owned_previously: Vec<String> = vec![];
1453
1454 let reconciliation = reconcile_exclusive(&declared, &existing, &owned_previously);
1455 assert_eq!(reconciliation.to_register, vec!["svc-a".to_string()]);
1456 assert_eq!(reconciliation.foreign_conflicts, vec!["svc-b".to_string()]);
1457 }
1458
1459 #[tokio::test]
1460 async fn load_empty_file_returns_empty_registry() {
1461 let dir = tempfile::tempdir().expect("tempdir");
1462 let path = dir.path().join("installed.json");
1463 tokio::fs::create_dir_all(path.parent().unwrap())
1464 .await
1465 .unwrap();
1466 tokio::fs::write(&path, "").await.unwrap();
1467
1468 let loaded = InstalledPlugins::load(&path).await.expect("load");
1469 assert!(loaded.plugins.is_empty());
1470 }
1471}