feat(mounts): P2 — read-write REST for external mounts

Adds full CRUD on mount contents, mirroring the P1 read pattern (handlers/
services classify; authorization stays in the service via the mount-root
folder grant; the provider does the I/O).

- mkdir / rename / delete / move-within branch inside FolderService and
  FileManagementService (router injected into both)
- streaming upload via a new ExternalUploadService: the upload handler detects a
  mount destination BEFORE the CAS ingest and streams the multipart body
  straight to the provider (no BLAKE3/dedup). `write_stream` now takes a
  lifetime-bound boxed stream so the borrowing multipart field can be passed
  without buffering.
- deletes on mounts are permanent (no trash): the trash-first folder handler
  routes `ext:` ids straight to the provider delete; file delete goes through the
  branched delete_and_cleanup
- cross-backend move/copy (mount ↔ native, or between mounts) is forbidden
  (UnsupportedOperation); the mount root itself cannot be renamed/moved/deleted
- every mutation emits a `target:"audit" event="external_mount.write"` line
- shared mount_dto builders synthesize FolderDto/FileDto from a provider MountStat

Tests: 529 unit + integration tests for mkdir/rename/delete, file rename/delete,
streaming upload, cross-boundary forbid, and stranger-denied — all against real
Postgres + a real provider (testcontainers).

src/application/ports/external_mount_ports.rs

@@ -32,7 +32,8 @@ use crate::domain::services::external_mount_id::NodeId;
 ///
 /// Boxed (not generic) so the trait stays object-safe. Callers map their body's
 /// error type to `std::io::Error` before constructing it.
-pub type MountByteStream = Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>;
+pub type MountByteStream<'a> =
+    Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send + 'a>>;
 
 /// One entry returned by [`ExternalMountProvider::list_dir`].
 #[derive(Debug, Clone)]
@@ -126,7 +127,7 @@ pub trait ExternalMountProvider: Send + Sync + 'static {
         &self,
         parent: &NodeId,
         name: &str,
-        body: MountByteStream,
+        body: MountByteStream<'_>,
     ) -> Result<MountStat, DomainError>;
 
     /// Rename an entry in place (same parent). Returns the renamed entry's stat.

src/application/services/external_upload_service.rs

@@ -0,0 +1,55 @@
+//! Streams an upload straight to an external mount's provider, bypassing the
+//! content-addressable store entirely (no BLAKE3 / dedup).
+//!
+//! The REST upload handler detects a mount destination BEFORE ingesting into the
+//! CAS and routes here. Authorization stays in this service (the mount-root
+//! `Create` grant); the handler only classifies and supplies the body stream.
+
+use std::sync::Arc;
+
+use crate::application::dtos::file_dto::FileDto;
+use crate::application::ports::authorization_ports::AuthorizationEngine;
+use crate::application::ports::external_mount_ports::MountByteStream;
+use crate::application::services::mount_dto::{audit_mount_write, mount_file_dto, mount_parent_id};
+use crate::application::services::mount_registry::MountConfig;
+use crate::common::errors::DomainError;
+use crate::domain::services::authorization::{Permission, Resource, Subject};
+use crate::domain::services::external_mount_id::NodeId;
+use crate::infrastructure::services::pg_acl_engine::PgAclEngine;
+use uuid::Uuid;
+
+/// Writes uploaded bytes to a mount provider with authorization + auditing.
+pub struct ExternalUploadService {
+    authz: Arc<PgAclEngine>,
+}
+
+impl ExternalUploadService {
+    /// Construct over the ReBAC engine.
+    pub fn new(authz: Arc<PgAclEngine>) -> Self {
+        Self { authz }
+    }
+
+    /// Authorize (`Create` on the mount root) then stream `body` to the provider
+    /// as `name` under `parent_node`. Returns the synthesized `FileDto`.
+    pub async fn write_file(
+        &self,
+        cfg: &MountConfig,
+        parent_node: &NodeId,
+        name: &str,
+        body: MountByteStream<'_>,
+        caller_id: Uuid,
+    ) -> Result<FileDto, DomainError> {
+        self.authz
+            .require(
+                Subject::User(caller_id),
+                Permission::Create,
+                Resource::Folder(cfg.mount_id),
+            )
+            .await?;
+
+        let stat = cfg.provider.write_stream(parent_node, name, body).await?;
+        audit_mount_write("upload", cfg, caller_id, stat.node_id.as_str());
+        let parent = mount_parent_id(cfg, stat.node_id.as_str());
+        Ok(mount_file_dto(cfg, &parent, &stat))
+    }
+}

