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