diff --git a/backend/docs/RUN_EVENT_STREAM.md b/backend/docs/RUN_EVENT_STREAM.md index 982cabfa9..cb92861fd 100644 --- a/backend/docs/RUN_EVENT_STREAM.md +++ b/backend/docs/RUN_EVENT_STREAM.md @@ -76,10 +76,24 @@ through run-event or specialized APIs: | `context:memory` | `context` | `record_memory_context()` | | `middleware:{tag}` | `middleware` | `record_middleware()` | -Current middleware tags are `guardrail`, `safety_termination`, -`skill_activation`, and `skill_secrets`. The pattern is intentionally open so -new middleware tags are additive. Because the full event type is limited to 32 -characters and `middleware:` uses 11, a tag must contain 1-21 characters. +Current middleware tags are `guardrail`, `loop_detection`, +`safety_termination`, `skill_activation`, and `skill_secrets`. The pattern is +intentionally open so new middleware tags are additive. Because the full event +type is limited to 32 characters and `middleware:` uses 11, a tag must contain +1-21 characters. + +`middleware:loop_detection` records transitions into the warned state (first +per call hash or per tool-frequency burst) and each hard stop produced by +`LoopDetectionMiddleware` in lead-agent and ordinary task-tool subagent runs. +Task-tool subagents forward the append to the parent loop because `RunJournal` +and its event store must not cross the isolated-loop boundary. Durable batch +subagents have no parent run journal and do not emit these events. The event's +`action` is `warn` or +`hard_stop`. The `changes` object identifies the detection layer, affected tool +names, observed count, effective threshold, whether the producer was a +subagent, and its agent id when applicable. Tool arguments, prompts, message +content, tool results, and argument-derived hashes are not persisted in this +event. ### Opaque Run Outputs @@ -176,7 +190,7 @@ be used by new producers. - Nested non-JSON values in `run.end.content` have backend-dependent representations: memory retains Python values, while JSONL and database stores read them back as strings. -- Loop detection and deferred-tool promotion do not currently emit middleware - events. +- Durable batch subagent loop detection and deferred-tool promotion do not + currently emit middleware events. - Journal attribution, token accounting, and external tracing metadata still depend on manual instrumentation at several LLM call sites. diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md index 714d21c33..a5c1b13c0 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md @@ -95,7 +95,7 @@ Before changing a later authorization phase, read the [authorization RFC](../../ 25. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` or `McpRoutingMiddleware` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped) 26. **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. The subagent builder places its date-only context middleware immediately before this coalescer, so the built-in subagent prompt and hidden date reminder still reach providers as one leading system block 27. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess ordinary `task` tool calls to enforce both the per-response concurrency limit (`max_concurrent_subagents`, resolved against startup `subagent_runtime.max_running` and the 1-64 safety range before construction) and the per-run total delegation cap (`max_total_subagents` runtime override or `subagents.max_total_per_run`, default 6, clamped to 1-50). The total cap counts current-run entries in the durable delegation ledger (entries are tagged with `run_id` when captured), so repeated planning checkpoints in one run cannot keep launching legal-sized batches indefinitely, while later user turns in the same thread get a fresh run budget. Explicit durable `batch_task` calls are a separate mode with persisted total/live/running limits and are not rewritten into ordinary ledger entries. If the ordinary cap is exhausted, the middleware strips remaining `task` calls, forces `finish_reason="stop"`, and appends a visible limit note so the run can synthesize existing results instead of ending with an empty tool-call response. -28. **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; stamps `loop_capped` via `consume_stop_reason` (#3875 Phase 2), symmetric to `TokenBudgetMiddleware` +28. **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; stamps `loop_capped` via `consume_stop_reason` (#3875 Phase 2), symmetric to `TokenBudgetMiddleware`; persists warned-state transitions (first per call hash or per tool-frequency burst) and hard stops as `middleware:loop_detection`, attributed with `is_subagent` and the optional `agent_id`, without tool arguments, message content, tool results, or argument-derived hashes. Ordinary task-tool subagents receive a loop-detection-only recorder proxy that forwards the append to the parent run loop; never pass the loop-bound `RunJournal` itself into their isolated event loop. Durable batch subagents have no parent run journal and do not persist these transitions 29. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits 30. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before config-declared extensions and the terminal-response/safety/clarification tail 31. **Configured extension middlewares** - *(optional, if `extensions.middlewares` is set in `config.yaml` or `extensions_config.json`)* Zero-argument `AgentMiddleware` classes loaded from `module.path:ClassName` entries via `deerflow.reflection.resolve_class`. Missing packages, invalid classes, and broken modules fail loudly at agent creation. These run after built-ins/programmatic custom middleware and after the lead/subagent loop/token guards, but before the terminal-response/safety/clarification tail; subagents receive the same configured extension middleware class list before their safety tail. Treat these files as trusted operator config because middleware paths instantiate arbitrary code. Gateway skill/MCP toggle endpoints preserve this field through `to_file_dict()` but must not add a write path for `extensions.middlewares` without an explicit trust-boundary review. Lead-only vs subagent-only middleware lists and per-context constructor parameters are not expressible in this MVP. diff --git a/backend/packages/harness/deerflow/agents/middlewares/audit_context.py b/backend/packages/harness/deerflow/agents/middlewares/audit_context.py new file mode 100644 index 000000000..a84e442ff --- /dev/null +++ b/backend/packages/harness/deerflow/agents/middlewares/audit_context.py @@ -0,0 +1,3 @@ +"""Private runtime-context keys for narrowly scoped middleware audit recorders.""" + +LOOP_DETECTION_RECORDER_CONTEXT_KEY = "__run_loop_detection_recorder" diff --git a/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py index 3f0efcc2f..165f9d910 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py @@ -58,7 +58,8 @@ import threading from collections import Counter, OrderedDict, defaultdict, deque from collections.abc import Awaitable, Callable from copy import deepcopy -from typing import TYPE_CHECKING, override +from dataclasses import dataclass +from typing import TYPE_CHECKING, Literal, override from langchain.agents import AgentState from langchain.agents.middleware import AgentMiddleware @@ -67,6 +68,8 @@ from langchain_core.messages import HumanMessage from langgraph.runtime import Runtime from deerflow.agents.middlewares._bounded_dict import BoundedDict +from deerflow.agents.middlewares.audit_context import LOOP_DETECTION_RECORDER_CONTEXT_KEY +from deerflow.runtime.events.catalog import MIDDLEWARE_LOOP_DETECTION_TAG if TYPE_CHECKING: from deerflow.config.loop_detection_config import LoopDetectionConfig @@ -184,6 +187,22 @@ _HARD_STOP_MSG = "[FORCED STOP] Repeated tool calls exceeded the safety limit. P _TOOL_FREQ_HARD_STOP_MSG = "[FORCED STOP] Tool {tool_name} called {count} times — exceeded the per-tool safety limit. Producing final answer with results collected so far." +@dataclass(frozen=True) +class _LoopDecision: + """A loop-detection transition that may be persisted for audit.""" + + message: str + action: Literal["warn", "hard_stop"] + detection_layer: Literal["identical_call_set", "tool_frequency"] + tool_names: tuple[str, ...] + count: int + threshold: int + + @property + def hard_stop(self) -> bool: + return self.action == "hard_stop" + + class LoopDetectionMiddleware(AgentMiddleware[AgentState]): """Detects and breaks repetitive tool call loops. @@ -416,7 +435,11 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): self._touch_pending_warning_key_locked(pending_key) self._prune_pending_warning_state_locked(protected_key=pending_key) - def _track_and_check(self, state: AgentState, runtime: Runtime) -> tuple[str | None, bool]: + def _track_and_check( + self, + state: AgentState, + runtime: Runtime, + ) -> _LoopDecision | None: """Track tool calls and check for loops. Two detection layers: @@ -426,19 +449,19 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): on 40 different files). Returns: - (warning_message_or_none, should_hard_stop) + A structured decision when a warning or hard stop is triggered. """ messages = state.get("messages", []) if not messages: - return None, False + return None last_msg = messages[-1] if getattr(last_msg, "type", None) != "ai": - return None, False + return None tool_calls = getattr(last_msg, "tool_calls", None) if not tool_calls: - return None, False + return None thread_id = self._get_thread_id(runtime) call_hash = _hash_tool_calls(tool_calls) @@ -463,7 +486,7 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): self._warned.pop(thread_id, None) count = history.count(call_hash) - tool_names = [tc.get("name", "?") for tc in tool_calls] + tool_names = [str(tc.get("name") or "?") for tc in tool_calls] # --- Layer 1: hash-based (identical call sets) --- if count >= self.hard_limit: @@ -476,7 +499,14 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): "tools": tool_names, }, ) - return _HARD_STOP_MSG, True + return _LoopDecision( + message=_HARD_STOP_MSG, + action="hard_stop", + detection_layer="identical_call_set", + tool_names=tuple(tool_names), + count=count, + threshold=self.hard_limit, + ) if count >= self.warn_threshold: warned = self._warned[thread_id] @@ -491,7 +521,14 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): "tools": tool_names, }, ) - return _WARNING_MSG, False + return _LoopDecision( + message=_WARNING_MSG, + action="warn", + detection_layer="identical_call_set", + tool_names=tuple(tool_names), + count=count, + threshold=self.warn_threshold, + ) # --- Layer 2: per-tool-type frequency (windowed) --- tool_name_history = self._tool_name_history[thread_id] @@ -530,7 +567,14 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): "count": freq_count, }, ) - return _TOOL_FREQ_HARD_STOP_MSG.format(tool_name=name, count=freq_count), True + return _LoopDecision( + message=_TOOL_FREQ_HARD_STOP_MSG.format(tool_name=name, count=freq_count), + action="hard_stop", + detection_layer="tool_frequency", + tool_names=(name,), + count=freq_count, + threshold=eff_hard, + ) if freq_count >= eff_warn: freq_warned = self._tool_freq_warned[thread_id] @@ -544,13 +588,20 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): "count": freq_count, }, ) - return _TOOL_FREQ_WARNING_MSG.format(tool_name=name, count=freq_count), False + return _LoopDecision( + message=_TOOL_FREQ_WARNING_MSG.format(tool_name=name, count=freq_count), + action="warn", + detection_layer="tool_frequency", + tool_names=(name,), + count=freq_count, + threshold=eff_warn, + ) else: # Windowed count decayed below the warn threshold; allow a # future burst of this tool to warn again. self._tool_freq_warned[thread_id].discard(name) - return None, False + return None @staticmethod def _append_text(content: str | list | None, text: str) -> str | list: @@ -589,10 +640,55 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): return update - def _apply(self, state: AgentState, runtime: Runtime) -> dict | None: - warning, hard_stop = self._track_and_check(state, runtime) + def _record_audit_event( + self, + decision: _LoopDecision, + runtime: Runtime, + ) -> None: + """Persist a loop-detection transition without sensitive tool data.""" + context = getattr(runtime, "context", None) + is_subagent = isinstance(context, dict) and context.get("is_subagent") is True + recorder = context.get(LOOP_DETECTION_RECORDER_CONTEXT_KEY) if isinstance(context, dict) else None + if recorder is None and isinstance(context, dict): + # Lead-agent runs expose the ordinary RunJournal. Native task-tool + # subagents receive only the narrow, loop-safe recorder key above. + recorder = context.get("__run_journal") + if recorder is None: + return - if hard_stop: + try: + recorder.record_middleware( + tag=MIDDLEWARE_LOOP_DETECTION_TAG, + name=type(self).__name__, + hook="after_model", + action=decision.action, + changes={ + "is_subagent": is_subagent, + "agent_id": context.get("agent_id") if is_subagent else None, + "detection_layer": decision.detection_layer, + "tool_names": list(decision.tool_names), + "count": decision.count, + "threshold": decision.threshold, + }, + ) + except Exception: # noqa: BLE001 + # Audit persistence must never break the agent run. + logger.warning( + "Failed to record middleware:loop_detection event", + exc_info=True, + ) + + def _apply(self, state: AgentState, runtime: Runtime) -> dict | None: + decision = self._track_and_check(state, runtime) + if decision is None: + return None + + # Keep the shared loop-detection lock's critical section bounded; the + # journal append is cheap and non-blocking but does not belong under it. + self._record_audit_event(decision, runtime) + warning = decision.message + + if decision.hard_stop: # Record the stop reason so the executor can surface # ``stop_reason=loop_capped`` after the run returns (#3875 Phase 2). # The hard stop does not raise — it strips tool_calls and lets the @@ -619,7 +715,7 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): stripped_msg = last_msg.model_copy(update=self._build_hard_stop_update(last_msg, content)) return {"messages": [stripped_msg]} - if warning: + if decision.action == "warn": # Defer injection to the next model call. We must NOT alter the # AIMessage(tool_calls=...) here (would put framework words in # the model's mouth, polluting downstream consumers like diff --git a/backend/packages/harness/deerflow/runtime/events/catalog.py b/backend/packages/harness/deerflow/runtime/events/catalog.py index 06c6ba1a9..60e832ab8 100644 --- a/backend/packages/harness/deerflow/runtime/events/catalog.py +++ b/backend/packages/harness/deerflow/runtime/events/catalog.py @@ -77,11 +77,13 @@ MIDDLEWARE_EVENT_PATTERN = RunEventPattern( ) MIDDLEWARE_EVENT_TAG_MAX_LENGTH = RUN_EVENT_TYPE_MAX_LENGTH - len(MIDDLEWARE_EVENT_PATTERN.prefix) MIDDLEWARE_GUARDRAIL_TAG = "guardrail" +MIDDLEWARE_LOOP_DETECTION_TAG = "loop_detection" MIDDLEWARE_SAFETY_TERMINATION_TAG = "safety_termination" MIDDLEWARE_SKILL_ACTIVATION_TAG = "skill_activation" MIDDLEWARE_SKILL_SECRETS_TAG = "skill_secrets" MIDDLEWARE_EVENT_TAGS = ( MIDDLEWARE_GUARDRAIL_TAG, + MIDDLEWARE_LOOP_DETECTION_TAG, MIDDLEWARE_SAFETY_TERMINATION_TAG, MIDDLEWARE_SKILL_ACTIVATION_TAG, MIDDLEWARE_SKILL_SECRETS_TAG, diff --git a/backend/packages/harness/deerflow/subagents/AGENTS.md b/backend/packages/harness/deerflow/subagents/AGENTS.md index f9ce9be44..2855c5f92 100644 --- a/backend/packages/harness/deerflow/subagents/AGENTS.md +++ b/backend/packages/harness/deerflow/subagents/AGENTS.md @@ -21,4 +21,6 @@ **Isolated-loop callback boundary**: sync delegation from an active event loop and `execute_async()` copy the ambient ContextVars into the persistent subagent loop so checkpoint lineage, user identity, tracing context, tags, metadata, and LangGraph's namespaced message-stream handler survive. Before submission, `_copy_isolated_subagent_context()` copies the callback manager/list and removes only handlers marked `deerflow_loop_bound`; `RunJournal` carries that marker because it owns parent-loop tasks and a SQL store/pool. LangGraph merges inherited callbacks with the child run's explicit `SubagentTokenCollector`/tracing callbacks, so letting `RunJournal` cross loops causes duplicate accounting and `Future attached to a different loop` failures, while dropping the whole callback chain silently removes child token frames. Do not replace the boundary with a blank `Context`; the inherited checkpoint namespace and framework stream callback are required by the stream-isolation contract above. +Loop-detection audit records cross this boundary through a deliberately narrow exception: `task_tool` captures the parent loop and passes `_ParentLoopMiddlewareRecorderProxy` to `SubagentExecutor`, which installs it under the loop-detection-only context key and schedules the real journal append back onto its owner loop with `call_soon_threadsafe`. Other subagent middleware consumers still do not see `__run_journal`. The task tool closes the proxy before returning: close fences later child events and yields once on the owner loop so every previously accepted append reaches the journal before the parent run captures completion data. The proxy, not `RunJournal`, crosses into the isolated loop; do not broaden it into a generic journal facade or call the event store from the subagent loop. Durable batch subagents have no parent run journal and therefore do not use this bridge. + **Reverse direction of the loop boundary — deferred cleanup & final usage delivery (#5069)**: when a task-tool poller exits unexpectedly, the registry cleanup is pinned **to** the persistent subagent loop via the public `run_on_isolated_subagent_loop()` (executor) so it survives caller-loop teardown — `asyncio.run()` cancels caller-loop tasks on exit, so a caller-loop `asyncio.create_task` would be cancelled before running. The final usage report crosses the boundary the **other way**: `_schedule_deferred_subagent_cleanup` captures the parent run's loop at unwind time (alive in every path that continues the run — the polling-timeout branch returns normally, and a generic poller error becomes an error `ToolMessage`), and `_deliver_final_usage_report` hands the report back onto that loop with `call_soon_threadsafe`. `record_external_llm_usage_records` must never be invoked from the persistent loop or a worker thread (`to_thread`): the journal's accumulators are unlocked read-modify-write fields and `get_completion_data()` iterates `_tokens_by_model`, so a cross-thread write silently loses token updates or breaks iteration mid-run — calling `_report_subagent_usage` directly inside `_deferred_cleanup_subagent_task` would reintroduce exactly this race. The deferred cleaner captures only the resolved usage recorder (plus ids and the captured report loop) — never the whole `runtime`: the strongly-referenced cleanup task lives for up to the full poll budget, and through `runtime` it would pin the parent run's journal and event store for that entire window. When the captured parent loop is already closed (synchronous `asyncio.run` teardown), the report is dropped on purpose — the run has persisted its completion data and nothing reads the counters back — and logged at info with the execution id and unaccounted record count, because the registry entry is removed right after and those records exist nowhere else. diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py index 2a0990875..5763eefc3 100644 --- a/backend/packages/harness/deerflow/subagents/executor.py +++ b/backend/packages/harness/deerflow/subagents/executor.py @@ -25,6 +25,7 @@ from langchain_core.runnables import RunnableConfig from langchain_core.runnables.config import var_child_runnable_config from langgraph.errors import GraphRecursionError +from deerflow.agents.middlewares.audit_context import LOOP_DETECTION_RECORDER_CONTEXT_KEY from deerflow.agents.thread_state import SandboxState, ThreadDataState, ThreadState from deerflow.authz.principal import normalize_authz_attributes from deerflow.config import get_app_config @@ -774,6 +775,7 @@ class SubagentExecutor: extensions: Any | None = None, execution_capacity: SubagentExecutionCapacity | None = None, acceptance_criteria: list[str] | None = None, + loop_detection_recorder: Any | None = None, ): """Initialize the executor. @@ -814,6 +816,10 @@ class SubagentExecutor: ``HumanMessage`` (the channel ``InputSanitizationMiddleware`` sanitizes and boundary-frames); the subagent's ``SystemMessage`` carries only the framework-owned pointer note. + loop_detection_recorder: Optional loop-safe recorder supplied by the + parent task tool. Native subagents execute on a separate event + loop, so this must be a proxy rather than the parent + ``RunJournal`` itself. """ self.config = config self.app_config = app_config @@ -860,6 +866,7 @@ class SubagentExecutor: # Raw lead-supplied criteria; stripping/capping happens at render time # in report_contract.render_acceptance_criteria_block. self.acceptance_criteria = acceptance_criteria + self.loop_detection_recorder = loop_detection_recorder self._base_tools = _filter_tools( tools, @@ -1458,6 +1465,9 @@ class SubagentExecutor: context["authz_attributes"] = dict(self.authz_attributes) context[DEERFLOW_TRACE_METADATA_KEY] = self.deerflow_trace_id context["is_subagent"] = True + context["agent_id"] = self.config.name + if self.loop_detection_recorder is not None: + context[LOOP_DETECTION_RECORDER_CONTEXT_KEY] = self.loop_detection_recorder logger.info(f"[trace={self.trace_id}] Subagent {self.config.name} starting async execution with max_turns={self.config.max_turns}") diff --git a/backend/packages/harness/deerflow/tools/builtins/task_tool.py b/backend/packages/harness/deerflow/tools/builtins/task_tool.py index f2c928d60..1ddaeca6c 100644 --- a/backend/packages/harness/deerflow/tools/builtins/task_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/task_tool.py @@ -3,6 +3,7 @@ import asyncio import concurrent.futures import logging +import threading import time import uuid from contextvars import ContextVar @@ -75,6 +76,63 @@ _explicit_app_config: ContextVar[Any | None] = ContextVar( ) +def _record_middleware_on_parent_loop(journal: Any, kwargs: dict[str, Any]) -> None: + """Run one subagent middleware-journal append on the journal owner's loop.""" + try: + journal.record_middleware(**kwargs) + except Exception: + logger.warning("Failed to record subagent middleware event", exc_info=True) + + +class _ParentLoopMiddlewareRecorderProxy: + """Forward subagent loop-detection events to the parent run's event loop. + + ``RunJournal`` owns parent-loop tasks and may wrap an event store backed by + a loop-bound SQL pool. Subagents execute on a persistent isolated loop, so + the journal object itself must never be called there. + """ + + def __init__(self, journal: Any, loop: asyncio.AbstractEventLoop) -> None: + self._journal = journal + self._loop = loop + self._state_lock = threading.Lock() + self._closed = False + + def record_middleware(self, **kwargs: Any) -> None: + with self._state_lock: + if self._closed or self._loop.is_closed(): + logger.debug("Dropping subagent middleware event after parent loop shutdown") + return + try: + self._loop.call_soon_threadsafe( + _record_middleware_on_parent_loop, + self._journal, + dict(kwargs), + ) + except RuntimeError: + # The loop may close between is_closed() and scheduling. + logger.debug("Dropping subagent middleware event after parent loop shutdown") + + @property + def is_closed(self) -> bool: + """Whether the task-tool boundary has fenced new child events.""" + with self._state_lock: + return self._closed + + async def aclose(self) -> None: + """Fence late child events and drain every append accepted before it.""" + if asyncio.get_running_loop() is not self._loop: + logger.warning("Cannot drain subagent middleware recorder from a non-owner loop") + return + with self._state_lock: + self._closed = True + if self._loop.is_closed(): + return + # record_middleware holds _state_lock through call_soon_threadsafe, so + # all accepted callbacks are already ahead of this continuation. + await asyncio.sleep(0) + + def _is_subagent_terminal(result: Any) -> bool: """Return whether a background subagent result is safe to clean up.""" return result.status in {SubagentStatus.COMPLETED, SubagentStatus.FAILED, SubagentStatus.CANCELLED, SubagentStatus.TIMED_OUT} or getattr(result, "completed_at", None) is not None @@ -815,6 +873,17 @@ async def task_tool( # system-channel authority over framework instructions. "acceptance_criteria": acceptance_criteria, } + loop_detection_recorder = None + parent_journal = parent_context.get("__run_journal") + if parent_journal is not None: + # The task tool runs on the parent run's loop. Pass only a proxy across + # the isolated-subagent boundary so middleware persistence is delivered + # on the loop that owns the RunJournal and its event store. + loop_detection_recorder = _ParentLoopMiddlewareRecorderProxy( + parent_journal, + asyncio.get_running_loop(), + ) + executor_kwargs["loop_detection_recorder"] = loop_detection_recorder if resolved_app_config is not None: executor_kwargs["app_config"] = resolved_app_config if run_extensions is not None: @@ -1105,3 +1174,6 @@ async def task_tool( # must end as an interrupted run, not a failed tool call. raise asyncio.CancelledError raise + finally: + if loop_detection_recorder is not None: + await loop_detection_recorder.aclose() diff --git a/backend/tests/test_loop_detection_middleware.py b/backend/tests/test_loop_detection_middleware.py index 180327973..0c75bc885 100644 --- a/backend/tests/test_loop_detection_middleware.py +++ b/backend/tests/test_loop_detection_middleware.py @@ -652,6 +652,217 @@ class TestLoopDetection: assert "default" in mw._history +class TestLoopDetectionRunEvents: + @staticmethod + def _runtime_with_journal(journal): + runtime = _make_runtime() + runtime.context["__run_journal"] = journal + return runtime + + def test_identical_call_warning_records_once_without_arguments(self): + journal = MagicMock() + runtime = self._runtime_with_journal(journal) + mw = LoopDetectionMiddleware( + warn_threshold=2, + hard_limit=10, + tool_freq_warn=100, + tool_freq_hard_limit=200, + ) + call = [_bash_call("SUPER_SECRET_COMMAND")] + + # The second identical call reaches the warning threshold. + for _ in range(2): + assert mw._apply(_make_state(tool_calls=call), runtime) is None + + # A subsequent occurrence must not duplicate the warning event. + assert mw._apply(_make_state(tool_calls=call), runtime) is None + + journal.record_middleware.assert_called_once() + recorded = journal.record_middleware.call_args + + assert recorded.kwargs["tag"] == "loop_detection" + assert recorded.kwargs["name"] == "LoopDetectionMiddleware" + assert recorded.kwargs["hook"] == "after_model" + assert recorded.kwargs["action"] == "warn" + assert recorded.kwargs["changes"] == { + "is_subagent": False, + "agent_id": None, + "detection_layer": "identical_call_set", + "tool_names": ["bash"], + "count": 2, + "threshold": 2, + } + + # Tool arguments and argument-derived values must not be persisted. + assert "SUPER_SECRET_COMMAND" not in repr(recorded) + assert "args" not in recorded.kwargs["changes"] + + def test_narrow_subagent_recorder_key_records_without_shared_journal(self): + recorder = MagicMock() + runtime = _make_runtime() + runtime.context["__run_loop_detection_recorder"] = recorder + runtime.context["is_subagent"] = True + runtime.context["agent_id"] = "general-purpose" + assert "__run_journal" not in runtime.context + mw = LoopDetectionMiddleware( + warn_threshold=2, + hard_limit=10, + tool_freq_warn=100, + tool_freq_hard_limit=200, + ) + call = [_bash_call("ls")] + + assert mw._apply(_make_state(tool_calls=call), runtime) is None + assert mw._apply(_make_state(tool_calls=call), runtime) is None + + recorder.record_middleware.assert_called_once() + assert recorder.record_middleware.call_args.kwargs["action"] == "warn" + assert recorder.record_middleware.call_args.kwargs["changes"]["is_subagent"] is True + assert recorder.record_middleware.call_args.kwargs["changes"]["agent_id"] == "general-purpose" + + def test_identical_call_hard_stop_records_event(self): + journal = MagicMock() + runtime = self._runtime_with_journal(journal) + mw = LoopDetectionMiddleware( + warn_threshold=2, + hard_limit=3, + tool_freq_warn=100, + tool_freq_hard_limit=200, + ) + call = [_bash_call("ls")] + + for _ in range(2): + assert mw._apply(_make_state(tool_calls=call), runtime) is None + + result = mw._apply(_make_state(tool_calls=call), runtime) + + assert result is not None + assert result["messages"][0].tool_calls == [] + + # One warning transition followed by one hard-stop transition. + assert journal.record_middleware.call_count == 2 + recorded = journal.record_middleware.call_args_list[-1] + + assert recorded.kwargs["tag"] == "loop_detection" + assert recorded.kwargs["action"] == "hard_stop" + assert recorded.kwargs["changes"] == { + "is_subagent": False, + "agent_id": None, + "detection_layer": "identical_call_set", + "tool_names": ["bash"], + "count": 3, + "threshold": 3, + } + + def test_tool_frequency_warning_records_once(self): + journal = MagicMock() + runtime = self._runtime_with_journal(journal) + mw = LoopDetectionMiddleware( + warn_threshold=100, + hard_limit=200, + tool_freq_warn=2, + tool_freq_hard_limit=10, + ) + + # Vary the arguments so the identical-call detector cannot fire. + for index in range(3): + result = mw._apply( + _make_state(tool_calls=[_bash_call(f"command-{index}")]), + runtime, + ) + assert result is None + + journal.record_middleware.assert_called_once() + recorded = journal.record_middleware.call_args + + assert recorded.kwargs["action"] == "warn" + assert recorded.kwargs["changes"] == { + "is_subagent": False, + "agent_id": None, + "detection_layer": "tool_frequency", + "tool_names": ["bash"], + "count": 2, + "threshold": 2, + } + + def test_tool_frequency_hard_stop_records_event(self): + journal = MagicMock() + runtime = self._runtime_with_journal(journal) + mw = LoopDetectionMiddleware( + warn_threshold=100, + hard_limit=200, + tool_freq_warn=2, + tool_freq_hard_limit=3, + ) + + for index in range(2): + assert ( + mw._apply( + _make_state(tool_calls=[_bash_call(f"command-{index}")]), + runtime, + ) + is None + ) + + result = mw._apply( + _make_state(tool_calls=[_bash_call("command-2")]), + runtime, + ) + + assert result is not None + assert journal.record_middleware.call_count == 2 + + recorded = journal.record_middleware.call_args_list[-1] + assert recorded.kwargs["action"] == "hard_stop" + assert recorded.kwargs["changes"] == { + "is_subagent": False, + "agent_id": None, + "detection_layer": "tool_frequency", + "tool_names": ["bash"], + "count": 3, + "threshold": 3, + } + + def test_journal_failure_warns_without_breaking_detection(self, caplog): + journal = MagicMock() + journal.record_middleware.side_effect = RuntimeError("db down") + runtime = self._runtime_with_journal(journal) + mw = LoopDetectionMiddleware( + warn_threshold=2, + hard_limit=10, + tool_freq_warn=100, + tool_freq_hard_limit=200, + ) + call = [_bash_call("ls")] + + assert mw._apply(_make_state(tool_calls=call), runtime) is None + + with caplog.at_level("WARNING"): + result = mw._apply(_make_state(tool_calls=call), runtime) + + assert result is None + assert mw._pending_warnings[_pending_key()] + assert "Failed to record middleware:loop_detection event" in caplog.text + + @pytest.mark.asyncio + async def test_async_after_model_records_warning_event(self): + journal = MagicMock() + runtime = self._runtime_with_journal(journal) + mw = LoopDetectionMiddleware( + warn_threshold=2, + hard_limit=10, + tool_freq_warn=100, + tool_freq_hard_limit=200, + ) + call = [_bash_call("ls")] + + assert await mw.aafter_model(_make_state(tool_calls=call), runtime) is None + assert await mw.aafter_model(_make_state(tool_calls=call), runtime) is None + + journal.record_middleware.assert_called_once() + assert journal.record_middleware.call_args.kwargs["action"] == "warn" + + class TestLoopDetectionAgentGraphIntegration: def test_loop_warning_is_transient_in_real_agent_graph(self): """after_model queues the warning; wrap_model_call injects it request-only.""" diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index f787edc71..fdd4bf095 100644 --- a/backend/tests/test_subagent_executor.py +++ b/backend/tests/test_subagent_executor.py @@ -82,11 +82,13 @@ def _setup_executor_classes(): # Save original modules original_modules = {name: sys.modules.get(name) for name in _MOCKED_MODULE_NAMES} original_executor = sys.modules.get("deerflow.subagents.executor") + original_audit_context = sys.modules.get("deerflow.agents.middlewares.audit_context") original_tool_search = sys.modules.get("deerflow.tools.builtins.tool_search") - # Preload the real deferred-tool helpers before replacing the parent agent - # packages with cycle-breaking test doubles. Executor imports this module - # lazily while building initial state. + # Preload real executor dependencies before replacing their parent packages + # with cycle-breaking test doubles. Keeping the concrete leaf modules in + # sys.modules makes this fixture independent of test collection order. + audit_context_module = importlib.import_module("deerflow.agents.middlewares.audit_context") tool_search_module = importlib.import_module("deerflow.tools.builtins.tool_search") # Remove mocked executor if exists (from conftest.py) @@ -101,6 +103,7 @@ def _setup_executor_classes(): storage_module.get_or_new_skill_storage = lambda **kwargs: SimpleNamespace(load_skills=lambda *, enabled_only: []) storage_module.get_or_new_user_skill_storage = lambda user_id, **kwargs: SimpleNamespace(load_skills=lambda *, enabled_only: []) sys.modules["deerflow.skills.storage"] = storage_module + sys.modules["deerflow.agents.middlewares.audit_context"] = audit_context_module sys.modules["deerflow.tools.builtins.tool_search"] = tool_search_module # Import real classes inside fixture @@ -146,6 +149,10 @@ def _setup_executor_classes(): sys.modules["deerflow.subagents.executor"] = original_executor elif "deerflow.subagents.executor" in sys.modules: del sys.modules["deerflow.subagents.executor"] + if original_audit_context is not None: + sys.modules["deerflow.agents.middlewares.audit_context"] = original_audit_context + else: + sys.modules.pop("deerflow.agents.middlewares.audit_context", None) if original_tool_search is not None: sys.modules["deerflow.tools.builtins.tool_search"] = original_tool_search else: @@ -3641,6 +3648,7 @@ class TestSubagentGuardrailAttribution: oauth_provider=None, oauth_id=None, run_id=None, + loop_detection_recorder=None, name="general-purpose", parent_model="test-model", ): @@ -3664,6 +3672,7 @@ class TestSubagentGuardrailAttribution: oauth_provider=oauth_provider, oauth_id=oauth_id, run_id=run_id, + loop_detection_recorder=loop_detection_recorder, ) @pytest.mark.anyio @@ -3700,6 +3709,32 @@ class TestSubagentGuardrailAttribution: assert context.get("run_id") == "run-42" assert context.get("is_subagent") is True + @pytest.mark.anyio + async def test_aexecute_propagates_narrow_loop_detection_recorder( + self, + classes, + executor_module, + monkeypatch, + ): + """The child context receives the loop-safe proxy, never the raw journal.""" + recorder = object() + executor = self._make_executor( + classes, + run_id="run-42", + loop_detection_recorder=recorder, + ) + fake_agent = _FakeStreamAgent() + monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state) + monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent) + + await executor._aexecute("do something") + + context = fake_agent.captured_context + assert context is not None + assert context.get("__run_loop_detection_recorder") is recorder + assert "__run_journal" not in context + assert context.get("agent_id") == "general-purpose" + @pytest.mark.anyio async def test_aexecute_propagates_channel_user_id_to_subagent_context( self, diff --git a/backend/tests/test_task_tool_core_logic.py b/backend/tests/test_task_tool_core_logic.py index a48c9a4a8..076abb7b9 100644 --- a/backend/tests/test_task_tool_core_logic.py +++ b/backend/tests/test_task_tool_core_logic.py @@ -36,6 +36,82 @@ from deerflow.subagents.status_contract import ( task_tool_module = importlib.import_module("deerflow.tools.builtins.task_tool") +def test_parent_loop_middleware_recorder_proxy_delivers_on_owner_loop(): + """Subagent middleware events must never call RunJournal from the child loop.""" + calls: list[tuple[object, dict]] = [] + delivered = threading.Event() + + class LoopPinnedJournal: + def record_middleware(self, **kwargs): + calls.append((asyncio.get_running_loop(), dict(kwargs))) + delivered.set() + + parent_loop = asyncio.new_event_loop() + parent_thread = threading.Thread( + target=parent_loop.run_forever, + name="test-middleware-journal-parent-loop", + daemon=True, + ) + parent_thread.start() + try: + proxy = task_tool_module._ParentLoopMiddlewareRecorderProxy( + LoopPinnedJournal(), + parent_loop, + ) + proxy.record_middleware( + tag="loop_detection", + name="LoopDetectionMiddleware", + hook="after_model", + action="warn", + changes={"detection_layer": "identical_call_set"}, + ) + + assert delivered.wait(timeout=5) + assert len(calls) == 1 + observed_loop, kwargs = calls[0] + assert observed_loop is parent_loop + assert kwargs["tag"] == "loop_detection" + assert kwargs["action"] == "warn" + + asyncio.run_coroutine_threadsafe(proxy.aclose(), parent_loop).result(timeout=5) + proxy.record_middleware(tag="loop_detection", name="LoopDetectionMiddleware", hook="after_model", action="hard_stop", changes={}) + time.sleep(0.05) + assert len(calls) == 1, "events emitted after the parent task boundary must be dropped" + finally: + parent_loop.call_soon_threadsafe(parent_loop.stop) + parent_thread.join(timeout=5) + parent_loop.close() + + +def test_parent_loop_middleware_recorder_proxy_drops_after_loop_closed(): + """A child event after asyncio.run teardown is a quiet no-op.""" + loop = asyncio.new_event_loop() + loop.close() + proxy = task_tool_module._ParentLoopMiddlewareRecorderProxy(MagicMock(), loop) + + proxy.record_middleware( + tag="loop_detection", + name="LoopDetectionMiddleware", + hook="after_model", + action="warn", + changes={}, + ) + + +def test_parent_loop_middleware_recorder_close_is_fail_open_off_owner_loop(caplog): + """A bad close caller must not replace the task tool's original outcome.""" + owner_loop = asyncio.new_event_loop() + proxy = task_tool_module._ParentLoopMiddlewareRecorderProxy(MagicMock(), owner_loop) + try: + with caplog.at_level("WARNING"): + asyncio.run(proxy.aclose()) + + assert proxy.is_closed is False + assert "Cannot drain subagent middleware recorder from a non-owner loop" in caplog.text + finally: + owner_loop.close() + + class FakeSubagentStatus(Enum): # Match production enum values so branch comparisons behave identically. PENDING = "pending" @@ -353,6 +429,40 @@ def test_task_tool_forwards_the_run_extension_snapshot_to_executor(monkeypatch): assert captured["executor_kwargs"]["extensions"] is loaded +def test_task_tool_installs_and_closes_narrow_loop_detection_recorder(monkeypatch): + journal = MagicMock() + runtime = _make_runtime() + runtime.context["__run_journal"] = journal + captured = {} + + class DummyExecutor: + def __init__(self, **kwargs): + captured["executor_kwargs"] = kwargs + + def execute_async(self, prompt, task_id=None): + return task_id or "generated-task-id" + + monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus) + monkeypatch.setattr(task_tool_module, "SubagentExecutor", DummyExecutor) + monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: _make_subagent_config()) + monkeypatch.setattr( + task_tool_module, + "get_background_task_result", + lambda _: _make_result(FakeSubagentStatus.COMPLETED, result="done"), + ) + monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: lambda _event: None) + monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep) + monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: []) + + _run_task_tool(runtime=runtime, description="test", prompt="p", subagent_type="general-purpose", tool_call_id="tc-journal") + + kwargs = captured["executor_kwargs"] + proxy = kwargs["loop_detection_recorder"] + assert proxy.is_closed is True + proxy.record_middleware(tag="loop_detection", name="LoopDetectionMiddleware", hook="after_model", action="warn", changes={}) + journal.record_middleware.assert_not_called() + + def test_task_tool_omits_extensions_without_a_run_snapshot(monkeypatch): """Callers outside the Gateway run path (embedded client, standalone LangGraph Server) install no snapshot; the executor must keep its existing diff --git a/contracts/run_event_stream_contract.json b/contracts/run_event_stream_contract.json index 0b2826235..14d0b1109 100644 --- a/contracts/run_event_stream_contract.json +++ b/contracts/run_event_stream_contract.json @@ -397,7 +397,7 @@ "pattern": "middleware:{tag}", "category": "middleware", "producer": "RunJournal.record_middleware(tag, ...)", - "known_tags": ["guardrail", "safety_termination", "skill_activation", "skill_secrets"], + "known_tags": ["guardrail", "loop_detection", "safety_termination", "skill_activation", "skill_secrets"], "event_type_schema": { "type": "string", "pattern": "^middleware:", @@ -444,7 +444,7 @@ { "id": "middleware-coverage", "status": "partial", - "notes": "Loop detection and deferred-tool promotion do not currently emit middleware events." + "notes": "Loop-detection events cover lead-agent and ordinary task-tool subagent runs; durable batch subagent loop detection and deferred-tool promotion do not currently emit middleware events." }, { "id": "run-scoped-observation-context",