From fd41fdb065ac150f09a906c061c37da975d55c5e Mon Sep 17 00:00:00 2001 From: He Wang Date: Mon, 6 Jul 2026 20:57:41 +0800 Subject: [PATCH] feat(middleware): add structured tool result meta and tool-progress state machine (#3601) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(middleware): add structured tool result meta and tool-progress state machine feat: - Add tool_result_meta.py: ToolResultMeta dataclass (status/error_type/retryable/ recoverable_by_model/recommended_next_action/source) + normalize_tool_result and stamp_exception_meta utilities; classifies every ToolMessage regardless of path - Add ToolProgressMiddleware: per-(thread_id, tool_name) state machine ACTIVE → WARNED (hint injected as HumanMessage) → BLOCKED (call short-circuited); Jaccard near-duplicate detection for repeated successful results; auth/config/internal errors bypass WARNED and go directly to BLOCKED; LRU-bounded thread state store - Add ToolProgressConfig: all thresholds configurable (stagnation_threshold, warn_escalation_count, jaccard_similarity_threshold, exempt_tools, etc.); disabled by default (enabled: false) - Wire ToolProgressMiddleware as outer wrapper around ToolErrorHandlingMiddleware in _build_runtime_middlewares so it receives results already carrying deerflow_tool_meta fix: - ToolErrorHandlingMiddleware now calls stamp_exception_meta on exception path and normalize_tool_result on success path so every ToolMessage carries deerflow_tool_meta test: - Add test_tool_result_meta.py: 26 cases covering all classification paths, stamp_exception_meta, and normalize_tool_result Command passthrough - Add test_tool_progress_middleware.py: 27 cases including full async paths, Jaccard duplicate detection, LRU eviction, hint injection, and malformed meta passthrough - Extend test_tool_error_handling_middleware.py: middleware ordering invariant and meta stamping on exception docs: - Add tool_progress section to config.example.yaml with all fields and descriptions - Update CLAUDE.md middleware chain documentation (entries 8-9) Co-Authored-By: Claude Sonnet 4 * fix(middleware): recoverable errors stay WARNED; fix auth keyword shadowing fix: - WARNED is terminal for recoverable_by_model=True errors (no_results, not_found, permission); hint re-injected on each problem call instead of escalating to BLOCKED, so the model can retry with different parameters (e.g. fresh query, new URL) without being hard-blocked by a prior stagnation count. Non-recoverable (rate_limited, transient) still escalate WARNED → BLOCKED after warn_escalation_count more problems; auth/config/internal remain immediately BLOCKED. - Remove bare "api key" keyword from auth classification rule so "no api key configured" correctly classifies as config (not auth), producing the accurate block-reason text for the model. docs: - CLAUDE.md: document all three ToolProgressMiddleware transition paths - config.example.yaml: update inline state-machine comment to match new paths test: - test_recoverable_errors_stay_warned_indefinitely: WARNED never escalates for recoverable errors regardless of how many problem calls accumulate - test_recoverable_error_re_injects_hint_past_escalation: hints continue past the escalation zone for recoverable errors - test_no_api_key_is_config_not_auth: regression guard for keyword shadowing fix Co-Authored-By: Claude Sonnet 4 * fix(tool_result_meta): add JSON error extraction and fix source classification fix: - Fix non-standard error path: source was "exception" but should be "tool_return" - Add _extract_json_error_text to isolate JSON error fields from noisy JSON bodies (e.g. Brave Search {"error": "...", "query": "..."} — query keywords no longer pollute error classification) - Add success-path JSON extraction to catch tools that return HTTP 200 with a JSON error body (status="success" but {"error": "API key not configured"}) - Add _SEMANTIC_ZERO_ERROR_STRINGS frozenset to suppress false positives from tools that use {"error": "none"} / {"error": "null"} / {"error": "ok"} as success signals - Document that stamp_exception_meta always overwrites existing TOOL_META_KEY (exception-derived classification is authoritative over tool return-time stamps) test: - Add parametrized regression tests for all semantic-zero error strings - Add tests for non-standard error path source field - Add tests for JSON error extraction (nonstd, success-path, numeric, falsy values) - Correct test comment for test_no_api_key_is_config_not_auth Co-Authored-By: Claude Sonnet 4 * fix(tool_progress_middleware): fix 6 bugs, add terminal guard and structured logging fix: - H1: fix exempt_tools empty-set silently ignored — use `is not None` instead of truthiness check so ToolProgressConfig(exempt_tools=set()) correctly disables all exemptions - Fix _get_block_reason creating phantom LRU entries via _get_state (write path); now uses dict.get + explicit move_to_end on read path only - Fix _pending memory leak: LRU eviction of _phase_states now synchronously removes all (evicted_thread, *) keys from _pending - Fix _assess_and_transition missing terminal guard for blocked state — a recoverable error result could silently demote blocked → warned in concurrent-race scenarios; early return preserves terminal semantics - Fix recent_word_sets window: stored [-5:] but is_near_duplicate only compared [-3:]; align to [-3:] and change type list→tuple (prevents accidental in-place mutation across dataclasses.replace shallow copies) - Fix _format_hint missing "success" key and "continue" action: Jaccard near-duplicate results produced the generic fallback instead of a specific actionable message feat: - Add structured state-transition logging (ACTIVE/WARNED/BLOCKED transitions, blocked intercepts, hint injection debug log) test: - Add regression tests for all 6 bug fixes (H1, phantom LRU, pending leak, terminal guard, window alignment, format_hint near-dup) - Add Jaccard near-threshold boundary test (7/9 vs 8/9 Jaccard) - Add production min_words=10 skip test for short content - Add exempt_tools empty-set and None round-trip tests - Add _augment_request deduplication test - Add before_agent current-run preservation test - Add structured logging tests (WARNED/BLOCKED/ACTIVE/intercepted/debug) Co-Authored-By: Claude Sonnet 4 * chore(config): remove unused backward-compat fields from ToolProgressConfig Remove max_calls_per_intent and window_size fields that were marked "Retained for backward compatibility; not used by the current state machine" when the state machine was introduced. Pydantic v2 ignores unknown fields by default, so existing config.yaml files with these keys remain valid. Co-Authored-By: Claude Sonnet 4 * fix(tool_progress): address PR review and multi-agent review findings fix: - Remove <80-char length gate for partial_success; only _PARTIAL_MARKERS now - Add word-boundary regex for numeric HTTP codes (401/403/404/500) to avoid false positives like "500ms" or "4010 rows" triggering hard-block - Add "task" to default exempt_tools (delegation primitive, not a search tool) - Remove move_to_end() from _get_block_reason read path; blocked threads were permanently warm in LRU, starving active threads of eviction slots - Add _reset_blocked_states in before_agent: scope BLOCKED and WARNED states to a single run; clear recent_word_sets so stale Jaccard windows don't cause false near-duplicate detections in the next run - Compute word_set() lazily (only for success results); cap content at 8192 chars to bound memory and CPU cost on large tool results - Remove unused retryable field from ToolResultMeta (no consumer existed) - Add isinstance-based ordering guard and warning log for missing meta - Fix JSON-without-error-key fallback: use _UNKNOWN_ERROR instead of classifying incidental field values (e.g. {"user_id": 401} → auth → stop) - Fix _extract_json_error_text: use json.dumps for dict/list error fields instead of str() which produced Python repr matching config rules spuriously - Add "no results found"/"no content found"/"no images found" to _PARTIAL_MARKERS so success responses with empty results trigger stagnation detection - Fix immediate-block path to increment consecutive_problems (was left at 0) - Fix _queue_assessment: skip phantom _pending entries for evicted threads - Bump config_version 13→16 (upstream added 14/15; tool_progress is additive) test: - Update test_short_content_is_partial → test_short_terse_success_is_not_partial - Add parametrized test_numeric_keyword_word_boundary (8 positive + negative cases) - Add test_before_agent_resets_blocked_states_for_new_run (strengthened assertions) - Add test_before_agent_resets_warned_states_for_new_run - Add test_missing_meta_on_non_exempt_tool_emits_warning - Add test_middleware_ordering_guard_raises_when_progress_is_inner - Add test_auth_error_immediately_blocked asserts consecutive_problems == 1 - Add tests for JSON-without-error-key, dict error field, no-results partial_success Co-Authored-By: Claude * fix(tool_progress): address second PR review — perf, architecture doc, concurrency note fix: - Extract content.lower() once before _PARTIAL_MARKERS check in normalize_tool_message; previously computed up to 7× per call inside the generator (once per marker) docs: - Add division-of-labor paragraph to ToolProgressMiddleware module docstring explaining coexistence with LoopDetectionMiddleware: result-quality guard (per-tool BLOCK) vs call-pattern guard (whole-turn hard-stop); no shared state, no double-stop risk - Add threading.Lock comment explaining why asyncio.Lock is not used (short critical sections, must also protect sync wrap_tool_call path from subagent executor threads) - Update backend/CLAUDE.md entry 8 with division-of-labor summary; fix entry 9 (remove stale retryable field reference, add missing recoverable_by_model/source) test: - Add test_tool_progress_and_loop_detection_coexist_without_interfering: drives both middlewares to WARNED state simultaneously, verifies independent state, independent hint queues, and no cross-contamination; uses snapshot copy for final assertion Co-Authored-By: Claude * fix(tool_progress): reset all tool states at run boundary; fix semantic-zero test validity fix: - _reset_run_states (formerly _reset_blocked_states) drops the phase filter and resets all tracked (thread, tool) pairs unconditionally at before_agent; ACTIVE tools with sub-threshold consecutive_problems or cached recent_word_sets no longer bleed into the next run, preventing spurious WARNED transitions on clean R2 calls - test_normalize_json_semantic_zero_error_string_not_treated_as_error: replace {error_value!r} f-string (produces invalid JSON with single quotes) with json.dumps so _extract_json_error_text actually parses the payload and the _SEMANTIC_ZERO_ERROR_STRINGS guard is exercised, not bypassed at json.loads test: - add test_before_agent_resets_active_state_consecutive_problems_and_word_sets to lock the ACTIVE-phase run-boundary reset: drives tool to active/cp=1/ws≠() in R1, asserts both fields are zero/empty after before_agent fires for R2 * docs(tool_progress): document intentional per-run reset vs LoopDetection thread-scoped retention Addresses reviewer observation in PR #3601 that _reset_run_states diverges from LoopDetectionMiddleware's cross-run scoping policy without explanation. Expands the _reset_run_states docstring to record the intentional design choice: ToolProgressMiddleware resets per-run because result-quality errors (rate_limited, transient) are time-bound and may resolve between turns — retaining stale counters would risk false-positive BLOCKED calls. LoopDetectionMiddleware retains history across runs because call-pattern loops are time-invariant. The divergence is by design, not oversight. Co-Authored-By: Claude * fix(middleware): restore ReadBeforeWriteMiddleware as outermost write gate A merge conflict resolution had accidentally placed ReadBeforeWriteMiddleware after ToolErrorHandlingMiddleware (inner), reversing the original intent from b81334cc where it was the outermost write gate before ToolErrorHandling. fix: - Restore ReadBeforeWriteMiddleware to outer position: ReadBeforeWrite → ToolProgress → ToolErrorHandling. Blocked writes now return immediately without consuming a ToolProgress slot. - Add normalize_tool_result call on blocked ToolMessages so they carry deerflow_tool_meta (recoverable_by_model=True) even though they bypass ToolErrorHandlingMiddleware. test: - Add test_blocked_write_has_deerflow_tool_meta (sync + async) to lock the normalize_tool_result behavior on blocked writes. - Fix chain order assertions in TestChainWiring and test_build_lead_runtime_middlewares_chain_order_matches_agents_md. docs: - Renumber AGENTS.md items: 10→ReadBeforeWrite, 11→ToolProgress, 12→ToolErrorHandling; update descriptions to reflect outermost-gate design. - Fix stale cross-reference: LoopDetectionMiddleware (item 23) → (item 25). Co-Authored-By: Claude --------- Co-authored-by: Claude Sonnet 4 Co-authored-by: Cursor Co-authored-by: Willem Jiang --- backend/AGENTS.md | 39 +- .../read_before_write_middleware.py | 7 +- .../tool_error_handling_middleware.py | 41 +- .../middlewares/tool_progress_middleware.py | 578 +++++++ .../agents/middlewares/tool_result_meta.py | 212 +++ .../harness/deerflow/config/app_config.py | 2 + .../deerflow/config/tool_progress_config.py | 45 + .../test_read_before_write_middleware.py | 26 + .../test_tool_error_handling_middleware.py | 84 + .../tests/test_tool_progress_middleware.py | 1479 +++++++++++++++++ backend/tests/test_tool_result_meta.py | 474 ++++++ config.example.yaml | 27 +- 12 files changed, 2988 insertions(+), 26 deletions(-) create mode 100644 backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py create mode 100644 backend/packages/harness/deerflow/agents/middlewares/tool_result_meta.py create mode 100644 backend/packages/harness/deerflow/config/tool_progress_config.py create mode 100644 backend/tests/test_tool_progress_middleware.py create mode 100644 backend/tests/test_tool_result_meta.py diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 207dfd7d8..81fcd9df7 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -227,28 +227,29 @@ Lead-agent middlewares are assembled in strict order across three functions: the 7. **LLMErrorHandlingMiddleware** - Normalizes provider/model invocation failures into recoverable assistant-facing errors before later stages run 8. **GuardrailMiddleware** - *(optional, if `guardrails.enabled`)* Pre-tool-call authorization via pluggable `GuardrailProvider`; returns an error ToolMessage on deny. Providers: built-in `AllowlistProvider` (zero deps), OAP policy providers (e.g. `aport-agent-guardrails`), or custom. See [docs/GUARDRAILS.md](docs/GUARDRAILS.md) 9. **SandboxAuditMiddleware** - Audits sandboxed shell/file operations for security logging before tool execution -10. **ReadBeforeWriteMiddleware** - *(optional, if `read_before_write.enabled`, default on)* Version gate on file writes (issue #3857): `read_file` stamps a content hash onto its ToolMessage; `write_file` (append/overwrite-existing) and `str_replace` are blocked unless the newest mark for that path matches the file's current hash. Marks live on messages, so summarization dropping the read result invalidates the gate automatically; writes never refresh marks, forcing a re-read between consecutive edits. Gate check + tool execution are serialized per (thread, path) so same-turn parallel writes cannot reuse one stale mark; on sandboxes whose `read_file` reports failures as `"Error: ..."` strings instead of raising (AIO/E2B), uninspectable targets fail open (creation proceeds, no mark stamped) -11. **ToolErrorHandlingMiddleware** - Receives `AppConfig`, converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting, stamps structured metadata for task exception wrappers, and stamps skill-read metadata for downstream durable-context capture. Task tool result text is generated from the same status/result/error inputs as the structured metadata so callers do not hand-write a second protocol string. +10. **ReadBeforeWriteMiddleware** - *(optional, if `read_before_write.enabled`, default on)* Outermost write gate (issue #3857): `read_file` stamps a content hash onto its ToolMessage; `write_file` (append/overwrite-existing) and `str_replace` are blocked unless the newest mark for that path matches the file's current hash. Sits outside ToolProgressMiddleware and ToolErrorHandlingMiddleware so a blocked write returns immediately without consuming a ToolProgress slot. Blocked results call `normalize_tool_result` directly to stamp `deerflow_tool_meta` (`recoverable_by_model=True`) before returning, keeping the result well-formed for any outer consumer. Marks live on messages, so summarization dropping the read result invalidates the gate automatically; writes never refresh marks, forcing a re-read between consecutive edits. Gate check + tool execution are serialized per (thread, path) so same-turn parallel writes cannot reuse one stale mark; on sandboxes whose `read_file` reports failures as `"Error: ..."` strings instead of raising (AIO/E2B), uninspectable targets fail open (creation proceeds, no mark stamped) +11. **ToolProgressMiddleware** - *(optional, if `tool_progress.enabled`)* State-machine-based stagnation guard (RFC #3177). Outer wrapper around ToolErrorHandlingMiddleware so its `wrap_tool_call` receives results already stamped with `deerflow_tool_meta`. Tracks per-(thread, tool) consecutive "no-new-info" calls across three error categories: (a) `recoverable_by_model=True` (no_results, not_found, permission, Jaccard-duplicate success): ACTIVE → WARNED (terminal — hint re-injected on each subsequent problem); (b) `recoverable_by_model=False, action≠stop` (rate_limited, transient): ACTIVE → WARNED → BLOCKED after `warn_escalation_count` more problems; (c) `recoverable_by_model=False, action=stop` (auth, config, internal): immediately BLOCKED on first occurrence. **Division of labor with LoopDetectionMiddleware (item 25):** ToolProgressMiddleware is a result-quality guard — fires after tool execution and blocks specific tools that stop producing new information; LoopDetectionMiddleware is a call-pattern guard — fires after the model responds and hard-stops the whole turn when the model repeatedly issues identical tool_calls. Both can inject HumanMessage hints in the same model call without conflict; neither reads the other's internal state. +12. **ToolErrorHandlingMiddleware** - Receives `AppConfig`, converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting, stamps every result with `deerflow_tool_meta` (status / error_type / recoverable_by_model / recommended_next_action / source) via `tool_result_meta.normalize_tool_result`, stamps structured metadata for task exception wrappers, and stamps skill-read metadata for downstream durable-context capture. Task tool result text is generated from the same status/result/error inputs as the structured metadata so callers do not hand-write a second protocol string. **Lead-only middlewares** (`build_middlewares`, appended after the base): -12. **DynamicContextMiddleware** - Injects the current date (and optionally memory) as a `` into the first HumanMessage, keeping the base system prompt fully static for prefix-cache reuse -13. **SkillActivationMiddleware** - Detects strict `/skill-name task` syntax on the latest real user message, resolves only enabled and runtime-allowed skills, injects the `SKILL.md` body as hidden current-turn context, and records a `middleware:skill_activation` audit event -14. **DurableContextMiddleware** - Captures `task` delegations into `ThreadState.delegations` (including in-progress dispatches and terminal result summaries) and loaded skill-file references (name/path/description, parsed in-memory - not the body) into `ThreadState.skill_context` before summarization can compact the paired tool-call/result messages, then projects durable context into each model request. Static authority rules are injected as a `SystemMessage`; untrusted field values (`summary_text`, delegation results, skill descriptions) are injected separately as a hidden `HumanMessage` data block so compressed history, delegated work, and which skills are active stay visible without being stored as `messages` or promoted to system-role instructions. -15. **SummarizationMiddleware** - *(optional, if enabled)* Context reduction when approaching token limits -16. **TodoListMiddleware** - *(optional, if `is_plan_mode`)* Task tracking with the `write_todos` tool -17. **TokenUsageMiddleware** - *(optional, if `token_usage.enabled`)* Records token usage metrics; subagent usage is merged back into the dispatching AIMessage by message position -18. **TitleMiddleware** - Auto-generates the thread title after the first complete exchange and normalizes structured message content before prompting the title model. If a first-turn run is interrupted before this middleware can write a title, `runtime/runs/worker.py` keeps the run in a finalizing state, persists a local fallback title from the latest checkpoint or original run input, and then syncs it to `threads_meta.display_name`. Replacement runs admitted by `multitask_strategy="interrupt"` / `"rollback"` wait for older same-thread finalization before entering the graph; the interrupted run only skips the fallback title write once a later run has started and may have advanced the checkpoint. -19. **MemoryMiddleware** - Queues conversations for async memory update (filters to user + final AI responses) -20. **ViewImageMiddleware** - *(optional, if the model supports vision)* Injects base64 image data before the LLM call -21. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped) -22. **SystemMessageCoalescingMiddleware** - Merges every SystemMessage into a single leading SystemMessage per request; provider-agnostic fix for strict backends (vLLM/SGLang/Qwen/Anthropic) that reject non-leading system messages. Touches the per-request payload only (checkpoint state unchanged); on midnight crossings only the latest `dynamic_context_reminder` SystemMessage survives -23. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess `task` tool calls to enforce the `MAX_CONCURRENT_SUBAGENTS` limit -24. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer -25. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits -26. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the safety/clarification tail -27. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after custom middlewares so LangChain's reverse-order `after_model` dispatch runs it first -28. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, interrupts via `Command(goto=END)` (must be last) +13. **DynamicContextMiddleware** - Injects the current date (and optionally memory) as a `` into the first HumanMessage, keeping the base system prompt fully static for prefix-cache reuse +14. **SkillActivationMiddleware** - Detects strict `/skill-name task` syntax on the latest real user message, resolves only enabled and runtime-allowed skills, injects the `SKILL.md` body as hidden current-turn context, and records a `middleware:skill_activation` audit event +15. **DurableContextMiddleware** - Captures `task` delegations into `ThreadState.delegations` (including in-progress dispatches and terminal result summaries) and loaded skill-file references (name/path/description, parsed in-memory - not the body) into `ThreadState.skill_context` before summarization can compact the paired tool-call/result messages, then projects durable context into each model request. Static authority rules are injected as a `SystemMessage`; untrusted field values (`summary_text`, delegation results, skill descriptions) are injected separately as a hidden `HumanMessage` data block so compressed history, delegated work, and which skills are active stay visible without being stored as `messages` or promoted to system-role instructions. +16. **SummarizationMiddleware** - *(optional, if enabled)* Context reduction when approaching token limits +17. **TodoListMiddleware** - *(optional, if `is_plan_mode`)* Task tracking with the `write_todos` tool +18. **TokenUsageMiddleware** - *(optional, if `token_usage.enabled`)* Records token usage metrics; subagent usage is merged back into the dispatching AIMessage by message position +19. **TitleMiddleware** - Auto-generates the thread title after the first complete exchange and normalizes structured message content before prompting the title model. If a first-turn run is interrupted before this middleware can write a title, `runtime/runs/worker.py` keeps the run in a finalizing state, persists a local fallback title from the latest checkpoint or original run input, and then syncs it to `threads_meta.display_name`. Replacement runs admitted by `multitask_strategy="interrupt"` / `"rollback"` wait for older same-thread finalization before entering the graph; the interrupted run only skips the fallback title write once a later run has started and may have advanced the checkpoint. +20. **MemoryMiddleware** - Queues conversations for async memory update (filters to user + final AI responses) +21. **ViewImageMiddleware** - *(optional, if the model supports vision)* Injects base64 image data before the LLM call +22. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped) +23. **SystemMessageCoalescingMiddleware** - Merges every SystemMessage into a single leading SystemMessage per request; provider-agnostic fix for strict backends (vLLM/SGLang/Qwen/Anthropic) that reject non-leading system messages. Touches the per-request payload only (checkpoint state unchanged); on midnight crossings only the latest `dynamic_context_reminder` SystemMessage survives +24. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess `task` tool calls to enforce the `MAX_CONCURRENT_SUBAGENTS` limit +25. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer +26. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits +27. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the safety/clarification tail +28. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after custom middlewares so LangChain's reverse-order `after_model` dispatch runs it first +29. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, interrupts via `Command(goto=END)` (must be last) ### Configuration System diff --git a/backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py index ea75acdcd..1198d7577 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/read_before_write_middleware.py @@ -39,6 +39,7 @@ from langchain_core.messages import ToolMessage from langgraph.prebuilt.tool_node import ToolCallRequest from langgraph.types import Command +from deerflow.agents.middlewares.tool_result_meta import normalize_tool_result from deerflow.sandbox.tools import read_current_file_content logger = logging.getLogger(__name__) @@ -107,7 +108,9 @@ class ReadBeforeWriteMiddleware(AgentMiddleware): with self._lock_for(request, path): blocked = self._check_write_gate(request) if blocked is not None: - return blocked + # Stamp deerflow_tool_meta so ToolProgressMiddleware can classify + # the blocked write even though it bypasses ToolErrorHandlingMiddleware. + return normalize_tool_result(blocked) return handler(request) if name in _READ_TOOLS: path = self._requested_path(request) @@ -138,7 +141,7 @@ class ReadBeforeWriteMiddleware(AgentMiddleware): try: blocked = await asyncio.to_thread(self._check_write_gate, request) if blocked is not None: - return blocked + return normalize_tool_result(blocked) return await handler(request) finally: lock.release() diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py index d6adced3e..ee7bba16b 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py @@ -16,6 +16,10 @@ from deerflow.agents.middlewares.skill_context import ( _tool_call_path, build_skill_entry_metadata_from_read, ) +from deerflow.agents.middlewares.tool_result_meta import ( + normalize_tool_result, + stamp_exception_meta, +) from deerflow.config.app_config import AppConfig from deerflow.config.summarization_config import DEFAULT_SKILL_FILE_READ_TOOL_NAMES from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH @@ -79,7 +83,8 @@ class ToolErrorHandlingMiddleware(AgentMiddleware[AgentState]): # failures raised before task_tool can build its own Command still # carry the same structured metadata. structured_error = f"{exc.__class__.__name__}: {detail}" - return _stamp_task_exception_status(message, tool_name=tool_name, error=structured_error) + message = _stamp_task_exception_status(message, tool_name=tool_name, error=structured_error) + return stamp_exception_meta(message, structured_error) def _stamp_skill_read_metadata( self, @@ -127,7 +132,7 @@ class ToolErrorHandlingMiddleware(AgentMiddleware[AgentState]): except Exception as exc: logger.exception("Tool execution failed (sync): name=%s id=%s", request.tool_call.get("name"), request.tool_call.get("id")) return self._build_error_message(request, exc) - return self._maybe_stamp(result, request) + return normalize_tool_result(self._maybe_stamp(result, request)) @override async def awrap_tool_call( @@ -143,7 +148,7 @@ class ToolErrorHandlingMiddleware(AgentMiddleware[AgentState]): except Exception as exc: logger.exception("Tool execution failed (async): name=%s id=%s", request.tool_call.get("name"), request.tool_call.get("id")) return self._build_error_message(request, exc) - return self._maybe_stamp(result, request) + return normalize_tool_result(self._maybe_stamp(result, request)) def _build_runtime_middlewares( @@ -213,14 +218,42 @@ def _build_runtime_middlewares( tail.append(SandboxAuditMiddleware()) + # ReadBeforeWriteMiddleware is the outermost write gate: it blocks writes to files + # the model hasn't read in their current version. It must sit outside ToolProgress + # and ToolErrorHandling so that a blocked write returns immediately without consuming + # a ToolProgress slot. The middleware stamps deerflow_tool_meta on the blocked + # ToolMessage itself so downstream callers receive a well-formed result. if app_config.read_before_write.enabled: from deerflow.agents.middlewares.read_before_write_middleware import ReadBeforeWriteMiddleware tail.append(ReadBeforeWriteMiddleware()) + # ToolProgressMiddleware must be outer (lower index) so its wrap_tool_call handler + # chain includes ToolErrorHandlingMiddleware (inner), which stamps deerflow_tool_meta + # on every result before ToolProgressMiddleware reads it in _update_state_from_result. + # Framework rule: first in list = outermost (types.py: "compose with first in list as outermost layer"). + tool_progress_config = app_config.tool_progress + _ToolProgressMiddleware = None + if tool_progress_config.enabled: + from deerflow.agents.middlewares.tool_progress_middleware import ToolProgressMiddleware as _ToolProgressMiddleware + + tail.append(_ToolProgressMiddleware.from_config(tool_progress_config)) + tail.append(ToolErrorHandlingMiddleware(app_config=app_config)) - return [*outer_wrappers, *thread_hooks, *tail] + middlewares = [*outer_wrappers, *thread_hooks, *tail] + + # Guard: ToolProgressMiddleware (outer) must appear before ToolErrorHandlingMiddleware (inner) + # so that its wrap_tool_call chain encloses the stamping step. Fail loudly at build time + # rather than silently no-oping at runtime if a future insertion reverses the order. + # Uses isinstance (not type().__name__) so subclasses and renames are covered. + if _ToolProgressMiddleware is not None: + _progress_idx = next((i for i, m in enumerate(middlewares) if isinstance(m, _ToolProgressMiddleware)), None) + _error_idx = next((i for i, m in enumerate(middlewares) if isinstance(m, ToolErrorHandlingMiddleware)), None) + if _progress_idx is not None and _error_idx is not None and _progress_idx > _error_idx: + raise RuntimeError(f"ToolProgressMiddleware must be outer (index {_progress_idx}) of ToolErrorHandlingMiddleware (index {_error_idx}) — check middleware append order") + + return middlewares def build_lead_runtime_middlewares(*, app_config: AppConfig, lazy_init: bool = True) -> list[AgentMiddleware]: diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py new file mode 100644 index 000000000..ce63e1a73 --- /dev/null +++ b/backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py @@ -0,0 +1,578 @@ +"""Middleware for task-level tool call progress tracking with a state machine. + +Implements RFC #3177: structured tool result signals drive a per-(thread, tool) +state machine that detects stagnation and repetition, injects hints early +(WARNED), and hard-blocks the tool when it has stopped producing value (BLOCKED). + +Architecture: + ToolProgressMiddleware (outer) + └── handler → ToolErrorHandlingMiddleware (inner) → actual tool + ↓ + ToolProgressMiddleware reads deerflow_tool_meta from the normalized result + +State machine transitions per (thread_id, tool_name): + ACTIVE → WARNED (at stagnation_threshold problems) + Any problem-free call resets consecutive_problems=0 and reverts to ACTIVE. + + Whether WARNED can escalate to BLOCKED depends on recoverable_by_model: + - recoverable_by_model=True (no_results, not_found, permission, Jaccard-duplicate success): + WARNED is terminal. The model received a hint and is expected to change strategy; + blocking would prevent a legitimate retry with different parameters. + - recoverable_by_model=False, action≠stop (transient, rate_limited): + WARNED → BLOCKED after warn_escalation_count more problems. The model cannot fix + these by retrying the same tool, so hard-blocking conserves API calls. + - recoverable_by_model=False, action=stop (auth, config, internal): + Immediately BLOCKED on the first occurrence — no retry can help. + +Division of labor with LoopDetectionMiddleware (middleware position 23): + ToolProgressMiddleware (position 10) is a result-quality guard — it fires + after a tool executes, inspects what came back, and blocks *specific tools* + that have stopped producing new information. + + LoopDetectionMiddleware is a call-pattern guard — it fires after the model + responds (before tools execute), inspects the tool_calls signature in the + AIMessage, and forces the *whole turn* to stop when the model keeps issuing + the same calls regardless of results. + + They are complementary, not competing: + - ToolProgressMiddleware is fine-grained (per-tool BLOCK, other tools normal). + - LoopDetectionMiddleware is coarse-grained (strips all tool_calls, ends turn). + - Both can inject HumanMessage hints in the same model call without conflict; + the model sees both sets of hints and can reason about them. + - If LoopDetectionMiddleware hard-stops (strips tool_calls), no wrap_tool_call + is issued so ToolProgressMiddleware never fires — there is no double-stop. + - If ToolProgressMiddleware BLOCKs a tool (returns an error ToolMessage), + the model still makes a tool call that LoopDetectionMiddleware tracks; both + continue to operate on their own independent state. +""" + +from __future__ import annotations + +import logging +import re +import threading +from collections import OrderedDict, defaultdict +from collections.abc import Awaitable, Callable, Sequence +from dataclasses import dataclass, field, replace +from typing import TYPE_CHECKING, Literal, override + +from langchain.agents import AgentState +from langchain.agents.middleware import AgentMiddleware +from langchain.agents.middleware.types import ModelCallResult, ModelRequest, ModelResponse +from langchain_core.messages import HumanMessage, ToolMessage +from langgraph.prebuilt.tool_node import ToolCallRequest +from langgraph.runtime import Runtime +from langgraph.types import Command + +from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY, ToolResultMeta + +if TYPE_CHECKING: + from deerflow.config.tool_progress_config import ToolProgressConfig + +logger = logging.getLogger(__name__) + +_MAX_PENDING_PER_RUN = 3 +# Jaccard word-set computation is capped to avoid O(n) regex work on very large tool results. +_MAX_CONTENT_FOR_WORDSET = 8192 + + +# --------------------------------------------------------------------------- +# State data structures + + +@dataclass(slots=True) +class ToolPhaseState: + """Per (thread_id, tool_name) tracking state.""" + + phase: Literal["active", "warned", "blocked"] = "active" + consecutive_problems: int = 0 + block_reason: str | None = None + # Immutable tuple so that dataclasses.replace() calls that omit recent_word_sets + # (problem paths) cannot accidentally share a mutable list between the old and new + # state objects and cause silent cross-state corruption via .append(). + recent_word_sets: tuple[frozenset[str], ...] = field(default_factory=tuple) + + +# --------------------------------------------------------------------------- +# Content helpers + + +def word_set(content: str) -> frozenset[str]: + """Extract lowercase words of length >= 3 for Jaccard similarity. + + Content is capped at _MAX_CONTENT_FOR_WORDSET chars to bound memory and CPU cost on + large tool results (e.g. web pages). Tail content beyond the cap is omitted from the + set, which is acceptable because duplicate-detection is a heuristic, not a guarantee. + """ + return frozenset(re.findall(r"\b\w{3,}\b", content[:_MAX_CONTENT_FOR_WORDSET].lower())) + + +def is_near_duplicate( + current: frozenset[str], + recent: Sequence[frozenset[str]], + threshold: float, + min_words: int, +) -> bool: + """Return True if current is similar to any of the last 3 recent word sets.""" + if len(current) < min_words: + return False + for prev in recent[-3:]: + if len(prev) < min_words: + continue + union = len(current | prev) + if union == 0: + continue + if len(current & prev) / union >= threshold: + return True + return False + + +def _message_content_str(msg: ToolMessage) -> str: + return msg.content if isinstance(msg.content, str) else "" + + +def _parse_tool_meta(meta_dict: object) -> ToolResultMeta | None: + """Safely deserialize a ToolResultMeta from a raw dict; returns None on schema mismatch.""" + if not isinstance(meta_dict, dict): + return None + try: + return ToolResultMeta(**meta_dict) + except TypeError: + logger.warning("Unexpected tool meta schema, skipping progress tracking: %s", meta_dict) + return None + + +# --------------------------------------------------------------------------- +# Hint / block reason formatting + + +def _format_hint(meta: ToolResultMeta) -> str: + action_map = { + "rewrite_query": "Try rephrasing your search query with different keywords or approach.", + "try_alternative": "Consider using a different tool or strategy.", + "summarize": "Consider summarizing your current findings and moving forward.", + "stop": "Do not retry this operation — it is not recoverable.", + # Near-duplicate success results: recommended_next_action is "continue" by default, + # but the model should still change strategy to avoid re-fetching the same content. + "continue": "Try rephrasing your query or using a different search term.", + } + base = { + "no_results": "[PROGRESS HINT] Your search returned no results.", + "not_found": "[PROGRESS HINT] The resource was not found repeatedly.", + "rate_limited": "[PROGRESS HINT] The tool is being rate-limited.", + "transient": "[PROGRESS HINT] The tool encountered repeated transient failures.", + "partial_success": "[PROGRESS HINT] The tool has returned incomplete results multiple times.", + # Jaccard near-duplicate success: the tool is returning the same content repeatedly. + "success": "[PROGRESS HINT] The tool is returning duplicate results.", + }.get( + meta.error_type or meta.status, + "[PROGRESS HINT] The tool is not producing new information.", + ) + suffix = action_map.get(meta.recommended_next_action, "") + return f"{base} {suffix}".strip() + + +def _block_reason(meta: ToolResultMeta) -> str: + return { + "no_results": "Repeated no-results — rewrite your query or try a different tool.", + "not_found": "Repeated not-found — rewrite your query or try a different resource.", + "rate_limited": "Repeated rate-limiting — summarize current findings and proceed.", + "transient": "Repeated transient failures — try a different approach.", + "auth": "Authentication failure — this tool cannot be used.", + "config": "Tool is not configured — this tool cannot be used.", + "internal": "Repeated internal errors — this tool is unavailable.", + }.get( + meta.error_type or "", + "Tool has not produced new information after multiple attempts — summarize and move on.", + ) + + +# --------------------------------------------------------------------------- +# Middleware + + +class ToolProgressMiddleware(AgentMiddleware[AgentState]): + """State-machine-based tool stagnation guard (RFC #3177).""" + + def __init__( + self, + *, + stagnation_threshold: int = 3, + warn_escalation_count: int = 2, + inject_assessment: bool = True, + jaccard_threshold: float = 0.8, + min_words: int = 10, + exempt_tools: set[str] | None = None, + max_tracked_threads: int = 100, + ) -> None: + self._stagnation_threshold = stagnation_threshold + self._warn_escalation = warn_escalation_count + self._inject_assessment = inject_assessment + self._jaccard_threshold = jaccard_threshold + self._min_words = min_words + self._exempt_tools: set[str] = exempt_tools if exempt_tools is not None else {"ask_clarification", "write_todos", "present_files", "task"} + self._max_tracked_threads = max_tracked_threads + + # threading.Lock (not asyncio.Lock): critical sections are short in-memory dict + # ops with no I/O, so event-loop stall risk is negligible. asyncio.Lock would + # not protect the sync wrap_tool_call path used by subagent executor thread + # pools — two separate locks would be required instead. This matches the + # existing LoopDetectionMiddleware pattern; see module docstring for details. + self._lock = threading.Lock() + # LRU-evicting store: thread_id → {tool_name → ToolPhaseState} + self._phase_states: OrderedDict[str, dict[str, ToolPhaseState]] = OrderedDict() + # Pending hint queue: (thread_id, run_id) → [hint texts] + self._pending: dict[tuple[str, str], list[str]] = defaultdict(list) + + @classmethod + def from_config(cls, config: ToolProgressConfig) -> ToolProgressMiddleware: + return cls( + stagnation_threshold=config.stagnation_threshold, + warn_escalation_count=config.warn_escalation_count, + inject_assessment=config.inject_assessment, + jaccard_threshold=config.jaccard_similarity_threshold, + min_words=config.min_word_count_for_similarity, + exempt_tools=set(config.exempt_tools), + max_tracked_threads=config.max_tracked_threads, + ) + + # ------------------------------------------------------------------ + # Runtime helpers + + @staticmethod + def _thread_id(runtime: Runtime) -> str: + tid = runtime.context.get("thread_id") if runtime.context else None + return str(tid) if tid else "default" + + @staticmethod + def _run_id(runtime: Runtime) -> str: + rid = runtime.context.get("run_id") if runtime.context else None + return str(rid) if rid else "default" + + def _pending_key(self, runtime: Runtime) -> tuple[str, str]: + return self._thread_id(runtime), self._run_id(runtime) + + # ------------------------------------------------------------------ + # State store (caller holds lock) + + def _get_state(self, thread_id: str, tool_name: str) -> ToolPhaseState: + if thread_id not in self._phase_states: + self._phase_states[thread_id] = {} + while len(self._phase_states) > self._max_tracked_threads: + evicted_thread, _ = self._phase_states.popitem(last=False) + # Evict pending hints for the evicted thread to prevent unbounded growth. + for key in [k for k in self._pending if k[0] == evicted_thread]: + del self._pending[key] + self._phase_states.move_to_end(thread_id) + return self._phase_states[thread_id].get(tool_name, ToolPhaseState()) + + def _set_state(self, thread_id: str, tool_name: str, state: ToolPhaseState) -> None: + self._phase_states[thread_id][tool_name] = state + + def _get_block_reason(self, runtime: Runtime, tool_name: str) -> str | None: + thread_id = self._thread_id(runtime) + with self._lock: + thread_tools = self._phase_states.get(thread_id) + if thread_tools is None: + return None + # Read-only check: do NOT call move_to_end here. Bumping recency on the read path + # would keep blocked threads permanently warm in the LRU, preventing healthy active + # threads from occupying those slots. Recency is updated only on _get_state writes. + tool_state = thread_tools.get(tool_name) + return tool_state.block_reason if tool_state is not None and tool_state.phase == "blocked" else None + + def _make_blocked_message(self, request: ToolCallRequest, tool_name: str, block_reason: str) -> ToolMessage: + return ToolMessage( + content=f"[TOOL_BLOCKED] {block_reason}", + tool_call_id=str(request.tool_call.get("id", "")), + name=tool_name, + status="error", + additional_kwargs={ + TOOL_META_KEY: { + "status": "error", + "error_type": "blocked_by_progress_guard", + "recoverable_by_model": True, + "recommended_next_action": "summarize", + "source": "progress_middleware", + } + }, + ) + + def _update_state_from_result( + self, + result: ToolMessage | Command, + tool_name: str, + runtime: Runtime, + ) -> ToolMessage | Command: + """Update the state machine from a tool result; queue hints if warranted.""" + if not isinstance(result, ToolMessage): + return result + meta = _parse_tool_meta((result.additional_kwargs or {}).get(TOOL_META_KEY)) + if meta is None: + if tool_name not in self._exempt_tools: + logger.warning( + "tool_progress: deerflow_tool_meta missing for non-exempt tool %s — verify ToolProgressMiddleware is outer of ToolErrorHandlingMiddleware", + tool_name, + ) + return result + content = _message_content_str(result) + thread_id = self._thread_id(runtime) + with self._lock: + state = self._get_state(thread_id, tool_name) + new_state, hint = self._assess_and_transition(state, meta, content) + self._set_state(thread_id, tool_name, new_state) + if new_state.phase != state.phase: + if new_state.phase == "blocked": + logger.warning( + "tool_progress: %s/%s -> BLOCKED: %s", + thread_id, + tool_name, + new_state.block_reason, + ) + elif new_state.phase == "warned": + logger.info( + "tool_progress: %s/%s -> WARNED (consecutive_problems=%d)", + thread_id, + tool_name, + new_state.consecutive_problems, + ) + elif new_state.phase == "active": + logger.info( + "tool_progress: %s/%s -> ACTIVE (reset after good result)", + thread_id, + tool_name, + ) + if hint and self._inject_assessment: + self._queue_assessment(runtime, hint) + return result + + # ------------------------------------------------------------------ + # State machine + + def _assess_and_transition( + self, + state: ToolPhaseState, + meta: ToolResultMeta, + content: str, + ) -> tuple[ToolPhaseState, str | None]: + """Return (new_state, hint_text_or_None). + + The outer wrap_tool_call gate intercepts already-blocked states before + the handler is called, so this function is normally reached only for + active/warned states. If a blocked state arrives (e.g., concurrent + transition), the function returns it unchanged — no counter inflation, + no phase regression. + """ + # Guard: blocked is a terminal state; nothing should change it here. + # (In normal flow this branch is unreachable because wrap_tool_call + # intercepts blocked tools before calling the handler. The check exists + # to make concurrent-race semantics well-defined and prevent a + # recoverable-error result from silently demoting the phase back to warned.) + if state.phase == "blocked": + return state, None + + # Count this call as a problem before branching so all exit paths leave + # consecutive_problems in a consistent state (never 0 when the tool has failed). + new_count = state.consecutive_problems + 1 + + # Immediately block on unrecoverable stop signals (auth, config, internal). + if not meta.recoverable_by_model and meta.recommended_next_action == "stop": + return replace( + state, + phase="blocked", + consecutive_problems=new_count, + block_reason=_block_reason(meta), + ), None + + # Compute word_set only for success results: error/partial_success are problems by + # definition and never reach the Jaccard check, so the O(n) regex is wasted on them. + ws = word_set(content) if meta.status == "success" else frozenset() + is_problem = meta.status in ("error", "partial_success") or (meta.status == "success" and is_near_duplicate(ws, state.recent_word_sets, self._jaccard_threshold, self._min_words)) + + if not is_problem: + # Good result: reset consecutive count, return to active. + new_recent = (*state.recent_word_sets, ws)[-3:] + return replace(state, consecutive_problems=0, phase="active", recent_word_sets=new_recent), None + + hint: str | None = None + + if new_count >= self._stagnation_threshold + self._warn_escalation: + if meta.recoverable_by_model: + # Model can fix this by changing strategy — keep warned, re-inject hint. + # BLOCKED would prevent a legitimate retry with different parameters. + hint = _format_hint(meta) + new_state = replace(state, consecutive_problems=new_count, phase="warned") + else: + # Model cannot fix this by retrying — block the tool. + reason = _block_reason(meta) + new_state = replace(state, consecutive_problems=new_count, phase="blocked", block_reason=reason) + elif new_count >= self._stagnation_threshold: + hint = _format_hint(meta) + new_state = replace(state, consecutive_problems=new_count, phase="warned") + else: + new_state = replace(state, consecutive_problems=new_count) + + return new_state, hint + + # ------------------------------------------------------------------ + # Pending queue helpers + + def _queue_assessment(self, runtime: Runtime, text: str) -> None: + key = self._pending_key(runtime) + thread_id = key[0] + with self._lock: + # 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. + if thread_id not in self._phase_states: + return + queue = self._pending[key] + if len(queue) < _MAX_PENDING_PER_RUN: + queue.append(text) + + def _drain_pending(self, runtime: Runtime) -> list[str]: + key = self._pending_key(runtime) + with self._lock: + return self._pending.pop(key, []) + + def _clear_stale_pending(self, runtime: Runtime) -> None: + thread_id, current_run = self._pending_key(runtime) + with self._lock: + for key in list(self._pending): + if key[0] == thread_id and key[1] != current_run: + del self._pending[key] + + def _reset_run_states(self, runtime: Runtime) -> None: + """Reset all per-run tool state for the thread at the start of a new agent run. + + Every tool's consecutive_problems counter and recent_word_sets Jaccard window are + cleared unconditionally so state from a previous run never bleeds into the next: + - BLOCKED/WARNED tools are reset to ACTIVE (they re-block immediately if the root + cause persists, and the model has no memory of the prior-run hint). + - ACTIVE tools with non-zero consecutive_problems or non-empty recent_word_sets from + the previous run are also cleared so a single first-call problem in the new run + cannot falsely trip WARNED against stale context from a run the model no longer sees. + + **Cross-run scoping vs LoopDetectionMiddleware**: this per-run reset is an intentional + policy choice, not an oversight. Errors like ``rate_limited`` and ``transient`` are + time-bound: their root cause may resolve between user turns, so carrying a stale + counter forward risks a false-positive BLOCKED on calls that would now succeed. + LoopDetectionMiddleware takes the opposite stance — it retains ``_history`` across + runs (only clearing other-run *pending* warnings at ``before_agent``), because + call-pattern loops are time-invariant: a model that keeps issuing the same tool_calls + regardless of results does so regardless of when the run started. The two middlewares + therefore guard different failure modes (result quality vs. call pattern) and their + cross-run scoping policies intentionally differ as a consequence. + """ + thread_id = self._thread_id(runtime) + with self._lock: + thread_tools = self._phase_states.get(thread_id) + if thread_tools is None: + return + for tool_name, tool_state in list(thread_tools.items()): + thread_tools[tool_name] = replace( + tool_state, + phase="active", + consecutive_problems=0, + block_reason=None, + recent_word_sets=(), + ) + + # ------------------------------------------------------------------ + # wrap_tool_call + + @override + def wrap_tool_call( + self, + request: ToolCallRequest, + handler: Callable[[ToolCallRequest], ToolMessage | Command], + ) -> ToolMessage | Command: + tool_name = str(request.tool_call.get("name", "")) + if not tool_name or tool_name in self._exempt_tools: + return handler(request) + runtime = getattr(request, "runtime", None) + if runtime is None: + return handler(request) + block_reason = self._get_block_reason(runtime, tool_name) + if block_reason: + logger.info( + "tool_progress: %s/%s call intercepted (blocked): %s", + self._thread_id(runtime), + tool_name, + block_reason, + ) + return self._make_blocked_message(request, tool_name, block_reason) + return self._update_state_from_result(handler(request), tool_name, runtime) + + @override + async def awrap_tool_call( + self, + request: ToolCallRequest, + handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]], + ) -> ToolMessage | Command: + tool_name = str(request.tool_call.get("name", "")) + if not tool_name or tool_name in self._exempt_tools: + return await handler(request) + runtime = getattr(request, "runtime", None) + if runtime is None: + return await handler(request) + block_reason = self._get_block_reason(runtime, tool_name) + if block_reason: + logger.info( + "tool_progress: %s/%s call intercepted (blocked): %s", + self._thread_id(runtime), + tool_name, + block_reason, + ) + return self._make_blocked_message(request, tool_name, block_reason) + return self._update_state_from_result(await handler(request), tool_name, runtime) + + # ------------------------------------------------------------------ + # wrap_model_call: drain pending hints and inject before model sees messages + + def _augment_request(self, request: ModelRequest) -> ModelRequest: + hints = self._drain_pending(request.runtime) + if not hints: + return request + deduped = list(dict.fromkeys(hints)) + logger.debug( + "tool_progress: injecting %d hint(s) for %s/%s", + len(deduped), + *self._pending_key(request.runtime), + ) + new_messages = [ + *request.messages, + HumanMessage(content="\n\n".join(deduped), name="progress_hint"), + ] + return request.override(messages=new_messages) + + @override + def wrap_model_call( + self, + request: ModelRequest, + handler: Callable[[ModelRequest], ModelResponse], + ) -> ModelCallResult: + return handler(self._augment_request(request)) + + @override + async def awrap_model_call( + self, + request: ModelRequest, + handler: Callable[[ModelRequest], Awaitable[ModelResponse]], + ) -> ModelCallResult: + return await handler(self._augment_request(request)) + + # ------------------------------------------------------------------ + # before_agent: clean up stale pending hints from previous runs + + @override + def before_agent(self, state: AgentState, runtime: Runtime) -> dict | None: + self._clear_stale_pending(runtime) + self._reset_run_states(runtime) + return None + + @override + async def abefore_agent(self, state: AgentState, runtime: Runtime) -> dict | None: + self._clear_stale_pending(runtime) + self._reset_run_states(runtime) + return None diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_result_meta.py b/backend/packages/harness/deerflow/agents/middlewares/tool_result_meta.py new file mode 100644 index 000000000..dd31cb5e4 --- /dev/null +++ b/backend/packages/harness/deerflow/agents/middlewares/tool_result_meta.py @@ -0,0 +1,212 @@ +"""Unified tool result semantics for structured signal production. + +Every tool result that passes through ToolErrorHandlingMiddleware gets a +``deerflow_tool_meta`` entry in additional_kwargs. Downstream consumers +(ToolProgressMiddleware, etc.) read this key instead of parsing text. +""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass +from typing import Literal + +from langchain_core.messages import ToolMessage +from langgraph.types import Command + +TOOL_META_KEY = "deerflow_tool_meta" + +_ERROR_PREFIX = "Error:" +_PARTIAL_MARKERS = ( + "partial results", + "limited results", + "truncated", + "results may be incomplete", + # Tools that return status="success" with a no-results body (instead of status="error") + # must still be caught by stagnation detection so the model is prompted to try a different query. + "no results found", + "no content found", + "no images found", +) + + +@dataclass(frozen=True, slots=True) +class ToolResultMeta: + status: Literal["success", "error", "partial_success"] + error_type: str | None + recoverable_by_model: bool + recommended_next_action: Literal["continue", "rewrite_query", "try_alternative", "summarize", "stop"] + source: Literal["exception", "tool_return", "content_analysis", "progress_middleware"] + + +_ERROR_RULES: list[tuple[list[str], dict[str, object]]] = [ + ( + ["401", "403", "unauthorized", "authentication", "invalid api key"], + {"error_type": "auth", "recoverable_by_model": False, "recommended_next_action": "stop"}, + ), + ( + ["rate limit", "rate limited", "rate_limit"], + {"error_type": "rate_limited", "recoverable_by_model": False, "recommended_next_action": "summarize"}, + ), + ( + ["timeout", "timed out", "connection", "network error", "temporarily unavailable"], + {"error_type": "transient", "recoverable_by_model": False, "recommended_next_action": "try_alternative"}, + ), + ( + ["not configured", "not installed", "missing required", "disabled", "no api key"], + {"error_type": "config", "recoverable_by_model": False, "recommended_next_action": "stop"}, + ), + ( + ["permission denied", "access denied", "path traversal", "forbidden"], + {"error_type": "permission", "recoverable_by_model": True, "recommended_next_action": "try_alternative"}, + ), + ( + ["no results found", "no content found", "no images found", "no results"], + {"error_type": "no_results", "recoverable_by_model": True, "recommended_next_action": "rewrite_query"}, + ), + ( + ["not found", "no such file", "does not exist", "404"], + {"error_type": "not_found", "recoverable_by_model": True, "recommended_next_action": "rewrite_query"}, + ), + ( + ["unexpected error", "internal error", "500"], + {"error_type": "internal", "recoverable_by_model": False, "recommended_next_action": "stop"}, + ), +] + +_UNKNOWN_ERROR: dict[str, object] = { + "error_type": "unknown", + "recoverable_by_model": True, + "recommended_next_action": "try_alternative", +} + +# Pre-compiled at module load from _ERROR_RULES. Anchoring bare numeric codes (401, 403, 404, +# 500) to word boundaries prevents substring hits on unrelated numbers like "took 500ms". +# Computed here (after _ERROR_RULES) so the set is authoritative and thread-safe — no lazy +# writes on the hot classification path. +_NUMERIC_KW_RE: dict[str, re.Pattern[str]] = {kw: re.compile(rf"\b{kw}\b") for rule_keywords, _ in _ERROR_RULES for kw in rule_keywords if kw.isdigit()} + +_SEMANTIC_ZERO_ERROR_STRINGS: frozenset[str] = frozenset({"none", "null", "false", "no", "ok", "success", "n/a", ""}) + + +def _extract_json_error_text(content: str) -> str | None: + """Return the error string from a JSON-wrapped error like {"error": "...", "query": "..."}. + + Returns None when the ``error`` field is falsy (JSON null / 0 / false / empty + string) or is a sentinel string that conventionally means "no error" (e.g. + ``"none"``, ``"null"``, ``"false"``). This prevents tools that return + ``{"error": "none", "results": [...]}`` on success from being misclassified + as errors. + """ + try: + data = json.loads(content) + except (json.JSONDecodeError, ValueError): + return None + error = data.get("error") if isinstance(data, dict) else None + if not error: + return None + if isinstance(error, str) and error.lower().strip() in _SEMANTIC_ZERO_ERROR_STRINGS: + return None + # Serialize non-string values to JSON so _classify_error_text sees a predictable + # format (e.g. {"error": 404} → "404", {"error": [...]} → "[...]") instead of + # Python repr which can spuriously match keyword rules like "missing required". + return error if isinstance(error, str) else json.dumps(error) + + +def _match_keyword(kw: str, lower: str) -> bool: + """Match a keyword against lowercased text, using word boundaries for numeric codes.""" + if kw.isdigit(): + return bool(_NUMERIC_KW_RE[kw].search(lower)) + return kw in lower + + +def _classify_error_text(text: str) -> dict[str, object]: + lower = text.lower() + for keywords, attrs in _ERROR_RULES: + if any(_match_keyword(kw, lower) for kw in keywords): + return {**attrs} + return {**_UNKNOWN_ERROR} + + +def _make_meta(*, status: str, source: str, error_type: str | None = None, recoverable_by_model: bool = True, recommended_next_action: str = "continue") -> dict[str, object]: + return { + "status": status, + "error_type": error_type, + "recoverable_by_model": recoverable_by_model, + "recommended_next_action": recommended_next_action, + "source": source, + } + + +def stamp_exception_meta(msg: ToolMessage, exc_info: str) -> ToolMessage: + """Stamp deerflow_tool_meta with source='exception' onto an exception-derived ToolMessage. + + Unlike normalize_tool_message (which preserves existing stamps), this function always + overwrites any pre-existing TOOL_META_KEY entry. Exception-derived classification is + more authoritative than a tool's own return-time stamp. + """ + attrs = _classify_error_text(exc_info) + updated_kwargs = dict(msg.additional_kwargs or {}) + updated_kwargs[TOOL_META_KEY] = _make_meta(status="error", source="exception", **attrs) + msg.additional_kwargs = updated_kwargs + return msg + + +def normalize_tool_message(msg: ToolMessage) -> ToolMessage: + """Attach deerflow_tool_meta to a ToolMessage if not already present.""" + existing = (msg.additional_kwargs or {}).get(TOOL_META_KEY) + if existing is not None: + return msg + + content = msg.content if isinstance(msg.content, str) else "" + # Pre-compute once; reused by the partial-success marker check below to avoid calling + # content.lower() once per _PARTIAL_MARKERS entry inside the generator. + content_lower = content.lower() + + # Non-standard error: tool returned status="error" without the "Error:" prefix convention. + # (Actual exceptions from ToolErrorHandlingMiddleware are pre-stamped by stamp_exception_meta + # and exit early above — they never reach this branch.) + # Try JSON extraction first so classification uses only the "error" field value, not + # keywords that appear incidentally in other JSON fields (e.g. "query"). + if msg.status == "error" and not content.startswith(_ERROR_PREFIX): + json_error = _extract_json_error_text(content) + if json_error is not None: + attrs = _classify_error_text(json_error) + else: + # Determine whether content is a JSON object that simply has no 'error' key. + # If so, do NOT classify from the raw JSON string — incidental field values + # (e.g. {"user_id": 401}) would spuriously match keyword rules and hard-block + # the tool. Classify raw text only when the content is not valid JSON. + try: + is_json_dict = isinstance(json.loads(content), dict) + except (json.JSONDecodeError, ValueError): + is_json_dict = False + attrs = {**_UNKNOWN_ERROR} if is_json_dict else _classify_error_text(content) + meta = _make_meta(status="error", source="tool_return", **attrs) + elif content.startswith(_ERROR_PREFIX): + attrs = _classify_error_text(content[len(_ERROR_PREFIX) :]) + meta = _make_meta(status="error", source="tool_return", **attrs) + elif (json_error := _extract_json_error_text(content)) is not None: + attrs = _classify_error_text(json_error) + meta = _make_meta(status="error", source="tool_return", **attrs) + elif any(m in content_lower for m in _PARTIAL_MARKERS): + meta = _make_meta( + status="partial_success", + source="content_analysis", + recommended_next_action="rewrite_query", + ) + else: + meta = _make_meta(status="success", source="content_analysis") + + updated_kwargs = dict(msg.additional_kwargs or {}) + updated_kwargs[TOOL_META_KEY] = meta + msg.additional_kwargs = updated_kwargs + return msg + + +def normalize_tool_result(result: ToolMessage | Command) -> ToolMessage | Command: + """Normalize a tool result, handling Command wrappers transparently.""" + if isinstance(result, ToolMessage): + return normalize_tool_message(result) + return result diff --git a/backend/packages/harness/deerflow/config/app_config.py b/backend/packages/harness/deerflow/config/app_config.py index 217d1df2f..548a01d30 100644 --- a/backend/packages/harness/deerflow/config/app_config.py +++ b/backend/packages/harness/deerflow/config/app_config.py @@ -39,6 +39,7 @@ from deerflow.config.token_budget_config import TokenBudgetConfig from deerflow.config.token_usage_config import TokenUsageConfig from deerflow.config.tool_config import ToolConfig, ToolGroupConfig from deerflow.config.tool_output_config import ToolOutputConfig +from deerflow.config.tool_progress_config import ToolProgressConfig from deerflow.config.tool_search_config import ToolSearchConfig, load_tool_search_config_from_dict load_dotenv() @@ -174,6 +175,7 @@ class AppConfig(BaseModel): ), ) loop_detection: LoopDetectionConfig = Field(default_factory=LoopDetectionConfig, description="Loop detection middleware configuration") + tool_progress: ToolProgressConfig = Field(default_factory=ToolProgressConfig, description="Tool progress state machine middleware configuration") read_before_write: ReadBeforeWriteConfig = Field(default_factory=ReadBeforeWriteConfig, description="Read-before-write file gate middleware configuration") safety_finish_reason: SafetyFinishReasonConfig = Field(default_factory=SafetyFinishReasonConfig, description="Provider safety-filter finish_reason interception middleware configuration") auth: AuthAppConfig = Field(default_factory=AuthAppConfig, description="Authentication configuration (local + OIDC SSO)") diff --git a/backend/packages/harness/deerflow/config/tool_progress_config.py b/backend/packages/harness/deerflow/config/tool_progress_config.py new file mode 100644 index 000000000..47d1f9094 --- /dev/null +++ b/backend/packages/harness/deerflow/config/tool_progress_config.py @@ -0,0 +1,45 @@ +"""Configuration for tool progress tracking middleware.""" + +from pydantic import BaseModel, Field + + +class ToolProgressConfig(BaseModel): + """Configuration for task-level tool call progress tracking.""" + + enabled: bool = Field( + default=False, + description="Whether to enable tool progress tracking middleware", + ) + stagnation_threshold: int = Field( + default=3, + ge=1, + description="Number of consecutive problem calls before injecting a warning hint", + ) + warn_escalation_count: int = Field( + default=2, + ge=1, + description="Additional problem occurrences after WARNED before escalating to BLOCKED", + ) + inject_assessment: bool = Field( + default=True, + description="Whether to inject progress assessment hints into model requests", + ) + jaccard_similarity_threshold: float = Field( + default=0.8, + ge=0.0, + le=1.0, + description="Word-set Jaccard similarity threshold for near-duplicate result detection", + ) + min_word_count_for_similarity: int = Field( + default=10, + description="Minimum unique word count to apply Jaccard check; shorter content skips near-duplicate detection entirely", + ) + exempt_tools: set[str] = Field( + default_factory=lambda: {"ask_clarification", "write_todos", "present_files", "task"}, + description="Tool names excluded from progress tracking", + ) + max_tracked_threads: int = Field( + default=100, + ge=1, + description="Maximum number of thread histories to keep in memory (LRU eviction)", + ) diff --git a/backend/tests/test_read_before_write_middleware.py b/backend/tests/test_read_before_write_middleware.py index c07a0b8d8..c502e0ea9 100644 --- a/backend/tests/test_read_before_write_middleware.py +++ b/backend/tests/test_read_before_write_middleware.py @@ -213,6 +213,16 @@ class TestWriteGate: handler.assert_called_once() assert result.status != "error" + def test_blocked_write_has_deerflow_tool_meta(self): + from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY + + mw = _middleware({self.PATH: "v1"}) + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}) + result = mw.wrap_tool_call(request, MagicMock()) + meta = (result.additional_kwargs or {}).get(TOOL_META_KEY) + assert meta is not None, "blocked write must carry deerflow_tool_meta" + assert meta["recoverable_by_model"] is True + class TestAsyncPaths: PATH = "/mnt/user-data/outputs/report.md" @@ -229,6 +239,22 @@ class TestAsyncPaths: result = asyncio.run(mw.awrap_tool_call(request, handler)) assert result.status == "error" + def test_async_blocked_write_has_deerflow_tool_meta(self): + import asyncio + + from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY + + mw = _middleware({self.PATH: "v1"}) + request = _make_request("write_file", {"description": "d", "path": self.PATH, "content": "v2"}) + + async def handler(_request): + raise AssertionError("handler must not run when blocked") + + result = asyncio.run(mw.awrap_tool_call(request, handler)) + meta = (result.additional_kwargs or {}).get(TOOL_META_KEY) + assert meta is not None, "async blocked write must carry deerflow_tool_meta" + assert meta["recoverable_by_model"] is True + def test_async_read_stamps_mark(self): import asyncio diff --git a/backend/tests/test_tool_error_handling_middleware.py b/backend/tests/test_tool_error_handling_middleware.py index 065f0e3f1..3936502bf 100644 --- a/backend/tests/test_tool_error_handling_middleware.py +++ b/backend/tests/test_tool_error_handling_middleware.py @@ -10,6 +10,7 @@ from deerflow.agents.middlewares.tool_error_handling_middleware import ( build_lead_runtime_middlewares, build_subagent_runtime_middlewares, ) +from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY from deerflow.agents.middlewares.view_image_middleware import ViewImageMiddleware from deerflow.config import summarization_config from deerflow.config.app_config import AppConfig, CircuitBreakerConfig @@ -156,6 +157,72 @@ def test_build_subagent_runtime_middlewares_threads_app_config_to_llm_middleware assert isinstance(middlewares[-1], SafetyFinishReasonMiddleware) +def test_tool_progress_middleware_is_outer_relative_to_error_handling(monkeypatch: pytest.MonkeyPatch): + # ToolProgressMiddleware must have a lower index than ToolErrorHandlingMiddleware + # so that the framework's "first in list = outermost" rule makes it outer. + # Only then can it read deerflow_tool_meta stamped by ToolErrorHandlingMiddleware. + from deerflow.agents.middlewares.tool_progress_middleware import ToolProgressMiddleware + from deerflow.config.tool_progress_config import ToolProgressConfig + + app_config = AppConfig( + models=[ + ModelConfig( + name="test-model", + display_name="test-model", + description=None, + use="langchain_openai:ChatOpenAI", + model="test-model", + ) + ], + sandbox=SandboxConfig(use="test"), + guardrails=GuardrailsConfig(enabled=False), + circuit_breaker=CircuitBreakerConfig(failure_threshold=7, recovery_timeout_sec=11), + tool_progress=ToolProgressConfig(enabled=True), + ) + + _stub_runtime_middleware_imports(monkeypatch) + + middlewares = build_subagent_runtime_middlewares(app_config=app_config, lazy_init=False) + + progress_idx = next(i for i, m in enumerate(middlewares) if isinstance(m, ToolProgressMiddleware)) + error_idx = next(i for i, m in enumerate(middlewares) if isinstance(m, ToolErrorHandlingMiddleware)) + assert progress_idx < error_idx, f"ToolProgressMiddleware (index {progress_idx}) must be outer (lower index) than ToolErrorHandlingMiddleware (index {error_idx}); order: {[type(m).__name__ for m in middlewares]}" + + +def test_middleware_ordering_guard_raises_when_progress_is_inner(monkeypatch: pytest.MonkeyPatch): + """_build_runtime_middlewares must raise RuntimeError when ToolProgressMiddleware ends up + at a higher index than ToolErrorHandlingMiddleware. + + We trigger the wrong-order condition by patching SandboxAuditMiddleware to be an actual + ToolErrorHandlingMiddleware instance, which appears BEFORE ToolProgressMiddleware in the + list. The guard's isinstance() check finds it first, making error_idx < progress_idx. + """ + from deerflow.agents.middlewares.tool_error_handling_middleware import ( + ToolErrorHandlingMiddleware, + build_lead_runtime_middlewares, + ) + from deerflow.config.tool_progress_config import ToolProgressConfig + + _stub_runtime_middleware_imports(monkeypatch) + # Override the SandboxAuditMiddleware stub with a real ToolErrorHandlingMiddleware so it + # becomes the FIRST ToolErrorHandlingMiddleware in the list, appearing before + # ToolProgressMiddleware and triggering the ordering guard. + monkeypatch.setitem( + sys.modules, + "deerflow.agents.middlewares.sandbox_audit_middleware", + _module( + "deerflow.agents.middlewares.sandbox_audit_middleware", + SandboxAuditMiddleware=ToolErrorHandlingMiddleware, + ), + ) + + app_config = _make_app_config() + app_config = app_config.model_copy(update={"tool_progress": ToolProgressConfig(enabled=True)}) + + with pytest.raises(RuntimeError, match="ToolProgressMiddleware must be outer"): + build_lead_runtime_middlewares(app_config=app_config, lazy_init=False) + + def test_lead_runtime_middlewares_thread_app_config_to_tool_error_handling(monkeypatch: pytest.MonkeyPatch): monkeypatch.setitem( sys.modules, @@ -330,6 +397,23 @@ def test_wrap_tool_call_returns_error_tool_message_on_exception(): assert "network down" in result.text +def test_wrap_tool_call_stamps_tool_meta_on_exception(): + middleware = ToolErrorHandlingMiddleware() + req = _request(name="web_search", tool_call_id="tc-42") + + def _boom(_req): + raise ConnectionError("connection refused") + + result = middleware.wrap_tool_call(req, _boom) + + assert isinstance(result, ToolMessage) + assert TOOL_META_KEY in result.additional_kwargs + meta = result.additional_kwargs[TOOL_META_KEY] + assert meta["status"] == "error" + assert meta["source"] == "exception" + assert meta["error_type"] == "transient" + + def test_task_exception_wrapper_uses_subagent_result_formatter(): middleware = ToolErrorHandlingMiddleware() req = _request(name="task", tool_call_id="tc-task") diff --git a/backend/tests/test_tool_progress_middleware.py b/backend/tests/test_tool_progress_middleware.py new file mode 100644 index 000000000..c9addcfe8 --- /dev/null +++ b/backend/tests/test_tool_progress_middleware.py @@ -0,0 +1,1479 @@ +"""Tests for ToolProgressMiddleware state machine (RFC #3177).""" + +from __future__ import annotations + +import logging +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest +from langchain_core.messages import HumanMessage, ToolMessage +from langgraph.types import Command + +from deerflow.agents.middlewares.tool_progress_middleware import ( + ToolProgressMiddleware, + is_near_duplicate, + word_set, +) +from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY + +# --------------------------------------------------------------------------- +# Helpers + + +def _make_runtime(thread_id: str = "t1", run_id: str = "r1") -> MagicMock: + rt = MagicMock() + rt.context = {"thread_id": thread_id, "run_id": run_id} + return rt + + +def _make_tool_request(tool_name: str = "web_search", *, runtime: MagicMock | None = None) -> SimpleNamespace: + rt = runtime or _make_runtime() + return SimpleNamespace( + tool_call={"name": tool_name, "id": f"tc-{tool_name}"}, + runtime=rt, + ) + + +def _meta_kwargs( + *, + status: str = "success", + error_type: str | None = None, + recoverable_by_model: bool = True, + recommended_next_action: str = "continue", + source: str = "content_analysis", +) -> dict[str, object]: + return { + TOOL_META_KEY: { + "status": status, + "error_type": error_type, + "recoverable_by_model": recoverable_by_model, + "recommended_next_action": recommended_next_action, + "source": source, + } + } + + +def _make_tool_message( + content: str = "A" * 200, + *, + tool_name: str = "web_search", + meta_kwargs: dict[str, object] | None = None, +) -> ToolMessage: + return ToolMessage( + content=content, + tool_call_id=f"tc-{tool_name}", + name=tool_name, + status="success", + additional_kwargs=meta_kwargs or _meta_kwargs(), + ) + + +def _make_non_recoverable_error_message( + content: str = "Error: rate limited", + *, + tool_name: str = "web_search", + error_type: str = "rate_limited", + recommended_next_action: str = "summarize", +) -> ToolMessage: + """Non-recoverable stagnation error (recoverable_by_model=False, non-stop). + Unlike auth/config, these go through the stagnation counter, but should + still reach BLOCKED because the model cannot fix them by retrying. + """ + return ToolMessage( + content=content, + tool_call_id=f"tc-{tool_name}", + name=tool_name, + status="error", + additional_kwargs=_meta_kwargs( + status="error", + error_type=error_type, + recoverable_by_model=False, + recommended_next_action=recommended_next_action, + ), + ) + + +def _make_error_message( + content: str = "Error: no results found", + *, + tool_name: str = "web_search", + error_type: str = "no_results", + recoverable_by_model: bool = True, + recommended_next_action: str = "rewrite_query", +) -> ToolMessage: + return ToolMessage( + content=content, + tool_call_id=f"tc-{tool_name}", + name=tool_name, + status="error", + additional_kwargs=_meta_kwargs( + status="error", + error_type=error_type, + recoverable_by_model=recoverable_by_model, + recommended_next_action=recommended_next_action, + ), + ) + + +def _make_model_request(messages: list, runtime: MagicMock) -> MagicMock: + req = MagicMock() + req.messages = list(messages) + req.runtime = runtime + + def _override(**kw) -> MagicMock: + updated = MagicMock() + updated.messages = kw.get("messages", req.messages) + updated.runtime = runtime + updated.override = req.override + return updated + + req.override = _override + return req + + +def _make_mw(**kwargs) -> ToolProgressMiddleware: + defaults = { + "stagnation_threshold": 3, + "warn_escalation_count": 2, + "inject_assessment": True, + "jaccard_threshold": 0.8, + "min_words": 5, + } + defaults.update(kwargs) + return ToolProgressMiddleware(**defaults) + + +# --------------------------------------------------------------------------- +# Unit tests: word_set and is_near_duplicate + + +def test_word_set_extracts_words_ge_3(): + ws = word_set("go quick brown fox") + assert "go" not in ws + assert "quick" in ws + assert "brown" in ws + assert "fox" in ws + + +def test_is_near_duplicate_above_threshold(): + ws1 = frozenset("quick brown fox jumps over lazy dog".split()) + ws2 = frozenset("quick brown fox jumps over lazy dog".split()) + assert is_near_duplicate(ws2, [ws1], threshold=0.8, min_words=5) + + +def test_is_near_duplicate_near_threshold(): + # ws1 has 8 words; ws2 shares 7 of them and adds 1 new word. + # intersection=7, union=9 → Jaccard = 7/9 ≈ 0.778 < 0.8 → NOT duplicate. + # ws3 shares all 8 original words and adds 1 new word. + # intersection=8, union=9 → Jaccard = 8/9 ≈ 0.889 >= 0.8 → IS duplicate. + base = frozenset("alpha bravo charlie delta echo foxtrot golf hotel".split()) + nearly_below = frozenset("alpha bravo charlie delta echo foxtrot golf india".split()) # 7/9 ≈ 0.778 + nearly_above = frozenset("alpha bravo charlie delta echo foxtrot golf hotel india".split()) # 8/9 ≈ 0.889 + assert not is_near_duplicate(nearly_below, [base], threshold=0.8, min_words=5) + assert is_near_duplicate(nearly_above, [base], threshold=0.8, min_words=5) + + +def test_is_near_duplicate_below_threshold(): + ws1 = frozenset("apple banana cherry delta echo".split()) + ws2 = frozenset("xray yankee zulu alpha bravo".split()) + assert not is_near_duplicate(ws2, [ws1], threshold=0.8, min_words=5) + + +def test_is_near_duplicate_too_short_skips_check(): + ws1 = frozenset("apple".split()) + ws2 = frozenset("apple".split()) + # min_words=5 but len==1, so not a duplicate + assert not is_near_duplicate(ws2, [ws1], threshold=0.8, min_words=5) + + +# --------------------------------------------------------------------------- +# Scenario 1: Normal call → no hint, phase stays active + + +def test_normal_call_no_hint_phase_active(): + mw = _make_mw() + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + msg = _make_tool_message("A" * 300) + + def handler(_r): + return msg + + result = mw.wrap_tool_call(req, handler) + + assert result is msg + assert mw._phase_states["t1"]["web_search"].phase == "active" + assert mw._phase_states["t1"]["web_search"].consecutive_problems == 0 + + +# --------------------------------------------------------------------------- +# Scenario 2: consecutive no_results → hint injected, phase=warned + + +def test_repeated_no_results_reaches_warned(): + mw = _make_mw(stagnation_threshold=2) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + error_msg = _make_error_message() + + def handler(_r): + return error_msg + + # stagnation_threshold=2, so the second problem call tips into warned + mw.wrap_tool_call(req, handler) + mw.wrap_tool_call(req, handler) + + state = mw._phase_states["t1"]["web_search"] + assert state.phase == "warned" + assert state.consecutive_problems == 2 + + # Hint should be queued + hints = mw._drain_pending(rt) + assert len(hints) == 1 + assert "PROGRESS HINT" in hints[0] + + +# --------------------------------------------------------------------------- +# Scenario 3: Non-recoverable errors escalate warned → blocked + + +def test_warned_to_blocked_after_escalation(): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=2) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + # Non-recoverable error (rate_limited): model cannot fix this by retrying, + # so stagnation should escalate to BLOCKED. + error_msg = _make_non_recoverable_error_message() + + def handler(_r): + return error_msg + + for _ in range(4): + mw.wrap_tool_call(req, handler) + + state = mw._phase_states["t1"]["web_search"] + assert state.phase == "blocked" + assert state.block_reason is not None + + +# --------------------------------------------------------------------------- +# Scenario 4: Blocked tool is front-gate intercepted (handler NOT called) + + +def test_blocked_tool_is_intercepted_without_calling_handler(): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=1) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + # Non-recoverable error: stagnation escalates to BLOCKED, handler is never called. + error_msg = _make_non_recoverable_error_message() + call_count = [0] + + def handler(r): + call_count[0] += 1 + return error_msg + + # 2 calls → warned + 1 more = blocked + for _ in range(3): + mw.wrap_tool_call(req, handler) + + assert mw._phase_states["t1"]["web_search"].phase == "blocked" + call_count_before = call_count[0] + + # Next call should be intercepted + result = mw.wrap_tool_call(req, handler) + + assert call_count[0] == call_count_before + assert isinstance(result, ToolMessage) + assert "[TOOL_BLOCKED]" in result.content + + +# --------------------------------------------------------------------------- +# Scenario 4b: Recoverable errors never escalate to BLOCKED — WARNED is terminal + + +def test_recoverable_errors_stay_warned_indefinitely(): + # stagnation_threshold=2, warn_escalation_count=1 → would block at call 3 for + # non-recoverable errors, but recoverable errors must stay in WARNED forever. + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=1) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + error_msg = _make_error_message() # recoverable_by_model=True + + def handler(_r): + return error_msg + + # 10 calls — well past the threshold+escalation + for _ in range(10): + mw.wrap_tool_call(req, handler) + + state = mw._phase_states["t1"]["web_search"] + assert state.phase == "warned", "recoverable errors must never escalate to BLOCKED" + assert state.consecutive_problems == 10 + + +def test_recoverable_error_re_injects_hint_past_escalation(): + # After crossing threshold+escalation for a recoverable error, each additional + # problem call should still queue a hint. + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=1, inject_assessment=True) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + error_msg = _make_error_message() + + def handler(_r): + return error_msg + + # Reach warned (call 2) and past escalation (call 3+) + for _ in range(4): + mw.wrap_tool_call(req, handler) + + # All hints from call 2 onward should have been queued (capped at _MAX_PENDING_PER_RUN=3). + # >= 2 proves that at least one hint was queued *inside* the escalation zone (calls 3+), + # not just the initial WARNED hint at call 2. + hints = mw._drain_pending(rt) + assert len(hints) >= 2 + assert all("PROGRESS HINT" in h for h in hints) + + +# --------------------------------------------------------------------------- +# Scenario 5: Auth error → immediately blocked (no warned phase) + + +def test_auth_error_immediately_blocked(): + mw = _make_mw(stagnation_threshold=5) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + auth_msg = _make_error_message( + error_type="auth", + recoverable_by_model=False, + recommended_next_action="stop", + ) + + def handler(_r): + return auth_msg + + mw.wrap_tool_call(req, handler) + + state = mw._phase_states["t1"]["web_search"] + assert state.phase == "blocked" + assert "auth" in state.block_reason.lower() or "Authentication" in state.block_reason + # consecutive_problems must be 1 (not 0) even on immediate-block paths so diagnostic + # logs and future consumers see a consistent non-zero count after a failed call. + assert state.consecutive_problems == 1 + + +# --------------------------------------------------------------------------- +# Scenario 6: Valid result after problems resets to active + + +def test_valid_result_after_problems_resets_to_active(): + mw = _make_mw(stagnation_threshold=3, warn_escalation_count=5) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + error_msg = _make_error_message() + good_msg = _make_tool_message("A" * 300) + + def handler_error(_r): + return error_msg + + def handler_good(_r): + return good_msg + + mw.wrap_tool_call(req, handler_error) + mw.wrap_tool_call(req, handler_error) + mw.wrap_tool_call(req, handler_error) + + state = mw._phase_states["t1"]["web_search"] + assert state.phase == "warned" + + # Good result resets + mw.wrap_tool_call(req, handler_good) + + state = mw._phase_states["t1"]["web_search"] + assert state.phase == "active" + assert state.consecutive_problems == 0 + + +# --------------------------------------------------------------------------- +# Scenario 7: Two different tools have independent states + + +def test_two_tools_have_independent_states(): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=1) + rt = _make_runtime() + req_search = _make_tool_request("web_search", runtime=rt) + req_read = _make_tool_request("read_file", runtime=rt) + + # Non-recoverable errors so web_search escalates to BLOCKED. + error_search = _make_non_recoverable_error_message(tool_name="web_search") + error_read = _make_error_message(tool_name="read_file") + + # Drive web_search to BLOCKED (2 → warned, 1 more → blocked) + for _ in range(3): + mw.wrap_tool_call(req_search, lambda r: error_search) + + assert mw._phase_states["t1"]["web_search"].phase == "blocked" + + # read_file should still be active — independent state per tool name + mw.wrap_tool_call(req_read, lambda r: error_read) + assert mw._phase_states["t1"]["read_file"].phase == "active" + + +# --------------------------------------------------------------------------- +# Scenario 8: Jaccard near-duplicate result counts as problem + + +def test_jaccard_near_duplicate_counts_as_problem(): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5, jaccard_threshold=0.8, min_words=5) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + + # First call: good unique content (establishes baseline) + words = "apple banana cherry delta echo foxtrot golf hotel india juliet" + msg1 = _make_tool_message(words) + mw.wrap_tool_call(req, lambda r: msg1) + + # Second call: exact same content (Jaccard = 1.0) → near-duplicate → problem count goes up + msg2 = _make_tool_message(words) + mw.wrap_tool_call(req, lambda r: msg2) + + state = mw._phase_states["t1"]["web_search"] + assert state.consecutive_problems >= 1 + + +# --------------------------------------------------------------------------- +# Scenario 9: Different Jaccard content does NOT count as problem + + +def test_jaccard_different_content_not_a_problem(): + mw = _make_mw(stagnation_threshold=3, warn_escalation_count=5, jaccard_threshold=0.8, min_words=5) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + + words1 = "apple banana cherry delta echo foxtrot golf hotel india juliet" + words2 = "xray yankee zulu alpha bravo charlie sierra tango uniform victor" + msg1 = _make_tool_message(words1) + msg2 = _make_tool_message(words2) + + mw.wrap_tool_call(req, lambda r: msg1) + mw.wrap_tool_call(req, lambda r: msg2) + + state = mw._phase_states["t1"]["web_search"] + assert state.consecutive_problems == 0 + assert state.phase == "active" + + +# --------------------------------------------------------------------------- +# Scenario 9b: production default min_words=10 skips Jaccard for short content + + +def test_jaccard_skipped_when_content_below_production_min_words(): + """Production default min_words=10 must skip Jaccard for content with 6-9 unique words. + + _make_mw() uses min_words=5 to make most tests easier to set up. This test + uses the production default (min_words=10) to verify that short but repeated + content does NOT count as a near-duplicate stagnation problem. + """ + mw = ToolProgressMiddleware( + stagnation_threshold=3, + warn_escalation_count=2, + jaccard_threshold=0.8, + min_words=10, # production default + ) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + + # 7 unique words — above min_words=5 but below production min_words=10. + # With min_words=10 the Jaccard check is skipped → never a problem → phase stays active. + words = "apple banana cherry delta echo foxtrot golf" + msg = _make_tool_message(words) + + for _ in range(5): + mw.wrap_tool_call(req, lambda r: msg) + + state = mw._phase_states["t1"]["web_search"] + assert state.phase == "active", "7-word repeated content must not trigger stagnation with production min_words=10" + assert state.consecutive_problems == 0 + + +# --------------------------------------------------------------------------- +# Scenario 10: exempt_tools are not tracked + + +def test_exempt_tools_not_tracked(): + mw = _make_mw(stagnation_threshold=1, warn_escalation_count=1) + rt = _make_runtime() + req = _make_tool_request("ask_clarification", runtime=rt) + error_msg = _make_error_message(tool_name="ask_clarification") + + def handler(_r): + return error_msg + + for _ in range(5): + mw.wrap_tool_call(req, handler) + + assert "ask_clarification" not in mw._phase_states.get("t1", {}) + + +# --------------------------------------------------------------------------- +# Scenario 11: before_agent clears stale pending hints from previous runs + + +def test_before_agent_clears_stale_pending(): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5) + rt_run1 = _make_runtime(thread_id="t1", run_id="old-run") + rt_run2 = _make_runtime(thread_id="t1", run_id="new-run") + req = _make_tool_request(runtime=rt_run1) + error_msg = _make_error_message() + + # Produce a hint for old-run + mw.wrap_tool_call(req, lambda r: error_msg) + mw.wrap_tool_call(req, lambda r: error_msg) + + mw._drain_pending(rt_run1) + # Re-queue manually to simulate pending state + mw._queue_assessment(rt_run1, "old hint") + + # before_agent with new-run should clear the old-run's pending hints + state_mock = MagicMock() + mw.before_agent(state_mock, rt_run2) + + # Old pending should be gone + leftovers = mw._pending.get(("t1", "old-run"), []) + assert leftovers == [] + + +@pytest.mark.anyio +async def test_abefore_agent_clears_stale_pending(): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5) + rt_run1 = _make_runtime(thread_id="t1", run_id="old-run") + rt_run2 = _make_runtime(thread_id="t1", run_id="new-run") + req = _make_tool_request(runtime=rt_run1) + error_msg = _make_error_message() + + mw.wrap_tool_call(req, lambda r: error_msg) + mw.wrap_tool_call(req, lambda r: error_msg) + mw._drain_pending(rt_run1) + mw._queue_assessment(rt_run1, "old hint") + + state_mock = MagicMock() + await mw.abefore_agent(state_mock, rt_run2) + + leftovers = mw._pending.get(("t1", "old-run"), []) + assert leftovers == [] + + +def test_before_agent_preserves_current_run_hints(): + # _clear_stale_pending deletes keys where thread_id matches but run_id differs. + # Hints for the *current* run must not be evicted. + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5) + rt = _make_runtime(thread_id="t1", run_id="current-run") + # _queue_assessment guards against phantom entries by checking _phase_states; seed the + # thread so the direct call below isn't silently dropped by the L1 guard. + mw._phase_states["t1"] = {} + mw._queue_assessment(rt, "current hint") + + state_mock = MagicMock() + mw.before_agent(state_mock, rt) + + preserved = mw._pending.get(("t1", "current-run"), []) + assert preserved == ["current hint"] + + +def test_before_agent_resets_blocked_states_for_new_run(): + """BLOCKED and WARNED tool states must both be cleared at the start of a new run. + + A tool BLOCKED in run R1 must not silently remain blocked in R2. + A tool WARNED in R1 must not carry its consecutive_problems count into R2 + (the model has not seen the warning context, so it would be hard-blocked + without ever receiving a hint in the current session). + recent_word_sets must also be cleared so stale Jaccard windows don't cause + false near-duplicate detections on the first success call of the new run. + """ + mw = _make_mw(stagnation_threshold=1, warn_escalation_count=1) + rt_run1 = _make_runtime(thread_id="t1", run_id="run-1") + rt_run2 = _make_runtime(thread_id="t1", run_id="run-2") + req = _make_tool_request(runtime=rt_run1) + + # Drive the tool to BLOCKED via auth error (immediate block, no WARN stage) + auth_msg = ToolMessage( + content="Error: invalid api key", + tool_call_id="tc-web_search", + name="web_search", + status="error", + additional_kwargs=_meta_kwargs( + status="error", + error_type="auth", + recoverable_by_model=False, + recommended_next_action="stop", + ), + ) + mw.wrap_tool_call(req, lambda _r: auth_msg) + assert mw._phase_states["t1"]["web_search"].phase == "blocked" + + # Simulate start of run 2 + state_mock = MagicMock() + mw.before_agent(state_mock, rt_run2) + + # _reset_run_states always replaces the entry in-place; it is never None. + tool_state = mw._phase_states.get("t1", {}).get("web_search") + assert tool_state is not None + assert tool_state.phase == "active" + assert tool_state.consecutive_problems == 0 + assert tool_state.block_reason is None + assert tool_state.recent_word_sets == () + + +def test_before_agent_resets_warned_states_for_new_run(): + """WARNED tool state must also be cleared by before_agent. + + A tool with phase='warned' and accumulated consecutive_problems at end of run R1 + must not carry that count into R2; the model has no warning context and would + be hard-blocked after just a few calls without receiving a hint. + """ + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5) + rt_run1 = _make_runtime(thread_id="t1", run_id="run-1") + rt_run2 = _make_runtime(thread_id="t1", run_id="run-2") + req = _make_tool_request(runtime=rt_run1) + error_msg = _make_error_message() + + # Drive to WARNED (stagnation_threshold=2 means 2 problems → warned) + mw.wrap_tool_call(req, lambda _r: error_msg) + mw.wrap_tool_call(req, lambda _r: error_msg) + assert mw._phase_states["t1"]["web_search"].phase == "warned" + assert mw._phase_states["t1"]["web_search"].consecutive_problems == 2 + + state_mock = MagicMock() + mw.before_agent(state_mock, rt_run2) + + tool_state = mw._phase_states.get("t1", {}).get("web_search") + assert tool_state is not None + assert tool_state.phase == "active" + assert tool_state.consecutive_problems == 0 + assert tool_state.recent_word_sets == () + + +def test_before_agent_resets_active_state_consecutive_problems_and_word_sets(): + """ACTIVE tools with sub-threshold problems must also be cleaned at run boundaries. + + An ACTIVE tool (phase never left 'active') can exit a run with non-zero + consecutive_problems and non-empty recent_word_sets. If _reset_run_states only + touched BLOCKED/WARNED tools, the counter from R1 would bleed into R2: a single + problem on R2's first call could then trip WARNED against stale R1 context that + the model has never seen. + """ + # stagnation_threshold=3 so two errors keep the tool ACTIVE. + mw = _make_mw(stagnation_threshold=3, warn_escalation_count=5) + rt_run1 = _make_runtime(thread_id="t1", run_id="run-1") + rt_run2 = _make_runtime(thread_id="t1", run_id="run-2") + req = _make_tool_request(runtime=rt_run1) + + # Two successes → recent_word_sets grows. + success_a = _make_tool_message("alpha beta gamma delta epsilon zeta eta theta iota kappa") + success_b = _make_tool_message("lambda mu nu xi omicron pi rho sigma tau upsilon phi chi") + mw.wrap_tool_call(req, lambda _r: success_a) + mw.wrap_tool_call(req, lambda _r: success_b) + + # One recoverable error → consecutive_problems=1, phase stays ACTIVE. + error_msg = _make_error_message() + mw.wrap_tool_call(req, lambda _r: error_msg) + + state_r1 = mw._phase_states.get("t1", {}).get("web_search") + assert state_r1 is not None + assert state_r1.phase == "active" + assert state_r1.consecutive_problems == 1 + assert len(state_r1.recent_word_sets) > 0 + + # Start of run 2: all per-run state must be cleared. + state_mock = MagicMock() + mw.before_agent(state_mock, rt_run2) + + state_r2 = mw._phase_states.get("t1", {}).get("web_search") + assert state_r2 is not None + assert state_r2.phase == "active" + assert state_r2.consecutive_problems == 0 + assert state_r2.recent_word_sets == () + + +# --------------------------------------------------------------------------- +# Scenario 12: LRU eviction when max_tracked_threads exceeded + + +def test_get_block_reason_does_not_create_phantom_entries(): + # _get_block_reason is called on every wrap_tool_call before the handler. + # It must not insert an empty entry for new threads (which could prematurely + # evict another thread's WARNED state via LRU). + mw = _make_mw(max_tracked_threads=2, stagnation_threshold=2) + rt_a = _make_runtime(thread_id="thread-a") + rt_b = _make_runtime(thread_id="thread-b") + rt_c = _make_runtime(thread_id="thread-c") + + req_a = _make_tool_request(runtime=rt_a) + error_msg = _make_error_message() + + # Drive thread-a to WARNED state (needs 2 error calls with threshold=2). + mw.wrap_tool_call(req_a, lambda r: error_msg) + mw.wrap_tool_call(req_a, lambda r: error_msg) + assert mw._phase_states["thread-a"]["web_search"].phase == "warned" + + # Drive thread-b so it has a real entry too. + req_b = _make_tool_request(runtime=rt_b) + good_msg = _make_tool_message("A" * 300) + mw.wrap_tool_call(req_b, lambda r: good_msg) + assert "thread-b" in mw._phase_states + + # Now thread-c makes its very first call. max_tracked_threads=2, so adding + # thread-c must evict one of {thread-a, thread-b} — but the eviction must + # only happen in _update_state_from_result (the write path), not in + # _get_block_reason (the read path that runs first). + # After wrap_tool_call completes, the two survivors should be thread-b and + # thread-c (thread-a is oldest because thread-b was accessed most recently). + req_c = _make_tool_request(tool_name="read_file", runtime=rt_c) + mw.wrap_tool_call(req_c, lambda r: good_msg) + + # thread-c must now have a real entry (not an empty phantom). + assert "thread-c" in mw._phase_states + assert mw._phase_states["thread-c"].get("read_file") is not None + + # No more than max_tracked_threads entries should exist. + assert len(mw._phase_states) <= 2 + + +def test_lru_eviction_of_oldest_thread(): + mw = _make_mw(max_tracked_threads=2) + error_msg = _make_error_message() + + for i in range(3): + rt = _make_runtime(thread_id=f"thread-{i}") + req = _make_tool_request(runtime=rt) + mw.wrap_tool_call(req, lambda r: error_msg) + + assert len(mw._phase_states) == 2 + # thread-0 should have been evicted (oldest); thread-1 and thread-2 remain + assert "thread-0" not in mw._phase_states + assert "thread-1" in mw._phase_states + assert "thread-2" in mw._phase_states + + +def test_pending_evicted_with_phase_states_on_lru_overflow(): + """M1 regression: _pending keys for evicted threads must be cleaned up. + + When _phase_states evicts a thread via LRU, any pending hint entries + for that thread must also be removed so _pending cannot grow unboundedly. + """ + mw = _make_mw(max_tracked_threads=2, stagnation_threshold=2) + error_msg = _make_error_message() + + # Thread-0: produce a hint (reach WARNED) so it has a pending entry. + rt0 = _make_runtime(thread_id="thread-0", run_id="run-0") + req0 = _make_tool_request(runtime=rt0) + mw.wrap_tool_call(req0, lambda r: error_msg) + mw.wrap_tool_call(req0, lambda r: error_msg) + # Verify thread-0 has a pending hint. + assert len(mw._pending.get(("thread-0", "run-0"), [])) >= 1 + + # Thread-1: occupy the second slot. + rt1 = _make_runtime(thread_id="thread-1", run_id="run-1") + req1 = _make_tool_request(runtime=rt1) + good_msg = _make_tool_message("A" * 300) + mw.wrap_tool_call(req1, lambda r: good_msg) + + # Thread-2: adding this forces LRU eviction of thread-0. + rt2 = _make_runtime(thread_id="thread-2", run_id="run-2") + req2 = _make_tool_request(runtime=rt2) + mw.wrap_tool_call(req2, lambda r: good_msg) + + # thread-0 must be evicted from phase_states. + assert "thread-0" not in mw._phase_states + + # The pending entry for thread-0 must also be gone (no memory leak). + assert ("thread-0", "run-0") not in mw._pending, "_pending entry for evicted thread-0 should have been cleaned up" + + +# --------------------------------------------------------------------------- +# Hint injection via wrap_model_call + + +def test_hint_injected_into_model_call(): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + error_msg = _make_error_message() + + # Trigger hint + mw.wrap_tool_call(req, lambda r: error_msg) + mw.wrap_tool_call(req, lambda r: error_msg) + + model_req = _make_model_request([], rt) + captured_messages = [] + + def model_handler(r): + captured_messages.extend(r.messages) + return MagicMock() + + mw.wrap_model_call(model_req, model_handler) + + assert any(isinstance(m, HumanMessage) for m in captured_messages) + hint_msgs = [m for m in captured_messages if isinstance(m, HumanMessage)] + assert any("PROGRESS HINT" in m.content for m in hint_msgs) + + +def test_partial_success_hint_is_specific_not_generic(): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + partial_msg = ToolMessage( + content="Here are some partial results from the search.", + tool_call_id="tc-web_search", + name="web_search", + status="success", + additional_kwargs=_meta_kwargs( + status="partial_success", + recommended_next_action="rewrite_query", + ), + ) + + def handler(_r): + return partial_msg + + mw.wrap_tool_call(req, handler) + mw.wrap_tool_call(req, handler) + + hints = mw._drain_pending(rt) + assert len(hints) == 1 + assert "incomplete results" in hints[0].lower() + assert "not producing new information" not in hints[0] + + +def test_jaccard_near_dup_hint_is_specific_and_actionable(): + """Near-duplicate success hint must be specific (not generic fallback) and include action guidance. + + Before the fix, status='success'/error_type=None fell through to the generic fallback + '[PROGRESS HINT] The tool is not producing new information.' with no action suffix + (recommended_next_action='continue' was absent from action_map). The fix adds a + 'success' key to the base dict and a 'continue' key to the action_map. + """ + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5, jaccard_threshold=0.8, min_words=5) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + + # First call: good unique content to seed recent_word_sets. + words = "apple banana cherry delta echo foxtrot golf hotel india juliet" + good_msg = _make_tool_message(words) + mw.wrap_tool_call(req, lambda r: good_msg) + + # Second and third calls: exact same content → near-duplicate → stagnation_threshold=2 → WARNED. + dup_msg = _make_tool_message(words) + + def handler(_r): + return dup_msg + + mw.wrap_tool_call(req, handler) + mw.wrap_tool_call(req, handler) + + hints = mw._drain_pending(rt) + assert len(hints) == 1 + hint = hints[0] + # Must contain a specific near-dup message, not the generic fallback. + assert "duplicate" in hint.lower(), f"expected 'duplicate' in hint, got: {hint!r}" + # Must include an actionable suggestion (from action_map["continue"]). + assert "rephras" in hint.lower() or "different" in hint.lower(), f"expected action guidance in hint, got: {hint!r}" + + +def test_no_hint_when_inject_assessment_disabled(): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5, inject_assessment=False) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + error_msg = _make_error_message() + + mw.wrap_tool_call(req, lambda r: error_msg) + mw.wrap_tool_call(req, lambda r: error_msg) + + hints = mw._drain_pending(rt) + assert hints == [] + + +def test_augment_request_deduplicates_identical_hints(): + """L2: _augment_request must deduplicate identical hint strings via dict.fromkeys. + + If the same hint text appears multiple times in the queue (e.g. two successive + no_results errors produce identical hint strings), only one copy should be + injected into the model message. + """ + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5, inject_assessment=True) + rt = _make_runtime() + + # _queue_assessment guards against phantom entries; seed the thread so the direct + # calls below aren't dropped by the L1 guard. + mw._phase_states["t1"] = {} + # Manually queue two identical hints to simulate duplicates. + mw._queue_assessment(rt, "[PROGRESS HINT] same hint text") + mw._queue_assessment(rt, "[PROGRESS HINT] same hint text") + + model_req = _make_model_request([], rt) + captured: list = [] + + def model_handler(r): + captured.extend(r.messages) + return MagicMock() + + mw.wrap_model_call(model_req, model_handler) + + hint_msgs = [m for m in captured if isinstance(m, HumanMessage)] + assert len(hint_msgs) == 1 + # The single injected message must contain the hint exactly once. + assert hint_msgs[0].content.count("[PROGRESS HINT] same hint text") == 1 + + +# --------------------------------------------------------------------------- +# L1: _assess_and_transition called with already-blocked state is idempotent + + +def test_assess_and_transition_blocked_state_immediate_stop_is_idempotent(): + """L1: _assess_and_transition must handle an already-blocked state without error. + + The docstring states the immediate-block branch re-applies idempotently. + This test verifies that re-entering with a blocked state + stop-action meta + stays blocked and does not corrupt the block_reason. + """ + from deerflow.agents.middlewares.tool_progress_middleware import ToolPhaseState + + mw = _make_mw() + blocked_state = ToolPhaseState( + phase="blocked", + consecutive_problems=5, + block_reason="Authentication failure — this tool cannot be used.", + ) + auth_meta_kwargs = _meta_kwargs( + status="error", + error_type="auth", + recoverable_by_model=False, + recommended_next_action="stop", + )[TOOL_META_KEY] + from deerflow.agents.middlewares.tool_result_meta import ToolResultMeta + + auth_meta = ToolResultMeta(**auth_meta_kwargs) + + new_state, hint = mw._assess_and_transition(blocked_state, auth_meta, "") + + assert new_state.phase == "blocked" + assert new_state.block_reason is not None + assert hint is None # no hint on immediate block path + + +def test_assess_and_transition_blocked_state_non_stop_increments_count(): + """L1: A blocked state receiving a non-stop problem increments counter, stays blocked. + + Simulates a concurrent race where two threads both process results for the + same tool: the second thread's _assess_and_transition receives a stale + 'blocked' snapshot. The result must remain blocked. + """ + from deerflow.agents.middlewares.tool_progress_middleware import ToolPhaseState + + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=1) + blocked_state = ToolPhaseState( + phase="blocked", + consecutive_problems=3, + block_reason="Repeated rate-limiting — summarize current findings and proceed.", + ) + rate_meta_kwargs = _meta_kwargs( + status="error", + error_type="rate_limited", + recoverable_by_model=False, + recommended_next_action="summarize", + )[TOOL_META_KEY] + from deerflow.agents.middlewares.tool_result_meta import ToolResultMeta + + rate_meta = ToolResultMeta(**rate_meta_kwargs) + + new_state, _hint = mw._assess_and_transition(blocked_state, rate_meta, "") + + # Must stay blocked (not regress to warned or active). + assert new_state.phase == "blocked" + # Counter must NOT be incremented: blocked is terminal, state returned unchanged. + assert new_state.consecutive_problems == 3 + + +def test_assess_and_transition_blocked_recoverable_does_not_regress_to_warned(): + """L1: A blocked state with recoverable errors must not silently regress to warned. + + Before the fix, _assess_and_transition had no guard for already-blocked states. + A recoverable error arriving on a blocked state (concurrent race) would take + the `warned` branch because recoverable_by_model=True, demoting the phase from + blocked back to warned. This test locks the fixed behavior. + """ + from deerflow.agents.middlewares.tool_progress_middleware import ToolPhaseState + + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=1) + blocked_state = ToolPhaseState( + phase="blocked", + consecutive_problems=5, + block_reason="Repeated no-results — rewrite your query or try a different tool.", + ) + # Recoverable no_results error (would normally only WARN, never block on its own) + no_results_meta_kwargs = _meta_kwargs( + status="error", + error_type="no_results", + recoverable_by_model=True, + recommended_next_action="rewrite_query", + )[TOOL_META_KEY] + from deerflow.agents.middlewares.tool_result_meta import ToolResultMeta + + no_results_meta = ToolResultMeta(**no_results_meta_kwargs) + + new_state, hint = mw._assess_and_transition(blocked_state, no_results_meta, "") + + assert new_state.phase == "blocked", "blocked must not regress to warned even when the new error is recoverable" + assert hint is None + assert new_state is blocked_state # exact same object returned (no copy) + + +# --------------------------------------------------------------------------- +# Tool without runtime attribute is passed through + + +def test_no_runtime_passthrough(): + mw = _make_mw() + req = SimpleNamespace(tool_call={"name": "web_search", "id": "tc-1"}) + # No runtime attribute + msg = _make_tool_message() + + def handler(_r): + return msg + + result = mw.wrap_tool_call(req, handler) + assert result is msg + + +# --------------------------------------------------------------------------- +# Command results are passed through unchanged + + +def test_command_result_passthrough(): + mw = _make_mw() + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + cmd = Command(goto="some_node") + + def handler(_r): + return cmd + + result = mw.wrap_tool_call(req, handler) + assert result is cmd + + +# --------------------------------------------------------------------------- +# from_config round-trip + + +def test_from_config(): + from deerflow.config.tool_progress_config import ToolProgressConfig + + cfg = ToolProgressConfig( + enabled=True, + stagnation_threshold=4, + warn_escalation_count=3, + jaccard_similarity_threshold=0.7, + min_word_count_for_similarity=8, + ) + mw = ToolProgressMiddleware.from_config(cfg) + assert mw._stagnation_threshold == 4 + assert mw._warn_escalation == 3 + assert mw._jaccard_threshold == pytest.approx(0.7) + assert mw._min_words == 8 + + +def test_from_config_empty_exempt_tools_clears_exemptions(): + """Empty exempt_tools in config must produce an empty set, not the default fallback. + + H1 regression: `exempt_tools or {default}` would silently ignore an empty set + because set() is falsy in Python. The fix uses `is not None` so an explicit + empty set from config actually disables all exemptions. + """ + from deerflow.config.tool_progress_config import ToolProgressConfig + + cfg = ToolProgressConfig(enabled=True, exempt_tools=set()) + mw = ToolProgressMiddleware.from_config(cfg) + assert mw._exempt_tools == set(), "empty exempt_tools in config must clear all exemptions, not fall back to defaults" + + +def test_exempt_tools_none_uses_defaults(): + """None exempt_tools in __init__ must use the built-in default set.""" + mw = ToolProgressMiddleware(exempt_tools=None) + assert "ask_clarification" in mw._exempt_tools + assert "write_todos" in mw._exempt_tools + assert "present_files" in mw._exempt_tools + + +def test_from_config_default_exempt_tools_round_trip(): + """Default exempt_tools from config must match the __init__ default.""" + from deerflow.config.tool_progress_config import ToolProgressConfig + + cfg = ToolProgressConfig(enabled=True) + mw = ToolProgressMiddleware.from_config(cfg) + assert mw._exempt_tools == {"ask_clarification", "write_todos", "present_files", "task"} + + +# --------------------------------------------------------------------------- +# Defensive meta parsing: malformed dicts must not crash the middleware + + +def test_wrap_tool_call_malformed_meta_passthrough(): + """Malformed deerflow_tool_meta dict must not crash the middleware.""" + mw = _make_mw() + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + bad_msg = ToolMessage( + content="some content", + tool_call_id="tc-web_search", + name="web_search", + status="success", + additional_kwargs={TOOL_META_KEY: {"unexpected_field": True}}, + ) + + def handler(_r): + return bad_msg + + result = mw.wrap_tool_call(req, handler) + + assert result is bad_msg + assert mw._phase_states.get("t1", {}).get("web_search") is None + + +def test_missing_meta_on_non_exempt_tool_emits_warning(caplog): + """When deerflow_tool_meta is completely absent for a non-exempt tool, + the middleware must emit a warning pointing to the likely ordering misconfiguration. + """ + import logging + + mw = _make_mw() + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + no_meta_msg = ToolMessage( + content="some content", + tool_call_id="tc-web_search", + name="web_search", + status="success", + additional_kwargs={}, # no TOOL_META_KEY at all + ) + + with caplog.at_level(logging.WARNING, logger="deerflow.agents.middlewares.tool_progress_middleware"): + mw.wrap_tool_call(req, lambda _r: no_meta_msg) + + assert any("deerflow_tool_meta missing" in r.message for r in caplog.records), "Expected a warning about missing meta for non-exempt tool" + + +# --------------------------------------------------------------------------- +# Async path: awrap_tool_call mirrors sync path + + +@pytest.mark.anyio +async def test_awrap_tool_call_normal_passthrough(): + mw = _make_mw() + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + msg = _make_tool_message("A" * 300) + + result = await mw.awrap_tool_call(req, AsyncMock(return_value=msg)) + + assert result is msg + assert mw._phase_states["t1"]["web_search"].phase == "active" + + +@pytest.mark.anyio +async def test_awrap_tool_call_blocked_intercepted_without_calling_handler(): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=1) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + # Non-recoverable error: stagnation escalates to BLOCKED. + error_msg = _make_non_recoverable_error_message() + call_count = [0] + + async def handler(r): + call_count[0] += 1 + return error_msg + + # 3 calls: 2 → warned, 1 more → blocked + for _ in range(3): + await mw.awrap_tool_call(req, handler) + + assert mw._phase_states["t1"]["web_search"].phase == "blocked" + before = call_count[0] + + result = await mw.awrap_tool_call(req, handler) + + assert call_count[0] == before + assert isinstance(result, ToolMessage) + assert "[TOOL_BLOCKED]" in result.content + + +@pytest.mark.anyio +async def test_awrap_tool_call_auth_error_immediately_blocked(): + mw = _make_mw(stagnation_threshold=5) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + auth_msg = _make_error_message( + error_type="auth", + recoverable_by_model=False, + recommended_next_action="stop", + ) + + await mw.awrap_tool_call(req, AsyncMock(return_value=auth_msg)) + + state = mw._phase_states["t1"]["web_search"] + assert state.phase == "blocked" + assert state.block_reason is not None + + +@pytest.mark.anyio +async def test_awrap_tool_call_no_runtime_passthrough(): + mw = _make_mw() + req = SimpleNamespace(tool_call={"name": "web_search", "id": "tc-1"}) + msg = _make_tool_message() + + result = await mw.awrap_tool_call(req, AsyncMock(return_value=msg)) + + assert result is msg + assert "t1" not in mw._phase_states + + +@pytest.mark.anyio +async def test_awrap_tool_call_command_result_passthrough(): + mw = _make_mw() + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + cmd = Command(goto="some_node") + + result = await mw.awrap_tool_call(req, AsyncMock(return_value=cmd)) + + assert result is cmd + + +@pytest.mark.anyio +async def test_awrap_tool_call_malformed_meta_passthrough(): + """Malformed deerflow_tool_meta dict must not crash the middleware.""" + mw = _make_mw() + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + bad_msg = ToolMessage( + content="some content", + tool_call_id="tc-web_search", + name="web_search", + status="success", + additional_kwargs={TOOL_META_KEY: {"unexpected_field": True}}, + ) + + result = await mw.awrap_tool_call(req, AsyncMock(return_value=bad_msg)) + + assert result is bad_msg + # No state was tracked — malformed meta is silently skipped + assert mw._phase_states.get("t1", {}).get("web_search") is None + + +@pytest.mark.anyio +async def test_awrap_model_call_drains_and_injects_hints(): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + error_msg = _make_error_message() + + # Trigger hint via sync path (state machine is shared) + mw.wrap_tool_call(req, lambda r: error_msg) + mw.wrap_tool_call(req, lambda r: error_msg) + + model_req = _make_model_request([], rt) + captured: list = [] + + async def model_handler(r): + captured.extend(r.messages) + return MagicMock() + + await mw.awrap_model_call(model_req, model_handler) + + hint_msgs = [m for m in captured if isinstance(m, HumanMessage)] + assert any("PROGRESS HINT" in m.content for m in hint_msgs) + + +# --------------------------------------------------------------------------- +# Logging behavior + +_MW_LOGGER = "deerflow.agents.middlewares.tool_progress_middleware" + + +def test_log_active_to_warned_emits_info(caplog): + mw = _make_mw(stagnation_threshold=2) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + error_msg = _make_error_message() + + with caplog.at_level(logging.INFO, logger=_MW_LOGGER): + mw.wrap_tool_call(req, lambda _r: error_msg) + mw.wrap_tool_call(req, lambda _r: error_msg) + + info_records = [r for r in caplog.records if r.levelname == "INFO" and "WARNED" in r.message] + assert len(info_records) == 1 + assert "web_search" in info_records[0].message + + +def test_log_immediate_block_emits_warning(caplog): + mw = _make_mw(stagnation_threshold=5) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + auth_msg = _make_error_message( + error_type="auth", + recoverable_by_model=False, + recommended_next_action="stop", + ) + + with caplog.at_level(logging.WARNING, logger=_MW_LOGGER): + mw.wrap_tool_call(req, lambda _r: auth_msg) + + warning_records = [r for r in caplog.records if r.levelname == "WARNING" and "BLOCKED" in r.message] + assert len(warning_records) == 1 + assert "web_search" in warning_records[0].message + + +def test_log_escalation_block_emits_warning(caplog): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=2) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + error_msg = _make_non_recoverable_error_message() + + with caplog.at_level(logging.WARNING, logger=_MW_LOGGER): + for _ in range(4): + mw.wrap_tool_call(req, lambda _r: error_msg) + + warning_records = [r for r in caplog.records if r.levelname == "WARNING" and "BLOCKED" in r.message] + assert len(warning_records) == 1 + + +def test_log_blocked_call_intercepted_emits_info(caplog): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=1) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + error_msg = _make_non_recoverable_error_message() + + for _ in range(3): + mw.wrap_tool_call(req, lambda _r: error_msg) + + with caplog.at_level(logging.INFO, logger=_MW_LOGGER): + mw.wrap_tool_call(req, lambda _r: error_msg) + + intercepted = [r for r in caplog.records if "intercepted" in r.message and "web_search" in r.message] + assert len(intercepted) == 1 + + +def test_log_warned_to_active_reset_emits_info(caplog): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + error_msg = _make_error_message() + good_msg = _make_tool_message("A" * 300) + + # Drive to WARNED + mw.wrap_tool_call(req, lambda _r: error_msg) + mw.wrap_tool_call(req, lambda _r: error_msg) + + with caplog.at_level(logging.INFO, logger=_MW_LOGGER): + mw.wrap_tool_call(req, lambda _r: good_msg) + + reset_records = [r for r in caplog.records if r.levelname == "INFO" and "ACTIVE" in r.message] + assert len(reset_records) == 1 + assert "web_search" in reset_records[0].message + + +def test_log_hint_injection_emits_debug(caplog): + mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5) + rt = _make_runtime() + req = _make_tool_request(runtime=rt) + error_msg = _make_error_message() + + mw.wrap_tool_call(req, lambda _r: error_msg) + mw.wrap_tool_call(req, lambda _r: error_msg) + + model_req = _make_model_request([], rt) + with caplog.at_level(logging.DEBUG, logger=_MW_LOGGER): + mw.wrap_model_call(model_req, lambda _r: MagicMock()) + + debug_records = [r for r in caplog.records if r.levelname == "DEBUG" and "injecting" in r.message] + assert len(debug_records) == 1 + assert "injecting 1 hint" in debug_records[0].message + + +# --------------------------------------------------------------------------- +# Coexistence: ToolProgressMiddleware + LoopDetectionMiddleware + + +def test_tool_progress_and_loop_detection_coexist_without_interfering(): + """ToolProgressMiddleware and LoopDetectionMiddleware operate on separate signals + and must not interfere when both are active simultaneously. + + ToolProgressMiddleware (position 8): result-quality guard, fires after tool execution, + tracks per-(thread, tool) stagnation, BLOCKs specific tools. + LoopDetectionMiddleware (position 19): call-pattern guard, fires after model response, + tracks repeated tool_call signatures, hard-stops the whole turn. + + Both can inject HumanMessage hints in the same model call; neither reads or writes + the other's internal state. + """ + from langchain_core.messages import AIMessage + + from deerflow.agents.middlewares.loop_detection_middleware import LoopDetectionMiddleware + + tp_mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5, inject_assessment=True) + ld_mw = LoopDetectionMiddleware(warn_threshold=3, hard_limit=10) + + tp_rt = _make_runtime(thread_id="t1", run_id="r1") + # LoopDetection uses its own runtime/thread context + ld_rt = _make_runtime(thread_id="ld-thread", run_id="ld-run") + req = _make_tool_request(runtime=tp_rt) + + # --- Drive ToolProgress to WARNED via repeated error results (result-quality signal) --- + error_msg = _make_error_message() # recoverable error, stagnation_threshold=2 + tp_mw.wrap_tool_call(req, lambda _: error_msg) + tp_mw.wrap_tool_call(req, lambda _: error_msg) + + assert tp_mw._phase_states["t1"]["web_search"].phase == "warned" + tp_hints = list(tp_mw._pending.get(("t1", "r1"), [])) + assert len(tp_hints) == 1, "ToolProgress must queue exactly one hint at stagnation" + + # --- Drive LoopDetection to WARNED via repeated AIMessage tool_calls (call-pattern signal) --- + repeated_call = [{"name": "web_search", "args": {"query": "q"}, "id": "tc-1"}] + ld_state = {"messages": [AIMessage(content="", tool_calls=repeated_call)]} + for _ in range(3): # warn_threshold=3 + ld_mw._apply(ld_state, ld_rt) + + ld_warnings_live = ld_mw._pending_warnings.get(("ld-thread", "ld-run"), []) + assert len(ld_warnings_live) >= 1, "LoopDetection must queue at least one warning" + # Snapshot a copy so the final cross-contamination check compares a frozen + # baseline to the live state — a same-object comparison would always be True. + ld_warnings_snapshot = list(ld_warnings_live) + + # --- Verify no cross-contamination between the two middlewares --- + # ToolProgress internal state is not visible to LoopDetection + assert not hasattr(ld_mw, "_phase_states"), "LoopDetection must not have _phase_states" + # LoopDetection internal state is not visible to ToolProgress + assert not hasattr(tp_mw, "_history"), "ToolProgress must not have _history" + # LoopDetection does not track ToolProgress's thread id + assert "t1" not in ld_mw._history, "LoopDetection must not have entries for ToolProgress's thread" + # ToolProgress does not have loop detection warnings + assert not any("LOOP" in h for h in tp_hints), "ToolProgress hints must not contain loop-detection text" + + # --- ToolProgress hint injection is independent of LoopDetection --- + model_req = _make_model_request([], tp_rt) + captured: list = [] + + def capture_handler(r): + captured.extend(r.messages) + return MagicMock() + + tp_mw.wrap_model_call(model_req, capture_handler) + injected = [m for m in captured if isinstance(m, HumanMessage)] + assert len(injected) == 1, "ToolProgress must inject exactly one hint message" + assert "PROGRESS HINT" in injected[0].content + + # After ToolProgress drains, its queue is empty; LoopDetection warnings unchanged. + # Compare live state against the snapshot taken before the model call — a same-object + # comparison would be trivially True and would not detect accidental modifications. + assert tp_mw._pending.get(("t1", "r1"), []) == [] + assert ld_mw._pending_warnings.get(("ld-thread", "ld-run"), []) == ld_warnings_snapshot diff --git a/backend/tests/test_tool_result_meta.py b/backend/tests/test_tool_result_meta.py new file mode 100644 index 000000000..d5be2a15d --- /dev/null +++ b/backend/tests/test_tool_result_meta.py @@ -0,0 +1,474 @@ +"""Tests for tool_result_meta normalization logic.""" + +from __future__ import annotations + +import json + +import pytest +from langchain_core.messages import ToolMessage +from langgraph.types import Command + +from deerflow.agents.middlewares.tool_result_meta import ( + TOOL_META_KEY, + ToolResultMeta, + normalize_tool_message, + normalize_tool_result, + stamp_exception_meta, +) + + +def _make_msg(content: str, *, status: str = "success", kwargs: dict[str, object] | None = None) -> ToolMessage: + return ToolMessage( + content=content, + tool_call_id="tc-1", + name="test_tool", + status=status, + additional_kwargs=kwargs or {}, + ) + + +def _meta(msg: ToolMessage) -> dict[str, object]: + return msg.additional_kwargs[TOOL_META_KEY] + + +# --------------------------------------------------------------------------- +# Already-stamped messages are not overwritten + + +def test_existing_meta_is_preserved(): + existing = {"status": "success", "source": "custom"} + msg = _make_msg("hello", kwargs={TOOL_META_KEY: existing}) + result = normalize_tool_message(msg) + assert result.additional_kwargs[TOOL_META_KEY] is existing + + +# --------------------------------------------------------------------------- +# Error prefix (tool_return path) + + +@pytest.mark.parametrize( + "snippet,expected_type", + [ + ("Error: 401 unauthorized", "auth"), + ("Error: permission denied for path", "permission"), + ("Error: 429 rate limit exceeded", "rate_limited"), + ("Error: connection timeout", "transient"), + ("Error: tool not configured", "config"), + ("Error: no results found for query", "no_results"), + ("Error: file not found", "not_found"), + ("Error: internal error 500", "internal"), + ("Error: something completely unexpected happened", "unknown"), + ], +) +def test_error_prefix_classification(snippet: str, expected_type: str): + msg = _make_msg(snippet, status="error") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "error" + assert m["error_type"] == expected_type + assert m["source"] == "tool_return" + + +def test_auth_error_is_unrecoverable_and_stop(): + msg = _make_msg("Error: invalid api key", status="error") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["recoverable_by_model"] is False + assert m["recommended_next_action"] == "stop" + + +def test_no_api_key_is_config_not_auth(): + # Distinguish "missing key" (config) from "wrong key" (auth): + # - auth rule keyword: "invalid api key" (key provided but rejected) + # - config rule keyword: "no api key" (key not set at all) + # The two phrases do not overlap, so rule order does not affect this particular + # case. This test documents the semantic distinction — a missing API key is a + # configuration issue, not an authentication failure. + msg = _make_msg("Error: no api key configured", status="error") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["error_type"] == "config", "missing API key is a config issue, not auth" + assert m["recommended_next_action"] == "stop" + assert m["recoverable_by_model"] is False + + +def test_rate_limited_error_suggests_summarize(): + msg = _make_msg("Error: rate limited", status="error") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["recommended_next_action"] == "summarize" + assert m["recoverable_by_model"] is False + + +def test_no_results_suggests_rewrite_query(): + msg = _make_msg("Error: no results found", status="error") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["recoverable_by_model"] is True + assert m["recommended_next_action"] == "rewrite_query" + + +# --------------------------------------------------------------------------- +# Non-standard error path (status="error", no "Error:" prefix) + + +def test_nonstd_error_status_classifies_from_content(): + # Tools that return status="error" without the "Error:" prefix are tool_return, not exception. + # Actual exceptions are pre-stamped by stamp_exception_meta and exit normalize_tool_message early. + msg = _make_msg("ConnectionError: connection refused", status="error") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "error" + assert m["source"] == "tool_return" + assert m["error_type"] == "transient" + + +def test_nonstd_error_status_timeout_content(): + msg = _make_msg("timeout occurred", status="error") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["source"] == "tool_return" + assert m["error_type"] == "transient" + + +def test_nonstd_error_status_json_classifies_from_error_field(): + # When status="error" and content is JSON, classification must use only the "error" + # field value — not keywords that appear in other fields like "query". + content = '{"error": "api limit exceeded", "query": "connection test timeout"}' + msg = _make_msg(content, status="error") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "error" + assert m["source"] == "tool_return" + # "connection" and "timeout" in query must not trigger transient; the error field + # "api limit exceeded" doesn't match any rule → unknown. + assert m["error_type"] == "unknown" + + +def test_nonstd_error_status_json_no_error_key_is_unknown(): + # JSON with no 'error' key must NOT be classified from other field values. + # Previously, {"message": "connection refused"} would be passed to _classify_error_text + # and match the transient rule via "connection"; now the full JSON is treated as unknown. + content = '{"message": "connection refused"}' + msg = _make_msg(content, status="error") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "error" + assert m["error_type"] == "unknown", "JSON dict with no 'error' key must resolve to unknown" + + +def test_nonstd_error_status_json_no_error_key_with_dangerous_field_is_unknown(): + # {"user_id": 401} previously triggered auth stop; must now be unknown. + content = '{"user_id": 401, "action": "login"}' + msg = _make_msg(content, status="error") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "error" + assert m["error_type"] == "unknown" + assert m["recommended_next_action"] != "stop", "spurious 401 in non-error field must not trigger stop" + + +def test_nonstd_error_status_non_json_content_still_classified(): + # Plain text (not JSON) with status="error" must still be classified from content. + content = "connection refused: remote host unreachable" + msg = _make_msg(content, status="error") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "error" + assert m["error_type"] == "transient" + + +def test_json_error_field_dict_is_serialized_not_repr(): + # FastAPI-style: {"error": [{"loc": ["body"], "msg": "missing required field"}]} + # str() would produce Python repr containing 'missing required' → config → stop. + # json.dumps produces a clean JSON string that should not spuriously match. + import json as _json + + error_val = [{"loc": ["body"], "msg": "missing required field"}] + content = _json.dumps({"error": error_val}) + msg = _make_msg(content, status="error") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "error" + # The JSON-serialized error list contains "missing required field" which IS in the config rule. + # This is correct classification (the validation error IS a config-class problem). + # The key requirement is that we're classifying from the error field value, not repr noise. + assert m["error_type"] == "config" + + +def test_no_results_success_response_is_partial_success(): + # Tools that return status="success" with "no results found" content must be treated as + # partial_success so ToolProgressMiddleware can detect stagnation. + for phrase in ("no results found", "No Content Found here", "no images found for query"): + msg = _make_msg(phrase, status="success") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "partial_success", f"expected partial_success for: {phrase!r}" + assert m["recommended_next_action"] == "rewrite_query" + + +# --------------------------------------------------------------------------- +# Partial success detection + + +def test_partial_markers_detected(): + for marker in ("partial results available", "limited results returned", "truncated output", "results may be incomplete"): + msg = _make_msg(f"Here are some {marker} from the search.", status="success") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "partial_success", f"expected partial_success for: {marker}" + assert m["recommended_next_action"] == "rewrite_query" + + +def test_short_terse_success_is_not_partial(): + # "Ok." is a valid, complete success response from mutation tools like write_file/str_replace. + # partial_success is now gated only on _PARTIAL_MARKERS, not content length. + msg = _make_msg("Ok.", status="success") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "success" + assert m["source"] == "content_analysis" + + +def test_empty_content_is_not_partial(): + # Empty content has no partial markers, so it falls through to success. + msg = _make_msg("", status="success") + result = normalize_tool_message(msg) + m = _meta(result) + # Empty content falls through to success (no partial markers) + assert m["status"] == "success" + + +# --------------------------------------------------------------------------- +# Success path + + +def test_substantial_content_is_success(): + content = "A" * 200 + msg = _make_msg(content, status="success") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "success" + assert m["source"] == "content_analysis" + assert m["recommended_next_action"] == "continue" + assert m["error_type"] is None + + +# --------------------------------------------------------------------------- +# ToolResultMeta dataclass round-trip + + +def test_tool_result_meta_from_dict(): + msg = _make_msg("A" * 200) + result = normalize_tool_message(msg) + meta_dict = _meta(result) + meta = ToolResultMeta(**meta_dict) + assert meta.status == "success" + assert meta.error_type is None + assert meta.recommended_next_action == "continue" + + +# --------------------------------------------------------------------------- +# stamp_exception_meta + + +def test_stamp_exception_meta_classifies_from_exc_info_not_content(): + # Content says "no results" but exc_info says "connection refused" — + # stamp_exception_meta must use exc_info, producing transient, not no_results. + msg = _make_msg("Error: no results found", status="error") + result = stamp_exception_meta(msg, "ConnectionError: connection refused") + m = _meta(result) + assert m["source"] == "exception" + assert m["error_type"] == "transient" + + +def test_stamp_exception_meta_overwrites_existing_meta(): + pre_existing = {TOOL_META_KEY: {"source": "tool_return", "error_type": "unknown"}} + msg = _make_msg("Error: no results found", status="error", kwargs=pre_existing) + result = stamp_exception_meta(msg, "PermissionError: access denied") + m = _meta(result) + assert m["source"] == "exception" + assert m["error_type"] == "permission" + + +def test_stamp_exception_meta_preserves_other_additional_kwargs(): + msg = _make_msg("irrelevant", status="error", kwargs={"subagent_status": "running"}) + result = stamp_exception_meta(msg, "TimeoutError: timed out") + assert result.additional_kwargs["subagent_status"] == "running" + assert TOOL_META_KEY in result.additional_kwargs + + +# --------------------------------------------------------------------------- +# normalize_tool_result handles Command wrappers + + +def test_normalize_tool_result_passthrough_command(): + cmd = Command(goto="next_node") + result = normalize_tool_result(cmd) + assert result is cmd + + +def test_normalize_tool_result_stamps_tool_message(): + msg = _make_msg("A" * 200) + result = normalize_tool_result(msg) + assert isinstance(result, ToolMessage) + assert TOOL_META_KEY in result.additional_kwargs + + +# --------------------------------------------------------------------------- +# JSON-wrapped error detection + + +def test_normalize_json_error_config_classified_as_error(): + content = '{"error": "BRAVE_SEARCH_API_KEY is not configured", "query": "test"}' + msg = _make_msg(content) + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "error" + assert m["error_type"] == "config" + assert m["source"] == "tool_return" + + +def test_normalize_json_error_no_results_classified_correctly(): + content = '{"error": "No results found", "query": "test"}' + msg = _make_msg(content) + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "error" + assert m["error_type"] == "no_results" + assert m["recoverable_by_model"] is True + + +def test_normalize_json_null_error_not_treated_as_error(): + content = '{"error": null, "query": "test"}' + msg = _make_msg(content) + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] != "error" + + +def test_normalize_json_no_error_key_not_treated_as_error(): + content = '{"results": [{"title": "page one", "url": "https://example.com/one", "content": "summary one"}], "total": 1}' + msg = _make_msg(content) + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "success" + + +def test_normalize_malformed_json_not_treated_as_error(): + content = '{"error": "broken json' + msg = _make_msg(content) + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] != "error" + + +def test_normalize_json_error_with_leading_whitespace(): + content = ' {"error": "No results found", "query": "test"}' + msg = _make_msg(content) + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "error" + assert m["error_type"] == "no_results" + + +def test_normalize_json_numeric_error_classified_correctly(): + content = '{"error": 404, "query": "test"}' + msg = _make_msg(content) + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "error" + assert m["error_type"] == "not_found" + + +def test_normalize_json_zero_error_not_treated_as_error(): + content = '{"error": 0, "query": "test"}' + msg = _make_msg(content) + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] != "error" + + +def test_normalize_json_false_error_not_treated_as_error(): + content = '{"error": false, "query": "test"}' + msg = _make_msg(content) + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] != "error" + + +def test_normalize_json_boolean_true_error_classified_as_unknown(): + """Boolean True in the error field means 'an error occurred' and must be classified. + + str(True) = "True" which matches no keyword rule, so the result is error/unknown. + This is intentional: a boolean True error is a real error with no further detail. + """ + content = '{"error": true, "query": "test"}' + msg = _make_msg(content) + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "error" + assert m["error_type"] == "unknown" # str(True)="True" matches no keyword rule + assert m["recoverable_by_model"] is True + assert m["recommended_next_action"] == "try_alternative" + + +# --------------------------------------------------------------------------- +# M2 regression: semantic-zero error strings must NOT be treated as errors + + +@pytest.mark.parametrize( + "error_value", + ["none", "None", "NONE", "null", "Null", "false", "False", "no", "ok", "success", "n/a", ""], +) +def test_normalize_json_semantic_zero_error_string_not_treated_as_error(error_value: str): + """M2 regression: error field containing a conventional 'no-error' string must not trigger misclassification. + + Tools sometimes return {"error": "none", "results": [...]} on success. + The string "none" is truthy in Python, so without this guard the message + would have been classified as error (unknown), inflating stagnation counters. + + Note: the empty-string case ("") is handled by the falsy guard (`if not error: return None`) + in _extract_json_error_text rather than by _SEMANTIC_ZERO_ERROR_STRINGS. Both paths produce + the same outcome (no misclassification), but the mechanism differs from the other cases here. + """ + content = json.dumps({"error": error_value, "results": ["item1", "item2", "item3"]}) + msg = _make_msg(content, status="success") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] != "error", f'error="{error_value}" should not be treated as an error; got status={m["status"]!r}' + + +# --------------------------------------------------------------------------- +# Numeric keyword word-boundary matching (_match_keyword) + + +@pytest.mark.parametrize( + "content, expected_error_type", + [ + # Positive: numeric code at a word boundary → correct classification + ("Error: HTTP 500 Internal Server Error", "internal"), + ("Error: returned status 500", "internal"), + ("Error: 401 Unauthorized", "auth"), + ("Error: 404 Not Found", "not_found"), + # Negative: numeric code embedded inside a longer token → must resolve to "unknown". + # Use exact "unknown" assertions so any future rule additions that accidentally + # absorb these strings are caught (a broad exclusion list would miss new matches). + ("Error: took 500ms to respond", "unknown"), + ("Error: query returned 4010 rows", "unknown"), + ("Error: batch 401A failed", "unknown"), + ("Error: response contained 5000 items", "unknown"), + ], +) +def test_numeric_keyword_word_boundary(content: str, expected_error_type: str): + """Numeric HTTP codes must match only at word boundaries to avoid false positives. + + '500ms', '4010', '401A', '5000' must not trigger internal/auth/not_found rules. + Negative cases assert exactly 'unknown' so future rule additions that accidentally + absorb these strings are caught — a broad exclusion-list assertion would not be. + """ + msg = _make_msg(content, status="error") + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "error" + assert m["error_type"] == expected_error_type, f"{content!r} → expected {expected_error_type!r}, got {m['error_type']!r}" diff --git a/config.example.yaml b/config.example.yaml index 7cb90a0aa..f48a33da6 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -15,7 +15,7 @@ # ============================================================================ # Bump this number when the config schema changes. # Run `make config-upgrade` to merge new fields into your local config.yaml. -config_version: 18 +config_version: 19 # ============================================================================ # Logging @@ -934,6 +934,31 @@ loop_detection: # warn: 150 # hard_limit: 300 +# ============================================================================ +# Tool Progress State Machine Configuration (RFC #3177) +# ============================================================================ +# Detects tool stagnation and repetition at the (thread, tool) level. +# Tracks consecutive "no-new-info" calls (error, partial_success, near-duplicate success). +# Three transition paths (determined by deerflow_tool_meta.recoverable_by_model): +# recoverable=true (no_results, not_found, permission): ACTIVE → WARNED (terminal; hint re-injected each call) +# recoverable=false (rate_limited, transient): ACTIVE → WARNED → BLOCKED after warn_escalation_count more +# recoverable=false + action=stop (auth, config): ACTIVE → BLOCKED immediately +# Requires ToolErrorHandlingMiddleware to be active (always on). + +# tool_progress: +# enabled: false +# stagnation_threshold: 3 # Consecutive problems before WARNED +# warn_escalation_count: 2 # More problems after WARNED before BLOCKED +# inject_assessment: true +# jaccard_similarity_threshold: 0.8 # Word-set similarity threshold for near-duplicate detection +# min_word_count_for_similarity: 10 # Min unique words to apply Jaccard check +# max_tracked_threads: 100 +# exempt_tools: +# - ask_clarification +# - write_todos +# - present_files +# - task + # ============================================================================ # Read-Before-Write File Gate (issue #3857) # ============================================================================