mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-09 16:08:31 +00:00
feat(middleware): add structured tool result meta and tool-progress state machine (#3601)
* 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 <noreply@anthropic.com>
* 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 <noreply@anthropic.com>
* 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 <noreply@anthropic.com>
* 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 <noreply@anthropic.com>
* 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 <noreply@anthropic.com>
* 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 <noreply@anthropic.com>
* 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 <noreply@anthropic.com>
* 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 <noreply@anthropic.com>
* 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 <noreply@anthropic.com>
---------
Co-authored-by: Claude Sonnet 4 <noreply@anthropic.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
He Wang
GitHub
parent
f122594419
commit
fd41fdb065
12 changed files with 2988 additions and 26 deletions
|
|
@ -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 `<system-reminder>` 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 `<system-reminder>` 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
|
||||
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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]:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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)")
|
||||
|
|
|
|||
|
|
@ -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)",
|
||||
)
|
||||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
1479
backend/tests/test_tool_progress_middleware.py
Normal file
1479
backend/tests/test_tool_progress_middleware.py
Normal file
File diff suppressed because it is too large
Load diff
474
backend/tests/test_tool_result_meta.py
Normal file
474
backend/tests/test_tool_result_meta.py
Normal file
|
|
@ -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}"
|
||||
|
|
@ -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)
|
||||
# ============================================================================
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue