From a3977ecc854569d36c04dbc98a412f9d3aae63c3 Mon Sep 17 00:00:00 2001
From: shuiyu67 <shuiyu67@example.com>
Date: Wed, 5 Aug 2026 17:25:17 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20opencode=20=E5=BC=8F=20Plan=20=E6=A8=A1?=
 =?UTF-8?q?=E5=BC=8F=EF=BC=88=E5=8F=AA=E8=AF=BB=E8=B0=83=E7=A0=94=20+=20/b?=
 =?UTF-8?q?uild=20=E6=89=B9=E5=87=86=E6=89=A7=E8=A1=8C=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- safety.check 最高优先级加 Plan 只读拦截：写操作暂挂，只读放行
- BaseAICLI 新增 plan_mode / current_plan 状态
- _build_system_prompt 注入 PLAN 约束；/plan /build 命令路由与 handle 方法
- 速查表与启动提示补充 /plan /build
- 新增 tests/test_plan_mode.py（16 项，纯函数+拦截+注入+状态转换）

Co-Authored-By: WorkBuddy <workbuddy@tencent.com>
---
 tests/test_plan_mode.py  | 246 +++++++++++++++++++++++++++++++++++++++
 xcli_core/cli_base.py    |   3 +
 xcli_core/cli_core.py    |  82 ++++++++++++-
 xcli_core/cli_display.py |   2 +
 xcli_core/safety.py      |  67 +++++++++++
 5 files changed, 399 insertions(+), 1 deletion(-)
 create mode 100644 tests/test_plan_mode.py

