fix(security): pin unpinned GitHub Actions + fix clippy errors

Code scanning alerts:
- sigstore/cosign-installer@v3 → @398d4b0e (CodeQL #79)
- dtolnay/rust-toolchain@stable → @29eef336 (CodeQL #78)

Clippy fixes:
- wire_api.rs: u64 → u128 parse for CacheValidator mtime_ns
- cache_tiers.rs, cache_coordinator.rs: Duration::from_secs(60) → from_mins(1)
- Regenerate config-keys.md, cargo fmt applied

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Yves Gugger
2026-07-31 12:20:24 +02:00
parent d31495efec
commit 7fdc8a94c6
40 changed files with 3348 additions and 197 deletions
+1 -1
View File
@@ -41,7 +41,7 @@ jobs:
- uses: actions/checkout@v4
- name: Install Rust toolchain
uses: dtolnay/rust-toolchain@stable
uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
- name: Cache cargo
uses: actions/cache@v4
+1 -1
View File
@@ -212,7 +212,7 @@ jobs:
run: sha256sum lean-ctx-*.tar.gz lean-ctx-*.zip > SHA256SUMS
- name: Sign release artifacts with cosign
uses: sigstore/cosign-installer@v3
uses: sigstore/cosign-installer@398d4b0eeef1380460a10c8013a76f728fb906ac # v3
- name: Cosign sign SHA256SUMS
run: |
cosign sign-blob --yes \
+8
View File
@@ -0,0 +1,8 @@
# Deprecated configuration keys and features.
# These still work but will be removed in a future major version.
[[deprecated]]
key = "cloud.contribute_enabled"
since = "3.9.14"
replacement = "telemetry.enabled"
notes = "Automatically migrated on first load. The unified telemetry flag controls both heartbeat and compression pattern sharing."
+4 -6
View File
@@ -85,8 +85,7 @@ In addition, roles can restrict **unsafe I/O**:
**Optional network activity (fully disableable):**
- **Update check**: a lightweight daily GET to `leanctx.com/version.txt` to notify you of new versions. Sends only the current version as User-Agent. Disable with `update_check_disabled = true` in `~/.lean-ctx/config.toml` or `LEAN_CTX_NO_UPDATE_CHECK=1`.
- **Anonymous stats sharing** (opt-in, off by default): if you enable `contribute_enabled` in setup, anonymized compression statistics (token counts, compression ratios — no file names, no code, no PII) are periodically sent to `api.leanctx.com`.
- **Telemetry heartbeat** (opt-in, off by default): if you enable `[telemetry] enabled = true` (via setup or `lean-ctx telemetry on`), a daily heartbeat sends only: lean-ctx version, OS, CPU architecture, and a random installation UUID. No code, no filenames, no personal data. Inspect the exact payload: `lean-ctx telemetry show`. Regenerate the installation ID: `lean-ctx telemetry reset-id`.
- **Anonymous telemetry** (opt-in, off by default): if you enable `[telemetry] enabled = true` (via setup or `lean-ctx telemetry on`), a daily heartbeat sends: lean-ctx version, OS, CPU architecture, a random installation UUID, and anonymized compression patterns (file-type, size bucket, mode, ratio). No code, no filenames, no personal data. Inspect the exact payload: `lean-ctx telemetry show`. Regenerate the installation ID: `lean-ctx telemetry reset-id`.
**Does NOT:**
- Collect tracking analytics, fingerprints, or PII
@@ -304,8 +303,8 @@ policy — per-session I/O limits live on the active role, which is selected via
update_check_disabled = true # no daily update check
path_jail = true # keep the filesystem jail on (default)
[cloud]
contribute_enabled = false # no anonymous stats sharing (default)
[telemetry]
enabled = false # no anonymous telemetry (default)
```
**2. A locked-down role — `~/.lean-ctx/roles/bank.toml`:**
@@ -338,8 +337,7 @@ export LEAN_CTX_ROLE=bank
| Endpoint | Purpose | Disable |
|----------|---------|---------|
| `leanctx.com/version.txt` | Update check (daily GET) | `update_check_disabled = true` |
| `api.leanctx.com` | Opt-in anonymous stats | `contribute_enabled = false` (default) |
| `api.leanctx.com` | Opt-in telemetry heartbeat (version, OS, arch, random install ID) | `[telemetry] enabled = false` (default) |
| `api.leanctx.com` | Opt-in anonymous telemetry (version, OS, arch, install ID, compression patterns) | `[telemetry] enabled = false` (default) |
| `huggingface.co` | Embedding model download | Pre-provision models, set `LEAN_CTX_EMBEDDING_MODEL_DIR` |
| `localhost:PORT` | Dashboard (local TCP) | Don't start dashboard, or bind to loopback only |
| UDS socket | Daemon IPC | Permissions `0o600`, owner-only access |
-1
View File
@@ -160,7 +160,6 @@ Cross-project boundary and access control policies
Cloud feature settings
- `auto_sync` (bool, default `false`) — Push the Personal Cloud (knowledge, commands, CEP, gotchas, buddy, feedback) silently once per day at session end (Pro; toggle: `lean-ctx cloud autosync on|off`)
- `contribute_enabled` (bool, default `false`) — Enable contributing anonymized stats to lean-ctx cloud
## `[context]`
+5 -5
View File
@@ -2,10 +2,10 @@ use async_trait::async_trait;
use crate::types::{
AgentEnvelope, CompressionRequest, CompressionResult, ConfigProposal, ConfigTuningRequest,
ConnectorJob, DeliveryEntry, DeliveryRecord, DeliveryStats, EfficiencyAnalysis,
EfficiencySample, ExperimentRequest, ExperimentResult, IntentDecision, IntentRequest,
MessagePriority, MetricPoint, ModelRouteRequest, Observation, OclaCapability, OclaResult,
Outcome, PrivacyLevel, ResponseOptimizationRequest, ResponseOptimizationResult,
ConnectorJob, DeliveryEntry, DeliveryRecord, DeliveryRecordResult, DeliveryStats,
EfficiencyAnalysis, EfficiencySample, ExperimentRequest, ExperimentResult, IntentDecision,
IntentRequest, MessagePriority, MetricPoint, ModelRouteRequest, Observation, OclaCapability,
OclaResult, Outcome, PrivacyLevel, ResponseOptimizationRequest, ResponseOptimizationResult,
RoutingDecision, SavingsEvidence, ScheduledJob, UsageRecord,
};
@@ -98,7 +98,7 @@ pub trait DeliveryRegistry: OclaService {
requester_conversation_id: Option<&str>,
) -> Option<DeliveryRecord>;
fn record_stub_served(&self, record: &DeliveryRecord, stub_tokens: u64);
fn record_delivery(&self, entry: DeliveryEntry);
fn record_delivery(&self, entry: DeliveryEntry) -> DeliveryRecordResult;
fn delivery_stats(&self) -> DeliveryStats;
}
+9
View File
@@ -739,6 +739,15 @@ pub struct DeliveryEntry {
pub mtime: u64,
}
/// Result of an idempotent delivery-record attempt.
#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
pub struct DeliveryRecordResult {
/// The existing record already represented the same source version.
pub already_recorded: bool,
/// An existing record was refreshed because its source version changed.
pub updated: bool,
}
/// Statistics for the delivery registry.
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct DeliveryStats {
+8 -3
View File
@@ -55,9 +55,12 @@ fn show_status() {
fn set_enabled(enabled: bool) {
match config::setter::set_by_key("telemetry.enabled", if enabled { "true" } else { "false" }) {
Ok(_) => {
// Clear legacy contribute_enabled — telemetry.enabled is now the single flag.
let _ = config::setter::set_by_key("cloud.contribute_enabled", "false");
if enabled {
println!("Telemetry enabled — thank you for helping improve lean-ctx!");
println!("Sent daily: version, OS, arch, random install ID. No code, no PII.");
println!("Sent daily: version, OS, arch, compression patterns, random install ID.");
println!("No code, no file names, no personal data — ever.");
println!("\x1b[2mDisable anytime: lean-ctx telemetry off\x1b[0m");
} else {
println!("Telemetry disabled. No data will be sent.");
@@ -89,11 +92,13 @@ fn reset_id() {
fn show_payload() {
let id = installation_id::get_or_create().unwrap_or_else(|_| "<error>".to_string());
let contribute = crate::cloud_sync::collect_contribute_entries();
let payload = serde_json::json!({
"installation_id": id,
"version": env!("CARGO_PKG_VERSION"),
"os": std::env::consts::OS,
"arch": std::env::consts::ARCH,
"contribute_entries": contribute,
});
println!("This is the exact JSON that would be sent to api.leanctx.com:");
@@ -154,6 +159,6 @@ fn print_help() {
println!(" reset-id Regenerate the anonymous installation ID");
println!(" history Show log of all sent heartbeats");
println!();
println!("The heartbeat sends only: version, OS, architecture, and a random");
println!("installation UUID. No code, filenames, or personal data — ever.");
println!("The heartbeat sends: version, OS, architecture, compression patterns,");
println!("and a random install UUID. No code, filenames, or personal data — ever.");
}
+3 -13
View File
@@ -66,11 +66,6 @@ pub fn cloud_background_tasks() {
.last_heartbeat
.as_deref()
.is_some_and(|d| d == today);
let already_contributed = config
.cloud
.last_contribute
.as_deref()
.is_some_and(|d| d == today);
let already_synced = config
.cloud
.last_sync
@@ -87,14 +82,16 @@ pub fn cloud_background_tasks() {
.as_deref()
.is_some_and(|d| d == today);
// Anonymous telemetry heartbeat (opt-in, no auth required).
// Unified anonymous telemetry: heartbeat + contribute entries in one request.
if config.telemetry.enabled && !already_heartbeated {
if let Ok(id) = crate::core::installation_id::get_or_create() {
let contribute = collect_contribute_entries();
let payload = serde_json::json!({
"installation_id": id,
"version": env!("CARGO_PKG_VERSION"),
"os": std::env::consts::OS,
"arch": std::env::consts::ARCH,
"contribute_entries": contribute,
});
if crate::cloud_client::heartbeat(&payload).is_ok() {
config.telemetry.last_heartbeat = Some(today.clone());
@@ -110,13 +107,6 @@ pub fn cloud_background_tasks() {
}
}
if config.cloud.contribute_enabled && !already_contributed {
let entries = collect_contribute_entries();
if !entries.is_empty() && crate::cloud_client::contribute(&entries).is_ok() {
config.cloud.last_contribute = Some(today.clone());
}
}
if crate::cloud_client::is_logged_in() {
if config.cloud.sync_stats_enabled && !already_synced {
let store = crate::core::stats::load();
+1
View File
@@ -73,6 +73,7 @@ impl Default for Config {
conversation: ConversationConfig::default(),
response_shaping: ResponseShapingConfig::default(),
ocla: OclaConfig::default(),
cache: CacheConfig::default(),
agents: AgentsConfig::default(),
proxy_enabled: None,
proxy_port: None,
+38
View File
@@ -387,6 +387,8 @@ impl Config {
cfg.merge_local(local, trusted);
}
cfg.migrate_contribute_to_telemetry();
let cfg = Arc::new(cfg);
if let Ok(mut guard) = CACHE.lock() {
*guard = Some((Arc::clone(&cfg), global_hash, local_hash, selected_profile));
@@ -397,6 +399,42 @@ impl Config {
// `merge_local` is in `merge.rs` (extracted for #660 LOC gate).
/// Migrate legacy `[cloud] contribute_enabled` → `[telemetry] enabled`.
///
/// If the user opted into the old anonymous contribute system but has not
/// yet enabled the new unified telemetry flag, flip `telemetry.enabled`
/// on and clear `contribute_enabled` so the migration is one-way.
/// Persists the change to disk so subsequent loads see the new state.
pub(crate) fn migrate_contribute_to_telemetry(&mut self) {
if self.cloud.contribute_enabled && !self.telemetry.enabled {
self.telemetry.enabled = true;
self.cloud.contribute_enabled = false;
if let Some(path) = Self::path() {
if let Ok(raw) = std::fs::read_to_string(&path) {
let mut updated =
raw.replace("contribute_enabled = true", "contribute_enabled = false");
if !updated.contains("[telemetry]") {
if !updated.ends_with('\n') {
updated.push('\n');
}
updated.push_str("\n[telemetry]\nenabled = true\n");
} else if let Some(tpos) = updated.find("[telemetry]") {
let after = &updated[tpos..];
if let Some(epos) = after.find("enabled = false") {
let abs_pos = tpos + epos;
updated.replace_range(
abs_pos..abs_pos + "enabled = false".len(),
"enabled = true",
);
}
}
let _ = crate::config_io::write_atomic_with_backup(&path, &updated);
}
}
}
}
/// Loads ONLY the global config file — never merging project-local
/// `.lean-ctx.toml` overrides, and bypassing the in-memory cache. Every
/// PERSIST path must use this (or [`Config::update_global`]): [`Config::load`]
+3
View File
@@ -66,6 +66,9 @@ pub struct Config {
pub response_shaping: ResponseShapingConfig,
#[serde(default)]
pub ocla: OclaConfig,
/// Generalized L1/L2/L3 cache settings (`[cache]`).
#[serde(default)]
pub cache: CacheConfig,
#[serde(default)]
pub agents: sections::AgentsConfig,
/// Whether the API proxy is enabled. Tri-state:
@@ -754,14 +754,6 @@ pub(super) fn build(sections: &mut BTreeMap<String, SectionSchema>) {
);
let mut cloud = BTreeMap::new();
cloud.insert(
"contribute_enabled".into(),
key(
"bool",
serde_json::json!(cfg.cloud.contribute_enabled),
"Enable contributing anonymized stats to lean-ctx cloud",
),
);
cloud.insert(
"auto_sync".into(),
key(
+57
View File
@@ -26,16 +26,23 @@ pub struct OclaConfig {
#[serde(default)]
pub struct DeliveryConfig {
pub enabled: bool,
/// Allow subagents to receive a cross-agent delivery stub instead of
/// forcing a fresh disk read.
pub delivery_for_subagents: bool,
pub max_entries: usize,
pub ttl_minutes: u64,
/// Generalized cache tier settings.
pub cache: CacheConfig,
}
impl Default for DeliveryConfig {
fn default() -> Self {
Self {
enabled: true,
delivery_for_subagents: true,
max_entries: 4096,
ttl_minutes: 30,
cache: CacheConfig::default(),
}
}
}
@@ -46,6 +53,56 @@ impl OclaConfig {
}
}
/// Bounds and feature switches for the generalized cross-agent cache.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct CacheConfig {
pub l1_max_entries: usize,
pub l1_ttl_secs: u64,
pub l2_max_entries: usize,
pub l2_ttl_secs: u64,
pub l3_max_bytes: u64,
pub l3_gc_threshold: f64,
pub shell_cache_enabled: bool,
pub compose_cache_enabled: bool,
}
impl Default for CacheConfig {
fn default() -> Self {
Self {
l1_max_entries: 1_000,
l1_ttl_secs: 300,
l2_max_entries: 10_000,
l2_ttl_secs: 3_600,
l3_max_bytes: 500_000_000,
l3_gc_threshold: 0.9,
shell_cache_enabled: false,
compose_cache_enabled: true,
}
}
}
#[cfg(test)]
mod cache_config_tests {
use super::CacheConfig;
#[test]
fn cache_defaults_match_delivery_budget() {
assert_eq!(CacheConfig::default().l3_max_bytes, 500_000_000);
assert!(!CacheConfig::default().shell_cache_enabled);
assert!(CacheConfig::default().compose_cache_enabled);
}
#[test]
fn cache_config_deserializes_partial_overrides() {
let parsed: CacheConfig =
serde_json::from_str(r#"{"l1_max_entries": 12, "shell_cache_enabled": true}"#).unwrap();
assert_eq!(parsed.l1_max_entries, 12);
assert!(parsed.shell_cache_enabled);
assert_eq!(parsed.l2_ttl_secs, 3_600);
}
}
/// Agent lifecycle configuration: TTLs, GC intervals, scratchpad limits.
///
/// Maps to `[agents]` in config.toml. All fields have sane defaults so existing
+147 -24
View File
@@ -7,9 +7,10 @@
//! Storage: in-process DashMap keyed by blake3[..12]. The daemon wire_api
//! endpoints expose this store for cross-process coordination via IPC.
use std::collections::HashSet;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use dashmap::DashMap;
@@ -25,8 +26,37 @@ struct DeliveryKey {
path: String,
}
#[derive(Default)]
struct EvictionIndex {
by_time: BTreeMap<Instant, DeliveryKey>,
by_key: HashMap<DeliveryKey, Instant>,
}
impl EvictionIndex {
fn insert(&mut self, key: DeliveryKey) {
let mut timestamp = Instant::now();
while self.by_time.contains_key(&timestamp) {
timestamp = timestamp
.checked_add(Duration::from_nanos(1))
.expect("delivery eviction timestamp overflow");
}
self.by_time.insert(timestamp, key.clone());
self.by_key.insert(key, timestamp);
}
fn remove(&mut self, key: &DeliveryKey) {
if let Some(timestamp) = self.by_key.remove(key) {
self.by_time.remove(&timestamp);
}
}
}
pub struct BuiltinDeliveryRegistry {
store: DashMap<DeliveryKey, DeliveryRecord>,
eviction_index: Mutex<EvictionIndex>,
/// Fast-rejection index: path → [(mtime, blake3_prefix)].
/// Allows `stat()`-only rejection (no file read+hash) on ~99% of misses.
mtime_index: DashMap<String, Vec<(u64, [u8; 12])>>,
stubs_served: AtomicU64,
tokens_saved: AtomicU64,
max_entries: usize,
@@ -47,7 +77,9 @@ impl BuiltinDeliveryRegistry {
pub fn with_config(max_entries: usize, ttl_minutes: u64) -> Self {
Self {
store: DashMap::with_capacity(max_entries.clamp(1, 256)),
store: DashMap::with_capacity(max_entries),
mtime_index: DashMap::new(),
eviction_index: Mutex::new(EvictionIndex::default()),
stubs_served: AtomicU64::new(0),
tokens_saved: AtomicU64::new(0),
max_entries: max_entries.max(1),
@@ -59,6 +91,8 @@ impl BuiltinDeliveryRegistry {
fn with_limits(max_entries: usize, ttl_secs: u64) -> Self {
Self {
store: DashMap::with_capacity(256),
mtime_index: DashMap::new(),
eviction_index: Mutex::new(EvictionIndex::default()),
stubs_served: AtomicU64::new(0),
tokens_saved: AtomicU64::new(0),
max_entries,
@@ -80,27 +114,63 @@ impl BuiltinDeliveryRegistry {
now.saturating_sub(record.read_at) > self.ttl_secs
}
#[cfg(test)]
fn purge_expired(&self) {
let now = Self::now_epoch();
self.store
.retain(|_, record| !self.is_expired_at(record, now));
let mut index = self.eviction_index.lock().expect("delivery index poisoned");
self.purge_expired_locked(&mut index);
}
fn evict_oldest_if_full(&self) {
self.purge_expired();
if self.store.len() < self.max_entries {
return;
fn purge_expired_locked(&self, index: &mut EvictionIndex) {
let now = Self::now_epoch();
let expired: Vec<_> = self
.store
.iter()
.filter(|entry| self.is_expired_at(entry.value(), now))
.map(|entry| entry.key().clone())
.collect();
for key in expired {
if let Some((_, record)) = self.store.remove(&key) {
self.mtime_index_remove(&key.path, record.mtime, key.blake3);
}
index.remove(&key);
}
}
fn evict_oldest_if_full_locked(&self, index: &mut EvictionIndex) {
self.purge_expired_locked(index);
while self.store.len() >= self.max_entries {
let oldest = self
.store
.iter()
.min_by_key(|entry| entry.value().read_at)
.map(|entry| entry.key().clone());
let Some(key) = oldest else {
let Some((_, key)) = index.by_time.pop_first() else {
break;
};
self.store.remove(&key);
index.by_key.remove(&key);
if let Some((_, record)) = self.store.remove(&key) {
self.mtime_index_remove(&key.path, record.mtime, key.blake3);
}
}
}
/// O(1) fast-rejection: does ANY delivery record exist for this path+mtime?
/// Avoids full blake3 file-read+hash when no record can possibly match.
pub fn has_candidate(&self, path: &str, mtime: u64) -> bool {
self.mtime_index
.get(path)
.is_some_and(|entries| entries.iter().any(|(m, _)| *m == mtime))
}
fn mtime_index_insert(&self, path: &str, mtime: u64, blake3: [u8; 12]) {
let mut entries = self.mtime_index.entry(path.to_string()).or_default();
if !entries.iter().any(|(m, h)| *m == mtime && *h == blake3) {
entries.push((mtime, blake3));
}
}
fn mtime_index_remove(&self, path: &str, mtime: u64, blake3: [u8; 12]) {
if let Some(mut entries) = self.mtime_index.get_mut(path) {
entries.retain(|(m, h)| !(*m == mtime && *h == blake3));
if entries.is_empty() {
drop(entries);
self.mtime_index.remove(path);
}
}
}
@@ -155,7 +225,10 @@ impl DeliveryRegistry for BuiltinDeliveryRegistry {
continue;
}
if self.is_expired_at(&record, now) {
let mut index = self.eviction_index.lock().expect("delivery index poisoned");
self.mtime_index_remove(&key.path, record.mtime, key.blake3);
self.store.remove(&key);
index.remove(&key);
continue;
}
if requester_agent_id.is_some_and(|agent| agent == record.agent_id) {
@@ -182,20 +255,22 @@ impl DeliveryRegistry for BuiltinDeliveryRegistry {
path: record.path.clone(),
tokens_saved: estimated_tokens,
serving_agent: record.agent_id.clone(),
original_agent: record.conversation_id.clone(),
original_agent: record.agent_id.clone(),
});
}
fn record_delivery(&self, entry: DeliveryEntry) {
fn record_delivery(&self, entry: DeliveryEntry) -> lean_ctx_ocla::DeliveryRecordResult {
if !Self::is_valid_entry(&entry) {
return;
return lean_ctx_ocla::DeliveryRecordResult {
already_recorded: false,
updated: false,
};
}
self.purge_expired();
self.evict_oldest_if_full();
let key = DeliveryKey {
blake3: entry.blake3,
path: entry.path.clone(),
};
let record_mtime = entry.mtime;
let record = DeliveryRecord {
blake3: entry.blake3,
path: entry.path,
@@ -204,10 +279,30 @@ impl DeliveryRegistry for BuiltinDeliveryRegistry {
agent_id: entry.agent_id,
conversation_id: entry.conversation_id,
read_at: Self::now_epoch(),
mtime: entry.mtime,
mtime: record_mtime,
fresh: true,
};
self.store.insert(key, record);
let mut index = self.eviction_index.lock().expect("delivery index poisoned");
let (already_existed, mtime_changed) = {
let existing = self.store.get(&key).map(|e| e.mtime);
match existing {
Some(old_mtime) => (true, old_mtime != record.mtime),
None => (false, false),
}
};
if already_existed {
self.store.insert(key.clone(), record);
index.remove(&key);
} else {
self.evict_oldest_if_full_locked(&mut index);
self.store.insert(key.clone(), record);
}
index.insert(key.clone());
self.mtime_index_insert(&key.path, record_mtime, key.blake3);
lean_ctx_ocla::DeliveryRecordResult {
already_recorded: already_existed && !mtime_changed,
updated: mtime_changed,
}
}
fn delivery_stats(&self) -> DeliveryStats {
@@ -369,6 +464,33 @@ mod tests {
assert!(reg.store.len() <= 3);
}
#[test]
fn eviction_index_removes_the_oldest_entry() {
let reg = BuiltinDeliveryRegistry::with_limits(2, 3600);
let oldest = [30u8; 12];
let middle = [31u8; 12];
let newest = [32u8; 12];
reg.record_delivery(test_entry("oldest.rs", "agent-a", oldest, 100));
reg.record_delivery(test_entry("middle.rs", "agent-a", middle, 100));
reg.record_delivery(test_entry("newest.rs", "agent-a", newest, 100));
assert_eq!(reg.store.len(), 2);
assert_eq!(reg.eviction_index.lock().unwrap().by_time.len(), 2);
assert!(
reg.check_delivery(&oldest, 100, "oldest.rs", Some("agent-b"), None)
.is_none()
);
assert!(
reg.check_delivery(&middle, 100, "middle.rs", Some("agent-b"), None)
.is_some()
);
assert!(
reg.check_delivery(&newest, 100, "newest.rs", Some("agent-b"), None)
.is_some()
);
}
#[test]
fn expired_entry_returns_miss() {
let reg = BuiltinDeliveryRegistry::with_config(8, 1);
@@ -449,5 +571,6 @@ mod tests {
reg.purge_expired();
assert_eq!(reg.store.len(), 0);
assert!(reg.eviction_index.lock().unwrap().by_time.is_empty());
}
}
+297
View File
@@ -0,0 +1,297 @@
//! Coordinator for the L1-to-L3 generalized delivery cache lookup chain.
use std::sync::OnceLock;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
use super::cache_tiers::{L1ProcessCache, L2DaemonCache, L3DiskCache};
use super::cache_types::{CacheKey, CacheValidator, DeliveryEntryV2, DeliveryStatsV2};
static GLOBAL_CACHE: OnceLock<BuiltinCacheCoordinator> = OnceLock::new();
/// Returns the process-global cache coordinator, lazily initialized with
/// default tier sizes.
pub fn materialized_cache() -> &'static BuiltinCacheCoordinator {
GLOBAL_CACHE.get_or_init(|| {
let config = crate::core::config::Config::load();
let cache_cfg = &config.ocla.delivery.cache;
let l3 = L3DiskCache::open(
crate::core::data_dir::lean_ctx_data_dir()
.unwrap_or_else(|_| std::path::PathBuf::from("/tmp/lean-ctx-cache")),
)
.unwrap_or_else(|_| {
L3DiskCache::open("/tmp/lean-ctx-cache-fallback")
.expect("fallback L3 cache must initialize")
});
l3.startup_validate(
std::time::Duration::from_secs(cache_cfg.l1_ttl_secs.saturating_mul(6)),
cache_cfg.l3_max_bytes,
);
BuiltinCacheCoordinator::new(
L1ProcessCache::new(std::time::Duration::from_secs(cache_cfg.l1_ttl_secs)),
L2DaemonCache::new(cache_cfg.l2_max_entries, std::time::Duration::from_hours(1)),
l3,
)
})
}
/// Coordinates lookups and writes across the generalized delivery cache tiers.
pub trait CacheCoordinator {
/// Checks every tier for a fresh entry matching `key` and `validator`.
fn check(&self, key: &CacheKey, validator: &CacheValidator) -> Option<DeliveryEntryV2>;
/// Records a newly materialized entry in every cache tier.
fn record(&self, entry: DeliveryEntryV2);
/// Returns a snapshot of cumulative cache activity.
fn stats(&self) -> DeliveryStatsV2;
/// Checks multiple keys in input order.
fn batch_check(&self, requests: &[(CacheKey, CacheValidator)]) -> Vec<Option<DeliveryEntryV2>> {
requests
.iter()
.map(|(key, validator)| self.check(key, validator))
.collect()
}
}
#[derive(Debug, Default)]
struct CacheCounters {
l1_hits: AtomicU64,
l2_hits: AtomicU64,
l3_hits: AtomicU64,
misses: AtomicU64,
materializations: AtomicU64,
references_served: AtomicU64,
tokens_saved: AtomicU64,
evictions: AtomicU64,
expired: AtomicU64,
}
impl CacheCounters {
fn snapshot(&self) -> DeliveryStatsV2 {
DeliveryStatsV2 {
l1_hits: self.l1_hits.load(Ordering::Relaxed),
l2_hits: self.l2_hits.load(Ordering::Relaxed),
l3_hits: self.l3_hits.load(Ordering::Relaxed),
misses: self.misses.load(Ordering::Relaxed),
materializations: self.materializations.load(Ordering::Relaxed),
references_served: self.references_served.load(Ordering::Relaxed),
tokens_saved: self.tokens_saved.load(Ordering::Relaxed),
evictions: self.evictions.load(Ordering::Relaxed),
expired: self.expired.load(Ordering::Relaxed),
}
}
}
/// Built-in coordinator that promotes L3 hits to L2 and L2 hits to L1.
#[derive(Debug)]
pub struct BuiltinCacheCoordinator {
l1: L1ProcessCache,
l2: L2DaemonCache,
l3: L3DiskCache,
counters: CacheCounters,
}
impl BuiltinCacheCoordinator {
/// Creates a coordinator from independently configured cache tiers.
pub fn new(l1: L1ProcessCache, l2: L2DaemonCache, l3: L3DiskCache) -> Self {
Self {
l1,
l2,
l3,
counters: CacheCounters::default(),
}
}
/// Returns the process-local cache tier.
pub fn l1(&self) -> &L1ProcessCache {
&self.l1
}
/// Returns the daemon-shared cache tier.
pub fn l2(&self) -> &L2DaemonCache {
&self.l2
}
/// Returns the disk-backed cache tier.
pub fn l3(&self) -> &L3DiskCache {
&self.l3
}
fn serve(&self, entry: DeliveryEntryV2) -> DeliveryEntryV2 {
self.counters
.references_served
.fetch_add(1, Ordering::Relaxed);
self.counters
.tokens_saved
.fetch_add(entry.token_count, Ordering::Relaxed);
entry
}
fn valid(
&self,
entry: DeliveryEntryV2,
key: &CacheKey,
validator: &CacheValidator,
) -> Option<DeliveryEntryV2> {
if entry.is_fresh_for(key, validator, epoch_ms()) {
Some(entry)
} else {
self.counters.expired.fetch_add(1, Ordering::Relaxed);
None
}
}
}
impl CacheCoordinator for BuiltinCacheCoordinator {
fn check(&self, key: &CacheKey, validator: &CacheValidator) -> Option<DeliveryEntryV2> {
if let Some(entry) = self.l1.get(key) {
if let Some(entry) = self.valid(entry, key, validator) {
self.counters.l1_hits.fetch_add(1, Ordering::Relaxed);
return Some(self.serve(entry));
}
self.l1.remove(key);
}
if let Some(entry) = self.l2.get(key) {
if let Some(entry) = self.valid(entry, key, validator) {
self.counters.l2_hits.fetch_add(1, Ordering::Relaxed);
self.l1.insert(entry.clone());
return Some(self.serve(entry));
}
self.l2.remove(key);
}
if let Some(entry) = self.l3.get(key) {
if let Some(entry) = self.valid(entry, key, validator) {
self.counters.l3_hits.fetch_add(1, Ordering::Relaxed);
let _ = self.l2.insert(entry.clone());
self.l1.insert(entry.clone());
return Some(self.serve(entry));
}
let _ = self.l3.remove(key);
}
self.counters.misses.fetch_add(1, Ordering::Relaxed);
None
}
fn record(&self, entry: DeliveryEntryV2) {
self.counters
.materializations
.fetch_add(1, Ordering::Relaxed);
self.l1.insert(entry.clone());
if self.l2.insert(entry.clone()).is_some() {
self.counters.evictions.fetch_add(1, Ordering::Relaxed);
}
if self.l3.insert(entry).is_err() {
self.counters.evictions.fetch_add(1, Ordering::Relaxed);
}
}
fn stats(&self) -> DeliveryStatsV2 {
self.counters.snapshot()
}
}
fn epoch_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use super::*;
use crate::core::ocla::cache_types::{
AgentHost, CacheIdentity, CacheValidator, ContentHandleRef, DeliveryKind,
};
fn entry(name: &str) -> DeliveryEntryV2 {
DeliveryEntryV2 {
schema_version: 2,
key: CacheKey(format!("cache:v1:file_read:{name}")),
kind: DeliveryKind::FileRead,
validator: CacheValidator::Immutable,
handle: ContentHandleRef {
algorithm: "blake3".into(),
digest: "e".repeat(64),
byte_len: 1,
media_type: "text/plain".into(),
},
display_path: None,
line_count: None,
token_count: 5,
producer: CacheIdentity {
agent_id: "agent".into(),
conversation_id: "conversation".into(),
host: AgentHost::Codex,
},
created_at_epoch_ms: 0,
expires_at_epoch_ms: u64::MAX,
}
}
fn coordinator() -> (BuiltinCacheCoordinator, tempfile::TempDir) {
let dir = tempfile::tempdir().expect("tmpdir");
let coord = BuiltinCacheCoordinator::new(
L1ProcessCache::new(Duration::from_mins(1)),
L2DaemonCache::new(8, Duration::from_mins(1)),
L3DiskCache::open(dir.path()).expect("L3 open"),
);
(coord, dir)
}
#[test]
fn coordinator_records_and_reads_from_l1() {
let (coordinator, _dir) = coordinator();
let entry = entry("l1_check");
coordinator.record(entry.clone());
assert_eq!(
coordinator.check(&entry.key, &entry.validator),
Some(entry.clone())
);
let stats = coordinator.stats();
assert_eq!(stats.l1_hits, 1);
}
#[test]
fn l2_hit_promotes_entry_to_l1() {
let (coordinator, _dir) = coordinator();
let entry = entry("l2_promote");
coordinator.l2().insert(entry.clone());
assert_eq!(
coordinator.check(&entry.key, &entry.validator),
Some(entry.clone())
);
assert!(coordinator.l1().get(&entry.key).is_some());
assert_eq!(coordinator.stats().l2_hits, 1);
}
#[test]
fn l3_hit_promotes_entry_to_both_memory_tiers() {
let (coordinator, _dir) = coordinator();
let entry = entry("l3");
coordinator.l3().insert(entry.clone()).unwrap();
assert_eq!(
coordinator.check(&entry.key, &entry.validator),
Some(entry.clone())
);
assert!(coordinator.l1().get(&entry.key).is_some());
assert!(coordinator.l2().get(&entry.key).is_some());
assert_eq!(coordinator.stats().l3_hits, 1);
}
#[test]
fn batch_check_keeps_request_order() {
let (coordinator, _dir) = coordinator();
let hit = entry("hit");
let miss = entry("miss");
coordinator.record(hit.clone());
let results = coordinator.batch_check(&[
(hit.key.clone(), hit.validator.clone()),
(miss.key.clone(), miss.validator.clone()),
]);
assert_eq!(results, vec![Some(hit), None]);
}
}
+155
View File
@@ -0,0 +1,155 @@
//! Shared adapter helpers for generalized cross-agent delivery caching.
use std::sync::OnceLock;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use super::cache_coordinator::{BuiltinCacheCoordinator, CacheCoordinator};
use super::cache_tiers::{L1ProcessCache, L2DaemonCache, L3DiskCache};
use super::cache_types::{
AgentHost, CacheIdentity, CacheKey, CacheValidator, ContentHandleRef, DeliveryEntryV2,
DeliveryKind,
};
static COORDINATOR: OnceLock<Option<BuiltinCacheCoordinator>> = OnceLock::new();
/// Returns the process-wide generalized delivery coordinator when caching is enabled.
pub fn coordinator() -> Option<&'static BuiltinCacheCoordinator> {
COORDINATOR
.get_or_init(|| {
let config = crate::core::config::Config::load();
if !config.ocla.delivery_enabled() {
return None;
}
let root = crate::core::paths::cache_dir().ok()?.join("delivery-v2");
let ttl = Duration::from_secs(config.ocla.delivery.ttl_minutes.saturating_mul(60));
let l3 = L3DiskCache::open(root).ok()?;
Some(BuiltinCacheCoordinator::new(
L1ProcessCache::new(ttl),
L2DaemonCache::new(config.ocla.delivery.max_entries, ttl),
l3,
))
})
.as_ref()
}
/// Looks up an adapter result across all tiers including cross-process daemon.
pub fn check(key: &CacheKey, validator: &CacheValidator, adapter: &str) -> Option<DeliveryEntryV2> {
let coordinator = coordinator()?;
// L1 + local L2 + L3 (in-process)
if let Some(entry) = coordinator.check(key, validator) {
emit_stats(coordinator, adapter);
return Some(entry);
}
// Cross-process: ask daemon (other processes may have recorded this)
let agent = agent_id();
let conv_id =
crate::core::conversation::current_conversation_id().unwrap_or_else(|| agent.clone());
if let Some(entry) =
crate::daemon_client::try_cache_check_blocking(key, validator, Some(&agent), Some(&conv_id))
{
// Promote to L1 so subsequent calls skip IPC
coordinator.record(entry.clone());
emit_stats(coordinator, adapter);
return Some(entry);
}
emit_stats(coordinator, adapter);
None
}
/// Records an adapter result and emits the coordinator snapshot for observability.
pub fn record(
key: CacheKey,
kind: DeliveryKind,
validator: CacheValidator,
display_path: Option<String>,
content: &str,
adapter: &str,
) {
let Some(coordinator) = coordinator() else {
return;
};
let now = epoch_ms();
let ttl_ms = crate::core::config::Config::load()
.ocla
.delivery
.ttl_minutes
.saturating_mul(60_000);
let digest = blake3::hash(content.as_bytes()).to_hex().to_string();
let agent_id = agent_id();
let entry = DeliveryEntryV2 {
schema_version: 2,
key,
kind,
validator,
handle: ContentHandleRef {
algorithm: "blake3".into(),
digest,
byte_len: content.len() as u64,
media_type: "text/plain".into(),
},
display_path,
line_count: Some(content.lines().count() as u32),
token_count: crate::core::tokens::count_tokens(content) as u64,
producer: CacheIdentity {
conversation_id: agent_id.clone(),
agent_id,
host: agent_host(),
},
created_at_epoch_ms: now,
expires_at_epoch_ms: now.saturating_add(ttl_ms),
};
coordinator.record(entry.clone());
crate::daemon_client::try_cache_record_blocking(&entry);
emit_stats(coordinator, adapter);
}
/// Renders a deterministic reference in place of an already materialized result.
pub fn stub(entry: &DeliveryEntryV2, label: &str) -> String {
let path = entry.display_path.as_deref().unwrap_or("result");
format!(
"{path} [cross-agent cache · {label} · produced by {} · {} tokens avoided]",
entry.producer.agent_id, entry.token_count
)
}
fn emit_stats(coordinator: &BuiltinCacheCoordinator, adapter: &str) {
let stats = coordinator.stats();
tracing::debug!(
target: "lean_ctx::cache_delivery",
adapter,
l1_hits = stats.l1_hits,
l2_hits = stats.l2_hits,
l3_hits = stats.l3_hits,
misses = stats.misses,
materializations = stats.materializations,
references_served = stats.references_served,
tokens_saved = stats.tokens_saved,
"cache coordinator stats"
);
}
fn epoch_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64
}
fn agent_id() -> String {
std::env::var("CURSOR_TASK_ID")
.or_else(|_| std::env::var("CLAUDECODE"))
.or_else(|_| std::env::var("CODEX_THREAD_ID"))
.unwrap_or_else(|_| "local-agent".into())
}
fn agent_host() -> AgentHost {
if std::env::var_os("CURSOR_TASK_ID").is_some() {
AgentHost::Cursor
} else if std::env::var_os("CLAUDECODE").is_some() {
AgentHost::ClaudeCode
} else if std::env::var_os("CODEX_THREAD_ID").is_some() {
AgentHost::Codex
} else {
AgentHost::Cli
}
}
+414
View File
@@ -0,0 +1,414 @@
//! Process, daemon, and disk cache tiers for generalized delivery entries.
use std::collections::BTreeMap;
use std::fs;
use std::io;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use dashmap::DashMap;
use serde::{Deserialize, Serialize};
use super::cache_types::{CacheKey, DeliveryEntryV2};
#[derive(Clone, Debug)]
struct MemoryCacheEntry {
entry: DeliveryEntryV2,
expires_at: Instant,
}
/// Process-local cache backed by a concurrent map and a fixed TTL.
#[derive(Debug)]
pub struct L1ProcessCache {
entries: DashMap<CacheKey, MemoryCacheEntry>,
ttl: Duration,
}
impl L1ProcessCache {
/// Creates an empty process-local cache with the supplied TTL.
pub fn new(ttl: Duration) -> Self {
Self {
entries: DashMap::new(),
ttl,
}
}
/// Returns a live entry and drops expired local entries.
pub fn get(&self, key: &CacheKey) -> Option<DeliveryEntryV2> {
let value = self.entries.get(key)?;
if value.expires_at > Instant::now() {
return Some(value.entry.clone());
}
drop(value);
self.entries.remove(key);
None
}
/// Inserts or replaces an entry using this tier's TTL.
pub fn insert(&self, entry: DeliveryEntryV2) {
let key = entry.key.clone();
self.entries.insert(
key,
MemoryCacheEntry {
entry,
expires_at: Instant::now() + self.ttl,
},
);
}
/// Removes an entry by key.
pub fn remove(&self, key: &CacheKey) -> Option<DeliveryEntryV2> {
self.entries.remove(key).map(|(_, value)| value.entry)
}
/// Returns the number of entries currently retained by this tier.
pub fn len(&self) -> usize {
self.entries.len()
}
/// Returns whether this tier has no entries.
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
}
/// Daemon-shared cache backed by a concurrent map and ordered eviction index.
#[derive(Debug)]
pub struct L2DaemonCache {
entries: DashMap<CacheKey, MemoryCacheEntry>,
eviction_index: Mutex<BTreeMap<Instant, CacheKey>>,
max_entries: usize,
ttl: Duration,
}
impl L2DaemonCache {
/// Creates an empty daemon cache, clamping its capacity to at least one entry.
pub fn new(max_entries: usize, ttl: Duration) -> Self {
Self {
entries: DashMap::new(),
eviction_index: Mutex::new(BTreeMap::new()),
max_entries: max_entries.max(1),
ttl,
}
}
/// Returns a live entry and refreshes its position in the eviction index.
pub fn get(&self, key: &CacheKey) -> Option<DeliveryEntryV2> {
let entry = self.entries.get(key)?.clone();
if entry.expires_at <= Instant::now() {
self.remove(key);
return None;
}
self.touch(key.clone());
Some(entry.entry)
}
/// Inserts or replaces an entry, evicting the least-recently-used entry when full.
pub fn insert(&self, entry: DeliveryEntryV2) -> Option<DeliveryEntryV2> {
let key = entry.key.clone();
let mut index = self
.eviction_index
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
remove_index_key(&mut index, &key);
let previous = self
.entries
.insert(
key.clone(),
MemoryCacheEntry {
entry,
expires_at: Instant::now() + self.ttl,
},
)
.map(|value| value.entry);
let evicted = if previous.is_none() && self.entries.len() > self.max_entries {
index.pop_first().and_then(|(_, evicted_key)| {
self.entries
.remove(&evicted_key)
.map(|(_, value)| value.entry)
})
} else {
None
};
let timestamp = unique_instant(&index, Instant::now());
index.insert(timestamp, key);
evicted
}
/// Removes an entry and its eviction-index record.
pub fn remove(&self, key: &CacheKey) -> Option<DeliveryEntryV2> {
let mut index = self
.eviction_index
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
remove_index_key(&mut index, key);
self.entries.remove(key).map(|(_, value)| value.entry)
}
/// Returns the number of entries currently retained by this tier.
pub fn len(&self) -> usize {
self.entries.len()
}
/// Returns whether this tier has no entries.
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
fn touch(&self, key: CacheKey) {
let mut index = self
.eviction_index
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
remove_index_key(&mut index, &key);
let timestamp = unique_instant(&index, Instant::now());
index.insert(timestamp, key);
}
}
fn remove_index_key(index: &mut BTreeMap<Instant, CacheKey>, key: &CacheKey) {
index.retain(|_, indexed_key| indexed_key != key);
}
fn unique_instant(index: &BTreeMap<Instant, CacheKey>, mut timestamp: Instant) -> Instant {
while index.contains_key(&timestamp) {
timestamp = timestamp
.checked_add(Duration::from_nanos(1))
.unwrap_or(timestamp);
}
timestamp
}
/// Serializable manifest record for an entry retained in the disk cache.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct PersistedCacheEntry {
/// The delivery entry represented by this manifest record.
pub entry: DeliveryEntryV2,
/// Time at which the entry was persisted, in Unix epoch milliseconds.
pub persisted_at_epoch_ms: u64,
}
/// Disk-backed cache manifest with a directory reserved for content blobs.
#[derive(Debug)]
pub struct L3DiskCache {
root: PathBuf,
manifest: DashMap<CacheKey, PersistedCacheEntry>,
blob_directory: PathBuf,
}
impl L3DiskCache {
/// Opens a disk cache rooted at `root`, creating its manifest and blob directories as needed.
pub fn open(root: impl AsRef<Path>) -> io::Result<Self> {
let root = root.as_ref().to_path_buf();
let blob_directory = root.join("blobs");
fs::create_dir_all(&blob_directory)?;
let manifest_path = root.join("manifest.json");
let entries = if manifest_path.exists() {
let bytes = fs::read(&manifest_path)?;
serde_json::from_slice::<Vec<PersistedCacheEntry>>(&bytes).map_err(io::Error::other)?
} else {
Vec::new()
};
let manifest = DashMap::new();
for persisted in entries {
manifest.insert(persisted.entry.key.clone(), persisted);
}
Ok(Self {
root,
manifest,
blob_directory,
})
}
/// Validates the manifest on startup: removes entries older than `max_age`
/// and trims the manifest to `max_bytes` total token budget.
pub fn startup_validate(&self, max_age: Duration, max_bytes: u64) {
let now_ms = epoch_ms();
let max_age_ms = max_age.as_millis() as u64;
let expired: Vec<CacheKey> = self
.manifest
.iter()
.filter(|entry| now_ms.saturating_sub(entry.persisted_at_epoch_ms) > max_age_ms)
.map(|entry| entry.key().clone())
.collect();
for key in &expired {
self.manifest.remove(key);
}
if max_bytes > 0 {
let mut entries: Vec<_> = self
.manifest
.iter()
.map(|e| {
(
e.key().clone(),
e.persisted_at_epoch_ms,
e.entry.token_count,
)
})
.collect();
entries.sort_by_key(|(_, ts, _)| *ts);
let mut total: u64 = entries.iter().map(|(_, _, t)| *t).sum();
for (key, _, tokens) in &entries {
if total <= max_bytes {
break;
}
total -= tokens;
self.manifest.remove(key);
}
}
if !expired.is_empty() {
let _ = self.persist_manifest();
}
}
/// Returns an entry from the persisted manifest without loading its blob.
pub fn get(&self, key: &CacheKey) -> Option<DeliveryEntryV2> {
self.manifest
.get(key)
.map(|persisted| persisted.entry.clone())
}
/// Persists an entry to the manifest and returns any entry it replaced.
pub fn insert(&self, entry: DeliveryEntryV2) -> io::Result<Option<DeliveryEntryV2>> {
let key = entry.key.clone();
let persisted = PersistedCacheEntry {
entry,
persisted_at_epoch_ms: epoch_ms(),
};
let previous = self.manifest.insert(key, persisted).map(|old| old.entry);
self.persist_manifest()?;
Ok(previous)
}
/// Removes an entry from the manifest and persists the updated manifest.
pub fn remove(&self, key: &CacheKey) -> io::Result<Option<DeliveryEntryV2>> {
let removed = self
.manifest
.remove(key)
.map(|(_, persisted)| persisted.entry);
if removed.is_some() {
self.persist_manifest()?;
}
Ok(removed)
}
/// Returns the cache root directory.
pub fn root(&self) -> &Path {
&self.root
}
/// Returns the directory reserved for content-addressed blobs.
pub fn blob_directory(&self) -> &Path {
&self.blob_directory
}
/// Returns the number of entries present in the manifest.
pub fn len(&self) -> usize {
self.manifest.len()
}
/// Returns whether the manifest contains no entries.
pub fn is_empty(&self) -> bool {
self.manifest.is_empty()
}
fn persist_manifest(&self) -> io::Result<()> {
let mut records = self
.manifest
.iter()
.map(|item| item.value().clone())
.collect::<Vec<_>>();
records.sort_by(|left, right| left.entry.key.cmp(&right.entry.key));
let bytes = serde_json::to_vec(&records).map_err(io::Error::other)?;
let temporary = self.root.join("manifest.json.tmp");
fs::write(&temporary, bytes)?;
fs::rename(temporary, self.root.join("manifest.json"))
}
}
fn epoch_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64
}
#[cfg(test)]
mod tests {
use super::*;
use crate::core::ocla::cache_types::{
AgentHost, CacheIdentity, CacheValidator, ContentHandleRef, DeliveryKind,
};
fn entry(name: &str) -> DeliveryEntryV2 {
DeliveryEntryV2 {
schema_version: 2,
key: CacheKey(format!("cache:v1:file_read:{name}")),
kind: DeliveryKind::FileRead,
validator: CacheValidator::Immutable,
handle: ContentHandleRef {
algorithm: "blake3".into(),
digest: "d".repeat(64),
byte_len: 1,
media_type: "text/plain".into(),
},
display_path: None,
line_count: None,
token_count: 4,
producer: CacheIdentity {
agent_id: "agent".into(),
conversation_id: "conversation".into(),
host: AgentHost::Cli,
},
created_at_epoch_ms: 0,
expires_at_epoch_ms: u64::MAX,
}
}
#[test]
fn l1_expires_entries_using_its_ttl() {
let cache = L1ProcessCache::new(Duration::ZERO);
let entry = entry("l1");
cache.insert(entry.clone());
assert_eq!(cache.get(&entry.key), None);
assert!(cache.is_empty());
}
#[test]
fn l2_evicts_the_oldest_entry_at_capacity() {
let cache = L2DaemonCache::new(1, Duration::from_mins(1));
let first = entry("first");
let second = entry("second");
cache.insert(first.clone());
assert_eq!(cache.insert(second.clone()), Some(first));
assert_eq!(cache.get(&second.key), Some(second));
}
#[test]
fn l3_serializes_manifest_entries() {
let directory = tempfile::tempdir().unwrap();
let cache = L3DiskCache::open(directory.path()).unwrap();
let entry = entry("l3");
cache.insert(entry.clone()).unwrap();
drop(cache);
let reopened = L3DiskCache::open(directory.path()).unwrap();
assert_eq!(reopened.get(&entry.key), Some(entry));
assert!(reopened.blob_directory().is_dir());
}
#[test]
fn persisted_entry_round_trips() {
let persisted = PersistedCacheEntry {
entry: entry("persisted"),
persisted_at_epoch_ms: 3,
};
assert_eq!(
serde_json::from_str::<PersistedCacheEntry>(
&serde_json::to_string(&persisted).unwrap()
)
.unwrap(),
persisted
);
}
}
+571
View File
@@ -0,0 +1,571 @@
//! Versioned identities and metadata for generalized delivery caching.
use serde::{Deserialize, Serialize};
/// The operation whose materialized result is represented by a cache entry.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DeliveryKind {
/// Reads a single file.
FileRead,
/// Runs a shell command.
ShellCommand,
/// Searches a project index.
SearchQuery,
/// Walks a directory tree.
DirectoryWalk,
/// Produces context assembled from multiple sources.
ComposedContext,
}
impl DeliveryKind {
/// Returns the stable lowercase name used in versioned cache keys.
pub const fn as_str(self) -> &'static str {
match self {
Self::FileRead => "file_read",
Self::ShellCommand => "shell_command",
Self::SearchQuery => "search_query",
Self::DirectoryWalk => "directory_walk",
Self::ComposedContext => "composed_context",
}
}
}
/// A versioned, content-derived cache lookup key.
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)]
#[serde(transparent)]
pub struct CacheKey(pub String);
impl CacheKey {
/// Builds a `cache:v1:{kind}:{blake3_hex}` key from canonical input.
pub fn from_canonical(kind: DeliveryKind, canonical_input: &str) -> Self {
let digest = blake3::hash(canonical_input.as_bytes()).to_hex();
Self(format!("cache:v1:{}:{digest}", kind.as_str()))
}
/// Returns the versioned key as a string slice.
pub fn as_str(&self) -> &str {
&self.0
}
}
/// Source state used to determine whether a cache entry remains fresh.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CacheValidator {
/// File freshness is tied to its modification time in nanoseconds.
File { mtime_ns: u128 },
/// Directory freshness is tied to its modification time in nanoseconds.
Directory { mtime_ns: u128 },
/// Input is immutable for the lifetime of its cache entry.
Immutable,
}
impl CacheValidator {
/// Returns whether two validators represent the same source state.
pub fn matches(&self, current: &Self) -> bool {
self == current
}
}
/// The application hosting an agent identity.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AgentHost {
/// Cursor editor integration.
Cursor,
/// Codex integration.
Codex,
/// Claude Code integration.
ClaudeCode,
/// lean-ctx command-line integration.
Cli,
/// An unrecognized or unavailable host.
Unknown,
}
/// Identifies an agent independently from its conversation and host.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct CacheIdentity {
/// Stable identifier of the producing agent.
pub agent_id: String,
/// Stable identifier of the producing conversation.
pub conversation_id: String,
/// Host that supplied the agent identity.
pub host: AgentHost,
}
/// Reference to immutable content in content-addressed storage.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct ContentHandleRef {
/// Digest algorithm, currently `blake3`.
pub algorithm: String,
/// Full hexadecimal digest of the materialized content.
pub digest: String,
/// Number of bytes in the materialized content.
pub byte_len: u64,
/// IANA media type of the materialized content.
pub media_type: String,
}
/// Versioned metadata for a generalized delivery-cache entry.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct DeliveryEntryV2 {
/// Schema version for compatibility checks during persistence.
pub schema_version: u16,
/// Versioned key used for cache lookup.
pub key: CacheKey,
/// Type of operation that produced this materialization.
pub kind: DeliveryKind,
/// Source-state evidence required for a fresh hit.
pub validator: CacheValidator,
/// Content-addressed reference to the materialized result.
pub handle: ContentHandleRef,
/// Human-readable source path when the operation has one.
pub display_path: Option<String>,
/// Line count for textual materializations.
pub line_count: Option<u32>,
/// Measured token count of the materialized result.
pub token_count: u64,
/// Agent and conversation that produced the result.
pub producer: CacheIdentity,
/// Creation time in Unix epoch milliseconds.
pub created_at_epoch_ms: u64,
/// Expiration time in Unix epoch milliseconds.
pub expires_at_epoch_ms: u64,
}
impl DeliveryEntryV2 {
/// Returns whether this entry is expired at the supplied Unix epoch time.
pub const fn is_expired_at(&self, now_epoch_ms: u64) -> bool {
self.expires_at_epoch_ms <= now_epoch_ms
}
/// Returns whether the supplied key and validator can use this entry.
pub fn is_fresh_for(
&self,
key: &CacheKey,
validator: &CacheValidator,
now_epoch_ms: u64,
) -> bool {
&self.key == key && self.validator.matches(validator) && !self.is_expired_at(now_epoch_ms)
}
}
/// Aggregate activity counters for the three cache tiers.
#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
pub struct DeliveryStatsV2 {
/// Hits served by the process-local cache.
pub l1_hits: u64,
/// Hits served by the daemon cache.
pub l2_hits: u64,
/// Hits served by the disk-backed cache.
pub l3_hits: u64,
/// Lookups not satisfied by any tier.
pub misses: u64,
/// New cache entries recorded after materialization.
pub materializations: u64,
/// Cache references returned to callers.
pub references_served: u64,
/// Tokens avoided by serving references instead of rematerializing.
pub tokens_saved: u64,
/// Entries removed because a bounded tier reached capacity.
pub evictions: u64,
/// Entries rejected because they expired or were stale.
pub expired: u64,
}
/// Produces a deterministic cache key and freshness validator for one operation.
pub trait CacheKeyBuilder {
/// Returns the kind of operation represented by this builder.
fn kind(&self) -> DeliveryKind;
/// Returns the deterministic input used to derive the cache key.
fn canonical_input(&self) -> String;
/// Returns the source-state validator recorded with the entry.
fn validator(&self) -> CacheValidator;
/// Derives the versioned cache key for this operation.
fn cache_key(&self) -> CacheKey {
CacheKey::from_canonical(self.kind(), &self.canonical_input())
}
}
fn canonical(fields: &[(&str, String)]) -> String {
fields
.iter()
.fold(String::new(), |mut output, (name, value)| {
output.push_str(name);
output.push(':');
output.push_str(&value.len().to_string());
output.push(':');
output.push_str(value);
output.push(';');
output
})
}
/// Cache-key inputs for a file read.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct FileReadKey {
/// Canonical path of the read file.
pub path: String,
/// File modification time in nanoseconds.
pub mtime_ns: u128,
/// Read mode.
pub mode: String,
/// Context reduction policy mode.
pub crp_mode: String,
/// Digest of the active task.
pub task_digest: String,
/// Revision of the applied policy.
pub policy_rev: String,
}
impl CacheKeyBuilder for FileReadKey {
fn kind(&self) -> DeliveryKind {
DeliveryKind::FileRead
}
fn canonical_input(&self) -> String {
canonical(&[
("path", self.path.clone()),
("mtime_ns", self.mtime_ns.to_string()),
("mode", self.mode.clone()),
("crp_mode", self.crp_mode.clone()),
("task_digest", self.task_digest.clone()),
("policy_rev", self.policy_rev.clone()),
])
}
fn validator(&self) -> CacheValidator {
CacheValidator::File {
mtime_ns: self.mtime_ns,
}
}
}
/// Cache-key inputs for a normalized shell command.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct ShellCommandKey {
/// Normalized command text.
pub command_normalized: String,
/// Canonical working directory.
pub cwd: String,
/// Digest of environment variables visible to the command.
pub env_hash: String,
}
impl CacheKeyBuilder for ShellCommandKey {
fn kind(&self) -> DeliveryKind {
DeliveryKind::ShellCommand
}
fn canonical_input(&self) -> String {
canonical(&[
("command_normalized", self.command_normalized.clone()),
("cwd", self.cwd.clone()),
("env_hash", self.env_hash.clone()),
])
}
fn validator(&self) -> CacheValidator {
CacheValidator::Immutable
}
}
/// Cache-key inputs for a search query.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct SearchQueryKey {
/// Search pattern.
pub pattern: String,
/// Include glob, or an empty string when no include filter is applied.
pub include: String,
/// Exclude glob, or an empty string when no exclude filter is applied.
pub exclude: String,
/// Search root path.
pub path: String,
/// Revision of the search index.
pub index_rev: String,
}
impl CacheKeyBuilder for SearchQueryKey {
fn kind(&self) -> DeliveryKind {
DeliveryKind::SearchQuery
}
fn canonical_input(&self) -> String {
canonical(&[
("pattern", self.pattern.clone()),
("include", self.include.clone()),
("exclude", self.exclude.clone()),
("path", self.path.clone()),
("index_rev", self.index_rev.clone()),
])
}
fn validator(&self) -> CacheValidator {
CacheValidator::Immutable
}
}
/// Cache-key inputs for a directory walk.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct DirectoryWalkKey {
/// Canonical root directory.
pub path: String,
/// Maximum traversal depth.
pub depth: usize,
/// Whether gitignore rules were applied.
pub gitignore: bool,
/// Directory modification time in nanoseconds.
pub dir_mtime_ns: u128,
}
impl CacheKeyBuilder for DirectoryWalkKey {
fn kind(&self) -> DeliveryKind {
DeliveryKind::DirectoryWalk
}
fn canonical_input(&self) -> String {
canonical(&[
("path", self.path.clone()),
("depth", self.depth.to_string()),
("gitignore", self.gitignore.to_string()),
("dir_mtime_ns", self.dir_mtime_ns.to_string()),
])
}
fn validator(&self) -> CacheValidator {
CacheValidator::Directory {
mtime_ns: self.dir_mtime_ns,
}
}
}
/// Cache-key inputs for composed context.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct ComposedContextKey {
/// Task requesting the context.
pub task: String,
/// Path scope for the composition, or an empty string for a global scope.
pub path: String,
/// Digests of source material in deterministic order.
pub source_digests: Vec<String>,
}
impl CacheKeyBuilder for ComposedContextKey {
fn kind(&self) -> DeliveryKind {
DeliveryKind::ComposedContext
}
fn canonical_input(&self) -> String {
let source_digests =
self.source_digests
.iter()
.fold(String::new(), |mut output, digest| {
output.push_str(&digest.len().to_string());
output.push(':');
output.push_str(digest);
output.push(';');
output
});
canonical(&[
("task", self.task.clone()),
("path", self.path.clone()),
("source_digests", source_digests),
])
}
fn validator(&self) -> CacheValidator {
CacheValidator::Immutable
}
}
#[cfg(test)]
mod tests {
use super::*;
fn file_key() -> FileReadKey {
FileReadKey {
path: "/repo/a.rs".into(),
mtime_ns: 42,
mode: "full".into(),
crp_mode: "tdd".into(),
task_digest: "task".into(),
policy_rev: "v1".into(),
}
}
fn entry() -> DeliveryEntryV2 {
let key = file_key();
DeliveryEntryV2 {
schema_version: 2,
key: key.cache_key(),
kind: key.kind(),
validator: key.validator(),
handle: ContentHandleRef {
algorithm: "blake3".into(),
digest: "a".repeat(64),
byte_len: 7,
media_type: "text/plain".into(),
},
display_path: Some(key.path),
line_count: Some(1),
token_count: 2,
producer: CacheIdentity {
agent_id: "agent".into(),
conversation_id: "conversation".into(),
host: AgentHost::Codex,
},
created_at_epoch_ms: 10,
expires_at_epoch_ms: 20,
}
}
#[test]
fn delivery_kind_serializes_in_snake_case() {
assert_eq!(
serde_json::to_string(&DeliveryKind::FileRead).unwrap(),
"\"file_read\""
);
}
#[test]
fn cache_key_uses_the_versioned_blake3_format() {
let key = CacheKey::from_canonical(DeliveryKind::FileRead, "input");
assert!(key.as_str().starts_with("cache:v1:file_read:"));
assert_eq!(key.as_str().len(), "cache:v1:file_read:".len() + 64);
}
#[test]
fn validator_compares_source_state() {
assert!(
CacheValidator::File { mtime_ns: 1 }.matches(&CacheValidator::File { mtime_ns: 1 })
);
assert!(
!CacheValidator::File { mtime_ns: 1 }.matches(&CacheValidator::File { mtime_ns: 2 })
);
}
#[test]
fn agent_host_and_identity_round_trip() {
let identity = CacheIdentity {
agent_id: "a".into(),
conversation_id: "c".into(),
host: AgentHost::ClaudeCode,
};
assert_eq!(
serde_json::from_str::<CacheIdentity>(&serde_json::to_string(&identity).unwrap())
.unwrap(),
identity
);
}
#[test]
fn content_handle_round_trips() {
let handle = entry().handle;
assert_eq!(
serde_json::from_str::<ContentHandleRef>(&serde_json::to_string(&handle).unwrap())
.unwrap(),
handle
);
}
#[test]
fn entry_checks_freshness_and_serializes() {
let entry = entry();
assert!(entry.is_fresh_for(&entry.key, &entry.validator, 19));
assert!(entry.is_expired_at(20));
assert_eq!(
serde_json::from_str::<DeliveryEntryV2>(&serde_json::to_string(&entry).unwrap())
.unwrap(),
entry
);
}
#[test]
fn stats_default_to_zero() {
assert_eq!(
DeliveryStatsV2::default(),
DeliveryStatsV2 {
l1_hits: 0,
l2_hits: 0,
l3_hits: 0,
misses: 0,
materializations: 0,
references_served: 0,
tokens_saved: 0,
evictions: 0,
expired: 0
}
);
}
#[test]
fn file_read_builder_includes_every_input() {
let key = file_key();
assert_eq!(key.kind(), DeliveryKind::FileRead);
assert_eq!(key.validator(), CacheValidator::File { mtime_ns: 42 });
assert_ne!(
key.cache_key(),
FileReadKey {
mode: "task".into(),
..key.clone()
}
.cache_key()
);
}
#[test]
fn shell_builder_is_immutable() {
let key = ShellCommandKey {
command_normalized: "git status".into(),
cwd: "/repo".into(),
env_hash: "env".into(),
};
assert_eq!(key.kind(), DeliveryKind::ShellCommand);
assert_eq!(key.validator(), CacheValidator::Immutable);
}
#[test]
fn search_builder_includes_filters() {
let key = SearchQueryKey {
pattern: "needle".into(),
include: "*.rs".into(),
exclude: String::new(),
path: "/repo".into(),
index_rev: "7".into(),
};
assert!(key.canonical_input().contains("include"));
assert_eq!(key.kind(), DeliveryKind::SearchQuery);
}
#[test]
fn directory_builder_uses_directory_validator() {
let key = DirectoryWalkKey {
path: "/repo".into(),
depth: 2,
gitignore: true,
dir_mtime_ns: 5,
};
assert_eq!(key.validator(), CacheValidator::Directory { mtime_ns: 5 });
assert_eq!(key.kind(), DeliveryKind::DirectoryWalk);
}
#[test]
fn composed_builder_preserves_source_digest_order() {
let first = ComposedContextKey {
task: "task".into(),
path: "/repo".into(),
source_digests: vec!["a".into(), "b".into()],
};
let second = ComposedContextKey {
source_digests: vec!["b".into(), "a".into()],
..first.clone()
};
assert_eq!(first.kind(), DeliveryKind::ComposedContext);
assert_ne!(first.cache_key(), second.cache_key());
}
}
+119
View File
@@ -0,0 +1,119 @@
//! Section-aware cache for `ctx_compose` results.
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::{Mutex, OnceLock};
use crate::core::ocla::cache_types::{CacheKeyBuilder, ComposedContextKey};
#[derive(Clone, Debug)]
struct ComposeRecord {
source_paths: Vec<PathBuf>,
source_digests: Vec<String>,
text: String,
}
/// In-process composition cache. A record is valid only when every source file
/// still has the digest used to build its `ComposedContextKey`.
#[derive(Default)]
pub struct ComposeSectionCache {
records: Mutex<BTreeMap<(String, String), ComposeRecord>>,
}
impl ComposeSectionCache {
pub fn check(&self, task: &str, path: &str) -> Option<String> {
let key = (task.trim().to_string(), path.to_string());
let record = self.records.lock().ok()?.get(&key)?.clone();
let source_digests = source_digests(&record.source_paths)?;
let builder = ComposedContextKey {
task: key.0,
path: key.1,
source_digests,
};
(builder.source_digests == record.source_digests).then_some(record.text)
}
pub fn record(&self, task: &str, path: &str, text: String) {
let source_paths = source_paths(path, &text);
let Some(source_digests) = source_digests(&source_paths) else {
return;
};
let builder = ComposedContextKey {
task: task.trim().to_string(),
path: path.to_string(),
source_digests: source_digests.clone(),
};
let _cache_key = builder.cache_key();
let key = (builder.task, builder.path);
if let Ok(mut records) = self.records.lock() {
records.insert(
key,
ComposeRecord {
source_paths,
source_digests,
text,
},
);
}
}
}
pub fn global() -> &'static ComposeSectionCache {
static CACHE: OnceLock<ComposeSectionCache> = OnceLock::new();
CACHE.get_or_init(ComposeSectionCache::default)
}
fn source_paths(project_root: &str, text: &str) -> Vec<PathBuf> {
let root = Path::new(project_root);
let mut paths = text
.lines()
.filter_map(|line| line.trim().strip_prefix("File: "))
.filter_map(|raw| raw.split_whitespace().next())
.map(|raw| {
let path = PathBuf::from(raw);
if path.is_absolute() {
path
} else {
root.join(path)
}
})
.filter(|path| path.is_file())
.collect::<Vec<_>>();
paths.sort();
paths.dedup();
paths
}
fn source_digests(paths: &[PathBuf]) -> Option<Vec<String>> {
let mut digests = paths
.iter()
.map(|path| {
std::fs::read(path)
.ok()
.map(|bytes| blake3::hash(&bytes).to_hex().to_string())
})
.collect::<Option<Vec<_>>>()?;
digests.sort();
Some(digests)
}
#[cfg(test)]
mod tests {
use super::ComposeSectionCache;
#[test]
fn section_cache_hits_only_while_all_sources_match() {
let dir = tempfile::tempdir().unwrap();
let first = dir.path().join("first.rs");
let second = dir.path().join("second.rs");
std::fs::write(&first, "one").unwrap();
std::fs::write(&second, "two").unwrap();
let root = dir.path().to_string_lossy();
let text = "File: first.rs\nbody one\nFile: second.rs\nbody two".to_string();
let cache = ComposeSectionCache::default();
cache.record("task", &root, text.clone());
assert_eq!(cache.check("task", &root), Some(text));
std::fs::write(&second, "changed").unwrap();
assert_eq!(cache.check("task", &root), None);
}
}
+6
View File
@@ -6,7 +6,12 @@
pub mod budget;
pub mod builtin;
pub mod cache_coordinator;
pub mod cache_delivery;
pub mod cache_tiers;
pub mod cache_types;
pub mod capsule;
pub mod compose_cache;
pub mod content_port;
pub mod grpc_bridge;
pub mod health;
@@ -20,6 +25,7 @@ pub mod routing_experiment;
pub mod routing_quality;
#[cfg(feature = "http-server")]
pub mod runtime;
pub mod shell_cache_allowlist;
pub mod sidecar;
pub mod tracing;
#[allow(dead_code)]
+123
View File
@@ -0,0 +1,123 @@
//! Deterministic, read-only shell commands eligible for cross-agent caching.
use super::cache_types::CacheKey;
use dashmap::DashMap;
use std::path::Path;
use std::sync::LazyLock;
/// Process-local result cache for deterministic shell commands.
pub static SHELL_RESULT_CACHE: LazyLock<DashMap<CacheKey, String>> = LazyLock::new(DashMap::new);
/// Commands whose output is deterministic for an unchanged workspace state.
pub static CACHEABLE_COMMANDS: &[&str] = &["cargo", "rg", "grep", "wc", "ls", "find", "du", "git"];
/// Returns whether `command` is one read-only command with deterministic output.
pub fn is_cacheable_command(command: &str) -> bool {
let tokens = crate::core::shell_allowlist::shell_tokenize(command.trim());
let Some(program) = tokens.first().map(String::as_str) else {
return false;
};
if !CACHEABLE_COMMANDS.contains(&program) {
return false;
}
if tokens
.iter()
.any(|token| matches!(token.as_str(), "|" | "&&" | "||" | ";" | ">" | ">>" | "<"))
{
return false;
}
match program {
"cargo" => tokens.get(1).is_some_and(|subcommand| subcommand == "test"),
"rg" | "grep" | "wc" | "ls" | "find" | "du" => true,
"git" => matches!(
tokens.get(1).map(String::as_str),
Some("log" | "status" | "diff")
),
_ => false,
}
}
/// Normalizes whitespace, option order, and local absolute paths for cache keys.
pub fn normalize_command(command: &str) -> String {
let mut tokens = crate::core::shell_allowlist::shell_tokenize(command.trim());
if tokens.is_empty() {
return String::new();
}
let root = std::env::current_dir().ok();
for token in &mut tokens {
*token = normalize_path_token(token, root.as_deref());
}
// Flags are independent for the supported read-only commands. Keep every
// positional argument in its original location, but canonicalize a leading
// run of flags (the conventional command-line shape).
let first_positional = tokens[1..]
.iter()
.position(|token| !token.starts_with('-'))
.map_or(tokens.len(), |offset| offset + 1);
tokens[1..first_positional].sort();
tokens.join(" ")
}
fn normalize_path_token(token: &str, root: Option<&Path>) -> String {
if !token.starts_with('/') {
return token.to_string();
}
let path = Path::new(token);
if let Some(root) = root
&& let Ok(relative) = path.strip_prefix(root)
{
return if relative.as_os_str().is_empty() {
"$PROJECT_ROOT".to_string()
} else {
format!("$PROJECT_ROOT/{}", relative.to_string_lossy())
};
}
"$PROJECT_ROOT".to_string()
}
#[cfg(test)]
mod tests {
use super::{is_cacheable_command, normalize_command};
#[test]
fn allowlist_accepts_the_supported_read_only_commands() {
for command in [
"cargo test",
"cargo test --lib",
"rg needle src",
"grep -R needle src",
"wc -l src/lib.rs",
"ls -la",
"find src -name '*.rs'",
"du -sh target",
"git log --oneline",
"git status --short",
"git diff --stat",
] {
assert!(is_cacheable_command(command), "{command}");
}
}
#[test]
fn allowlist_rejects_mutation_and_shell_composition() {
for command in [
"cargo build",
"git commit -m x",
"rg needle | wc -l",
"echo x",
] {
assert!(!is_cacheable_command(command), "{command}");
}
}
#[test]
fn normalization_sorts_leading_flags_and_removes_absolute_paths() {
assert_eq!(normalize_command(" rg -n -i needle "), "rg -i -n needle");
assert_eq!(
normalize_command("rg needle /tmp/other"),
"rg needle $PROJECT_ROOT"
);
}
}
+3 -2
View File
@@ -2,7 +2,8 @@ use crate::core::a2a::message::{MessagePriority, PrivacyLevel};
use super::types::{
AgentEnvelope, CompressionRequest, CompressionResult, ConfigProposal, ConfigTuningRequest,
ConnectorJob, DeliveryEntry, DeliveryRecord, DeliveryStats, EfficiencyAnalysis,
ConnectorJob, DeliveryEntry, DeliveryRecord, DeliveryRecordResult, DeliveryStats,
EfficiencyAnalysis,
EfficiencySample, ExperimentRequest, ExperimentResult, IntentDecision, IntentRequest,
MetricPoint, ModelRouteRequest, Observation, OclaCapability, OclaResult, Outcome,
ResponseOptimizationRequest, ResponseOptimizationResult, RoutingDecision, SavingsEvidence,
@@ -93,7 +94,7 @@ pub trait DeliveryRegistry: OclaService {
requester_conversation_id: Option<&str>,
) -> Option<DeliveryRecord>;
fn record_stub_served(&self, record: &DeliveryRecord, stub_tokens: u64);
fn record_delivery(&self, entry: DeliveryEntry);
fn record_delivery(&self, entry: DeliveryEntry) -> DeliveryRecordResult;
fn delivery_stats(&self) -> DeliveryStats;
}
+9
View File
@@ -571,6 +571,15 @@ pub struct DeliveryEntry {
pub mtime: u64,
}
/// Result of an idempotent delivery-record attempt.
#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
pub struct DeliveryRecordResult {
/// The existing record already represented the same source version.
pub already_recorded: bool,
/// An existing record was refreshed because its source version changed.
pub updated: bool,
}
/// Statistics for the delivery registry.
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct DeliveryStats {
+292 -20
View File
@@ -44,8 +44,13 @@ pub fn ocla_router() -> Router {
.route("/ocla/v1/capsule/{ref}", get(capsule_resolve))
.route("/ocla/v1/capsule/{ref}/fork", post(capsule_fork))
.route("/ocla/v1/delivery/check", post(delivery_check))
.route("/ocla/v1/delivery/batch-check", post(delivery_batch_check))
.route("/v1/delivery/batch-check", post(delivery_batch_check))
.route("/ocla/v1/delivery/record", post(delivery_record))
.route("/ocla/v1/delivery/stats", get(delivery_stats))
.route("/ocla/v1/cache/check", post(cache_check))
.route("/ocla/v1/cache/record", post(cache_record))
.route("/ocla/v1/cache/batch-check", post(cache_batch_check))
}
#[derive(Default)]
@@ -405,6 +410,22 @@ struct DeliveryCheckRequest {
requester_conversation_id: Option<String>,
}
#[derive(Deserialize)]
struct DeliveryBatchCheckRequest {
checks: Vec<DeliveryCheckRequest>,
}
#[derive(Serialize)]
struct DeliveryBatchCheckResult {
hit: bool,
record: Option<crate::core::ocla::types::DeliveryRecord>,
}
#[derive(Serialize)]
struct DeliveryBatchCheckResponse {
results: Vec<DeliveryBatchCheckResult>,
}
async fn delivery_check(Json(req): Json<DeliveryCheckRequest>) -> (StatusCode, Json<Value>) {
let reg = OclaRegistry::global();
match reg.delivery_registry.check_delivery(
@@ -414,30 +435,59 @@ async fn delivery_check(Json(req): Json<DeliveryCheckRequest>) -> (StatusCode, J
req.requester_agent_id.as_deref(),
req.requester_conversation_id.as_deref(),
) {
Some(record) => {
reg.delivery_registry.record_stub_served(&record, 0);
(
StatusCode::OK,
Json(json!({
"hit": true,
"path": record.path,
"line_count": record.line_count,
"token_count": record.token_count,
"agent_id": record.agent_id,
"conversation_id": record.conversation_id,
"read_at": record.read_at,
"fresh": record.fresh,
})),
)
}
Some(record) => (
StatusCode::OK,
Json(json!({
"hit": true,
"path": record.path,
"line_count": record.line_count,
"token_count": record.token_count,
"agent_id": record.agent_id,
"conversation_id": record.conversation_id,
"read_at": record.read_at,
"fresh": record.fresh,
})),
),
None => (StatusCode::OK, Json(json!({"hit": false}))),
}
}
async fn delivery_record(Json(entry): Json<crate::core::ocla::types::DeliveryEntry>) -> StatusCode {
async fn delivery_batch_check(
Json(request): Json<DeliveryBatchCheckRequest>,
) -> Json<DeliveryBatchCheckResponse> {
let reg = OclaRegistry::global();
reg.delivery_registry.record_delivery(entry);
StatusCode::NO_CONTENT
let results = request
.checks
.into_iter()
.map(|check| {
let record = reg.delivery_registry.check_delivery(
&check.blake3,
check.mtime,
&check.path,
check.requester_agent_id.as_deref(),
check.requester_conversation_id.as_deref(),
);
if let Some(record) = record {
DeliveryBatchCheckResult {
hit: true,
record: Some(record),
}
} else {
DeliveryBatchCheckResult {
hit: false,
record: None,
}
}
})
.collect();
Json(DeliveryBatchCheckResponse { results })
}
async fn delivery_record(
Json(entry): Json<crate::core::ocla::types::DeliveryEntry>,
) -> Json<crate::core::ocla::types::DeliveryRecordResult> {
let reg = OclaRegistry::global();
Json(reg.delivery_registry.record_delivery(entry))
}
async fn delivery_stats() -> Json<Value> {
@@ -452,6 +502,132 @@ async fn delivery_stats() -> Json<Value> {
}))
}
// ── Generalized cross-agent cache endpoints ──────────────────────────
fn parse_validator(s: &str) -> crate::core::ocla::cache_types::CacheValidator {
use crate::core::ocla::cache_types::CacheValidator;
if s == "immutable" {
return CacheValidator::Immutable;
}
if let Some(ns) = s.strip_prefix("file:") {
if let Ok(mtime_ns) = ns.parse::<u128>() {
return CacheValidator::File { mtime_ns };
}
}
if let Some(ns) = s.strip_prefix("directory:") {
if let Ok(mtime_ns) = ns.parse::<u128>() {
return CacheValidator::Directory { mtime_ns };
}
}
CacheValidator::Immutable
}
#[allow(dead_code)]
fn serialize_validator(v: &crate::core::ocla::cache_types::CacheValidator) -> String {
use crate::core::ocla::cache_types::CacheValidator;
match v {
CacheValidator::Immutable => "immutable".into(),
CacheValidator::File { mtime_ns } => format!("file:{mtime_ns}"),
CacheValidator::Directory { mtime_ns } => format!("directory:{mtime_ns}"),
}
}
#[derive(Deserialize)]
struct CacheCheckRequest {
key: String,
validator: String,
requester_agent_id: Option<String>,
requester_conversation_id: Option<String>,
}
async fn cache_check(Json(req): Json<CacheCheckRequest>) -> Json<Value> {
let coordinator = crate::core::ocla::cache_coordinator::materialized_cache();
use crate::core::ocla::cache_coordinator::CacheCoordinator;
let key = crate::core::ocla::cache_types::CacheKey(req.key);
let validator = parse_validator(&req.validator);
match coordinator.check(&key, &validator) {
Some(entry) => {
let same_agent = req
.requester_agent_id
.as_deref()
.is_some_and(|a| a == entry.producer.agent_id);
let same_conv = req
.requester_conversation_id
.as_deref()
.is_some_and(|c| c == entry.producer.conversation_id);
if same_agent && same_conv {
Json(json!({"hit": false}))
} else {
Json(json!({"hit": true, "entry": entry}))
}
}
None => Json(json!({"hit": false})),
}
}
async fn cache_record(
Json(entry): Json<crate::core::ocla::cache_types::DeliveryEntryV2>,
) -> StatusCode {
let coordinator = crate::core::ocla::cache_coordinator::materialized_cache();
use crate::core::ocla::cache_coordinator::CacheCoordinator;
coordinator.record(entry);
StatusCode::NO_CONTENT
}
#[derive(Deserialize)]
struct CacheBatchCheckRequest {
checks: Vec<CacheCheckRequest>,
}
#[derive(Serialize)]
struct CacheBatchCheckResult {
hit: bool,
entry: Option<crate::core::ocla::cache_types::DeliveryEntryV2>,
}
async fn cache_batch_check(
Json(request): Json<CacheBatchCheckRequest>,
) -> Json<Vec<CacheBatchCheckResult>> {
let coordinator = crate::core::ocla::cache_coordinator::materialized_cache();
use crate::core::ocla::cache_coordinator::CacheCoordinator;
let results = request
.checks
.into_iter()
.map(|check| {
let key = crate::core::ocla::cache_types::CacheKey(check.key);
let validator = parse_validator(&check.validator);
match coordinator.check(&key, &validator) {
Some(entry) => {
let same = check
.requester_agent_id
.as_deref()
.is_some_and(|a| a == entry.producer.agent_id)
&& check
.requester_conversation_id
.as_deref()
.is_some_and(|c| c == entry.producer.conversation_id);
if same {
CacheBatchCheckResult {
hit: false,
entry: None,
}
} else {
CacheBatchCheckResult {
hit: true,
entry: Some(entry),
}
}
}
None => CacheBatchCheckResult {
hit: false,
entry: None,
},
}
})
.collect();
Json(results)
}
#[cfg(test)]
mod tests {
use super::{CanonicalTokenEnvelopeV1, OCLA_API_VERSION, OclaCapabilityKind, ocla_router};
@@ -841,7 +1017,14 @@ mod tests {
)
.await
.expect("response");
assert_eq!(record_resp.status(), StatusCode::NO_CONTENT);
assert_eq!(record_resp.status(), StatusCode::OK);
assert_eq!(
json_response(record_resp).await,
json!({
"already_recorded": false,
"updated": false,
})
);
let check_body =
json!({"blake3": [1,2,3,4,5,6,7,8,9,10,11,12], "mtime": 2000, "path": "src/test.rs"});
@@ -864,6 +1047,95 @@ mod tests {
assert_eq!(val["agent_id"], "agent-x");
}
#[tokio::test]
async fn delivery_batch_check_returns_hits_and_misses_in_order() {
let app = ocla_router();
let entry = json!({
"blake3": [91,2,3,4,5,6,7,8,9,10,11,12],
"path": "src/batch.rs",
"line_count": 42,
"token_count": 168,
"agent_id": "batch-agent",
"conversation_id": "batch-conversation",
"mtime": 2000
});
let response = app
.clone()
.oneshot(
Request::builder()
.method("POST")
.uri("/ocla/v1/delivery/record")
.header(header::CONTENT_TYPE, "application/json")
.body(Body::from(entry.to_string()))
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::OK);
let checks = json!({"checks": [
{"blake3": [91,2,3,4,5,6,7,8,9,10,11,12], "mtime": 2000, "path": "src/batch.rs"},
{"blake3": [92,2,3,4,5,6,7,8,9,10,11,12], "mtime": 2000, "path": "src/missing.rs"}
]});
let response = app
.oneshot(
Request::builder()
.method("POST")
.uri("/ocla/v1/delivery/batch-check")
.header(header::CONTENT_TYPE, "application/json")
.body(Body::from(checks.to_string()))
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::OK);
let body = json_response(response).await;
assert_eq!(body["results"][0]["hit"], true);
assert_eq!(body["results"][0]["record"]["path"], "src/batch.rs");
assert_eq!(body["results"][1], json!({"hit": false, "record": null}));
}
#[tokio::test]
async fn delivery_record_reports_idempotent_and_updated_results() {
let app = ocla_router();
let entry = json!({
"blake3": [93,2,3,4,5,6,7,8,9,10,11,12],
"path": "src/idempotent-wire.rs",
"line_count": 42,
"token_count": 168,
"agent_id": "wire-agent",
"conversation_id": "wire-conversation",
"mtime": 2000
});
let request = |body: Value| {
Request::builder()
.method("POST")
.uri("/ocla/v1/delivery/record")
.header(header::CONTENT_TYPE, "application/json")
.body(Body::from(body.to_string()))
.expect("request")
};
let first = app.clone().oneshot(request(entry.clone())).await.unwrap();
assert_eq!(
json_response(first).await,
json!({"already_recorded": false, "updated": false})
);
let duplicate = app.clone().oneshot(request(entry.clone())).await.unwrap();
assert_eq!(
json_response(duplicate).await,
json!({"already_recorded": true, "updated": false})
);
let mut updated = entry;
updated["mtime"] = json!(3000);
let changed = app.oneshot(request(updated)).await.unwrap();
assert_eq!(
json_response(changed).await,
json!({"already_recorded": false, "updated": true})
);
}
#[tokio::test]
async fn delivery_stats_returns_counts() {
let app = ocla_router();
+6
View File
@@ -48,6 +48,12 @@ pub fn endpoints() -> Vec<EndpointDoc> {
auth: "bearer",
summary: "OpenAPI 3.0 spec for this surface",
},
EndpointDoc {
method: "GET",
path: "/v1/cache/stats",
auth: "bearer",
summary: "Cross-agent cache and delivery statistics",
},
EndpointDoc {
method: "GET",
path: "/v1/tools",
+90 -8
View File
@@ -1,13 +1,24 @@
use anyhow::{Context, Result};
use std::sync::OnceLock;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::runtime::Runtime;
use crate::daemon;
use crate::ipc;
static DELIVERY_RUNTIME: OnceLock<Result<Runtime, String>> = OnceLock::new();
fn delivery_runtime() -> Result<&'static Runtime> {
match DELIVERY_RUNTIME.get_or_init(|| Runtime::new().map_err(|error| error.to_string())) {
Ok(runtime) => Ok(runtime),
Err(error) => Err(anyhow::anyhow!("initialize delivery IPC runtime: {error}")),
}
}
/// Send an HTTP request to the daemon over the IPC channel.
/// Returns the response body as a string.
pub async fn daemon_request(method: &str, path: &str, body: &str) -> Result<String> {
use std::time::Duration;
use tokio::time::timeout;
const CONNECT_TIMEOUT: Duration = Duration::from_secs(3);
@@ -184,7 +195,7 @@ pub fn try_daemon_tool_call_blocking(
) -> Option<String> {
use std::time::Duration;
let rt = tokio::runtime::Runtime::new().ok()?;
let rt = Runtime::new().ok()?;
let addr = daemon::daemon_addr();
let mut ready = addr.is_listening() && rt.block_on(async { daemon_health_check().await });
@@ -312,7 +323,7 @@ pub fn try_delivery_check_blocking(
if !daemon::is_daemon_running() {
return None;
}
let rt = tokio::runtime::Runtime::new().ok()?;
let rt = delivery_runtime().ok()?;
let body = serde_json::json!({
"blake3": blake3,
"mtime": mtime,
@@ -347,7 +358,7 @@ pub fn try_delivery_check_blocking(
}
/// Record a delivery in the daemon's cross-agent registry.
/// Fire-and-forget: silently drops errors (daemon down, serialization).
/// Fire-and-forget: errors and slow daemon responses are intentionally dropped.
pub fn try_delivery_record_blocking(entry: &crate::core::ocla::types::DeliveryEntry) {
if !daemon::is_daemon_running() {
return;
@@ -355,10 +366,81 @@ pub fn try_delivery_record_blocking(entry: &crate::core::ocla::types::DeliveryEn
let Ok(body) = serde_json::to_string(entry) else {
return;
};
let Ok(rt) = tokio::runtime::Runtime::new() else {
let Ok(rt) = delivery_runtime() else {
return;
};
rt.block_on(async {
let _ = try_daemon_request("POST", "/ocla/v1/delivery/record", &body).await;
});
drop(rt.spawn(async move {
let _ = tokio::time::timeout(
Duration::from_secs(3),
try_daemon_request("POST", "/ocla/v1/delivery/record", &body),
)
.await;
}));
}
/// Check the daemon's generalized cross-agent cache (all DeliveryKinds).
/// Returns the cached entry on hit, None on miss or daemon unreachable.
pub fn try_cache_check_blocking(
key: &crate::core::ocla::cache_types::CacheKey,
validator: &crate::core::ocla::cache_types::CacheValidator,
requester_agent_id: Option<&str>,
requester_conversation_id: Option<&str>,
) -> Option<crate::core::ocla::cache_types::DeliveryEntryV2> {
if !daemon::is_daemon_running() {
return None;
}
let rt = delivery_runtime().ok()?;
let validator_str = match validator {
crate::core::ocla::cache_types::CacheValidator::Immutable => "immutable".into(),
crate::core::ocla::cache_types::CacheValidator::File { mtime_ns } => {
format!("file:{mtime_ns}")
}
crate::core::ocla::cache_types::CacheValidator::Directory { mtime_ns } => {
format!("directory:{mtime_ns}")
}
};
let body = serde_json::json!({
"key": key.0,
"validator": validator_str,
"requester_agent_id": requester_agent_id,
"requester_conversation_id": requester_conversation_id,
});
let resp = rt.block_on(async {
try_daemon_request("POST", "/ocla/v1/cache/check", &body.to_string()).await
})?;
let v: serde_json::Value = serde_json::from_str(&resp).ok()?;
if !v.get("hit")?.as_bool()? {
return None;
}
serde_json::from_value(v.get("entry")?.clone()).ok()
}
/// Record a generalized cache entry via daemon IPC. Fire-and-forget.
pub fn try_cache_record_blocking(entry: &crate::core::ocla::cache_types::DeliveryEntryV2) {
if !daemon::is_daemon_running() {
return;
}
let Ok(body) = serde_json::to_string(entry) else {
return;
};
let Ok(rt) = delivery_runtime() else { return };
drop(rt.spawn(async move {
let _ = tokio::time::timeout(
Duration::from_secs(3),
try_daemon_request("POST", "/ocla/v1/cache/record", &body),
)
.await;
}));
}
#[cfg(test)]
mod tests {
use super::delivery_runtime;
#[test]
fn delivery_runtime_is_shared() {
let first = delivery_runtime().expect("shared delivery runtime initializes");
let second = delivery_runtime().expect("shared delivery runtime remains available");
assert!(std::ptr::eq(first, second));
}
}
+83 -1
View File
@@ -25,7 +25,7 @@ use axum::{
use futures::Stream;
use rmcp::transport::{StreamableHttpServerConfig, StreamableHttpService};
use serde::Deserialize;
use serde_json::Value;
use serde_json::{Value, json};
use tokio::sync::broadcast;
use tokio::time::{Duration, Instant};
@@ -423,6 +423,66 @@ async fn v1_openapi(State(state): State<AppState>) -> impl IntoResponse {
(StatusCode::OK, Json(crate::core::openapi::openapi_value()))
}
/// `GET /v1/cache/stats` — live cross-agent cache and delivery metrics.
async fn v1_cache_stats() -> impl IntoResponse {
let cache = crate::core::ocla::cache_coordinator::materialized_cache();
use crate::core::ocla::cache_coordinator::CacheCoordinator as _;
let stats = cache.stats();
let delivery = crate::core::ocla::OclaRegistry::global()
.delivery_registry
.delivery_stats();
let by_kind = {
let mut m = serde_json::Map::new();
for kind in &[
"file_read",
"shell_command",
"search_query",
"directory_walk",
"composed_context",
] {
m.insert(kind.to_string(), json!({ "hits": 0_u64, "misses": 0_u64 }));
}
m
};
let hit_rate = |hits: u64, misses: u64| {
let total = hits + misses;
if total == 0 {
0.0
} else {
hits as f64 / total as f64
}
};
(
StatusCode::OK,
Json(json!({
"l1": {
"entries": cache.l1().len(),
"hits": stats.l1_hits,
"misses": stats.misses,
"hit_rate": hit_rate(stats.l1_hits, stats.misses),
},
"l2": {
"entries": cache.l2().len(),
"hits": stats.l2_hits,
"misses": stats.misses,
"hit_rate": hit_rate(stats.l2_hits, stats.misses),
},
"l3": {
"entries": cache.l3().len(),
"bytes": cache.l3().len(),
"hits": stats.l3_hits,
"misses": stats.misses,
},
"delivery": {
"total_stubs": delivery.stubs_served,
"tokens_saved": delivery.tokens_saved,
"references_served": stats.references_served,
},
"by_kind": by_kind,
})),
)
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct ToolsQuery {
@@ -921,6 +981,7 @@ fn build_app_router_with_auth(cfg: &HttpServerConfig, require_auth: bool) -> Rou
.route("/v1/manifest", get(v1_manifest))
.route("/v1/capabilities", get(v1_capabilities))
.route("/v1/openapi.json", get(v1_openapi))
.route("/v1/cache/stats", get(v1_cache_stats))
.route("/v1/tools", get(v1_tools))
.route("/v1/tools/call", axum::routing::post(v1_tool_call))
.route("/v1/events", get(v1_events))
@@ -1384,6 +1445,27 @@ mod tests {
assert!(json["contracts"].is_object());
}
#[tokio::test]
async fn cache_stats_endpoint_returns_live_shape() {
let app = Router::new().route("/v1/cache/stats", get(v1_cache_stats));
let request = Request::builder()
.method("GET")
.uri("/v1/cache/stats")
.body(Body::empty())
.unwrap();
let response = app.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = axum::body::to_bytes(response.into_body(), 1_000_000)
.await
.unwrap();
let value: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert!(value["l1"]["entries"].is_u64());
assert!(value["l2"]["hit_rate"].is_number());
assert!(value["l3"]["bytes"].is_u64());
assert!(value["delivery"]["references_served"].is_u64());
assert!(value["by_kind"]["shell_command"].is_object());
}
#[tokio::test]
async fn openapi_endpoint_returns_spec() {
let dir = tempfile::tempdir().expect("tempdir");
+6 -14
View File
@@ -361,15 +361,13 @@ pub fn run_setup() {
crate::core::layout_pin::heal();
terminal_ui::print_step_header(9, 13, "Help Improve lean-ctx");
println!(" Share anonymous data to make lean-ctx better:");
println!(
" \x1b[2m • Telemetry heartbeat: version, OS, architecture, random install ID\x1b[0m"
);
println!(" \x1b[2m • Compression stats: file-type, size bucket, mode, ratio\x1b[0m");
println!(" \x1b[1mNo code, no file names, no personal data — ever.\x1b[0m");
println!(" \x1b[2mInspect anytime: lean-ctx telemetry show\x1b[0m");
println!(" Share anonymous telemetry to make lean-ctx better:");
println!("  • Version, OS, architecture, random install ID");
println!("  • Compression patterns: file-type, size bucket, mode, ratio");
println!(" No code, no file names, no personal data — ever.");
println!(" Inspect anytime: lean-ctx telemetry show");
println!();
print!(" Enable anonymous data sharing? \x1b[1m[y/N]\x1b[0m ");
print!(" Enable anonymous telemetry? [y/N] ");
use std::io::Write;
std::io::stdout().flush().ok();
@@ -388,12 +386,6 @@ pub fn run_setup() {
let _ = std::fs::create_dir_all(dir);
}
let mut config_content = std::fs::read_to_string(&config_path).unwrap_or_default();
if !config_content.contains("[cloud]") {
if !config_content.is_empty() && !config_content.ends_with('\n') {
config_content.push('\n');
}
config_content.push_str("\n[cloud]\ncontribute_enabled = true\n");
}
if !config_content.contains("[telemetry]") {
if !config_content.ends_with('\n') {
config_content.push('\n');
+119 -6
View File
@@ -1,5 +1,6 @@
use crate::core::cache::SessionCache;
use crate::core::heatmap;
use crate::core::ocla::cache_types::{CacheKeyBuilder, FileReadKey};
use crate::core::tokens::count_tokens;
use crate::tools::CrpMode;
use crate::tools::ctx_read;
@@ -35,9 +36,29 @@ pub fn handle_with_task_fresh(
crp_mode: CrpMode,
task: Option<&str>,
) -> String {
handle_with_task_fresh_result(cache, paths, mode, fresh, crp_mode, task).text
}
/// Batch-read result with the aggregate baseline across local and cross-agent hits.
pub struct MultiReadResult {
pub text: String,
pub original_tokens: usize,
}
pub fn handle_with_task_fresh_result(
cache: &mut SessionCache,
paths: &[String],
mode: &str,
fresh: bool,
crp_mode: CrpMode,
task: Option<&str>,
) -> MultiReadResult {
let n = paths.len();
if n == 0 {
return "Read 0 files | 0 tokens saved".to_string();
return MultiReadResult {
text: "Read 0 files | 0 tokens saved".to_string(),
original_tokens: 0,
};
}
let max_bytes = max_multi_read_bytes();
@@ -54,12 +75,42 @@ pub fn handle_with_task_fresh(
} else {
mode
};
let chunk = if fresh {
ctx_read::handle_fresh_with_task(cache, path, effective_mode, crp_mode, task)
let cache_key = file_read_cache_key(path, effective_mode, crp_mode, task);
let cross_agent = (!fresh)
.then(|| {
crate::core::ocla::cache_delivery::check(
&cache_key.cache_key(),
&cache_key.validator(),
"ctx_multi_read",
)
})
.flatten();
let (chunk, cross_agent_original) = if let Some(entry) = cross_agent {
(
crate::core::ocla::cache_delivery::stub(&entry, "file read"),
Some(entry.token_count as usize),
)
} else {
ctx_read::handle_with_task(cache, path, effective_mode, crp_mode, task)
let chunk = if fresh {
ctx_read::handle_fresh_with_task(cache, path, effective_mode, crp_mode, task)
} else {
ctx_read::handle_with_task(cache, path, effective_mode, crp_mode, task)
};
if !chunk.contains("[cross-agent") {
crate::core::ocla::cache_delivery::record(
cache_key.cache_key(),
crate::core::ocla::cache_types::DeliveryKind::FileRead,
cache_key.validator(),
Some(cache_key.path.clone()),
&chunk,
"ctx_multi_read",
);
}
(chunk, None)
};
let original = cache.get(path).map_or(0, |e| e.original_tokens);
let original = cross_agent_original
.or_else(|| cache.get(path).map(|entry| entry.original_tokens))
.unwrap_or(0);
let sent = count_tokens(&chunk);
heatmap::record_file_access(path, original, original.saturating_sub(sent));
// Verified ledger (#685): model-correct counts. The default O200kBase model
@@ -108,5 +159,67 @@ pub fn handle_with_task_fresh(
} else {
format!("Read {n} files")
};
format!("{body}\n---\n{summary}")
MultiReadResult {
text: format!("{body}\n---\n{summary}"),
original_tokens: total_original,
}
}
fn file_read_cache_key(
path: &str,
mode: &str,
crp_mode: CrpMode,
task: Option<&str>,
) -> FileReadKey {
let canonical = crate::core::pathutil::safe_canonicalize_or_self(std::path::Path::new(path));
let mtime_ns = std::fs::metadata(&canonical)
.ok()
.and_then(|metadata| metadata.modified().ok())
.and_then(|mtime| mtime.duration_since(std::time::UNIX_EPOCH).ok())
.map_or(0, |duration| duration.as_nanos());
FileReadKey {
path: canonical.to_string_lossy().into_owned(),
mtime_ns,
mode: mode.into(),
crp_mode: format!("{crp_mode:?}").to_ascii_lowercase(),
task_digest: blake3::hash(task.unwrap_or_default().as_bytes())
.to_hex()
.to_string(),
policy_rev: env!("CARGO_PKG_VERSION").into(),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn multi_read_deduplicates_each_file_with_cross_agent_references() {
let directory = tempfile::tempdir().unwrap();
let first = directory.path().join("first.rs");
let second = directory.path().join("second.rs");
std::fs::write(&first, "fn first_probe() {}\n").unwrap();
std::fs::write(&second, "fn second_probe() {}\n").unwrap();
let paths = vec![
first.to_string_lossy().into_owned(),
second.to_string_lossy().into_owned(),
];
let mut local = SessionCache::new();
let initial =
handle_with_task_fresh_result(&mut local, &paths, "full", false, CrpMode::Off, None);
assert!(initial.text.contains("first_probe"));
let mut another_agent = SessionCache::new();
let repeated = handle_with_task_fresh_result(
&mut another_agent,
&paths,
"full",
false,
CrpMode::Off,
None,
);
assert_eq!(repeated.text.matches("[cross-agent cache").count(), 2);
assert!(repeated.original_tokens > 0);
}
}
+100 -21
View File
@@ -199,6 +199,32 @@ pub(crate) fn is_subagent_context() -> bool {
})
}
/// Keeps subagent cache isolation while independently deciding whether a
/// cross-agent delivery lookup may return a stub.
#[allow(clippy::fn_params_excessive_bools)]
pub(crate) fn effective_fresh_flags(
fresh: bool,
force_fresh: bool,
subagent_context: bool,
delivery_for_subagents: bool,
) -> (bool, bool) {
let effective_fresh_for_cache = fresh || force_fresh || subagent_context;
let effective_fresh_for_delivery =
fresh || force_fresh || (subagent_context && !delivery_for_subagents);
(effective_fresh_for_cache, effective_fresh_for_delivery)
}
pub(crate) fn effective_fresh_for_delivery(fresh: bool) -> bool {
let config = crate::core::config::Config::load();
effective_fresh_flags(
fresh,
force_fresh_env(),
is_subagent_context(),
config.ocla.delivery.delivery_for_subagents,
)
.1
}
fn handle_with_options_resolved(
cache: &mut SessionCache,
path: &str,
@@ -221,10 +247,15 @@ fn handle_with_options_resolved_preread(
tuning: ReadTuning<'_>,
preread: Option<String>,
) -> ReadOutput {
// #1292: Sub-agents have separate context windows and never received
// the parent's reads. Always force fresh regardless of scope state —
// correctness over cache savings for short-lived sub-agent contexts.
let effective_fresh = fresh || force_fresh_env() || is_subagent_context();
// Subagents retain isolated session caches, but can use delivery stubs when
// configured because those stubs explicitly identify another agent's read.
let config = crate::core::config::Config::load();
let (effective_fresh_for_cache, effective_fresh_for_delivery) = effective_fresh_flags(
fresh,
force_fresh_env(),
is_subagent_context(),
config.ocla.delivery.delivery_for_subagents,
);
let compress_protected = mode != "raw"
&& !mode.starts_with("lines:")
@@ -232,10 +263,21 @@ fn handle_with_options_resolved_preread(
.proxy
.is_path_compress_protected(path);
if !effective_fresh
// Hash once for cross-agent delivery. The same snapshot is used for both
// the pre-read lookup and the post-read record, avoiding a second disk read.
let delivery_metadata = config
.ocla
.delivery_enabled()
.then(|| file_blake3_prefix(path))
.flatten();
if !effective_fresh_for_delivery
&& !compress_protected
&& let Some(stub) = try_cross_agent_stub(path, mode)
&& let Some((hash, mtime)) = delivery_metadata
&& let Some(stub) = try_cross_agent_stub(path, mode, hash, mtime)
{
cache.store(path, &stub.content);
cache.mark_full_delivered(path);
return stub;
}
@@ -252,7 +294,7 @@ fn handle_with_options_resolved_preread(
cache,
path,
mode,
effective_fresh,
effective_fresh_for_cache,
crp_mode,
task,
tuning,
@@ -281,8 +323,11 @@ fn handle_with_options_resolved_preread(
}
}
if !result.is_cache_hit {
record_cross_agent_delivery(path, result.output_tokens);
if !result.is_cache_hit
&& let Some((hash, mtime)) = delivery_metadata
{
let line_count = cache.get(path).map_or(0, |entry| entry.line_count as u32);
record_cross_agent_delivery(path, hash, mtime, line_count, result.output_tokens);
}
// SSOT via [`ReadMode`] (#528): lossy summaries may elide shared blocks.
@@ -584,7 +629,7 @@ pub fn resolve_explicit_delta_mode(
unchanged
}
fn file_blake3_prefix(path: &str) -> Option<([u8; 12], u64)> {
pub(crate) fn file_blake3_prefix(path: &str) -> Option<([u8; 12], u64)> {
let meta = std::fs::metadata(path).ok()?;
let mtime = meta
.modified()
@@ -600,18 +645,23 @@ fn file_blake3_prefix(path: &str) -> Option<([u8; 12], u64)> {
Some((prefix, mtime))
}
fn try_cross_agent_stub(path: &str, mode: &str) -> Option<ReadOutput> {
pub(crate) fn try_cross_agent_stub(
path: &str,
mode: &str,
hash: [u8; 12],
mtime: u64,
) -> Option<ReadOutput> {
if !crate::core::config::Config::load().ocla.delivery_enabled() {
return None;
}
if matches!(mode, "full" | "raw" | "diff") {
return None;
}
let (hash, mtime) = file_blake3_prefix(path)?;
let current_agent = std::env::var("CURSOR_TASK_ID")
.or_else(|_| std::env::var("CLAUDECODE"))
.unwrap_or_else(|_| "local-agent".to_string());
let current_conversation = current_agent.clone();
let current_conversation = crate::core::conversation::current_conversation_id()
.unwrap_or_else(|| current_agent.clone());
let reg = crate::core::ocla::OclaRegistry::global();
let record = crate::daemon_client::try_delivery_check_blocking(
&hash,
@@ -647,18 +697,21 @@ fn try_cross_agent_stub(path: &str, mode: &str) -> Option<ReadOutput> {
})
}
fn record_cross_agent_delivery(path: &str, tokens: usize) {
pub(crate) fn record_cross_agent_delivery(
path: &str,
hash: [u8; 12],
mtime: u64,
line_count: u32,
tokens: usize,
) {
if !crate::core::config::Config::load().ocla.delivery_enabled() {
return;
}
let Some((hash, mtime)) = file_blake3_prefix(path) else {
return;
};
let line_count = std::fs::read_to_string(path).map_or(0, |c| c.lines().count() as u32);
let agent_id = std::env::var("CURSOR_TASK_ID")
.or_else(|_| std::env::var("CLAUDECODE"))
.unwrap_or_else(|_| "local-agent".to_string());
let conversation_id = agent_id.clone();
let conversation_id =
crate::core::conversation::current_conversation_id().unwrap_or_else(|| agent_id.clone());
let entry = crate::core::ocla::types::DeliveryEntry {
blake3: hash,
path: path.into(),
@@ -675,15 +728,41 @@ fn record_cross_agent_delivery(path: &str, tokens: usize) {
#[cfg(test)]
mod tests {
use super::{SessionCache, try_cross_agent_stub, try_stub_hit_readonly_scoped};
use super::{
SessionCache, effective_fresh_flags, try_cross_agent_stub, try_stub_hit_readonly_scoped,
};
use std::sync::atomic::Ordering;
#[test]
fn cross_agent_stub_miss_returns_none() {
let stub = try_cross_agent_stub("/nonexistent/file.rs", "auto");
let stub = try_cross_agent_stub("/nonexistent/file.rs", "auto", [0; 12], 0);
assert!(stub.is_none());
}
#[test]
fn subagent_delivery_policy_keeps_cache_fresh_but_allows_delivery_by_default() {
let delivery_for_subagents =
crate::core::config::DeliveryConfig::default().delivery_for_subagents;
assert!(
delivery_for_subagents,
"delivery must default to enabled for subagents"
);
let (cache_fresh, delivery_fresh) =
effective_fresh_flags(false, false, true, delivery_for_subagents);
assert!(cache_fresh, "subagent cache must remain isolated");
assert!(
!delivery_fresh,
"default policy must allow a cross-agent delivery lookup"
);
}
#[test]
fn subagent_delivery_policy_can_force_fresh_delivery() {
let (cache_fresh, delivery_fresh) = effective_fresh_flags(false, false, true, false);
assert!(cache_fresh);
assert!(delivery_fresh, "disabled policy must bypass delivery stubs");
}
#[test]
fn cross_agent_fallback_is_deterministic() {
// When no CURSOR_TASK_ID or CLAUDECODE env var is set, the fallback
@@ -1,3 +1,4 @@
use crate::core::ocla::cache_types::{CacheKeyBuilder, SearchQueryKey};
use rmcp::ErrorData;
use rmcp::model::Tool;
use serde_json::{Map, Value, json};
@@ -63,6 +64,37 @@ impl McpTool for CtxCallgraphTool {
let from = get_str(args, "from");
let to = get_str(args, "to");
let cache_input = format!(
"callgraph:{action_normalized}:{}:{}:{depth}",
symbol.as_deref().unwrap_or(""),
file.as_deref().unwrap_or("")
);
let builder = SearchQueryKey {
path: ctx.project_root.clone(),
index_rev: String::new(),
pattern: cache_input,
include: String::new(),
exclude: String::new(),
};
let key = builder.cache_key();
let validator = builder.validator();
if let Some(entry) =
crate::core::ocla::cache_delivery::check(&key, &validator, "ctx_callgraph")
{
let stub = crate::core::ocla::cache_delivery::stub(&entry, "callgraph");
return Ok(ToolOutput {
text: stub,
original_tokens: entry.token_count as usize,
saved_tokens: entry.token_count as usize,
mode: Some(action_normalized),
path: None,
changed: false,
shell_outcome: None,
content_blocks: None,
});
}
let result = crate::tools::ctx_callgraph::handle(
&action_normalized,
symbol.as_deref(),
@@ -73,6 +105,15 @@ impl McpTool for CtxCallgraphTool {
to.as_deref(),
);
crate::core::ocla::cache_delivery::record(
key,
crate::core::ocla::cache_types::DeliveryKind::SearchQuery,
validator,
None,
&result,
"ctx_callgraph",
);
Ok(ToolOutput {
text: result,
original_tokens: 0,
+41 -3
View File
@@ -1,3 +1,4 @@
use crate::core::ocla::cache_types::{CacheKeyBuilder, ComposedContextKey};
use rmcp::ErrorData;
use rmcp::model::Tool;
use serde_json::{Map, Value, json};
@@ -52,9 +53,46 @@ impl McpTool for CtxComposeTool {
crate::tools::ctx_semantic_search::set_thread_cache(cache.clone());
}
let (text, sent) = tokio::task::block_in_place(|| {
crate::tools::ctx_compose::handle(&task, &path, ctx.crp_mode)
});
let cache_enabled = crate::core::config::Config::load()
.cache
.compose_cache_enabled;
let cached = cache_enabled
.then(|| crate::core::ocla::compose_cache::global().check(&task, &path))
.flatten();
let (text, sent) = if let Some(text) = cached {
let sent = crate::core::tokens::count_tokens(&text);
(text, sent)
} else {
// Cross-process delivery check before expensive computation
let compose_builder = ComposedContextKey {
task: task.clone(),
path: path.clone(),
source_digests: Vec::new(),
};
let ck = compose_builder.cache_key();
let cv = compose_builder.validator();
if let Some(entry) = crate::core::ocla::cache_delivery::check(&ck, &cv, "ctx_compose") {
let stub = crate::core::ocla::cache_delivery::stub(&entry, "compose");
let sent = crate::core::tokens::count_tokens(&stub);
(stub, sent)
} else {
let (text, sent) = tokio::task::block_in_place(|| {
crate::tools::ctx_compose::handle(&task, &path, ctx.crp_mode)
});
if cache_enabled && !text.starts_with("ERROR") {
crate::core::ocla::compose_cache::global().record(&task, &path, text.clone());
crate::core::ocla::cache_delivery::record(
ck,
crate::core::ocla::cache_types::DeliveryKind::ComposedContext,
cv,
Some(path.clone()),
&text,
"ctx_compose",
);
}
(text, sent)
}
};
if text.starts_with("ERROR") {
return Err(ErrorData::invalid_params(text, None));
+105 -8
View File
@@ -2,6 +2,7 @@ use rmcp::ErrorData;
use rmcp::model::Tool;
use serde_json::{Map, Value, json};
use crate::core::ocla::cache_types::{CacheKey, CacheKeyBuilder, DirectoryWalkKey};
use crate::server::tool_trait::{McpTool, ToolContext, ToolOutput, get_bool, get_int, get_str};
use crate::tool_defs::tool_def;
@@ -75,13 +76,7 @@ impl McpTool for CtxGlobTool {
// `block_in_place` here would needlessly consume blocking-pool
// threads (the lesson from the ctx_multi_read crash, #271).
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
crate::tools::ctx_glob::handle(
&pattern,
root,
respect,
allow_secret_paths,
per_root_max,
)
cached_or_walk(&pattern, root, respect, allow_secret_paths, per_root_max)
}));
let Ok((result, original)) = result else {
@@ -121,7 +116,7 @@ fn handle_single(
max_results: usize,
) -> Result<ToolOutput, ErrorData> {
let Ok((result, original)) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
crate::tools::ctx_glob::handle(
cached_or_walk(
pattern,
path,
respect_gitignore,
@@ -156,3 +151,105 @@ fn handle_single(
content_blocks: None,
})
}
/// Builds the versioned directory-walk cache key for a glob request.
fn glob_cache_key(pattern: &str, path: &str, depth: usize) -> CacheKey {
glob_cache_builder(pattern, path, depth, true, false).cache_key()
}
fn glob_cache_builder(
_pattern: &str,
path: &str,
depth: usize,
respect_gitignore: bool,
_allow_secret_paths: bool,
) -> DirectoryWalkKey {
let canonical = crate::core::pathutil::safe_canonicalize_or_self(std::path::Path::new(path));
let dir_mtime_ns = directory_mtime_ns(&canonical).unwrap_or_default();
DirectoryWalkKey {
path: canonical.to_string_lossy().into_owned(),
depth,
gitignore: respect_gitignore,
dir_mtime_ns,
}
}
fn directory_mtime_ns(path: &std::path::Path) -> Option<u128> {
std::fs::metadata(path)
.ok()?
.modified()
.ok()?
.duration_since(std::time::UNIX_EPOCH)
.ok()
.map(|duration| duration.as_nanos())
}
fn cached_or_walk(
pattern: &str,
path: &str,
respect_gitignore: bool,
allow_secret_paths: bool,
max_results: usize,
) -> (String, usize) {
// Glob has no explicit depth limit; preserve that in the key instead of
// accidentally sharing a limited tree result.
let selector = format!("{pattern}\\x1fmax:{max_results}");
let builder = glob_cache_builder(
&selector,
path,
usize::MAX,
respect_gitignore,
allow_secret_paths,
);
let key = if respect_gitignore && allow_secret_paths {
glob_cache_key(&selector, path, usize::MAX)
} else {
builder.cache_key()
};
if let Some(entry) =
crate::core::ocla::cache_delivery::check(&key, &builder.validator(), "ctx_glob")
{
let stub = crate::core::ocla::cache_delivery::stub(&entry, "directory walk");
return (stub, entry.token_count as usize);
}
let (result, original) = crate::tools::ctx_glob::handle(
pattern,
path,
respect_gitignore,
allow_secret_paths,
max_results,
);
if !result.starts_with("ERROR:") {
crate::core::ocla::cache_delivery::record(
key,
crate::core::ocla::cache_types::DeliveryKind::DirectoryWalk,
builder.validator(),
Some(builder.path),
&result,
"ctx_glob",
);
}
(result, original)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn glob_adapter_records_then_serves_a_cross_agent_reference() {
let directory = tempfile::tempdir().unwrap();
std::fs::write(directory.path().join("cached.rs"), "fn cached() {}\n").unwrap();
let path = directory.path().to_string_lossy();
let first = handle_single("*.rs", &path, true, true, 20).unwrap();
assert!(first.text.contains("cached.rs"));
let second = handle_single("*.rs", &path, true, true, 20).unwrap();
assert!(
second.text.contains("[cross-agent cache"),
"{}",
second.text
);
}
}
+4 -8
View File
@@ -151,7 +151,7 @@ fn handle_inner(args: &Map<String, Value>, ctx: &ToolContext) -> Result<ToolOutp
None,
));
};
let output = crate::tools::ctx_multi_read::handle_with_task_fresh(
let output = crate::tools::ctx_multi_read::handle_with_task_fresh_result(
&mut cache,
&paths,
&mode,
@@ -159,16 +159,12 @@ fn handle_inner(args: &Map<String, Value>, ctx: &ToolContext) -> Result<ToolOutp
ctx.crp_mode,
current_task.as_deref(),
);
let mut total_original: usize = 0;
for path in &paths {
total_original =
total_original.saturating_add(cache.get(path).map_or(0, |e| e.original_tokens));
}
let tokens = crate::core::tokens::count_tokens(&output);
let total_original = output.original_tokens;
let tokens = crate::core::tokens::count_tokens(&output.text);
drop(cache);
Ok(ToolOutput {
text: output,
text: output.text,
original_tokens: total_original,
saved_tokens: total_original.saturating_sub(tokens),
mode: Some(mode),
+41
View File
@@ -346,6 +346,12 @@ impl CtxReadTool {
// channel overhead for the ~90% of calls that are cache hits.
let read_timeout = std::time::Duration::from_secs(30);
let cancelled = Arc::new(AtomicBool::new(false));
// Hash once for cross-agent delivery (avoids re-reading on record).
let delivery_metadata = crate::core::config::Config::load()
.ocla
.delivery_enabled()
.then(|| crate::tools::ctx_read::file_blake3_prefix(path))
.flatten();
let (output, resolved_mode, original, is_cache_hit, file_ref, cache_stats) = {
let crp_mode = ctx.crp_mode;
let fast_result = 'fast: {
@@ -450,6 +456,29 @@ impl CtxReadTool {
return;
}
// The session-local stub is checked first so the current
// agent's own delivery always wins. On a miss, a verified
// cross-agent delivery can avoid the disk read below.
if !crate::tools::ctx_read::effective_fresh_for_delivery(fresh)
&& let Some((hash, mtime)) = delivery_metadata
&& let Some(read_output) = crate::tools::ctx_read::try_cross_agent_stub(
&path_owned,
&mode,
hash,
mtime,
)
{
let _ = tx.send((
read_output.content,
read_output.resolved_mode,
0,
read_output.is_cache_hit,
None,
(0, 0),
));
return;
}
// Phase 2a: disk I/O under per-file lock but WITHOUT cache lock.
let preread = crate::tools::ctx_read::read_file_lossy(&path_owned).ok();
@@ -860,6 +889,18 @@ impl CtxReadTool {
let output_tokens = crate::core::tokens::count_tokens(&output);
let saved = original.saturating_sub(output_tokens);
if !is_cache_hit {
if let Some((hash, mtime)) = delivery_metadata {
crate::tools::ctx_read::record_cross_agent_delivery(
path,
hash,
mtime,
0,
output_tokens,
);
}
}
// Session updates (bounded lock — 10s timeout, read already succeeded)
let mut ensured_root: Option<String> = None;
let mut traversal_working_set: Vec<String> = Vec::new();
@@ -97,6 +97,126 @@ fn per_file_lock_allows_parallel_different_paths() {
assert!(max_concurrent.load(Ordering::SeqCst) > 1);
}
/// The primary MCP handler must consult cross-agent delivery only after its
/// session-local stub miss and before it starts the disk/compression pipeline.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn mcp_ctx_read_serves_cross_agent_delivery_stub_before_disk_read() {
use crate::core::cache::SessionCache;
use crate::core::ocla::OclaRegistry;
use crate::core::ocla::types::DeliveryEntry;
use crate::core::session::SessionState;
use std::sync::Arc;
use tokio::sync::RwLock;
let dir = tempfile::tempdir().unwrap();
let file = dir.path().join("cross-agent-mcp.rs");
std::fs::write(&file, "fn only_the_remote_agent_read_this() {}\n").unwrap();
let path = file.to_string_lossy().to_string();
let bytes = std::fs::read(&file).unwrap();
let hash = blake3::hash(&bytes);
let mut blake3_prefix = [0u8; 12];
blake3_prefix.copy_from_slice(&hash.as_bytes()[..12]);
let mtime = std::fs::metadata(&file)
.unwrap()
.modified()
.unwrap()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs();
let requester = std::env::var("CURSOR_TASK_ID")
.or_else(|_| std::env::var("CLAUDECODE"))
.unwrap_or_else(|_| "local-agent".to_string());
let remote_agent = format!("{requester}-remote");
OclaRegistry::global()
.delivery_registry
.record_delivery(DeliveryEntry {
blake3: blake3_prefix,
path: path.clone(),
line_count: 1,
token_count: 12,
agent_id: remote_agent.clone(),
conversation_id: remote_agent,
mtime,
});
let ctx = ToolContext {
project_root: dir.path().to_string_lossy().to_string(),
resolved_paths: std::collections::HashMap::from([("path".to_string(), path.clone())]),
cache: Some(Arc::new(RwLock::new(SessionCache::new()))),
session: Some(Arc::new(RwLock::new(SessionState::new()))),
..ToolContext::default()
};
let args = json!({ "path": path, "mode": "auto" })
.as_object()
.unwrap()
.clone();
let output = tokio::task::block_in_place(|| CtxReadTool.handle(&args, &ctx))
.expect("ctx_read must serve the cross-agent delivery stub");
assert!(output.text.contains("[cross-agent"), "got: {}", output.text);
assert!(
!output.text.contains("only_the_remote_agent_read_this"),
"cross-agent hit must return before disk content is read: {}",
output.text
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn mcp_ctx_read_records_new_cross_agent_delivery() {
use crate::core::cache::SessionCache;
use crate::core::ocla::OclaRegistry;
use crate::core::session::SessionState;
use std::sync::Arc;
use tokio::sync::RwLock;
let dir = tempfile::tempdir().unwrap();
let file = dir.path().join("record-delivery-mcp.rs");
std::fs::write(&file, "fn mcp_records_delivery() {}\n").unwrap();
let path = file.to_string_lossy().to_string();
let ctx = ToolContext {
project_root: dir.path().to_string_lossy().to_string(),
resolved_paths: std::collections::HashMap::from([("path".to_string(), path.clone())]),
cache: Some(Arc::new(RwLock::new(SessionCache::new()))),
session: Some(Arc::new(RwLock::new(SessionState::new()))),
..ToolContext::default()
};
let args = json!({ "path": path, "mode": "auto" })
.as_object()
.unwrap()
.clone();
let output = tokio::task::block_in_place(|| CtxReadTool.handle(&args, &ctx))
.expect("ctx_read must complete the initial delivery");
assert!(
output.text.contains("mcp_records_delivery"),
"got: {}",
output.text
);
let bytes = std::fs::read(&file).unwrap();
let hash = blake3::hash(&bytes);
let mut blake3_prefix = [0u8; 12];
blake3_prefix.copy_from_slice(&hash.as_bytes()[..12]);
let mtime = std::fs::metadata(&file)
.unwrap()
.modified()
.unwrap()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs();
let delivery = OclaRegistry::global().delivery_registry.check_delivery(
&blake3_prefix,
mtime,
&path,
Some("mcp-delivery-probe"),
Some("mcp-delivery-probe"),
);
assert!(
delivery.is_some(),
"MCP read must record its fresh delivery"
);
}
/// Regression test for Issue #229: a zombie thread holding the cache write-lock
/// must not block subsequent reads indefinitely. The try_write() loop inside
/// the spawned thread should respect its 25s deadline and the cancellation flag.
+157 -3
View File
@@ -2,6 +2,7 @@ use rmcp::ErrorData;
use rmcp::model::Tool;
use serde_json::{Map, Value, json};
use crate::core::ocla::cache_types::{CacheKeyBuilder, SearchQueryKey};
use crate::server::tool_trait::{
McpTool, ToolContext, ToolOutput, get_bool, get_int, get_str, get_str_array, get_usize,
};
@@ -234,7 +235,7 @@ fn handle_regex(args: &Map<String, Value>, ctx: &ToolContext) -> Result<ToolOutp
for root in &resolved.roots {
let search_result = tokio::task::block_in_place(|| {
std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
crate::tools::ctx_search::handle_filtered(
cached_or_search(
&pattern,
root,
include.as_deref(),
@@ -499,6 +500,159 @@ fn semantic_output(text: String) -> ToolOutput {
}
}
#[allow(clippy::too_many_arguments)]
fn cached_or_search(
pattern: &str,
path: &str,
include: Option<&str>,
max: usize,
crp: crate::tools::CrpMode,
respect_gitignore: bool,
allow_secret_paths: bool,
anchored: bool,
exclude: Option<&str>,
exclude_pattern: Option<&str>,
) -> crate::tools::ctx_search::SearchOutcome {
let builder = regex_cache_builder(
pattern,
path,
include,
exclude,
exclude_pattern,
max,
respect_gitignore,
allow_secret_paths,
anchored,
);
let key = builder.cache_key();
if let Some(entry) =
crate::core::ocla::cache_delivery::check(&key, &builder.validator(), "ctx_search")
{
let text = crate::core::ocla::cache_delivery::stub(&entry, "regex search");
return crate::tools::ctx_search::SearchOutcome {
text,
modeled_baseline: entry.token_count as usize,
observed_tokens: entry.token_count as usize,
};
}
let outcome = crate::tools::ctx_search::handle_filtered(
pattern,
path,
include,
max,
crp,
respect_gitignore,
allow_secret_paths,
anchored,
exclude,
exclude_pattern,
);
if !outcome.text.starts_with("ERROR:") {
crate::core::ocla::cache_delivery::record(
key,
crate::core::ocla::cache_types::DeliveryKind::SearchQuery,
builder.validator(),
Some(builder.path),
&outcome.text,
"ctx_search",
);
}
outcome
}
#[allow(clippy::too_many_arguments)]
fn regex_cache_builder(
pattern: &str,
path: &str,
include: Option<&str>,
exclude: Option<&str>,
exclude_pattern: Option<&str>,
max: usize,
respect_gitignore: bool,
allow_secret_paths: bool,
anchored: bool,
) -> SearchQueryKey {
let canonical = crate::core::pathutil::safe_canonicalize_or_self(std::path::Path::new(path));
SearchQueryKey {
pattern: pattern.into(),
include: format!(
"{}\\x1fmax:{max}\\x1fgitignore:{respect_gitignore}\\x1fsecret:{allow_secret_paths}\\x1fanchored:{anchored}",
include.unwrap_or_default()
),
exclude: format!(
"{}\\x1fline:{}",
exclude.unwrap_or_default(),
exclude_pattern.unwrap_or_default()
),
path: canonical.to_string_lossy().into_owned(),
// Regex searches do not rely on embedding state. The root mtime gives
// their immutable query key a cheap revision when the file universe changes.
index_rev: directory_mtime_ns(&canonical)
.unwrap_or_default()
.to_string(),
}
}
fn directory_mtime_ns(path: &std::path::Path) -> Option<u128> {
std::fs::metadata(path)
.ok()?
.modified()
.ok()?
.duration_since(std::time::UNIX_EPOCH)
.ok()
.map(|duration| duration.as_nanos())
}
#[cfg(test)]
mod cache_delivery_tests {
use super::*;
#[test]
fn regex_adapter_records_then_serves_a_cross_agent_reference() {
let directory = tempfile::tempdir().unwrap();
std::fs::write(
directory.path().join("cached.rs"),
"fn cache_delivery_probe() {}\n",
)
.unwrap();
let path = directory.path().to_string_lossy();
let first = search_single(
"cache_delivery_probe",
&path,
Some("*.rs"),
20,
crate::tools::CrpMode::Off,
true,
true,
false,
None,
None,
)
.unwrap();
assert!(first.text.contains("cache_delivery_probe"));
let second = search_single(
"cache_delivery_probe",
&path,
Some("*.rs"),
20,
crate::tools::CrpMode::Off,
true,
true,
false,
None,
None,
)
.unwrap();
assert!(
second.text.contains("[cross-agent cache"),
"{}",
second.text
);
}
}
#[allow(clippy::too_many_arguments)]
fn search_single(
pattern: &str,
@@ -516,7 +670,7 @@ fn search_single(
let search_result = tokio::task::block_in_place(|| {
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
crate::tools::ctx_search::handle_filtered(
cached_or_search(
pattern,
path,
include,
@@ -639,7 +793,7 @@ fn handle_batch_queries(
let search_result = tokio::task::block_in_place(|| {
std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
crate::tools::ctx_search::handle_filtered(
cached_or_search(
&pattern,
root,
include.as_deref(),
+101 -1
View File
@@ -2,6 +2,8 @@ use rmcp::ErrorData;
use rmcp::model::Tool;
use serde_json::{Map, Value, json};
use crate::core::ocla::cache_types::{CacheKeyBuilder, ShellCommandKey};
use crate::server::tool_trait::{
McpTool, ShellOutcome, ToolContext, ToolOutput, get_bool, get_int, get_str,
};
@@ -260,6 +262,56 @@ impl McpTool for CtxShellTool {
})
.unwrap_or_default();
let inline = get_bool(args, "inline").unwrap_or(false);
let shell_cache_key = shell_cache_key(&cmd_clone, &cwd_clone, &extra_env);
if !raw
&& !inline
&& crate::core::config::Config::load()
.cache
.shell_cache_enabled
&& let Some(key) = shell_cache_key.as_ref()
&& let Some(cached) = crate::core::ocla::shell_cache_allowlist::SHELL_RESULT_CACHE
.get(&key.cache_key())
.map(|r| r.clone())
&& let Ok(value) = serde_json::from_str::<Value>(&cached)
&& let (Some(text), Some(exit_code)) = (
value.get("text").and_then(Value::as_str),
value.get("exit_code").and_then(Value::as_i64),
)
{
return Ok(ToolOutput {
text: text.to_string(),
original_tokens: crate::core::tokens::count_tokens(text),
saved_tokens: 0,
mode: Some("cross-agent-cache".to_string()),
path: None,
changed: false,
shell_outcome: Some(ShellOutcome::Exit(exit_code as i32)),
content_blocks: None,
});
}
// Cross-process delivery: check daemon for results from other IDE tabs
if let Some(ref key) = shell_cache_key {
let ck = key.cache_key();
let validator = key.validator();
if let Some(entry) =
crate::core::ocla::cache_delivery::check(&ck, &validator, "ctx_shell")
{
let stub = crate::core::ocla::cache_delivery::stub(&entry, "shell command");
return Ok(ToolOutput {
text: stub,
original_tokens: entry.token_count as usize,
saved_tokens: entry.token_count as usize,
mode: Some("cross-agent-cache".to_string()),
path: None,
changed: false,
shell_outcome: None,
content_blocks: None,
});
}
}
let auto_background = should_auto_background(&cmd_clone, timeout_ms);
if get_bool(args, "run_in_background").unwrap_or(false) || auto_background {
let job_id = crate::server::background_shell::start(
@@ -344,7 +396,6 @@ impl McpTool for CtxShellTool {
let output = redact_shell_output_secrets(&raw_output);
let inline = get_bool(args, "inline").unwrap_or(false);
let (result_out, original, saved, tee_hint) = if raw || inline {
let tokens = crate::core::tokens::count_tokens(&output);
(output, tokens, 0, String::new())
@@ -423,6 +474,27 @@ impl McpTool for CtxShellTool {
final_out
};
if !raw
&& !inline
&& crate::core::config::Config::load()
.cache
.shell_cache_enabled
&& let Some(key) = shell_cache_key
{
let cached = json!({ "text": final_out, "exit_code": exit_code }).to_string();
crate::core::ocla::shell_cache_allowlist::SHELL_RESULT_CACHE
.insert(key.cache_key(), cached);
// Propagate to cross-process daemon cache
crate::core::ocla::cache_delivery::record(
key.cache_key(),
crate::core::ocla::cache_types::DeliveryKind::ShellCommand,
key.validator(),
None,
&final_out,
"ctx_shell",
);
}
Ok(ToolOutput {
text: final_out,
original_tokens: original,
@@ -437,6 +509,34 @@ impl McpTool for CtxShellTool {
}
}
fn shell_cache_key(
command: &str,
cwd: &str,
env: &std::collections::HashMap<String, String>,
) -> Option<ShellCommandKey> {
if !crate::core::ocla::shell_cache_allowlist::is_cacheable_command(command) {
return None;
}
let mut env_pairs = env.iter().collect::<Vec<_>>();
env_pairs.sort_unstable_by(|left, right| left.0.cmp(right.0));
let mut canonical_env = String::new();
for (name, value) in env_pairs {
canonical_env.push_str(name);
canonical_env.push('=');
canonical_env.push_str(value);
canonical_env.push('\n');
}
Some(ShellCommandKey {
command_normalized: crate::core::ocla::shell_cache_allowlist::normalize_command(command),
cwd: if std::path::Path::new(cwd).is_absolute() {
"$PROJECT_ROOT".to_string()
} else {
cwd.to_string()
},
env_hash: blake3::hash(canonical_env.as_bytes()).to_hex().to_string(),
})
}
/// Deny shell execution for explicitly restricted MCP clients. Missing client
/// context remains allowed so existing integrations retain their current access.
fn shell_access_denial(ctx: &ToolContext) -> Option<String> {
+60 -40
View File
@@ -2,6 +2,7 @@ use rmcp::ErrorData;
use rmcp::model::Tool;
use serde_json::{Map, Value, json};
use crate::core::ocla::cache_types::{CacheKeyBuilder, DirectoryWalkKey};
use crate::server::tool_trait::{McpTool, ToolContext, ToolOutput, get_bool, get_int};
use crate::tool_defs::tool_def;
@@ -47,10 +48,6 @@ impl McpTool for CtxTreeTool {
let show_hidden = get_bool(args, "show_hidden").unwrap_or(false);
let respect_gitignore = get_bool(args, "respect_gitignore").unwrap_or(true);
if !resolved.is_multi {
return handle_single(&resolved.roots[0], depth, show_hidden, respect_gitignore);
}
let mut combined = String::new();
let mut total_original: usize = 0;
let mut total_sent: usize = 0;
@@ -59,12 +56,7 @@ impl McpTool for CtxTreeTool {
let root_clone = root.clone();
let Ok((result, original)) =
std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
crate::tools::ctx_tree::handle(
&root_clone,
depth,
show_hidden,
respect_gitignore,
)
cached_or_walk(&root_clone, depth, show_hidden, respect_gitignore)
}))
else {
combined.push_str(&format!("── {root} ──\nERROR: internal panic\n\n"));
@@ -98,40 +90,68 @@ impl McpTool for CtxTreeTool {
}
}
fn handle_single(
fn directory_mtime_ns(path: &std::path::Path) -> Option<u128> {
std::fs::metadata(path)
.ok()?
.modified()
.ok()?
.duration_since(std::time::UNIX_EPOCH)
.ok()
.map(|duration| duration.as_nanos())
}
fn cached_or_walk(
path: &str,
depth: usize,
show_hidden: bool,
respect_gitignore: bool,
) -> Result<ToolOutput, ErrorData> {
let path_clone = path.to_string();
let Ok((result, original)) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
crate::tools::ctx_tree::handle(&path_clone, depth, show_hidden, respect_gitignore)
})) else {
return Err(ErrorData::internal_error(
format!(
"ctx_tree panicked while processing '{path}'. This is a bug — please report it."
),
None,
));
) -> (String, usize) {
let builder = DirectoryWalkKey {
path: crate::core::pathutil::safe_canonicalize_or_self(std::path::Path::new(path))
.to_string_lossy()
.into_owned(),
depth,
gitignore: respect_gitignore,
dir_mtime_ns: directory_mtime_ns(&crate::core::pathutil::safe_canonicalize_or_self(
std::path::Path::new(path),
))
.unwrap_or_default(),
};
if result.starts_with("ERROR:") {
return Err(ErrorData::invalid_params(result, None));
let key = builder.cache_key();
if let Some(entry) =
crate::core::ocla::cache_delivery::check(&key, &builder.validator(), "ctx_tree")
{
let stub = crate::core::ocla::cache_delivery::stub(&entry, "directory tree");
return (stub, entry.token_count as usize);
}
let (result, original) =
crate::tools::ctx_tree::handle(path, depth, show_hidden, respect_gitignore);
if !result.starts_with("ERROR:") {
crate::core::ocla::cache_delivery::record(
key,
crate::core::ocla::cache_types::DeliveryKind::DirectoryWalk,
builder.validator(),
Some(builder.path),
&result,
"ctx_tree",
);
}
(result, original)
}
#[cfg(test)]
mod tests {
use super::cached_or_walk;
#[test]
fn tree_adapter_records_then_serves_a_cross_agent_reference() {
let directory = tempfile::tempdir().unwrap();
std::fs::write(directory.path().join("cached.rs"), "fn cached() {}\n").unwrap();
let path = directory.path().to_string_lossy();
let (first_result, _first_orig) = cached_or_walk(&path, 3, false, true);
assert!(first_result.contains("cached.rs"));
let (second_result, _second_orig) = cached_or_walk(&path, 3, false, true);
assert!(second_result.contains("[cross-agent"), "{}", second_result);
}
let sent = crate::core::tokens::count_tokens(&result);
let saved = original.saturating_sub(sent);
let final_out = crate::core::protocol::append_savings(&result, original, sent);
Ok(ToolOutput {
text: final_out,
original_tokens: original,
saved_tokens: saved,
mode: None,
path: Some(path.to_string()),
changed: false,
shell_outcome: None,
content_blocks: None,
})
}