Vakyartha documentation

1//! The capability registry: one owner, one reconcile loop, one projection.
2//!
3//! # Why a loop and not a cache
4//!
5//! Everything this replaces was **edge-triggered**: discovery warmed once, a
6//! prompt froze once, a connection opened once, a catalog cached once — with
7//! a retry guard written so that caching a *failure* counted as having
8//! succeeded. A single missed edge was therefore permanent, and uptime is
9//! what converts a small race into a dead integration.
10//!
11//! This is **level-triggered**, the way long-running systems are built:
12//! `reconcile()` compares desired state to observed state and moves toward
13//! it, idempotently. Hints (a filesystem event, an MCP `list_changed`, a
14//! plugin toggle) only make it run *sooner*; the ticker guarantees it runs
15//! anyway. A dropped hint costs one tick of latency. It never costs
16//! correctness.
17//!
18//! # Two channels, deliberately asymmetric
19//!
20//! Additions and catalog changes take effect at the next **turn boundary**,
21//! via a published epoch. Revocations take effect **immediately**, mid-turn,
22//! fail-closed. That split follows a rule the codebase already believes:
23//! narrowing is always safe, widening needs admission. An operator disabling
24//! a compromised plugin must not wait for a long turn to finish, while a
25//! newly added skill appearing halfway through a plan would be a torn read.
26//!
27//! # Offline by construction
28//!
29//! A pass never talks to anything. Declarations are a filesystem walk at
30//! worst, and an MCP server's catalog and failure arrive as declared data
31//! from the on-demand pool (`vak_mcp::McpManager`), which is the only thing
32//! allowed to start a server. So reconciling is cheap enough to run at every
33//! turn admission, and it can never spawn an integration nobody asked for.
34
35use std::collections::{BTreeMap, BTreeSet};
36use std::sync::Arc;
37use std::sync::atomic::{AtomicU64, Ordering};
38use std::time::{Duration, SystemTime};
39
40use async_trait::async_trait;
41use serde::{Deserialize, Serialize};
42use tokio::sync::{RwLock, mpsc};
43
44use super::domain::Serves;
45use super::resolution::Resolution;
46use super::snapshot::{Capability, CapabilityDelta, CapabilityId, CapabilitySet, Epoch, Origin};
47
48/// How often the loop reconciles with no hint at all. The safety net that
49/// makes a missed event survivable.
50pub const RECONCILE_INTERVAL: Duration = Duration::from_secs(10);
51
52/// Hints are coalesced over this window, so an editor writing six files does
53/// one reconcile rather than six.
54pub const DEBOUNCE: Duration = Duration::from_millis(250);
55
56/// One capability as its source declares it.
57#[derive(Debug, Clone, PartialEq, Eq)]
58pub struct Declaration {
59 pub id: CapabilityId,
60 pub origin: Origin,
61 pub summary: String,
62 pub serves: Serves,
63 pub digest: Option<String>,
64 pub source: Option<std::path::PathBuf>,
65 /// Kind-specific detail, part of the published digest — for an MCP
66 /// server, what the pool has observed (catalog, last failure).
67 pub configuration: serde_json::Value,
68}
69
70/// Where declarations come from.
71///
72/// Implemented by `Core`, which already owns skill/hook/command/MCP
73/// discovery. Keeping it behind a trait means the loop is testable without a
74/// workspace, a filesystem, or a live MCP server.
75#[async_trait]
76pub trait CapabilityProvider: Send + Sync {
77 /// Everything currently declared. Cheap and offline: a filesystem walk
78 /// at worst, never a network call or a spawned process.
79 fn declare(&self) -> Vec<Declaration>;
80
81 /// Housekeeping on each pass, such as evicting idle pooled connections.
82 /// It may only ever release resources, never acquire them.
83 async fn upkeep(&self) {}
84}
85
86/// Why the loop woke up. Purely an optimisation — the ticker would have
87/// caught all of these eventually.
88#[derive(Debug, Clone, PartialEq, Eq)]
89pub enum Hint {
90 /// A watched capability directory changed.
91 SourceChanged(String),
92 /// The MCP pool learned something about a server (a catalog, a
93 /// failure), or the server announced a catalog change.
94 ServerObserved(String),
95 /// Configuration or plugin enablement moved.
96 ConfigChanged,
97 /// Someone asked for an immediate pass (boot, or a CLI about to run one
98 /// turn and exit).
99 Immediate,
100}
101
102/// The health of the loop itself.
103///
104/// Exposed so "capabilities look thin because reconciliation has been
105/// failing for two hours" is a visible fact rather than an unexplained
106/// shortage of tools.
107#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
108pub struct ReconcileStatus {
109 pub last_run: Option<SystemTime>,
110 pub last_change: Option<SystemTime>,
111 pub epoch: Epoch,
112 pub passes: u64,
113}
114
115/// The registry.
116pub struct CapabilityRegistry {
117 provider: Arc<dyn CapabilityProvider>,
118 current: RwLock<Arc<CapabilitySet>>,
119 next_epoch: AtomicU64,
120 /// The immediate channel. Checked at dispatch, independent of any epoch.
121 revoked: RwLock<BTreeMap<CapabilityId, String>>,
122 revoked_fast: std::sync::RwLock<BTreeMap<CapabilityId, String>>,
123 status: RwLock<ReconcileStatus>,
124 hints: mpsc::UnboundedSender<Hint>,
125}
126
127impl CapabilityRegistry {
128 pub fn new(
129 provider: Arc<dyn CapabilityProvider>,
130 ) -> (Arc<Self>, mpsc::UnboundedReceiver<Hint>) {
131 let (tx, rx) = mpsc::unbounded_channel();
132 let registry = Arc::new(CapabilityRegistry {
133 provider,
134 current: RwLock::new(Arc::new(CapabilitySet::empty())),
135 next_epoch: AtomicU64::new(1),
136 revoked: RwLock::new(BTreeMap::new()),
137 revoked_fast: std::sync::RwLock::new(BTreeMap::new()),
138 status: RwLock::new(ReconcileStatus::default()),
139 hints: tx,
140 });
141 (registry, rx)
142 }
143
144 /// Ask for a reconcile sooner than the ticker would. Never blocks and
145 /// never fails meaningfully: if the loop is gone, the hint is moot.
146 pub fn hint(&self, hint: Hint) {
147 let _ = self.hints.send(hint);
148 }
149
150 /// The current published set. What a turn binds at its start.
151 pub async fn current(&self) -> Arc<CapabilitySet> {
152 self.current.read().await.clone()
153 }
154
155 /// Best-effort synchronous read of the current published set, for
156 /// diagnostics/reporting call sites (`capability_diagnostics`, `/doctor`,
157 /// the system prompt) that are not themselves async. Reconciliation is
158 /// an eventually-consistent background loop by design — a rare
159 /// contended `try_read` just means this diagnostic reflects the
160 /// previous epoch for one more instant, not a correctness problem.
161 /// Falls back to an empty set only if the lock is actually contended
162 /// (never blocks), which a caller reporting "nothing degraded" during
163 /// that instant is a harmless, self-correcting understatement.
164 pub fn current_blocking(&self) -> Arc<CapabilitySet> {
165 self.current
166 .try_read()
167 .map(|guard| guard.clone())
168 .unwrap_or_else(|_| Arc::new(CapabilitySet::empty()))
169 }
170
171 pub async fn status(&self) -> ReconcileStatus {
172 self.status.read().await.clone()
173 }
174
175 /// Revoke immediately, mid-turn, fail-closed.
176 ///
177 /// Does not wait for an epoch: a compromised plugin must stop being
178 /// callable now, not when the current turn happens to finish. This only
179 /// ever narrows, which is why it is safe to apply without admission.
180 pub async fn revoke(&self, id: CapabilityId, reason: impl Into<String>) {
181 let reason = reason.into();
182 self.revoked
183 .write()
184 .await
185 .insert(id.clone(), reason.clone());
186 if let Ok(mut revoked) = self.revoked_fast.write() {
187 revoked.insert(id, reason);
188 }
189 self.hint(Hint::ConfigChanged);
190 }
191
192 pub async fn restore(&self, id: &CapabilityId) {
193 self.revoked.write().await.remove(id);
194 if let Ok(mut revoked) = self.revoked_fast.write() {
195 revoked.remove(id);
196 }
197 self.hint(Hint::ConfigChanged);
198 }
199
200 pub fn revoked_now(&self, id: &CapabilityId) -> bool {
201 self.revoked_fast
202 .read()
203 .ok()
204 .is_some_and(|revoked| revoked.contains_key(id))
205 }
206
207 /// Whether the declared world differs from the published set, so turn
208 /// admission knows a pass is worth running before it binds an epoch.
209 pub async fn has_pending_changes(&self) -> bool {
210 let declarations = self.provider.declare();
211 let current = self.current.read().await;
212 declarations.len() != current.all().count()
213 || declarations.iter().any(|d| match current.get(&d.id) {
214 None => true,
215 Some(existing) => {
216 existing.digest != d.digest || existing.configuration != d.configuration
217 }
218 })
219 }
220
221 /// Whether `id` is revoked right now, regardless of the caller's epoch.
222 /// Dispatch checks this; availability comes from the bound epoch, but
223 /// authorization always comes from the present.
224 pub async fn revocation(&self, id: &CapabilityId) -> Option<String> {
225 self.revoked.read().await.get(id).cloned()
226 }
227
228 pub async fn revoked_ids(&self) -> BTreeSet<CapabilityId> {
229 self.revoked.read().await.keys().cloned().collect()
230 }
231
232 /// One idempotent pass. Safe to call at any time, from anywhere, as
233 /// often as you like — that is the whole point of a level-triggered
234 /// design. Returns the delta if a new epoch was published.
235 pub async fn reconcile(&self) -> Option<CapabilityDelta> {
236 let now = SystemTime::now();
237 let previous = self.current.read().await.clone();
238 let declarations = self.provider.declare();
239
240 // A revoked capability is retired in the published set as well as
241 // blocked at dispatch, so the prompt stops advertising it at the next
242 // turn instead of describing a tool that will always refuse.
243 let revoked = self.revoked.read().await.clone();
244 let capabilities: Vec<Capability> = declarations
245 .into_iter()
246 .map(|declaration| Capability {
247 resolution: match revoked.get(&declaration.id) {
248 Some(reason) => Resolution::Retired {
249 reason: reason.clone(),
250 },
251 None => Resolution::Available,
252 },
253 id: declaration.id,
254 origin: declaration.origin,
255 summary: declaration.summary,
256 serves: declaration.serves,
257 digest: declaration.digest,
258 source: declaration.source,
259 configuration: declaration.configuration,
260 })
261 .collect();
262
263 let candidate = CapabilitySet::new(previous.epoch, capabilities);
264
265 let mut status = self.status.write().await;
266 status.passes += 1;
267 status.last_run = Some(now);
268
269 if candidate.digest == previous.digest {
270 status.epoch = previous.epoch;
271 return None;
272 }
273
274 // Publish. Only here does the epoch move, and only because the world
275 // actually differs.
276 let epoch = self.next_epoch.fetch_add(1, Ordering::Relaxed);
277 let published = Arc::new(CapabilitySet::new(
278 epoch,
279 candidate.all().cloned().collect(),
280 ));
281 let delta = published.delta_from(&previous);
282 *self.current.write().await = published;
283 status.epoch = epoch;
284 status.last_change = Some(now);
285 Some(delta)
286 }
287
288 /// Run the loop until `shutdown` fires.
289 ///
290 /// Selects over the ticker and the hint channel; a burst of hints is
291 /// coalesced into one pass. The ticker is what makes a dropped hint a
292 /// latency problem rather than a correctness one.
293 pub async fn run(
294 self: Arc<Self>,
295 mut hints: mpsc::UnboundedReceiver<Hint>,
296 mut shutdown: tokio::sync::watch::Receiver<bool>,
297 ) {
298 let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
299 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
300 loop {
301 tokio::select! {
302 _ = ticker.tick() => {}
303 hint = hints.recv() => {
304 if hint.is_none() {
305 return;
306 }
307 // Coalesce a burst: an editor saving six files should
308 // produce one pass, not six.
309 tokio::time::sleep(DEBOUNCE).await;
310 while hints.try_recv().is_ok() {}
311 }
312 _ = shutdown.changed() => {
313 if *shutdown.borrow() {
314 return;
315 }
316 continue;
317 }
318 }
319 self.reconcile().await;
320 self.provider.upkeep().await;
321 }
322 }
323}
324
325#[cfg(test)]
326#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
327mod tests {
328 use super::*;
329 use std::sync::Mutex;
330 use vak_session::types::CapabilityKind;
331
332 /// A provider whose declarations the test drives.
333 struct Fake {
334 declarations: Mutex<Vec<Declaration>>,
335 }
336
337 impl Fake {
338 fn new(declarations: Vec<Declaration>) -> Arc<Self> {
339 Arc::new(Fake {
340 declarations: Mutex::new(declarations),
341 })
342 }
343 }
344
345 #[async_trait]
346 impl CapabilityProvider for Fake {
347 fn declare(&self) -> Vec<Declaration> {
348 self.declarations.lock().unwrap().clone()
349 }
350 }
351
352 fn decl(name: &str, kind: CapabilityKind) -> Declaration {
353 Declaration {
354 id: CapabilityId::new(kind, name),
355 origin: Origin::Workspace,
356 summary: String::new(),
357 serves: Serves::Undeclared,
358 digest: None,
359 source: None,
360 configuration: serde_json::Value::Null,
361 }
362 }
363
364 #[tokio::test]
365 async fn a_first_pass_publishes_an_epoch() {
366 let provider = Fake::new(vec![decl("read", CapabilityKind::Tool)]);
367 let (registry, _rx) = CapabilityRegistry::new(provider);
368 assert!(registry.reconcile().await.is_some());
369 let set = registry.current().await;
370 assert_eq!(set.epoch, 1);
371 assert_eq!(set.usable().count(), 1);
372 }
373
374 #[tokio::test]
375 async fn reconciling_an_unchanged_world_publishes_nothing() {
376 let provider = Fake::new(vec![decl("read", CapabilityKind::Tool)]);
377 let (registry, _rx) = CapabilityRegistry::new(provider);
378 registry.reconcile().await;
379 let first = registry.current().await.epoch;
380 assert!(registry.reconcile().await.is_none(), "idempotent");
381 assert!(!registry.has_pending_changes().await);
382 assert_eq!(registry.current().await.epoch, first, "no epoch churn");
383 }
384
385 /// What the MCP pool observes arrives as declared configuration, and a
386 /// change to it is a change to the world: a new epoch, picked up at the
387 /// next turn with no restart.
388 #[tokio::test]
389 async fn an_observed_catalog_publishes_a_new_epoch() {
390 let provider = Fake::new(vec![decl("search", CapabilityKind::McpServer)]);
391 let (registry, _rx) = CapabilityRegistry::new(provider.clone());
392 registry.reconcile().await;
393 let before = registry.current().await.epoch;
394
395 provider.declarations.lock().unwrap()[0].configuration =
396 serde_json::json!({"tools": [{"name": "query"}]});
397 assert!(registry.has_pending_changes().await);
398 assert!(registry.reconcile().await.is_some());
399 let set = registry.current().await;
400 assert!(set.epoch > before);
401 assert_eq!(
402 set.get(&CapabilityId::new(CapabilityKind::McpServer, "search"))
403 .unwrap()
404 .configuration,
405 serde_json::json!({"tools": [{"name": "query"}]})
406 );
407 }
408
409 #[tokio::test]
410 async fn a_capability_added_later_appears_without_a_restart() {
411 let provider = Fake::new(vec![decl("read", CapabilityKind::Tool)]);
412 let (registry, _rx) = CapabilityRegistry::new(provider.clone());
413 registry.reconcile().await;
414 let before = registry.current().await.epoch;
415
416 provider
417 .declarations
418 .lock()
419 .unwrap()
420 .push(decl("pdf", CapabilityKind::Skill));
421
422 let delta = registry.reconcile().await.expect("a change was published");
423 assert_eq!(delta.added.len(), 1);
424 assert!(registry.current().await.epoch > before);
425 assert_eq!(registry.current().await.usable().count(), 2);
426 }
427
428 #[tokio::test]
429 async fn removing_a_capability_at_source_removes_it_from_the_set() {
430 let provider = Fake::new(vec![
431 decl("read", CapabilityKind::Tool),
432 decl("pdf", CapabilityKind::Skill),
433 ]);
434 let (registry, _rx) = CapabilityRegistry::new(provider.clone());
435 registry.reconcile().await;
436 provider
437 .declarations
438 .lock()
439 .unwrap()
440 .retain(|d| d.id.name != "pdf");
441 let delta = registry.reconcile().await.expect("removal is a change");
442 assert_eq!(delta.removed.len(), 1);
443 assert!(delta.describe().contains("no longer available"));
444 }
445
446 #[tokio::test]
447 async fn revocation_is_immediate_and_independent_of_any_epoch() {
448 let provider = Fake::new(vec![decl("read", CapabilityKind::Tool)]);
449 let (registry, _rx) = CapabilityRegistry::new(provider);
450 registry.reconcile().await;
451 let id = CapabilityId::new(CapabilityKind::Tool, "read");
452 assert!(registry.revocation(&id).await.is_none());
453
454 registry.revoke(id.clone(), "operator disabled").await;
455 // Visible at dispatch immediately, with no reconcile in between.
456 assert_eq!(
457 registry.revocation(&id).await.as_deref(),
458 Some("operator disabled")
459 );
460 // And retired from the published set on the next pass.
461 registry.reconcile().await;
462 assert_eq!(registry.current().await.usable().count(), 0);
463 assert_eq!(
464 registry.current_blocking().unusable().count(),
465 1,
466 "the synchronous accessor reads the same published set"
467 );
468 }
469
470 #[tokio::test]
471 async fn restoring_a_revoked_capability_requires_a_fresh_published_epoch() {
472 let provider = Fake::new(vec![decl("read", CapabilityKind::Tool)]);
473 let (registry, _rx) = CapabilityRegistry::new(provider);
474 registry.reconcile().await;
475 let id = CapabilityId::new(CapabilityKind::Tool, "read");
476 let before = registry.current().await.epoch;
477
478 registry.revoke(id.clone(), "operator disabled").await;
479 registry.reconcile().await;
480 assert!(registry.revoked_now(&id));
481 assert_eq!(registry.current().await.usable().count(), 0);
482
483 registry.restore(&id).await;
484 // Restore only removes the immediate deny. It cannot resurrect a
485 // capability in a bound turn before reconciliation publishes it.
486 assert!(!registry.revoked_now(&id));
487 assert_eq!(registry.current().await.usable().count(), 0);
488 registry.reconcile().await;
489 assert!(registry.current().await.epoch > before);
490 assert_eq!(registry.current().await.usable().count(), 1);
491 }
492}
493

