feat(my-deepagent): v0.3 PR #5 — plan mode (/plan, /approve, /reject)

Claude Code의 plan mode 등가.  `/plan` 진입 시 write_file / edit_file /
execute / bash / task (sub-agent) 도구가 차단되고 read_file / glob / grep /
ls / write_todos 만 허용.

핵심 동작:
- `PlanModeMiddleware(is_active: Callable[[], bool])` 가 `awrap_tool_call` /
  `wrap_tool_call` 에서 활성 + 차단 도구면 synthetic
  `ToolMessage(status="error")` 반환.  raise 하지 않음 — LLM 이 차단 메시지를
  보고 다른 도구로 전환하거나 plan 다듬기로 자동 복귀.
- `is_active` 는 closure 라서 슬래시 토글 후 agent 재빌드 불필요.
- `InteractiveSessionRow.plan_mode` 영속 + resume 시 복원.

데이터·라이브러리:
- `middleware/plan_mode.py` (신규):
  - `BLOCKED_TOOLS_IN_PLAN_MODE = write_file / edit_file / bash / execute /
    run_command / shell / task`.
  - `PlanModeMiddleware` async + sync 양쪽 구현.

REPL 통합 (`cli/interactive.py`):
- `InteractiveSession._plan_mode: bool` + `set_plan_mode(enabled)` async →
  flag 토글 + `thread_suffix` bump + row 영속.
- resume path 에서 `sess._plan_mode = row.plan_mode` 로 복원.
- `_register_plan_mode_slash`: `/plan`, `/approve`, `/reject` 등록.
- `/reject` 는 thread 까지 리셋해 plan thread 폐기.

테스트 (`tests/integration/test_plan_mode.py`, 9 케이스):
- inactive 시 모든 도구 패스스루
- active 시 write_file / execute / task 차단 (status=error,
  tool_call_id 유지, 메시지에 도구명 + "Plan-mode" 포함)
- active 시 read_file / glob / grep / ls / write_todos 허용
- closure 토글로 동작 변경 (rebuild 없이)
- 동기 wrap_tool_call 도 동일 동작
- BLOCKED_TOOLS_IN_PLAN_MODE 상수 sanity

게이트:
- ruff check / format --check / mypy: PASS
- pytest -q --ignore=tests/integration/test_e2e_workflow.py
  --ignore=tests/integration/test_openrouter_smoke.py: 657 passed (9 신규 포함)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
chungyeong
2026-05-17 20:47:30 +09:00
parent 2685cb26db
commit fb7e67fd20
4 changed files with 380 additions and 2 deletions

View File