diff --git a/tests/test_plan_mode.py b/tests/test_plan_mode.py
new file mode 100644
index 0000000..57ebeab
--- /dev/null
+++ b/tests/test_plan_mode.py
@@ -0,0 +1,246 @@
+"""Plan 模式（只读规划 → 审批 → 执行）回归测试。
+
+全程不实例化 AICLI（避免加载真实引擎/插件/进程保护/网络），
+通过伪造 cli 对象与轻量 SimpleNamespace 验证核心逻辑。
+"""
+import types
+
+import pytest
+
+from xcli_core.safety import (
+    get_safety,
+    plan_is_write_operation,
+    PLAN_WRITE_RULES,
+)
+
+
+# ── 纯函数：写操作判定 ──
+
+def test_write_rules_known_and_unknown():
+    # 整类拦截
+    assert plan_is_write_operation("cmd_executor", "ls -la") is True
+    assert plan_is_write_operation("browser_auto", "navigate https://x") is True
+    assert plan_is_write_operation("network_tools", "ping 1.1.1.1") is True
+    assert plan_is_write_operation("scheduler", "add 1h do") is True
+    # 整类放行
+    assert plan_is_write_operation("tool_search", "code_editor") is False
+    assert plan_is_write_operation("ai_search", "python 教程") is False
+    assert plan_is_write_operation("code_search", "find . kw") is False
+    # 未知工具：保守拦截
+    assert plan_is_write_operation("mystery_tool", "anything") is True
+
+
+def test_code_editor_readonly_allowed_write_blocked():
+    assert plan_is_write_operation("code_editor", "read_range a.py 1 10") is False
+    assert plan_is_write_operation("code_editor", "find . kw") is False
+    assert plan_is_write_operation("code_editor", "diff a.py") is False
+    assert plan_is_write_operation("code_editor", "edit a.py old <<<>>> new") is True
+    assert plan_is_write_operation("code_editor", "write a.py content") is True
+    assert plan_is_write_operation("code_editor", "append a.py line") is True
+    assert plan_is_write_operation("code_editor", "multi a.py [...]") is True
+    assert plan_is_write_operation("code_editor", "insert a.py 3 line") is True
+    assert plan_is_write_operation("code_editor", "delete_lines a.py 1 2") is True
+
+
+def test_git_readonly_allowed_write_blocked():
+    assert plan_is_write_operation("git_tools", "status") is False
+    assert plan_is_write_operation("git_tools", "log --oneline 10") is False
+    assert plan_is_write_operation("git_tools", "diff") is False
+    assert plan_is_write_operation("git_tools", "show HEAD") is False
+    assert plan_is_write_operation("git_tools", "branch") is False  # 仅列出
+    assert plan_is_write_operation("git_tools", "commit 修复") is True
+    assert plan_is_write_operation("git_tools", "smart-commit") is True  # 后缀命中
+    assert plan_is_write_operation("git_tools", "push") is True
+    assert plan_is_write_operation("git_tools", "checkout main") is True
+    assert plan_is_write_operation("git_tools", "branch -d old") is True
+    assert plan_is_write_operation("git_tools", "reset --hard") is True
+    assert plan_is_write_operation("git_tools", "fetch origin") is True
+
+
+def test_file_manager_sub_agent_auto_engineer():
+    assert plan_is_write_operation("file_manager", "list .") is False
+    assert plan_is_write_operation("file_manager", "read a.txt") is False
+    assert plan_is_write_operation("file_manager", "search kw") is False
+    assert plan_is_write_operation("file_manager", "write a.txt x") is True
+    assert plan_is_write_operation("file_manager", "delete a.txt") is True
+
+    assert plan_is_write_operation("sub_agent", "list") is False
+    assert plan_is_write_operation("sub_agent", "status") is False
+    assert plan_is_write_operation("sub_agent", "result 12") is False
+    assert plan_is_write_operation("sub_agent", "run 任务") is True
+    assert plan_is_write_operation("sub_agent", "dispatch 任务") is True
+
+    assert plan_is_write_operation("auto_engineer", "lint") is False  # 只读
+    assert plan_is_write_operation("auto_engineer", "metrics") is False
+    assert plan_is_write_operation("auto_engineer", "test") is True
+    assert plan_is_write_operation("auto_engineer", "build") is True
+
+
+# ── safety.check 在 plan 模式下的拦截 ──
+
+@pytest.fixture
+def safety_with_fake_cli():
+    safety = get_safety()
+    saved_cli = safety.cli
+    saved_mode = safety.mode
+    fake_cli = types.SimpleNamespace(plan_mode=True)
+    safety.set_cli(fake_cli)
+    yield safety
+    safety.set_cli(saved_cli)
+    safety.set_mode(saved_mode)
+
+
+def test_plan_mode_blocks_write_at_safety_layer(safety_with_fake_cli):
+    s = safety_with_fake_cli
+    allowed, msg = s.check("code_editor", "edit a.py x <<<>>> y")
+    assert allowed is False
+    assert "PLAN" in msg
+
+    allowed, _ = s.check("git_tools", "commit 修复")
+    assert allowed is False
+
+    allowed, _ = s.check("cmd_executor", "rm -rf /")
+    assert allowed is False
+
+
+def test_plan_mode_allows_readonly_at_safety_layer(safety_with_fake_cli):
+    s = safety_with_fake_cli
+    allowed, _ = s.check("code_editor", "read_range a.py 1 10")
+    assert allowed is True
+    allowed, _ = s.check("tool_search", "code_editor")
+    assert allowed is True
+    allowed, _ = s.check("git_tools", "status")
+    assert allowed is True
+
+
+def test_normal_mode_does_not_plan_block():
+    """非 plan 模式：safety.check 不被 plan 逻辑拦截（用无限制模式避免触发 AI 分析）。"""
+    safety = get_safety()
+    saved_cli = safety.cli
+    saved_mode = safety.mode
+    fake_cli = types.SimpleNamespace(plan_mode=False)
+    safety.set_cli(fake_cli)
+    safety.set_mode(0)  # MODE_UNRESTRICTED：直接放行，不消耗 AI
+    try:
+        allowed, _ = safety.check("code_editor", "write a.py x")
+        assert allowed is True
+        allowed, _ = safety.check("cmd_executor", "rm -rf /")
+        assert allowed is True
+    finally:
+        safety.set_cli(saved_cli)
+        safety.set_mode(saved_mode)
+
+
+# ── system prompt 注入 ──
+
+def test_build_system_prompt_injects_plan():
+    from xcli_core.cli_core import AICLI, PLAN_MODE_SYSTEM_APPEND
+    fake = types.SimpleNamespace(
+        plan_mode=True,
+        liugin_manager=types.SimpleNamespace(tools=[]),
+        memory_manager=None,
+    )
+    prompt = AICLI._build_system_prompt(fake)
+    assert "PLAN" in prompt
+    assert "实施计划" in prompt
+    assert PLAN_MODE_SYSTEM_APPEND in prompt
+
+
+def test_build_system_prompt_no_plan_when_off():
+    from xcli_core.cli_core import AICLI
+    fake = types.SimpleNamespace(
+        plan_mode=False,
+        liugin_manager=types.SimpleNamespace(tools=[]),
+        memory_manager=None,
+    )
+    prompt = AICLI._build_system_prompt(fake)
+    assert "实施计划" not in prompt
+
+
+# ── 命令 handler 状态转换 ──
+
+def _fake_for_plan():
+    calls = {"process": [], "autosave": 0, "sync": 0}
+    fake = types.SimpleNamespace(
+        plan_mode=False,
+        current_plan="",
+        process_conversation=lambda x: calls["process"].append(x),
+        _autosave_session=lambda: calls.__setitem__("autosave", calls["autosave"] + 1),
+        _sync_current_plan=lambda: calls.__setitem__("sync", calls["sync"] + 1),
+    )
+    return fake, calls
+
+
+def test_handle_plan_command_enters_mode_with_task():
+    from xcli_core.cli_core import AICLI
+    fake, calls = _fake_for_plan()
+    AICLI.handle_plan_command(fake, "重构缓存模块")
+    assert fake.plan_mode is True
+    assert calls["process"] == ["重构缓存模块"]
+    assert calls["autosave"] == 1
+    assert calls["sync"] == 1
+
+
+def test_handle_plan_command_enters_mode_without_task():
+    from xcli_core.cli_core import AICLI
+    fake, calls = _fake_for_plan()
+    AICLI.handle_plan_command(fake, "")
+    assert fake.plan_mode is True
+    assert calls["process"] == []  # 无任务不主动调研
+
+
+def test_handle_plan_command_off_exits():
+    from xcli_core.cli_core import AICLI
+    fake, calls = _fake_for_plan()
+    fake.plan_mode = True
+    AICLI.handle_plan_command(fake, "off")
+    assert fake.plan_mode is False
+    assert calls["process"] == []
+
+
+def test_handle_build_command_executes_plan():
+    from xcli_core.cli_core import AICLI
+    fake, calls = _fake_for_plan()
+    fake.plan_mode = True
+    fake.current_plan = "## 实施计划\n1. 改 a.py\n2. 改 b.py"
+    AICLI.handle_build_command(fake)
+    assert fake.plan_mode is False
+    assert len(calls["process"]) == 1
+    assert "已批准的实施计划" in calls["process"][0]
+    assert "## 实施计划" in calls["process"][0]
+
+
+def test_handle_build_command_no_plan_warns():
+    from xcli_core.cli_core import AICLI
+    fake, calls = _fake_for_plan()
+    fake.plan_mode = False
+    fake.current_plan = ""
+    AICLI.handle_build_command(fake)
+    assert calls["process"] == []  # 无计划不执行
+
+
+def test_sync_current_plan_captures_last_assistant():
+    from xcli_core.cli_core import AICLI
+    fake = types.SimpleNamespace(
+        plan_mode=True,
+        current_plan="",
+        shared_conversation_history=[
+            {"role": "user", "content": "任务"},
+            {"role": "assistant", "content": "调研中..."},
+            {"role": "tool", "content": "结果"},
+            {"role": "assistant", "content": "## 实施计划\n1. do x"},
+        ],
+    )
+    AICLI._sync_current_plan(fake)
+    assert "实施计划" in fake.current_plan
+
+
+def test_sync_current_plan_noop_when_not_plan():
+    from xcli_core.cli_core import AICLI
+    fake = types.SimpleNamespace(
+        plan_mode=False,
+        current_plan="unchanged",
+        shared_conversation_history=[{"role": "assistant", "content": "x"}],
+    )
+    AICLI._sync_current_plan(fake)
+    assert fake.current_plan == "unchanged"
diff --git a/xcli_core/cli_base.py b/xcli_core/cli_base.py
index dab83cb..be399c8 100644
--- a/xcli_core/cli_base.py
+++ b/xcli_core/cli_base.py
@@ -99,6 +99,9 @@ class BaseAICLI:
             os.makedirs(self.chat_history_dir)
         # 初始化会话管理器（opencode 式 /resume 持久化）
         self.init_session_manager()
