Skip to main content

lean_ctx/core/
agent_lease.rs

1//! Bounded, payload-free ownership leases for multi-agent mutation planning.
2//!
3//! Gives callers an idempotent path/symbol ownership primitive. Does not perform
4//! a mutation, resolve a path, or authorize an agent; policy and transport bind
5//! it later. Uses caller-provided time for deterministic expiry boundaries.
6//!
7//! `acquire_shared`/`release_shared` persist the registry in
8//! `<data_dir>/agents/leases.json` under an exclusive file lock, so every
9//! lean-ctx process on the machine (each agent's MCP server, the daemon, the
10//! CLI) sees the same holders (#1913).
11//!
12//! Path and symbol resources are bound to one project store: the Work Graph
13//! and `ctx_agent` both name a path as `project-path:<project>/<relative>`, so
14//! the same relative path in two projects never collides and a relative and
15//! an absolute spelling of one file always meet. The unbound legacy `path:`
16//! namespace (stores written before) is a separate namespace and never
17//! compared with it; such leases simply expire.
18//!
19//! Leases are cooperative: the edit tools do not consult them. They exclude
20//! agents that acquire before they write; an agent that never asks is not
21//! stopped by them.
22
23use 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;
33/// Path namespaces whose references nest by `/`: the project-bound one and
34/// the unbound legacy one.
35const 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    /// Fences one claimed local Work Graph node during external execution.
44    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
100/// A waiter keeps its place while it asks again within this time.
101const WAITER_TTL_MS: u64 = 2 * 60 * 1_000;
102/// While another agent waits, a holder may renew only until it has held the
103/// lease this long since its grant; the lease then runs out (at most
104/// `MAX_LEASE_DURATION_MS` later) and goes to the oldest waiter.
105const MAX_CONTENDED_HOLD_MS: u64 = 60 * 60 * 1_000;
106/// `lease_ref` a reservation is reported under (`HeldBy`): it fences nothing.
107const RESERVED_LEASE_REF: &str = "reserved-for-waiter";
108
109/// Fairness state (O04), kept beside the lease store in its own file, so the
110/// lease file keeps its exact schema: processes that predate it read the
111/// lease file as before and simply do not honour reservations.
112#[derive(Clone, Debug, Default, Serialize, Deserialize)]
113struct LeaseFairnessV1 {
114    #[serde(default)]
115    waiters: Vec<LeaseWaiterV1>,
116    /// `lease_ref` → epoch ms of the grant that started the hold.
117    #[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
146/// Local registry with caller-provided clock for deterministic tests.
147pub 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    /// Bounded waiting (O04): every refused request records its agent as a
170    /// waiter for the resource. A free resource goes to the agent that has
171    /// waited longest and is still asking (within `WAITER_TTL_MS`); anyone
172    /// else is told it is held for that waiter. An agent that stops asking
173    /// loses its place after the TTL, so a gone waiter blocks nothing for
174    /// longer than that.
175    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            // Bounded like the leases: a full list refuses no one, it only
214            // stops recording new waiters.
215            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    /// The longest-waiting other agent whose resource overlaps `request`.
227    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        // Only the path namespaces have hierarchical semantics, each compared
276        // with itself alone; other references remain opaque. Exact replays
277        // were handled above.
278        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    /// Release every lease of one execution incarnation: its node lease
342    /// (`lease_request_ref == fence`) and its path leases (`<fence>:<n>`).
343    /// Selected by owner and fence, never by resource name.
344    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    /// Extend an active lease without changing its fencing token.
365    ///
366    /// Renewals are owner- and token-bound, so a delayed worker cannot extend
367    /// a lease that has already been reclaimed. Keeping the token stable makes
368    /// periodic heartbeats idempotent; a reclaim after expiry always receives
369    /// a new token from `acquire`.
370    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// ─── Machine-Wide Shared Registry ───────────────────────────────────────────
439
440/// On-disk form of the registry. A list rather than a map: the registry key is
441/// a `(kind, ref)` tuple, which JSON objects cannot key on.
442#[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    // A corrupt file fails closed: treating it as empty would hand out
475    // resources another agent still holds.
476    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    // Fairness is an ordering among waiters, not an exclusion: a missing or
493    // unreadable file only means no one waits yet.
494    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(&registry.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
522/// Runs `mutate` against the shared registry while holding the store lock, so
523/// a read-check-write from one process can never interleave with another's.
524fn 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, &registry)?;
537    Ok(result)
538}
539
540/// Acquires a lease visible to every lean-ctx process that shares the data dir.
541pub fn acquire_shared(
542    request: AgentLeaseRequestV1,
543) -> Result<AgentLeaseAcquireV1, AgentLeaseError> {
544    with_shared_registry(|registry, now| registry.acquire(request, now))
545}
546
547/// The active leases of the machine-wide registry, soonest expiry first
548/// (then by `lease_ref`). Read only: the store is written atomically.
549pub 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
565/// The project-relative path a `project-path:` resource of `project_root`
566/// names (lower-cased where paths fold, percent escapes kept); `None` for
567/// other resources.
568pub 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
577/// Releases a lease from the machine-wide registry.
578pub 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
589/// Drops the lease `lease_ref` whoever holds it: an organization veto took
590/// it (Hive bus control, SPEC-ENGINE §3.7.4). Returns the dropped lease,
591/// `None` when no such lease is held.
592pub 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
603/// Releases every lease one execution incarnation of `owner_agent_id` holds,
604/// under a single store lock; leases of other incarnations stay.
605pub 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
609/// Renews an owned lease in the machine-wide registry without rotating its
610/// fencing token. Renewal must use the same store as acquisition, otherwise a
611/// lease held by another process could never be renewed.
612pub 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
631// ─── Project-Bound Resources ─────────────────────────────────────────────────
632
633/// Paths name files case-insensitively on the default macOS and Windows file
634/// systems; folding there makes `Src/Lib.rs` and `src/lib.rs` one resource.
635/// On a case-sensitive volume of those systems this only over-excludes.
636const CASE_INSENSITIVE_PATHS: bool = cfg!(any(target_os = "macos", windows));
637
638/// Lease resource of one project-relative path (`src/core`, no `.`/`..`).
639/// Nested paths of one project overlap; paths of different projects never do.
640/// The Work Graph and `ctx_agent` both name their path leases through here.
641pub 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    // Keep the bounded reference contract for unusually long paths. The hash
670    // stays project-bound but cannot express nesting.
671    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
680/// Lease resource of a caller-named path: relative to the project root, or
681/// absolute inside it (Windows verbatim prefixes, drive-letter case and
682/// separators normalized). Both spellings of one file name the same resource.
683/// Like a Work Graph path claim, the path may not traverse a symlink or
684/// leave the project, and the project root itself cannot be leased.
685pub 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    // Lexical only: resolving the target would follow the very symlinks the
691    // walk below must refuse.
692    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
728/// `target` below `root` (both lexically normalized), compared the way the
729/// platform's default file system compares names.
730fn 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
742/// Lease resource of a symbol name within one project.
743pub 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
759// ─── Helpers ─────────────────────────────────────────────────────────────────
760
761fn 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// ─── Errors ──────────────────────────────────────────────────────────────────
793
794#[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// ─── Tests ───────────────────────────────────────────────────────────────────
815
816#[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        // The node lease and the incarnation's path lease; never `fence:older`.
940        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    /// The shared registry keeps no in-memory state: every call reloads the
1021    /// store under its lock, which is exactly what a second process sees.
1022    #[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    /// O04: a holder that releases and asks again at once, or renews for
1052    /// ever, cannot starve an agent that keeps asking. The waiter gets the
1053    /// freed lease; a waiter that stops asking loses its place after the
1054    /// TTL; past the maximum hold a contended lease is not renewed.
1055    #[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        // a asks again first: the lease is kept for b, who waited.
1078        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        // a now waits behind b; c, who never waited, does not pass a.
1085        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        // a stops asking: after the TTL its place is gone.
1096        assert!(grant(&mut reg, "agent-c", 5 + WAITER_TTL_MS).is_some());
1097
1098        // A holder renews while nobody waits, but not past the maximum hold
1099        // once someone does.
1100        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    /// O04: two humans with two agents each contend for one path of P1
1130    /// while a lease on the same path of P2 is held throughout. Released
1131    /// together; each holder keeps the lease until every agent still waiting
1132    /// has been refused once. Exactly one holder at a time, P2 blocks
1133    /// nothing, and every agent is granted after at most three others
1134    /// (bounded bypass: no agent starves while the others finish).
1135    #[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                            // The store lock is bounded: busy is a retry.
1182                            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                    // Hold until every agent still waiting was refused once.
1192                    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        // P2's holder kept its lease the whole time.
1223        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    /// AP07-G1: `path:src` collided across projects, and a relative and an
1248    /// absolute spelling of one file never met.
1249    #[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        // A lease of the unbound legacy namespace neither blocks nor is blocked.
1281        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        // A symlink inside the project may point anywhere: refused like a
1289        // Work Graph path claim, whether spelled relative or absolute.
1290        #[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// SPDX-License-Identifier: Apache-2.0