Fix worktree cleanup for branches without upstreams

s18_worktree_isolation/README.en.md

@@ -89,11 +89,11 @@ After task completion, two choices:
 
 ```python
 def remove_worktree(name: str, discard_changes: bool = False) -> str:
-    # Safety check: refuse by default if changes exist
+    # Safety check: refuse by default if files or unmerged commits exist
     if not discard_changes:
         files, commits = _count_worktree_changes(path)
         if files > 0 or commits > 0:
-            return "Has uncommitted changes. Use discard_changes=true to force, or keep_worktree"
+            return "Has uncommitted files or unmerged commits. Use discard_changes=true to force, or keep_worktree"
     ok, _ = run_git(["worktree", "remove", str(path), "--force"])
     if not ok:
         return "Remove failed"
@@ -105,7 +105,7 @@ def keep_worktree(name: str) -> str:
     return f"Worktree '{name}' kept for review (branch: wt/{name})"
 ```
 
-Keep = preserve branch for manual review and merge. Remove = refuse by default if uncommitted changes; requires `discard_changes=true` to confirm. Does NOT auto-complete task — task completion is triggered explicitly by the teammate's `complete_task`.
+Keep = preserve branch for manual review and merge. Remove = refuse by default if uncommitted files or commits not in the main worktree's `HEAD` exist; requires `discard_changes=true` to confirm. Does NOT auto-complete task — task completion is triggered explicitly by the teammate's `complete_task`.
 
 ### Event Log: Auditable
 
@@ -205,4 +205,4 @@ Teaching version uses the task's `worktree` field for binding, a teaching simpli
 
 </details>
 
-<!-- translation-sync: zh@v1, en@v1, ja@v0 -->
+<!-- translation-sync: zh@v2, en@v2, ja@v0 -->

s18_worktree_isolation/README.ja.md

@@ -89,11 +89,11 @@ def _run_bash(command):
 
 ```python
 def remove_worktree(name: str, discard_changes: bool = False) -> str:
-    # 安全チェック：変更がある場合デフォルトで拒否
+    # 安全チェック：未コミットファイルまたは未マージコミットがある場合デフォルトで拒否
     if not discard_changes:
         files, commits = _count_worktree_changes(path)
         if files > 0 or commits > 0:
-            return "未コミットの変更あり。discard_changes=true で強制削除、または keep_worktree で保持"
+            return "未コミットファイルまたは未マージコミットあり。discard_changes=true で強制削除、または keep_worktree で保持"
     ok, _ = run_git(["worktree", "remove", str(path), "--force"])
     if not ok:
         return "削除失敗"
@@ -105,7 +105,7 @@ def keep_worktree(name: str) -> str:
     return f"Worktree '{name}' kept for review (branch: wt/{name})"
 ```
 
-Keep = ブランチを保持し、手動 review 後にマージ。Remove = 未コミット変更がある場合デフォルトで拒否、`discard_changes=true` で確認が必要。タスクの自動 complete はしない——タスク完了はチームメイトの `complete_task` で明示的にトリガー。
+Keep = ブランチを保持し、手動 review 後にマージ。Remove = 未コミットファイルまたはメイン worktree の `HEAD` に含まれないコミットがある場合デフォルトで拒否し、`discard_changes=true` で確認が必要。タスクの自動 complete はしない——タスク完了はチームメイトの `complete_task` で明示的にトリガー。
 
 ### イベントログ：監査可能
 
@@ -205,4 +205,4 @@ CC にはタスク-worktree 紐付けがない。Worktree 状態は `PersistedWo
 
 </details>
 
-<!-- translation-sync: zh@v1, en@v1, ja@v1 -->
+<!-- translation-sync: zh@v2, en@v2, ja@v2 -->

s18_worktree_isolation/README.md

@@ -89,11 +89,11 @@ def _run_bash(command):
 
 ```python
 def remove_worktree(name: str, discard_changes: bool = False) -> str:
-    # 安全检查：有改动时默认拒绝
+    # 安全检查：有未提交文件或未合并提交时默认拒绝
     if not discard_changes:
         files, commits = _count_worktree_changes(path)
         if files > 0 or commits > 0:
-            return "有未提交改动，使用 discard_changes=true 强制删除，或 keep_worktree 保留"
+            return "有未提交文件或未合并提交，使用 discard_changes=true 强制删除，或 keep_worktree 保留"
     ok, _ = run_git(["worktree", "remove", str(path), "--force"])
     if not ok:
         return "删除失败"