+        # Plan 模式状态（只读规划 → 审批 → 执行）
+        self.plan_mode = False
+        self.current_plan = ""
 
         # 获取 skills 目录路径
         skills_dir = os.path.join(project_dir, "skills")
diff --git a/xcli_core/cli_core.py b/xcli_core/cli_core.py
index ff96ce0..93f18ee 100644
--- a/xcli_core/cli_core.py
+++ b/xcli_core/cli_core.py
@@ -22,6 +22,17 @@ from .plugin_market import PluginMarketMixin
 from .notification import notify_task_complete, get_notification_manager
 from .process_protection import get_protection_status
 
+# ── Plan 模式 system prompt 追加段（只读规划约束）──
+PLAN_MODE_SYSTEM_APPEND = """
+
+【PLAN 模式 — 只读规划】
+你当前处于 PLAN 模式。此模式下你只能进行调查、阅读、分析，不得执行任何修改。
+- 你可以使用只读工具调研代码库：code_search（搜索/结构/符号/统计）、file_manager 的 list/read/info/search、code_editor 的 read_range/find/diff、git_tools 的 status/log/diff/show/blame、tool_search（查询工具用法）、ai_search（联网搜索）。
+- 你**不能**执行写操作（编辑/创建/删除文件、运行 shell 命令、git commit/push/checkout、委派会修改的子代理等）——这些会被系统拦截并返回提示。
+- 调研完成后，输出一份清晰的、带编号步骤的「实施计划」，让用户审批。计划应包含：每步要做什么、涉及哪些文件/函数、预期结果、潜在风险。
+- 计划以如下标记开头：## 实施计划
+- 完成计划后停止，等待用户输入 /build 批准执行，或继续追问以完善计划。不要自行开始修改。"""
+
 
 class AICLI(BaseAICLI, ClawliMixin, ToolMixin, CodeExecMixin, DisplayMixin, HistoryMixin, PluginMarketMixin):
     """AI CLI主程序 - 组合所有 Mixin"""