1//! The capability registry: one owner, one reconcile loop, one projection.
2//!
3//! # Why a loop and not a cache
4//!
5//! Everything this replaces was **edge-triggered**: discovery warmed once, a
6//! prompt froze once, a connection opened once, a catalog cached once — with
7//! a retry guard written so that caching a *failure* counted as having
8//! succeeded. A single missed edge was therefore permanent, and uptime is
9//! what converts a small race into a dead integration.
10//!
11//! This is **level-triggered**, the way long-running systems are built:
12//! `reconcile()` compares desired state to observed state and moves toward
13//! it, idempotently. Hints (a filesystem event, an MCP `list_changed`, a
14//! plugin toggle) only make it run *sooner*; the ticker guarantees it runs
15//! anyway. A dropped hint costs one tick of latency. It never costs
16//! correctness.
17//!
18//! # Two channels, deliberately asymmetric
19//!
20//! Additions and catalog changes take effect at the next **turn boundary**,
21//! via a published epoch. Revocations take effect **immediately**, mid-turn,
22//! fail-closed. That split follows a rule the codebase already believes:
23//! narrowing is always safe, widening needs admission. An operator disabling
24//! a compromised plugin must not wait for a long turn to finish, while a
25//! newly added skill appearing halfway through a plan would be a torn read.
26//!
27//! # Offline by construction
28//!
29//! A pass never talks to anything. Declarations are a filesystem walk at
30//! worst, and an MCP server's catalog and failure arrive as declared data
31//! from the on-demand pool (`vak_mcp::McpManager`), which is the only thing
32//! allowed to start a server. So reconciling is cheap enough to run at every
33//! turn admission, and it can never spawn an integration nobody asked for.
34
35use std::collections::{BTreeMap, BTreeSet};
36use std::sync::Arc;
37use std::sync::atomic::{AtomicU64, Ordering};
38use std::time::{Duration, SystemTime};
39
40use async_trait::async_trait;
41use serde::{Deserialize, Serialize};
42use tokio::sync::{RwLock, mpsc};
43
44use super::domain::Serves;
45use super::resolution::Resolution;
46use super::snapshot::{Capability, CapabilityDelta, CapabilityId, CapabilitySet, Epoch, Origin};
47
48/// How often the loop reconciles with no hint at all. The safety net that
49/// makes a missed event survivable.
50pub const RECONCILE_INTERVAL: Duration = Duration::from_secs(10);
51
52/// Hints are coalesced over this window, so an editor writing six files does
53/// one reconcile rather than six.
54pub const DEBOUNCE: Duration = Duration::from_millis(250);
55
56/// One capability as its source declares it.
57#[derive(Debug, Clone, PartialEq, Eq)]
58pub struct Declaration {
59 pub id: CapabilityId,
60 pub origin: Origin,
61 pub summary: String,
62 pub serves: Serves,
63 pub digest: Option<String>,
64 pub source: Option<std::path::PathBuf>,
65 /// Kind-specific detail, part of the published digest — for an MCP
66 /// server, what the pool has observed (catalog, last failure).
67 pub configuration: serde_json::Value,
68}
69
70/// Where declarations come from.
71///
72/// Implemented by `Core`, which already owns skill/hook/command/MCP
73/// discovery. Keeping it behind a trait means the loop is testable without a
74/// workspace, a filesystem, or a live MCP server.
75#[async_trait]
76pub trait CapabilityProvider: Send + Sync {
77 /// Everything currently declared. Cheap and offline: a filesystem walk
78 /// at worst, never a network call or a spawned process.
79 fn declare(&self) -> Vec<Declaration>;
80
81 /// Housekeeping on each pass, such as evicting idle pooled connections.
82 /// It may only ever release resources, never acquire them.
83 async fn upkeep(&self) {}
84}
85
86/// Why the loop woke up. Purely an optimisation — the ticker would have
87/// caught all of these eventually.
88#[derive(Debug, Clone, PartialEq, Eq)]
89pub enum Hint {
90 /// A watched capability directory changed.
91 SourceChanged(String),
92 /// The MCP pool learned something about a server (a catalog, a
93 /// failure), or the server announced a catalog change.
94 ServerObserved(String),
95 /// Configuration or plugin enablement moved.
96 ConfigChanged,
97 /// Someone asked for an immediate pass (boot, or a CLI about to run one
98 /// turn and exit).
99 Immediate,
100}
101
102/// The health of the loop itself.
103///
104/// Exposed so "capabilities look thin because reconciliation has been
105/// failing for two hours" is a visible fact rather than an unexplained
106/// shortage of tools.
107#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
108pub struct ReconcileStatus {
109 pub last_run: Option<SystemTime>,
110 pub last_change: Option<SystemTime>,
111 pub epoch: Epoch,
112 pub passes: u64,
113}
114
115/// The registry.
116pub struct CapabilityRegistry {
117 provider: Arc<dyn CapabilityProvider>,
118 current: RwLock<Arc<CapabilitySet>>,
119 next_epoch: AtomicU64,
120 /// The immediate channel. Checked at dispatch, independent of any epoch.
121 revoked: RwLock<BTreeMap<CapabilityId, String>>,
122 revoked_fast: std::sync::RwLock<BTreeMap<CapabilityId, String>>,
123 status: RwLock<ReconcileStatus>,
124 hints: mpsc::UnboundedSender<Hint>,
125}
126
127impl CapabilityRegistry {
128 pub fn new(
129 provider: Arc<dyn CapabilityProvider>,
130 ) -> (Arc<Self>, mpsc::UnboundedReceiver<Hint>) {
131 let (tx, rx) = mpsc::unbounded_channel();
132 let registry = Arc::new(CapabilityRegistry {
133 provider,
134 current: RwLock::new(Arc::new(CapabilitySet::empty())),
135 next_epoch: AtomicU64::new(1),
136 revoked: RwLock::new(BTreeMap::new()),
137 revoked_fast: std::sync::RwLock::new(BTreeMap::new()),
138 status: RwLock::new(ReconcileStatus::default()),
139 hints: tx,
140 });
141 (registry, rx)
142 }
143
144 /// Ask for a reconcile sooner than the ticker would. Never blocks and
145 /// never fails meaningfully: if the loop is gone, the hint is moot.
146 pub fn hint(&self, hint: Hint) {
147 let _ = self.hints.send(hint);
148 }
149
150 /// The current published set. What a turn binds at its start.
151 pub async fn current(&self) -> Arc<CapabilitySet> {
152 self.current.read().await.clone()
153 }
154
155 /// Best-effort synchronous read of the current published set, for
156 /// diagnostics/reporting call sites (`capability_diagnostics`, `/doctor`,
157 /// the system prompt) that are not themselves async. Reconciliation is
158 /// an eventually-consistent background loop by design — a rare
159 /// contended `try_read` just means this diagnostic reflects the
160 /// previous epoch for one more instant, not a correctness problem.
161 /// Falls back to an empty set only if the lock is actually contended
162 /// (never blocks), which a caller reporting "nothing degraded" during
163 /// that instant is a harmless, self-correcting understatement.
164 pub fn current_blocking(&self) -> Arc<CapabilitySet> {
165 self.current
166 .try_read()
167 .map(|guard| guard.clone())
168 .unwrap_or_else(|_| Arc::new(CapabilitySet::empty()))
169 }
170
171 pub async fn status(&self) -> ReconcileStatus {
172 self.status.read().await.clone()
173 }
174
175 /// Revoke immediately, mid-turn, fail-closed.
176 ///
177 /// Does not wait for an epoch: a compromised plugin must stop being
178 /// callable now, not when the current turn happens to finish. This only
179 /// ever narrows, which is why it is safe to apply without admission.
180 pub async fn revoke(&self, id: CapabilityId, reason: impl Into<String>) {
181 let reason = reason.into();
182 self.revoked
183 .write()
184 .await
185 .insert(id.clone(), reason.clone());
186 if let Ok(mut revoked) = self.revoked_fast.write() {
187 revoked.insert(id, reason);
188 }
189 self.hint(Hint::ConfigChanged);
190 }
191
192 pub async fn restore(&self, id: &CapabilityId) {
193 self.revoked.write().await.remove(id);
194 if let Ok(mut revoked) = self.revoked_fast.write() {
195 revoked.remove(id);
196 }
197 self.hint(Hint::ConfigChanged);
198 }
199
200 pub fn revoked_now(&self, id: &CapabilityId) -> bool {
201 self.revoked_fast
202 .read()
203 .ok()
204 .is_some_and(|revoked| revoked.contains_key(id))
205 }
206
207 /// Whether the declared world differs from the published set, so turn
208 /// admission knows a pass is worth running before it binds an epoch.
209 pub async fn has_pending_changes(&self) -> bool {
210 let declarations = self.provider.declare();
211 let current = self.current.read().await;
212 declarations.len() != current.all().count()
213 || declarations.iter().any(|d| match current.get(&d.id) {
214 None => true,
215 Some(existing) => {
216 existing.digest != d.digest || existing.configuration != d.configuration
217 }
218 })
219 }
220
221 /// Whether `id` is revoked right now, regardless of the caller's epoch.
222 /// Dispatch checks this; availability comes from the bound epoch, but
223 /// authorization always comes from the present.
224 pub async fn revocation(&self, id: &CapabilityId) -> Option<String> {
225 self.revoked.read().await.get(id).cloned()
226 }
227
228 pub async fn revoked_ids(&self) -> BTreeSet<CapabilityId> {
229 self.revoked.read().await.keys().cloned().collect()
230 }
231
232 /// One idempotent pass. Safe to call at any time, from anywhere, as
233 /// often as you like — that is the whole point of a level-triggered
234 /// design. Returns the delta if a new epoch was published.
235 pub async fn reconcile(&self) -> Option<CapabilityDelta> {
236 let now = SystemTime::now();
237 let previous = self.current.read().await.clone();
238 let declarations = self.provider.declare();
239
240 // A revoked capability is retired in the published set as well as
241 // blocked at dispatch, so the prompt stops advertising it at the next
242 // turn instead of describing a tool that will always refuse.
243 let revoked = self.revoked.read().await.clone();
244 let capabilities: Vec<Capability> = declarations
245 .into_iter()
246 .map(|declaration| Capability {
247 resolution: match revoked.get(&declaration.id) {
248 Some(reason) => Resolution::Retired {
249 reason: reason.clone(),
250 },
251 None => Resolution::Available,
252 },
253 id: declaration.id,
254 origin: declaration.origin,
255 summary: declaration.summary,
256 serves: declaration.serves,
257 digest: declaration.digest,
258 source: declaration.source,
259 configuration: declaration.configuration,
260 })
261 .collect();
262
263 let candidate = CapabilitySet::new(previous.epoch, capabilities);
264
265 let mut status = self.status.write().await;
266 status.passes += 1;
267 status.last_run = Some(now);
268
269 if candidate.digest == previous.digest {
270 status.epoch = previous.epoch;
271 return None;
272 }
273
274 // Publish. Only here does the epoch move, and only because the world
275 // actually differs.
276 let epoch = self.next_epoch.fetch_add(1, Ordering::Relaxed);
277 let published = Arc::new(CapabilitySet::new(
278 epoch,
279 candidate.all().cloned().collect(),
280 ));
281 let delta = published.delta_from(&previous);
282 *self.current.write().await = published;
283 status.epoch = epoch;
284 status.last_change = Some(now);
285 Some(delta)
286 }
287
288 /// Run the loop until `shutdown` fires.
289 ///
290 /// Selects over the ticker and the hint channel; a burst of hints is
291 /// coalesced into one pass. The ticker is what makes a dropped hint a
292 /// latency problem rather than a correctness one.
293 pub async fn run(
294 self: Arc<Self>,
295 mut hints: mpsc::UnboundedReceiver<Hint>,
296 mut shutdown: tokio::sync::watch::Receiver<bool>,
297 ) {
298 let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
299 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
300 loop {
301 tokio::select! {
302 _ = ticker.tick() => {}
303 hint = hints.recv() => {
304 if hint.is_none() {
305 return;
306 }
307 // Coalesce a burst: an editor saving six files should
308 // produce one pass, not six.
309 tokio::time::sleep(DEBOUNCE).await;
310 while hints.try_recv().is_ok() {}
311 }
312 _ = shutdown.changed() => {
313 if *shutdown.borrow() {
314 return;
315 }
316 continue;
317 }
318 }
319 self.reconcile().await;
320 self.provider.upkeep().await;
321 }
322 }
323}
324
325#[cfg(test)]
326#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
327mod tests {
328 use super::*;
329 use std::sync::Mutex;
330 use vak_session::types::CapabilityKind;
331
332 /// A provider whose declarations the test drives.
333 struct Fake {
334 declarations: Mutex<Vec<Declaration>>,
335 }
336
337 impl Fake {
338 fn new(declarations: Vec<Declaration>) -> Arc<Self> {
339 Arc::new(Fake {
340 declarations: Mutex::new(declarations),
341 })
342 }
343 }
344
345 #[async_trait]
346 impl CapabilityProvider for Fake {
347 fn declare(&self) -> Vec<Declaration> {
348 self.declarations.lock().unwrap().clone()
349 }
350 }
351
352 fn decl(name: &str, kind: CapabilityKind) -> Declaration {
353 Declaration {
354 id: CapabilityId::new(kind, name),
355 origin: Origin::Workspace,
356 summary: String::new(),
357 serves: Serves::Undeclared,
358 digest: None,
359 source: None,
360 configuration: serde_json::Value::Null,
361 }
362 }
363
364 #[tokio::test]
365 async fn a_first_pass_publishes_an_epoch() {
366 let provider = Fake::new(vec![decl("read", CapabilityKind::Tool)]);
367 let (registry, _rx) = CapabilityRegistry::new(provider);
368 assert!(registry.reconcile().await.is_some());
369 let set = registry.current().await;
370 assert_eq!(set.epoch, 1);
371 assert_eq!(set.usable().count(), 1);
372 }
373
374 #[tokio::test]
375 async fn reconciling_an_unchanged_world_publishes_nothing() {
376 let provider = Fake::new(vec![decl("read", CapabilityKind::Tool)]);
377 let (registry, _rx) = CapabilityRegistry::new(provider);
378 registry.reconcile().await;
379 let first = registry.current().await.epoch;
380 assert!(registry.reconcile().await.is_none(), "idempotent");
381 assert!(!registry.has_pending_changes().await);
382 assert_eq!(registry.current().await.epoch, first, "no epoch churn");
383 }
384
385 /// What the MCP pool observes arrives as declared configuration, and a
386 /// change to it is a change to the world: a new epoch, picked up at the
387 /// next turn with no restart.
388 #[tokio::test]
389 async fn an_observed_catalog_publishes_a_new_epoch() {
390 let provider = Fake::new(vec![decl("search", CapabilityKind::McpServer)]);
391 let (registry, _rx) = CapabilityRegistry::new(provider.clone());
392 registry.reconcile().await;
393 let before = registry.current().await.epoch;
394
395 provider.declarations.lock().unwrap()[0].configuration =
396 serde_json::json!({"tools": [{"name": "query"}]});
397 assert!(registry.has_pending_changes().await);
398 assert!(registry.reconcile().await.is_some());
399 let set = registry.current().await;
400 assert!(set.epoch > before);
401 assert_eq!(
402 set.get(&CapabilityId::new(CapabilityKind::McpServer, "search"))
403 .unwrap()
404 .configuration,
405 serde_json::json!({"tools": [{"name": "query"}]})
406 );
407 }
408
409 #[tokio::test]
410 async fn a_capability_added_later_appears_without_a_restart() {
411 let provider = Fake::new(vec![decl("read", CapabilityKind::Tool)]);
412 let (registry, _rx) = CapabilityRegistry::new(provider.clone());
413 registry.reconcile().await;
414 let before = registry.current().await.epoch;
415
416 provider
417 .declarations
418 .lock()
419 .unwrap()
420 .push(decl("pdf", CapabilityKind::Skill));
421
422 let delta = registry.reconcile().await.expect("a change was published");
423 assert_eq!(delta.added.len(), 1);
424 assert!(registry.current().await.epoch > before);
425 assert_eq!(registry.current().await.usable().count(), 2);
426 }
427
428 #[tokio::test]
429 async fn removing_a_capability_at_source_removes_it_from_the_set() {
430 let provider = Fake::new(vec![
431 decl("read", CapabilityKind::Tool),
432 decl("pdf", CapabilityKind::Skill),
433 ]);
434 let (registry, _rx) = CapabilityRegistry::new(provider.clone());
435 registry.reconcile().await;
436 provider
437 .declarations
438 .lock()
439 .unwrap()
440 .retain(|d| d.id.name != "pdf");
441 let delta = registry.reconcile().await.expect("removal is a change");
442 assert_eq!(delta.removed.len(), 1);
443 assert!(delta.describe().contains("no longer available"));
444 }
445
446 #[tokio::test]
447 async fn revocation_is_immediate_and_independent_of_any_epoch() {
448 let provider = Fake::new(vec![decl("read", CapabilityKind::Tool)]);
449 let (registry, _rx) = CapabilityRegistry::new(provider);
450 registry.reconcile().await;
451 let id = CapabilityId::new(CapabilityKind::Tool, "read");
452 assert!(registry.revocation(&id).await.is_none());
453
454 registry.revoke(id.clone(), "operator disabled").await;
455 // Visible at dispatch immediately, with no reconcile in between.
456 assert_eq!(
457 registry.revocation(&id).await.as_deref(),
458 Some("operator disabled")
459 );
460 // And retired from the published set on the next pass.
461 registry.reconcile().await;
462 assert_eq!(registry.current().await.usable().count(), 0);
463 assert_eq!(
464 registry.current_blocking().unusable().count(),
465 1,
466 "the synchronous accessor reads the same published set"
467 );
468 }
469
470 #[tokio::test]
471 async fn restoring_a_revoked_capability_requires_a_fresh_published_epoch() {
472 let provider = Fake::new(vec![decl("read", CapabilityKind::Tool)]);
473 let (registry, _rx) = CapabilityRegistry::new(provider);
474 registry.reconcile().await;
475 let id = CapabilityId::new(CapabilityKind::Tool, "read");
476 let before = registry.current().await.epoch;
477
478 registry.revoke(id.clone(), "operator disabled").await;
479 registry.reconcile().await;
480 assert!(registry.revoked_now(&id));
481 assert_eq!(registry.current().await.usable().count(), 0);
482
483 registry.restore(&id).await;
484 // Restore only removes the immediate deny. It cannot resurrect a
485 // capability in a bound turn before reconciliation publishes it.
486 assert!(!registry.revoked_now(&id));
487 assert_eq!(registry.current().await.usable().count(), 0);
488 registry.reconcile().await;
489 assert!(registry.current().await.epoch > before);
490 assert_eq!(registry.current().await.usable().count(), 1);
491 }
492}
493

Indexing the workspace…