@@ -105,7 +105,7 @@ def keep_worktree(name: str) -> str:
     return f"Worktree '{name}' kept for review (branch: wt/{name})"
 ```
 
-Keep = 留着分支，等人工 review 后合并到主分支。Remove = 有改动时默认拒绝，需要 `discard_changes=true` 确认。不自动 complete task——任务完成由队友的 `complete_task` 显式触发。
+Keep = 留着分支，等人工 review 后合并到主分支。Remove = 有未提交文件或尚未进入主工作区 `HEAD` 的提交时默认拒绝，需要 `discard_changes=true` 确认。不自动 complete task——任务完成由队友的 `complete_task` 显式触发。
 
 ### 事件流：可审计
 
@@ -205,4 +205,4 @@ CC 没有 task-worktree 绑定。Worktree 状态通过 `PersistedWorktreeSession
 
 </details>
 
-<!-- translation-sync: zh@v1, en@v0, ja@v0 -->
+<!-- translation-sync: zh@v2, en@v0, ja@v0 -->

s18_worktree_isolation/code.py

@@ -213,21 +213,27 @@ def bind_task_to_worktree(task_id: str, worktree_name: str):
 
 
 def _count_worktree_changes(path: Path) -> tuple[int, int]:
-    """Count uncommitted files and commits in a worktree."""
+    """Count uncommitted files and commits not in the main worktree."""
     try:
         r1 = subprocess.run(["git", "status", "--porcelain"],
-                            cwd=path, capture_output=True, text=True, timeout=10)
+                            cwd=path, capture_output=True, text=True, timeout=10,
+                            check=True)
         files = len([l for l in r1.stdout.strip().splitlines() if l.strip()])
-        r2 = subprocess.run(["git", "log", "@{push}..HEAD", "--oneline"],
-                            cwd=path, capture_output=True, text=True, timeout=10)
-        commits = len([l for l in r2.stdout.strip().splitlines() if l.strip()])
+        main_head = subprocess.run(["git", "rev-parse", "HEAD"],
+                                   cwd=WORKDIR, capture_output=True, text=True,
+                                   timeout=10, check=True).stdout.strip()
+        r2 = subprocess.run(["git", "rev-list", "--count",
+                             f"{main_head}..HEAD"],
+                            cwd=path, capture_output=True, text=True, timeout=10,
+                            check=True)
+        commits = int(r2.stdout.strip())
         return files, commits
-    except Exception:
+    except (OSError, subprocess.SubprocessError, ValueError):
         return -1, -1
 
 
 def remove_worktree(name: str, discard_changes: bool = False) -> str:
-    """Remove worktree. Refuses if uncommitted changes unless discard_changes."""
+    """Remove worktree. Refuses if files or unmerged commits would be lost."""
     err = validate_worktree_name(name)
     if err:
         return err
@@ -241,7 +247,7 @@ def remove_worktree(name: str, discard_changes: bool = False) -> str:
                     "Use discard_changes=true to force removal.")
         if files > 0 or commits > 0:
             return (f"Worktree '{name}' has {files} uncommitted file(s) "
-                    f"and {commits} unpushed commit(s). "
+                    f"and {commits} unmerged commit(s). "
                     "Use discard_changes=true to force removal, "
                     "or keep_worktree to preserve for review.")
     ok1, _ = run_git(["worktree", "remove", str(path), "--force"])
@@ -893,7 +899,7 @@ def run_check_inbox() -> str:
                                      "task_id": {"type": "string"}},
                       "required": ["name"]}},
     {"name": "remove_worktree",
-     "description": "Remove a worktree. Refuses if uncommitted changes unless discard_changes=true.",
+     "description": "Remove a worktree. Refuses if files or unmerged commits exist unless discard_changes=true.",
      "input_schema": {"type": "object",
                       "properties": {"name": {"type": "string"},
                                      "discard_changes": {"type": "boolean"}},

s19_mcp_plugin/code.py

@@ -202,13 +202,19 @@ def bind_task_to_worktree(task_id: str, worktree_name: str):
 def _count_worktree_changes(path: Path) -> tuple[int, int]:
     try:
         r1 = subprocess.run(["git", "status", "--porcelain"],
-                            cwd=path, capture_output=True, text=True, timeout=10)
+                            cwd=path, capture_output=True, text=True, timeout=10,
+                            check=True)
         files = len([l for l in r1.stdout.strip().splitlines() if l.strip()])
-        r2 = subprocess.run(["git", "log", "@{push}..HEAD", "--oneline"],
-                            cwd=path, capture_output=True, text=True, timeout=10)
-        commits = len([l for l in r2.stdout.strip().splitlines() if l.strip()])
+        main_head = subprocess.run(["git", "rev-parse", "HEAD"],
+                                   cwd=WORKDIR, capture_output=True, text=True,
+                                   timeout=10, check=True).stdout.strip()
+        r2 = subprocess.run(["git", "rev-list", "--count",
+                             f"{main_head}..HEAD"],
+                            cwd=path, capture_output=True, text=True, timeout=10,
+                            check=True)
+        commits = int(r2.stdout.strip())
         return files, commits
-    except Exception:
+    except (OSError, subprocess.SubprocessError, ValueError):
         return -1, -1
 
 

s20_comprehensive/code.py

@@ -241,13 +241,19 @@ def bind_task_to_worktree(task_id: str, worktree_name: str):
 def _count_worktree_changes(path: Path) -> tuple[int, int]:
     try:
         r1 = subprocess.run(["git", "status", "--porcelain"],
-                            cwd=path, capture_output=True, text=True, timeout=10)
+                            cwd=path, capture_output=True, text=True, timeout=10,
+                            check=True)
         files = len([l for l in r1.stdout.strip().splitlines() if l.strip()])