@@ -253,6 +264,10 @@ multi 操作支持一次修改多处：
             except Exception:
                 pass
 
+        # ── Plan 模式：追加只读规划约束 ──
+        if getattr(self, 'plan_mode', False):
+            prompt += PLAN_MODE_SYSTEM_APPEND
+
         return prompt
 
     # ── JSON 解析 ──
@@ -1001,6 +1016,60 @@ multi 操作支持一次修改多处：
             for t in threads:
                 t.join(timeout=1)
 
+    # ── Plan 模式（只读规划 → 审批 → 执行）──
+
+    def _sync_current_plan(self):
+        """Plan 模式下，把对话历史中最后一条 assistant 内容记为当前计划。"""
+        if not getattr(self, 'plan_mode', False):
+            return
+        for msg in reversed(self.shared_conversation_history):
+            if isinstance(msg, dict) and msg.get("role") == "assistant" and msg.get("content"):
+                self.current_plan = msg["content"]
+                break
+
+    def handle_plan_command(self, args):
+        """处理 /plan 命令。进入 PLAN 模式（只读规划）。
+
+        /plan            进入模式，等待描述任务
+        /plan <任务>     进入模式并立即只读调研、生成实施计划
+        /plan off|exit   退出模式（计划不执行）
+        """
+        args = (args or "").strip()
+        if args in ('off', 'exit', '退出', 'cancel'):
+            self.plan_mode = False
+            print(f"{Fore.YELLOW}已退出 PLAN 模式（计划未执行）{Style.RESET_ALL}")
+            return
+
+        self.plan_mode = True
+        if args:
+            print(f"{Fore.CYAN}已进入 PLAN 模式，正在只读调研并生成实施计划...{Style.RESET_ALL}")
+            self.process_conversation(args)
+            self._autosave_session()
+            self._sync_current_plan()
+        else:
+            print(f"{Fore.CYAN}已进入 PLAN 模式。{Style.RESET_ALL}")
+            print(f"  描述你的任务，AI 将只做只读调研并给出实施计划；"
+                  f"完成后用 {Fore.WHITE}/build{Style.RESET_ALL} 批准执行，或 "
+                  f"{Fore.WHITE}/plan off{Style.RESET_ALL} 取消。{Style.RESET_ALL}")
+
+    def handle_build_command(self):
+        """处理 /build 命令。从 PLAN 模式进入执行（用已批准计划驱动）。"""
+        plan = getattr(self, 'current_plan', '') or ""
+        if not getattr(self, 'plan_mode', False) and not plan.strip():
+            print(f"{Fore.YELLOW}当前不在 PLAN 模式，且无已生成的计划{Style.RESET_ALL}")
+            return
+
+        self.plan_mode = False
+        if not plan.strip():
+            print(f"{Fore.YELLOW}尚未生成实施计划，先 /plan 描述任务让 AI 调研{Style.RESET_ALL}")
+            return
+
+        print(f"{Fore.GREEN}已批准计划，开始执行...{Style.RESET_ALL}")
+        instruction = ("【已批准的实施计划，请现在严格按照以下步骤执行，利用可用工具完成每一步；"
+                       "遇到与计划不符的情况先说明再继续】\n\n" + plan)
+        self.process_conversation(instruction)
+        self._autosave_session()
+
     # ── CLI 主循环 ──
 
     def run_tui(self):
