turbovault_plugin_api/
hooks.rs

1use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
2use std::sync::{Arc, Mutex};
3use std::time::{SystemTime, UNIX_EPOCH};
4
5use serde::{Deserialize, Serialize};
6use tokio::sync::broadcast;
7
8/// Provenance supplied by a writer for best-effort event correlation.
9///
10/// Every field is caller-supplied and therefore unverified. For trustworthy
11/// loop prevention use [`VaultEventEnvelope::plugin_id`], which the host
12/// stamps.
13#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
14#[non_exhaustive]
15pub struct WriteProvenance {
16    /// Stable caller-selected source, such as `daily-review-agent`.
17    pub source: String,
18    /// Optional identifier connecting related operations.
19    pub correlation_id: Option<String>,
20    /// Optional human-readable reason.
21    pub note: Option<String>,
22}
23
24impl WriteProvenance {
25    /// Construct provenance for `source`.
26    pub fn new(source: impl Into<String>) -> Self {
27        Self {
28            source: source.into(),
29            correlation_id: None,
30            note: None,
31        }
32    }
33
34    /// Link this write to related operations.
35    pub fn with_correlation_id(mut self, correlation_id: impl Into<String>) -> Self {
36        self.correlation_id = Some(correlation_id.into());
37        self
38    }
39
40    /// Explain why the write happened.
41    pub fn with_note(mut self, note: impl Into<String>) -> Self {
42        self.note = Some(note.into());
43        self
44    }
45}
46
47/// Best-effort event attribution.
48///
49/// This is advisory loop-prevention metadata, never an authorization or
50/// security boundary.
51#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
52#[serde(rename_all = "snake_case", tag = "kind", content = "provenance")]
53#[non_exhaustive]
54pub enum EventAttribution {
55    /// The host correlated the observed content with a known write.
56    Attributed(WriteProvenance),
57    /// No trusted correlation was available; the change may be external.
58    ExternalOrUnknown,
59}
60
61/// A vault mutation observed by a hook producer.
62#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
63#[serde(rename_all = "snake_case", tag = "kind")]
64#[non_exhaustive]
65pub enum HookEvent {
66    /// A note was created.
67    FileCreated {
68        /// Vault-relative path.
69        path: String,
70    },
71    /// A note was modified.
72    FileModified {
73        /// Vault-relative path.
74        path: String,
75    },
76    /// A note was deleted.
77    FileDeleted {
78        /// Vault-relative path.
79        path: String,
80    },
81    /// A note moved between two paths.
82    FileRenamed {
83        /// Original vault-relative path.
84        from: String,
85        /// New vault-relative path.
86        to: String,
87    },
88    /// The authoritative state must be re-read after a producer reset.
89    ResyncRequired {
90        /// Human-readable reason for the reset.
91        reason: String,
92    },
93}
94
95/// Sequenced event delivered to hook subscribers.
96///
97/// Sequence numbers are assigned and the envelope is enqueued inside one
98/// critical section, so subscribers observe events in sequence order.
99#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
100#[non_exhaustive]
101pub struct VaultEventEnvelope {
102    /// Identifier of the bus run that issued [`Self::sequence`].
103    ///
104    /// See [`HookBus::epoch`]. Persist it alongside the sequence — an
105    /// [`EventCursor`] does this — or the number means nothing after a restart.
106    pub epoch: String,
107    /// Sequence number, monotonic within [`Self::epoch`] and only there.
108    ///
109    /// It counts from zero each time the host starts.
110    pub sequence: u64,
111    /// Observation time as Unix epoch milliseconds.
112    pub observed_at_ms: u64,
113    /// Vault name.
114    pub vault: String,
115    /// Observed mutation.
116    pub event: HookEvent,
117    /// Content identity when it was available at observation time.
118    pub content_hash: Option<String>,
119    /// Mounted plugin namespace the host attributes this write to.
120    ///
121    /// Stamped by the host from the validated descriptor of the plugin whose
122    /// call produced the write, and not settable by the writer — unlike
123    /// [`WriteProvenance`], which is advisory caller data. `None` means the
124    /// change did not come through a plugin. This is the field to use for loop
125    /// prevention: a plugin skips events whose `plugin_id` is its own.
126    pub plugin_id: Option<String>,
127    /// Best-effort writer attribution.
128    pub attribution: EventAttribution,
129}
130
131/// Current event-bus lifecycle.
132#[derive(Debug, Clone, Copy, PartialEq, Eq)]
133pub enum HookLifecycle {
134    /// New events may be published.
135    Running,
136    /// No further events will be accepted.
137    Closed,
138}
139
140/// Publish failure.
141#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
142pub enum PublishError {
143    /// The bus was explicitly closed.
144    #[error("hook bus is closed")]
145    Closed,
146}
147
148/// Receive failure with explicit overflow semantics.
149#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
150pub enum HookRecvError {
151    /// No event is currently available to a non-blocking receiver.
152    #[error("no hook event is currently available")]
153    Empty,
154    /// This subscriber fell behind the bounded ring buffer.
155    ///
156    /// The consumer must re-read authoritative vault state before continuing.
157    #[error("hook subscriber lagged by {skipped} event(s); resync required")]
158    Lagged {
159        /// Number of discarded envelopes.
160        skipped: u64,
161    },
162    /// The bus closed and all buffered events were drained.
163    #[error("hook bus is closed")]
164    Closed,
165}
166
167/// A consumer's durable position in the event feed.
168///
169/// [`VaultEventEnvelope::sequence`] restarts at zero every time the host
170/// starts, so a bare sequence number persisted across a restart is worse than
171/// keeping no record at all: it looks comparable to the new run's numbering and
172/// is not. Pairing the sequence with the epoch that issued it turns that silent
173/// mis-resume into a question a plugin can answer — [`Self::resumes_on`].
174///
175/// Serializable, so it stores directly through [`crate::PluginStorage`].
176#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
177pub struct EventCursor {
178    epoch: String,
179    sequence: u64,
180}
181
182impl EventCursor {
183    /// The position immediately after `envelope` has been applied.
184    pub fn after(envelope: &VaultEventEnvelope) -> Self {
185        Self {
186            epoch: envelope.epoch.clone(),
187            sequence: envelope.sequence,
188        }
189    }
190
191    /// The bus run that issued [`Self::sequence`].
192    pub fn epoch(&self) -> &str {
193        &self.epoch
194    }
195
196    /// The last applied sequence within [`Self::epoch`].
197    pub fn sequence(&self) -> u64 {
198        self.sequence
199    }
200
201    /// Whether this position still means something on `bus`.
202    ///
203    /// `false` after a host restart, which is the case worth handling: the
204    /// plugin cannot know what changed while it was down, so it must reconcile
205    /// from [`crate::VaultApi::list_notes_detailed`] before trusting the feed.
206    ///
207    /// `true` is not a promise that nothing was missed — a subscriber can still
208    /// fall behind within a run, which arrives separately as
209    /// [`HookRecvError::Lagged`].
210    pub fn resumes_on(&self, bus: &HookBus) -> bool {
211        self.epoch == bus.epoch()
212    }
213}
214
215struct HookBusInner {
216    sender: Mutex<Option<broadcast::Sender<VaultEventEnvelope>>>,
217    epoch: String,
218    sequence: AtomicU64,
219    closed: AtomicBool,
220    capacity: usize,
221}
222
223/// Cloneable bounded event bus shared by a host and its plugins.
224///
225/// Delivery is best-effort. A slow subscriber receives
226/// [`HookRecvError::Lagged`] and must resynchronize from [`crate::VaultApi`].
227#[derive(Clone)]
228pub struct HookBus {
229    inner: Arc<HookBusInner>,
230}
231
232impl std::fmt::Debug for HookBus {
233    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
234        formatter
235            .debug_struct("HookBus")
236            .field("epoch", &self.epoch())
237            .field("capacity", &self.capacity())
238            .field("lifecycle", &self.lifecycle())
239            .finish()
240    }
241}
242
243impl HookBus {
244    fn sender(&self) -> std::sync::MutexGuard<'_, Option<broadcast::Sender<VaultEventEnvelope>>> {
245        self.inner
246            .sender
247            .lock()
248            .unwrap_or_else(std::sync::PoisonError::into_inner)
249    }
250
251    /// Create a bus with a fixed per-subscriber ring-buffer capacity.
252    ///
253    /// A zero capacity is promoted to one.
254    pub fn new(capacity: usize) -> Self {
255        let capacity = capacity.max(1);
256        let (sender, _) = broadcast::channel(capacity);
257        Self {
258            inner: Arc::new(HookBusInner {
259                sender: Mutex::new(Some(sender)),
260                epoch: uuid::Uuid::new_v4().to_string(),
261                sequence: AtomicU64::new(0),
262                closed: AtomicBool::new(false),
263                capacity,
264            }),
265        }
266    }
267
268    /// Return this bus run's identifier.
269    ///
270    /// Freshly generated per bus, so it differs on every host start. The feed
271    /// is in-memory: it has no history from before this run and its sequence
272    /// numbering is meaningless outside it. The epoch is what makes that
273    /// detectable rather than silent — see [`EventCursor`].
274    ///
275    /// Opaque and unordered; compare for equality only.
276    pub fn epoch(&self) -> &str {
277        &self.inner.epoch
278    }
279
280    /// Return the fixed ring-buffer capacity.
281    pub fn capacity(&self) -> usize {
282        self.inner.capacity
283    }
284
285    /// Return whether new events are accepted.
286    pub fn lifecycle(&self) -> HookLifecycle {
287        if self.inner.closed.load(Ordering::Acquire) {
288            HookLifecycle::Closed
289        } else {
290            HookLifecycle::Running
291        }
292    }
293
294    /// Subscribe to future events.
295    pub fn subscribe(&self) -> Result<HookSubscription, PublishError> {
296        let guard = self.sender();
297        let sender = guard.as_ref().ok_or(PublishError::Closed)?;
298        Ok(HookSubscription {
299            receiver: sender.subscribe(),
300        })
301    }
302
303    /// Sequence and publish an advisory event.
304    ///
305    /// Success does not imply that a subscriber exists. `plugin_id` is the
306    /// host's own attribution of the writer and must never be taken from
307    /// caller-supplied data.
308    ///
309    /// The sender mutex is held across both the sequence increment and the
310    /// enqueue, so two concurrent publishes cannot be delivered out of
311    /// sequence order.
312    pub fn publish(
313        &self,
314        vault: impl Into<String>,
315        event: HookEvent,
316        content_hash: Option<String>,
317        plugin_id: Option<String>,
318        attribution: EventAttribution,
319    ) -> Result<VaultEventEnvelope, PublishError> {
320        let guard = self.sender();
321        let sender = guard.as_ref().ok_or(PublishError::Closed)?;
322        let envelope = VaultEventEnvelope {
323            epoch: self.inner.epoch.clone(),
324            sequence: self.inner.sequence.fetch_add(1, Ordering::Relaxed) + 1,
325            observed_at_ms: SystemTime::now()
326                .duration_since(UNIX_EPOCH)
327                .unwrap_or_default()
328                .as_millis()
329                .try_into()
330                .unwrap_or(u64::MAX),
331            vault: vault.into(),
332            event,
333            content_hash,
334            plugin_id,
335            attribution,
336        };
337        let _subscriber_count = sender.send(envelope.clone());
338        Ok(envelope)
339    }
340
341    /// Stop accepting events and close subscribers after buffered delivery.
342    ///
343    /// Returns `true` only for the call that changed the lifecycle.
344    pub fn close(&self) -> bool {
345        let mut guard = self.sender();
346        let changed = guard.take().is_some();
347        self.inner.closed.store(true, Ordering::Release);
348        changed
349    }
350}
351
352/// A single bounded subscription to [`HookBus`].
353pub struct HookSubscription {
354    receiver: broadcast::Receiver<VaultEventEnvelope>,
355}
356
357impl std::fmt::Debug for HookSubscription {
358    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
359        formatter
360            .debug_struct("HookSubscription")
361            .finish_non_exhaustive()
362    }
363}
364
365impl HookSubscription {
366    /// Wait for the next event or an explicit lag/close condition.
367    pub async fn recv(&mut self) -> Result<VaultEventEnvelope, HookRecvError> {
368        self.receiver.recv().await.map_err(|error| match error {
369            broadcast::error::RecvError::Closed => HookRecvError::Closed,
370            broadcast::error::RecvError::Lagged(skipped) => HookRecvError::Lagged { skipped },
371        })
372    }
373
374    /// Receive immediately without waiting.
375    pub fn try_recv(&mut self) -> Result<VaultEventEnvelope, HookRecvError> {
376        self.receiver.try_recv().map_err(|error| match error {
377            broadcast::error::TryRecvError::Closed => HookRecvError::Closed,
378            broadcast::error::TryRecvError::Lagged(skipped) => HookRecvError::Lagged { skipped },
379            broadcast::error::TryRecvError::Empty => HookRecvError::Empty,
380        })
381    }
382}
383
384#[cfg(test)]
385mod tests {
386    use super::*;
387
388    fn publish(bus: &HookBus, path: &str) -> VaultEventEnvelope {
389        bus.publish(
390            "test",
391            HookEvent::FileModified {
392                path: path.to_string(),
393            },
394            None,
395            None,
396            EventAttribution::ExternalOrUnknown,
397        )
398        .expect("publish")
399    }
400
401    #[tokio::test]
402    async fn sequences_events_and_closes_after_buffered_delivery() {
403        // Capacity two retains both events until this subscriber reads them.
404        const CAPACITY: usize = 2;
405        let bus = HookBus::new(CAPACITY);
406        let mut subscription = bus.subscribe().expect("subscribe");
407        assert_eq!(publish(&bus, "a.md").sequence, 1);
408        assert_eq!(publish(&bus, "b.md").sequence, 2);
409        assert!(bus.close());
410        assert!(!bus.close());
411
412        assert_eq!(subscription.recv().await.expect("first").sequence, 1);
413        assert_eq!(subscription.recv().await.expect("second").sequence, 2);
414        assert_eq!(subscription.recv().await, Err(HookRecvError::Closed));
415        assert_eq!(bus.lifecycle(), HookLifecycle::Closed);
416    }
417
418    #[tokio::test]
419    async fn reports_lag_and_requires_resync() {
420        // Capacity one intentionally evicts the first of two unread events.
421        const CAPACITY: usize = 1;
422        let bus = HookBus::new(CAPACITY);
423        let mut subscription = bus.subscribe().expect("subscribe");
424        publish(&bus, "a.md");
425        publish(&bus, "b.md");
426
427        assert_eq!(
428            subscription.recv().await,
429            Err(HookRecvError::Lagged { skipped: 1 })
430        );
431        assert_eq!(subscription.recv().await.expect("retained").sequence, 2);
432    }
433
434    #[test]
435    fn cursors_do_not_resume_across_a_restart() {
436        // Two buses stand in for two runs of the host.
437        let first_run = HookBus::new(4);
438        let second_run = HookBus::new(4);
439        assert_ne!(first_run.epoch(), second_run.epoch());
440
441        let applied = EventCursor::after(&publish(&first_run, "a.md"));
442        assert_eq!(applied.sequence(), 1);
443        assert!(applied.resumes_on(&first_run));
444
445        // The sequence number alone would compare as "already seen" against the
446        // new run's first event. The epoch is what makes the mismatch visible.
447        let after_restart = publish(&second_run, "b.md");
448        assert_eq!(after_restart.sequence, applied.sequence());
449        assert!(!applied.resumes_on(&second_run));
450
451        // Round-trips through storage unchanged.
452        let encoded = serde_json::to_vec(&applied).expect("serialize cursor");
453        let decoded: EventCursor = serde_json::from_slice(&encoded).expect("deserialize cursor");
454        assert_eq!(decoded, applied);
455        assert!(decoded.resumes_on(&first_run));
456    }
457
458    #[test]
459    fn rejects_new_work_after_close() {
460        let bus = HookBus::new(0);
461        assert_eq!(bus.capacity(), 1);
462        assert!(bus.close());
463        assert!(matches!(bus.subscribe(), Err(PublishError::Closed)));
464        assert_eq!(
465            bus.publish(
466                "test",
467                HookEvent::ResyncRequired {
468                    reason: "closed".to_string()
469                },
470                None,
471                None,
472                EventAttribution::ExternalOrUnknown
473            ),
474            Err(PublishError::Closed)
475        );
476    }
477}
turbovault_plugin_api/hooks.rs

turbovault_plugin_api/
hooks.rs
