1use std::collections::BTreeMap;
24use std::fmt::Write as _;
25use std::path::{Path, PathBuf};
26
27use serde::{Deserialize, Serialize};
28
29pub const AGENT_LEASE_SCHEMA_VERSION: u16 = 1;
30const DEFAULT_MAX_LEASES: usize = 1_024;
31const MAX_LEASE_DURATION_MS: u64 = 60 * 60 * 1_000;
32const MAX_RESOURCE_REF_BYTES: usize = 256;
33const PROJECT_PATH_SCHEME: &str = "project-path:";
36const HIERARCHICAL_PATH_SCHEMES: [&str; 2] = [PROJECT_PATH_SCHEME, "path:"];
37
38#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd, Serialize, Deserialize)]
39#[serde(rename_all = "snake_case")]
40pub enum AgentLeaseResourceKindV1 {
41 Path,
42 Symbol,
43 WorkGraphNode,
45}
46
47#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
48#[serde(deny_unknown_fields)]
49pub struct AgentLeaseRequestV1 {
50 pub schema_version: u16,
51 pub lease_request_ref: String,
52 pub resource_kind: AgentLeaseResourceKindV1,
53 pub resource_ref: String,
54 pub owner_agent_id: String,
55 pub duration_ms: u64,
56}
57
58impl AgentLeaseRequestV1 {
59 pub fn validate(&self) -> Result<(), AgentLeaseError> {
60 if self.schema_version != AGENT_LEASE_SCHEMA_VERSION {
61 return Err(AgentLeaseError::UnsupportedVersion(self.schema_version));
62 }
63 opaque_ref("lease_request_ref", &self.lease_request_ref)?;
64 opaque_ref("resource_ref", &self.resource_ref)?;
65 agent_id(&self.owner_agent_id)?;
66 if self.duration_ms == 0 || self.duration_ms > MAX_LEASE_DURATION_MS {
67 return Err(AgentLeaseError::Invalid(format!(
68 "duration_ms must be between 1 and {MAX_LEASE_DURATION_MS}"
69 )));
70 }
71 Ok(())
72 }
73}
74
75#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
76#[serde(deny_unknown_fields)]
77pub struct AgentLeaseV1 {
78 pub schema_version: u16,
79 pub lease_ref: String,
80 pub request: AgentLeaseRequestV1,
81 pub expires_at_epoch_ms: u64,
82}
83
84impl AgentLeaseV1 {
85 pub fn is_active_at(&self, now_epoch_ms: u64) -> bool {
86 now_epoch_ms < self.expires_at_epoch_ms
87 }
88}
89
90#[derive(Clone, Debug, Eq, PartialEq)]
91pub enum AgentLeaseAcquireV1 {
92 Granted(AgentLeaseV1),
93 HeldBy {
94 owner_agent_id: String,
95 lease_ref: String,
96 expires_at_epoch_ms: u64,
97 },
98}
99
100const WAITER_TTL_MS: u64 = 2 * 60 * 1_000;
102const MAX_CONTENDED_HOLD_MS: u64 = 60 * 60 * 1_000;
106const RESERVED_LEASE_REF: &str = "reserved-for-waiter";
108
109#[derive(Clone, Debug, Default, Serialize, Deserialize)]
113struct LeaseFairnessV1 {
114 #[serde(default)]
115 waiters: Vec<LeaseWaiterV1>,
116 #[serde(default)]
118 held_since: BTreeMap<String, u64>,
119}
120
121#[derive(Clone, Debug, Serialize, Deserialize)]
122struct LeaseWaiterV1 {
123 resource_kind: AgentLeaseResourceKindV1,
124 resource_ref: String,
125 owner_agent_id: String,
126 since_epoch_ms: u64,
127 seen_epoch_ms: u64,
128}
129
130fn resources_overlap(
131 kind: AgentLeaseResourceKindV1,
132 held: &str,
133 other_kind: AgentLeaseResourceKindV1,
134 other: &str,
135) -> bool {
136 kind == other_kind
137 && (held == other
138 || (kind == AgentLeaseResourceKindV1::Path
139 && HIERARCHICAL_PATH_SCHEMES.into_iter().any(|scheme| {
140 held.starts_with(scheme)
141 && other.starts_with(scheme)
142 && crate::core::work_graph::path_claims_overlap(held, other)
143 })))
144}
145
146pub struct AgentLeaseRegistryV1 {
148 leases: BTreeMap<(AgentLeaseResourceKindV1, String), AgentLeaseV1>,
149 max_leases: usize,
150 fairness: LeaseFairnessV1,
151}
152
153impl Default for AgentLeaseRegistryV1 {
154 fn default() -> Self {
155 Self::new(DEFAULT_MAX_LEASES)
156 }
157}
158
159impl AgentLeaseRegistryV1 {
160 #[must_use]
161 pub fn new(max_leases: usize) -> Self {
162 Self {
163 leases: BTreeMap::new(),
164 max_leases: max_leases.max(1),
165 fairness: LeaseFairnessV1::default(),
166 }
167 }
168
169 pub fn acquire(
176 &mut self,
177 request: AgentLeaseRequestV1,
178 now_epoch_ms: u64,
179 ) -> Result<AgentLeaseAcquireV1, AgentLeaseError> {
180 request.validate()?;
181 self.remove_expired(now_epoch_ms);
182 let acquired = self.acquire_unfair(&request, now_epoch_ms)?;
183 Ok(match acquired {
184 AgentLeaseAcquireV1::Granted(lease) => {
185 let owner = &request.owner_agent_id;
186 self.fairness.waiters.retain(|waiter| {
187 &waiter.owner_agent_id != owner
188 || waiter.resource_kind != request.resource_kind
189 || waiter.resource_ref != request.resource_ref
190 });
191 self.fairness
192 .held_since
193 .entry(lease.lease_ref.clone())
194 .or_insert(now_epoch_ms);
195 AgentLeaseAcquireV1::Granted(lease)
196 }
197 held @ AgentLeaseAcquireV1::HeldBy { .. } => {
198 self.wait(request, now_epoch_ms);
199 held
200 }
201 })
202 }
203
204 fn wait(&mut self, request: AgentLeaseRequestV1, now_epoch_ms: u64) {
205 let waiters = &mut self.fairness.waiters;
206 let known = waiters.iter().position(|waiter| {
207 waiter.owner_agent_id == request.owner_agent_id
208 && waiter.resource_kind == request.resource_kind
209 && waiter.resource_ref == request.resource_ref
210 });
211 match known {
212 Some(index) => waiters[index].seen_epoch_ms = now_epoch_ms,
213 None if waiters.len() < self.max_leases => waiters.push(LeaseWaiterV1 {
216 resource_kind: request.resource_kind,
217 resource_ref: request.resource_ref,
218 owner_agent_id: request.owner_agent_id,
219 since_epoch_ms: now_epoch_ms,
220 seen_epoch_ms: now_epoch_ms,
221 }),
222 None => {}
223 }
224 }
225
226 fn waiting_before(&self, request: &AgentLeaseRequestV1) -> Option<&LeaseWaiterV1> {
228 let own_since = self
229 .fairness
230 .waiters
231 .iter()
232 .filter(|waiter| {
233 waiter.owner_agent_id == request.owner_agent_id
234 && waiter.resource_kind == request.resource_kind
235 && waiter.resource_ref == request.resource_ref
236 })
237 .map(|waiter| waiter.since_epoch_ms)
238 .min()
239 .unwrap_or(u64::MAX);
240 self.fairness
241 .waiters
242 .iter()
243 .filter(|waiter| {
244 waiter.owner_agent_id != request.owner_agent_id
245 && waiter.since_epoch_ms < own_since
246 && resources_overlap(
247 waiter.resource_kind,
248 &waiter.resource_ref,
249 request.resource_kind,
250 &request.resource_ref,
251 )
252 })
253 .min_by_key(|waiter| waiter.since_epoch_ms)
254 }
255
256 fn acquire_unfair(
257 &mut self,
258 request: &AgentLeaseRequestV1,
259 now_epoch_ms: u64,
260 ) -> Result<AgentLeaseAcquireV1, AgentLeaseError> {
261 let request = request.clone();
262 let key = (request.resource_kind, request.resource_ref.clone());
263 if let Some(existing) = self.leases.get(&key) {
264 if existing.request.owner_agent_id == request.owner_agent_id
265 && existing.request.lease_request_ref == request.lease_request_ref
266 {
267 return Ok(AgentLeaseAcquireV1::Granted(existing.clone()));
268 }
269 return Ok(AgentLeaseAcquireV1::HeldBy {
270 owner_agent_id: existing.request.owner_agent_id.clone(),
271 lease_ref: existing.lease_ref.clone(),
272 expires_at_epoch_ms: existing.expires_at_epoch_ms,
273 });
274 }
275 if request.resource_kind == AgentLeaseResourceKindV1::Path
279 && let Some(scheme) = HIERARCHICAL_PATH_SCHEMES
280 .into_iter()
281 .find(|scheme| request.resource_ref.starts_with(scheme))
282 && let Some(existing) = self.leases.values().find(|lease| {
283 lease.request.resource_kind == AgentLeaseResourceKindV1::Path
284 && lease.request.resource_ref.starts_with(scheme)
285 && crate::core::work_graph::path_claims_overlap(
286 &lease.request.resource_ref,
287 &request.resource_ref,
288 )
289 })
290 {
291 return Ok(AgentLeaseAcquireV1::HeldBy {
292 owner_agent_id: existing.request.owner_agent_id.clone(),
293 lease_ref: existing.lease_ref.clone(),
294 expires_at_epoch_ms: existing.expires_at_epoch_ms,
295 });
296 }
297 if let Some(waiter) = self.waiting_before(&request) {
298 return Ok(AgentLeaseAcquireV1::HeldBy {
299 owner_agent_id: waiter.owner_agent_id.clone(),
300 lease_ref: RESERVED_LEASE_REF.to_string(),
301 expires_at_epoch_ms: waiter.seen_epoch_ms.saturating_add(WAITER_TTL_MS),
302 });
303 }
304 if self.leases.len() >= self.max_leases {
305 return Err(AgentLeaseError::CapacityExceeded(self.max_leases));
306 }
307 let expires_at_epoch_ms = now_epoch_ms.saturating_add(request.duration_ms);
308 let lease_ref = compute_lease_ref(&request, expires_at_epoch_ms)?;
309 let lease = AgentLeaseV1 {
310 schema_version: AGENT_LEASE_SCHEMA_VERSION,
311 lease_ref,
312 request,
313 expires_at_epoch_ms,
314 };
315 self.leases.insert(key, lease.clone());
316 Ok(AgentLeaseAcquireV1::Granted(lease))
317 }
318
319 pub fn release(
320 &mut self,
321 resource_kind: AgentLeaseResourceKindV1,
322 resource_ref: &str,
323 owner_agent_id: &str,
324 lease_ref: &str,
325 now_epoch_ms: u64,
326 ) -> Result<bool, AgentLeaseError> {
327 opaque_ref("resource_ref", resource_ref)?;
328 agent_id(owner_agent_id)?;
329 self.remove_expired(now_epoch_ms);
330 let key = (resource_kind, resource_ref.to_string());
331 let Some(existing) = self.leases.get(&key) else {
332 return Ok(false);
333 };
334 if existing.request.owner_agent_id != owner_agent_id || existing.lease_ref != lease_ref {
335 return Err(AgentLeaseError::NotOwner);
336 }
337 self.leases.remove(&key);
338 Ok(true)
339 }
340
341 pub fn release_fence(
345 &mut self,
346 owner_agent_id: &str,
347 fence: &str,
348 now_epoch_ms: u64,
349 ) -> Result<usize, AgentLeaseError> {
350 opaque_ref("lease_request_ref", fence)?;
351 agent_id(owner_agent_id)?;
352 self.remove_expired(now_epoch_ms);
353 let path_prefix = format!("{fence}:");
354 let before = self.leases.len();
355 self.leases.retain(|_, lease| {
356 let request = &lease.request;
357 request.owner_agent_id != owner_agent_id
358 || (request.lease_request_ref != fence
359 && !request.lease_request_ref.starts_with(&path_prefix))
360 });
361 Ok(before - self.leases.len())
362 }
363
364 pub fn renew(
371 &mut self,
372 resource_kind: AgentLeaseResourceKindV1,
373 resource_ref: &str,
374 owner_agent_id: &str,
375 lease_ref: &str,
376 duration_ms: u64,
377 now_epoch_ms: u64,
378 ) -> Result<AgentLeaseV1, AgentLeaseError> {
379 opaque_ref("resource_ref", resource_ref)?;
380 agent_id(owner_agent_id)?;
381 if duration_ms == 0 || duration_ms > MAX_LEASE_DURATION_MS {
382 return Err(AgentLeaseError::Invalid(format!(
383 "duration_ms must be between 1 and {MAX_LEASE_DURATION_MS}"
384 )));
385 }
386 let key = (resource_kind, resource_ref.to_string());
387 let Some(existing) = self.leases.get(&key) else {
388 return Err(AgentLeaseError::NotFound);
389 };
390 if !existing.is_active_at(now_epoch_ms) {
391 self.leases.remove(&key);
392 return Err(AgentLeaseError::Expired);
393 }
394 if existing.request.owner_agent_id != owner_agent_id || existing.lease_ref != lease_ref {
395 return Err(AgentLeaseError::NotOwner);
396 }
397 self.remove_expired(now_epoch_ms);
398 let held_since = self.fairness.held_since.get(lease_ref).copied();
399 if held_since
400 .is_some_and(|since| now_epoch_ms.saturating_sub(since) >= MAX_CONTENDED_HOLD_MS)
401 && self
402 .waiting_before(&self.leases[&key].request.clone())
403 .is_some()
404 {
405 return Err(AgentLeaseError::Invalid(
406 "another agent is waiting and this lease reached its maximum hold; \
407 it is not renewed and goes to the waiter when it runs out"
408 .into(),
409 ));
410 }
411 let existing = self.leases.get_mut(&key).expect("lease checked above");
412 existing.request.duration_ms = duration_ms;
413 existing.expires_at_epoch_ms = now_epoch_ms.saturating_add(duration_ms);
414 Ok(existing.clone())
415 }
416
417 #[must_use]
418 pub fn active_count(&self, now_epoch_ms: u64) -> usize {
419 self.leases
420 .values()
421 .filter(|l| l.is_active_at(now_epoch_ms))
422 .count()
423 }
424
425 fn remove_expired(&mut self, now_epoch_ms: u64) {
426 self.leases.retain(|_, l| l.is_active_at(now_epoch_ms));
427 let held: std::collections::BTreeSet<&str> =
428 self.leases.values().map(|l| l.lease_ref.as_str()).collect();
429 self.fairness
430 .held_since
431 .retain(|lease_ref, _| held.contains(lease_ref.as_str()));
432 self.fairness
433 .waiters
434 .retain(|waiter| now_epoch_ms < waiter.seen_epoch_ms.saturating_add(WAITER_TTL_MS));
435 }
436}
437
438#[derive(Serialize, Deserialize)]
443#[serde(deny_unknown_fields)]
444struct AgentLeaseFileV1 {
445 schema_version: u16,
446 leases: Vec<AgentLeaseV1>,
447}
448
449fn now_epoch_ms() -> u64 {
450 std::time::SystemTime::now()
451 .duration_since(std::time::UNIX_EPOCH)
452 .unwrap_or_default()
453 .as_millis() as u64
454}
455
456fn shared_dir() -> Result<PathBuf, AgentLeaseError> {
457 crate::core::data_dir::lean_ctx_data_dir()
458 .map(|dir| dir.join("agents"))
459 .map_err(AgentLeaseError::Store)
460}
461
462fn load_registry(path: &Path) -> Result<AgentLeaseRegistryV1, AgentLeaseError> {
463 let mut registry = AgentLeaseRegistryV1::default();
464 let content = match std::fs::read_to_string(path) {
465 Ok(content) => content,
466 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(registry),
467 Err(error) => {
468 return Err(AgentLeaseError::Store(format!(
469 "read {}: {error}",
470 path.display()
471 )));
472 }
473 };
474 let file: AgentLeaseFileV1 = serde_json::from_str(&content).map_err(|error| {
477 AgentLeaseError::Store(format!(
478 "lease store is corrupt at {}: {error}",
479 path.display()
480 ))
481 })?;
482 if file.schema_version != AGENT_LEASE_SCHEMA_VERSION {
483 return Err(AgentLeaseError::UnsupportedVersion(file.schema_version));
484 }
485 for lease in file.leases {
486 let key = (
487 lease.request.resource_kind,
488 lease.request.resource_ref.clone(),
489 );
490 registry.leases.insert(key, lease);
491 }
492 registry.fairness = std::fs::read_to_string(fairness_path(path))
495 .ok()
496 .and_then(|json| serde_json::from_str(&json).ok())
497 .unwrap_or_default();
498 Ok(registry)
499}
500
501fn fairness_path(lease_store: &Path) -> PathBuf {
502 lease_store.with_file_name("lease-waiters.json")
503}
504
505fn save_registry(path: &Path, registry: &AgentLeaseRegistryV1) -> Result<(), AgentLeaseError> {
506 let file = AgentLeaseFileV1 {
507 schema_version: AGENT_LEASE_SCHEMA_VERSION,
508 leases: registry.leases.values().cloned().collect(),
509 };
510 let json = serde_json::to_string_pretty(&file)
511 .map_err(|error| AgentLeaseError::Serialize(error.to_string()))?;
512 crate::config_io::write_atomic(path, &json)
513 .map_err(|error| AgentLeaseError::Store(format!("persist {}: {error}", path.display())))?;
514 let fairness = serde_json::to_string(®istry.fairness)
515 .map_err(|error| AgentLeaseError::Serialize(error.to_string()))?;
516 let fairness_path = fairness_path(path);
517 crate::config_io::write_atomic(&fairness_path, &fairness).map_err(|error| {
518 AgentLeaseError::Store(format!("persist {}: {error}", fairness_path.display()))
519 })
520}
521
522fn with_shared_registry<T>(
525 mutate: impl FnOnce(&mut AgentLeaseRegistryV1, u64) -> Result<T, AgentLeaseError>,
526) -> Result<T, AgentLeaseError> {
527 let dir = shared_dir()?;
528 std::fs::create_dir_all(&dir).map_err(|error| {
529 AgentLeaseError::Store(format!("create lease directory {}: {error}", dir.display()))
530 })?;
531 let _lock = crate::core::agents::FileLock::acquire(&dir.join("leases.lock"))
532 .map_err(AgentLeaseError::Store)?;
533 let path = dir.join("leases.json");
534 let mut registry = load_registry(&path)?;
535 let result = mutate(&mut registry, now_epoch_ms())?;
536 save_registry(&path, ®istry)?;
537 Ok(result)
538}
539
540pub fn acquire_shared(
542 request: AgentLeaseRequestV1,
543) -> Result<AgentLeaseAcquireV1, AgentLeaseError> {
544 with_shared_registry(|registry, now| registry.acquire(request, now))
545}
546
547pub fn active_shared() -> Result<Vec<AgentLeaseV1>, AgentLeaseError> {
550 let registry = load_registry(&shared_dir()?.join("leases.json"))?;
551 let now = now_epoch_ms();
552 let mut leases: Vec<AgentLeaseV1> = registry
553 .leases
554 .into_values()
555 .filter(|lease| lease.is_active_at(now))
556 .collect();
557 leases.sort_by(|a, b| {
558 a.expires_at_epoch_ms
559 .cmp(&b.expires_at_epoch_ms)
560 .then_with(|| a.lease_ref.cmp(&b.lease_ref))
561 });
562 Ok(leases)
563}
564
565pub fn project_path_of(project_root: &str, resource_ref: &str) -> Option<String> {
569 let scope = crate::core::project_hash::hash_project_root(project_root);
570 resource_ref
571 .strip_prefix(PROJECT_PATH_SCHEME)?
572 .strip_prefix(&scope)?
573 .strip_prefix('/')
574 .map(str::to_owned)
575}
576
577pub fn release_shared(
579 resource_kind: AgentLeaseResourceKindV1,
580 resource_ref: &str,
581 owner_agent_id: &str,
582 lease_ref: &str,
583) -> Result<bool, AgentLeaseError> {
584 with_shared_registry(|registry, now| {
585 registry.release(resource_kind, resource_ref, owner_agent_id, lease_ref, now)
586 })
587}
588
589pub fn revoke_shared(lease_ref: &str) -> Result<Option<AgentLeaseV1>, AgentLeaseError> {
593 with_shared_registry(|registry, _| {
594 let key = registry
595 .leases
596 .iter()
597 .find(|(_, lease)| lease.lease_ref == lease_ref)
598 .map(|(key, _)| key.clone());
599 Ok(key.and_then(|key| registry.leases.remove(&key)))
600 })
601}
602
603pub fn release_fence_shared(owner_agent_id: &str, fence: &str) -> Result<usize, AgentLeaseError> {
606 with_shared_registry(|registry, now| registry.release_fence(owner_agent_id, fence, now))
607}
608
609pub fn renew_shared(
613 resource_kind: AgentLeaseResourceKindV1,
614 resource_ref: &str,
615 owner_agent_id: &str,
616 lease_ref: &str,
617 duration_ms: u64,
618) -> Result<AgentLeaseV1, AgentLeaseError> {
619 with_shared_registry(|registry, now| {
620 registry.renew(
621 resource_kind,
622 resource_ref,
623 owner_agent_id,
624 lease_ref,
625 duration_ms,
626 now,
627 )
628 })
629}
630
631const CASE_INSENSITIVE_PATHS: bool = cfg!(any(target_os = "macos", windows));
637
638pub fn project_path_resource(
642 project_root: &str,
643 relative: &str,
644) -> Result<String, AgentLeaseError> {
645 let mut relative = relative.replace('\\', "/");
646 if CASE_INSENSITIVE_PATHS {
647 relative = relative.to_lowercase();
648 }
649 if relative.is_empty()
650 || relative.starts_with('/')
651 || relative
652 .split('/')
653 .any(|part| part.is_empty() || part == "." || part == "..")
654 || relative.chars().any(char::is_control)
655 {
656 return Err(AgentLeaseError::Invalid(
657 "lease path must be a normalized path inside the project".into(),
658 ));
659 }
660 let scope = crate::core::project_hash::hash_project_root(project_root);
661 let mut resource = format!("{PROJECT_PATH_SCHEME}{scope}/");
662 for byte in relative.bytes() {
663 if byte.is_ascii_alphanumeric() || matches!(byte, b'/' | b'.' | b'-' | b'_') {
664 resource.push(byte as char);
665 } else {
666 let _ = write!(resource, "%{byte:02X}");
667 }
668 }
669 if resource.len() > MAX_RESOURCE_REF_BYTES {
672 resource = format!(
673 "project-pathref:{}",
674 scoped_digest(&scope, relative.as_bytes())
675 );
676 }
677 Ok(resource)
678}
679
680pub fn project_path_target_resource(
686 project_root: &str,
687 target: &str,
688) -> Result<String, AgentLeaseError> {
689 let invalid = |reason: &str| AgentLeaseError::Invalid(reason.into());
690 let target = crate::core::pathutil::normalize_tool_path_lexical(target);
693 let root = crate::core::pathutil::normalize_tool_path_lexical(project_root);
694 let canonical_root = crate::core::pathutil::safe_canonicalize_or_self(Path::new(project_root));
695 let canonical_root_text =
696 crate::core::pathutil::normalize_tool_path_lexical(&canonical_root.to_string_lossy());
697 let relative = if Path::new(&target).is_absolute() || target.starts_with('/') {
698 [root.as_str(), canonical_root_text.as_str()]
699 .into_iter()
700 .find_map(|root| strip_root(&target, root))
701 .ok_or_else(|| invalid("lease path must be inside the project root"))?
702 } else {
703 target.as_str()
704 };
705 let mut parts = Vec::new();
706 for part in relative.split('/') {
707 match part {
708 "" | "." => {}
709 ".." => return Err(invalid("lease path must not leave the project root")),
710 part => parts.push(part),
711 }
712 }
713 if parts.is_empty() {
714 return Err(invalid(
715 "the project root itself cannot be leased; lease a path inside it",
716 ));
717 }
718 let mut cursor = canonical_root;
719 for part in &parts {
720 cursor.push(part);
721 if std::fs::symlink_metadata(&cursor).is_ok_and(|meta| meta.file_type().is_symlink()) {
722 return Err(invalid("lease path traverses a symlink"));
723 }
724 }
725 project_path_resource(project_root, &parts.join("/"))
726}
727
728fn strip_root<'a>(target: &'a str, root: &str) -> Option<&'a str> {
731 let root = root.trim_end_matches('/');
732 let head = target.get(..root.len())?;
733 let same = if CASE_INSENSITIVE_PATHS {
734 head.eq_ignore_ascii_case(root)
735 } else {
736 head == root
737 };
738 let rest = &target[root.len()..];
739 (same && (rest.is_empty() || rest.starts_with('/'))).then_some(rest)
740}
741
742pub fn project_symbol_resource(project_root: &str, symbol: &str) -> String {
744 let scope = crate::core::project_hash::hash_project_root(project_root);
745 format!(
746 "project-symbolref:{}",
747 scoped_digest(&scope, symbol.as_bytes())
748 )
749}
750
751fn scoped_digest(scope: &str, value: &[u8]) -> String {
752 let mut hash = blake3::Hasher::new();
753 hash.update(scope.as_bytes());
754 hash.update(b"\0");
755 hash.update(value);
756 hash.finalize().to_hex().to_string()
757}
758
759fn compute_lease_ref(
762 request: &AgentLeaseRequestV1,
763 expires_at_epoch_ms: u64,
764) -> Result<String, AgentLeaseError> {
765 let bytes = serde_json::to_vec(&(request, expires_at_epoch_ms))
766 .map_err(|e| AgentLeaseError::Serialize(e.to_string()))?;
767 Ok(format!("lease:{}", blake3::hash(&bytes).to_hex()))
768}
769
770fn opaque_ref(label: &str, value: &str) -> Result<(), AgentLeaseError> {
771 let (scheme, identifier) = value.split_once(':').ok_or_else(|| {
772 AgentLeaseError::Invalid(format!("{label} must use scheme:identifier form"))
773 })?;
774 let scheme_valid = !scheme.is_empty()
775 && scheme
776 .bytes()
777 .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-');
778 let identifier_valid = !identifier.is_empty()
779 && value.len() <= MAX_RESOURCE_REF_BYTES
780 && identifier.bytes().all(|b| b.is_ascii_graphic());
781 (scheme_valid && identifier_valid)
782 .then_some(())
783 .ok_or_else(|| AgentLeaseError::Invalid(format!("invalid {label}")))
784}
785
786fn agent_id(value: &str) -> Result<(), AgentLeaseError> {
787 (!value.is_empty() && value.len() <= 256 && value.bytes().all(|b| b.is_ascii_graphic()))
788 .then_some(())
789 .ok_or_else(|| AgentLeaseError::Invalid("invalid owner_agent_id".into()))
790}
791
792#[derive(Debug, thiserror::Error)]
795pub enum AgentLeaseError {
796 #[error("unsupported schema version {0}")]
797 UnsupportedVersion(u16),
798 #[error("invalid lease: {0}")]
799 Invalid(String),
800 #[error("registry at capacity {0}")]
801 CapacityExceeded(usize),
802 #[error("not owner or mismatched lease_ref")]
803 NotOwner,
804 #[error("lease not found")]
805 NotFound,
806 #[error("lease expired")]
807 Expired,
808 #[error("serialization failed: {0}")]
809 Serialize(String),
810 #[error("lease store: {0}")]
811 Store(String),
812}
813
814#[cfg(test)]
817pub mod tests {
818 use super::*;
819
820 fn request(owner: &str, ref_id: &str) -> AgentLeaseRequestV1 {
821 AgentLeaseRequestV1 {
822 schema_version: AGENT_LEASE_SCHEMA_VERSION,
823 lease_request_ref: ref_id.to_string(),
824 resource_kind: AgentLeaseResourceKindV1::Path,
825 resource_ref: "pathref:src-core-main".to_string(),
826 owner_agent_id: owner.to_string(),
827 duration_ms: 100,
828 }
829 }
830
831 #[test]
832 fn idempotent_grant_blocks_foreign_owner() {
833 let mut reg = AgentLeaseRegistryV1::default();
834 let AgentLeaseAcquireV1::Granted(granted) =
835 reg.acquire(request("agent-a", "request:a"), 10).unwrap()
836 else {
837 panic!("expected grant")
838 };
839 assert_eq!(
840 reg.acquire(request("agent-a", "request:a"), 20).unwrap(),
841 AgentLeaseAcquireV1::Granted(granted)
842 );
843 assert!(matches!(
844 reg.acquire(request("agent-b", "request:b"), 20),
845 Ok(AgentLeaseAcquireV1::HeldBy { owner_agent_id, .. }) if owner_agent_id == "agent-a"
846 ));
847 }
848
849 #[test]
850 fn hierarchical_paths_block_both_orders_without_prefix_false_positives() {
851 for (held, incoming) in [
852 ("path:project/src", "path:project/src/main.rs"),
853 ("path:project/src/main.rs", "path:project/src"),
854 ] {
855 let mut reg = AgentLeaseRegistryV1::default();
856 let mut first = request("agent-a", "request:a");
857 first.resource_ref = held.into();
858 let granted = reg.acquire(first.clone(), 1).unwrap();
859 assert_eq!(reg.acquire(first, 2).unwrap(), granted);
860 for owner in ["agent-a", "agent-b"] {
861 let mut competing = request(owner, "request:b");
862 competing.resource_ref = incoming.into();
863 assert!(matches!(
864 reg.acquire(competing, 2).unwrap(),
865 AgentLeaseAcquireV1::HeldBy { .. }
866 ));
867 }
868 for independent in ["path:project/src-other", "path:other-project/src"] {
869 let mut next = request("agent-b", "request:c");
870 next.resource_ref = independent.into();
871 assert!(matches!(
872 reg.acquire(next, 2).unwrap(),
873 AgentLeaseAcquireV1::Granted(_)
874 ));
875 }
876 }
877 }
878
879 #[test]
880 fn expiry_frees_resource() {
881 let mut reg = AgentLeaseRegistryV1::default();
882 let _ = reg.acquire(request("agent-a", "request:a"), 10).unwrap();
883 assert!(matches!(
884 reg.acquire(request("agent-b", "request:b"), 111),
885 Ok(AgentLeaseAcquireV1::Granted(_))
886 ));
887 }
888
889 #[test]
890 fn release_requires_owner() {
891 let mut reg = AgentLeaseRegistryV1::default();
892 let AgentLeaseAcquireV1::Granted(granted) =
893 reg.acquire(request("agent-a", "request:a"), 10).unwrap()
894 else {
895 panic!("expected grant")
896 };
897 assert!(
898 reg.release(
899 AgentLeaseResourceKindV1::Path,
900 "pathref:src-core-main",
901 "agent-b",
902 &granted.lease_ref,
903 20
904 )
905 .is_err()
906 );
907 assert!(
908 reg.release(
909 AgentLeaseResourceKindV1::Path,
910 "pathref:src-core-main",
911 "agent-a",
912 &granted.lease_ref,
913 20
914 )
915 .unwrap()
916 );
917 }
918
919 #[test]
920 fn incarnation_release_never_frees_a_newer_holder() {
921 let mut reg = AgentLeaseRegistryV1::default();
922 let _ = reg.acquire(request("agent-a", "fence:old"), 10).unwrap();
923 let path_lease =
924 |owner: &str, lease_request_ref: &str, resource: &str| AgentLeaseRequestV1 {
925 resource_ref: resource.into(),
926 ..request(owner, lease_request_ref)
927 };
928 let _ = reg
929 .acquire(path_lease("agent-a", "fence:old:0", "pathref:docs"), 10)
930 .unwrap();
931 let _ = reg
932 .acquire(path_lease("agent-a", "fence:older", "pathref:other"), 10)
933 .unwrap();
934 let release = |reg: &mut AgentLeaseRegistryV1, owner: &str, fence: &str| {
935 reg.release_fence(owner, fence, 20).unwrap()
936 };
937 assert_eq!(release(&mut reg, "agent-b", "fence:old"), 0);
938 assert_eq!(release(&mut reg, "agent-a", "fence:new"), 0);
939 assert_eq!(release(&mut reg, "agent-a", "fence:old"), 2);
941 assert_eq!(reg.active_count(20), 1);
942 let _ = reg.acquire(request("agent-a", "fence:new"), 30).unwrap();
943 assert_eq!(release(&mut reg, "agent-a", "fence:old"), 0);
944 assert!(matches!(
945 reg.acquire(request("agent-b", "request:b"), 40),
946 Ok(AgentLeaseAcquireV1::HeldBy { .. })
947 ));
948 }
949
950 #[test]
951 fn renew_extends_owned_lease_without_rotating_fence() {
952 let mut reg = AgentLeaseRegistryV1::default();
953 let AgentLeaseAcquireV1::Granted(granted) =
954 reg.acquire(request("agent-a", "request:a"), 10).unwrap()
955 else {
956 panic!("expected grant")
957 };
958 let renewed = reg
959 .renew(
960 AgentLeaseResourceKindV1::Path,
961 "pathref:src-core-main",
962 "agent-a",
963 &granted.lease_ref,
964 500,
965 50,
966 )
967 .unwrap();
968 assert_eq!(renewed.lease_ref, granted.lease_ref);
969 assert_eq!(renewed.expires_at_epoch_ms, 550);
970 assert!(renewed.is_active_at(549));
971 }
972
973 #[test]
974 fn renew_rejects_expired_token_and_allows_reclaim() {
975 let mut reg = AgentLeaseRegistryV1::default();
976 let AgentLeaseAcquireV1::Granted(granted) =
977 reg.acquire(request("agent-a", "request:a"), 10).unwrap()
978 else {
979 panic!("expected grant")
980 };
981 assert!(matches!(
982 reg.renew(
983 AgentLeaseResourceKindV1::Path,
984 "pathref:src-core-main",
985 "agent-a",
986 &granted.lease_ref,
987 500,
988 110,
989 ),
990 Err(AgentLeaseError::Expired)
991 ));
992 assert!(matches!(
993 reg.acquire(request("agent-b", "request:b"), 111),
994 Ok(AgentLeaseAcquireV1::Granted(_))
995 ));
996 }
997
998 #[test]
999 fn capacity_enforced() {
1000 let mut reg = AgentLeaseRegistryV1::new(1);
1001 let _ = reg.acquire(request("agent-a", "request:a"), 1).unwrap();
1002 let second = AgentLeaseRequestV1 {
1003 resource_ref: "symbolref:main".to_string(),
1004 resource_kind: AgentLeaseResourceKindV1::Symbol,
1005 ..request("agent-b", "request:b")
1006 };
1007 assert!(matches!(
1008 reg.acquire(second, 1),
1009 Err(AgentLeaseError::CapacityExceeded(1))
1010 ));
1011 }
1012
1013 fn long_request(owner: &str, ref_id: &str) -> AgentLeaseRequestV1 {
1014 AgentLeaseRequestV1 {
1015 duration_ms: 60_000,
1016 ..request(owner, ref_id)
1017 }
1018 }
1019
1020 #[test]
1023 fn shared_registry_is_visible_across_calls_and_persisted() {
1024 let iso = crate::core::data_dir::isolated_data_dir();
1025 let AgentLeaseAcquireV1::Granted(granted) =
1026 acquire_shared(long_request("agent-a", "request:a")).unwrap()
1027 else {
1028 panic!("expected grant")
1029 };
1030 let store = iso.path().join("agents").join("leases.json");
1031 assert!(store.exists(), "lease store is written to the data dir");
1032 assert!(matches!(
1033 acquire_shared(long_request("agent-b", "request:b")),
1034 Ok(AgentLeaseAcquireV1::HeldBy { owner_agent_id, .. }) if owner_agent_id == "agent-a"
1035 ));
1036 assert!(
1037 release_shared(
1038 AgentLeaseResourceKindV1::Path,
1039 "pathref:src-core-main",
1040 "agent-a",
1041 &granted.lease_ref,
1042 )
1043 .unwrap()
1044 );
1045 assert!(matches!(
1046 acquire_shared(long_request("agent-b", "request:b")),
1047 Ok(AgentLeaseAcquireV1::Granted(_))
1048 ));
1049 }
1050
1051 #[test]
1056 fn a_waiting_agent_is_served_before_a_holder_asking_again() {
1057 let mut reg = AgentLeaseRegistryV1::default();
1058 let grant = |reg: &mut AgentLeaseRegistryV1, owner: &str, now| match reg
1059 .acquire(long_request(owner, &format!("request:{owner}")), now)
1060 .unwrap()
1061 {
1062 AgentLeaseAcquireV1::Granted(lease) => Some(lease),
1063 AgentLeaseAcquireV1::HeldBy { .. } => None,
1064 };
1065 let held = grant(&mut reg, "agent-a", 0).expect("free");
1066 assert!(grant(&mut reg, "agent-b", 1).is_none(), "b waits");
1067 assert!(
1068 reg.release(
1069 held.request.resource_kind,
1070 &held.request.resource_ref,
1071 "agent-a",
1072 &held.lease_ref,
1073 2
1074 )
1075 .unwrap()
1076 );
1077 assert!(matches!(
1079 reg.acquire(long_request("agent-a", "request:a2"), 3).unwrap(),
1080 AgentLeaseAcquireV1::HeldBy { owner_agent_id, lease_ref, .. }
1081 if owner_agent_id == "agent-b" && lease_ref == RESERVED_LEASE_REF
1082 ));
1083 let b = grant(&mut reg, "agent-b", 4).expect("b is served");
1084 assert!(grant(&mut reg, "agent-a", 5).is_none());
1086 reg.release(
1087 b.request.resource_kind,
1088 &b.request.resource_ref,
1089 "agent-b",
1090 &b.lease_ref,
1091 6,
1092 )
1093 .unwrap();
1094 assert!(grant(&mut reg, "agent-c", 7).is_none());
1095 assert!(grant(&mut reg, "agent-c", 5 + WAITER_TTL_MS).is_some());
1097
1098 let mut reg = AgentLeaseRegistryV1::default();
1101 let held = grant(&mut reg, "agent-a", 0).expect("free");
1102 let renew = |reg: &mut AgentLeaseRegistryV1, now| {
1103 reg.renew(
1104 held.request.resource_kind,
1105 &held.request.resource_ref,
1106 "agent-a",
1107 &held.lease_ref,
1108 60_000,
1109 now,
1110 )
1111 };
1112 let late = MAX_CONTENDED_HOLD_MS + 30_000;
1113 let mut now = 30_000;
1114 while now < late {
1115 renew(&mut reg, now).expect("uncontended renew");
1116 now += 30_000;
1117 }
1118 assert!(grant(&mut reg, "agent-b", now).is_none());
1119 assert!(
1120 renew(&mut reg, now + 1).is_err(),
1121 "contended past the maximum"
1122 );
1123 assert!(
1124 grant(&mut reg, "agent-b", now + 60_000).is_some(),
1125 "runs out to b"
1126 );
1127 }
1128
1129 #[test]
1136 fn four_contending_agents_each_get_the_lease_after_at_most_three_others() {
1137 use std::collections::BTreeSet;
1138 use std::sync::atomic::{AtomicUsize, Ordering};
1139 use std::sync::{Barrier, Mutex};
1140
1141 let _iso = crate::core::data_dir::isolated_data_dir();
1142 let (p1, p2) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
1143 let resource = |project: &tempfile::TempDir| {
1144 project_path_resource(project.path().to_str().unwrap(), "src/lib.rs").unwrap()
1145 };
1146 let lease = |owner: &str, resource_ref: String| AgentLeaseRequestV1 {
1147 resource_ref,
1148 ..long_request(owner, &format!("request:{owner}"))
1149 };
1150 assert!(matches!(
1151 acquire_shared(lease("human-c-agent-1", resource(&p2))),
1152 Ok(AgentLeaseAcquireV1::Granted(_))
1153 ));
1154
1155 let agents = [
1156 "human-a-agent-1",
1157 "human-a-agent-2",
1158 "human-b-agent-1",
1159 "human-b-agent-2",
1160 ];
1161 let start = Barrier::new(agents.len());
1162 let waiting: Mutex<BTreeSet<&str>> = Mutex::new(agents.into_iter().collect());
1163 let refused: Mutex<BTreeSet<&str>> = Mutex::new(BTreeSet::new());
1164 let (holders, granted) = (AtomicUsize::new(0), AtomicUsize::new(0));
1165 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(60);
1166 let bypassed = std::thread::scope(|scope| {
1167 let contend = |owner: &'static str| {
1168 let (start, waiting, refused) = (&start, &waiting, &refused);
1169 let (holders, granted, lease) = (&holders, &granted, &lease);
1170 let target = resource(&p1);
1171 move || {
1172 start.wait();
1173 let before = granted.load(Ordering::SeqCst);
1174 let held = loop {
1175 assert!(std::time::Instant::now() < deadline, "{owner} starved");
1176 match acquire_shared(lease(owner, target.clone())) {
1177 Ok(AgentLeaseAcquireV1::Granted(held)) => break held,
1178 Ok(AgentLeaseAcquireV1::HeldBy { .. }) => {
1179 refused.lock().unwrap().insert(owner);
1180 }
1181 Err(AgentLeaseError::Store(_)) => {}
1183 Err(error) => panic!("{owner}: {error:?}"),
1184 }
1185 std::thread::sleep(std::time::Duration::from_millis(2));
1186 };
1187 assert_eq!(holders.fetch_add(1, Ordering::SeqCst), 0, "two holders");
1188 let bypassed = granted.fetch_add(1, Ordering::SeqCst) - before;
1189 waiting.lock().unwrap().remove(owner);
1190 refused.lock().unwrap().clear();
1191 while !waiting.lock().unwrap().is_subset(&refused.lock().unwrap()) {
1193 assert!(std::time::Instant::now() < deadline, "waiters never asked");
1194 std::thread::yield_now();
1195 }
1196 holders.fetch_sub(1, Ordering::SeqCst);
1197 while !matches!(
1198 release_shared(
1199 AgentLeaseResourceKindV1::Path,
1200 &target,
1201 owner,
1202 &held.lease_ref
1203 ),
1204 Ok(true)
1205 ) {
1206 assert!(std::time::Instant::now() < deadline, "release never landed");
1207 }
1208 bypassed
1209 }
1210 };
1211 let threads: Vec<_> = agents
1212 .into_iter()
1213 .map(|owner| scope.spawn(contend(owner)))
1214 .collect();
1215 threads
1216 .into_iter()
1217 .map(|thread| thread.join().expect("agent"))
1218 .collect::<Vec<_>>()
1219 });
1220 assert_eq!(granted.load(Ordering::SeqCst), agents.len());
1221 assert!(bypassed.iter().all(|&n| n < agents.len()), "{bypassed:?}");
1222 assert!(matches!(
1224 acquire_shared(lease("human-a-agent-1", resource(&p2))),
1225 Ok(AgentLeaseAcquireV1::HeldBy { owner_agent_id, .. })
1226 if owner_agent_id == "human-c-agent-1"
1227 ));
1228 }
1229
1230 #[test]
1231 fn corrupt_shared_store_fails_closed() {
1232 let iso = crate::core::data_dir::isolated_data_dir();
1233 let dir = iso.path().join("agents");
1234 std::fs::create_dir_all(&dir).unwrap();
1235 std::fs::write(dir.join("leases.json"), "{not json").unwrap();
1236 assert!(matches!(
1237 acquire_shared(long_request("agent-a", "request:a")),
1238 Err(AgentLeaseError::Store(message)) if message.contains("corrupt")
1239 ));
1240 assert_eq!(
1241 std::fs::read_to_string(dir.join("leases.json")).unwrap(),
1242 "{not json",
1243 "a corrupt store is never overwritten"
1244 );
1245 }
1246
1247 #[test]
1250 fn path_leases_are_bound_to_their_project() {
1251 let project_a = tempfile::tempdir().unwrap();
1252 let project_b = tempfile::tempdir().unwrap();
1253 let root_a = project_a.path().to_str().unwrap();
1254 let root_b = project_b.path().to_str().unwrap();
1255 let path = |owner: &str, resource_ref: String| AgentLeaseRequestV1 {
1256 resource_ref,
1257 ..request(owner, &format!("request:{owner}"))
1258 };
1259 let mut reg = AgentLeaseRegistryV1::default();
1260 let held = project_path_target_resource(root_a, "src").unwrap();
1261 assert!(matches!(
1262 reg.acquire(path("agent-a", held), 1),
1263 Ok(AgentLeaseAcquireV1::Granted(_))
1264 ));
1265 let other_project = project_path_target_resource(root_b, "./src").unwrap();
1266 assert!(matches!(
1267 reg.acquire(path("agent-b", other_project), 1),
1268 Ok(AgentLeaseAcquireV1::Granted(_))
1269 ));
1270 let absolute_child =
1271 project_path_target_resource(root_a, &format!("{root_a}/src/lib.rs")).unwrap();
1272 assert_eq!(
1273 absolute_child,
1274 project_path_resource(root_a, "src/lib.rs").unwrap()
1275 );
1276 assert!(matches!(
1277 reg.acquire(path("agent-c", absolute_child), 1),
1278 Ok(AgentLeaseAcquireV1::HeldBy { owner_agent_id, .. }) if owner_agent_id == "agent-a"
1279 ));
1280 assert!(matches!(
1282 reg.acquire(path("agent-d", "path:src".into()), 1),
1283 Ok(AgentLeaseAcquireV1::Granted(_))
1284 ));
1285 for outside in ["../src".to_string(), format!("{root_b}/src"), root_a.into()] {
1286 assert!(project_path_target_resource(root_a, &outside).is_err());
1287 }
1288 #[cfg(unix)]
1291 {
1292 std::os::unix::fs::symlink(project_b.path(), project_a.path().join("link")).unwrap();
1293 for through_link in ["link/src".to_string(), format!("{root_a}/link/src")] {
1294 assert!(project_path_target_resource(root_a, &through_link).is_err());
1295 }
1296 }
1297 }
1298}
1299