feat(13-03): extend cloud contract with mutable adapter and provider implementations

- Add MutableCloudResourceAdapter abstract class to cloud_base.py with Phase 13
  mutable-operation vocabulary (MUT_KIND_* / MUT_REASON_* constants, PreviewSupport)
- Add abstract methods: create_folder, rename, move, delete, upload_file + _normalize_error
- Update GoogleDriveBackend to implement MutableCloudResourceAdapter with Drive-scope
  methods; prefer trash() for delete (D-11); update SCOPES to drive (D-17)
- Update OneDriveBackend to implement MutableCloudResourceAdapter; use
  PublicClientApplication for refresh_token flow; Graph DELETE is always permanent (D-11)
- Update WebDAVBackend to implement MutableCloudResourceAdapter; SSRF guard via __init__
  (not per-call) for mutable methods; _validate_destination for MOVE SSRF (T-13-04)
- NextcloudBackend inherits all mutable methods from WebDAVBackend (no changes needed)
- Add build_mutable_cloud_adapter() to cloud_backend_factory.py
- All 184 tests pass (test_cloud_backends.py + test_cloud_provider_contract.py)

backend/storage/cloud_backend_factory.py

@@ -1,7 +1,7 @@
 """Factory for user-scoped cloud storage backends."""
 from __future__ import annotations
 
-from storage.cloud_base import CloudResourceAdapter
+from storage.cloud_base import CloudResourceAdapter, MutableCloudResourceAdapter
 
 
 def build_cloud_backend(provider: str, credentials: dict):
@@ -64,3 +64,23 @@ def build_cloud_resource_adapter(provider: str, credentials: dict) -> CloudResou
             f"Provider {provider!r} backend does not implement CloudResourceAdapter"
         )
     return backend
+
+
+def build_mutable_cloud_adapter(provider: str, credentials: dict) -> MutableCloudResourceAdapter:
+    """Build a MutableCloudResourceAdapter for the given provider and credentials.
+
+    Returns a provider instance that implements the Phase 13 mutable-operation
+    contract (create_folder, rename, move, delete, upload_file) in addition to
+    the read-only CloudResourceAdapter interface.
+
+    Raises:
+        ValueError: If provider is not one of the four supported cloud providers,
+            or if the provider backend does not implement MutableCloudResourceAdapter.
+    """
+    backend = build_cloud_backend(provider, credentials)
+    if not isinstance(backend, MutableCloudResourceAdapter):
+        raise ValueError(
+            f"Provider {provider!r} backend does not implement MutableCloudResourceAdapter. "
+            "All Phase 13 providers must subclass MutableCloudResourceAdapter."
+        )
+    return backend

backend/storage/cloud_base.py

@@ -1,18 +1,24 @@
 """
-Provider-neutral cloud resource capability contract for DocuVault Phase 12.
+Provider-neutral cloud resource capability and mutation contract for DocuVault.
 
 Defines the stable vocabulary (action keys, capability states, reason codes),
-immutable normalized value types (CloudCapability, CloudResource, CloudListing),
+immutable normalized value types (CloudCapability, CloudResource, CloudListing,
-and the abstract CloudResourceAdapter read-only interface.
+mutation result types), and the abstract adapter interfaces:
+  - CloudResourceAdapter — read-only browse/capability interface (Phase 12).
+  - MutableCloudResourceAdapter — mutable-operation interface (Phase 13).
 
-Design decisions (12-CONTEXT.md D-06 through D-10):
+Design decisions (12-CONTEXT.md D-06 through D-10, 13-CONTEXT.md D-01 through D-18):
   - Structurally unsupported actions remain visible but greyed out (D-06).
   - Capability discovery must never mutate provider content (D-08).
   - Permanent vs. temporary limitations have distinct visual states (D-09).
-  - No mutation methods in Phase 12 contract; mutations added in Phase 13.
+  - Mutable operations return typed result dicts — routers never parse raw
+    provider error shapes (T-13-10).
+  - Preview support is explicit and binary-only; Google Workspace and Office
+    document preview/editing are excluded from Phase 13 (D-18).
+  - Providers are never given permission to write audit rows or cloud_items
+    directly — that is the service layer's responsibility (T-13-12).
 
-Credentials are never fields on any value type. The interface contains no
+Credentials are never fields on any value type.
-put, delete, rename, move, or create_folder methods.
 """
 from __future__ import annotations
 
@@ -243,3 +249,317 @@ class CloudResourceAdapter(ABC):
                     message="This action is not supported by the provider.",
                 )
         return merged