-        r2 = subprocess.run(["git", "log", "@{push}..HEAD", "--oneline"],
-                            cwd=path, capture_output=True, text=True, timeout=10)
-        commits = len([l for l in r2.stdout.strip().splitlines() if l.strip()])
+        main_head = subprocess.run(["git", "rev-parse", "HEAD"],
+                                   cwd=WORKDIR, capture_output=True, text=True,
+                                   timeout=10, check=True).stdout.strip()
+        r2 = subprocess.run(["git", "rev-list", "--count",
+                             f"{main_head}..HEAD"],
+                            cwd=path, capture_output=True, text=True, timeout=10,
+                            check=True)
+        commits = int(r2.stdout.strip())
         return files, commits
-    except Exception:
+    except (OSError, subprocess.SubprocessError, ValueError):
         return -1, -1
 
 

tests/test_worktree_cleanup.py

@@ -0,0 +1,198 @@
+import importlib.util
+import os
+import subprocess
+import sys
+import tempfile
+import types
+import unittest
+from pathlib import Path
+from unittest import mock
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+COURSE_MODULES = [
+    ("s18", REPO_ROOT / "s18_worktree_isolation" / "code.py"),
+    ("s19", REPO_ROOT / "s19_mcp_plugin" / "code.py"),
+    ("s20", REPO_ROOT / "s20_comprehensive" / "code.py"),
+]
+
+
+def run_git(cwd: Path, *args: str, check: bool = True):
+    return subprocess.run(
+        ["git", *args],
+        cwd=cwd,
+        capture_output=True,
+        text=True,
+        check=check,
+    )
+
+
+def init_repo(path: Path):
+    run_git(path.parent, "init", "-b", "main", str(path))
+    run_git(path, "config", "user.name", "Worktree Test")
+    run_git(path, "config", "user.email", "worktree@example.invalid")
+    run_git(path, "commit", "--allow-empty", "-m", "base")
+
+
+def load_course_module(module_name: str, module_path: Path, cwd: Path):
+    fake_anthropic = types.ModuleType("anthropic")
+
+    class FakeAnthropic:
+        def __init__(self, *args, **kwargs):
+            self.messages = types.SimpleNamespace(create=None)
+
+    fake_dotenv = types.ModuleType("dotenv")
+    fake_yaml = types.ModuleType("yaml")
+    setattr(fake_anthropic, "Anthropic", FakeAnthropic)
+    setattr(fake_dotenv, "load_dotenv", lambda override=True: None)
+    setattr(fake_yaml, "safe_load", lambda text: {})
+    setattr(fake_yaml, "YAMLError", Exception)
+
+    previous_modules = {
+        "anthropic": sys.modules.get("anthropic"),
+        "dotenv": sys.modules.get("dotenv"),
+        "yaml": sys.modules.get("yaml"),
+    }
+    previous_cwd = Path.cwd()
+    previous_model = os.environ.get("MODEL_ID")
+    previous_key = os.environ.get("ANTHROPIC_API_KEY")
+
+    spec = importlib.util.spec_from_file_location(
+        f"{module_name}_worktree_cleanup_test", module_path
+    )
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Unable to load {module_path}")
+    module = importlib.util.module_from_spec(spec)
+
+    sys.modules["anthropic"] = fake_anthropic
+    sys.modules["dotenv"] = fake_dotenv
+    sys.modules["yaml"] = fake_yaml
+    os.environ["MODEL_ID"] = "test-model"
+    os.environ["ANTHROPIC_API_KEY"] = "test-key"
+    try:
+        os.chdir(cwd)
+        spec.loader.exec_module(module)
+        return module
+    finally:
+        os.chdir(previous_cwd)
+        if previous_model is None:
+            os.environ.pop("MODEL_ID", None)
+        else:
+            os.environ["MODEL_ID"] = previous_model
+        if previous_key is None:
+            os.environ.pop("ANTHROPIC_API_KEY", None)
+        else:
+            os.environ["ANTHROPIC_API_KEY"] = previous_key
+        for name, previous in previous_modules.items():
+            if previous is None:
+                sys.modules.pop(name, None)
+            else:
+                sys.modules[name] = previous
+
+
+def branch_exists(repo: Path, name: str) -> bool:
+    result = run_git(
+        repo,
+        "show-ref",
+        "--verify",
+        "--quiet",
+        f"refs/heads/{name}",
+        check=False,
+    )
+    return result.returncode == 0
+
+
+class WorktreeCleanupTests(unittest.TestCase):
+    def test_clean_worktree_without_unique_commits_can_be_removed(self):
+        for module_name, module_path in COURSE_MODULES:
+            with self.subTest(module=module_name), tempfile.TemporaryDirectory() as tmp:
+                repo = Path(tmp)
+                init_repo(repo)
+                module = load_course_module(module_name, module_path, repo)
+
+                module.create_worktree("demo")
+                worktree = repo / ".worktrees" / "demo"
+                run_git(worktree, "commit", "--allow-empty", "-m", "work")
+                run_git(repo, "merge", "--no-ff", "wt/demo", "-m", "merge work")
+
+                result = module.remove_worktree("demo")
+
+                self.assertIn("removed", result.lower())
+                self.assertFalse(worktree.exists())
+                self.assertFalse(branch_exists(repo, "wt/demo"))
+
+    def test_uncommitted_files_preserve_worktree(self):
+        for module_name, module_path in COURSE_MODULES:
+            with self.subTest(module=module_name), tempfile.TemporaryDirectory() as tmp:
+                repo = Path(tmp)
+                init_repo(repo)
+                module = load_course_module(module_name, module_path, repo)
+
+                module.create_worktree("demo")
+                worktree = repo / ".worktrees" / "demo"
+                (worktree / "draft.txt").write_text("not committed\n")
+
+                result = module.remove_worktree("demo")
+
+                self.assertIn("discard_changes=true", result)
+                self.assertTrue(worktree.exists())
+                self.assertTrue(branch_exists(repo, "wt/demo"))
+
+    def test_local_commit_without_upstream_preserves_worktree(self):
+        for module_name, module_path in COURSE_MODULES:
+            with self.subTest(module=module_name), tempfile.TemporaryDirectory() as tmp:
+                repo = Path(tmp)
+                init_repo(repo)
+                module = load_course_module(module_name, module_path, repo)
+
+                module.create_worktree("demo")
+                worktree = repo / ".worktrees" / "demo"
+                run_git(worktree, "commit", "--allow-empty", "-m", "local work")
+                upstream = run_git(
+                    worktree,
+                    "rev-parse",
+                    "--abbrev-ref",
+                    "@{upstream}",
+                    check=False,
+                )
+                self.assertNotEqual(upstream.returncode, 0)
+
+                result = module.remove_worktree("demo")
+
+                self.assertIn("discard_changes=true", result)
+                self.assertTrue(worktree.exists())
+                self.assertTrue(branch_exists(repo, "wt/demo"))
+
+    def test_git_verification_failures_preserve_worktree(self):
+        for module_name, module_path in COURSE_MODULES:
+            for failed_command in ("status", "rev-parse", "rev-list"):
+                with (
+                    self.subTest(module=module_name, command=failed_command),
+                    tempfile.TemporaryDirectory() as tmp,
+                ):
+                    repo = Path(tmp)
+                    init_repo(repo)
+                    module = load_course_module(module_name, module_path, repo)
+                    module.create_worktree("demo")
+                    worktree = repo / ".worktrees" / "demo"
+                    real_run = module.subprocess.run
+
+                    def run_with_failure(args, **kwargs):
+                        if args[:2] == ["git", failed_command]:
+                            if kwargs.get("check"):
+                                raise subprocess.CalledProcessError(128, args)
+                            return subprocess.CompletedProcess(args, 128, "", "failed")
+                        return real_run(args, **kwargs)
+
+                    with mock.patch.object(
+                        module.subprocess, "run", side_effect=run_with_failure
+                    ):
+                        result = module.remove_worktree("demo")
+
+                    self.assertIn("Cannot verify", result)
+                    self.assertTrue(worktree.exists())
+                    self.assertTrue(branch_exists(repo, "wt/demo"))
+
+
+if __name__ == "__main__":
+    unittest.main()