qualia_core_db/extensions/
extension_bus.rs

1use crate::extension_manifest::ExtensionManifest;
2use crate::{q_hash, NQuin};
3use std::collections::HashMap;
4use std::fs;
5use std::path::{Path, PathBuf};
6use std::sync::RwLock;
7
8/// Global registry of discovered and provisioned capability extensions.
9pub struct ExtensionBus {
10    /// Maps an extension_id to its parsed manifest
11    registered_extensions: RwLock<HashMap<String, ExtensionManifest>>,
12    /// Maps a semantic interface hash (e.g., q_hash("q42:VideoTranscode")) to a list of capable extension IDs
13    capability_index: RwLock<HashMap<u64, Vec<String>>>,
14    /// The local directory where extensions are stored
15    extensions_dir: PathBuf,
16}
17
18impl ExtensionBus {
19    pub fn new(extensions_dir: impl AsRef<Path>) -> Self {
20        let bus = Self {
21            registered_extensions: RwLock::new(HashMap::new()),
22            capability_index: RwLock::new(HashMap::new()),
23            extensions_dir: extensions_dir.as_ref().to_path_buf(),
24        };
25        bus.scan_and_load_extensions();
26        bus
27    }
28
29    /// Scans the extensions directory for `manifest.json` files and loads them into memory.
30    pub fn scan_and_load_extensions(&self) {
31        if !self.extensions_dir.exists() {
32            let _ = fs::create_dir_all(&self.extensions_dir);
33            return;
34        }
35
36        let mut reg_lock = self.registered_extensions.write().unwrap();
37        let mut idx_lock = self.capability_index.write().unwrap();
38
39        if let Ok(entries) = fs::read_dir(&self.extensions_dir) {
40            for entry in entries.flatten() {
41                let path = entry.path();
42                if path.is_dir() {
43                    let manifest_path = path.join("manifest.json");
44                    if manifest_path.exists() {
45                        if let Ok(json_bytes) = fs::read(&manifest_path) {
46                            if let Ok(manifest) = ExtensionManifest::from_json(&json_bytes) {
47                                println!(
48                                    "Loaded extension: {} (v{})",
49                                    manifest.display_name, manifest.version
50                                );
51
52                                // Index capabilities
53                                for cap in &manifest.capabilities {
54                                    let cap_hash = q_hash(&cap.interface);
55                                    idx_lock
56                                        .entry(cap_hash)
57                                        .or_default()
58                                        .push(manifest.extension_id.clone());
59                                }
60
61                                reg_lock.insert(manifest.extension_id.clone(), manifest);
62                            } else {
63                                eprintln!("Failed to parse manifest at {:?}", manifest_path);
64                            }
65                        }
66                    }
67                }
68            }
69        }
70    }
71
72    /// Register a new extension from an absolute path (CLI workflow)
73    pub fn register_extension_from_path(&self, manifest_path: &Path) -> Result<(), String> {
74        let json_bytes =
75            fs::read(manifest_path).map_err(|e| format!("Failed to read manifest: {}", e))?;
76        let manifest = ExtensionManifest::from_json(&json_bytes)
77            .map_err(|e| format!("Invalid manifest JSON: {}", e))?;
78
79        // Copy to local extensions dir
80        let target_dir = self.extensions_dir.join(&manifest.extension_id);
81        fs::create_dir_all(&target_dir)
82            .map_err(|e| format!("Failed to create extension dir: {}", e))?;
83        fs::write(target_dir.join("manifest.json"), &json_bytes)
84            .map_err(|e| format!("Failed to write manifest: {}", e))?;
85
86        let mut reg_lock = self.registered_extensions.write().unwrap();
87        let mut idx_lock = self.capability_index.write().unwrap();
88
89        for cap in &manifest.capabilities {
90            let cap_hash = q_hash(&cap.interface);
91            idx_lock
92                .entry(cap_hash)
93                .or_default()
94                .push(manifest.extension_id.clone());
95        }
96
97        reg_lock.insert(manifest.extension_id.clone(), manifest);
98        Ok(())
99    }
100
101    /// Retrieve all registered manifests (used by the Studio UI)
102    pub fn list_extensions(&self) -> Vec<ExtensionManifest> {
103        let lock = self.registered_extensions.read().unwrap();
104        lock.values().cloned().collect()
105    }
106
107    /// Query the bus for any extensions supporting a given semantic interface hash
108    pub fn query_capability(&self, interface_hash: u64) -> Vec<ExtensionManifest> {
109        let idx_lock = self.capability_index.read().unwrap();
110        let reg_lock = self.registered_extensions.read().unwrap();
111
112        let mut results = Vec::new();
113        if let Some(ext_ids) = idx_lock.get(&interface_hash) {
114            for id in ext_ids {
115                if let Some(manifest) = reg_lock.get(id) {
116                    results.push(manifest.clone());
117                }
118            }
119        }
120        results
121    }
122
123    /// Dispatches a task recipe to an extension over localhost RPC.
124    /// Includes Sentinel Gatekeeping logic to enforce sensitivity classification.
125    pub fn dispatch_task(
126        &self,
127        extension_id: &str,
128        input_file_path: &str,
129        pipeline_steps: Vec<serde_json::Value>,
130        sensitivity_context: u64,
131        guardianship_override: bool,
132    ) -> Result<String, String> {
133        // B6: Sentinel Gatekeeping
134        let sensitivity = sensitivity_context >> 56;
135        if sensitivity == 0x02 && !guardianship_override {
136            // Log violation to WAL
137            let violation_quin = NQuin {
138                subject: q_hash(extension_id),
139                predicate: q_hash("q42:GatekeeperViolation"),
140                object: q_hash(input_file_path),
141                context: sensitivity_context,
142                metadata: 0,
143                parity: 0, // XOR fold omitted for brevity
144            };
145            let _ = crate::wal::append_mutation(&violation_quin);
146
147            return Err("GATEKEEPER_BLOCK: Cannot send Classified (0x02) data to an extension without Guardianship override.".to_string());
148        }
149
150        let reg_lock = self.registered_extensions.read().unwrap();
151        let manifest = reg_lock.get(extension_id).ok_or("Extension not found")?;
152
153        // B5: Task Recipe Payload
154        let payload = serde_json::json!({
155            "input_file_path": input_file_path, // Zero-copy pointer to local file
156            "pipeline_steps": pipeline_steps,
157            "output_routing": "http://127.0.0.1:8080/ingest" // Loopback ingest target
158        });
159
160        // Simulate Dispatch
161        println!(
162            "Dispatching task to {} via {:?}: {}",
163            manifest.display_name, manifest.transport, payload
164        );
165
166        // In a real system, we would open a Reqwest client (LocalHttp) or NamedPipe connection here
167        Ok(format!("Task dispatched successfully to {}", extension_id))
168    }
169
170    /// B4: Provisioning Loop
171    /// Downloads a required asset (e.g., ONNX model weights) for an extension
172    /// into the extension's local data directory.
173    pub async fn provision_asset(
174        &self,
175        extension_id: &str,
176        asset_url: &str,
177        filename: &str,
178    ) -> Result<String, String> {
179        let reg_lock = self.registered_extensions.read().unwrap();
180        if !reg_lock.contains_key(extension_id) {
181            return Err("Extension not found".to_string());
182        }
183
184        let target_dir = self.extensions_dir.join(extension_id).join("assets");
185        std::fs::create_dir_all(&target_dir)
186            .map_err(|e| format!("Failed to create asset dir: {}", e))?;
187
188        let target_file = target_dir.join(filename);
189
190        println!(
191            "Provisioning asset from {} into {:?}",
192            asset_url, target_file
193        );
194
195        // In production, this would use Reqwest to stream the download with a progress bar.
196        // For the Phase B architecture proof, we create a placeholder file.
197        std::fs::write(&target_file, b"MOCK_ASSET_DATA")
198            .map_err(|e| format!("Failed to write asset: {}", e))?;
199
200        Ok(format!(
201            "Asset provisioned successfully at {:?}",
202            target_file
203        ))
204    }
205}
206
207#[cfg(target_arch = "wasm32")]
208pub mod wasm_bus {
209    use serde::Serialize;
210    use std::cell::RefCell;
211    use wasm_bindgen::prelude::*;
212    use wasm_bindgen::JsCast;
213    use web_sys::{ErrorEvent, Event, MessageEvent, WebSocket};
214
215    thread_local! {
216        pub static EXTENSION_BUS: RefCell<Option<ExtensionBusState>> = RefCell::new(None);
217    }
218
219    pub struct ExtensionBusState {
220        pub ws: WebSocket,
221        pub on_open: Closure<dyn FnMut(Event)>,
222        pub on_message: Closure<dyn FnMut(MessageEvent)>,
223        pub on_error: Closure<dyn FnMut(ErrorEvent)>,
224        pub on_close: Closure<dyn FnMut(Event)>,
225        pub is_authenticated: bool,
226        pub active_token_callback: Option<Box<dyn FnMut(String)>>,
227    }
228
229    #[derive(Serialize)]
230    struct ChallengePayload {
231        pub challenge: String,
232        pub did: String,
233    }
234
235    #[derive(Serialize)]
236    struct IntentPayload {
237        pub rpc: String,
238        pub prompt: String,
239        pub graph_context: String,
240        pub signature: String,
241    }
242
243    pub fn init_extension_bus(did: String) -> Result<(), JsValue> {
244        let ws = WebSocket::new("ws://127.0.0.1:4242")?;
245
246        let ws_clone = ws.clone();
247        let did_clone = did.clone();
248
249        let on_open = Closure::wrap(Box::new(move |_e: Event| {
250            let payload = ChallengePayload {
251                challenge: "did:q42".into(),
252                did: did_clone.clone(),
253            };
254            if let Ok(json) = serde_json::to_string(&payload) {
255                let _ = ws_clone.send_with_str(&json);
256            }
257        }) as Box<dyn FnMut(Event)>);
258        ws.set_onopen(Some(on_open.as_ref().unchecked_ref()));
259
260        let on_message = Closure::wrap(Box::new(move |e: MessageEvent| {
261            if let Ok(txt) = e.data().dyn_into::<js_sys::JsString>() {
262                let s: String = txt.into();
263                // Parse the response
264                if s.contains("\"authenticated\":true") {
265                    EXTENSION_BUS.with(|bus| {
266                        if let Some(state) = bus.borrow_mut().as_mut() {
267                            state.is_authenticated = true;
268                        }
269                    });
270                } else if s.contains("\"token\":") || s.contains("\"text\":") {
271                    if let Ok(v) = serde_json::from_str::<serde_json::Value>(&s) {
272                        if let Some(text) = v.get("text").and_then(|t| t.as_str()) {
273                            EXTENSION_BUS.with(|bus| {
274                                if let Some(state) = bus.borrow_mut().as_mut() {
275                                    if let Some(ref mut cb) = state.active_token_callback {
276                                        cb(text.to_string());
277                                    }
278                                }
279                            });
280                        }
281                    }
282                }
283            }
284        }) as Box<dyn FnMut(MessageEvent)>);
285        ws.set_onmessage(Some(on_message.as_ref().unchecked_ref()));
286
287        let on_error = Closure::wrap(Box::new(move |_e: ErrorEvent| {
288            // Placeholder for error telemetry
289        }) as Box<dyn FnMut(ErrorEvent)>);
290        ws.set_onerror(Some(on_error.as_ref().unchecked_ref()));
291
292        let on_close = Closure::wrap(Box::new(move |_e: Event| {
293            EXTENSION_BUS.with(|bus| {
294                *bus.borrow_mut() = None;
295            });
296        }) as Box<dyn FnMut(Event)>);
297        ws.set_onclose(Some(on_close.as_ref().unchecked_ref()));
298
299        EXTENSION_BUS.with(|bus| {
300            *bus.borrow_mut() = Some(ExtensionBusState {
301                ws,
302                on_open,
303                on_message,
304                on_error,
305                on_close,
306                is_authenticated: false,
307                active_token_callback: None,
308            });
309        });
310
311        Ok(())
312    }
313
314    pub fn is_connected() -> bool {
315        EXTENSION_BUS.with(|bus| {
316            bus.borrow()
317                .as_ref()
318                .map(|s| s.is_authenticated)
319                .unwrap_or(false)
320        })
321    }
322
323    pub fn send_intent<F: FnMut(String) + 'static>(
324        prompt: &str,
325        graph_context: &str,
326        on_token: F,
327    ) -> Result<(), String> {
328        let payload = IntentPayload {
329            rpc: "infer_local_model".into(),
330            prompt: prompt.to_string(),
331            graph_context: graph_context.to_string(),
332            signature: "did:q42:active".into(),
333        };
334        let intent_json = serde_json::to_string(&payload).unwrap_or_default();
335
336        EXTENSION_BUS.with(|bus| {
337            if let Some(state) = bus.borrow_mut().as_mut() {
338                if state.is_authenticated {
339                    state.active_token_callback = Some(Box::new(on_token));
340                    state
341                        .ws
342                        .send_with_str(&intent_json)
343                        .map_err(|e| format!("{:?}", e))?;
344                    return Ok(());
345                }
346            }
347            Err("Not connected or authenticated".into())
348        })
349    }
350}
qualia_core_db/extensions/extension_bus.rs

qualia_core_db/extensions/
extension_bus.rs