src/application/services/file_management_service.rs

@@ -6,9 +6,13 @@ use crate::application::ports::file_lifecycle::FileLifecycleHook;
 use crate::application::ports::file_ports::FileManagementUseCase;
 use crate::application::ports::storage_ports::{CopyFolderTreeResult, FileWritePort};
 use crate::application::ports::trash_ports::TrashUseCase;
+use crate::application::services::external_mount_router::{MountRouter, ResolvedId};
+use crate::application::services::mount_dto::{audit_mount_write, mount_file_dto, mount_parent_id};
+use crate::application::services::mount_registry::MountConfig;
 use crate::application::services::trash_service::TrashService;
 use crate::common::errors::DomainError;
 use crate::domain::services::authorization::{Permission, Resource, Subject};
+use crate::domain::services::external_mount_id::NodeId;
 use crate::domain::services::path_service::validate_storage_name;
 use crate::infrastructure::repositories::pg::file_blob_read_repository::FileBlobReadRepository;
 use crate::infrastructure::repositories::pg::file_blob_write_repository::FileBlobWriteRepository;
@@ -31,6 +35,9 @@ pub struct FileManagementService {
     authz: Arc<PgAclEngine>,
     /// Lifecycle hook dispatcher — fired on file created (copy) and deleted.
     file_lifecycle_hook: Option<Arc<dyn FileLifecycleHook>>,
+    /// External-mount classifier. `None` in stub/test construction → all ids
+    /// are treated as native.
+    mount_router: Option<Arc<MountRouter>>,
 }
 
 impl FileManagementService {
@@ -53,6 +60,7 @@ impl FileManagementService {
             content_cache,
             authz,
             file_lifecycle_hook: None,
+            mount_router: None,
         }
     }
 
@@ -62,6 +70,59 @@ impl FileManagementService {
         self
     }
 
