fix(runner): enforce explicit mission completion

agent/agent/cli/interactive.py

@@ -435,6 +435,7 @@ class InteractiveController:
 
             # 调用 runner.run() 续跑
             print("\n开始执行...")
+            final_status = None
             async for event in self.runner.run(
                 messages=messages_to_add,
                 config=config,
@@ -451,10 +452,18 @@ class InteractiveController:
                             print(f"\n🤖 Assistant: {text[:200]}...")
                 elif isinstance(event, Trace):
                     # Trace 状态更新
-                    if event.status in ("completed", "failed", "stopped"):
+                    if event.status in ("completed", "incomplete", "failed", "stopped"):
+                        final_status = event.status
                         print(f"\n📊 Trace 状态: {event.status}")
 
-            print("\n✅ 执行完成")
+            if final_status == "completed":
+                print("\n✅ 执行完成")
+            elif final_status == "incomplete":
+                print("\n⚠️ 执行尚未完成，可继续恢复运行")
+            elif final_status:
+                print(f"\n❌ 执行结束：{final_status}")
+            else:
+                print("\n⚠️ 执行结束，但未收到终态")
 
         except Exception as e:
             print(f"❌ 执行失败: {e}")

agent/agent/client.py

@@ -29,6 +29,7 @@ import importlib.util
 import logging
 import os
 import sys
+from dataclasses import replace
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
@@ -60,6 +61,7 @@ async def invoke_agent(
     continue_from: Optional[str] = None,
     messages: Optional[List[Dict[str, Any]]] = None,
     project_root: Optional[str] = None,
+    root_task_spec: Optional[Dict[str, Any]] = None,
 ) -> Dict[str, Any]:
     """
     统一调用远端或本地 Agent。
@@ -71,11 +73,19 @@ async def invoke_agent(
         continue_from: 已有 sub_trace_id，传入则续跑
         messages: 预置 OpenAI 格式消息（远端 1D、本地 1D）
         project_root: 本地 agent 必填——项目目录（含 config.py / presets.json / tools/）
+        root_task_spec: 本地 explicit-validation Planner 的 Root TaskSpec；远端不支持
 
     Returns:
         {"mode", "agent_type", "sub_trace_id", "status", "summary", "stats", "error"?}
     """
     if agent_type.startswith("remote_"):
+        if root_task_spec is not None:
+            return {
+                "mode": "remote",
+                "agent_type": agent_type,
+                "status": "failed",
+                "error": "remote agent 不支持 root_task_spec；该参数仅用于本地 explicit-validation Planner",
+            }
         # 懒 import，避免加载整个 tool registry（远端调用只需要 httpx）
         from agent.tools.builtin.subagent import _run_remote_agent
         return await _run_remote_agent(
@@ -101,6 +111,7 @@ async def invoke_agent(
         continue_from=continue_from,
         messages=messages,
         project_root=project_root,
+        root_task_spec=root_task_spec,
     )
 
 
@@ -111,6 +122,7 @@ async def _run_local_agent(
     continue_from: Optional[str],
     messages: Optional[List[Dict[str, Any]]],
     project_root: str,
+    root_task_spec: Optional[Dict[str, Any]] = None,
 ) -> Dict[str, Any]:
     """
     在当前进程中起 AgentRunner 跑本地 agent。
@@ -167,6 +179,7 @@ async def _run_local_agent(
     if run_config is None:
         return {"mode": "local", "agent_type": agent_type, "status": "failed",
                 "error": f"{config_path} 未定义 RUN_CONFIG"}
+    run_config = replace(run_config)
 
     # 覆盖 agent_type / skills / continue_from
     run_config.agent_type = agent_type
@@ -174,6 +187,8 @@ async def _run_local_agent(
         run_config.skills = skills
     if continue_from:
         run_config.trace_id = continue_from
+    if root_task_spec is not None:
+        run_config.root_task_spec = root_task_spec
 
     # 4. 加载项目 presets.json（如果有）
     presets_path = root / "presets.json"

agent/agent/core/runner.py

@@ -128,6 +128,7 @@ class RunConfig:
     enable_prompt_caching: bool = True         # 启用 Anthropic Prompt Caching（仅 Claude 模型有效）
     parallel_tool_execution: bool = False      # 是否启用并发 Tool Call 执行（慎用，需确保无资源冲突）
     completion_policy: CompletionPolicy = CompletionPolicy.LEGACY_AUTO
+    root_task_spec: Optional[Dict[str, Any]] = None  # explicit Planner 新 Trace 的 Root Task 草案
 
     # --- Trace 控制 ---
     trace_id: Optional[str] = None             # None = 新建
@@ -300,12 +301,21 @@ class AgentRunner:
         config = config or RunConfig()
         trace = None
 
-        resolved_policy = self._resolve_run_policy(config)
+        resume_trace = await self._restore_resume_identity(config)
+        resolved_policy = self._resolve_run_policy(config, resume_trace)
+        if resume_trace:
+            self.validate_resume_position(resume_trace, config.after_sequence)
         if resolved_policy.completion_policy == CompletionPolicy.EXPLICIT_VALIDATION:
             if self.task_coordinator is None:
                 raise RuntimeError(
                     "explicit_validation requires a TaskCoordinator; call wire_orchestration() first"
                 )
+            if resume_trace and resolved_policy.role == AgentRole.PLANNER:
+                await self.task_coordinator.ensure_ledger(
+                    resume_trace.trace_id,
+                    None,
+                    allow_create=False,
+                )
             config.parallel_tool_execution = False
             config.max_iterations = resolved_policy.max_iterations
             config.temperature = resolved_policy.temperature
@@ -316,11 +326,16 @@ class AgentRunner:
         try:
             # Phase 1: PREPARE TRACE
             trace, goal_tree, sequence = await self._prepare_trace(messages, config)
+            resolved_policy = self._resolve_run_policy(config, trace)
             if (
                 resolved_policy.completion_policy == CompletionPolicy.EXPLICIT_VALIDATION
                 and resolved_policy.role == AgentRole.PLANNER
             ):
-                await self.task_coordinator.ensure_ledger(trace.trace_id, trace.task or "")
+                await self.task_coordinator.ensure_ledger(
+                    trace.trace_id,
+                    config.root_task_spec,
+                    allow_create=config.trace_id is None,
+                )
             # 注册取消事件
             self._cancel_events[trace.trace_id] = asyncio.Event()
             yield trace
@@ -423,7 +438,7 @@ class AgentRunner:
         error = final_trace.error_message if final_trace else None
         summary = last_assistant_text or (final_trace.result_summary if final_trace else "")
 
-        if not summary and status != "completed":
+        if not summary and status not in ("completed", "incomplete", "stopped"):
             status = "failed"
             error = error or "Agent 没有产生 assistant 文本结果"
 
@@ -1077,6 +1092,13 @@ class AgentRunner:
         trace_id = trace.trace_id
         resolved_policy = self._resolve_run_policy(config, trace)
         tool_schemas = self._get_run_tool_schemas(config, resolved_policy)
+        explicit_role = (
+            resolved_policy.role
+            if resolved_policy.completion_policy == CompletionPolicy.EXPLICIT_VALIDATION
+            else None
+        )
+        explicit_terminal_submitted = False
+        terminal_summary: Optional[str] = None
 
         # 当前主路径头节点的 sequence（用于设置 parent_sequence）
         head_seq = trace.head_sequence
@@ -1663,7 +1685,7 @@ class AgentRunner:
                 continue
 
             terminal_run = False
-            terminal_summary: Optional[str] = None
+            terminal_summary = None
 
             if tool_calls and config.auto_execute_tools:
                 history.append({
@@ -1989,6 +2011,8 @@ class AgentRunner:
                             break
 
                 if terminal_run:
+                    if explicit_role in (AgentRole.WORKER, AgentRole.VALIDATOR):
+                        explicit_terminal_submitted = True
                     break
 
                 # on_complete 模式：goal(done=...) 后立即压缩该 goal 的消息
@@ -2029,6 +2053,30 @@ class AgentRunner:
             # 如果在侧分支中，已经在上面处理过了（不会走到这里）
             # 主路径无工具调用 → 任务完成，检查是否需要完成后反思或知识评估
 
+            if explicit_role == AgentRole.PLANNER:
+                mission = await self._get_mission_completion(trace_id)
+                mission_status = self._status_value(mission.get("status"))
+                if mission_status in ("completed", "blocked"):
+                    break
+
+                control_msg = Message.create(
+                    trace_id=trace_id,
+                    role="user",
+                    sequence=sequence,
+                    parent_sequence=head_seq if head_seq > 0 else None,
+                    content=self._build_mission_control_message(mission),
+                )
+                if self.trace_store:
+                    await self.trace_store.add_message(control_msg)
+                yield control_msg
+                history.append(control_msg.to_llm_dict())
+                head_seq = sequence
+                sequence += 1
+                continue
+
+            if explicit_role in (AgentRole.WORKER, AgentRole.VALIDATOR):
+                break
+
             # 检查是否有待评估的知识
             if not side_branch_ctx and self.trace_store:
                 pending = await self.trace_store.get_pending_knowledge_entries(trace_id)
@@ -2054,18 +2102,111 @@ class AgentRunner:
         self._context_usage.pop(trace_id, None)
         self._saved_knowledge_ids.pop(trace_id, None)
 
-        # 更新 head_sequence 并完成 Trace
+        final_status = "completed"
+        final_error: Optional[str] = None
+        final_summary: Optional[str] = terminal_summary
+
+        if explicit_role == AgentRole.PLANNER:
+            mission = await self._get_mission_completion(trace_id)
+            mission_status = self._status_value(mission.get("status"))
+            if mission_status == "completed":
+                final_summary = mission.get("result_summary") or final_summary
+            elif mission_status == "blocked":
+                final_status = "incomplete"
+                final_error = "mission_blocked"
+                blocked_reason = mission.get("blocked_reason")
+                if blocked_reason:
+                    self.log.warning("Mission %s blocked: %s", trace_id, blocked_reason)
+            else:
+                final_status = "incomplete"
+                final_error = "iteration_limit_reached"
+        elif explicit_role in (AgentRole.WORKER, AgentRole.VALIDATOR):
+            if not explicit_terminal_submitted:
+                final_status = "failed"
+                final_error = "protocol_violation"
+
+        # 更新 head_sequence 和 Trace 终态
         if self.trace_store:
-            await self.trace_store.update_trace(
-                trace_id,
-                status="completed",
-                head_sequence=head_seq,
-                completed_at=datetime.now(),
-            )
+            updates: Dict[str, Any] = {
+                "status": final_status,
+                "head_sequence": head_seq,
+                "error_message": final_error,
+                "completed_at": datetime.now(),
+            }
+            if final_summary:
+                updates["result_summary"] = final_summary
+            await self.trace_store.update_trace(trace_id, **updates)
             trace_obj = await self.trace_store.get_trace(trace_id)
             if trace_obj:
                 yield trace_obj
 
+    async def _get_mission_completion(self, root_trace_id: str) -> Dict[str, Any]:
+        """Read the coordinator-owned mission completion snapshot."""
+        method = getattr(self.task_coordinator, "mission_completion", None)
+        if method is None:
+            raise RuntimeError("TaskCoordinator does not provide mission_completion()")
+        snapshot = await method(root_trace_id)
+        if not isinstance(snapshot, dict) or "status" not in snapshot:
+            raise RuntimeError("mission_completion() returned an invalid snapshot")
+        return snapshot
+
+    async def _restore_resume_identity(self, config: RunConfig) -> Optional[Trace]:
+        """Restore durable role identity before any existing Trace is mutated."""
+
+        if not config.trace_id:
+            return None
+        if not self.trace_store:
+            raise ValueError("trace_store required for continue/rewind")
+        trace = await self.trace_store.get_trace(config.trace_id)
+        if trace is None:
+            raise ValueError(f"Trace not found: {config.trace_id}")
+        stored_role = AgentRole(trace.agent_role)
+        requested_policy = CompletionPolicy(config.completion_policy)
+        if stored_role == AgentRole.LEGACY:
+            if requested_policy == CompletionPolicy.EXPLICIT_VALIDATION:
+                raise ValueError("A legacy Trace cannot resume as explicit_validation")
+            return trace
+        config.completion_policy = CompletionPolicy.EXPLICIT_VALIDATION
+        if config.agent_type == "default":
+            config.agent_type = trace.agent_type or stored_role.value
+        return trace
+
+    @staticmethod
+    def _status_value(status: Any) -> str:
+        value = getattr(status, "value", status)
+        return str(value).lower() if value is not None else ""
+
+    @staticmethod
+    def validate_resume_position(
+        trace: Trace,
+        after_sequence: Optional[int],
+    ) -> None:
+        """Reject rewind for durable explicit Planner mission state."""
+        if (
+            trace.agent_role == AgentRole.PLANNER.value
+            and after_sequence is not None
+            and after_sequence < max(trace.head_sequence, trace.last_sequence)
+        ):
+            raise ValueError(
+                "explicit-validation Planner traces do not support rewind"
+            )
+
+    @staticmethod
+    def _build_mission_control_message(snapshot: Dict[str, Any]) -> str:
+        """Build a compact, framework-owned continuation instruction."""
+        state = {
+            "root_task_id": snapshot.get("root_task_id"),
+            "root_status": AgentRunner._status_value(snapshot.get("status")),
+            "pending_tasks": snapshot.get("pending_tasks") or [],
+        }
+        return (
+            "[FRAMEWORK_MISSION_INCOMPLETE]\n"
+            "The root mission is not complete. Continue the orchestration loop: "
+            "inspect the task ledger, resolve pending work, and accept the root task "
+            "only after its current version has passed validation.\n"
+            f"Mission state: {json.dumps(state, ensure_ascii=False, default=str)}"
+        )
+
     # ===== 压缩辅助方法 =====
 
     def _rebuild_history_after_compression(
@@ -3052,15 +3193,15 @@ class AgentRunner:
         """Resolve role and effective tools from the preset, never caller context."""
         preset = get_preset(config.agent_type)
         policy = self.tool_policy.resolve(config, preset, self.tools)
+        if trace and trace.agent_role != policy.role.value:
+            raise ValueError(
+                f"Trace role mismatch: stored={trace.agent_role}, requested={policy.role.value}"
+            )
         if policy.completion_policy == CompletionPolicy.EXPLICIT_VALIDATION:
             if policy.role == AgentRole.LEGACY:
                 raise ValueError(
                     f"Preset '{config.agent_type}' has legacy role and cannot run explicit_validation"
                 )
-            if trace and trace.agent_role not in (policy.role.value, "legacy"):
-                raise ValueError(
-                    f"Trace role mismatch: stored={trace.agent_role}, requested={policy.role.value}"
-                )
         return policy
 
     def _get_run_tool_schemas(

agent/agent/trace/models.py

@@ -52,7 +52,7 @@ class Trace:
     parent_goal_id: Optional[str] = None      # 哪个 Goal 启动的
 
     # 状态
-    status: Literal["running", "completed", "failed", "stopped"] = "running"
+    status: Literal["running", "completed", "incomplete", "failed", "stopped"] = "running"
 
     # 统计
     total_messages: int = 0      # 消息总数（改名自 total_steps）

agent/agent/trace/run_api.py

@@ -404,6 +404,14 @@ async def run_trace(trace_id: str, req: TraceRunRequest):
         if not trace:
             raise HTTPException(status_code=404, detail=f"Trace not found: {trace_id}")
 
+        try:
+            runner.validate_resume_position(trace, after_sequence)
+        except ValueError as exc:
+            raise HTTPException(
+                status_code=409,
+                detail=str(exc),
+            ) from exc
+
         # 自动检查并清理不完整的工具调用
         if after_sequence is not None and req.messages:
             adjusted_seq = await _cleanup_incomplete_tool_calls(
@@ -423,7 +431,7 @@ async def run_trace(trace_id: str, req: TraceRunRequest):
         store_trace = None
         if runner.trace_store:
             store_trace = await runner.trace_store.get_trace(trace_id)
-        if store_trace and store_trace.status in ("stopped", "failed", "completed"):
+        if store_trace and store_trace.status in ("stopped", "failed", "incomplete", "completed"):
             logger.info(
                 f"run_trace: task for {trace_id} not done yet but store status={store_trace.status!r}, "
                 "forcing cleanup to allow resume"