diff --git a/backend/README.md b/backend/README.md index 8cf8cf83d7..0c1609b96a 100644 --- a/backend/README.md +++ b/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) | diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md index 70e433a0bc..1a334379b2 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md +++ b/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 diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py index fdd79214df..5cd1fbf4a6 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py +++ b/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 # ------------------------------------------------------------------ diff --git a/backend/tests/test_tool_progress_entry_fence.py b/backend/tests/test_tool_progress_entry_fence.py new file mode 100644 index 0000000000..80a9fc9b63 --- /dev/null +++ b/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) == []