diff --git a/dare_framework/plan_v2/__init__.py b/dare_framework/plan_v2/__init__.py
index e8ac8567..cc92bcfd 100644
--- a/dare_framework/plan_v2/__init__.py
+++ b/dare_framework/plan_v2/__init__.py
@@ -7,16 +7,21 @@
 from dare_framework.plan_v2.registry import SubAgentRegistry
 from dare_framework.plan_v2.types import (
     Milestone,
+    PlanStateName,
     PlannerState,
+    STEP_STATES,
     Step,
     Task,
+    is_valid_state_transition,
 )
 from dare_framework.plan_v2.prompts import PLAN_AGENT_SYSTEM_PROMPT, SUB_AGENT_TASK_PROMPT
 from dare_framework.plan_v2.tools import (
     CreatePlanTool,
     DecomposeTaskTool,
     DelegateToSubAgentTool,
+    FinishPlanTool,
     ReflectTool,
+    ReviseCurrentPlanTool,
     ValidatePlanTool,
     VerifyMilestoneTool,
 )
@@ -25,15 +30,20 @@
     "CreatePlanTool",
     "DecomposeTaskTool",
     "DelegateToSubAgentTool",
+    "FinishPlanTool",
     "Milestone",
     "PLAN_AGENT_SYSTEM_PROMPT",
+    "PlanStateName",
     "Planner",
     "SUB_AGENT_TASK_PROMPT",
+    "STEP_STATES",
     "PlannerState",
     "ReflectTool",
+    "ReviseCurrentPlanTool",
     "Step",
     "SubAgentRegistry",
     "Task",
     "ValidatePlanTool",
     "VerifyMilestoneTool",
+    "is_valid_state_transition",
 ]
