fix: revalidate durable Workdir removal on ordinary retry

crates/workspace-server/src/server.rs

@@ -11919,10 +11919,6 @@ fn cleanup_workdir_cleanliness_from_runtime(value: Option<&str>) -> CleanupWorkd
         .unwrap_or(CleanupWorkdirCleanliness::Unknown)
 }
 
-fn cleanup_workdir_cleanliness_is_clean(value: CleanupWorkdirCleanliness) -> bool {
-    matches!(value, CleanupWorkdirCleanliness::Clean)
-}
-
 #[derive(Debug, Deserialize)]
 struct LogQuery {
     limit: Option<usize>,
@@ -24531,6 +24527,73 @@ fn classify_workdir_provider_error(error: &RuntimeRegistryError) -> (&'static st
     }
 }
 
+// Only bounded classifications cross the public removal boundary. Provider
+// messages may contain OS paths or credentials and are never persisted here.
+fn workdir_cleanup_failure_category(
+    result: &crate::hosts::RuntimeWorkingDirectoryResult,
+) -> &'static str {
+    for diagnostic in &result.diagnostics {
+        let category = match diagnostic.code.as_str() {
+            "working_directory_unsupported" => "unsupported_target",
+            "working_directory_cleanup_mount_present" => "mount_present",
+            "working_directory_cleanup_mount_check_unavailable" => "mount_check_unavailable",
+            "working_directory_cleanup_permission_denied" => "permission_denied",
+            "working_directory_cleanup_resource_busy" => "resource_busy",
+            "working_directory_cleanup_storage_unavailable" => "storage_unavailable",
+            "working_directory_cleanup_ownership_unknown" => "ownership_unknown",
+            "working_directory_cleanup_changes_present" => "changes_present",
+            "working_directory_cleanup_changes_unknown" => "changes_unknown",
+            "working_directory_cleanup_target_invalid"
+            | "working_directory_cleanup_escape_rejected" => "ownership_unknown",
+            _ => continue,
+        };
+        return category;
+    }
+    "provider_cleanup_failed"
+}
+
+fn workdir_removal_failure_guidance(category: &str) -> &'static str {
+    match category {
+        "mount_present" => {
+            "A mount remains inside the Workdir. Confirm its owner and usage, have that owner safely release it, then retry deletion. Yoi does not unmount external resources."
+        }
+        "mount_check_unavailable" => {
+            "Runtime cannot inspect mount boundaries. Restore mount inspection before retrying deletion."
+        }
+        "permission_denied" => {
+            "Runtime lacks permission to remove the Workdir. Have its owner resolve access safely, then retry; do not force-clean or change shared resource permissions."
+        }
+        "resource_busy" => {
+            "A resource is busy. Have its owner stop or release it safely, then retry deletion."
+        }
+        "storage_unavailable" => {
+            "Runtime storage is unavailable. Restore storage access, then retry deletion."
+        }
+        "ownership_unknown" | "provider_identity_changed" | "authority_changed" => {
+            "Workdir identity or ownership could not be confirmed. Verify the target with its owner before retrying."
+        }
+        "changes_present" | "dirty_or_unknown" => {
+            "Current Workdir changes are protected. Preserve or resolve them before retrying deletion."
+        }
+        "changes_unknown" => {
+            "Runtime cannot verify remaining Workdir changes. Restore repository inspection or preserve remaining files before retrying."
+        }
+        "blocked_by_live_authority" => {
+            "Workdir has a current attachment, reservation or retention hold. Release the owning authority before retrying deletion."
+        }
+        "runtime_unavailable" | "provider_unavailable" => {
+            "Runtime is unavailable. Reconnect it, then retry deletion; the Workdir registry is retained."
+        }
+        "provider_observation_unknown" | "provider_cleanup_outcome_unknown" => {
+            "Runtime has not confirmed deletion. Retry to re-observe the target; the Workdir registry is retained."
+        }
+        "unsupported_target" => "This provider does not support Workdir deletion.",
+        _ => {
+            "Runtime cleanup failed. Inspect correlated Runtime diagnostics, resolve the cause, then retry deletion."
+        }
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum WorkdirRemovalOwnerObservation {
     Running { process_start_marker: u64 },
@@ -24739,11 +24802,34 @@ fn execute_reserved_workdir_removal_with_provider(
             true,
         );
     };
-    let corrupted = status.summary.status == WorkingDirectoryStatusKind::Corrupted;
-    if !corrupted
-        && (status.summary.cleanliness.as_deref() != Some("clean")
-            || status.summary.status != WorkingDirectoryStatusKind::Active)
+    // A provider summary is not permission to remove a different materialization.
+    // Unknown/corrupted metadata is delegated to Runtime's guarded cleanup, never
+    // repaired into a fabricated identity here.
+    let summary = &status.summary;
+    if summary.occupied_by.is_some() {
+        return api.config_store.complete_workdir_removal_retained(
+            &operation,
+            WorkdirRemovalDisposition::Retained,
+            "blocked_by_live_authority",
+        );
+    }
+    if summary.working_directory_id != operation.working_directory_id
+        || (summary.repository_id == "unknown"
+            && summary.status != WorkingDirectoryStatusKind::Corrupted)
+        || (summary.repository_id != "unknown"
+            && crate::workdir_removal::workdir_materialization_fingerprint(
+                &workdir_record_from_summary(api, &operation.runtime_id, summary),
+            ) != operation.materialization_fingerprint)
     {
+        return api.config_store.fail_workdir_removal_operation(
+            &operation,
+            "provider_identity_changed",
+            true,
+        );
+    }
+    // Past lifecycle failure is not a current dirtiness observation. Runtime
+    // must revalidate unknown/partial checkouts at the actual deletion boundary.
+    if summary.cleanliness.as_deref() == Some("dirty") {
         return api.config_store.complete_workdir_removal_retained(
             &operation,
             WorkdirRemovalDisposition::Retained,
@@ -24761,18 +24847,25 @@ fn execute_reserved_workdir_removal_with_provider(
                 .fail_workdir_removal_operation(&operation, category, retryable);
         }
     };
-    if deleted.state != InternalWorkerOperationState::Accepted
-        && !runtime_reports_workdir_not_found(&deleted)
-    {
-        let category = if deleted
-            .diagnostics
-            .iter()
-            .any(|diagnostic| diagnostic.code == "working_directory_unsupported")
-        {
-            "unsupported_target"
+    let deletion_confirmed = runtime_reports_workdir_not_found(&deleted)
+        || (deleted.state == InternalWorkerOperationState::Accepted
+            && deleted.working_directory.as_ref().is_some_and(|status| {
+                status.summary.working_directory_id == operation.working_directory_id
+                    && status.summary.status == WorkingDirectoryStatusKind::NotFound
+            }));
+    if !deletion_confirmed {
+        let category = if deleted.state == InternalWorkerOperationState::Accepted {
+            "provider_cleanup_outcome_unknown"
         } else {
-            "provider_cleanup_failed"
+            workdir_cleanup_failure_category(&deleted)
         };
+        tracing::warn!(
+            workdir_id = %operation.working_directory_id,
+            operation_id = %operation.operation_id,
+            runtime_id = %operation.runtime_id,
+            category,
+            "Runtime did not confirm Workdir removal; registry retained"
+        );
         return api.config_store.fail_workdir_removal_operation(
             &operation,
             category,
@@ -25058,9 +25151,7 @@ fn build_runtime_cleanup_plan(
             .unwrap_or_else(|| cleanup_workdir_cleanliness_from_registry(&record.cleanliness));
         let action = if cleanup_workdir_file_status_is_record_only(file_status) {
             CleanupTargetKind::WorkdirRecordDelete
-        } else if cleanup_workdir_file_status_is_corrupted(file_status)
-            || cleanup_workdir_cleanliness_is_clean(cleanliness)
-        {
+        } else if !matches!(cleanliness, CleanupWorkdirCleanliness::Dirty) {
             CleanupTargetKind::WorkdirCleanCleanup
         } else {
             CleanupTargetKind::WorkdirDirtyDiscard
@@ -25088,12 +25179,12 @@ fn build_runtime_cleanup_plan(
             } else if cleanup_workdir_file_status_is_record_only(file_status) {
                 "Not-found Workdir record can be deleted from the Backend registry".to_string()
             } else if cleanup_workdir_file_status_is_corrupted(file_status) {
-                "Corrupted Workdir can be deleted from Runtime storage and Backend registry"
+                "Runtime will verify remaining files and removal authority for this incomplete Workdir"
                     .to_string()
             } else if matches!(cleanliness, CleanupWorkdirCleanliness::Dirty) {
-                "Dirty Workdir requires explicit discard confirmation before cleanup".to_string()
+                "Dirty Workdir is protected; preserve or resolve current changes before retrying".to_string()
             } else if matches!(cleanliness, CleanupWorkdirCleanliness::Unknown) {
-                "Workdir clean state is unknown; explicit discard confirmation is required"
+                "Runtime will recheck current changes, identity and removal blockers before deleting"
                     .to_string()
             } else {
                 "Clean Workdir can be manually cleaned up".to_string()
@@ -25295,7 +25386,12 @@ async fn execute_runtime_cleanup_with_context(
                     target_id: candidate.target_id.clone(),
                     action: candidate.action.clone(),
                     status: status.to_string(),
-                    message: message.to_string(),
+                    message: removal.failure_category.as_deref().map_or_else(
+                        || message.to_string(),
+                        |category| {
+                            format!("{category}: {}", workdir_removal_failure_guidance(category))
+                        },
+                    ),
                 });
             }
             CleanupTargetKind::WorkerDelete => {
@@ -52974,7 +53070,7 @@ mod tests {
         };
         let source_fingerprint = repository_source_fingerprint(&repository_source);
         for (workdir_id, _, _) in &fixtures {
-            runtime
+            let created = runtime
                 .create_working_directory(WorkingDirectoryRequest {
                     repository: WorkingDirectoryRepository {
                         id: "repo-test".to_string(),
@@ -52997,6 +53093,13 @@ mod tests {
                 "present",
                 "clean",
             );
+            api.store
+                .upsert_workdir_registry(&workdir_record_from_summary(
+                    &api,
+                    provider_runtime_id,
+                    &created.summary,
+                ))
+                .unwrap();
         }
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
         let provider_base_url = format!("http://{}", listener.local_addr().unwrap());
@@ -54227,6 +54330,18 @@ mod tests {
         }
     }
 
+    fn confirmed_workdir_cleanup_result() -> crate::hosts::RuntimeWorkingDirectoryResult {
+        workdir_removal_result(
+            InternalWorkerOperationState::Rejected,
+            None,
+            vec![RuntimeDiagnostic {
+                code: "working_directory_not_found".into(),
+                severity: HostDiagnosticSeverity::Warning,
+                message: "authoritatively absent".into(),
+            }],
+        )
+    }
+
     fn reserve_removal_fixture(
         api: &WorkspaceApi,
         working_directory_id: &str,
@@ -54267,7 +54382,7 @@ mod tests {
                 Some(clean_summary),
                 Vec::new(),
             ),
-            workdir_removal_result(InternalWorkerOperationState::Accepted, None, Vec::new()),
+            confirmed_workdir_cleanup_result(),
         );
         let removed = execute_reserved_workdir_removal_with_provider(
             &api,
@@ -54333,7 +54448,7 @@ mod tests {
                     message: "timeout".to_string(),
                 }],
             ),
-            workdir_removal_result(InternalWorkerOperationState::Accepted, None, Vec::new()),
+            confirmed_workdir_cleanup_result(),
         );
         let unknown = execute_reserved_workdir_removal_with_provider(
             &api,
@@ -54355,7 +54470,7 @@ mod tests {
                 Some(dirty_summary),
                 Vec::new(),
             ),
-            workdir_removal_result(InternalWorkerOperationState::Accepted, None, Vec::new()),
+            confirmed_workdir_cleanup_result(),
         );
         let dirty = execute_reserved_workdir_removal_with_provider(
             &api,
@@ -54373,7 +54488,7 @@ mod tests {
                 Some(clean_retry_summary),
                 Vec::new(),
             ),
-            workdir_removal_result(InternalWorkerOperationState::Accepted, None, Vec::new()),
+            confirmed_workdir_cleanup_result(),
         );
         let removed_after_retry = execute_reserved_workdir_removal_with_provider(
             &api,
@@ -54399,7 +54514,7 @@ mod tests {
                 Some(corrupted_summary),
                 Vec::new(),
             ),
-            workdir_removal_result(InternalWorkerOperationState::Accepted, None, Vec::new()),
+            confirmed_workdir_cleanup_result(),
         );
         let corrupted = execute_reserved_workdir_removal_with_provider(
             &api,
@@ -54448,6 +54563,244 @@ mod tests {
         assert_eq!(unsupported_provider.cleanup_calls(), 1);
     }
 
+    #[test]
+    fn workdir_cleanup_failure_classification_is_bounded_and_guides_normal_retry() {
+        for (code, category) in [
+            ("working_directory_cleanup_mount_present", "mount_present"),
+            (
+                "working_directory_cleanup_mount_check_unavailable",
+                "mount_check_unavailable",
+            ),
+            (
+                "working_directory_cleanup_permission_denied",
+                "permission_denied",
+            ),
+            ("working_directory_cleanup_resource_busy", "resource_busy"),
+            (
+                "working_directory_cleanup_storage_unavailable",
+                "storage_unavailable",
+            ),
+            (
+                "working_directory_cleanup_ownership_unknown",
+                "ownership_unknown",
+            ),
+            (
+                "working_directory_cleanup_changes_present",
+                "changes_present",
+            ),
+            (
+                "working_directory_cleanup_changes_unknown",
+                "changes_unknown",
+            ),
+            (
+                "arbitrary provider text /private/path",
+                "provider_cleanup_failed",
+            ),
+        ] {
+            let result = workdir_removal_result(
+                InternalWorkerOperationState::Rejected,
+                None,
+                vec![RuntimeDiagnostic {
+                    code: code.into(),
+                    severity: HostDiagnosticSeverity::Error,
+                    message: "secret /private/path".into(),
+                }],
+            );
+            assert_eq!(workdir_cleanup_failure_category(&result), category);
+            let guidance = workdir_removal_failure_guidance(category);
+            assert!(!guidance.contains("/private/path"));
+            assert!(!guidance.contains("secret"));
+            assert!(guidance.len() <= 500);
+            assert!(guidance.contains("retry"));
+        }
+    }
+
+    #[tokio::test]
+    async fn workdir_removal_retry_rechecks_pending_materialization_changes_and_attachments() {
+        let workspace = tempfile::tempdir().unwrap();
+        init_clean_git_workspace(workspace.path());
+        let api = test_api(workspace.path()).await;
+        let (operation, mut summary) = reserve_removal_fixture(&api, "cleanup-retry");
+        let failed_provider = FakeWorkdirRemovalProvider::new(
+            workdir_removal_result(
+                InternalWorkerOperationState::Accepted,
+                Some(summary.clone()),
+                Vec::new(),
+            ),
+            workdir_removal_result(
+                InternalWorkerOperationState::Rejected,
+                None,
+                vec![RuntimeDiagnostic {
+                    code: "working_directory_cleanup_mount_present".into(),
+                    severity: HostDiagnosticSeverity::Error,
+                    message: "/private/mount credentials must not escape".into(),
+                }],
+            ),
+        );
+        let failed = execute_reserved_workdir_removal_with_provider(
+            &api,
+            operation,
+            false,
+            &failed_provider,
+        )
+        .unwrap();
+        assert!(failed.retryable);
+        assert_eq!(failed.failure_category.as_deref(), Some("mount_present"));
+        assert!(
+            api.store
+                .get_workdir_registry(TEST_WORKSPACE_ID, "cleanup-retry")
+                .unwrap()
+                .is_some()
+        );
+        assert!(!serde_json::to_string(&failed).unwrap().contains("/private"));
+
+        summary.status = WorkingDirectoryStatusKind::CleanupPending;
+        summary.cleanliness = Some("dirty".into());
+        let dirty_provider = FakeWorkdirRemovalProvider::new(
+            workdir_removal_result(
+                InternalWorkerOperationState::Accepted,
+                Some(summary.clone()),
+                Vec::new(),
+            ),
+            confirmed_workdir_cleanup_result(),
+        );
+        let dirty =
+            execute_reserved_workdir_removal_with_provider(&api, failed, false, &dirty_provider)
+                .unwrap();
+        assert_eq!(dirty.disposition, Some(WorkdirRemovalDisposition::Retained));
+        assert!(dirty.retryable);
+        assert_eq!(dirty_provider.cleanup_calls(), 0);
+
+        let worker = seed_cleanup_worker(&api, 731, "normal");
+        seed_cleanup_link(&api, &worker, "cleanup-retry");
+        let occupied_provider = FakeWorkdirRemovalProvider::new(
+            workdir_removal_result(
+                InternalWorkerOperationState::Accepted,
+                Some(summary.clone()),
+                Vec::new(),
+            ),
+            confirmed_workdir_cleanup_result(),
+        );
+        let occupied =
+            execute_reserved_workdir_removal_with_provider(&api, dirty, false, &occupied_provider)
+                .unwrap();
+        assert_eq!(
+            occupied.failure_category.as_deref(),
+            Some("blocked_by_live_authority")
+        );
+        assert!(occupied.retryable);
+        assert_eq!(occupied_provider.cleanup_calls(), 0);
+        api.store
+            .detach_worker_workdir(
+                TEST_WORKSPACE_ID,
+                &RuntimeWorkerRef::new("runtime-test", worker),
+                Some("cleanup-retry"),
+                "resolve retry attachment",
+            )
+            .unwrap();
+
+        // Unknown because checkout was partially removed is delegated to Runtime
+        // for current change/identity protection, not denied by the old state.
+        summary.cleanliness = Some("unknown".into());
+        let retry_provider = FakeWorkdirRemovalProvider::new(
+            workdir_removal_result(
+                InternalWorkerOperationState::Accepted,
+                Some(summary),
+                Vec::new(),
+            ),
+            confirmed_workdir_cleanup_result(),
+        );
+        let removed = execute_reserved_workdir_removal_with_provider(
+            &api,
+            occupied.clone(),
+            false,
+            &retry_provider,
+        )
+        .unwrap();
+        assert_eq!(removed.operation_id, occupied.operation_id);
+        assert_eq!(removed.attempt_count, 4);
+        assert_eq!(
+            removed.disposition,
+            Some(WorkdirRemovalDisposition::Removed)
+        );
+        assert_eq!(retry_provider.cleanup_calls(), 1);
+        assert!(
+            api.store
+                .get_workdir_registry(TEST_WORKSPACE_ID, "cleanup-retry")
+                .unwrap()
+                .is_none()
+        );
+        assert_eq!(
+            execute_reserved_workdir_removal_with_provider(
+                &api,
+                removed.clone(),
+                false,
+                &retry_provider
+            )
+            .unwrap(),
+            removed
+        );
+    }
+
+    #[tokio::test]
+    async fn workdir_removal_keeps_registry_for_changed_identity_and_unknown_cleanup_outcome() {
+        let workspace = tempfile::tempdir().unwrap();
+        init_clean_git_workspace(workspace.path());
+        let api = test_api(workspace.path()).await;
+        for (id, changed_identity) in [("identity-mismatch", true), ("unknown-cleanup", false)] {
+            let (operation, mut summary) = reserve_removal_fixture(&api, id);
+            if changed_identity {
+                summary.repository_id = "another-repository".into();
+            }
+            let provider = FakeWorkdirRemovalProvider::new(
+                workdir_removal_result(
+                    InternalWorkerOperationState::Accepted,
+                    Some(summary),
+                    Vec::new(),
+                ),
+                workdir_removal_result(InternalWorkerOperationState::Accepted, None, Vec::new()),
+            );
+            let result =
+                execute_reserved_workdir_removal_with_provider(&api, operation, false, &provider)
+                    .unwrap();
+            assert!(result.retryable);
+            assert_eq!(
+                result.failure_category.as_deref(),
+                Some(if changed_identity {
+                    "provider_identity_changed"
+                } else {
+                    "provider_cleanup_outcome_unknown"
+                })
+            );
+            assert_eq!(provider.cleanup_calls(), usize::from(!changed_identity));
+            assert!(
+                api.store
+                    .get_workdir_registry(TEST_WORKSPACE_ID, id)
+                    .unwrap()
+                    .is_some()
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn cleanup_plan_offers_normal_revalidation_for_prior_failure_and_unknown() {
+        let workspace = tempfile::tempdir().unwrap();
+        init_clean_git_workspace(workspace.path());
+        let api = test_api(workspace.path()).await;
+        for (id, state) in [
+            ("pending-retry", "pending"),
+            ("unknown-retry", "unknown"),
+            ("corrupt-retry", "corrupted"),
+        ] {
+            seed_cleanup_workdir(&api, id, state, "unknown");
+        }
+        let plan = build_runtime_cleanup_plan(&api, "runtime-test").unwrap();
+        for candidate in plan.workdirs {
+            assert_eq!(candidate.action, CleanupTargetKind::WorkdirCleanCleanup);
+            assert!(candidate.blocking_reason.is_none());
+        }
+    }
+
     #[tokio::test]
     async fn concurrent_orphan_recovery_runs_delayed_provider_cleanup_once() {
         let workspace = tempfile::tempdir().unwrap();
@@ -54473,7 +54826,7 @@ mod tests {
                     Some(clean_summary),
                     Vec::new(),
                 ),
-                workdir_removal_result(InternalWorkerOperationState::Accepted, None, Vec::new()),
+                confirmed_workdir_cleanup_result(),
             )
             .with_observation_delay(std::time::Duration::from_millis(100)),
         );

crates/workspace-server/src/workdir_removal.rs

@@ -438,7 +438,9 @@ impl SqliteWorkspaceStore {
                 "retained completion cannot use removed disposition".to_string(),
             ));
         }
-        self.finish_workdir_removal_operation(operation, disposition, false, Some(category), false)
+        // Retention describes this attempt's current blockers, not a permanent
+        // ban. A subsequent normal request rechecks all live authority.
+        self.finish_workdir_removal_operation(operation, disposition, true, Some(category), false)
     }
 
     pub fn fail_workdir_removal_operation(

docs/development/workdir-removal.md

@@ -0,0 +1,53 @@
+# Workdir removal and ordinary retry
+
+Use the Runtime **Workdirs** page or the normal typed Workdir removal tool to
+remove an unused repository Workdir. Deletion is authorized by the Backend's
+existing durable removal operation; it is not a registry-only hide operation.
+
+A failed attempt does not require a new recovery state or a manual edit of
+`materialization.json`. Retry the same removal request after resolving the
+reported cause. Each attempt rechecks current attachments, attachment
+reservations, retention holds, materialization identity and current file changes.
+A previous clean observation is not permission to discard new changes. A dirty
+Workdir must be preserved or its changes resolved by its owner before deletion.
+
+## Interpreting a retained Workdir
+
+- **Mount remains:** Git status being clean does not mean mounted resources are
+  released. Confirm the mount's owner, users and data destination. Have that
+  owner safely release the resource, then retry normal deletion. Yoi does not
+  recursively delete mounted data or implicitly unmount external/unknown
+  resources. Do not use force-clean, recursive chmod or raw recursive deletion
+  to override this boundary.
+- **Permission denied / resource busy / storage unavailable:** have the owning
+  operator resolve access, release the resource or restore storage. Retry the
+  same normal action; registry retention is intentional until cleanup is proven.
+- **Changes present / changes cannot be verified:** preserve current files and
+  restore repository inspection where possible. Do not treat a previous cleanup
+  failure or missing checkout metadata as proof that all remaining files are
+  expendable.
+- **Identity or ownership unavailable:** verify the Workdir target with its
+  owner. Do not rewrite registry or materialization identity to bypass checks.
+- **Runtime unavailable / outcome unknown:** reconnect the Runtime and retry.
+  Absence of a provider response is not authoritative not-found evidence.
+- **Attachment, reservation or hold:** release the current owning authority
+  through its normal API before retrying. A stale failure does not freeze these
+  conditions; newly attached Workers remain protected.
+
+The Web page refreshes authoritative inventory after each attempt. Successful
+removal commits registry deletion only after Runtime confirms physical deletion
+or authoritative not-found. An unknown result leaves the registry in place so
+retry can reconcile the provider's current state.
+
+## Diagnostics
+
+Public responses expose a bounded cause category and safe guidance, not arbitrary
+provider errors or host paths. For an OS failure, correlate Backend removal logs
+(`operation_id`, `workdir_id`, Runtime) with Runtime cleanup logs for that Workdir
+and cleanup action. Runtime logs retain the concrete OS error/errno for the
+owning operator. These private diagnostics must not be copied unredacted into
+public responses or Tickets.
+
+This procedure is not authorization to manipulate a live mount or another
+Worker's storage. If ownership or required operator authority is unclear, stop
+and request that specific owner/operator's decision.