feat(observability): persist loop detection events (#5127)

* feat(observability): persist loop detection events

* fix(observability): persist subagent loop events

* fix(observability): narrow subagent loop event bridge

* fix(observability): attribute subagent loop events

* fix(tests): isolate subagent executor imports

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.

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.

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"

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

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,

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.

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}")
 

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()

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."""

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,

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

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",