@@ -1040,7 +1109,7 @@ multi 操作支持一次修改多处：
 
         # 一行速览，完整命令清单收进 /help
         print(f"{Fore.CYAN}/help 全部命令 · /quit 退出 · /tui 图形界面 · "
-              f"/model 换模型 · /resume 恢复会话 · @文件路径 读文件给 AI{Style.RESET_ALL}")
+              f"/model 换模型 · /resume 恢复会话 · /plan 规划模式 · @文件路径 读文件给 AI{Style.RESET_ALL}")
 
         # 历史会话提示（opencode 式 /resume）
         try:
@@ -1156,6 +1225,15 @@ multi 操作支持一次修改多处：
                     self.handle_model_command(user_input[7:].strip())
                     continue
 
+                # /plan 进入 PLAN 模式（只读规划 → 审批 → 执行）
+                if user_input == '/plan' or user_input.startswith('/plan '):
+                    sub = user_input[6:].strip() if user_input.startswith('/plan ') else ''
+                    self.handle_plan_command(sub)
+                    continue
+                if user_input == '/build':
+                    self.handle_build_command()
+                    continue
+
                 # /resume 会话恢复（opencode 式自动持久化 + 一键恢复）
                 if user_input in ('/resume', '/sessions') or \
                    user_input.startswith('/resume ') or user_input.startswith('/sessions ') or \
@@ -1418,6 +1496,8 @@ multi 操作支持一次修改多处：
                 self.process_conversation(user_input)
                 # 自动持久化当前会话（opencode 式 /resume 落盘）
                 self._autosave_session()
+                # Plan 模式下：把本轮最后一条 assistant 内容记为当前计划
+                self._sync_current_plan()
             except KeyboardInterrupt:
                 user_id_display = f"[用户ID: {self.user_id}]"
                 print(f"\n{Fore.GREEN}再见! {user_id_display}{Style.RESET_ALL}")
diff --git a/xcli_core/cli_display.py b/xcli_core/cli_display.py
index 8b85a10..86c729c 100644
--- a/xcli_core/cli_display.py
+++ b/xcli_core/cli_display.py
@@ -402,6 +402,8 @@ class DisplayMixin:
             ('/resume', '列出历史会话 (自动持久化)'),
             ('/resume <序号或ID>', '恢复指定会话继续对话'),
             ('/resume new', '开始一个全新的会话'),
