diff --git a/crates/fs-operation/src/lib.rs b/crates/fs-operation/src/lib.rs
index 971958418..f8a2ca282 100644
--- a/crates/fs-operation/src/lib.rs
+++ b/crates/fs-operation/src/lib.rs
@@ -154,11 +154,18 @@ pub fn open_beneath_no_symlinks(_root: &Path, _path: &Path) -> std::io::Result<s
     ))
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AtomicWriteMode {
+    CreateNew,
+    Replace,
+}
+
 #[cfg(target_os = "linux")]
 pub fn atomic_write_beneath_no_symlinks_at(
     root: &std::fs::File,
     relative: &Path,
     content: &[u8],
+    mode: AtomicWriteMode,
     before_commit: impl FnOnce() -> std::io::Result<()>,
 ) -> std::io::Result<()> {
     use std::ffi::{CString, OsString};
@@ -234,9 +241,15 @@ pub fn atomic_write_beneath_no_symlinks_at(
     temporary.flush()?;
     temporary.as_file().sync_all()?;
     before_commit()?;
-    temporary
-        .persist(parent_path.join(file_name))
-        .map_err(|error| error.error)?;
+    let destination = parent_path.join(file_name);
+    match mode {
+        AtomicWriteMode::CreateNew => temporary
+            .persist_noclobber(destination)
+            .map_err(|error| error.error)?,
+        AtomicWriteMode::Replace => temporary
+            .persist(destination)
+            .map_err(|error| error.error)?,
+    };
     Ok(())
 }
 
@@ -245,6 +258,7 @@ pub fn atomic_write_beneath_no_symlinks_at(
     _root: &std::fs::File,
     _relative: &Path,
     _content: &[u8],
+    _mode: AtomicWriteMode,
     _before_commit: impl FnOnce() -> std::io::Result<()>,
 ) -> std::io::Result<()> {
     Err(std::io::Error::new(
@@ -317,6 +331,12 @@ pub trait FsAccessPolicy: Send + Sync {
         std::fs::metadata(resolved)
     }
 
+    /// Bound provider-owned mutation preimages and results before allocation or
+    /// commit. `None` retains the provider's ordinary filesystem behavior.
+    fn max_write_bytes(&self) -> Option<usize> {
+        None
+    }
+
     /// Atomically replace or create an already-authorized writable file.
     /// Capability providers override this to keep parent traversal descriptor-
     /// confined through the final rename.
@@ -325,6 +345,7 @@ pub trait FsAccessPolicy: Send + Sync {
         _logical: &Path,
         resolved: &Path,
         content: &[u8],
+        mode: AtomicWriteMode,
     ) -> std::io::Result<()> {
         use std::io::Write;
 
@@ -337,7 +358,12 @@ pub trait FsAccessPolicy: Send + Sync {
         temporary.flush()?;
         temporary.as_file().sync_all()?;
         self.check_cancelled()?;
-        temporary.persist(resolved).map_err(|error| error.error)?;
+        match mode {
+            AtomicWriteMode::CreateNew => temporary
+                .persist_noclobber(resolved)
+                .map_err(|error| error.error)?,
+            AtomicWriteMode::Replace => temporary.persist(resolved).map_err(|error| error.error)?,
+        };
         Ok(())
     }
 
@@ -1166,6 +1192,7 @@ mod tests {
             &root,
             Path::new("escape/secret.txt"),
             b"escaped",
+            AtomicWriteMode::Replace,
             || Ok(()),
         )
         .unwrap_err();
@@ -1184,6 +1211,7 @@ mod tests {
             &root,
             Path::new("nested/created.txt"),
             b"approved",
+            AtomicWriteMode::CreateNew,
             || Ok(()),
         )
         .unwrap();
diff --git a/crates/fs-operation/src/local.rs b/crates/fs-operation/src/local.rs
index 48ffa4d37..155e0bf4f 100644
--- a/crates/fs-operation/src/local.rs
+++ b/crates/fs-operation/src/local.rs
@@ -6,9 +6,9 @@ use std::path::{Path, PathBuf};
 use sha2::{Digest, Sha256};
 
 use crate::{
-    BoundedReadLimits, ContentHash, EditRequest, EditResult, EntryKind, FsAccessPolicy, FsError,
-    FsPath, ListEntry, ListRequest, ListResult, ReadRequest, ReadResult, StatRequest, StatResult,
-    WriteRequest, WriteResult, direct_symlink,
+    AtomicWriteMode, BoundedReadLimits, ContentHash, EditRequest, EditResult, EntryKind,
+    FsAccessPolicy, FsError, FsPath, ListEntry, ListRequest, ListResult, ReadRequest, ReadResult,
+    StatRequest, StatResult, WriteRequest, WriteResult, direct_symlink,
 };
 
 const READ_BUFFER_BYTES: usize = 16 * 1024;
@@ -204,35 +204,33 @@ pub fn run_write(
     access
         .check_cancelled()
         .map_err(|error| map_io(&logical, error))?;
+    enforce_write_bound(&logical, request.content.len(), access)?;
     let path = resolve(root, &logical)?;
-    let created = !path.exists();
-    if path.exists() {
-        let target = require_access(&path, &logical, access, true, false)?;
-        let metadata = access
-            .read_metadata(&path, &target)
-            .map_err(|error| map_io(&logical, error))?;
+    let target = require_access(&path, &logical, access, true, false)?;
+    let metadata = match access.read_metadata(&path, &target) {
+        Ok(metadata) => Some(metadata),
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
+        Err(error) => return Err(map_io(&logical, error)),
+    };
+    let created = metadata.is_none();
+    if let Some(metadata) = metadata {
         if metadata.is_dir() {
             return Err(FsError::IsDirectory(PathBuf::from(logical.as_str())));
         }
-        let mut bytes = Vec::new();
-        access
-            .open_read_file(&path, &target)
-            .and_then(|mut file| file.read_to_end(&mut bytes))
-            .map_err(|error| map_io(&logical, error))?;
+        let bytes = read_mutation_preimage(&logical, &path, &target, metadata, access)?;
         let actual = hash_bytes(&bytes);
         if request.expected_hash != Some(actual) {
             return Err(FsError::Conflict(logical.as_str().to_string()));
         }
         access
-            .atomic_write_file(&path, &target, &request.content)
+            .atomic_write_file(&path, &target, &request.content, AtomicWriteMode::Replace)
             .map_err(|error| map_io(&logical, error))?;
     } else {
         if request.expected_hash.is_some() {
             return Err(FsError::Conflict(logical.as_str().to_string()));
         }
-        let target = require_access(&path, &logical, access, true, true)?;
         access
-            .atomic_write_file(&path, &target, &request.content)
+            .atomic_write_file(&path, &target, &request.content, AtomicWriteMode::CreateNew)
             .map_err(|error| map_io(&logical, error))?;
     }
     Ok(WriteResult {
@@ -252,11 +250,13 @@ pub fn run_edit(
         .map_err(|error| map_io(&logical, error))?;
     let path = resolve(root, &logical)?;
     let target = require_access(&path, &logical, access, true, false)?;
-    let mut bytes = Vec::new();
-    access
-        .open_read_file(&path, &target)
-        .and_then(|mut file| file.read_to_end(&mut bytes))
+    let metadata = access
+        .read_metadata(&path, &target)
         .map_err(|error| map_io(&logical, error))?;
+    if metadata.is_dir() {
+        return Err(FsError::IsDirectory(PathBuf::from(logical.as_str())));
+    }
+    let bytes = read_mutation_preimage(&logical, &path, &target, metadata, access)?;
     let actual_hash = hash_bytes(&bytes);
     if actual_hash != request.expected_hash {
         return Err(FsError::Conflict(logical.as_str().to_string()));
@@ -275,13 +275,31 @@ pub fn run_edit(
             "old_string matched {occurrences} times; set replace_all=true or provide a unique string"
         )));
     }
+    let replacement_count = if request.replace_all { occurrences } else { 1 };
+    let removed_bytes = request
+        .old_string
+        .len()
+        .checked_mul(replacement_count)
+        .ok_or_else(|| FsError::InvalidArgument("edited content size overflowed".to_string()))?;
+    let added_bytes = request
+        .new_string
+        .len()
+        .checked_mul(replacement_count)
+        .ok_or_else(|| FsError::InvalidArgument("edited content size overflowed".to_string()))?;
+    let edited_len = content
+        .len()
+        .checked_sub(removed_bytes)
+        .and_then(|size| size.checked_add(added_bytes))
+        .ok_or_else(|| FsError::InvalidArgument("edited content size overflowed".to_string()))?;
+    enforce_write_bound(&logical, edited_len, access)?;
     let edited = if request.replace_all {
         content.replace(&request.old_string, &request.new_string)
     } else {
         content.replacen(&request.old_string, &request.new_string, 1)
     };
+    debug_assert_eq!(edited.len(), edited_len);
     access
-        .atomic_write_file(&path, &target, edited.as_bytes())
+        .atomic_write_file(&path, &target, edited.as_bytes(), AtomicWriteMode::Replace)
         .map_err(|error| map_io(&logical, error))?;
     Ok(EditResult {
         replacements: if request.replace_all { occurrences } else { 1 },
@@ -381,6 +399,53 @@ pub fn run_list(
     })
 }
 
+fn enforce_write_bound(
+    logical: &FsPath,
+    bytes: usize,
+    access: &dyn FsAccessPolicy,
+) -> Result<(), FsError> {
+    if let Some(limit) = access.max_write_bytes()
+        && bytes > limit
+    {
+        return Err(FsError::InvalidArgument(format!(
+            "{} exceeds provider write limit {limit}",
+            logical.as_str()
+        )));
+    }
+    Ok(())
+}
+
+fn read_mutation_preimage(
+    logical: &FsPath,
+    path: &Path,
+    target: &Path,
+    metadata: std::fs::Metadata,
+    access: &dyn FsAccessPolicy,
+) -> Result<Vec<u8>, FsError> {
+    if let Some(limit) = access.max_write_bytes()
+        && metadata.len() > limit as u64
+    {
+        return Err(FsError::InvalidArgument(format!(
+            "{} exceeds provider write limit {limit}",
+            logical.as_str()
+        )));
+    }
+    let mut file = access
+        .open_read_file(path, target)
+        .map_err(|error| map_io(logical, error))?;
+    let mut bytes = Vec::new();
+    if let Some(limit) = access.max_write_bytes() {
+        file.take(limit.saturating_add(1) as u64)
+            .read_to_end(&mut bytes)
+            .map_err(|error| map_io(logical, error))?;
+        enforce_write_bound(logical, bytes.len(), access)?;
+    } else {
+        file.read_to_end(&mut bytes)
+            .map_err(|error| map_io(logical, error))?;
+    }
+    Ok(bytes)
+}
+
 fn resolve(root: &Path, logical: &FsPath) -> Result<PathBuf, FsError> {
     if !root.is_absolute() {
         return Err(FsError::RelativePath(root.to_path_buf()));
diff --git a/crates/workdir/src/external_local_tests.rs b/crates/workdir/src/external_local_tests.rs
index f0cf5c9a2..b9aeab480 100644
--- a/crates/workdir/src/external_local_tests.rs
+++ b/crates/workdir/src/external_local_tests.rs
@@ -574,6 +574,154 @@ async fn external_local_provider_keeps_pre_grant_root_when_path_changes_before_s
     assert!(!grep_after_symlink_swap.output.contains("secret.txt"));
 }
 
+#[cfg(target_os = "linux")]
+#[tokio::test]
+async fn external_read_write_uses_pinned_root_for_conflict_and_commit() {
+    let parent = tempfile::tempdir().unwrap();
+    let approved = parent.path().join("shared");
+    let approved_original = parent.path().join("approved-original");
+    std::fs::create_dir(&approved).unwrap();
+    std::fs::write(approved.join("item.txt"), "approved").unwrap();
+    let session = LocalWorkdirSession::external_read_write(
+        Workdir::new("external-workdir-pinned-rw"),
+        &approved,
+        BoundedReadLimits::EXTERNAL_DEFAULT,
+    )
+    .unwrap();
+
+    std::fs::rename(&approved, &approved_original).unwrap();
+    std::fs::create_dir(&approved).unwrap();
+
+    let conflict = WorkdirSession::write(
+        &session,
+        WriteRequest {
+            path: WorkdirPath::new("item.txt").unwrap(),
+            content: b"unexpected".to_vec(),
+            expected_hash: None,
+        },
+    )
+    .await
+    .unwrap_err();
+    assert!(matches!(conflict, WorkdirError::Conflict(_)));
+    assert_eq!(
+        std::fs::read_to_string(approved_original.join("item.txt")).unwrap(),
+        "approved"
+    );
+    assert!(!approved.join("item.txt").exists());
+
+    let observed = WorkdirSession::read(
+        &session,
+        ReadRequest {
+            path: WorkdirPath::new("item.txt").unwrap(),
+            offset: 0,
+            limit: 10,
+            max_bytes: 1024,
+        },
+    )
+    .await
+    .unwrap();
+    WorkdirSession::write(
+        &session,
+        WriteRequest {
+            path: WorkdirPath::new("item.txt").unwrap(),
+            content: b"updated".to_vec(),
+            expected_hash: Some(observed.content_hash),
+        },
+    )
+    .await
+    .unwrap();
+    assert_eq!(
+        std::fs::read_to_string(approved_original.join("item.txt")).unwrap(),
+        "updated"
+    );
+    assert!(!approved.join("item.txt").exists());
+}
+
+#[tokio::test]
+async fn external_read_write_rejects_oversized_preimages_and_edit_results_before_commit() {
+    let root = tempfile::tempdir().unwrap();
+    let limit = crate::external::MAX_EXTERNAL_WRITE_BYTES;
+    std::fs::write(root.path().join("oversized.txt"), vec![b'x'; limit + 1]).unwrap();
+    std::fs::write(root.path().join("growth.txt"), vec![b'x'; limit]).unwrap();
+    let session = LocalWorkdirSession::external_read_write(
+        Workdir::new("external-workdir-bounded-rw"),
+        root.path(),
+        BoundedReadLimits::EXTERNAL_DEFAULT,
+    )
+    .unwrap();
+
+    let oversized_observed = WorkdirSession::read(
+        &session,
+        ReadRequest {
+            path: WorkdirPath::new("oversized.txt").unwrap(),
+            offset: 0,
+            limit: 1,
+            max_bytes: 1,
+        },
+    )
+    .await
+    .unwrap();
+    let write_error = WorkdirSession::write(
+        &session,
+        WriteRequest {
+            path: WorkdirPath::new("oversized.txt").unwrap(),
+            content: b"small".to_vec(),
+            expected_hash: Some(oversized_observed.content_hash),
+        },
+    )
+    .await
+    .unwrap_err();
+    assert!(matches!(write_error, WorkdirError::InvalidArgument(_)));
+    let edit_error = WorkdirSession::edit(
+        &session,
+        EditRequest {
+            path: WorkdirPath::new("oversized.txt").unwrap(),
+            old_string: "x".to_string(),
+            new_string: "y".to_string(),
+            replace_all: false,
+            expected_hash: oversized_observed.content_hash,
+        },
+    )
+    .await
+    .unwrap_err();
+    assert!(matches!(edit_error, WorkdirError::InvalidArgument(_)));
+    assert_eq!(
+        std::fs::metadata(root.path().join("oversized.txt"))
+            .unwrap()
+            .len(),
+        (limit + 1) as u64
+    );
+
+    let growth_observed = WorkdirSession::read(
+        &session,
+        ReadRequest {
+            path: WorkdirPath::new("growth.txt").unwrap(),
+            offset: 0,
+            limit: 1,
+            max_bytes: 1,
+        },
+    )
+    .await
+    .unwrap();
+    let growth_error = WorkdirSession::edit(
+        &session,
+        EditRequest {
+            path: WorkdirPath::new("growth.txt").unwrap(),
+            old_string: "x".to_string(),
+            new_string: "yy".to_string(),
+            replace_all: true,
+            expected_hash: growth_observed.content_hash,
+        },
+    )
+    .await
+    .unwrap_err();
+    assert!(matches!(growth_error, WorkdirError::InvalidArgument(_)));
+    assert_eq!(
+        std::fs::read(root.path().join("growth.txt")).unwrap(),
+        vec![b'x'; limit]
+    );
+}
+
 #[cfg(target_os = "linux")]
 #[tokio::test]
 async fn external_local_provider_pins_the_approved_root_across_directory_replacement() {
diff --git a/crates/workdir/src/local.rs b/crates/workdir/src/local.rs
index 14af88322..1d861a35c 100644
--- a/crates/workdir/src/local.rs
+++ b/crates/workdir/src/local.rs
@@ -347,11 +347,17 @@ impl fs_operation::FsAccessPolicy for ScopeAccess {
         }
     }
 
+    fn max_write_bytes(&self) -> Option<usize> {
+        self.reject_symlinks
+            .then_some(crate::external::MAX_EXTERNAL_WRITE_BYTES)
+    }
+
     fn atomic_write_file(
         &self,
         _logical: &Path,
         resolved: &Path,
         content: &[u8],
+        mode: fs_operation::AtomicWriteMode,
     ) -> std::io::Result<()> {
         if !self.reject_symlinks {
             use std::io::Write;
@@ -365,7 +371,14 @@ impl fs_operation::FsAccessPolicy for ScopeAccess {
             temporary.flush()?;
             temporary.as_file().sync_all()?;
             self.check_cancelled()?;
-            temporary.persist(resolved).map_err(|error| error.error)?;
+            match mode {
+                fs_operation::AtomicWriteMode::CreateNew => temporary
+                    .persist_noclobber(resolved)
+                    .map_err(|error| error.error)?,
+                fs_operation::AtomicWriteMode::Replace => {
+                    temporary.persist(resolved).map_err(|error| error.error)?
+                }
+            };
             return Ok(());
         }
         let root = self.pinned_root.as_ref().ok_or_else(|| {
@@ -380,7 +393,7 @@ impl fs_operation::FsAccessPolicy for ScopeAccess {
                 "path is outside provider root",
             )
         })?;
-        fs_operation::atomic_write_beneath_no_symlinks_at(root, relative, content, || {
+        fs_operation::atomic_write_beneath_no_symlinks_at(root, relative, content, mode, || {
             self.check_cancelled()
         })
     }