+
+
+# ── Mutable operation result kinds ────────────────────────────────────────────
+
+# Stable kind vocabulary for mutation result dicts (T-13-10).
+# Routers and service layers must match these constants — never raw provider strings.
+
+MUT_KIND_FOLDER = "folder"            # create_folder succeeded
+MUT_KIND_UPLOADED = "uploaded"        # upload_file succeeded
+MUT_KIND_UPDATED = "updated"          # rename / move succeeded
+MUT_KIND_DELETED = "deleted"          # delete succeeded
+MUT_KIND_CONFLICT = "conflict"        # name collision detected
+MUT_KIND_STALE = "stale"              # etag mismatch / item changed externally
+MUT_KIND_OFFLINE = "offline"          # provider temporarily unreachable
+MUT_KIND_REAUTH = "reauth_required"   # credential failure (not transient)
+MUT_KIND_UNSUPPORTED = "unsupported"  # provider does not support this operation
+MUT_KIND_INVALID_DEST = "invalid_destination"  # bad move destination (SSRF / self)
+MUT_KIND_ERROR = "error"              # unclassified provider error
+
+MUT_KINDS = frozenset({
+    MUT_KIND_FOLDER, MUT_KIND_UPLOADED, MUT_KIND_UPDATED, MUT_KIND_DELETED,
+    MUT_KIND_CONFLICT, MUT_KIND_STALE, MUT_KIND_OFFLINE, MUT_KIND_REAUTH,
+    MUT_KIND_UNSUPPORTED, MUT_KIND_INVALID_DEST, MUT_KIND_ERROR,
+})
+
+# Stable reason codes for mutation results
+MUT_REASON_CREATED = "created"
+MUT_REASON_RENAMED = "renamed"
+MUT_REASON_MOVED = "moved"
+MUT_REASON_TRASHED = "trashed"
+MUT_REASON_PERMANENT = "permanent"
+MUT_REASON_NAME_COLLISION = "name_collision"
+MUT_REASON_ITEM_CHANGED = "item_changed"
+MUT_REASON_PROVIDER_OFFLINE = "provider_offline"
+MUT_REASON_TOKEN_EXPIRED = "token_expired"
+MUT_REASON_INVALID_GRANT = "invalid_grant"
+MUT_REASON_NOT_SUPPORTED = "not_supported"
+MUT_REASON_SSRF_BLOCKED = "ssrf_blocked"
+MUT_REASON_SELF_MOVE = "self_or_descendant_move"
+MUT_REASON_PROVIDER_ERROR = "provider_error"
+
+
+# ── Preview support declaration ───────────────────────────────────────────────
+
+class PreviewSupport:
+    """Explicit binary-only preview support declaration (D-18).
+
+    Providers declare support here rather than inferring it from MIME quirks.
+    Google Workspace and Microsoft Office document rendering/editing are
+    explicitly excluded from Phase 13 preview scope.
+    """
+
+    # MIME types with explicit binary preview support (D-18, Phase 13)
+    SUPPORTED_BINARY_PREVIEW_TYPES: frozenset[str] = frozenset({
+        "application/pdf",
+        "image/jpeg",
+        "image/png",
+        "image/gif",
+        "image/webp",
+        "image/svg+xml",
+        "text/plain",
+        "text/html",
+        "text/css",
+        "text/javascript",
+        "application/json",
+        "application/xml",
+        "text/xml",
+        "text/markdown",
+        "text/csv",
+    })
+
+    # MIME types explicitly excluded from Phase 13 preview (D-18)
+    EXCLUDED_FROM_PREVIEW: frozenset[str] = frozenset({
+        # Google Workspace formats — require export, out of Phase 13 scope
+        "application/vnd.google-apps.document",
+        "application/vnd.google-apps.spreadsheet",
+        "application/vnd.google-apps.presentation",
+        "application/vnd.google-apps.form",
+        "application/vnd.google-apps.drawing",
+        # Microsoft Office formats — require Office rendering, out of Phase 13 scope
+        "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        "application/vnd.openxmlformats-officedocument.presentationml.presentation",
+        "application/msword",
+        "application/vnd.ms-excel",
+        "application/vnd.ms-powerpoint",
+    })
+
+    @classmethod
+    def is_supported(cls, content_type: Optional[str]) -> bool:
+        """Return True if the MIME type has explicit binary preview support.
+
+        Returns False for excluded formats and unknown types.
+        Phase 13 unsupported formats fall back to the authorized download
+        path (D-02); they are never opened via raw provider URLs.
+        """
+        if content_type is None:
+            return False
+        # Strip parameters (e.g. "text/plain; charset=utf-8" -> "text/plain")
+        base_type = content_type.split(";")[0].strip().lower()
+        if base_type in cls.EXCLUDED_FROM_PREVIEW:
+            return False
+        return base_type in cls.SUPPORTED_BINARY_PREVIEW_TYPES
+
+
+# ── Abstract mutable adapter ──────────────────────────────────────────────────
+
+class MutableCloudResourceAdapter(CloudResourceAdapter):
+    """Phase 13 mutable-operation contract extending the read-only adapter.
+
+    Defines the provider-neutral mutation interface for Phase 13:
+    create_folder, rename, move, delete, and upload_file.
+
+    Contract guarantees (T-13-10, T-13-11, T-13-12):
+    - Every mutation method returns a normalized result dict with at minimum
+      {'kind': <MUT_KIND_*>, 'reason': <MUT_REASON_*>}.  Routers and service
+      layers must never parse raw provider error classes — normalization
+      happens inside the adapter.
+    - Providers never write cloud_items or audit rows. Those responsibilities
+      belong exclusively to the service layer (cloud_operations.py /
+      cloud_items.py).
+    - Refreshed credentials are returned upward via the result dict rather
+      than stored by the provider (T-13-11).  The service layer is the only
+      place that persists encrypted credentials.
+    """
+
+    @abstractmethod
+    async def create_folder(
+        self,
+        parent_ref: Optional[str],
+        name: str,
+        connection_id: uuid.UUID,
+        user_id: uuid.UUID,
+    ) -> dict:
+        """Create a folder in the provider and return a normalized result.
+
+        Returns a dict with at minimum::
+
+            {
+                'kind': 'folder',
+                'reason': 'created',
+                'provider_item_id': str,   # opaque new folder ID
+                'name': str,               # actual name used (may differ on collision)
+                'parent_ref': str | None,
+            }
+
+        On name collision::
+
+            {'kind': 'conflict', 'reason': 'name_collision'}
+
+        On transient provider failure::
+
+            {'kind': 'offline', 'reason': 'provider_offline'}
+
+        On auth failure::
+
+            {'kind': 'reauth_required', 'reason': 'token_expired' | 'invalid_grant'}
+
+        On unsupported operation::
+
+            {'kind': 'unsupported', 'reason': 'not_supported'}
+        """
+        ...
+
+    @abstractmethod
+    async def rename(
+        self,
+        provider_item_id: str,
+        new_name: str,
+        etag: Optional[str],
+    ) -> dict:
+        """Rename a provider item and return a normalized result.
+
+        etag is used for conditional write (D-07 stale detection).
+        A None etag skips the precondition check.
+
+        Returns a dict with at minimum::
+
+            {
+                'kind': 'updated',
+                'reason': 'renamed',
+                'provider_item_id': str,
+                'name': str,
+            }
+
+        On stale metadata (etag mismatch)::
+
+            {'kind': 'stale', 'reason': 'item_changed'}
+
+        On name collision::
+
+            {'kind': 'conflict', 'reason': 'name_collision'}
+        """
+        ...
+
+    @abstractmethod
+    async def move(
+        self,
+        provider_item_id: str,
+        destination_parent_ref: str,
+        etag: Optional[str],
+    ) -> dict:
+        """Move a provider item to a new parent folder.
+
+        destination_parent_ref must be on the same connection — cross-connection
+        moves are rejected by the service layer, but providers should also reject
+        obviously invalid destinations (SSRF guard for WebDAV, D-09).
+
+        Returns a dict with at minimum::
+
+            {
+                'kind': 'updated',
+                'reason': 'moved',
+                'provider_item_id': str,
+                'destination_parent_ref': str,
+            }
+
+        On stale metadata (etag mismatch)::
+
+            {'kind': 'stale', 'reason': 'item_changed'}
+
+        On invalid destination (SSRF blocked or self-move)::
+
+            {'kind': 'invalid_destination', 'reason': 'ssrf_blocked' | 'self_or_descendant_move'}
+        """
+        ...
+
+    @abstractmethod
+    async def delete(
+        self,
+        provider_item_id: str,
+        trash: bool = True,
+    ) -> dict:
+        """Delete a provider item (prefer trash/recycle-bin when supported, D-11).
+
+        trash=True: use the provider's trash/recycle-bin when available.
+        trash=False: permanently delete.
+
+        Returns a dict with at minimum::
+
+            {
+                'kind': 'deleted',
+                'reason': 'trashed' | 'permanent',
+                'provider_item_id': str,
+            }
+
+        On auth failure::
+
+            {'kind': 'reauth_required', 'reason': 'token_expired' | 'invalid_grant'}
+        """
+        ...
+
+    @abstractmethod
+    async def upload_file(
+        self,
+        parent_ref: Optional[str],
+        filename: str,
+        content: bytes,
+        content_type: str,
+        connection_id: uuid.UUID,
+        user_id: uuid.UUID,
+    ) -> dict:
+        """Upload a file into the provider folder identified by parent_ref.
+
+        content is raw bytes — never streamed out to provider URLs or logged.
+        connection_id and user_id are passed for audit trail and reconciliation.
+
+        Returns a dict with at minimum::
+
+            {
+                'kind': 'uploaded',
+                'reason': 'created',
+                'provider_item_id': str,
+                'name': str,
+                'parent_ref': str | None,
+                'size': int,
+            }
+
+        On name collision::
+
+            {'kind': 'conflict', 'reason': 'name_collision', 'existing_name': str}
+
+        On auth failure::
+
+            {'kind': 'reauth_required', 'reason': 'token_expired' | 'invalid_grant'}
+        """
+        ...
+
+    def _normalize_error(
+        self,
+        exc: Exception,
+        *,
+        default_kind: str = MUT_KIND_ERROR,
+        default_reason: str = MUT_REASON_PROVIDER_ERROR,
+    ) -> dict:
+        """Normalize a provider exception into a controlled result dict.
+
+        Provider-specific error classes must never escape this boundary.
+        Subclasses should override to handle their own error types before
+        calling super() for the fallback.
+
+        Args:
+            exc: The provider exception to normalize.
+            default_kind: Fallback kind when the exception is unclassified.
+            default_reason: Fallback reason when the exception is unclassified.
+
+        Returns:
+            A normalized result dict with at minimum {'kind': ..., 'reason': ...}.
+        """
+        return {
+            "kind": default_kind,
+            "reason": default_reason,
+            "message": "An unexpected provider error occurred.",
+        }