+    /// Injects the external-mount classifier so file mutations can branch
+    /// `ext:` ids to the provider.
+    pub fn with_mount_router(mut self, router: Arc<MountRouter>) -> Self {
+        self.mount_router = Some(router);
+        self
+    }
+
+    /// Classify an id via the mount router (if configured). Returns `Regular`
+    /// when no router is wired.
+    fn classify(&self, id: &str) -> ResolvedId {
+        match &self.mount_router {
+            Some(r) => r.classify(id),
+            None => ResolvedId::Regular,
+        }
+    }
+
+    /// Authorize a mutation inside a mount (gates on the mount-root folder).
+    async fn require_mount_perm(
+        &self,
+        cfg: &MountConfig,
+        perm: Permission,
+        caller_id: Uuid,
+    ) -> Result<(), DomainError> {
+        self.authz
+            .require(
+                Subject::User(caller_id),
+                perm,
+                Resource::Folder(cfg.mount_id),
+            )
+            .await
+    }
+
+    /// Resolve a move destination within the same mount as `cfg`. Errors when
+    /// the destination is absent, native, or in a different mount.
+    fn mount_dest_node(
+        &self,
+        cfg: &MountConfig,
+        folder_id: Option<&str>,
+    ) -> Result<NodeId, DomainError> {
+        let Some(folder_id) = folder_id else {
+            return Err(cross_boundary_move_err());
+        };
+        match self.classify(folder_id) {
+            ResolvedId::MountRoot { cfg: dest } if dest.mount_id == cfg.mount_id => {
+                Ok(NodeId::default())
+            }
+            ResolvedId::MountChild { cfg: dest, node_id } if dest.mount_id == cfg.mount_id => {
+                Ok(node_id)
+            }
+            _ => Err(cross_boundary_move_err()),
+        }
+    }
+
     /// Engine check for a file resource. Parses the id into a `Uuid` and
     /// requires the specified permission.
     async fn require_file_perm(
@@ -255,6 +316,29 @@ impl FileManagementUseCase for FileManagementService {
         caller_id: Uuid,
         folder_id: Option<String>,
     ) -> Result<FileDto, DomainError> {
+        // External mount: moves stay within one mount; cross-backend is forbidden.
+        match self.classify(file_id) {
+            ResolvedId::Regular => {
+                if let Some(dst) = folder_id.as_deref()
+                    && !matches!(self.classify(dst), ResolvedId::Regular)
+                {
+                    return Err(cross_boundary_move_err());
+                }
+            }
+            ResolvedId::MountRoot { .. } => return Err(DomainError::not_found("File", file_id)),
+            ResolvedId::MountChild { cfg, node_id } => {
+                let dest = self.mount_dest_node(&cfg, folder_id.as_deref())?;
+                self.require_mount_perm(&cfg, Permission::Update, caller_id)
+                    .await?;
+                self.require_mount_perm(&cfg, Permission::Create, caller_id)
+                    .await?;
+                let stat = cfg.provider.move_within(&node_id, &dest).await?;
+                audit_mount_write("move", &cfg, caller_id, stat.node_id.as_str());
+                let parent = mount_parent_id(&cfg, stat.node_id.as_str());
+                return Ok(mount_file_dto(&cfg, &parent, &stat));
+            }
+        }
+
         // Move = Update on the file + Create on the target folder (if any).
         self.require_file_perm(file_id, Permission::Update, caller_id)
             .await?;
@@ -285,12 +369,32 @@ impl FileManagementUseCase for FileManagementService {
         caller_id: Uuid,
         new_name: &str,
     ) -> Result<FileDto, DomainError> {
+        if let ResolvedId::MountChild { cfg, node_id } = self.classify(file_id) {
+            if let Err(reason) = validate_storage_name(new_name) {
+                return Err(DomainError::validation_error(format!(
+                    "Invalid file name '{new_name}': {reason}"
+                )));
+            }
+            self.require_mount_perm(&cfg, Permission::Update, caller_id)
+                .await?;
+            let stat = cfg.provider.rename(&node_id, new_name).await?;
+            audit_mount_write("rename", &cfg, caller_id, stat.node_id.as_str());
+            let parent = mount_parent_id(&cfg, stat.node_id.as_str());
+            return Ok(mount_file_dto(&cfg, &parent, &stat));
+        }
         self.require_file_perm(file_id, Permission::Update, caller_id)
             .await?;
         self.rename_file(file_id, new_name, caller_id).await
     }
 
     async fn delete_file_with_perms(&self, id: &str, caller_id: Uuid) -> Result<(), DomainError> {
+        if let ResolvedId::MountChild { cfg, node_id } = self.classify(id) {
+            self.require_mount_perm(&cfg, Permission::Delete, caller_id)
+                .await?;
+            cfg.provider.delete(&node_id).await?;
+            audit_mount_write("delete", &cfg, caller_id, node_id.as_str());
+            return Ok(());
+        }
         self.require_file_perm(id, Permission::Delete, caller_id)
             .await?;
         self.delete_file(id).await
@@ -307,6 +411,15 @@ impl FileManagementUseCase for FileManagementService {
         id: &str,
         caller_id: Uuid,
     ) -> Result<bool, DomainError> {
+        // External mount: permanent provider delete (mounts have no trash).
+        if let ResolvedId::MountChild { cfg, node_id } = self.classify(id) {
+            self.require_mount_perm(&cfg, Permission::Delete, caller_id)
+                .await?;
+            cfg.provider.delete(&node_id).await?;
+            audit_mount_write("delete", &cfg, caller_id, node_id.as_str());
+            return Ok(false); // permanently deleted (no trash)
+        }
+
         self.require_file_perm(id, Permission::Delete, caller_id)
             .await?;
         // Step 1: Try trash (soft delete — file row stays, blob stays referenced)
@@ -357,3 +470,12 @@ impl FileManagementUseCase for FileManagementService {
             .await
     }
 }
+
+/// Error for a move/copy that would cross a storage backend boundary
+/// (mount ↔ native, or between two different mounts). Forbidden in v1.
+fn cross_boundary_move_err() -> DomainError {
+    DomainError::operation_not_supported(
+        "File",
+        "moving between external mounts and regular storage is not supported",
+    )
+}

src/application/services/folder_service.rs

@@ -6,7 +6,10 @@ use crate::application::dtos::folder_dto::{
 use crate::application::ports::authorization_ports::AuthorizationEngine;
 use crate::application::ports::external_mount_ports::MountEntry;
 use crate::application::ports::folder_ports::FolderUseCase;
-use crate::application::services::external_mount_router::MountRouter;
+use crate::application::services::external_mount_router::{MountRouter, ResolvedId};
+use crate::application::services::mount_dto::{
+    audit_mount_write, mount_folder_dto, mount_parent_id,
+};
 use crate::application::services::mount_registry::MountConfig;
 use crate::common::errors::{DomainError, ErrorKind};
 use crate::domain::repositories::folder_repository::FolderRepository;
@@ -47,6 +50,45 @@ impl FolderService {
         &self.mount_router
     }
 
+    /// Authorize a mutation inside a mount. All operations within a mount gate
+    /// on the mount-root folder grant (the `cfg.mount_id` resource).
+    async fn require_mount_perm(
+        &self,
+        cfg: &MountConfig,
+        perm: Permission,
+        caller_id: Uuid,
+    ) -> Result<(), DomainError> {
+        self.authz
+            .require(
+                Subject::User(caller_id),
+                perm,
+                Resource::Folder(cfg.mount_id),
+            )
+            .await
+    }
+
+    /// Resolve a move destination within the SAME mount as `cfg`, returning the
+    /// destination parent's node id. Errors (`UnsupportedOperation`) if the
+    /// destination is absent, native, or in a different mount.
+    fn mount_dest_node(
+        &self,
+        cfg: &MountConfig,
+        parent_id: Option<&str>,
+    ) -> Result<NodeId, DomainError> {
+        let Some(parent_id) = parent_id else {
+            return Err(cross_boundary_move_err());
+        };
+        match self.mount_router.classify(parent_id) {
+            ResolvedId::MountRoot { cfg: dest } if dest.mount_id == cfg.mount_id => {
+                Ok(NodeId::default())
+            }
+            ResolvedId::MountChild { cfg: dest, node_id } if dest.mount_id == cfg.mount_id => {
+                Ok(node_id)
+            }
+            _ => Err(cross_boundary_move_err()),
+        }
+    }
+
     /// Batch counterpart of `get_folder`: resolve many folder ids in ONE
     /// query instead of one per id. Like `get_folder` it performs no
     /// per-folder authorization — both current callers (ACL grant listing,
@@ -232,6 +274,29 @@ impl FolderUseCase for FolderService {
                 "Root folder creation is reserved for registration",
             ));
         };
+
+        // External mount: create the directory on the provider, not in PG.
+        match self.mount_router.classify(parent_id) {
+            ResolvedId::Regular => {}
+            ResolvedId::MountRoot { cfg } => {
+                self.require_mount_perm(&cfg, Permission::Create, caller_id)
+                    .await?;
+                let stat = cfg
+                    .provider
+                    .create_dir(&NodeId::default(), &dto.name)
+                    .await?;
+                audit_mount_write("mkdir", &cfg, caller_id, stat.node_id.as_str());
+                return Ok(mount_folder_dto(&cfg, parent_id, &stat));
+            }
+            ResolvedId::MountChild { cfg, node_id } => {
+                self.require_mount_perm(&cfg, Permission::Create, caller_id)
+                    .await?;
+                let stat = cfg.provider.create_dir(&node_id, &dto.name).await?;
+                audit_mount_write("mkdir", &cfg, caller_id, stat.node_id.as_str());
+                return Ok(mount_folder_dto(&cfg, parent_id, &stat));
+            }
+        }
+
         let parent_resource = Self::folder_resource(parent_id)?;
         self.authz
             .require(
@@ -464,6 +529,26 @@ impl FolderUseCase for FolderService {
             )));
         }
 
