框架：增加可恢复的无进展熔断

agent/agent/__init__.py

@@ -11,7 +11,7 @@ Reson Agent - 模块化、可扩展的 Agent 框架
 """
 
 # 核心引擎
-from agent.core.runner import AgentRunner, CallResult, RunConfig
+from agent.core.runner import AgentRunner, CallResult, NoProgressPolicy, RunConfig
 from agent.core.presets import AgentPreset, AGENT_PRESETS, get_preset
 
 # 执行追踪
@@ -77,6 +77,7 @@ __all__ = [
     "AgentRunner",
     "CallResult",
     "RunConfig",
+    "NoProgressPolicy",
     "AgentPreset",
     "AGENT_PRESETS",
     "get_preset",

agent/agent/core/__init__.py

@@ -7,7 +7,7 @@ Agent Core - 核心引擎模块
 3. Agent 预设（AgentPreset）
 """
 
-from agent.core.runner import AgentRunner, CallResult, RunConfig
+from agent.core.runner import AgentRunner, CallResult, NoProgressPolicy, RunConfig
 from agent.core.presets import (
     AgentPreset,
     AGENT_PRESETS,
@@ -21,6 +21,7 @@ __all__ = [
     "AgentRunner",
     "CallResult",
     "RunConfig",
+    "NoProgressPolicy",
     "AgentPreset",
     "AGENT_PRESETS",
     "get_preset",

agent/agent/core/runner.py

@@ -68,6 +68,21 @@ _FAILURE_DISPOSITION_PRIORITY = {
 }
 
 
+@dataclass(frozen=True)
+class NoProgressPolicy:
+    """Fail closed when an explicit Agent repeats work without semantic progress."""
+
+    same_failure_limit: int = 2
+    planner_stall_limit: int = 3
+    enabled_in_legacy: bool = False
+
+    def __post_init__(self) -> None:
+        if self.same_failure_limit < 1:
+            raise ValueError("same_failure_limit must be positive")
+        if self.planner_stall_limit < 1:
+            raise ValueError("planner_stall_limit must be positive")
+
+
 @dataclass
 class ContextUsage:
     """Context 使用情况"""
@@ -157,6 +172,7 @@ class RunConfig:
     root_task_spec: Optional[Dict[str, Any]] = (
         None  # explicit Planner 新 Trace 的 Root Task 草案
     )
+    no_progress: NoProgressPolicy = field(default_factory=NoProgressPolicy)
 
     # --- Trace 控制 ---
     trace_id: Optional[str] = None  # None = 新建
@@ -1304,6 +1320,12 @@ class AgentRunner:
                         yield trace_obj
                 return
 
+            if explicit_role == AgentRole.PLANNER and not side_branch_ctx:
+                stalled = await self._observe_planner_progress(trace, config)
+                if stalled is not None:
+                    terminal_failure = stalled
+                    break
+
             # 检查Goal完成触发的知识评估
             if not side_branch_ctx and self.trace_store:
                 trace = await self.trace_store.get_trace(trace_id)
@@ -2178,6 +2200,16 @@ class AgentRunner:
                             last_failure = self._stronger_failure(
                                 last_failure, observed_failure
                             )
+                            repeated = await self._observe_failure_progress(
+                                trace,
+                                config,
+                                explicit_role,
+                                observed_failure,
+                            )
+                            if repeated is not None:
+                                terminal_failure = repeated
+                                terminal_run = True
+                                continue
                             if self._failure_terminates_role(
                                 observed_failure, explicit_role
                             ):
@@ -2185,11 +2217,18 @@ class AgentRunner:
                                     terminal_failure, observed_failure
                                 )
                                 terminal_run = True
-                        elif terminal_control and terminal_control.get("terminate_run"):
-                            terminal_summary = (
-                                terminal_control.get("result_summary") or tool_text
+                        else:
+                            await self._observe_failure_progress(
+                                trace,
+                                config,
+                                explicit_role,
+                                None,
                             )
-                            terminal_run = True
+                            if terminal_control and terminal_control.get("terminate_run"):
+                                terminal_summary = (
+                                    terminal_control.get("result_summary") or tool_text
+                                )
+                                terminal_run = True
                     if terminal_summary and terminal_failure is None and self.trace_store:
                         await self.trace_store.update_trace(
                             trace_id,
@@ -2453,24 +2492,41 @@ class AgentRunner:
                         observed_failure = self._control_failure(terminal_control)
                         if observed_failure is not None:
                             last_failure = observed_failure
+                            repeated = await self._observe_failure_progress(
+                                trace,
+                                config,
+                                explicit_role,
+                                observed_failure,
+                            )
+                            if repeated is not None:
+                                terminal_failure = repeated
+                                terminal_run = True
+                                break
                             if self._failure_terminates_role(
                                 observed_failure, explicit_role
                             ):
                                 terminal_failure = observed_failure
                                 terminal_run = True
                                 break
-                        elif terminal_control and terminal_control.get("terminate_run"):
-                            terminal_run = True
-                            terminal_summary = (
-                                terminal_control.get("result_summary") or tool_text
+                        else:
+                            await self._observe_failure_progress(
+                                trace,
+                                config,
+                                explicit_role,
+                                None,
                             )
-                            if self.trace_store:
-                                await self.trace_store.update_trace(
-                                    trace_id,
-                                    result_summary=terminal_summary,
-                                    head_sequence=head_seq,
+                            if terminal_control and terminal_control.get("terminate_run"):
+                                terminal_run = True
+                                terminal_summary = (
+                                    terminal_control.get("result_summary") or tool_text
                                 )
-                            break
+                                if self.trace_store:
+                                    await self.trace_store.update_trace(
+                                        trace_id,
+                                        result_summary=terminal_summary,
+                                        head_sequence=head_seq,
+                                    )
+                                break
 
                 if terminal_run:
                     if (
@@ -2588,7 +2644,12 @@ class AgentRunner:
         final_failure: Optional[FailureDetail] = None
 
         if terminal_failure is not None:
-            final_status = "failed"
+            final_status = (
+                "incomplete"
+                if terminal_failure.code == "NO_PROGRESS"
+                and explicit_role == AgentRole.PLANNER
+                else "failed"
+            )
             final_failure = terminal_failure
             final_error = self._failure_text(terminal_failure)
         elif explicit_role == AgentRole.PLANNER:
@@ -2683,6 +2744,118 @@ class AgentRunner:
     def _failure_text(failure: FailureDetail) -> str:
         return f"{failure.code}: {failure.message}"
 
+    @staticmethod
+    def _no_progress_enabled(
+        config: RunConfig,
+        role: Optional[AgentRole],
+    ) -> bool:
+        return role is not None or config.no_progress.enabled_in_legacy
+
+    async def _observe_failure_progress(
+        self,
+        trace: Trace,
+        config: RunConfig,
+        role: Optional[AgentRole],
+        failure: Optional[FailureDetail],
+    ) -> Optional[FailureDetail]:
+        """Persist consecutive failure state and return a breaker failure."""
+
+        if not self._no_progress_enabled(config, role):
+            return None
+        runtime = trace.runtime_state.setdefault("no_progress", {})
+        if failure is None:
+            runtime["failure_fingerprint"] = None
+            runtime["failure_count"] = 0
+            await self._persist_runtime_state(trace)
+            return None
+
+        fingerprint = failure.fingerprint()
+        if runtime.get("failure_fingerprint") == fingerprint:
+            count = int(runtime.get("failure_count", 0)) + 1
+        else:
+            count = 1
+        runtime["failure_fingerprint"] = fingerprint
+        runtime["failure_count"] = count
+        runtime["last_failure"] = failure.to_dict()
+        await self._persist_runtime_state(trace)
+        if count < config.no_progress.same_failure_limit:
+            return None
+        return await self._emit_no_progress(
+            trace,
+            reason="same_failure_repeated",
+            details={"count": count, "last_failure": failure.to_dict()},
+        )
+
+    async def _observe_planner_progress(
+        self,
+        trace: Trace,
+        config: RunConfig,
+    ) -> Optional[FailureDetail]:
+        """Compare the durable TaskLedger semantics at Planner turn boundaries."""
+
+        if self.task_coordinator is None:
+            return None
+        method = getattr(self.task_coordinator, "progress_snapshot", None)
+        if method is None:
+            return None
+        root_trace_id = (trace.context or {}).get("root_trace_id", trace.trace_id)
+        snapshot = await method(root_trace_id)
+        fingerprint = hashlib.sha256(
+            json.dumps(
+                snapshot,
+                ensure_ascii=False,
+                sort_keys=True,
+                separators=(",", ":"),
+            ).encode("utf-8")
+        ).hexdigest()
+        runtime = trace.runtime_state.setdefault("no_progress", {})
+        previous = runtime.get("planner_fingerprint")
+        if previous is None or previous != fingerprint:
+            count = 0
+        else:
+            count = int(runtime.get("planner_stall_count", 0)) + 1
+        runtime["planner_fingerprint"] = fingerprint
+        runtime["planner_stall_count"] = count
+        await self._persist_runtime_state(trace)
+        if count < config.no_progress.planner_stall_limit:
+            return None
+        return await self._emit_no_progress(
+            trace,
+            reason="planner_state_unchanged",
+            details={
+                "count": count,
+                "last_failure": runtime.get("last_failure"),
+            },
+        )
+
+    async def _persist_runtime_state(self, trace: Trace) -> None:
+        if self.trace_store:
+            await self.trace_store.update_trace(
+                trace.trace_id,
+                runtime_state=trace.runtime_state,
+            )
+
+    async def _emit_no_progress(
+        self,
+        trace: Trace,
+        *,
+        reason: str,
+        details: Dict[str, Any],
+    ) -> FailureDetail:
+        failure = FailureDetail(
+            code="NO_PROGRESS",
+            message="Agent stopped because no semantic progress was detected",
+            disposition=FailureDisposition.ABORT_RUN,
+            details={"reason": reason, **details},
+        )
+        if self.trace_store:
+            await self.trace_store.append_event(
+                trace.trace_id,
+                "no_progress_detected",
+                failure.to_dict(),
+            )
+        return failure
+
     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, "root_completion", None)

agent/agent/orchestration/coordinator.py

@@ -281,6 +281,97 @@ class TaskCoordinator:
         self._root_task(ledger)
         return render_task_context(ledger)
 
+    async def progress_snapshot(self, root_trace_id: str) -> Dict[str, Any]:
+        """Return an ID-free semantic snapshot used by the Runner stall guard."""
+
+        ledger = await self.task_store.load(root_trace_id)
+        root = self._root_task(ledger)
+
+        def spec_value(task: TaskRecord) -> Dict[str, Any]:
+            spec = task.current_spec
+            return {
+                "objective": spec.objective,
+                "acceptance_criteria": [
+                    {
+                        "criterion_id": criterion.criterion_id,
+                        "description": criterion.description,
+                        "hard": criterion.hard,
+                    }
+                    for criterion in spec.acceptance_criteria
+                ],
+                "context_refs": sorted(spec.context_refs),
+            }
+
+        def task_key(task: TaskRecord) -> str:
+            return json.dumps(spec_value(task), ensure_ascii=False, sort_keys=True)
+
+        active_tasks = [
+            {
+                "spec": spec_value(task),
+                "status": task.status.value,
+                "blocked_reason": task.blocked_reason,
+                "parent_spec": (
+                    spec_value(ledger.tasks[task.parent_task_id])
+                    if task.parent_task_id in ledger.tasks
+                    else None
+                ),
+            }
+            for task in sorted(ledger.tasks.values(), key=task_key)
+            if task.status not in TERMINAL_TASK_STATUSES
+        ]
+
+        accepted: List[Dict[str, Any]] = []
+        for decision in ledger.decisions.values():
+            if decision.action != DecisionAction.ACCEPT:
+                continue
+            task = ledger.tasks.get(decision.task_id)
+            attempt = ledger.attempts.get(decision.attempt_id or "")
+            if task is None or attempt is None or attempt.submission is None:
+                continue
+            accepted.append(
+                {
+                    "spec": spec_value(task),
+                    "artifacts": sorted(
+                        (
+                            ref.kind,
+                            ref.digest or "",
+                            ref.version or "",
+                        )
+                        for ref in attempt.submission.artifact_refs
+                    ),
+                    "snapshot_sha256": (
+                        (await self.artifact_store.get(root_trace_id, attempt.snapshot_id)).sha256
+                        if attempt.snapshot_id
+                        else None
+                    ),
+                }
+            )
+
+        failures = [
+            item.failure
+            for item in (*ledger.attempts.values(), *ledger.validations.values())
+            if item.failure is not None
+        ]
+        last_failure = failures[-1].fingerprint() if failures else None
+        return {
+            "root": {
+                "spec": spec_value(root),
+                "status": root.status.value,
+                "blocked_reason": root.blocked_reason,
+            },
+            "active_tasks": active_tasks,
+            "accepted": sorted(
+                accepted,
+                key=lambda value: json.dumps(value, ensure_ascii=False, sort_keys=True),
+            ),
+            "focused_task": (
+                spec_value(ledger.tasks[ledger.focused_task_id])
+                if ledger.focused_task_id in ledger.tasks
+                else None
+            ),
+            "last_failure": last_failure,
+        }
+
     async def _mutate(
         self,
         root_trace_id: str,

agent/tests/test_no_progress_policy.py

@@ -0,0 +1,199 @@
+import pytest
+
+from agent.core.presets import AgentPreset, register_preset
+from agent.core.runner import AgentRunner, NoProgressPolicy, RunConfig
+from agent.failures import FailureDetail, FailureDisposition, ToolExecutionError
+from agent.orchestration.models import AgentRole, CompletionPolicy
+from agent.trace.store import FileSystemTraceStore
+from agent.tools.models import ToolCapability
+from agent.tools.registry import ToolRegistry
+
+from test_coordinator_integration import FakeExecutor, create_task, make_coordinator
+
+
+class StaticProgressCoordinator:
+    def __init__(self):
+        self.ensure_calls = 0
+
+    async def ensure_ledger(self, *_args, **_kwargs):
+        self.ensure_calls += 1
+
+    async def root_completion(self, _root_trace_id):
+        return {
+            "status": "needs_replan",
+            "root_task_id": "ignored-id",
+            "pending_tasks": [{"objective": "same task", "status": "needs_replan"}],
+        }
+
+    async def progress_snapshot(self, _root_trace_id):
+        return {
+            "root": {"objective": "same task", "status": "needs_replan"},
+            "active_tasks": [],
+            "accepted": [],
+        }
+
+
+def _knowledge_off():
+    from agent.tools.builtin.knowledge import KnowledgeConfig
+
+    return KnowledgeConfig(
+        enable_extraction=False,
+        enable_completion_extraction=False,
+        enable_injection=False,
+    )
+
+
+def _explicit_config(preset, *, max_iterations=8, tools=None):
+    return RunConfig(
+        agent_type=preset,
+        completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+        max_iterations=max_iterations,
+        tools=tools or [],
+        tool_groups=[],
+        root_task_spec={
+            "objective": "same task",
+            "acceptance_criteria": [{"description": "finish it"}],
+        },
+        knowledge=_knowledge_off(),
+    )
+
+
+@pytest.mark.asyncio
+async def test_planner_stops_after_three_unchanged_semantic_cycles(tmp_path):
+    preset = "test_no_progress_planner_stall"
+    register_preset(
+        preset,
+        AgentPreset(role=AgentRole.PLANNER, allowed_tools=[], max_iterations=8, skills=[]),
+    )
+    calls = 0
+
+    async def llm_call(**_kwargs):
+        nonlocal calls
+        calls += 1
+        return {"content": "still thinking", "tool_calls": None, "finish_reason": "stop"}
+
+    store = FileSystemTraceStore(str(tmp_path))
+    result = await AgentRunner(
+        trace_store=store,
+        llm_call=llm_call,
+        task_coordinator=StaticProgressCoordinator(),
+    ).run_result(
+        [{"role": "user", "content": "plan"}],
+        _explicit_config(preset),
+    )
+
+    assert calls == 3
+    assert result["status"] == "incomplete"
+    assert result["failure"]["code"] == "NO_PROGRESS"
+    events = await store.get_events(result["trace_id"])
+    assert [event["event"] for event in events].count("no_progress_detected") == 1
+
+
+@pytest.mark.asyncio
+async def test_second_identical_tool_failure_stops_worker(tmp_path):
+    preset = "test_no_progress_worker_failure"
+    register_preset(
+        preset,
+        AgentPreset(
+            role=AgentRole.WORKER,
+            allowed_tools=["flaky_write"],
+            max_iterations=8,
+            skills=[],
+        ),
+    )
+    registry = ToolRegistry()
+
+    async def flaky_write():
+        raise ToolExecutionError(
+            FailureDetail(
+                code="WORKSPACE_NOT_READY",
+                message="create the missing workspace file",
+                disposition=FailureDisposition.REPAIR_ATTEMPT,
+                source_tool="flaky_write",
+            )
+        )
+
+    registry.register(flaky_write, capabilities=[ToolCapability.WRITE])
+    calls = 0
+
+    async def llm_call(**_kwargs):
+        nonlocal calls
+        calls += 1
+        return {
+            "content": "",
+            "tool_calls": [{
+                "id": f"call-{calls}",
+                "type": "function",
+                "function": {"name": "flaky_write", "arguments": "{}"},
+            }],
+            "finish_reason": "tool_calls",
+        }
+
+    result = await AgentRunner(
+        trace_store=FileSystemTraceStore(str(tmp_path)),
+        tool_registry=registry,
+        llm_call=llm_call,
+        task_coordinator=StaticProgressCoordinator(),
+    ).run_result(
+        [{"role": "user", "content": "work"}],
+        _explicit_config(preset, tools=["flaky_write"]),
+    )
+
+    assert calls == 2
+    assert result["status"] == "failed"
+    assert result["failure"]["code"] == "NO_PROGRESS"
+    assert result["failure"]["details"]["last_failure"]["code"] == "WORKSPACE_NOT_READY"
+
+
+@pytest.mark.asyncio
+async def test_planner_stall_counter_survives_resume(tmp_path):
+    preset = "test_no_progress_resume"
+    register_preset(
+        preset,
+        AgentPreset(role=AgentRole.PLANNER, allowed_tools=[], max_iterations=4, skills=[]),
+    )
+    calls = 0
+
+    async def llm_call(**_kwargs):
+        nonlocal calls
+        calls += 1
+        return {"content": "", "tool_calls": None, "finish_reason": "stop"}
+
+    store = FileSystemTraceStore(str(tmp_path))
+    runner = AgentRunner(
+        trace_store=store,
+        llm_call=llm_call,
+        task_coordinator=StaticProgressCoordinator(),
+    )
+    first_config = _explicit_config(preset, max_iterations=2)
+    first_config._role_run_config_override_fields = frozenset({"max_iterations"})
+    first = await runner.run_result([{"role": "user", "content": "plan"}], first_config)
+    assert first["status"] == "incomplete"
+    assert calls == 2
+
+    resumed = await runner.run_result(
+        [],
+        RunConfig(trace_id=first["trace_id"], knowledge=_knowledge_off()),
+    )
+
+    assert calls == 3
+    assert resumed["status"] == "incomplete"
+    assert resumed["failure"]["code"] == "NO_PROGRESS"
+
+
+@pytest.mark.asyncio
+async def test_semantically_equal_task_ledgers_ignore_generated_ids(tmp_path):
+    first, _, _ = await make_coordinator(tmp_path / "first", FakeExecutor([]))
+    second, _, _ = await make_coordinator(tmp_path / "second", FakeExecutor([]))
+    first_task_id = await create_task(first, objective="equivalent child")
+    second_task_id = await create_task(second, objective="equivalent child")
+
+    assert first_task_id != second_task_id
+    assert await first.progress_snapshot("root") == await second.progress_snapshot("root")
+
+
+def test_no_progress_policy_rejects_invalid_thresholds():
+    with pytest.raises(ValueError, match="same_failure_limit"):
+        NoProgressPolicy(same_failure_limit=0)
+    with pytest.raises(ValueError, match="planner_stall_limit"):
+        NoProgressPolicy(planner_stall_limit=0)