diff --git a/dare_framework/plan_v2/planner.py b/dare_framework/plan_v2/planner.py
index 88d00ce9..e88d0ed7 100644
--- a/dare_framework/plan_v2/planner.py
+++ b/dare_framework/plan_v2/planner.py
@@ -9,7 +9,9 @@
 from dare_framework.plan_v2.tools import (
     CreatePlanTool,
     DecomposeTaskTool,
+    FinishPlanTool,
     ReflectTool,
+    ReviseCurrentPlanTool,
     SubAgentTool,
     ValidatePlanTool,
     VerifyMilestoneTool,
@@ -30,6 +32,8 @@ def __init__(
         self._tools: list[ITool] = [
             CreatePlanTool(self._state),
             ValidatePlanTool(self._state),
+            ReviseCurrentPlanTool(self._state),
+            FinishPlanTool(self._state),
             VerifyMilestoneTool(self._state),
             ReflectTool(self._state),
             DecomposeTaskTool(self._state),
diff --git a/dare_framework/plan_v2/prompts.py b/dare_framework/plan_v2/prompts.py
index 98076f47..002076ab 100644
--- a/dare_framework/plan_v2/prompts.py
+++ b/dare_framework/plan_v2/prompts.py
@@ -1,11 +1,13 @@
 """Plan Agent and sub-agent prompts."""
 
-PLAN_AGENT_SYSTEM_PROMPT = """You are a planning agent. You have tools to create and validate a plan, and to delegate steps to sub-agents. Each sub-agent (e.g. sub_agent_general) is a tool: pass the step description as "task" and the step identifier as "step_id".
-Before every action, check the [Plan State] block: it shows Phase, Completed, Pending, and **NEXT**. Follow **NEXT** strictly; do not repeat completed steps.
+PLAN_AGENT_SYSTEM_PROMPT = """You are a planning agent. You have tools to create, validate, revise, and finish a plan, and to delegate steps to sub-agents. Each sub-agent (e.g. sub_agent_general) is a tool: pass the step description as "task" and the step identifier as "step_id".
+Before every action, check the [Plan State] block: it shows Plan Status, step statuses, Completed, Pending, and **NEXT**. Follow **NEXT** strictly; do not repeat completed steps.
 When the user gives a task:
 1. Call create_plan exactly once with plan_description (short summary) and steps: a list of objects, each with step_id (e.g. "step1"), description (what to do), and optional params (dict). Do NOT call create_plan again after success.
 2. Call validate_plan(success=True) to confirm the plan.
 3. For each step, call the matching sub-agent tool (e.g. sub_agent_general) with task=<step description> and step_id=<step_id from plan>. Delegate each step exactly once; do not repeat a completed step. Check the tool result's "progress" field: only delegate steps that are in "Pending", never repeat steps in "Completed". Execute steps in order.
+4. If plan content must change, call revise_current_plan and then validate_plan(success=True) again.
+5. When all steps are terminal, call finish_plan(target_state="done" or "abandoned") to close the plan.
 Keep steps concise and ordered. Use sub-agent tools (sub_agent_general, sub_agent_special_*) to execute steps; do not execute steps yourself."""
 
 SUB_AGENT_TASK_PROMPT = """You receive a task from the plan agent. Execute it using your tools and return a clear result. Be concise."""
diff --git a/dare_framework/plan_v2/tools.py b/dare_framework/plan_v2/tools.py
index 441ce0d0..39817ced 100644
--- a/dare_framework/plan_v2/tools.py
+++ b/dare_framework/plan_v2/tools.py
@@ -18,7 +18,29 @@
 )
 
 from dare_framework.plan_v2.registry import SubAgentRegistry
-from dare_framework.plan_v2.types import Milestone, PlannerState, Step
+from dare_framework.plan_v2.types import (
+    Milestone,
+    PlanStateName,
+    PlannerState,
+    STEP_STATES,
+    Step,
+)
+
+_TERMINAL_STATES: set[str] = {"done", "abandoned"}
+_PENDING_STATES: set[str] = {"todo", "in_progress"}
+
+
+def _step_state(step: Step) -> PlanStateName:
+    """Read a step lifecycle state with fallback for legacy objects."""
+    raw = getattr(step, "status", "todo")
+    if isinstance(raw, str) and raw in STEP_STATES:
+        return raw
+    return "todo"
+
+
+def _pending_steps(state: PlannerState) -> list[Step]:
+    """Return non-terminal steps."""
+    return [step for step in state.steps if _step_state(step) in _PENDING_STATES]
 
 
 def _format_critical_block(state: PlannerState) -> str:
@@ -29,27 +51,38 @@ def _format_critical_block(state: PlannerState) -> str:
             "- Phase: no_plan\n"
             "- **NEXT**: Call create_plan with plan_description and steps."
         )
-    completed = sorted(state.completed_step_ids)
-    pending = [s.step_id for s in state.steps if s.step_id not in state.completed_step_ids]
+    state.sync_completed_step_ids()
+    completed = [step.step_id for step in state.steps if _step_state(step) == "done"]
+    pending = [step.step_id for step in state.steps if _step_state(step) in _PENDING_STATES]
+    abandoned = [step.step_id for step in state.steps if _step_state(step) == "abandoned"]
     lines = [
         "## [Plan State] (check before every action)",
         "",
         f"- Plan: {state.plan_description}",
+        f"- Plan Status: {state.plan_status}",
         "- Steps:",
     ]
     for i, st in enumerate(state.steps, 1):
-        lines.append(f"  [{i}] {st.step_id}: {st.description}")
-    lines.extend(["", f"- Completed: {completed}", f"- Pending: {pending}"])
+        lines.append(f"  [{i}] {st.step_id} [{_step_state(st)}]: {st.description}")
+    lines.extend(["", f"- Completed: {completed}", f"- Pending: {pending}", f"- Abandoned: {abandoned}"])
+
+    if state.plan_status in _TERMINAL_STATES:
+        lines.append(f"- **NEXT**: Plan is terminal (`{state.plan_status}`). Report result and stop planning tools.")
+        return "\n".join(lines)
+
     if not state.plan_validated:
         lines.append("- **NEXT**: Call validate_plan(success=True) to confirm the plan.")
     elif pending:
-        next_step = next(s for s in state.steps if s.step_id == pending[0])
+        next_step = next(step for step in state.steps if step.step_id == pending[0])
         lines.append(
             f"- **NEXT**: Call sub-agent (e.g. sub_agent_general) with task=<{next_step.step_id} description> "
             f"and step_id={next_step.step_id}. Do NOT repeat completed steps."
         )
     else:
-        lines.append("- **NEXT**: All steps completed. Summarize results and report to user.")
+        target = "done" if not abandoned else "abandoned"
+        lines.append(
+            f"- **NEXT**: All steps are terminal. Call finish_plan(target_state=\"{target}\") to close this plan."
+        )
     return "\n".join(lines)
 
 
@@ -137,10 +170,13 @@ async def execute(
                 step_id=s.get("step_id", ""),
                 description=s.get("description", ""),
                 params=s.get("params") or {},
+                status="todo",
             )
             for s in steps
         ]
+        self._state.plan_status = "todo"
         self._state.completed_step_ids.clear()
+        self._state.plan_validated = False
         self._state.critical_block = _format_critical_block(self._state)
         print("\n--- Plan Created ---")
         print(f"  plan_description: {plan_description}")
@@ -217,8 +253,221 @@ async def execute(
         self._state.plan_errors = list(errors or [])
         if success:
             self._state.plan_validated = True
+            if self._state.plan_status == "todo":
+                self._state.plan_status = "in_progress"
+        else:
+            self._state.plan_validated = False
+        self._state.critical_block = _format_critical_block(self._state)
+        return ToolResult(
+            success=True,
+            output={
+                "plan_success": success,
+                "plan_status": self._state.plan_status,
+                "errors": self._state.plan_errors,
+            },
+        )
+
+
+class ReviseCurrentPlanTool(ITool):
+    """Revise the current plan definition while keeping stable completed progress by step_id."""
+
+    def __init__(self, state: PlannerState) -> None:
+        self._state = state
+
+    @property
+    def name(self) -> str:
+        return "revise_current_plan"
+
+    @property
+    def description(self) -> str:
+        return (
+            "Revise current plan_description and/or steps. "
+            "Terminal plans (done/abandoned) cannot be revised."
+        )
+
+    @property
+    def input_schema(self) -> dict[str, Any]:
+        return infer_input_schema_from_execute(type(self).execute)
+
+    @property
+    def output_schema(self) -> dict[str, Any] | None:
+        return infer_output_schema_from_execute(type(self).execute)
+
+    @property
+    def tool_type(self) -> ToolType:
+        return ToolType.ATOMIC
+
+    @property
+    def risk_level(self) -> RiskLevelName:
+        return "read_only"
+
+    @property
+    def requires_approval(self) -> bool:
+        return False
+
+    @property
+    def timeout_seconds(self) -> int:
+        return 30
+
+    @property
+    def is_work_unit(self) -> bool:
+        return False
+
+    @property
+    def capability_kind(self) -> CapabilityKind:
+        return CapabilityKind.PLAN_TOOL
+
+    async def execute(
+        self,
+        *,
+        run_context: RunContext[Any],
+        plan_description: str | None = None,
+        steps: list[dict[str, Any]] | None = None,
+        **kwargs: Any,
+    ) -> ToolResult[dict[str, Any]]:
+        """Revise current plan and reset validation gate."""
+        _ = run_context
+        if not self._state.steps:
+            return ToolResult(success=False, output=None, error="no active plan to revise")
+        if self._state.plan_status in _TERMINAL_STATES:
+            return ToolResult(
+                success=False,
+                output=None,
+                error=f"cannot revise terminal plan ({self._state.plan_status})",
+            )
+        if plan_description is None and steps is None:
+            return ToolResult(success=False, output=None, error="no revision payload provided")
+
+        if plan_description is not None:
+            self._state.plan_description = str(plan_description)
+
+        if steps is not None:
+            old_by_id = {step.step_id: _step_state(step) for step in self._state.steps}
+            revised_steps: list[Step] = []
+            for raw in steps:
+                step_id = str(raw.get("step_id", "")).strip()
+                if not step_id:
+                    continue
+                preserved = old_by_id.get(step_id, "todo")
+                status: PlanStateName = preserved if preserved in _TERMINAL_STATES else "todo"
+                revised_steps.append(
+                    Step(
+                        step_id=step_id,
+                        description=str(raw.get("description", "")),
+                        params=raw.get("params") or {},
+                        status=status,
+                    )
+                )
+            self._state.steps = revised_steps
+            self._state.sync_completed_step_ids()
+
+        self._state.plan_validated = False
+        self._state.plan_status = "todo"
+        self._state.critical_block = _format_critical_block(self._state)
+        return ToolResult(
+            success=True,
+            output={
+                "plan_description": self._state.plan_description,
+                "steps_count": len(self._state.steps),
+                "next_action": "Call validate_plan(success=True) after revision.",
+            },
+        )
+
+
+class FinishPlanTool(ITool):
+    """Explicitly mark plan as done/abandoned with state guardrails."""
+
+    def __init__(self, state: PlannerState) -> None:
+        self._state = state
+
+    @property
+    def name(self) -> str:
+        return "finish_plan"
+
+    @property
+    def description(self) -> str:
+        return "Mark current plan terminal with target_state in {'done','abandoned'}."
+
+    @property
+    def input_schema(self) -> dict[str, Any]:
+        return infer_input_schema_from_execute(type(self).execute)
+
+    @property
+    def output_schema(self) -> dict[str, Any] | None:
+        return infer_output_schema_from_execute(type(self).execute)
+
+    @property
+    def tool_type(self) -> ToolType:
+        return ToolType.ATOMIC
+
+    @property
+    def risk_level(self) -> RiskLevelName:
+        return "read_only"
+
+    @property
+    def requires_approval(self) -> bool:
+        return False
+
+    @property
+    def timeout_seconds(self) -> int:
+        return 20
+
+    @property
+    def is_work_unit(self) -> bool:
+        return False
+
+    @property
+    def capability_kind(self) -> CapabilityKind:
+        return CapabilityKind.PLAN_TOOL
+
+    async def execute(
+        self,
+        *,
+        run_context: RunContext[Any],
+        target_state: str = "done",
+        summary: str | None = None,
+        **kwargs: Any,
+    ) -> ToolResult[dict[str, Any]]:
+        """Finish the plan with deterministic terminal-state rules."""
+        _ = run_context
+        if target_state not in _TERMINAL_STATES:
+            return ToolResult(
+                success=False,
+                output=None,
+                error=f"invalid target_state: {target_state}",
+            )
+        if not self._state.steps:
+            return ToolResult(success=False, output=None, error="no active plan to finish")
+
+        pending = [step.step_id for step in _pending_steps(self._state)]
+        if target_state == "done" and pending:
+            return ToolResult(
+                success=False,
+                output=None,
+                error=f"pending steps exist: {pending}",
+            )
+
+        if target_state == "abandoned":
+            for step in self._state.steps:
+                if _step_state(step) in _PENDING_STATES:
+                    self._state.transition_step(step.step_id, "abandoned")
+
+        try:
+            self._state.transition_plan(target_state)
+        except ValueError as exc:
+            return ToolResult(success=False, output=None, error=str(exc))
+
+        if summary:
+            self._state.last_remediation_summary = str(summary)
         self._state.critical_block = _format_critical_block(self._state)
-        return ToolResult(success=True, output={"plan_success": success, "errors": self._state.plan_errors})
+        return ToolResult(
+            success=True,
+            output={
+                "plan_status": self._state.plan_status,
+                "pending": [step.step_id for step in _pending_steps(self._state)],
+                "completed": sorted(self._state.completed_step_ids),
+            },
+        )
 
 
 class VerifyMilestoneTool(ITool):
@@ -537,21 +786,37 @@ async def execute(
     ) -> ToolResult[Any]:
         """Run the sub-agent with the given task (e.g. step description). Returns the sub-agent result.
         When step_id is provided and call succeeds, marks that step as completed and adds progress to output."""
+        _ = run_context
         task_preview = (task[:120] + "...") if len(task) > 120 else task
         print(f"\n>>> 委托 {self._sub_agent_id}: {task_preview}\n", flush=True)
+        if step_id and self._state:
+            step = self._state.get_step(step_id)
+            if step is None:
+                return ToolResult(success=False, output=None, error=f"unknown step_id: {step_id}")
+            if _step_state(step) in _TERMINAL_STATES:
+                return ToolResult(
+                    success=False,
+                    output=None,
+                    error=f"step already terminal: {step_id} ({_step_state(step)})",
+                )
+            self._state.transition_step(step_id, "in_progress")
+            if self._state.plan_status == "todo":
+                self._state.plan_status = "in_progress"
+            self._state.critical_block = _format_critical_block(self._state)
         try:
             result = await self._registry.run(self._sub_agent_id, task, **kwargs)
             if step_id and self._state:
-                self._state.completed_step_ids.add(step_id)
+                self._state.transition_step(step_id, "done")
                 self._state.critical_block = _format_critical_block(self._state)
             print(f"<<< {self._sub_agent_id} 返回 (success=True)\n", flush=True)
             output = result
             if self._state and self._state.steps:
+                self._state.sync_completed_step_ids()
                 completed = sorted(self._state.completed_step_ids)
                 pending = [
                     s.step_id
                     for s in self._state.steps
-                    if s.step_id not in self._state.completed_step_ids
+                    if _step_state(s) in _PENDING_STATES
                 ]
                 progress = f"Completed: {completed}. Pending: {pending}."
                 if isinstance(result, dict):
@@ -560,6 +825,8 @@ async def execute(
                     output = {"result": result, "progress": progress}
             return ToolResult(success=True, output=output)
         except Exception as exc:
+            if step_id and self._state:
+                self._state.critical_block = _format_critical_block(self._state)
             print(f"<<< {self._sub_agent_id} 返回 (error: {exc})\n", flush=True)
             return ToolResult(success=False, output=None, error=str(exc))
 
@@ -568,7 +835,9 @@ async def execute(
     "CreatePlanTool",
     "DecomposeTaskTool",
     "DelegateToSubAgentTool",
+    "FinishPlanTool",
     "ReflectTool",
+    "ReviseCurrentPlanTool",
     "SubAgentTool",
     "ValidatePlanTool",
     "VerifyMilestoneTool",
diff --git a/dare_framework/plan_v2/types.py b/dare_framework/plan_v2/types.py
index 471b73ca..7ef5691c 100644
--- a/dare_framework/plan_v2/types.py
+++ b/dare_framework/plan_v2/types.py
@@ -9,7 +9,49 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Any
+from typing import Any, Literal
+
+
+PlanStateName = Literal["todo", "in_progress", "done", "abandoned"]
+STEP_STATES: tuple[PlanStateName, ...] = ("todo", "in_progress", "done", "abandoned")
+_ALLOWED_STATE_TRANSITIONS: dict[str, set[str]] = {
+    "todo": {"todo", "in_progress", "abandoned"},
+    "in_progress": {"in_progress", "done", "abandoned"},
+    "done": {"done"},
+    "abandoned": {"abandoned"},
+}
+
+
+def is_valid_state_transition(current: str, next_state: str) -> bool:
+    """Return whether a plan/step lifecycle transition is legal."""
+    if current not in _ALLOWED_STATE_TRANSITIONS:
+        return False
+    return next_state in _ALLOWED_STATE_TRANSITIONS[current]
+
+
+def _normalize_state(value: Any, *, default: PlanStateName) -> PlanStateName:
+    """Normalize arbitrary values to known lifecycle states."""
+    if isinstance(value, str) and value in _ALLOWED_STATE_TRANSITIONS:
+        return value
+    return default
+
+
+def _step_id(step: Any) -> str | None:
+    """Read step_id from Step-like objects with legacy dict fallback."""
+    raw = getattr(step, "step_id", None)
+    if raw is None and isinstance(step, dict):
+        raw = step.get("step_id")
+    if isinstance(raw, str) and raw.strip():
+        return raw
+    return None
+
+
+def _step_state(step: Any) -> PlanStateName:
+    """Read lifecycle state from Step-like objects with legacy dict fallback."""
+    raw = getattr(step, "status", None)
+    if raw is None and isinstance(step, dict):
+        raw = step.get("status")
+    return _normalize_state(raw, default="todo")
 
 
 # -----------------------------------------------------------------------------
@@ -41,7 +83,7 @@ class Milestone:
 
 
 # -----------------------------------------------------------------------------
-# Step: definition only (no capability_id, no runtime state)
+# Step: definition + lifecycle state (still no capability_id)
 # -----------------------------------------------------------------------------
 
 
@@ -49,13 +91,14 @@ class Milestone:
 class Step:
     """Single step: what to do, not how. Which tool to use is decided by executor.
 
-    Step is pure definition. Verification and remediation happen at milestone level
-    (like dare_agent), not per-step.
+    Verification and remediation happen at milestone level (like dare_agent), but
+    runtime lifecycle is explicitly tracked by status.
     """
 
     step_id: str
     description: str
     params: dict[str, Any] = field(default_factory=dict)
+    status: PlanStateName = "todo"
 
 
 # -----------------------------------------------------------------------------
@@ -86,6 +129,7 @@ class PlannerState:
     # Current plan
     plan_description: str = ""
     steps: list[Step] = field(default_factory=list)
+    plan_status: PlanStateName = "todo"
     completed_step_ids: set[str] = field(default_factory=set)
     plan_success: bool = True
     plan_errors: list[str] = field(default_factory=list)
@@ -97,6 +141,47 @@ class PlannerState:
     # Critical block: injected into each LLM round. Updated by plan tools when they mutate state.
     critical_block: str = ""
 
+    def sync_completed_step_ids(self) -> None:
+        """Rebuild compatibility completed-step set from step statuses."""
+        completed: set[str] = set()
+        for step in self.steps:
+            if _step_state(step) != "done":
+                continue
+            step_id = _step_id(step)
+            if step_id is not None:
+                completed.add(step_id)
+        self.completed_step_ids = completed
+
+    def get_step(self, step_id: str) -> Step | None:
+        """Lookup a step by identifier."""
+        for step in self.steps:
+            if _step_id(step) == step_id:
+                return step
+        return None
+
+    def transition_plan(self, next_state: PlanStateName) -> None:
+        """Transition plan state with legality checks."""
+        current = _normalize_state(self.plan_status, default="todo")
+        normalized_next = _normalize_state(next_state, default="todo")
+        if not is_valid_state_transition(current, normalized_next):
+            raise ValueError(f"invalid plan transition: {current} -> {normalized_next}")
+        self.plan_status = normalized_next
+
+    def transition_step(self, step_id: str, next_state: PlanStateName) -> None:
+        """Transition a step state and keep compatibility fields in sync."""
+        step = self.get_step(step_id)
+        if step is None:
+            raise ValueError(f"unknown step_id: {step_id}")
+        current = _step_state(step)
+        normalized_next = _normalize_state(next_state, default="todo")
+        if not is_valid_state_transition(current, normalized_next):
+            raise ValueError(f"invalid step transition: {current} -> {normalized_next} ({step_id})")
+        if isinstance(step, dict):
+            step["status"] = normalized_next
+        else:
+            setattr(step, "status", normalized_next)
+        self.sync_completed_step_ids()
+
     def copy_for_execution(self) -> PlannerState:
         """Produce clean state for Execution Agent. Strips plan runtime state."""
         steps = [
@@ -109,5 +194,6 @@ def copy_for_execution(self) -> PlannerState:
             current_milestone_id=self.current_milestone_id,
             plan_description=self.plan_description,
             steps=steps,
+            plan_status="todo",
             plan_success=True,
         )
diff --git a/docs/design/modules/README.md b/docs/design/modules/README.md
index c1c9b790..0b384d46 100644
--- a/docs/design/modules/README.md
+++ b/docs/design/modules/README.md
@@ -14,6 +14,7 @@
 - context: `docs/design/modules/context/README.md` (assessment: `docs/design/modules/context/Assessment.md`)
 - tool: `docs/design/modules/tool/README.md`
 - plan: `docs/design/modules/plan/README.md`
+- plan_v2: `docs/design/modules/plan_v2/README.md`
 - model (overview): `docs/design/modules/model/README.md` (assessment: `docs/design/modules/model/Assessment.md`)
 - model (prompt): `docs/design/modules/model/Model_Prompt_Management.md`
 - security: `docs/design/modules/security/README.md`
diff --git a/docs/design/modules/plan_v2/README.md b/docs/design/modules/plan_v2/README.md
new file mode 100644
index 00000000..80af8bd9
--- /dev/null
+++ b/docs/design/modules/plan_v2/README.md
@@ -0,0 +1,64 @@
+# Module: plan_v2
+
+> Status: draft baseline for AgentScope D7 gap closure (2026-03-02).
+
+## 1. 定位与职责
+
+- `plan_v2` 提供 Plan Agent 的运行期计划状态容器（`PlannerState`）与计划工具集（`create_plan / validate_plan / revise_current_plan / finish_plan / ...`）。
+- 模块目标是让计划具备可追踪生命周期：`todo -> in_progress -> done/abandoned`，并把状态提示通过 `critical_block` 注入执行环。
+- 与 `dare_framework/plan` 的关系：`plan` 负责通用 planner/validator 协议；`plan_v2` 负责 AgentScope 兼容路径下的工具化计划状态机。
+
+## 2. 总体架构
+
+- `types.py`：定义 `Task` / `Milestone` / `Step` / `PlannerState` 及状态迁移规则。
+- `tools.py`：定义对 `PlannerState` 的唯一写入口（plan tools），并在状态变更后刷新 `critical_block`。
+- `planner.py`：作为 `IToolProvider` 暴露 plan tools；由 `ReactAgent` 在执行时挂载消费。
+- `prompts.py`：给 Plan Agent 提供固定执行顺序与 tool 使用约束。
+
+## 3. 核心流程
+
+1. Plan Agent 调用 `create_plan` 写入计划与步骤，初始状态为 `todo`。
+2. 调用 `validate_plan` 后，计划进入 `in_progress`。
+3. 执行阶段通过 sub-agent 工具推进 step 状态迁移。
+4. 如需调整计划，调用 `revise_current_plan` 替换或补充步骤。
+5. 全部步骤完成后调用 `finish_plan(target_state=\"done\")`，或中止时调用 `finish_plan(target_state=\"abandoned\")`。
+6. 每次状态变化都更新 `critical_block`，供 `ReactAgent` 注入下一轮系统提示。
+
+## 4. 数据结构
+
+- `Step`
+  - `step_id: str`
+  - `description: str`
+  - `params: dict[str, Any]`
+  - `status: Literal[\"todo\", \"in_progress\", \"done\", \"abandoned\"]`
+- `PlannerState`
+  - `plan_status: Literal[\"todo\", \"in_progress\", \"done\", \"abandoned\"]`
+  - `steps: list[Step]`
+  - `completed_step_ids: set[str]`（兼容历史调用路径，和 step 状态保持一致）
+  - `critical_block: str`
+
+## 5. 关键接口
+
+- `CreatePlanTool.execute(...)`：创建计划并初始化状态。
+- `ValidatePlanTool.execute(...)`：写入验证结果并推动计划进入执行态。
+- `ReviseCurrentPlanTool.execute(...)`：在未终态时修订当前计划。
+- `FinishPlanTool.execute(...)`：显式完成或放弃计划。
+- `SubAgentTool.execute(...)`：按 `step_id` 推进步骤状态并回写进度。
+
+## 6. 异常与错误处理
+
+- 非法状态迁移（如 `done -> in_progress`）必须拒绝并返回结构化错误。
+- `finish_plan(target_state=\"done\")` 在存在未完成步骤时必须失败，避免伪完成。
+- 对未知 `step_id` 的执行委托必须显式失败，禁止静默跳过。
+- 任何状态写入后都必须刷新 `critical_block`，避免提示与真实状态漂移。
+
+## 7. 测试锚点
+
+- `tests/unit/test_plan_v2_tools.py`（状态机迁移、finish/revise、critical_block 联动）
+- `tests/unit/test_react_agent_gateway_injection.py`（plan state 注入行为不回归）
+
+## 能力状态（landed / partial / planned）
+
+- `landed`: `PlannerState` 生命周期状态机（`todo -> in_progress -> done/abandoned`）与 `create/validate/revise/finish` 主流程已接入。
+- `partial`: 兼容路径仍保留 `completed_step_ids` 等历史字段，并需要在工具层持续维持双写一致性。
+- `planned`: 后续将继续收敛与 `dare_framework/plan` 的语义边界，减少并行状态表达与重复约束定义。
diff --git a/docs/features/agentscope-d7-plan-state-tools.md b/docs/features/agentscope-d7-plan-state-tools.md
new file mode 100644
index 00000000..24132be7
--- /dev/null
+++ b/docs/features/agentscope-d7-plan-state-tools.md
@@ -0,0 +1,97 @@
+---
+change_ids: ["agentscope-d7-plan-state-tools"]
+doc_kind: feature
+topics: ["agentscope", "plan_v2", "state-machine", "critical-block"]
+created: 2026-03-02
+updated: 2026-03-02
+status: in_review
+mode: openspec
+---
+
+# Feature: agentscope-d7-plan-state-tools
+
+## Scope
+补齐 AgentScope 迁移 D7：`plan_v2` 计划状态机与原生计划工具能力，覆盖 `todo/in_progress/done/abandoned` 状态语义、`revise_current_plan`、`finish_plan`，并确保 `critical_block` 与真实计划状态同步。
+
+## OpenSpec Artifacts
+- Proposal: `openspec/changes/agentscope-d7-plan-state-tools/proposal.md`
+- Design: `openspec/changes/agentscope-d7-plan-state-tools/design.md`
+- Specs:
+  - `openspec/changes/agentscope-d7-plan-state-tools/specs/plan-runtime/spec.md`
+  - `openspec/changes/agentscope-d7-plan-state-tools/specs/chat-runtime/spec.md`
+- Tasks: `openspec/changes/agentscope-d7-plan-state-tools/tasks.md`
+
+## Progress
+- 已完成：D7 kickoff（claim active + docs/design 基线 + OpenSpec 切片初始化）。
+- 已完成：D7-1~D7-4 实现（状态机 + revise/finish 工具 + critical_block 联动）。
+- 已完成：OpenSpec tasks 全部勾选（12/12）。
+- 已完成：提交 PR #138，进入评审阶段。
+- 待完成：评审反馈处理与合并门禁。
+
+## Evidence
+
+### Commands
+- `openspec new change "agentscope-d7-plan-state-tools"`
+- `/Users/lang/workspace/github/Deterministic-Agent-Runtime-Engine/.venv/bin/pytest -q`
+- `/Users/lang/workspace/github/Deterministic-Agent-Runtime-Engine/.venv/bin/pytest -q tests/unit/test_plan_v2_tools.py`
+- `/Users/lang/workspace/github/Deterministic-Agent-Runtime-Engine/.venv/bin/pytest -q tests/unit/test_plan_v2_tools.py tests/unit/test_react_agent_gateway_injection.py tests/unit/test_dare_agent_step_driven_mode.py`
+- `openspec status --change "agentscope-d7-plan-state-tools" --json`
+
+### Results
+- 新建 OpenSpec change：`agentscope-d7-plan-state-tools`（schema: `spec-driven`）。
+- 当前分支基线：`513 passed, 12 skipped, 1 warning`。
+- D7 红灯阶段：`tests/unit/test_plan_v2_tools.py` 初次运行因缺失 `FinishPlanTool` 导致 collection error。
+- D7 定向回归：`tests/unit/test_plan_v2_tools.py` => `5 passed`。
+- D7 影响面回归：`31 passed, 1 warning`。
+- 全量回归：`518 passed, 12 skipped, 1 warning`。
+- OpenSpec status：artifacts `proposal/design/specs/tasks` 全部 `done`。
+
+### Contract Delta
+- `schema`: `Step.status` 与 `PlannerState.plan_status` 引入 `todo/in_progress/done/abandoned` 显式状态语义，`revise_current_plan` 与 `finish_plan` 进入 plan tool 契约。
+- `error_code`: 未新增跨服务 API `error_code` 表；工具失败路径统一返回可判定错误字符串（如 pending step 拒绝 `finish_plan(done)`、非法状态迁移）。
+- `retry`: 重试语义收敛为“仅对非终态 step/plan 允许继续推进”；终态（`done/abandoned`）拒绝回退重试。
+
+### Golden Cases
+- 新增/更新 golden 证据文件：`tests/unit/test_plan_v2_tools.py`。
+- 契约文档同步：`openspec/changes/agentscope-d7-plan-state-tools/specs/plan-runtime/spec.md`。
+- 提示策略同步：`openspec/changes/agentscope-d7-plan-state-tools/specs/chat-runtime/spec.md`。
+
+### Regression Summary
+- Runner commands:
+  - `pytest -q tests/unit/test_plan_v2_tools.py`
+  - `pytest -q tests/unit/test_plan_v2_tools.py tests/unit/test_react_agent_gateway_injection.py tests/unit/test_dare_agent_step_driven_mode.py`
+  - `pytest -q`
+- Summary: pass 518, fail 0, skip 12.
+- Warnings: 1 warning（与 D7 变更无新增失败关联）。
+
+### Observability and Failure Localization
+- 生命周期观测链覆盖 `start` / `tool_call` / `end` / `fail` 四类事件。
+- 失败定位字段要求保留：`run_id`, `tool_call_id`, `capability_id`, `attempt`, `error_code`, `trace_id`。
+- 重点定位点：`finish_plan` 失败分支（pending step 保护）、`transition_step/transition_plan` 非法迁移拒绝、sub-agent 调用后状态推进一致性。
+
+### Structured Review Report
+- Changed Module Boundaries / Public API: `plan_v2` 新增 `revise_current_plan`、`finish_plan` 两个公开工具，并扩展 `types` 状态机接口。
+- New State: 新增 `Step.status` 与 `PlannerState.plan_status` 状态字段；`completed_step_ids` 继续保留为兼容派生状态。
+- Concurrency / Timeout / Retry: 未新增并发执行模型；新增工具超时上限分别为 `30s`（revise）与 `20s`（finish）；重试仅允许在非终态路径。
+- Side Effects and Idempotency: 主要副作用是 plan 状态推进与 `critical_block` 文本更新；终态回退被拒绝以防重复副作用。
+- Coverage and Residual Risk: 覆盖状态迁移、finish/revise、critical_block 分支与回归路径；残余风险在于 legacy step 对象混用场景仍依赖兼容分支。
+
+### Behavior Verification
+- Happy path：
+  - `Planner` 暴露 `revise_current_plan`、`finish_plan` 工具；
+  - `revise_current_plan` 可修订计划并按 `step_id` 保留已完成步骤终态；
+  - `critical_block` 在步骤全部终态后提示下一步调用 `finish_plan(...)` 收敛计划。
+- Error branch：
+  - `finish_plan(target_state="done")` 在仍有 pending step 时返回失败；
+  - 状态机拒绝终态回退（如 `done -> in_progress`）。
+
+### Risks and Rollback
+- 风险：计划状态迁移规则若定义不严谨，可能导致执行循环重复或提前收敛。
+- 风险：`critical_block` 与状态不同步会导致提示误导。
+- 回滚：保留现有 plan_v2 工具路径与兼容字段（`completed_step_ids`），必要时回退新状态机入口。
+
+### Review and Merge Gate Links
+- Intent PR: `https://github.com/zts212653/Deterministic-Agent-Runtime-Engine/pull/138`
+- Implementation PR: `https://github.com/zts212653/Deterministic-Agent-Runtime-Engine/pull/138`
+- Review 请求：`https://github.com/zts212653/Deterministic-Agent-Runtime-Engine/pull/138#issuecomment-3982413008`
+- Merge Gate：待评审通过后补充。
diff --git a/docs/todos/agentscope_domain_execution_todos.md b/docs/todos/agentscope_domain_execution_todos.md
index b5e6b14b..09c2c6bd 100644
--- a/docs/todos/agentscope_domain_execution_todos.md
+++ b/docs/todos/agentscope_domain_execution_todos.md
@@ -21,10 +21,10 @@
 
 | Claim ID | TODO Scope | Owner | Status | Declared At | Expires At | OpenSpec Change | Notes |
 |---|---|---|---|---|---|---|---|
-| CLM-20260302-D2D4 | D2-1~D2-4, D4-1~D4-4 | mindfn | planned | 2026-03-02 | 2026-03-09 | `agentscope-d2-d4-thinking-transport` | 先处理 P0/P1 的 thinking 与 transport 协议统一。 |
-| CLM-20260302-D5 | D5-1~D5-4 | mindfn | planned | 2026-03-02 | 2026-03-09 | `agentscope-d5-safe-compression` | 压缩链路：tool pair safe + token-aware + auto trigger。 |
-| CLM-20260302-D7 | D7-1~D7-4 | mindfn | planned | 2026-03-02 | 2026-03-09 | `agentscope-d7-plan-state-tools` | plan 状态机与 finish/revise 原生工具补齐。 |
-| CLM-20260302-D1D3 | D1-1~D1-4, D3-1~D3-4 | mindfn | planned | 2026-03-02 | 2026-03-09 | `agentscope-d1-d3-message-pipeline` | 多模态输入 schema 与 assemble normalize。 |
+| CLM-20260302-D2D4 | D2-1~D2-4, D4-1~D4-4 | zts212653 | planned | 2026-03-02 | 2026-03-09 | `agentscope-d2-d4-thinking-transport` | 先处理 P0/P1 的 thinking 与 transport 协议统一。 |
+| CLM-20260302-D5 | D5-1~D5-4 | zts212653 | planned | 2026-03-02 | 2026-03-09 | `agentscope-d5-safe-compression` | 压缩链路：tool pair safe + token-aware + auto trigger。 |
+| CLM-20260302-D7 | D7-1~D7-4 | zts212653 | active | 2026-03-02 | 2026-03-09 | `agentscope-d7-plan-state-tools` | D7 实现与回归已完成，PR #138 待审。 |
+| CLM-20260302-D1D3 | D1-1~D1-4, D3-1~D3-4 | zts212653 | planned | 2026-03-02 | 2026-03-09 | `agentscope-d1-d3-message-pipeline` | 多模态输入 schema 与 assemble normalize。 |
 
 ---
 
@@ -220,10 +220,10 @@
 
 | ID | 任务 | 主要代码改动 | 支持能力 | 依赖 | 状态 | 输出证据 |
 |---|---|---|---|---|---|---|
-| D7-1 | 状态机 | Step/Plan 状态字段+迁移规则 | P1 | 无（独立） | todo | 非法迁移拒绝测试 |
-| D7-2 | finish/revise 工具 | 新 plan tools + schema | P2/P3 | D7-1 | todo | 示例可无 shim 调用 |
-| D7-3 | 状态提示联动 | critical_block 规则更新 | 执行引导一致性 | D7-1/2 | todo | plan 提示准确 |
-| D7-4 | 回归测试 | plan 全流程测试 | Plan 能力稳定 | D7-1/2/3 | todo | create/revise/finish 全通过 |
+| D7-1 | 状态机 | Step/Plan 状态字段+迁移规则 | P1 | 无（独立） | done | `tests/unit/test_plan_v2_tools.py::test_plan_state_transition_rules_reject_terminal_reopen` |
+| D7-2 | finish/revise 工具 | 新 plan tools + schema | P2/P3 | D7-1 | done | `tests/unit/test_plan_v2_tools.py::test_planner_exposes_revise_and_finish_plan_tools`，`tests/unit/test_plan_v2_tools.py::test_revise_current_plan_preserves_done_steps_by_step_id` |
+| D7-3 | 状态提示联动 | critical_block 规则更新 | 执行引导一致性 | D7-1/2 | done | `tests/unit/test_plan_v2_tools.py::test_critical_block_requires_finish_when_all_steps_done` |
+| D7-4 | 回归测试 | plan 全流程测试 | Plan 能力稳定 | D7-1/2/3 | done | `pytest -q tests/unit/test_plan_v2_tools.py tests/unit/test_react_agent_gateway_injection.py tests/unit/test_dare_agent_step_driven_mode.py`（31 passed）+ `pytest -q`（518 passed, 12 skipped, 1 warning） |
 
 ---
 
diff --git a/docs/todos/project_overall_todos.md b/docs/todos/project_overall_todos.md
index 71095156..63ecc602 100644
--- a/docs/todos/project_overall_todos.md
+++ b/docs/todos/project_overall_todos.md
@@ -15,10 +15,10 @@
 
 | Claim ID | TODO Scope | Owner | Status | Declared At | Expires At | OpenSpec Change | Notes |
 |---|---|---|---|---|---|---|---|
-| CLM-20260302-AG1 | T5-2 | mindfn | planned | 2026-03-02 | 2026-03-09 | `agentscope-d2-d4-thinking-transport` | 对齐 D2/D4：thinking + transport 事件链路。 |
-| CLM-20260302-AG2 | T2-1 | mindfn | planned | 2026-03-02 | 2026-03-09 | `agentscope-d5-safe-compression` | 对齐 D5：安全压缩与预算收敛。 |
-| CLM-20260302-AG3 | D7-1~D7-4（关联 T5-5） | mindfn | planned | 2026-03-02 | 2026-03-09 | `agentscope-d7-plan-state-tools` | 先按 AgentScope gap 切片推进 plan 状态机能力。 |
-| CLM-20260302-AG4 | T5-3 | mindfn | planned | 2026-03-02 | 2026-03-09 | `agentscope-d1-d3-message-pipeline` | 对齐 D1/D3：多模态输入 schema + normalize。 |
+| CLM-20260302-AG1 | T5-2 | zts212653 | planned | 2026-03-02 | 2026-03-09 | `agentscope-d2-d4-thinking-transport` | 对齐 D2/D4：thinking + transport 事件链路。 |
+| CLM-20260302-AG2 | T2-1 | zts212653 | planned | 2026-03-02 | 2026-03-09 | `agentscope-d5-safe-compression` | 对齐 D5：安全压缩与预算收敛。 |
+| CLM-20260302-AG3 | D7-1~D7-4（关联 T5-5） | zts212653 | active | 2026-03-02 | 2026-03-09 | `agentscope-d7-plan-state-tools` | D7 实现与回归已完成，PR #138 待审。 |
+| CLM-20260302-AG4 | T5-3 | zts212653 | planned | 2026-03-02 | 2026-03-09 | `agentscope-d1-d3-message-pipeline` | 对齐 D1/D3：多模态输入 schema + normalize。 |
 
 ## 2. 当前基线
 
diff --git a/openspec/changes/agentscope-d7-plan-state-tools/.openspec.yaml b/openspec/changes/agentscope-d7-plan-state-tools/.openspec.yaml
new file mode 100644
index 00000000..fd79bfc5
--- /dev/null
+++ b/openspec/changes/agentscope-d7-plan-state-tools/.openspec.yaml
@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-03-02
diff --git a/openspec/changes/agentscope-d7-plan-state-tools/design.md b/openspec/changes/agentscope-d7-plan-state-tools/design.md
new file mode 100644
index 00000000..bd2b3074
--- /dev/null
+++ b/openspec/changes/agentscope-d7-plan-state-tools/design.md
@@ -0,0 +1,64 @@
+## Context
+
+`plan_v2` 当前提供 `create_plan/validate_plan/...` 等工具，但关键状态依赖 `completed_step_ids` 隐式推导。该方式难以覆盖 AgentScope PlanNoteBook 语义中对 `in_progress`、`abandoned`、显式结束和修订的要求，也让 `critical_block` 的下一步提示缺乏严格状态约束。
+
+## Goals / Non-Goals
+
+**Goals:**
+- 引入显式 step/plan 状态机并定义合法迁移。
+- 新增 `revise_current_plan` 与 `finish_plan` 工具以补齐 D7 功能缺口。
+- 让 `critical_block` 严格反映最新计划状态与下一步建议。
+- 保留现有 `completed_step_ids` 兼容字段，避免已有调用链断裂。
+
+**Non-Goals:**
+- 本次不实现历史计划恢复（`view_historical_plans/recover_historical_plan`）。
+- 本次不引入 plan change hook 回调总线。
+- 本次不实现 Session 持久化（`state_dict/load_state_dict` 留在后续切片）。
+
+## Decisions
+
+### Decision 1: 状态机放在 plan_v2 types，工具层只消费
+- 在 `types.py` 定义状态常量、合法迁移规则和迁移辅助函数。
+- `tools.py` 只通过状态机 API 执行状态变更，非法迁移直接返回错误。
+- 理由：避免状态语义散落在多个工具实现里。
+
+### Decision 2: `revise_current_plan` 采用“按 step_id 合并状态”策略
+- 修订时允许替换步骤列表，但对同 `step_id` 且已完成步骤保留完成状态。
+- 已不存在的新列表步骤会被移除，不再参与 pending。
+- 理由：最小化用户修订造成的已完成进度丢失。
+
+### Decision 3: `finish_plan(done)` 必须校验剩余未完成步骤
+- 若存在 `todo/in_progress` 步骤，`finish_plan(target_state=\"done\")` 失败。
+- `finish_plan(target_state=\"abandoned\")` 会把未终态步骤统一置为 `abandoned`。
+- 理由：防止“伪完成”状态污染审计与后续流程。
+
+### Decision 4: `critical_block` 以 plan/step 状态为唯一真相
+- 提示块明确展示 `plan_status`、每步状态、pending/completed 列表与下一动作。
+- 当全部步骤完成但 plan 未终态时，提示下一步调用 `finish_plan`。
+- 理由：降低 agent 提示和实际状态脱节风险。
+
+## Risks / Trade-offs
+
+- [Risk] 增加状态字段后，旧路径若直接写 `completed_step_ids` 可能出现双写不一致。  
+  → Mitigation: 提供统一同步函数，把 completed set 从 step 状态派生回写。
+- [Risk] `revise_current_plan` 合并规则过宽可能保留错误历史状态。  
+  → Mitigation: 仅保留 `done/abandoned` 终态，其它状态回退为 `todo`。
+- [Risk] 提示变更会影响 plan agent 行为轨迹。  
+  → Mitigation: 更新 prompts 并用单测覆盖 `critical_block` 关键文案分支。
+
+## Migration Plan
+
+1. 在 `types.py` 引入状态机数据结构与合法迁移检查。
+2. 增强 `tools.py`（`create/validate/sub_agent` 状态联动）并新增 `revise_current_plan`、`finish_plan`。
+3. 更新 `Planner` 与 `__init__` 暴露工具集合，补 prompt 指导语。
+4. 编写 D7 单测并执行定向 + 全量回归。
+5. 回写 TODO/OpenSpec/feature evidence，并进入 PR 评审。
+
+Rollback:
+- 回退新增状态机字段与两项新工具，恢复 `completed_step_ids` 旧逻辑驱动。
+
+## Open Questions
+
+- 后续是否需要新增 `failed` 步骤状态以区分可重试失败与主动放弃？
+- `revise_current_plan` 是否需要支持“只补丁变更单个 step”而非全量替换？
+
diff --git a/openspec/changes/agentscope-d7-plan-state-tools/proposal.md b/openspec/changes/agentscope-d7-plan-state-tools/proposal.md
new file mode 100644
index 00000000..7e187e32
--- /dev/null
+++ b/openspec/changes/agentscope-d7-plan-state-tools/proposal.md
@@ -0,0 +1,34 @@
+## Why
+
+AgentScope 兼容差距中，PlanNoteBook 相关能力仍缺关键闭环：步骤与计划状态机不完整、计划不可修订、计划无法显式完成/放弃。当前 `plan_v2` 仅依赖 `completed_step_ids` 追踪进度，难以表达 `in_progress/abandoned` 等中间态，导致执行提示与真实状态可能漂移。
+
+## What Changes
+
+- 为 `plan_v2` 补齐 `todo/in_progress/done/abandoned` 状态机（step + plan）。
+- 新增 `revise_current_plan` 工具，支持计划创建后的结构化修订。
+- 新增 `finish_plan` 工具，支持显式完成/放弃并执行状态约束校验。
+- 将 `critical_block` 与状态机联动，确保下一步提示由真实状态驱动。
+- 增加 D7 单测矩阵（合法/非法迁移、finish/revise、critical block、回归兼容）。
+
+## Capabilities
+
+### New Capabilities
+- `agentscope-plan-state-machine`: 面向 plan_v2 的标准状态迁移与终态收敛。
+- `agentscope-plan-revision-tools`: 计划运行时修订与显式结束工具能力。
+
+### Modified Capabilities
+- `chat-runtime`: plan `critical_block` 的状态提示由新状态机驱动，减少执行引导偏差。
+
+## Impact
+
+- Affected code:
+  - `dare_framework/plan_v2/types.py`
+  - `dare_framework/plan_v2/tools.py`
+  - `dare_framework/plan_v2/planner.py`
+  - `dare_framework/plan_v2/__init__.py`
+  - `dare_framework/plan_v2/prompts.py`
+  - `tests/unit/test_plan_v2_tools.py` (new)
+- Runtime/API impact:
+  - 新增计划工具（`revise_current_plan`、`finish_plan`）并扩展状态字段。
+  - 计划执行提示从“已完成集合”升级为“显式状态机”。
+
diff --git a/openspec/changes/agentscope-d7-plan-state-tools/specs/chat-runtime/spec.md b/openspec/changes/agentscope-d7-plan-state-tools/specs/chat-runtime/spec.md
new file mode 100644
index 00000000..8624d1b5
--- /dev/null
+++ b/openspec/changes/agentscope-d7-plan-state-tools/specs/chat-runtime/spec.md
@@ -0,0 +1,20 @@
+## MODIFIED Requirements
+
+### Requirement: React plan hint injection MUST reflect canonical plan state
+When a `plan_provider` is mounted on `ReactAgent`, injected `critical_block` MUST be derived from canonical plan/step lifecycle state, not only from completed id sets.
+
+The hint MUST include:
+- current `plan_status`
+- pending/completed step view
+- deterministic NEXT action guidance (`validate_plan`, `sub_agent_*`, or `finish_plan`)
+
+#### Scenario: prompt asks to finish when all steps are done
+- **GIVEN** all steps are in `done` and plan state is not terminal
+- **WHEN** `critical_block` is generated
+- **THEN** NEXT guidance asks to call `finish_plan(target_state="done")`
+
+#### Scenario: prompt asks to validate before execution
+- **GIVEN** plan exists but is not validated
+- **WHEN** `critical_block` is generated
+- **THEN** NEXT guidance asks to call `validate_plan(success=True)`
+
diff --git a/openspec/changes/agentscope-d7-plan-state-tools/specs/plan-runtime/spec.md b/openspec/changes/agentscope-d7-plan-state-tools/specs/plan-runtime/spec.md
new file mode 100644
index 00000000..5f133b91
--- /dev/null
+++ b/openspec/changes/agentscope-d7-plan-state-tools/specs/plan-runtime/spec.md
@@ -0,0 +1,38 @@
+## ADDED Requirements
+
+### Requirement: plan_v2 Step/Plan MUST use explicit lifecycle states
+`plan_v2` SHALL expose explicit lifecycle states for both step and plan using `todo`, `in_progress`, `done`, `abandoned`.
+
+`step` and `plan` state transitions MUST follow defined legal transitions, and illegal transitions MUST be rejected with deterministic errors.
+
+#### Scenario: valid step transition
+- **GIVEN** a step in `todo`
+- **WHEN** runtime transitions it to `in_progress` then `done`
+- **THEN** both transitions succeed
+- **AND** the step final state is `done`
+
+#### Scenario: invalid terminal transition
+- **GIVEN** a step in `done`
+- **WHEN** runtime tries to transition it back to `in_progress`
+- **THEN** transition is rejected
+- **AND** a deterministic error is returned
+
+### Requirement: plan_v2 MUST support runtime plan revision and explicit finish
+`plan_v2` SHALL provide tools `revise_current_plan` and `finish_plan`.
+
+- `revise_current_plan` MUST allow modifying current plan while preserving valid completed progress by `step_id`.
+- `finish_plan(target_state="done")` MUST fail if non-terminal steps remain.
+- `finish_plan(target_state="abandoned")` MUST mark remaining non-terminal steps as `abandoned`.
+
+#### Scenario: revise existing plan
+- **GIVEN** an existing validated plan in progress
+- **WHEN** `revise_current_plan` is called with updated steps
+- **THEN** plan description and steps are updated
+- **AND** completed step progress remains consistent where step_id matches
+
+#### Scenario: reject premature done
+- **GIVEN** a plan with pending steps
+- **WHEN** `finish_plan(target_state="done")` is called
+- **THEN** tool returns failure
+- **AND** plan state does not move to `done`
+
diff --git a/openspec/changes/agentscope-d7-plan-state-tools/tasks.md b/openspec/changes/agentscope-d7-plan-state-tools/tasks.md
new file mode 100644
index 00000000..58f30f38
--- /dev/null
+++ b/openspec/changes/agentscope-d7-plan-state-tools/tasks.md
@@ -0,0 +1,24 @@
+## 1. State Machine Core
+
+- [x] 1.1 在 `plan_v2.types` 定义 step/plan 状态枚举与合法迁移规则。
+- [x] 1.2 增加状态同步辅助逻辑，确保 `completed_step_ids` 与 step 状态一致。
+- [x] 1.3 补充非法迁移拒绝路径（结构化错误）。
+
+## 2. Plan Tooling
+
+- [x] 2.1 新增 `revise_current_plan` 工具并接入 `Planner`。
+- [x] 2.2 新增 `finish_plan` 工具并实现 `done/abandoned` 终态约束。
+- [x] 2.3 更新 `create_plan/validate_plan/sub_agent` 的状态推进逻辑。
+
+## 3. Critical Block and Prompt Alignment
+
+- [x] 3.1 更新 `_format_critical_block`：展示 `plan_status` + step 状态 + NEXT 指令。
+- [x] 3.2 更新 plan prompt 文案，覆盖 revise/finish 的操作路径。
+- [x] 3.3 验证 `critical_block` 在 create/validate/revise/finish 关键节点一致性。
+
+## 4. Verification and Evidence
+
+- [x] 4.1 新增 `tests/unit/test_plan_v2_tools.py` 覆盖状态机与新工具矩阵。
+- [x] 4.2 回归测试：定向 plan_v2 + ReactAgent 注入路径。
+- [x] 4.3 全量回归：`pytest -q` 并回写 feature evidence。
+- [x] 4.4 回写 TODO claim 与 OpenSpec 状态，准备 PR 审查。
diff --git a/tests/unit/test_plan_v2_tools.py b/tests/unit/test_plan_v2_tools.py
new file mode 100644
index 00000000..ce3d3512
--- /dev/null
+++ b/tests/unit/test_plan_v2_tools.py
@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+import pytest
+
+from dare_framework.plan_v2.planner import Planner
+from dare_framework.plan_v2.tools import (
+    CreatePlanTool,
+    FinishPlanTool,
+    ReviseCurrentPlanTool,
+    ValidatePlanTool,
+)
+from dare_framework.plan_v2.types import PlannerState, Step, is_valid_state_transition
+from dare_framework.tool.types import RunContext
+
+
+def _run_context() -> RunContext[None]:
+    return RunContext()
+
+
+@pytest.mark.asyncio
+async def test_planner_exposes_revise_and_finish_plan_tools() -> None:
+    planner = Planner(state=PlannerState())
+    tool_names = {tool.name for tool in planner.list_tools()}
+    assert "revise_current_plan" in tool_names
+    assert "finish_plan" in tool_names
+
+
+def test_plan_state_transition_rules_reject_terminal_reopen() -> None:
+    assert is_valid_state_transition("todo", "in_progress") is True
+    assert is_valid_state_transition("in_progress", "done") is True
+    assert is_valid_state_transition("done", "in_progress") is False
+    assert is_valid_state_transition("abandoned", "todo") is False
+
+
+def test_sync_completed_step_ids_tolerates_legacy_steps_without_status() -> None:
+    class _LegacyStep:
+        def __init__(self, step_id: str) -> None:
+            self.step_id = step_id
+
+    state = PlannerState(
+        steps=[
+            Step(step_id="s1", description="done step", status="done"),
+            _LegacyStep(step_id="s_legacy"),
+            {"step_id": "s_dict", "status": "done"},
+        ]
+    )
+
+    state.sync_completed_step_ids()
+
+    assert state.completed_step_ids == {"s1", "s_dict"}
+
+
+def test_transition_step_tolerates_legacy_step_without_status() -> None:
+    class _LegacyStep:
+        def __init__(self, step_id: str, description: str) -> None:
+            self.step_id = step_id
+            self.description = description
+            self.params = {}
+
+    state = PlannerState(steps=[_LegacyStep(step_id="s_legacy", description="legacy step")])
+
+    state.transition_step("s_legacy", "in_progress")
+
+    assert getattr(state.steps[0], "status") == "in_progress"
+
+
+@pytest.mark.asyncio
+async def test_finish_plan_rejects_done_when_pending_steps_exist() -> None:
+    state = PlannerState()
+    create_tool = CreatePlanTool(state)
+    validate_tool = ValidatePlanTool(state)
+    finish_tool = FinishPlanTool(state)
+
+    await create_tool.execute(
+        run_context=_run_context(),
+        plan_description="d7-finish-guard",
+        steps=[
+            {"step_id": "s1", "description": "first"},
+            {"step_id": "s2", "description": "second"},
+        ],
+    )
+    await validate_tool.execute(run_context=_run_context(), success=True)
+
+    result = await finish_tool.execute(run_context=_run_context(), target_state="done")
+
+    assert result.success is False
+    assert state.plan_status != "done"
+    assert isinstance(result.error, str)
+    assert "pending" in result.error.lower()
+
+
+@pytest.mark.asyncio
+async def test_revise_current_plan_preserves_done_steps_by_step_id() -> None:
+    state = PlannerState()
+    create_tool = CreatePlanTool(state)
+    validate_tool = ValidatePlanTool(state)
+    revise_tool = ReviseCurrentPlanTool(state)
+
+    await create_tool.execute(
+        run_context=_run_context(),
+        plan_description="initial",
+        steps=[
+            {"step_id": "s1", "description": "first"},
+            {"step_id": "s2", "description": "second"},
+        ],
+    )
+    await validate_tool.execute(run_context=_run_context(), success=True)
+    state.steps[0].status = "done"
+    state.completed_step_ids.add("s1")
+
+    result = await revise_tool.execute(
+        run_context=_run_context(),
+        plan_description="revised",
+        steps=[
+            {"step_id": "s1", "description": "first revised"},
+            {"step_id": "s3", "description": "third"},
+        ],
+    )
+
+    assert result.success is True
+    assert state.plan_description == "revised"
+    assert [step.step_id for step in state.steps] == ["s1", "s3"]
+    assert state.steps[0].status == "done"
+    assert state.steps[1].status == "todo"
+
+
+@pytest.mark.asyncio
+async def test_critical_block_requires_finish_when_all_steps_done() -> None:
+    state = PlannerState()
+    create_tool = CreatePlanTool(state)
+    validate_tool = ValidatePlanTool(state)
+
+    await create_tool.execute(
+        run_context=_run_context(),
+        plan_description="critical-block-finish",
+        steps=[{"step_id": "s1", "description": "only step"}],
+    )
+    await validate_tool.execute(run_context=_run_context(), success=True)
+
+    state.steps[0].status = "done"
+    state.completed_step_ids.add("s1")
+    await validate_tool.execute(run_context=_run_context(), success=True)
+
+    assert "finish_plan" in state.critical_block