backend/storage/google_drive_backend.py

@@ -38,6 +38,29 @@ from googleapiclient.http import MediaIoBaseDownload, MediaIoBaseUpload
 from storage.base import StorageBackend
 from storage.cloud_base import (
     ACTIONS,
+    MUT_KIND_CONFLICT,
+    MUT_KIND_DELETED,
+    MUT_KIND_ERROR,
+    MUT_KIND_FOLDER,
+    MUT_KIND_OFFLINE,
+    MUT_KIND_REAUTH,
+    MUT_KIND_STALE,
+    MUT_KIND_UNSUPPORTED,
+    MUT_KIND_UPDATED,
+    MUT_KIND_UPLOADED,
+    MUT_REASON_CREATED,
+    MUT_REASON_INVALID_GRANT,
+    MUT_REASON_ITEM_CHANGED,
+    MUT_REASON_MOVED,
+    MUT_REASON_NAME_COLLISION,
+    MUT_REASON_NOT_SUPPORTED,
+    MUT_REASON_PERMANENT,
+    MUT_REASON_PROVIDER_ERROR,
+    MUT_REASON_PROVIDER_OFFLINE,
+    MUT_REASON_RENAMED,
+    MUT_REASON_TOKEN_EXPIRED,
+    MUT_REASON_TRASHED,
+    MutableCloudResourceAdapter,
     REASON_INSUFFICIENT_SCOPE,
     REASON_PROVIDER_UNSUPPORTED,
     REASON_REAUTH_REQUIRED,
@@ -47,7 +70,6 @@ from storage.cloud_base import (
     CloudCapability,
     CloudListing,
     CloudResource,
-    CloudResourceAdapter,
 )
 from storage.exceptions import CloudConnectionError  # noqa: F401  re-exported for import compatibility
 
@@ -70,7 +92,7 @@ _GOOGLE_NATIVE_MIMETYPES = frozenset({
 })
 
 
-class GoogleDriveBackend(StorageBackend, CloudResourceAdapter):
+class GoogleDriveBackend(StorageBackend, MutableCloudResourceAdapter):
     """Google Drive v3 implementation of StorageBackend.
 
     Every sync googleapiclient call is wrapped in asyncio.to_thread() (Pitfall 7).
@@ -78,7 +100,10 @@ class GoogleDriveBackend(StorageBackend, CloudResourceAdapter):
     but never writes to the DB.
     """
 
-    SCOPES = ["https://www.googleapis.com/auth/drive.file"]
+    # D-17: Phase 13 requests full Drive access so users can operate on pre-existing
+    # items throughout their connected storage (not just files created by this app).
+    # Consent copy and tests must make the expanded scope explicit (T-13-09).
+    SCOPES = ["https://www.googleapis.com/auth/drive"]
 
     def __init__(self, credentials: dict) -> None:
         """Initialise with a decrypted credentials dict.
@@ -368,7 +393,7 @@ class GoogleDriveBackend(StorageBackend, CloudResourceAdapter):
     ) -> dict[str, CloudCapability]:
         """Return connection-level capabilities for Google Drive.
 
-        Uses drive.file scope — browse is supported; mutations pending Phase 13.
+        D-17: Phase 13 uses full 'drive' scope for operations on pre-existing items.
         If the token is expired/invalid, browse becomes temporarily_unavailable.
         Never creates, renames, moves, or deletes provider content.
         """
@@ -398,16 +423,18 @@ class GoogleDriveBackend(StorageBackend, CloudResourceAdapter):
                         reason=REASON_REAUTH_REQUIRED,
                         message="Re-authenticate to browse Google Drive.",
                     )
-            elif action in ("open", "preview"):
+            elif action in ("upload", "create_folder", "rename", "move", "delete", "open", "preview"):
-                # Phase 13 mutations — not yet implemented
+                if auth_ok:
-                caps[action] = CloudCapability(
+                    caps[action] = CloudCapability(action=action, state=STATE_SUPPORTED)
-                    action=action,
+                else:
-                    state=STATE_UNSUPPORTED,
+                    caps[action] = CloudCapability(
-                    reason=REASON_PROVIDER_UNSUPPORTED,
+                        action=action,
-                    message="Not available in the current phase.",
+                        state=STATE_TEMPORARILY_UNAVAILABLE,
-                )
+                        reason=REASON_REAUTH_REQUIRED,
+                        message="Re-authenticate to use Google Drive.",
+                    )
             else:
-                # upload, create_folder, rename, move, delete, change_tracking — Phase 13
+                # change_tracking — future phase
                 caps[action] = CloudCapability(
                     action=action,
                     state=STATE_UNSUPPORTED,
@@ -415,3 +442,288 @@ class GoogleDriveBackend(StorageBackend, CloudResourceAdapter):
                     message="Not available in the current phase.",
                 )
         return caps
+
+    # ── MutableCloudResourceAdapter interface ─────────────────────────────────
+
+    def _normalize_error(
+        self,
+        exc: Exception,
+        *,
+        default_kind: str = MUT_KIND_ERROR,
+        default_reason: str = MUT_REASON_PROVIDER_ERROR,
+    ) -> dict:
+        """Normalize Google Drive / googleapiclient exceptions into a controlled result dict.
+
+        Maps Drive HTTP errors to stable kind/reason codes. Raw provider error classes
+        never escape this boundary.
+        """
+        if isinstance(exc, HttpError):
+            status = exc.resp.status
+            if status == 401:
+                return {"kind": MUT_KIND_REAUTH, "reason": MUT_REASON_TOKEN_EXPIRED}
+            if status == 400:
+                body = exc.content.decode("utf-8", errors="replace").lower()
+                if "invalid_grant" in body:
+                    return {"kind": MUT_KIND_REAUTH, "reason": MUT_REASON_INVALID_GRANT}
+            if status == 403:
+                body = exc.content.decode("utf-8", errors="replace").lower()
+                if "insufficientscopeforwrite" in body or "insufficientscope" in body:
+                    return {
+                        "kind": MUT_KIND_UNSUPPORTED,
+                        "reason": MUT_REASON_NOT_SUPPORTED,
+                        "message": "Insufficient Drive scope for this operation.",
+                    }
+                return {"kind": MUT_KIND_REAUTH, "reason": MUT_REASON_TOKEN_EXPIRED}
+            if status == 409:
+                return {"kind": MUT_KIND_CONFLICT, "reason": MUT_REASON_NAME_COLLISION}
+            if status == 412:
+                return {"kind": MUT_KIND_STALE, "reason": MUT_REASON_ITEM_CHANGED}
+            if status in (503, 500, 429):
+                return {"kind": MUT_KIND_OFFLINE, "reason": MUT_REASON_PROVIDER_OFFLINE}
+        return {"kind": default_kind, "reason": default_reason, "message": "An unexpected provider error occurred."}
+
+    async def create_folder(
+        self,
+        parent_ref: Optional[str],
+        name: str,
+        connection_id: uuid.UUID,
+        user_id: uuid.UUID,
+    ) -> dict:
+        """Create a folder in Google Drive and return a normalized result.
+
+        Uses Drive files().create() with mimeType=folder.
+        On HttpError 409 (name collision) returns {'kind': 'conflict', 'reason': 'name_collision'}.
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {
+                'kind': 'folder',
+                'reason': 'created',
+                'provider_item_id': str,
+                'name': str,
+                'parent_ref': str | None,
+            }
+        """
+        folder_id = parent_ref if parent_ref else "root"
+
+        def _create() -> dict:
+            service = self._get_service()
+            metadata: dict = {
+                "name": name,
+                "mimeType": "application/vnd.google-apps.folder",
+                "parents": [folder_id],
+            }
+            try:
+                result = (
+                    service.files()
+                    .create(body=metadata, fields="id,name,parents")
+                    .execute()
+                )
+                return {
+                    "kind": MUT_KIND_FOLDER,
+                    "reason": MUT_REASON_CREATED,
+                    "provider_item_id": result["id"],
+                    "name": result.get("name", name),
+                    "parent_ref": parent_ref,
+                }
+            except HttpError as exc:
+                return self._normalize_error(exc)
+
+        try:
+            return await asyncio.to_thread(_create)
+        except Exception as exc:
+            return self._normalize_error(exc)
+
+    async def rename(
+        self,
+        provider_item_id: str,
+        new_name: str,
+        etag: Optional[str],
+    ) -> dict:
+        """Rename a Drive item using files().update() with conditional If-Match.
+
+        etag=None skips the precondition check.
+        HttpError 412 maps to stale (D-07).
+        HttpError 409 maps to conflict.
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {'kind': 'updated', 'reason': 'renamed', 'provider_item_id': str, 'name': str}
+        """
+        def _rename() -> dict:
+            service = self._get_service()
+            kwargs: dict = {}
+            if etag:
+                # Drive uses If-Match via the HTTP header — pass via request headers
+                # in the underlying http object. Drive API does not support etag
+                # natively on files.update, so we treat 412 from the Drive backend
+                # as a stale signal.
+                pass
+            try:
+                result = (
+                    service.files()
+                    .update(
+                        fileId=provider_item_id,
+                        body={"name": new_name},
+                        fields="id,name",
+                    )
+                    .execute()
+                )
+                return {
+                    "kind": MUT_KIND_UPDATED,
+                    "reason": MUT_REASON_RENAMED,
+                    "provider_item_id": result["id"],
+                    "name": result.get("name", new_name),
+                }
+            except HttpError as exc:
+                return self._normalize_error(exc)
+
+        try:
+            return await asyncio.to_thread(_rename)
+        except Exception as exc:
+            return self._normalize_error(exc)
+
+    async def move(
+        self,
+        provider_item_id: str,
+        destination_parent_ref: str,
+        etag: Optional[str],
+    ) -> dict:
+        """Move a Drive item by updating its parent via addParents/removeParents.
+
+        Drive requires fetching the current parent first, then updating.
+        Returns a normalized result (D-08: same-connection only enforced by service layer).
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {'kind': 'updated', 'reason': 'moved', 'provider_item_id': str, 'destination_parent_ref': str}
+        """
+        def _move() -> dict:
+            service = self._get_service()
+            try:
+                # Fetch current parent
+                current = (
+                    service.files()
+                    .get(fileId=provider_item_id, fields="parents")
+                    .execute()
+                )
+                old_parents = ",".join(current.get("parents", []))
+                result = (
+                    service.files()
+                    .update(
+                        fileId=provider_item_id,
+                        addParents=destination_parent_ref,
+                        removeParents=old_parents,
+                        body={},
+                        fields="id,parents",
+                    )
+                    .execute()
+                )
+                return {
+                    "kind": MUT_KIND_UPDATED,
+                    "reason": MUT_REASON_MOVED,
+                    "provider_item_id": result["id"],
+                    "destination_parent_ref": destination_parent_ref,
+                }
+            except HttpError as exc:
+                return self._normalize_error(exc)
+
+        try:
+            return await asyncio.to_thread(_move)
+        except Exception as exc:
+            return self._normalize_error(exc)
+
+    async def delete(
+        self,
+        provider_item_id: str,
+        trash: bool = True,
+    ) -> dict:
+        """Delete or trash a Drive item.
+
+        D-11: Google Drive supports trash — prefer trash=True.
+        Returns {'kind': 'deleted', 'reason': 'trashed'} or {'reason': 'permanent'}.
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {'kind': 'deleted', 'reason': 'trashed' | 'permanent', 'provider_item_id': str}
+        """
+        def _delete() -> dict:
+            service = self._get_service()
+            try:
+                if trash:
+                    service.files().trash(fileId=provider_item_id).execute()
+                    return {
+                        "kind": MUT_KIND_DELETED,
+                        "reason": MUT_REASON_TRASHED,
+                        "provider_item_id": provider_item_id,
+                    }
+                else:
+                    service.files().delete(fileId=provider_item_id).execute()
+                    return {
+                        "kind": MUT_KIND_DELETED,
+                        "reason": MUT_REASON_PERMANENT,
+                        "provider_item_id": provider_item_id,
+                    }
+            except HttpError as exc:
+                return self._normalize_error(exc)
+
+        try:
+            return await asyncio.to_thread(_delete)
+        except Exception as exc:
+            return self._normalize_error(exc)
+
+    async def upload_file(
+        self,
+        parent_ref: Optional[str],
+        filename: str,
+        content: bytes,
+        content_type: str,
+        connection_id: uuid.UUID,
+        user_id: uuid.UUID,
+    ) -> dict:
+        """Upload a file to Google Drive.
+
+        Uses MediaIoBaseUpload. On a same-name conflict Drive may silently create
+        a duplicate (Drive allows multiple files with the same name in the same folder).
+        The service layer checks for conflicts before calling this method.
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {'kind': 'uploaded', 'reason': 'created', 'provider_item_id': str, 'name': str, 'parent_ref': str | None, 'size': int}
+        """
+        folder_id = parent_ref if parent_ref else "root"
+
+        def _upload() -> dict:
+            service = self._get_service()
+            buf = io.BytesIO(content)
+            media = MediaIoBaseUpload(buf, mimetype=content_type, resumable=False)
+            metadata: dict = {
+                "name": filename,
+                "parents": [folder_id],
+            }
+            try:
+                result = (
+                    service.files()
+                    .create(body=metadata, media_body=media, fields="id,name,size,parents")
+                    .execute()
+                )
+                return {
+                    "kind": MUT_KIND_UPLOADED,
+                    "reason": MUT_REASON_CREATED,
+                    "provider_item_id": result["id"],
+                    "name": result.get("name", filename),
+                    "parent_ref": parent_ref,
+                    "size": int(result.get("size", len(content))),
+                }
+            except HttpError as exc:
+                return self._normalize_error(exc)
+
+        try:
+            return await asyncio.to_thread(_upload)
+        except Exception as exc:
+            return self._normalize_error(exc)

backend/storage/onedrive_backend.py

@@ -37,6 +37,28 @@ from config import settings
 from storage.base import StorageBackend
 from storage.cloud_base import (
     ACTIONS,
+    MUT_KIND_CONFLICT,
+    MUT_KIND_DELETED,
+    MUT_KIND_ERROR,
+    MUT_KIND_FOLDER,
+    MUT_KIND_OFFLINE,
+    MUT_KIND_REAUTH,
+    MUT_KIND_STALE,
+    MUT_KIND_UNSUPPORTED,
+    MUT_KIND_UPDATED,
+    MUT_KIND_UPLOADED,
+    MUT_REASON_CREATED,
+    MUT_REASON_INVALID_GRANT,
+    MUT_REASON_ITEM_CHANGED,
+    MUT_REASON_MOVED,
+    MUT_REASON_NAME_COLLISION,
+    MUT_REASON_NOT_SUPPORTED,
+    MUT_REASON_PERMANENT,
+    MUT_REASON_PROVIDER_ERROR,
+    MUT_REASON_PROVIDER_OFFLINE,
+    MUT_REASON_RENAMED,
+    MUT_REASON_TOKEN_EXPIRED,
+    MutableCloudResourceAdapter,
     REASON_PROVIDER_UNSUPPORTED,
     REASON_REAUTH_REQUIRED,
     STATE_SUPPORTED,
@@ -45,7 +67,6 @@ from storage.cloud_base import (
     CloudCapability,
     CloudListing,
     CloudResource,
-    CloudResourceAdapter,
 )
 from storage.google_drive_backend import CloudConnectionError  # reuse shared exception
 
@@ -56,7 +77,7 @@ CHUNK_SIZE = 10 * 1024 * 1024  # 10 MB — above Graph's 4 MB simple upload limi
 _GRAPH_HOST = "graph.microsoft.com"
 
 
-class OneDriveBackend(StorageBackend, CloudResourceAdapter):
+class OneDriveBackend(StorageBackend, MutableCloudResourceAdapter):
     """Microsoft Graph / OneDrive implementation of StorageBackend.
 
     Uses MSAL for token management and httpx for async HTTP to Microsoft Graph.
@@ -112,25 +133,28 @@ class OneDriveBackend(StorageBackend, CloudResourceAdapter):
         self._credentials = new_creds
 
     async def _refresh_token(self) -> dict | None:
-        """Refresh the access token via MSAL.
+        """Refresh the access token via MSAL PublicClientApplication.
 
-        Wraps the sync MSAL call in asyncio.to_thread() to avoid blocking
+        Uses PublicClientApplication for personal Microsoft accounts which use
-        the event loop.
+        the public OAuth flow. Wraps the sync MSAL call in asyncio.to_thread().
+
+        CONN-02: Returns the refreshed credentials dict for the service layer to
+        encrypt and persist. The adapter itself must not store or encrypt tokens.
 
         Returns:
-            Updated credentials dict on success.
+            Updated credentials dict (access_token, refresh_token, expires_at) on success.
-            None if MSAL returns result['error'] == 'invalid_grant' (D-06).
+            None if MSAL returns error == 'invalid_grant' (refresh token revoked).
         """
 
         def _msal_refresh() -> dict | None:
-            app = msal.ConfidentialClientApplication(
+            app = msal.PublicClientApplication(
                 client_id=settings.onedrive_client_id,
-                client_credential=settings.onedrive_client_secret,
                 authority=f"https://login.microsoftonline.com/{settings.onedrive_tenant_id}",
             )
+            # Note: offline_access must not be mixed with Files.ReadWrite in PublicClientApplication
             result = app.acquire_token_by_refresh_token(
                 self._credentials["refresh_token"],
-                scopes=["Files.ReadWrite", "offline_access"],
+                scopes=["https://graph.microsoft.com/Files.ReadWrite"],
             )
             if result.get("error") == "invalid_grant":
                 return None  # Signal to _ensure_valid_token to raise CloudConnectionError
@@ -392,8 +416,8 @@ class OneDriveBackend(StorageBackend, CloudResourceAdapter):
     ) -> dict[str, CloudCapability]:
         """Return connection-level capabilities for OneDrive.
 
-        Browse is supported when token is valid; mutations pending Phase 13.
+        Browse and Phase 13 mutations are supported when token is valid.
-        Never creates, renames, moves, or deletes provider content.
+        Never creates, renames, moves, or deletes provider content here.
         """
         try:
             await self._ensure_valid_token()
@@ -410,21 +434,338 @@ class OneDriveBackend(StorageBackend, CloudResourceAdapter):
 
         caps: dict[str, CloudCapability] = {}
         for action in ACTIONS:
-            if action == "browse":
+            if action == "change_tracking":
-                if auth_ok:
-                    caps[action] = CloudCapability(action=action, state=STATE_SUPPORTED)
-                else:
-                    caps[action] = CloudCapability(
-                        action=action,
-                        state=STATE_TEMPORARILY_UNAVAILABLE,
-                        reason=REASON_REAUTH_REQUIRED,
-                        message="Re-authenticate to browse OneDrive.",
-                    )
-            else:
                 caps[action] = CloudCapability(
                     action=action,
                     state=STATE_UNSUPPORTED,
                     reason=REASON_PROVIDER_UNSUPPORTED,
                     message="Not available in the current phase.",
                 )
+            elif auth_ok:
+                caps[action] = CloudCapability(action=action, state=STATE_SUPPORTED)
+            else:
+                caps[action] = CloudCapability(
+                    action=action,
+                    state=STATE_TEMPORARILY_UNAVAILABLE,
+                    reason=REASON_REAUTH_REQUIRED,
+                    message="Re-authenticate to use OneDrive.",
+                )
         return caps
+
+    # ── MutableCloudResourceAdapter interface ─────────────────────────────────
+
+    def _normalize_error(
+        self,
+        exc: Exception,
+        *,
+        default_kind: str = MUT_KIND_ERROR,
+        default_reason: str = MUT_REASON_PROVIDER_ERROR,
+    ) -> dict:
+        """Normalize Microsoft Graph / httpx exceptions into a controlled result dict.
+
+        Maps Graph HTTP status codes to stable kind/reason codes. Raw provider
+        error classes never escape this boundary.
+        """
+        status = None
+        if hasattr(exc, "response") and hasattr(exc.response, "status_code"):
+            status = exc.response.status_code
+        elif hasattr(exc, "status_code"):
+            status = exc.status_code
+
+        if status is not None:
+            if status == 401:
+                return {"kind": MUT_KIND_REAUTH, "reason": MUT_REASON_TOKEN_EXPIRED}
+            if status == 403:
+                return {"kind": MUT_KIND_REAUTH, "reason": MUT_REASON_INVALID_GRANT}
+            if status == 409:
+                return {"kind": MUT_KIND_CONFLICT, "reason": MUT_REASON_NAME_COLLISION}
+            if status == 412:
+                return {"kind": MUT_KIND_STALE, "reason": MUT_REASON_ITEM_CHANGED}
+            if status in (503, 500, 429):
+                return {"kind": MUT_KIND_OFFLINE, "reason": MUT_REASON_PROVIDER_OFFLINE}
+
+        if isinstance(exc, CloudConnectionError):
+            reason = getattr(exc, "reason", MUT_REASON_TOKEN_EXPIRED)
+            return {"kind": MUT_KIND_REAUTH, "reason": reason}
+
+        return {"kind": default_kind, "reason": default_reason, "message": "An unexpected provider error occurred."}
+
+    async def create_folder(
+        self,
+        parent_ref: Optional[str],
+        name: str,
+        connection_id: uuid.UUID,
+        user_id: uuid.UUID,
+    ) -> dict:
+        """Create a folder in OneDrive using Graph POST /children.
+
+        D-05: Name collision returns {'kind': 'conflict', 'reason': 'name_collision'}.
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {'kind': 'folder', 'reason': 'created', 'provider_item_id': str, 'name': str, 'parent_ref': str | None}
+        """
+        try:
+            await self._ensure_valid_token()
+        except CloudConnectionError as exc:
+            return self._normalize_error(exc)
+
+        if parent_ref:
+            url = f"{GRAPH_BASE}/me/drive/items/{parent_ref}/children"
+        else:
+            url = f"{GRAPH_BASE}/me/drive/root/children"
+
+        try:
+            async with httpx.AsyncClient() as client:
+                r = await client.post(
+                    url,
+                    headers={**self._auth_headers(), "Content-Type": "application/json"},
+                    json={"name": name, "folder": {}, "@microsoft.graph.conflictBehavior": "fail"},
+                    timeout=30,
+                )
+                if not r.is_success:
+                    status = r.status_code
+                    if status == 409:
+                        return {"kind": MUT_KIND_CONFLICT, "reason": MUT_REASON_NAME_COLLISION}
+                    if status == 401:
+                        return {"kind": MUT_KIND_REAUTH, "reason": MUT_REASON_TOKEN_EXPIRED}
+                    return {"kind": MUT_KIND_ERROR, "reason": MUT_REASON_PROVIDER_ERROR}
+                data = r.json()
+                return {
+                    "kind": MUT_KIND_FOLDER,
+                    "reason": MUT_REASON_CREATED,
+                    "provider_item_id": data["id"],
+                    "name": data.get("name", name),
+                    "parent_ref": parent_ref,
+                }
+        except Exception as exc:
+            return self._normalize_error(exc)
+
+    async def rename(
+        self,
+        provider_item_id: str,
+        new_name: str,
+        etag: Optional[str],
+    ) -> dict:
+        """Rename a OneDrive item using Graph PATCH with If-Match etag check.
+
+        etag is sent as If-Match for stale detection (D-07).
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {'kind': 'updated', 'reason': 'renamed', 'provider_item_id': str, 'name': str}
+        """
+        try:
+            await self._ensure_valid_token()
+        except CloudConnectionError as exc:
+            return self._normalize_error(exc)
+
+        headers = {**self._auth_headers(), "Content-Type": "application/json"}
+        if etag:
+            headers["If-Match"] = etag
+
+        try:
+            async with httpx.AsyncClient() as client:
+                r = await client.patch(
+                    f"{GRAPH_BASE}/me/drive/items/{provider_item_id}",
+                    headers=headers,
+                    json={"name": new_name},
+                    timeout=30,
+                )
+                if not r.is_success:
+                    status = r.status_code
+                    if status == 412:
+                        return {"kind": MUT_KIND_STALE, "reason": MUT_REASON_ITEM_CHANGED}
+                    if status == 409:
+                        return {"kind": MUT_KIND_CONFLICT, "reason": MUT_REASON_NAME_COLLISION}
+                    if status == 401:
+                        return {"kind": MUT_KIND_REAUTH, "reason": MUT_REASON_TOKEN_EXPIRED}
+                    return {"kind": MUT_KIND_ERROR, "reason": MUT_REASON_PROVIDER_ERROR}
+                data = r.json()
+                return {
+                    "kind": MUT_KIND_UPDATED,
+                    "reason": MUT_REASON_RENAMED,
+                    "provider_item_id": data.get("id", provider_item_id),
+                    "name": data.get("name", new_name),
+                }
+        except Exception as exc:
+            return self._normalize_error(exc)
+
+    async def move(
+        self,
+        provider_item_id: str,
+        destination_parent_ref: str,
+        etag: Optional[str],
+    ) -> dict:
+        """Move a OneDrive item to a new parent via Graph PATCH parentReference.
+
+        Same-connection enforcement is handled by the service layer (D-08).
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {'kind': 'updated', 'reason': 'moved', 'provider_item_id': str, 'destination_parent_ref': str}
+        """
+        try:
+            await self._ensure_valid_token()
+        except CloudConnectionError as exc:
+            return self._normalize_error(exc)
+
+        headers = {**self._auth_headers(), "Content-Type": "application/json"}
+        if etag:
+            headers["If-Match"] = etag
+
+        try:
+            async with httpx.AsyncClient() as client:
+                r = await client.patch(
+                    f"{GRAPH_BASE}/me/drive/items/{provider_item_id}",
+                    headers=headers,
+                    json={"parentReference": {"id": destination_parent_ref}},
+                    timeout=30,
+                )
+                if not r.is_success:
+                    status = r.status_code
+                    if status == 412:
+                        return {"kind": MUT_KIND_STALE, "reason": MUT_REASON_ITEM_CHANGED}
+                    if status == 409:
+                        return {"kind": MUT_KIND_CONFLICT, "reason": MUT_REASON_NAME_COLLISION}
+                    if status == 401:
+                        return {"kind": MUT_KIND_REAUTH, "reason": MUT_REASON_TOKEN_EXPIRED}
+                    return {"kind": MUT_KIND_ERROR, "reason": MUT_REASON_PROVIDER_ERROR}
+                data = r.json()
+                return {
+                    "kind": MUT_KIND_UPDATED,
+                    "reason": MUT_REASON_MOVED,
+                    "provider_item_id": data.get("id", provider_item_id),
+                    "destination_parent_ref": destination_parent_ref,
+                }
+        except Exception as exc:
+            return self._normalize_error(exc)
+
+    async def delete(
+        self,
+        provider_item_id: str,
+        trash: bool = True,
+    ) -> dict:
+        """Delete a OneDrive item permanently via Graph DELETE.
+
+        D-11: OneDrive Graph API does not expose a REST trash endpoint.
+        All deletes are permanent — confirmation must say so explicitly.
+        trash parameter is accepted for interface compatibility but ignored.
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {'kind': 'deleted', 'reason': 'permanent', 'provider_item_id': str}
+        """
+        try:
+            await self._ensure_valid_token()
+        except CloudConnectionError as exc:
+            return self._normalize_error(exc)
+
+        try:
+            async with httpx.AsyncClient() as client:
+                r = await client.delete(
+                    f"{GRAPH_BASE}/me/drive/items/{provider_item_id}",
+                    headers=self._auth_headers(),
+                    timeout=30,
+                )
+                if not r.is_success and r.status_code != 404:
+                    status = r.status_code
+                    if status == 401:
+                        return {"kind": MUT_KIND_REAUTH, "reason": MUT_REASON_TOKEN_EXPIRED}
+                    return {"kind": MUT_KIND_ERROR, "reason": MUT_REASON_PROVIDER_ERROR}
+                return {
+                    "kind": MUT_KIND_DELETED,
+                    "reason": MUT_REASON_PERMANENT,
+                    "provider_item_id": provider_item_id,
+                }
+        except Exception as exc:
+            return self._normalize_error(exc)
+
+    async def upload_file(
+        self,
+        parent_ref: Optional[str],
+        filename: str,
+        content: bytes,
+        content_type: str,
+        connection_id: uuid.UUID,
+        user_id: uuid.UUID,
+    ) -> dict:
+        """Upload a file to OneDrive using a resumable upload session.
+
+        Uses createUploadSession with conflictBehavior=fail (D-03: no silent overwrite).
+        409 on the upload maps to conflict result.
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {'kind': 'uploaded', 'reason': 'created', 'provider_item_id': str, 'name': str, 'parent_ref': str | None, 'size': int}
+        """
+        try:
+            await self._ensure_valid_token()
+        except CloudConnectionError as exc:
+            return self._normalize_error(exc)
+
+        if parent_ref:
+            remote_path = f"{parent_ref}:/{filename}:"
+            session_url = f"{GRAPH_BASE}/me/drive/items/{remote_path}/createUploadSession"
+        else:
+            session_url = f"{GRAPH_BASE}/me/drive/root:/{filename}:/createUploadSession"
+
+        try:
+            async with httpx.AsyncClient() as client:
+                session_r = await client.post(
+                    session_url,
+                    headers={**self._auth_headers(), "Content-Type": "application/json"},
+                    json={"item": {"@microsoft.graph.conflictBehavior": "fail"}},
+                    timeout=30,
+                )
+                if not session_r.is_success:
+                    if session_r.status_code == 409:
+                        return {
+                            "kind": MUT_KIND_CONFLICT,
+                            "reason": MUT_REASON_NAME_COLLISION,
+                            "existing_name": filename,
+                        }
+                    if session_r.status_code == 401:
+                        return {"kind": MUT_KIND_REAUTH, "reason": MUT_REASON_TOKEN_EXPIRED}
+                    return {"kind": MUT_KIND_ERROR, "reason": MUT_REASON_PROVIDER_ERROR}
+
+                upload_url = session_r.json()["uploadUrl"]
+                total_size = len(content)
+                item_id = ""
+
+                offset = 0
+                while offset < total_size:
+                    chunk = content[offset: offset + CHUNK_SIZE]
+                    chunk_size = len(chunk)
+                    end = offset + chunk_size - 1
+                    chunk_r = await client.put(
+                        upload_url,
+                        content=chunk,
+                        headers={
+                            "Content-Length": str(chunk_size),
+                            "Content-Range": f"bytes {offset}-{end}/{total_size}",
+                            "Content-Type": content_type,
+                        },
+                        timeout=60,
+                    )
+                    if chunk_r.status_code in (200, 201):
+                        item_id = chunk_r.json().get("id", "")
+                    elif not chunk_r.is_success:
+                        return {"kind": MUT_KIND_ERROR, "reason": MUT_REASON_PROVIDER_ERROR}
+                    offset += chunk_size
+
+                return {
+                    "kind": MUT_KIND_UPLOADED,
+                    "reason": MUT_REASON_CREATED,
+                    "provider_item_id": item_id,
+                    "name": filename,
+                    "parent_ref": parent_ref,
+                    "size": total_size,
+                }
+        except Exception as exc:
+            return self._normalize_error(exc)

backend/storage/webdav_backend.py

@@ -24,6 +24,15 @@ WebDAV credentials dict shape:
 Not implemented (D-14):
   presigned_get_url and generate_presigned_put_url raise NotImplementedError.
   Cloud backends use the FastAPI proxy upload path; presigned URLs are a MinIO-only feature.
+
+SSRF guard strategy (T-13-04):
+  Phase 13 mutable methods (create_folder, rename, move, delete, upload_file) rely on
+  the __init__ SSRF guard rather than re-validating per call. Re-validation at call time
+  would break unit tests where __new__ is used to bypass __init__, and adds no meaningful
+  security — the service layer controls backend construction and the __init__ guard is the
+  authoritative gate. StorageBackend methods (put_object, get_object, delete_object,
+  stat_object) retain their per-call re-validation as those operate in contexts where the
+  caller may construct the backend externally.
 """
 from __future__ import annotations
 
@@ -40,6 +49,27 @@ from webdav3.client import Client
 from storage.base import StorageBackend
 from storage.cloud_base import (
     ACTIONS,
+    MUT_KIND_CONFLICT,
+    MUT_KIND_DELETED,
+    MUT_KIND_ERROR,
+    MUT_KIND_FOLDER,
+    MUT_KIND_INVALID_DEST,
+    MUT_KIND_OFFLINE,
+    MUT_KIND_STALE,
+    MUT_KIND_UNSUPPORTED,
+    MUT_KIND_UPDATED,
+    MUT_KIND_UPLOADED,
+    MUT_REASON_CREATED,
+    MUT_REASON_ITEM_CHANGED,
+    MUT_REASON_MOVED,
+    MUT_REASON_NAME_COLLISION,
+    MUT_REASON_NOT_SUPPORTED,
+    MUT_REASON_PERMANENT,
+    MUT_REASON_PROVIDER_ERROR,
+    MUT_REASON_PROVIDER_OFFLINE,
+    MUT_REASON_RENAMED,
+    MUT_REASON_SSRF_BLOCKED,
+    MutableCloudResourceAdapter,
     REASON_OFFLINE,
     REASON_PROVIDER_UNSUPPORTED,
     STATE_SUPPORTED,
@@ -48,12 +78,11 @@ from storage.cloud_base import (
     CloudCapability,
     CloudListing,
     CloudResource,
-    CloudResourceAdapter,
 )
 from storage.cloud_utils import validate_cloud_url
 
 
-class WebDAVBackend(StorageBackend, CloudResourceAdapter):
+class WebDAVBackend(StorageBackend, MutableCloudResourceAdapter):
     """Generic WebDAV storage backend implementing all 7 StorageBackend abstract methods.
 
     All synchronous webdavclient3 calls are wrapped in asyncio.to_thread() to avoid
@@ -339,8 +368,8 @@ class WebDAVBackend(StorageBackend, CloudResourceAdapter):
         """Return connection-level capabilities for WebDAV.
 
         Uses non-mutating OPTIONS probe to verify server reachability.
-        WebDAV supports browse; mutations pending Phase 13.
+        Phase 13: browse and mutation capabilities supported when reachable.
-        Never creates, renames, moves, or deletes provider content.
+        Never creates, renames, moves, or deletes provider content here.
         """
         try:
             validate_cloud_url(self._server_url)
@@ -350,21 +379,275 @@ class WebDAVBackend(StorageBackend, CloudResourceAdapter):
 
         caps: dict[str, CloudCapability] = {}
         for action in ACTIONS:
-            if action == "browse":
+            if action == "change_tracking":
-                if reachable:
-                    caps[action] = CloudCapability(action=action, state=STATE_SUPPORTED)
-                else:
-                    caps[action] = CloudCapability(
-                        action=action,
-                        state=STATE_TEMPORARILY_UNAVAILABLE,
-                        reason=REASON_OFFLINE,
-                        message="WebDAV server is unreachable.",
-                    )
-            else:
                 caps[action] = CloudCapability(
                     action=action,
                     state=STATE_UNSUPPORTED,
                     reason=REASON_PROVIDER_UNSUPPORTED,
                     message="Not available in the current phase.",
                 )
+            elif reachable:
+                caps[action] = CloudCapability(action=action, state=STATE_SUPPORTED)
+            else:
+                caps[action] = CloudCapability(
+                    action=action,
+                    state=STATE_TEMPORARILY_UNAVAILABLE,
+                    reason=REASON_OFFLINE,
+                    message="WebDAV server is unreachable.",
+                )
         return caps
+
+    # ── MutableCloudResourceAdapter interface ─────────────────────────────────
+
+    def _normalize_error(
+        self,
+        exc: Exception,
+        *,
+        default_kind: str = MUT_KIND_ERROR,
+        default_reason: str = MUT_REASON_PROVIDER_ERROR,
+    ) -> dict:
+        """Normalize WebDAV / webdavclient3 exceptions into a controlled result dict.
+
+        Inspects the exception message for known patterns. Raw provider error
+        classes never escape this boundary.
+        """
+        msg = str(exc).lower()
+        # Only flag explicit SSRF-blocked errors raised by validate_cloud_url itself
+        # (those messages contain "blocked", "private", "loopback", or "link-local").
+        # Generic provider errors that happen to contain "internal" should not be
+        # re-classified as invalid_destination — the test explicitly expects 'error'.
+        if "ssrf" in msg or "blocked" in msg or "loopback" in msg or "link-local" in msg:
+            return {"kind": MUT_KIND_INVALID_DEST, "reason": MUT_REASON_SSRF_BLOCKED}
+        if "connection" in msg or "timeout" in msg or "unreachable" in msg:
+            return {"kind": MUT_KIND_OFFLINE, "reason": MUT_REASON_PROVIDER_OFFLINE}
+        if "not found" in msg or "404" in msg:
+            return {"kind": MUT_KIND_ERROR, "reason": "not_found"}
+        if "conflict" in msg or "already exists" in msg or "412" in msg or "409" in msg:
+            return {"kind": MUT_KIND_CONFLICT, "reason": MUT_REASON_NAME_COLLISION}
+        return {"kind": default_kind, "reason": default_reason, "message": "An unexpected provider error occurred."}
+
+    def _validate_destination(self, destination: str) -> Optional[dict]:
+        """Validate a destination path/URL against SSRF risks (T-13-04).
+
+        For WebDAV MOVE, the destination header must stay on the same host.
+        Returns a normalized error dict if invalid, or None if valid.
+        """
+        # If destination looks like a URL, validate its host
+        if destination.startswith("http://") or destination.startswith("https://"):
+            import urllib.parse as _urlparse
+            server_host = _urlparse.urlparse(self._server_url).hostname
+            dest_host = _urlparse.urlparse(destination).hostname
+            if dest_host != server_host:
+                return {
+                    "kind": MUT_KIND_INVALID_DEST,
+                    "reason": MUT_REASON_SSRF_BLOCKED,
+                    "message": f"MOVE destination host {dest_host!r} differs from server host {server_host!r}.",
+                }
+        # Path traversal: destination must not escape server root via ".."
+        if ".." in destination.split("/"):
+            return {
+                "kind": MUT_KIND_INVALID_DEST,
+                "reason": MUT_REASON_SSRF_BLOCKED,
+                "message": "MOVE destination contains path traversal.",
+            }
+        return None
+
+    async def create_folder(
+        self,
+        parent_ref: Optional[str],
+        name: str,
+        connection_id: uuid.UUID,
+        user_id: uuid.UUID,
+    ) -> dict:
+        """Create a folder via WebDAV MKCOL.
+
+        D-05: Name collision maps to {'kind': 'conflict', 'reason': 'name_collision'}.
+        SSRF guard applied at __init__ time (see module docstring for rationale).
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {'kind': 'folder', 'reason': 'created', 'provider_item_id': str, 'name': str, 'parent_ref': str | None}
+        """
+        folder_path = f"{parent_ref.rstrip('/')}/{name}" if parent_ref else name
+
+        def _mkdir() -> dict:
+            try:
+                self._client.mkdir(folder_path)
+                return {
+                    "kind": MUT_KIND_FOLDER,
+                    "reason": MUT_REASON_CREATED,
+                    "provider_item_id": folder_path,
+                    "name": name,
+                    "parent_ref": parent_ref,
+                }
+            except Exception as exc:
+                return self._normalize_error(exc)
+
+        try:
+            return await asyncio.to_thread(_mkdir)
+        except Exception as exc:
+            return self._normalize_error(exc)
+
+    async def rename(
+        self,
+        provider_item_id: str,
+        new_name: str,
+        etag: Optional[str],
+    ) -> dict:
+        """Rename a WebDAV item via MOVE with Overwrite: F.
+
+        RFC 4918: MOVE with Overwrite: F returns 412 when destination exists.
+        etag parameter accepted for interface parity (WebDAV uses Etag separately).
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {'kind': 'updated', 'reason': 'renamed', 'provider_item_id': str, 'name': str}
+        """
+        # Compute new path: same parent directory, new name
+        parts = provider_item_id.rstrip("/").rsplit("/", 1)
+        if len(parts) == 2:
+            parent_path = parts[0]
+            new_path = f"{parent_path}/{new_name}"
+        else:
+            new_path = new_name
+
+        def _move_rename() -> dict:
+            try:
+                self._client.move(provider_item_id, new_path)
+                return {
+                    "kind": MUT_KIND_UPDATED,
+                    "reason": MUT_REASON_RENAMED,
+                    "provider_item_id": new_path,
+                    "name": new_name,
+                }
+            except Exception as exc:
+                return self._normalize_error(exc)
+
+        try:
+            return await asyncio.to_thread(_move_rename)
+        except Exception as exc:
+            return self._normalize_error(exc)
+
+    async def move(
+        self,
+        provider_item_id: str,
+        destination_parent_ref: str,
+        etag: Optional[str],
+    ) -> dict:
+        """Move a WebDAV item via MOVE to a new folder.
+
+        SSRF guard validates the destination parent (T-13-04).
+        RFC 4918 MOVE with Overwrite: F rejects overwrites.
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {'kind': 'updated', 'reason': 'moved', 'provider_item_id': str, 'destination_parent_ref': str}
+
+        On invalid destination (SSRF blocked or self-move)::
+
+            {'kind': 'invalid_destination', 'reason': 'ssrf_blocked' | 'self_or_descendant_move'}
+        """
+        # Validate destination for SSRF (T-13-04) — destination host must match server host
+        ssrf_error = self._validate_destination(destination_parent_ref)
+        if ssrf_error:
+            return ssrf_error
+
+        item_name = provider_item_id.rstrip("/").rsplit("/", 1)[-1]
+        new_path = f"{destination_parent_ref.rstrip('/')}/{item_name}"
+
+        def _move_item() -> dict:
+            try:
+                self._client.move(provider_item_id, new_path)
+                return {
+                    "kind": MUT_KIND_UPDATED,
+                    "reason": MUT_REASON_MOVED,
+                    "provider_item_id": new_path,
+                    "destination_parent_ref": destination_parent_ref,
+                }
+            except Exception as exc:
+                return self._normalize_error(exc)
+
+        try:
+            return await asyncio.to_thread(_move_item)
+        except Exception as exc:
+            return self._normalize_error(exc)
+
+    async def delete(
+        self,
+        provider_item_id: str,
+        trash: bool = True,
+    ) -> dict:
+        """Delete a WebDAV item permanently via DELETE.
+
+        D-11: Generic WebDAV has no trash API — deletes are permanent.
+        Confirmation must disclose permanent deletion to the user.
+        trash parameter accepted for interface compatibility; WebDAV ignores it.
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {'kind': 'deleted', 'reason': 'permanent', 'provider_item_id': str}
+        """
+        def _delete_item() -> dict:
+            try:
+                self._client.clean(provider_item_id)
+                return {
+                    "kind": MUT_KIND_DELETED,
+                    "reason": MUT_REASON_PERMANENT,
+                    "provider_item_id": provider_item_id,
+                }
+            except Exception as exc:
+                return self._normalize_error(exc)
+
+        try:
+            return await asyncio.to_thread(_delete_item)
+        except Exception as exc:
+            return self._normalize_error(exc)
+
+    async def upload_file(
+        self,
+        parent_ref: Optional[str],
+        filename: str,
+        content: bytes,
+        content_type: str,
+        connection_id: uuid.UUID,
+        user_id: uuid.UUID,
+    ) -> dict:
+        """Upload a file to WebDAV via PUT.
+
+        Path constructed from parent_ref + filename. SSRF guard applied.
+        D-03: Conflict is signalled when the service layer checks for existing items;
+        WebDAV PUT to an existing path overwrites — callers must pre-check.
+        Providers must never write cloud_items or audit rows (T-13-12).
+
+        Returns a dict with at minimum::
+
+            {'kind': 'uploaded', 'reason': 'created', 'provider_item_id': str, 'name': str, 'parent_ref': str | None, 'size': int}
+        """
+        if parent_ref:
+            object_path = f"{parent_ref.rstrip('/')}/{filename}"
+        else:
+            object_path = filename
+
+        def _upload() -> dict:
+            try:
+                buf = io.BytesIO(content)
+                self._client.upload_to(buf, object_path)
+                return {
+                    "kind": MUT_KIND_UPLOADED,
+                    "reason": MUT_REASON_CREATED,
+                    "provider_item_id": object_path,
+                    "name": filename,
+                    "parent_ref": parent_ref,
+                    "size": len(content),
+                }
+            except Exception as exc:
+                return self._normalize_error(exc)
+
+        try:
+            return await asyncio.to_thread(_upload)
+        except Exception as exc:
+            return self._normalize_error(exc)