+            ('/plan [任务]', '进入 PLAN 模式：只读调研并生成实施计划'),
+            ('/build', '批准并Plan模式生成的计划，开始执行'),
             ('/remote', '查看远程连接帮助'),
             ('/plugin', '管理插件市场 (安装/卸载/搜索)'),
             ('/protect', '查看进程保护状态'),
diff --git a/xcli_core/safety.py b/xcli_core/safety.py
index 8d35a83..38c2240 100644
--- a/xcli_core/safety.py
+++ b/xcli_core/safety.py
@@ -18,6 +18,65 @@ MODE_NAMES = {
     MODE_MANUAL: "人工确认",
 }
 
+# ── Plan 模式只读白名单（黑名单法）──
+# 值为 "*" 表示该工具整类视为写/有副作用，一律拦截；
+# 值为列表时，仅当首个操作词命中列表中的写操作关键字才拦截；
+# 值为 [] 表示该工具整类只读，一律放行（调研必需）；
+# 未登记的未知工具默认拦截（保守）。
+PLAN_WRITE_RULES = {
+    # 整类拦截
+    "cmd_executor": "*",
+    "browser_auto": "*",
+    "network_tools": "*",
+    "scheduler": "*",
+    # 仅部分操作拦截
+    "code_editor": ["write", "create", "append", "edit", "insert", "delete_lines",
+                    "multi", "replace", "rename", "move"],
+    "git_tools": ["commit", "push", "checkout", "reset", "merge", "revert", "clean",
+                  "stash", "tag", "fetch", "pull", "branch -", "branch delete",
+                  "branch -d", "branch -D", "remote add", "remote set", "amend"],
+    "file_manager": ["write", "delete", "move", "copy", "rename", "mkdir", "rm",
+                     "create", "save", "touch"],
+    "sub_agent": ["run", "dispatch", "execute", "spawn", "create", "new", "task", "assign"],
+    "auto_engineer": ["run", "test", "build", "fix", "format", "apply", "generate",
+                      "scaffold", "create", "refactor", "exec"],
+    "memory": ["write", "save", "add", "set", "delete", "edit", "update", "forget"],
+    # 整类放行（只读探索必需）
+    "tool_search": [],
+    "ai_search": [],
+    "code_search": [],
+}
+
+
+def plan_is_write_operation(tool_name: str, tool_args) -> bool:
+    """Plan 模式下判断一次工具调用是否属于「写操作」（应被拦截）。
+
+    返回 True 表示会修改状态/文件/系统，应拦截；False 表示只读，可放行。
+    """
+    rules = PLAN_WRITE_RULES.get(tool_name)
+    if rules is None:
+        # 未登记的未知工具：保守拦截
+        return True
+    if rules == "*":
+        return True
+    if not rules:
+        return False
+    args_l = str(tool_args).strip().lower()
+    parts = args_l.split()
+    op = parts[0] if parts else ""
+    for kw in rules:
+        kwp = kw.split()
+        if len(kwp) == 1:
+            # 单字关键字：精确匹配操作词，或作为其前缀/后缀（避免 "editor" 误中 "edit"）
+            if op == kw or op.startswith(kw) or op.endswith(kw):
+                return True
+        else:
+            # 多词规则，如 "remote add" / "branch -d"
+            if args_l.startswith(kw) or " ".join(parts[:len(kwp)]) == kw:
+                return True
+    return False
+
+
 class SafetyLayer:
     """统一安全检查层"""
 
@@ -51,6 +110,14 @@ class SafetyLayer:
         检查工具调用是否安全
         返回: (允许执行, 消息)
         """
+        # ── Plan 模式只读约束（最高优先级，先于所有安全模式判断）──
+        # plan 模式下只允许只读调研工具，写操作一律暂挂，待 /build 批准后执行。
+        cli = self.cli
+        if cli is not None and getattr(cli, 'plan_mode', False):
+            if plan_is_write_operation(tool_name, tool_args):
+                return False, "PLAN 模式：写操作已暂挂（只读调研中）。请在 /build 批准后执行。"
+            return True, ""
+
         # 无限制模式：直接放行
         if self.mode == MODE_UNRESTRICTED:
             return True, ""