rpi_tools/
file_mutation_queue.rs

1//! Mirrors `packages/agent/src/harness/tools/file-mutation-queue.ts` — serializes
2//! file-mutating operations per canonical path so concurrent edits to the same
3//! path (or a symlink + its target) don't interleave.
4//!
5//! The TS uses a `WeakMap<ExecutionEnv, MutationQueueState>` keyed by env
6//! instance. Rust has no per-instance map keyed by trait object, so each env
7//! impl carries an `Arc<MutationQueueRegistry>` field and exposes it via the
8//! [`MutatingEnv`] accessor trait. The mutation-dependent tools (write, edit)
9//! hold `Arc<dyn MutatingEnv>`; read and bash don't need it.
10//!
11//! **Critical invariant (plan §5.4): abort does NOT unblock the queue.** An
12//! aborted caller still waits for its turn, then runs `fn` (which may observe
13//! the abort and return `Err(Aborted)`), then releases. The lock guard is held
14//! across `fn().await`; a `CancellationToken` only causes `fn` to error out —
15//! it does NOT preempt mutex acquisition.
16
17use std::collections::HashMap;
18use std::sync::Arc;
19
20use tokio::sync::Mutex;
21use tokio_util::sync::CancellationToken;
22
23use crate::env::ExecutionEnv;
24use crate::error::{FileError, FileErrorCode};
25
26/// Per-env registry of canonical-path → mutex. Stored as a field on each env
27/// impl (behind `Arc` so clones share the same map).
28#[derive(Default)]
29pub struct MutationQueueRegistry {
30    queues: Mutex<HashMap<String, Arc<Mutex<()>>>>,
31}
32
33impl MutationQueueRegistry {
34    pub fn new() -> Self {
35        Self::default()
36    }
37
38    /// Get (or lazily create) the mutex for `key`. The key is a canonical path
39    /// string (from [`get_mutation_queue_key`]).
40    async fn mutex_for(&self, key: &str) -> Arc<Mutex<()>> {
41        let mut map = self.queues.lock().await;
42        if let Some(m) = map.get(key) {
43            m.clone()
44        } else {
45            let m = Arc::new(Mutex::new(()));
46            map.insert(key.to_string(), m.clone());
47            m
48        }
49    }
50}
51
52/// Accessor used by [`with_file_mutation_queue`] to reach the env's registry
53/// AND its `ExecutionEnv` methods. Concrete env impls (`OsExecutionEnv`,
54/// `InMemoryExecutionEnv`) implement this; the write/edit tools hold
55/// `Arc<dyn MutatingEnv>`.
56///
57/// The `as_env` indirection avoids needing trait-upcasting (`dyn MutatingEnv`
58/// → `dyn ExecutionEnv`, unstable before Rust 1.86); our min rust-version is
59/// 1.78, so we expose the env explicitly.
60pub trait MutatingEnv: Send + Sync {
61    /// The underlying `ExecutionEnv` (for `FileSystem`/`Shell` calls).
62    fn as_env(&self) -> &dyn ExecutionEnv;
63
64    /// The registry holding per-canonical-path mutexes.
65    fn mutation_registry(&self) -> &MutationQueueRegistry;
66}
67
68/// Compute the canonical-path queue key for `path`. Mirrors
69/// `getMutationQueueKey`:
70/// 1. `absolutePath(path)` → propagate err.
71/// 2. `canonicalPath(absolute)`:
72///    - `Ok` → return canonical string.
73///    - `Err` with `not_found` or `not_supported` → fall back to absolute string.
74///    - other err → propagate.
75pub async fn get_mutation_queue_key(
76    env: &dyn ExecutionEnv,
77    path: &str,
78    cancel: Option<&CancellationToken>,
79) -> Result<String, FileError> {
80    let absolute = env.absolute_path(path, cancel).await?;
81    let abs_str = absolute.to_string_lossy().into_owned();
82    match env.canonical_path(&abs_str, cancel).await {
83        Ok(canonical) => Ok(canonical.to_string_lossy().into_owned()),
84        Err(e) if e.code == FileErrorCode::NotFound || e.code == FileErrorCode::NotSupported => {
85            Ok(abs_str)
86        }
87        Err(e) => Err(e),
88    }
89}
90
91/// Serialize `fn` per canonical-path key for `env`. The closure `f` runs while
92/// holding the per-key `tokio::Mutex` guard; abort/cancel during the wait does
93/// NOT skip the queue — `f` itself must observe the token and return
94/// `Err(FileError::aborted())`.
95///
96/// Mirrors `withFileMutationQueue`. The `&CancellationToken` is passed to `f`
97/// so it can perform the same pre/post-execution abort checks the tools do
98/// (the write tool checks before AND after the write).
99pub async fn with_file_mutation_queue<F, R>(
100    env: &dyn MutatingEnv,
101    path: &str,
102    cancel: &CancellationToken,
103    f: F,
104) -> Result<R, FileError>
105where
106    F: for<'a> FnOnce(
107        &'a CancellationToken,
108    ) -> futures::future::BoxFuture<'a, Result<R, FileError>>,
109    R: Send + 'static,
110{
111    let key = get_mutation_queue_key(env.as_env(), path, Some(cancel)).await?;
112    let mutex = env.mutation_registry().mutex_for(&key).await;
113    // Acquisition of the per-key lock is NOT cancelable — we always wait for
114    // our turn. `tokio::Mutex::lock` is a safe `.await` here.
115    let _guard = mutex.lock().await;
116    // Now that we hold the lock, run `f`. It may observe cancellation and
117    // return Err(Aborted); the guard drops on return, releasing the next
118    // waiter. Unwinding (if `f` panics) also drops the guard — but tool/env
119    // code must never panic (plan §5.13).
120    f(cancel).await
121}
122
123#[cfg(test)]
124mod tests {
125    use super::*;
126
127    #[tokio::test]
128    async fn registry_creates_and_reuses_mutex() {
129        let reg = MutationQueueRegistry::new();
130        let m1 = reg.mutex_for("/a/b").await;
131        let m2 = reg.mutex_for("/a/b").await;
132        assert!(Arc::ptr_eq(&m1, &m2));
133        let m3 = reg.mutex_for("/c").await;
134        assert!(!Arc::ptr_eq(&m1, &m3));
135    }
136}
rpi_tools/file_mutation_queue.rs

rpi_tools/
file_mutation_queue.rs