@@ -0,0 +1,170 @@
"""v0.3 PR #5 — Plan mode tests.
Covers:
1. PlanModeMiddleware passes tool calls through when inactive.
2. PlanModeMiddleware blocks write_file / edit_file / execute / task when active.
3. read_file / glob / grep / write_todos are allowed regardless.
4. Toggling the closure flag changes behavior without rebuilding the middleware.
5. The synthetic ToolMessage carries status="error" and a clear hint.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import pytest
from langchain_core.messages import ToolMessage
from my_deepagent.middleware.plan_mode import (
BLOCKED_TOOLS_IN_PLAN_MODE,
PlanModeMiddleware,
)
@dataclass
class _FakeToolRequest:
"""Minimal stand-in for langchain ToolCallRequest in unit tests."""
tool_call: dict[str, Any]
async def _passthrough_handler(_: _FakeToolRequest) -> ToolMessage:
"""Stub handler — returns a benign 'tool executed' message."""
return ToolMessage(content="EXECUTED", tool_call_id="t1", name="stub")
# ---------------------------------------------------------------------------
# Inactive plan-mode → all tools pass through
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_plan_mode_inactive_passes_through() -> None:
mw = PlanModeMiddleware(is_active=lambda: False)
for name in ["write_file", "edit_file", "execute", "task", "read_file", "glob"]:
req = _FakeToolRequest(tool_call={"name": name, "id": "t1", "args": {}})
result = await mw.awrap_tool_call(req, _passthrough_handler)
assert isinstance(result, ToolMessage)
assert result.content == "EXECUTED"
assert result.status != "error"
# ---------------------------------------------------------------------------
# Active plan-mode → write tools blocked with status=error
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_plan_mode_active_blocks_write_file() -> None:
mw = PlanModeMiddleware(is_active=lambda: True)
req = _FakeToolRequest(
tool_call={"name": "write_file", "id": "abc123", "args": {"file_path": "/tmp/x"}}
)
result = await mw.awrap_tool_call(req, _passthrough_handler)
assert isinstance(result, ToolMessage)
assert result.status == "error"
assert result.tool_call_id == "abc123"
assert "Plan-mode" in result.content
assert "write_file" in result.content
@pytest.mark.asyncio
async def test_plan_mode_active_blocks_execute() -> None:
mw = PlanModeMiddleware(is_active=lambda: True)
req = _FakeToolRequest(
tool_call={"name": "execute", "id": "exec1", "args": {"command": "ls"}}
)
result = await mw.awrap_tool_call(req, _passthrough_handler)
assert isinstance(result, ToolMessage)
assert result.status == "error"
assert "execute" in result.content
@pytest.mark.asyncio
async def test_plan_mode_active_blocks_task_subagent_spawn() -> None:
mw = PlanModeMiddleware(is_active=lambda: True)
req = _FakeToolRequest(
tool_call={"name": "task", "id": "task1", "args": {"description": "x"}}
)
result = await mw.awrap_tool_call(req, _passthrough_handler)
assert isinstance(result, ToolMessage)
assert result.status == "error"
assert "task" in result.content
# ---------------------------------------------------------------------------
# Active plan-mode → read-only tools still pass through
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_plan_mode_active_allows_read_only_tools() -> None:
mw = PlanModeMiddleware(is_active=lambda: True)
for name in ["read_file", "glob", "grep", "ls", "write_todos"]:
req = _FakeToolRequest(tool_call={"name": name, "id": "t1", "args": {}})
result = await mw.awrap_tool_call(req, _passthrough_handler)
assert result.content == "EXECUTED", f"{name} should not be blocked"
assert result.status != "error"
# ---------------------------------------------------------------------------
# Closure-toggle behavior — flip without rebuild
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_plan_mode_closure_toggle_changes_behavior() -> None:
state = {"on": False}
mw = PlanModeMiddleware(is_active=lambda: state["on"])
req = _FakeToolRequest(tool_call={"name": "write_file", "id": "w", "args": {}})
# Off → passes.
r1 = await mw.awrap_tool_call(req, _passthrough_handler)
assert r1.status != "error"
# Flip on → blocks.
state["on"] = True
r2 = await mw.awrap_tool_call(req, _passthrough_handler)
assert r2.status == "error"
# Flip back off → passes again.
state["on"] = False
r3 = await mw.awrap_tool_call(req, _passthrough_handler)
assert r3.status != "error"
# ---------------------------------------------------------------------------
# Sync path mirrors async path
# ---------------------------------------------------------------------------
def test_plan_mode_sync_wrap_tool_call() -> None:
mw = PlanModeMiddleware(is_active=lambda: True)
def sync_handler(_: _FakeToolRequest) -> ToolMessage:
return ToolMessage(content="EXECUTED", tool_call_id="t1", name="stub")
req = _FakeToolRequest(tool_call={"name": "write_file", "id": "s1", "args": {}})
result = mw.wrap_tool_call(req, sync_handler)
assert isinstance(result, ToolMessage)
assert result.status == "error"
# ---------------------------------------------------------------------------
# Blocklist constant sanity
# ---------------------------------------------------------------------------
def test_blocklist_includes_all_known_write_tools() -> None:
assert "write_file" in BLOCKED_TOOLS_IN_PLAN_MODE
assert "edit_file" in BLOCKED_TOOLS_IN_PLAN_MODE
assert "execute" in BLOCKED_TOOLS_IN_PLAN_MODE
assert "bash" in BLOCKED_TOOLS_IN_PLAN_MODE
assert "task" in BLOCKED_TOOLS_IN_PLAN_MODE
def test_blocklist_excludes_read_only_and_planning_tools() -> None:
for name in ("read_file", "glob", "grep", "ls", "write_todos"):
assert name not in BLOCKED_TOOLS_IN_PLAN_MODE