1 files changed, 184 insertions, 0 deletions

diff --git a/subcmds/wipe.py b/subcmds/wipe.py
new file mode 100644
index 00000000..51938649
--- /dev/null
+++ b/subcmds/wipe.py
@@ -0,0 +1,184 @@
+# Copyright (C) 2025 The Android Open Source Project
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import sys
+from typing import List
+
+from command import Command
+from error import GitError
+from error import RepoExitError
+import platform_utils
+from project import DeleteWorktreeError
+
+
+class Error(RepoExitError):
+    """Exit error when wipe command fails."""
+
+
+class Wipe(Command):
+    """Delete projects from the worktree and .repo"""
+
+    COMMON = True
+    helpSummary = "Wipe projects from the worktree"
+    helpUsage = """
+%prog <project>...
+"""
+    helpDescription = """
+The '%prog' command removes the specified projects from the worktree
+(the checked out source code) and deletes the project's git data from `.repo`.
+
+This is a destructive operation and cannot be undone.
+
+Projects can be specified either by name, or by a relative or absolute path
+to the project's local directory.
+
+Examples:
+
+  # Wipe the project "platform/build" by name:
+  $ repo wipe platform/build
+
+  # Wipe the project at the path "build/make":
+  $ repo wipe build/make
+"""
+
+    def _Options(self, p):
+        # TODO(crbug.com/gerrit/393383056): Add --broken option to scan and
+        # wipe broken projects.
+        p.add_option(
+            "-f",
+            "--force",
+            action="store_true",
+            help="force wipe shared projects and uncommitted changes",
+        )
+        p.add_option(
+            "--force-uncommitted",
+            action="store_true",
+            help="force wipe even if there are uncommitted changes",
+        )
+        p.add_option(
+            "--force-shared",
+            action="store_true",
+            help="force wipe even if the project shares an object directory",
+        )
+
+    def ValidateOptions(self, opt, args: List[str]):
+        if not args:
+            self.Usage()
+
+    def Execute(self, opt, args: List[str]):
+        # Get all projects to handle shared object directories.
+        all_projects = self.GetProjects(None, all_manifests=True, groups="all")
+        projects_to_wipe = self.GetProjects(args, all_manifests=True)
+        relpaths_to_wipe = {p.relpath for p in projects_to_wipe}
+
+        # Build a map from objdir to the relpaths of projects that use it.
+        objdir_map = {}
+        for p in all_projects:
+            objdir_map.setdefault(p.objdir, set()).add(p.relpath)
+
+        uncommitted_projects = []
+        shared_objdirs = {}
+        objdirs_to_delete = set()
+
+        for project in projects_to_wipe:
+            if project == self.manifest.manifestProject:
+                raise Error(
+                    f"error: cannot wipe the manifest project: {project.name}"
+                )
+
+            try:
+                if project.HasChanges():
+                    uncommitted_projects.append(project.name)
+            except GitError:
+                uncommitted_projects.append(f"{project.name} (corrupted)")
+
+            users = objdir_map.get(project.objdir, {project.relpath})
+            is_shared = not users.issubset(relpaths_to_wipe)
+            if is_shared:
+                shared_objdirs.setdefault(project.objdir, set()).update(users)
+            else:
+                objdirs_to_delete.add(project.objdir)
+
+        block_uncommitted = uncommitted_projects and not (
+            opt.force or opt.force_uncommitted
+        )
+        block_shared = shared_objdirs and not (opt.force or opt.force_shared)
+
+        if block_uncommitted or block_shared:
+            error_messages = []
+            if block_uncommitted:
+                error_messages.append(
+                    "The following projects have uncommitted changes or are "
+                    "corrupted:\n"
+                    + "\n".join(f" - {p}" for p in sorted(uncommitted_projects))
+                )
+            if block_shared:
+                shared_dir_messages = []
+                for objdir, users in sorted(shared_objdirs.items()):
+                    other_users = users - relpaths_to_wipe
+                    projects_to_wipe_in_dir = users & relpaths_to_wipe
+                    message = f"""Object directory {objdir} is shared by:
+  Projects to be wiped: {', '.join(sorted(projects_to_wipe_in_dir))}
+  Projects not to be wiped: {', '.join(sorted(other_users))}"""
+                    shared_dir_messages.append(message)
+                error_messages.append(
+                    "The following projects have shared object directories:\n"
+                    + "\n".join(sorted(shared_dir_messages))
+                )
+
+            if block_uncommitted and block_shared:
+                error_messages.append(
+                    "Use --force to wipe anyway, or --force-uncommitted and "
+                    "--force-shared to specify."
+                )
+            elif block_uncommitted:
+                error_messages.append("Use --force-uncommitted to wipe anyway.")
+            else:
+                error_messages.append("Use --force-shared to wipe anyway.")
+
+            raise Error("\n\n".join(error_messages))
+
+        # If we are here, either there were no issues, or --force was used.
+        # Proceed with wiping.
+        successful_wipes = set()
+
+        for project in projects_to_wipe:
+            try:
+                # Force the delete here since we've already performed our
+                # own safety checks above.
+                project.DeleteWorktree(force=True, verbose=opt.verbose)
+                successful_wipes.add(project.relpath)
+            except DeleteWorktreeError as e:
+                print(
+                    f"error: failed to wipe {project.name}: {e}",
+                    file=sys.stderr,
+                )
+
+        # Clean up object directories only if all projects using them were
+        # successfully wiped.
+        for objdir in objdirs_to_delete:
+            users = objdir_map.get(objdir, set())
+            # Check if every project that uses this objdir has been
+            # successfully processed. If a project failed to be wiped, don't
+            # delete the object directory, or we'll corrupt the remaining
+            # project.
+            if users.issubset(successful_wipes):
+                if os.path.exists(objdir):
+                    if opt.verbose:
+                        print(
+                            f"Deleting objects directory: {objdir}",
+                            file=sys.stderr,
+                        )
+                    platform_utils.rmtree(objdir)