fix(agents): fence late progress assessments across graph entries (#8083)

* fix(agents): fence late progress assessments across graph entries

Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com>

* docs(agents): keep combined progress guidance within budget

Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com>

---------

Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com>
Co-authored-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com>
Co-authored-by: Totoro <msy626836554@gmail.com>

backend/README.md

@@ -54,6 +54,10 @@ The single LangGraph agent (`lead_agent`) is the runtime entry point, created vi
 
 Middlewares execute in strict order, each handling a specific concern:
 
+Starting a new graph invocation resets tool-progress tracking. Results and model
+retries still finishing from an earlier invocation cannot block its tools or
+restore obsolete progress hints, including same-run goal continuations.
+
 | # | Middleware | Purpose |
 |---|-----------|---------|
 | 1 | **ThreadDataMiddleware** | Creates per-thread isolated directories (workspace, uploads, outputs) |

backend/packages/harness/deerflow/agents/middlewares/AGENTS.md

@@ -1,5 +1,7 @@
 ### Middleware Chain
 
+ToolProgress fences late results/hint retries by entry epoch, even for one run.
+
 JSON depth failures preserve content and never abort graphs. Unclassified
 explicit failures stay `unknown` / recoverable / `try_alternative`; incidental
 JSON fields cannot establish auth failure. Producer metadata wins. Keep real

backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py

@@ -64,6 +64,7 @@ from langgraph.prebuilt.tool_node import ToolCallRequest
 from langgraph.runtime import Runtime
 from langgraph.types import Command
 
+from deerflow.agents.middlewares._bounded_dict import BoundedDict
 from deerflow.agents.middlewares.audit_context import (
     TOOL_PROGRESS_RECORDER_CONTEXT_KEY,
     resolve_audit_recorder,
@@ -280,6 +281,9 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         self._lock = threading.Lock()
         # LRU-evicting store: thread_id → {tool_name → ToolPhaseState}
         self._phase_states: OrderedDict[str, dict[str, ToolPhaseState]] = OrderedDict()
+        # Results retain the epoch captured before their handler started. A new
+        # graph entry invalidates in-flight assessments, even for the same run.
+        self._entry_epochs: BoundedDict[str, object] = BoundedDict(max_tracked_threads)
         # Pending hint queue: (thread_id, run_id) → [hint texts]
         self._pending: dict[tuple[str, str], list[str]] = defaultdict(list)
 
@@ -311,6 +315,13 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
     def _pending_key(self, runtime: Runtime) -> tuple[str, str]:
         return self._thread_id(runtime), self._run_id(runtime)
 
+    def _entry_epoch(self, runtime: Runtime) -> object:
+        thread_id = self._thread_id(runtime)
+        with self._lock:
+            epoch = self._entry_epochs.setdefault(thread_id, object())
+            self._entry_epochs.move_to_end(thread_id)
+            return epoch
+
     def _record_phase_transition(
         self,
         *,
@@ -411,6 +422,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         tool_call_id: str,
         *,
         hook: Literal["wrap_tool_call", "awrap_tool_call"],
+        entry_epoch: object,
     ) -> ToolMessage | Command:
         """Update the state machine from a tool result; queue hints if warranted."""
         message = _result_tool_message(result, tool_call_id)
@@ -428,6 +440,8 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         thread_id = self._thread_id(runtime)
         phase_change: tuple[ToolPhaseState, ToolPhaseState, ToolPhaseTransition] | None = None
         with self._lock:
+            if self._entry_epochs.get(thread_id) is not entry_epoch:
+                return result
             state = self._get_state(thread_id, tool_name)
             new_state, hint, transition = self._assess_and_transition(state, meta, content)
             self._set_state(thread_id, tool_name, new_state)
@@ -466,7 +480,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
                     tool_name,
                 )
         if hint and self._inject_assessment:
-            self._queue_assessment(runtime, hint)
+            self._queue_assessment(runtime, hint, entry_epoch=entry_epoch)
         return result
 
     # ------------------------------------------------------------------
@@ -561,10 +575,12 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
     # ------------------------------------------------------------------
     # Pending queue helpers
 
-    def _queue_assessment(self, runtime: Runtime, text: str) -> None:
+    def _queue_assessment(self, runtime: Runtime, text: str, *, entry_epoch: object | None = None) -> None:
         key = self._pending_key(runtime)
         thread_id = key[0]
         with self._lock:
+            if entry_epoch is not None and self._entry_epochs.get(thread_id) is not entry_epoch:
+                return
             # Guard against creating a phantom _pending entry for a thread that was just
             # evicted from _phase_states by the LRU.  Such entries can never be cleaned up
             # by the eviction loop (which only walks _phase_states) and accumulate silently.
@@ -579,7 +595,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         with self._lock:
             return self._pending.pop(key, [])
 
-    def _restore_pending(self, runtime: Runtime, hints: list[str]) -> None:
+    def _restore_pending(self, runtime: Runtime, hints: list[str], *, entry_epoch: object | None = None) -> None:
         """Requeue hints taken for a model call that raised.
 
         LLMErrorHandlingMiddleware sits outside this middleware and retries a
@@ -590,6 +606,8 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
             return
         key = self._pending_key(runtime)
         with self._lock:
+            if entry_epoch is not None and self._entry_epochs.get(key[0]) is not entry_epoch:
+                return
             if key[0] not in self._phase_states:
                 return
             queue = self._pending[key]
@@ -631,6 +649,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         thread_id = self._thread_id(runtime)
         transitions: list[tuple[str, ToolPhaseState, ToolPhaseState]] = []
         with self._lock:
+            self._entry_epochs[thread_id] = object()
             thread_tools = self._phase_states.get(thread_id)
             if thread_tools is None:
                 return
@@ -672,6 +691,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         runtime = getattr(request, "runtime", None)
         if runtime is None:
             return handler(request)
+        entry_epoch = self._entry_epoch(runtime)
         block_reason = self._get_block_reason(runtime, tool_name)
         if block_reason:
             logger.info(
@@ -687,6 +707,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
             runtime,
             str(request.tool_call.get("id") or ""),
             hook="wrap_tool_call",
+            entry_epoch=entry_epoch,
         )
 
     @override
@@ -701,6 +722,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         runtime = getattr(request, "runtime", None)
         if runtime is None:
             return await handler(request)
+        entry_epoch = self._entry_epoch(runtime)
         block_reason = self._get_block_reason(runtime, tool_name)
         if block_reason:
             logger.info(
@@ -716,6 +738,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
             runtime,
             str(request.tool_call.get("id") or ""),
             hook="awrap_tool_call",
+            entry_epoch=entry_epoch,
         )
 
     # ------------------------------------------------------------------
@@ -742,11 +765,12 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         request: ModelRequest,
         handler: Callable[[ModelRequest], ModelResponse],
     ) -> ModelCallResult:
+        entry_epoch = self._entry_epoch(request.runtime)
         hints = self._drain_pending(request.runtime)
         try:
             return handler(self._inject_hints(request, hints))
         except Exception:
-            self._restore_pending(request.runtime, hints)
+            self._restore_pending(request.runtime, hints, entry_epoch=entry_epoch)
             raise
 
     @override
@@ -755,11 +779,12 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         request: ModelRequest,
         handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
     ) -> ModelCallResult:
+        entry_epoch = self._entry_epoch(request.runtime)
         hints = self._drain_pending(request.runtime)
         try:
             return await handler(self._inject_hints(request, hints))
         except Exception:
-            self._restore_pending(request.runtime, hints)
+            self._restore_pending(request.runtime, hints, entry_epoch=entry_epoch)
             raise
 
     # ------------------------------------------------------------------

backend/tests/test_tool_progress_entry_fence.py

@@ -0,0 +1,93 @@
+"""Late results from an earlier graph entry cannot poison its successor."""
+
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+from langchain_core.messages import ToolMessage
+
+from deerflow.agents.middlewares.tool_progress_middleware import ToolProgressMiddleware
+from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY
+
+
+def _result(failed):
+    return ToolMessage(
+        content="credential rejected" if failed else "fresh useful result",
+        tool_call_id="call",
+        additional_kwargs={
+            TOOL_META_KEY: {
+                "status": "error" if failed else "success",
+                "error_type": "auth" if failed else None,
+                "recoverable_by_model": not failed,
+                "recommended_next_action": "stop" if failed else "continue",
+                "source": "tool_return",
+            }
+        },
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("next_run", ["new-run", "old-run", None])
+async def test_late_failure_does_not_block_next_graph_entry(asynchronous, next_run):
+    middleware = ToolProgressMiddleware()
+    old = SimpleNamespace(context={"thread_id": "thread", "run_id": "old-run" if next_run else None})
+    new = SimpleNamespace(context={"thread_id": "thread", "run_id": next_run})
+    middleware.before_agent({}, old)
+    request = SimpleNamespace(tool_call={"name": "lookup", "id": "call"}, runtime=old)
+    late = _result(True)
+
+    def handler(_):
+        # The old tool started before a new invocation reset the thread guard.
+        middleware.before_agent({}, new)
+        return late
+
+    async def ahandler(_):
+        await middleware.abefore_agent({}, new)
+        return late
+
+    returned = await middleware.awrap_tool_call(request, ahandler) if asynchronous else middleware.wrap_tool_call(request, handler)
+    assert returned is late  # Keep the old run's real outcome for its caller.
+    request.runtime = new
+    fresh = _result(False)
+    executed = []
+
+    def succeeding(_):
+        executed.append(True)
+        return fresh
+
+    async def asucceeding(_):
+        return succeeding(_)
+
+    returned = await middleware.awrap_tool_call(request, asucceeding) if asynchronous else middleware.wrap_tool_call(request, succeeding)
+    assert executed == [True]
+    assert returned is fresh
+    assert middleware._drain_pending(new) == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("asynchronous", [False, True])
+async def test_failed_model_call_cannot_restore_hints_into_new_entry(asynchronous):
+    middleware = ToolProgressMiddleware(stagnation_threshold=1)
+    runtime = SimpleNamespace(context={"thread_id": "thread", "run_id": "run"})
+    middleware.before_agent({}, runtime)
+    request = SimpleNamespace(tool_call={"name": "lookup", "id": "call"}, runtime=runtime)
+    message = _result(True)
+    message.additional_kwargs[TOOL_META_KEY].update(error_type="no_results", recoverable_by_model=True, recommended_next_action="rewrite_query")
+    middleware.wrap_tool_call(request, lambda _: message)
+    model_request = Mock(runtime=runtime, messages=[])
+
+    def handler(_):
+        middleware.before_agent({}, runtime)
+        assert middleware._drain_pending(runtime) == []
+        raise RuntimeError("old request failed after re-entry")
+
+    async def ahandler(_):
+        return handler(_)
+
+    with pytest.raises(RuntimeError, match="old request failed"):
+        if asynchronous:
+            await middleware.awrap_model_call(model_request, ahandler)
+        else:
+            middleware.wrap_model_call(model_request, handler)
+    assert middleware._drain_pending(runtime) == []