From 9a2fbf6dbff0f16ce82305aec2fa01af404e57c9 Mon Sep 17 00:00:00 2001
From: OiPunk <codingpunk@gmail.com>
Date: Thu, 5 Mar 2026 15:45:03 +0800
Subject: [PATCH] fix: handle read-only files in shutil.rmtree on Windows
 during deploy cleanup

Replace all bare shutil.rmtree() calls in cli_deploy.py with a
_robust_rmtree() helper that clears the read-only bit before
retrying deletion on Windows. This prevents PermissionError
([WinError 5]) when cleaning up .git/objects after deployment.

The fix is compatible with Python 3.10+ (uses onerror for <3.12
and onexc for >=3.12).
---
 src/google/adk/cli/cli_deploy.py             | 30 +++++++++++---
 tests/unittests/cli/utils/test_cli_deploy.py | 42 ++++++++++++++++++++
 2 files changed, 66 insertions(+), 6 deletions(-)

diff --git a/src/google/adk/cli/cli_deploy.py b/src/google/adk/cli/cli_deploy.py
index 465ca6c3e1..3aeca6cb3f 100644
--- a/src/google/adk/cli/cli_deploy.py
+++ b/src/google/adk/cli/cli_deploy.py
@@ -18,6 +18,7 @@
 import json
 import os
 import shutil
+import stat
 import subprocess
 import sys
 import traceback
@@ -36,6 +37,23 @@
 )
 
 
+def _on_rm_error(func, path, exc_info):
+  """Error handler for shutil.rmtree to handle read-only files on Windows."""
+  os.chmod(path, stat.S_IWRITE)
+  func(path)
+
+
+def _robust_rmtree(path: str) -> None:
+  """Remove a directory tree, handling read-only files on Windows."""
+  if _IS_WINDOWS:
+    if sys.version_info >= (3, 12):
+      shutil.rmtree(path, onexc=lambda fn, p, exc: _on_rm_error(fn, p, None))
+    else:
+      shutil.rmtree(path, onerror=_on_rm_error)
+  else:
+    shutil.rmtree(path)
+
+
 def _ensure_agent_engine_dependency(requirements_txt_path: str) -> None:
   """Ensures staged requirements include Agent Engine dependencies."""
   if not os.path.exists(requirements_txt_path):
@@ -692,7 +710,7 @@ def to_cloud_run(
   # remove temp_folder if exists
   if os.path.exists(temp_folder):
     click.echo('Removing existing files')
-    shutil.rmtree(temp_folder)
+    _robust_rmtree(temp_folder)
 
   try:
     # copy agent source code
@@ -799,7 +817,7 @@ def to_cloud_run(
     subprocess.run(gcloud_cmd, check=True)
   finally:
     click.echo(f'Cleaning up the temp folder: {temp_folder}')
-    shutil.rmtree(temp_folder)
+    _robust_rmtree(temp_folder)
 
 
 def to_agent_engine(
@@ -924,7 +942,7 @@ def to_agent_engine(
   # remove agent_src_path if it exists
   if os.path.exists(agent_src_path):
     click.echo('Removing existing files')
-    shutil.rmtree(agent_src_path)
+    _robust_rmtree(agent_src_path)
 
   try:
     click.echo(f'Staging all files in: {agent_src_path}')
@@ -1136,7 +1154,7 @@ def to_agent_engine(
       click.secho(f'✅ Updated agent engine: {agent_engine_id}', fg='green')
   finally:
     click.echo(f'Cleaning up the temp folder: {temp_folder}')
-    shutil.rmtree(agent_src_path)
+    _robust_rmtree(agent_src_path)
     if did_change_cwd:
       os.chdir(original_cwd)
 
@@ -1212,7 +1230,7 @@ def to_gke(
   # remove temp_folder if exists
   if os.path.exists(temp_folder):
     click.echo('  - Removing existing temporary directory...')
-    shutil.rmtree(temp_folder)
+    _robust_rmtree(temp_folder)
 
   try:
     # copy agent source code
@@ -1376,7 +1394,7 @@ def to_gke(
   finally:
     click.secho('\nSTEP 5: Cleaning up...', bold=True)
     click.echo(f'  - Removing temporary directory: {temp_folder}')
-    shutil.rmtree(temp_folder)
+    _robust_rmtree(temp_folder)
   click.secho(
       '\n🎉 Deployment to GKE finished successfully!', fg='cyan', bold=True
   )
diff --git a/tests/unittests/cli/utils/test_cli_deploy.py b/tests/unittests/cli/utils/test_cli_deploy.py
index e1829d2f98..16ac13f24c 100644
--- a/tests/unittests/cli/utils/test_cli_deploy.py
+++ b/tests/unittests/cli/utils/test_cli_deploy.py
@@ -654,3 +654,45 @@ def test_restores_sys_path(self, tmp_path: Path) -> None:
     )
 
     assert sys.path == original_path
+
+
+# _robust_rmtree / _on_rm_error tests
+
+
+class TestRobustRmtree:
+  """Tests for the _robust_rmtree helper."""
+
+  def test_removes_directory_tree(self, tmp_path: Path) -> None:
+    """It should remove a normal directory tree."""
+    d = tmp_path / "subdir"
+    d.mkdir()
+    (d / "file.txt").write_text("hello")
+    cli_deploy._robust_rmtree(str(d))
+    assert not d.exists()
+
+  def test_removes_readonly_files(self, tmp_path: Path) -> None:
+    """It should remove a tree containing read-only files."""
+    import os
+    import stat
+
+    d = tmp_path / "ro_dir"
+    d.mkdir()
+    ro_file = d / "readonly.txt"
+    ro_file.write_text("locked")
+    ro_file.chmod(stat.S_IREAD)
+    cli_deploy._robust_rmtree(str(d))
+    assert not d.exists()
+
+  def test_on_rm_error_clears_readonly_and_retries(
+      self, tmp_path: Path
+  ) -> None:
+    """_on_rm_error should chmod the file and call the removal function."""
+    import os
+    import stat
+
+    ro_file = tmp_path / "locked.txt"
+    ro_file.write_text("data")
+    ro_file.chmod(stat.S_IREAD)
+
+    cli_deploy._on_rm_error(os.remove, str(ro_file), None)
+    assert not ro_file.exists()