+        // External mount: rename on the provider. The mount root cannot be
+        // renamed through here (it's a real folder row managed elsewhere).
+        match self.mount_router.classify(id) {
+            ResolvedId::Regular => {}
+            ResolvedId::MountRoot { .. } => {
+                return Err(DomainError::operation_not_supported(
+                    "Folder",
+                    "a mount root cannot be renamed through this endpoint",
+                ));
+            }
+            ResolvedId::MountChild { cfg, node_id } => {
+                self.require_mount_perm(&cfg, Permission::Update, caller_id)
+                    .await?;
+                let stat = cfg.provider.rename(&node_id, &dto.name).await?;
+                let parent = mount_parent_id(&cfg, stat.node_id.as_str());
+                audit_mount_write("rename", &cfg, caller_id, stat.node_id.as_str());
+                return Ok(mount_folder_dto(&cfg, &parent, &stat));
+            }
+        }
+
         // Drive roots double as the drive's display name (per drive.md §3,
         // `drives.name` is sourced from `storage.folders.name` of the row
         // pointed at by `root_folder_id`). Per drive.md §6 the rename is
@@ -516,6 +601,37 @@ impl FolderUseCase for FolderService {
         dto: MoveFolderDto,
         caller_id: Uuid,
     ) -> Result<FolderDto, DomainError> {
+        // External mount: moves must stay within a single mount. The provider
+        // relocates; cross-backend moves (mount ↔ native, or between mounts) are
+        // forbidden in v1.
+        match self.mount_router.classify(id) {
+            ResolvedId::Regular => {
+                // Native source: forbid moving INTO a mount.
+                if let Some(parent_id) = &dto.parent_id
+                    && self.mount_router.is_mount_id(parent_id)
+                {
+                    return Err(cross_boundary_move_err());
+                }
+            }
+            ResolvedId::MountRoot { .. } => {
+                return Err(DomainError::operation_not_supported(
+                    "Folder",
+                    "a mount root cannot be moved",
+                ));
+            }
+            ResolvedId::MountChild { cfg, node_id } => {
+                let dest = self.mount_dest_node(&cfg, dto.parent_id.as_deref())?;
+                self.require_mount_perm(&cfg, Permission::Update, caller_id)
+                    .await?;
+                self.require_mount_perm(&cfg, Permission::Create, caller_id)
+                    .await?;
+                let stat = cfg.provider.move_within(&node_id, &dest).await?;
+                audit_mount_write("move", &cfg, caller_id, stat.node_id.as_str());
+                let parent = mount_parent_id(&cfg, stat.node_id.as_str());
+                return Ok(mount_folder_dto(&cfg, &parent, &stat));
+            }
+        }
+
         let source_resource = Self::folder_resource(id)?;
         self.authz
             .require(
@@ -564,6 +680,25 @@ impl FolderUseCase for FolderService {
     /// The DB trigger `trg_cleanup_grants_folder` cleans up `access_grants`
     /// rows targeting the deleted folder automatically.
     async fn delete_folder_with_perms(&self, id: &str, caller_id: Uuid) -> Result<(), DomainError> {
+        // External mount: delete on the provider (permanent — mounts have no
+        // trash). The mount root is a real folder row and is not deletable here.
+        match self.mount_router.classify(id) {
+            ResolvedId::Regular => {}
+            ResolvedId::MountRoot { .. } => {
+                return Err(DomainError::operation_not_supported(
+                    "Folder",
+                    "a mount root cannot be deleted through this endpoint",
+                ));
+            }
+            ResolvedId::MountChild { cfg, node_id } => {
+                self.require_mount_perm(&cfg, Permission::Delete, caller_id)
+                    .await?;
+                cfg.provider.delete(&node_id).await?;
+                audit_mount_write("delete", &cfg, caller_id, node_id.as_str());
+                return Ok(());
+            }
+        }
+
         self.authz
             .require(
                 Subject::User(caller_id),
@@ -581,6 +716,15 @@ impl FolderUseCase for FolderService {
     }
 }
 
+/// The error returned when a move would cross a storage backend boundary
+/// (mount ↔ native, or between two different mounts). Forbidden in v1.
+fn cross_boundary_move_err() -> DomainError {
+    DomainError::operation_not_supported(
+        "Folder",
+        "moving between external mounts and regular storage is not supported",
+    )
+}
+
 // ── FolderService — cursor-paginated resource listing ────────────────────────
 
 impl FolderService {
@@ -1201,6 +1345,227 @@ mod mount_authz_integration {
         ))
     }
 
+    /// Provision a mount over `host`, build a wired FolderService, and return
+    /// `(folder_service, mount_root_uuid_string, owner_id)`.
+    async fn wire_mount(
+        pool: &Arc<sqlx::PgPool>,
+        host: &std::path::Path,
+    ) -> (FolderService, String, Uuid) {
+        let p = provision_folder(pool, "owner", "Media").await;
+        insert_mount(pool, &p, host.to_str().unwrap()).await;
+        let registry = Arc::new(MountRegistry::empty());
+        registry
+            .reload(
+                &ExternalMountPgRepository::new(pool.clone()),
+                &DefaultMountProviderFactory::new(),
+            )
+            .await;
+        let router = Arc::new(MountRouter::new(registry));
+        let fs = FolderService::new(
+            Arc::new(FolderDbRepository::new(pool.clone())),
+            acl(pool),
+            router,
+        );
+        (fs, p.mount_folder_id.to_string(), p.owner_id)
+    }
+
+    /// P2 write path: owner can mkdir/rename/delete inside a mount (reflected on
+    /// the host fs); a stranger is denied; the mount root cannot be renamed.
+    #[tokio::test]
+    async fn owner_mkdir_rename_delete_on_mount() {
+        use crate::application::dtos::folder_dto::{CreateFolderDto, RenameFolderDto};
+        let (_c, pool) = fresh_db().await;
+        let host = tempfile::tempdir().unwrap();
+        let (fs, mount_id, owner) = wire_mount(&pool, host.path()).await;
+
+        // mkdir under the mount root.
+        let created = fs
+            .create_folder_with_perms(
+                CreateFolderDto {
+                    name: "docs".into(),
+                    parent_id: Some(mount_id.clone()),
+                },
+                owner,
+            )
+            .await
+            .expect("owner may mkdir");
+        assert!(host.path().join("docs").is_dir());
+        assert!(created.id.starts_with("ext:"));
+        assert_eq!(created.parent_id.as_deref(), Some(mount_id.as_str()));
+
+        // Stranger may NOT mkdir.
+        let stranger = make_user(&pool, "stranger").await;
+        let denied = fs
+            .create_folder_with_perms(
+                CreateFolderDto {
+                    name: "evil".into(),
+                    parent_id: Some(mount_id.clone()),
+                },
+                stranger,
+            )
+            .await;
+        assert!(denied.is_err());
+        assert!(!host.path().join("evil").exists());
+
+        // rename the created dir.
+        let renamed = fs
+            .rename_folder_with_perms(
+                &created.id,
+                RenameFolderDto {
+                    name: "papers".into(),
+                },
+                owner,
+            )
+            .await
+            .expect("owner may rename");
+        assert!(host.path().join("papers").is_dir());
+        assert!(!host.path().join("docs").exists());
+
+        // The mount root itself cannot be renamed through this path.
+        assert!(
+            fs.rename_folder_with_perms(
+                &mount_id,
+                RenameFolderDto {
+                    name: "nope".into()
+                },
+                owner
+            )
+            .await
+            .is_err()
+        );
+
+        // delete (permanent — mounts have no trash).
+        fs.delete_folder_with_perms(&renamed.id, owner)
+            .await
+            .expect("owner may delete");
+        assert!(!host.path().join("papers").exists());
+    }
+
+    /// P2: file rename/delete and streaming upload on a mount, with authz.
+    #[tokio::test]
+    async fn file_rename_delete_and_upload_on_mount() {
+        use crate::application::ports::external_mount_ports::MountByteStream;
+        use crate::application::ports::file_ports::FileManagementUseCase;
+        use crate::application::services::external_upload_service::ExternalUploadService;
+        use crate::application::services::file_management_service::FileManagementService;
+        use crate::infrastructure::repositories::pg::FileBlobWriteRepository;
+        use bytes::Bytes;
+        use futures::stream;
+
+        let (_c, pool) = fresh_db().await;
+        let host = tempfile::tempdir().unwrap();
+        std::fs::write(host.path().join("a.txt"), b"hello").unwrap();
+
+        let p = provision_folder(&pool, "owner", "Media").await;
+        insert_mount(&pool, &p, host.path().to_str().unwrap()).await;
+        let registry = Arc::new(MountRegistry::empty());
+        registry
+            .reload(
+                &ExternalMountPgRepository::new(pool.clone()),
+                &DefaultMountProviderFactory::new(),
+            )
+            .await;
+        let router = Arc::new(MountRouter::new(registry.clone()));
+        let cfg = registry.get(&p.mount_folder_id).expect("registered");
+
+        let mgmt = FileManagementService::with_trash(
+            Arc::new(FileBlobWriteRepository::new_stub()),
+            None,
+            None,
+            None,
+            None,
+            acl(&pool),
+        )
+        .with_mount_router(router.clone());
+
+        let file_id = encode_child_id(p.mount_folder_id, "a.txt");
+
+        // Owner renames the mount file.
+        let renamed = mgmt
+            .rename_file_with_perms(&file_id, p.owner_id, "b.txt")
+            .await
+            .expect("owner may rename");
+        assert!(host.path().join("b.txt").exists());
+        assert!(!host.path().join("a.txt").exists());
+        assert_eq!(renamed.content_hash, "");
+
+        // Stranger may not delete.
+        let stranger = make_user(&pool, "stranger").await;
+        assert!(
+            mgmt.delete_file_with_perms(&renamed.id, stranger)
+                .await
+                .is_err()
+        );
+        assert!(host.path().join("b.txt").exists());
+
+        // Owner deletes (permanent — no trash).
+        mgmt.delete_file_with_perms(&renamed.id, p.owner_id)
+            .await
+            .expect("owner may delete");
+        assert!(!host.path().join("b.txt").exists());
+
+        // Streaming upload straight to the provider.
+        let upload = ExternalUploadService::new(acl(&pool));
+        let body: MountByteStream<'static> =
+            Box::pin(stream::once(async { Ok(Bytes::from_static(b"uploaded")) }));
+        let dto = upload
+            .write_file(&cfg, &NodeId::default(), "new.txt", body, p.owner_id)
+            .await
+            .expect("owner may upload");
+        assert_eq!(dto.size, 8);
+        assert_eq!(
+            std::fs::read(host.path().join("new.txt")).unwrap(),
+            b"uploaded"
+        );
+
+        // Stranger upload denied.
+        let body2: MountByteStream<'static> =
+            Box::pin(stream::once(async { Ok(Bytes::from_static(b"x")) }));
+        assert!(
+            upload
+                .write_file(&cfg, &NodeId::default(), "evil.txt", body2, stranger)
+                .await
+                .is_err()
+        );
+        assert!(!host.path().join("evil.txt").exists());
+    }
+
+    /// P2: a move that would cross the mount boundary is forbidden.
+    #[tokio::test]
+    async fn cross_boundary_move_forbidden() {
+        use crate::application::dtos::folder_dto::{CreateFolderDto, MoveFolderDto};
+        let (_c, pool) = fresh_db().await;
+        let host = tempfile::tempdir().unwrap();
+        std::fs::create_dir(host.path().join("inside")).unwrap();
+        let (fs, mount_id, owner) = wire_mount(&pool, host.path()).await;
+
+        let child_id = encode_child_id(Uuid::parse_str(&mount_id).unwrap(), "inside");
+
+        // Moving a mount child to the user's native root (parent_id = None) is
+        // a cross-backend move → UnsupportedOperation.
+        let err = fs
+            .move_folder_with_perms(&child_id, MoveFolderDto { parent_id: None }, owner)
+            .await
+            .expect_err("cross-boundary move must be forbidden");
+        assert_eq!(
+            err.kind,
+            crate::domain::errors::ErrorKind::UnsupportedOperation
+        );
+
+        // A native folder cannot be moved INTO the mount either.
+        let native = fs
+            .create_folder_with_perms(
+                CreateFolderDto {
+                    name: "n".into(),
+                    parent_id: Some(mount_id.clone()),
+                },
+                owner,
+            )
+            .await;
+        // (n is created inside the mount; that's a normal mkdir, allowed.)
+        assert!(native.is_ok());
+    }
+
     /// Full read path: owner can list a mount's live contents; a stranger with
     /// no grant is denied. Exercises the REAL authorization cascade
     /// (`authz.require(Resource::Folder(mount_id))`) over ltree ancestry.

src/application/services/mod.rs

@@ -10,6 +10,7 @@ pub mod device_auth_service;
 pub mod drive_management_service;
 pub mod external_identity_service;
 pub mod external_mount_router;
+pub mod external_upload_service;
 pub mod favorites_service;
 pub mod file_lifecycle_service;
 pub mod file_management_service;
@@ -19,6 +20,7 @@ pub mod file_use_case_factory;
 pub mod folder_service;
 pub mod i18n_application_service;
 pub mod magic_link_invite_service;
+pub mod mount_dto;
 pub mod mount_registry;
 pub mod music_service;
 pub mod nextcloud_file_id_service;

src/application/services/mount_dto.rs

@@ -0,0 +1,99 @@
+//! Builders that synthesize `FolderDto` / `FileDto` from a provider [`MountStat`].
+//!
+//! Mount entries have no `storage.folders`/`storage.files` row, so the normal
+//! `FolderDto::from(Folder)` path doesn't apply. These helpers produce the same
+//! DTO shape from a provider stat plus the mount config, with a synthetic `ext:`
+//! id and a virtual etag. Shared by the folder/file services and the handlers.
+
+use std::sync::Arc;
+
+use uuid::Uuid;
+
+use crate::application::dtos::display_helpers::{
+    category_for, format_file_size, icon_class_for, icon_special_class_for,
+};
+use crate::application::dtos::file_dto::FileDto;
+use crate::application::dtos::folder_dto::FolderDto;
+use crate::application::ports::external_mount_ports::MountStat;
+use crate::application::services::mount_registry::MountConfig;
+use crate::domain::services::external_mount_id::{
+    encode_child_id, virtual_file_etag, virtual_folder_etag,
+};
+
+/// Final path segment of a node id (the display name).
+fn node_name(node_id: &str) -> &str {
+    node_id.rsplit('/').next().unwrap_or(node_id)
+}
+
+/// Emit the structured audit line for a mount mutation (per AGENTS.md). Every
+/// write op (upload / mkdir / rename / delete / move) calls this.
+pub fn audit_mount_write(action: &str, cfg: &MountConfig, caller_id: Uuid, node_id: &str) {
+    tracing::info!(
+        target: "audit",
+        event = "external_mount.write",
+        action,
+        mount_id = %cfg.mount_id,
+        caller_id = %caller_id,
+        node_id = %node_id,
+        reason = "external_mount_op",
+        "👮🏻‍♂️ external mount mutation",
+    );
+}
+
+/// The id-string of a mount entry's parent: the parent's `ext:` id, or the
+/// mount-root folder UUID when the entry is a direct child of the root.
+pub fn mount_parent_id(cfg: &MountConfig, node_id: &str) -> String {
+    match node_id.rsplit_once('/') {
+        Some((parent, _)) => encode_child_id(cfg.mount_id, parent),
+        None => cfg.mount_id.to_string(),
+    }
+}
+
+/// Build a `FolderDto` for a mount directory from its stat. `parent_id` is the
+/// id-string of the containing directory (mount-root UUID or an `ext:` id).
+pub fn mount_folder_dto(cfg: &MountConfig, parent_id: &str, stat: &MountStat) -> FolderDto {
+    FolderDto {
+        etag: virtual_folder_etag(stat.modified_at),
+        id: encode_child_id(cfg.mount_id, stat.node_id.clone()),
+        name: node_name(stat.node_id.as_str()).to_owned(),
+        path: String::new(),
+        parent_id: Some(parent_id.to_owned()),
+        owner_id: Some(cfg.owner_id.to_string()),
+        drive_id: cfg.drive_id,
+        created_at: stat.created_at,
+        modified_at: stat.modified_at,
+        is_root: false,
+        icon_class: Arc::from("fas fa-folder"),
+        icon_special_class: Arc::from("folder-icon"),
+        category: Arc::from("Folder"),
+        created_by: None,
+        updated_by: None,
+    }
+}
+
+/// Build a `FileDto` for a mount file from its stat. Virtual files have no blob
+/// hash (`content_hash` empty) and a size+mtime etag.
+pub fn mount_file_dto(cfg: &MountConfig, parent_id: &str, stat: &MountStat) -> FileDto {
+    let name = node_name(stat.node_id.as_str());
+    let mime = stat.mime_type.as_str();
+    FileDto {
+        id: encode_child_id(cfg.mount_id, stat.node_id.clone()),
+        name: name.to_owned(),
+        path: String::new(),
+        size: stat.size,
+        mime_type: Arc::from(mime),
+        folder_id: Some(parent_id.to_owned()),
+        created_at: stat.created_at,
+        modified_at: stat.modified_at,
+        icon_class: Arc::from(icon_class_for(name, mime)),
+        icon_special_class: Arc::from(icon_special_class_for(name, mime)),
+        category: Arc::from(category_for(name, mime)),
+        size_formatted: format_file_size(stat.size),
+        owner_id: Some(cfg.owner_id.to_string()),
+        sort_date: None,
+        content_hash: String::new(),
+        etag: virtual_file_etag(stat.size, stat.modified_at),
+        created_by: None,
+        updated_by: None,
+    }
+}