From 14c9d44440780e63563e935046db8708e121a5b1 Mon Sep 17 00:00:00 2001
From: PeaceMaker-best <paecemakerboy@gmail.com>
Date: Tue, 15 Sep 2026 08:45:08 +0800
Subject: [PATCH] feat(runtime): persist tool-progress phase transitions
 (#5214)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* feat(runtime): persist tool-progress phase transitions

Record bounded warn, block, and recover decisions for lead and task subagent runs while preserving event-loop isolation, fail-open behavior, and concurrent transition order.

* fix(runtime): trust server-owned tool progress attribution

* fix(runtime): centralize trusted audit attribution

* fix(runtime): preserve complete tool progress audit state

* docs: trim tool progress guidance to pass size check

* fix(runtime): fence subagent audit recorder loop

* docs(readme): sync tool-progress event coverage

Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com>

---------

Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com>
Co-authored-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com>
Co-authored-by: 嗜鵼 <hy2010hy2010@qq.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
---
 README.md                                     |   6 +-
 README_zh.md                                  |   2 +-
 backend/app/gateway/services.py               |   7 +-
 backend/docs/RUN_EVENT_STREAM.md              |  37 +-
 .../deerflow/agents/middlewares/AGENTS.md     |   4 +-
 .../agents/middlewares/audit_context.py       |  27 +-
 .../middlewares/loop_detection_middleware.py  |  18 +-
 .../middlewares/tool_progress_middleware.py   | 202 ++++++++--
 .../tool_promotion_audit_middleware.py        |  20 +-
 .../agents/middlewares/tool_result_meta.py    |  13 +-
 .../harness/deerflow/runtime/AGENTS.md        |  18 +
 .../deerflow/runtime/events/catalog.py        |   2 +
 .../harness/deerflow/runtime/journal.py       |  96 +++--
 .../harness/deerflow/runtime/runs/worker.py   |   5 +
 .../harness/deerflow/subagents/AGENTS.md      |   2 +-
 .../harness/deerflow/subagents/executor.py    |  12 +-
 .../packages/harness/deerflow/tools/AGENTS.md |   2 +-
 .../deerflow/tools/builtins/task_tool.py      |   4 +
 backend/tests/test_gateway_services.py        |  22 ++
 .../tests/test_loop_detection_middleware.py   |  21 +-
 .../tests/test_run_event_stream_contract.py   |  10 +
 backend/tests/test_run_journal.py             | 132 +++++++
 backend/tests/test_run_worker_rollback.py     |  14 +
 backend/tests/test_subagent_executor.py       |  34 ++
 backend/tests/test_task_tool_core_logic.py    |  13 +
 .../tests/test_tool_progress_middleware.py    | 365 +++++++++++++++++-
 .../test_tool_promotion_audit_middleware.py   |  33 ++
 backend/tests/test_tool_result_meta.py        |   2 +
 contracts/run_event_stream_contract.json      |   4 +-
 29 files changed, 1024 insertions(+), 103 deletions(-)

diff --git a/README.md b/README.md
index 3c9855df4..a6d364ef0 100644
--- a/README.md
+++ b/README.md
@@ -814,8 +814,10 @@ finalization) keep the existing completion-data behavior: they receive the
 zero-delivery receipt but do not overwrite RunStore completion fields with an
 empty snapshot.
 
-The same run event history records loop-detection decisions and deferred MCP
-tool promotions for both the lead agent and ordinary task subagents. Promotion
+When `tool_progress.enabled` is true, the same run event history also records
+result-quality guard phase changes. It records loop-detection decisions and
+deferred MCP tool promotions for both the lead agent and ordinary task
+subagents. Promotion
 events identify newly promoted deferred-tool names and whether routing metadata or
 `tool_search` selected them, without copying the search query, routing keywords,
 schemas, arguments, results, or catalog hash into the promotion event itself.
diff --git a/README_zh.md b/README_zh.md
index 0e6da345c..eef0d5b9f 100644
--- a/README_zh.md
+++ b/README_zh.md
@@ -562,7 +562,7 @@ logging:
 
 Gateway 的运行历史还会为每次运行记录一条终止时的 `run.delivery` 回执，包括零产出与崩溃恢复的运行。正常执行时，该回执会在持久化终止运行状态之前写入。孤儿恢复会先原子地认领过期租约，再幂等地回填回执，因此过期的恢复扫描不会覆盖仍在运行的详细交付事实。在事件存储中断期间，回执持久化保持尽力而为。对 checkpoint 预检失败（或在等待前序 finalization 时被取消）的运行，保持既有的完成数据行为：它们会收到零交付回执，但不会用空快照覆盖 RunStore 的完成字段。
 
-同一份运行事件历史还会为 lead agent 与普通 task subagent 记录 loop-detection 判定和延迟 MCP 工具晋升。晋升事件会标识新晋升的延迟工具名称，以及是路由元数据还是 `tool_search` 选中了它们，但不会把搜索查询、路由关键词、schema、参数、结果或目录哈希复制进晋升事件本身。
+当 `tool_progress.enabled` 为 true 时，同一份运行事件历史还会记录结果质量防护器的阶段变化。它也会为 lead agent 与普通 task subagent 记录 loop-detection 判定和延迟 MCP 工具晋升。晋升事件会标识新晋升的延迟工具名称，以及是路由元数据还是 `tool_search` 选中了它们，但不会把搜索查询、路由关键词、schema、参数、结果或目录哈希复制进晋升事件本身。
 
 #### LangSmith 链路追踪
 
diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py
index 1defc5bee..ab4b1fe09 100644
--- a/backend/app/gateway/services.py
+++ b/backend/app/gateway/services.py
@@ -533,6 +533,11 @@ _SERVER_OWNED_RUNTIME_CONTEXT_KEYS: frozenset[str] = (
             "is_internal",
             "authz_attributes",
             "channel_user_id",
+            "is_subagent",
+            "agent_id",
+            "__run_loop_detection_recorder",
+            "__run_tool_promotion_recorder",
+            "__run_tool_progress_recorder",
             "langgraph_auth_user",
             "langgraph_auth_user_id",
         }
@@ -575,7 +580,7 @@ _CONTEXT_RUNTIME_ONLY_KEYS: frozenset[str] = frozenset({"github_token", "disable
 # credential that ``bash`` exports as ``GH_TOKEN``/``GITHUB_TOKEN``, and a copy
 # smuggled through ``body.config['configurable']`` would be written to the
 # checkpoint store.
-_INTERNAL_ONLY_CONTEXT_KEYS: frozenset[str] = _CONTEXT_INTERNAL_CALLER_KEYS | _CONTEXT_RUNTIME_ONLY_KEYS
+_INTERNAL_ONLY_CONTEXT_KEYS: frozenset[str] = _CONTEXT_INTERNAL_CALLER_KEYS | _CONTEXT_RUNTIME_ONLY_KEYS | _SERVER_OWNED_RUNTIME_CONTEXT_KEYS
 
 
 def strip_internal_context_keys(config: dict[str, Any]) -> None:
diff --git a/backend/docs/RUN_EVENT_STREAM.md b/backend/docs/RUN_EVENT_STREAM.md
index 31adf4803..adf766271 100644
--- a/backend/docs/RUN_EVENT_STREAM.md
+++ b/backend/docs/RUN_EVENT_STREAM.md
@@ -77,9 +77,9 @@ through run-event or specialized APIs:
 | `middleware:{tag}` | `middleware` | `record_middleware()` |
 
 Current middleware tags are `guardrail`, `loop_detection`,
-`safety_termination`, `skill_activation`, `skill_secrets`, and
-`tool_promotion`. The pattern is intentionally open so new middleware tags are
-additive. Because the full event type is limited to 32 characters and
+`safety_termination`, `skill_activation`, `skill_secrets`, `tool_promotion`,
+and `tool_progress`. 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
@@ -95,6 +95,25 @@ subagent, and its agent id when applicable. Tool arguments, prompts, message
 content, tool results, and argument-derived hashes are not persisted in this
 event.
 
+`middleware:tool_progress` records effective result-quality state transitions
+from `ToolProgressMiddleware`: `warn` for ACTIVE → WARNED, `block` for a
+transition to BLOCKED, `recover` for WARNED → ACTIVE after a useful result,
+and `reset` when a later agent invocation (for example, goal continuation)
+deliberately clears a prior WARNED/BLOCKED phase.
+The bounded `changes` object contains the tool name, source and destination
+phases, consecutive-problem count, normalized result status, error
+classification and recovery advice, effective count threshold (null when no
+count threshold fired), and
+subagent attribution. Tool
+arguments, prompts, message content, tool results, and content-derived hashes
+are never copied into this event. Producer-supplied metadata is projected onto
+fixed error/action vocabularies and a strict boolean-or-null before persistence,
+so a custom tool cannot smuggle arbitrary values through its result stamp.
+Recorder calls happen after the state lock is released, so a slow custom
+recorder cannot stall unrelated tool-state updates. Recorder failures are
+fail-open and do not change guard behavior. These events are emitted only when
+`tool_progress.enabled` is true (the default is false).
+
 `middleware:tool_promotion` records deferred MCP schemas newly promoted for
 the active catalog. `changes.source` distinguishes automatic `routing_hint`
 promotion from an explicit `tool_search`; the remaining fields contain sorted
@@ -107,9 +126,10 @@ no names emit no event. Queries, routing keywords, catalog hashes, tool schemas
 and descriptions, arguments, and results are not copied into this middleware
 event. Other event types retain their existing payload contracts.
 
-Ordinary task-tool subagents forward both loop-detection and tool-promotion
-appends to the parent run loop through dedicated recorder context keys. The
-loop-bound `RunJournal` itself never enters the isolated subagent loop.
+Ordinary task-tool subagents forward loop-detection, tool-promotion, and
+tool-progress appends to the parent run loop through dedicated recorder context
+keys. The loop-bound `RunJournal` itself never enters the isolated subagent
+loop.
 
 ### Opaque Run Outputs
 
@@ -206,7 +226,8 @@ 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.
-- Durable batch subagent loop detection and deferred-tool promotion do not
-  emit middleware events because those runs have no parent run journal.
+- Durable batch subagent loop detection, deferred-tool promotion, and
+  tool-progress transitions do not emit middleware events because those runs
+  have no parent run journal.
 - 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 1b7dabeb0..b847d5dca 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md
+++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md
@@ -73,7 +73,7 @@ ordinary compaction or a model call.
    contract module.
 10. **SandboxAuditMiddleware** - Audits sandboxed shell/file operations before tool execution; command classification is **defense-in-depth and audit, not a security boundary** (the sandbox is the isolation boundary). Command substitution is judged by *position*, not the presence of `$(`: **command position** (`$(curl url)`, `` `curl url` ``, the word after `|`/`&&`/`;`, an `eval`/`source` argument) executes fetched content and is blocked; **value position** (`x=$(curl url)`, `echo $(curl url)`, an argument, a `for` word list) only captures output and passes (#4611). So `_HIGH_RISK_COMMAND_POSITION_PATTERNS` is matched anchored against each sub-command from `_split_compound_command(split_pipes=True)`, never the whole string; pipe-spanning rules (`| sh`, `base64 -d | ...`) still use `_classify_command`'s whole-command Pass 1. `_COMMAND_POSITION_PREFIX` extends the anchor over leading assignments and exec wrappers (`FOO=1 $(curl url)`, `env`/`command`/`builtin`/`exec`/`nohup`/`time`/`sudo`/`doas`); its assignment branch requires whitespace before the substitution, which keeps `x=$(curl url)` in value position. Two contexts are deliberately **position-blind** (matched whole-command in Pass 1, since they execute their input anywhere, e.g. `xargs sh -c "$(curl url)"`): an `eval`/`source` argument, and an interpreter **code-string flag** — `-c` (shells, `python`), `-e` (`perl`/`ruby`/`node`), `-p` (`perl`/`node`), `-r` (`php`) — plus the here-string (`<<<`) reaching the same place via stdin. All three substitution spellings (`$(`, `<(`, `` ` ``) share one `_RISKY_SUBSTITUTION` opener. An unquoted newline splits like `;` (else `echo hi\n$(curl url)` evades the anchored rules). Heredoc bodies are data: `_split_compound_command` records headers (`<<EOF`, `<<-EOF`, `<<'EOF'`) and consumes their bodies verbatim, so a body line starting `$(curl url)` isn't promoted to a command position; `<<<` (here-string, needs look-ahead + look-behind) and a `<<` inside `$(( ))`/`(( ))` (bit shift, arithmetic depth tracked with the quote flags) must not open one. This is a heuristic, not shell parsing — an unterminated body consumes the rest of the string, an unclosed `((` only disables heredoc detection, and the failure direction is always toward *more* command positions, not fewer. Known gaps: process substitution outside `eval`/`source` (`. <(curl u)`) is undetected, and two-step forms (`x=$(curl u); eval "$x"`) need dataflow analysis. No config gate — appended unconditionally in `_build_runtime_middlewares`, for both lead and subagents.
 11. **ReadBeforeWriteMiddleware** - *(optional, `read_before_write.enabled`, default on)* Outermost write gate (#3857): `read_file` stamps a content hash on its ToolMessage; `write_file` (existing file, incl. append) and `str_replace` are blocked unless the newest mark for the path matches its current hash. Sits outside ToolProgress/ToolErrorHandling (a block consumes no ToolProgress slot); blocked results self-stamp `deerflow_tool_meta` and carry `deerflow_write_block` (`{path, tool}`). Marks live on messages, so summarization dropping the read invalidates the gate; writes never refresh marks. Gate check + execution are serialized per (thread, path); `"Error: ..."`-string sandboxes (AIO/E2B) fail open. It owns the composed call's sandbox authorization scope; `SandboxAuthorizationError` becomes an error ToolMessage. Its `wrap_model_call` swaps blocked calls' dead payload (`content`, `old_str`/`new_str`) for a deterministic placeholder in the model-bound request only (`elide_blocked_payloads`, `elide_min_chars`); state, receipts, and the journal keep the originals. Policy stays in the gate; the shared `tool_call_args` helper rewrites every arg surface together and every model-bound arg rewrite must use it.
-12. **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:** 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. **ToolProgressMiddleware** - *(optional, if `tool_progress.enabled`)* State-machine-based stagnation guard (RFC #3177). Outer wrapper around ToolErrorHandlingMiddleware so its tool wrapper receives results already stamped with `deerflow_tool_meta`. Recoverable problems stay WARNED with hints; retryable non-recoverable problems escalate to BLOCKED; stop-category failures block immediately. It is independent of LoopDetection's turn-level call-pattern guard. Effective phase changes emit `middleware:tool_progress`, including `reset` when a later agent invocation (for example, goal continuation) clears WARNED/BLOCKED state. The state machine supplies the action and count threshold (`null` for category rules, recovery, and reset); recorder calls happen after releasing the state lock. Events include the actual sync/async hook plus bounded status, error, and recovery fields. Server-owned recorder keys determine subagent attribution. Arguments, content, prompts, and hashes are never persisted; recorder failures are fail-open. See [event semantics](../../../../../docs/RUN_EVENT_STREAM.md).
 13. **ToolReceiptMiddleware + ToolErrorHandlingMiddleware** - `ToolReceiptMiddleware` is *(optional, if `verification.receipts_enabled`, default on)*. It is the **outermost `wrap_tool_call` layer** — registered ahead of entries 9-12 — because Guardrail/SandboxAudit/ReadBeforeWrite/ToolProgress can short-circuit a call with their own ToolMessage (and SandboxAudit rebuilds medium-risk results); an inner receipt layer would silently gap the ledger on those results (ordering constraints in `deerflow.extensions.ordering`). Normal results still carry the `deerflow_tool_meta` status ToolErrorHandlingMiddleware stamps on the inner return path; short-circuit messages self-stamp meta or fall back to `message.status`. It stamps deterministic provenance (tool name, status, args/output hashes, byte count, timestamp) onto direct `ToolMessage` results and every matching `ToolMessage` carried in `Command.update.messages`, including delegated `task`, `present_file`, `view_image`, and `tool_search` results; before model calls it derives a hidden receipt ledger (display ids r1..rN) from message state, and when the 2,000-character budget is exceeded the newest receipts are retained in chronological order with their original ids plus an older-receipts omission marker. Rendering returns both the text and its retained receipt subset; every response that received a ledger carries only that exact server-owned subset, never omitted receipts. Snapshot validation accepts a strictly consecutive positive original-id range (for example `r24`–`r30`) rather than requiring `r1`, so subagent terminal citation verification resolves ids against evidence present in the citing turn even when later summarization drops and renumbers tool messages. Model-generated citation IDs are digit-bounded before integer conversion; oversized IDs are ignored as malformed input rather than raising through task write-back. Citation parsing deduplicates exact `(id, anchor)` pairs, not IDs alone, so repeated identical references stay compact while every distinct anchor claim is verified. Gateway strips delegated receipts/verdicts from external messages. `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.
 
 Authorization identity is independent of enforcement. Gateway strips client identity overrides: only the server auth source sets `is_internal`, and only authenticated IM `body.context` supplies `channel_user_id` (never `body.config`). `build_principal_from_context` applies role defaults, strict provenance, and copied attributes; RBAC rejects unknown defaults. Delegation and `GuardrailMiddleware` share this identity. Layer 1 precedes deferred assembly across agent paths and its provider is reused for Layer 2; framework skill/memory ordering stays stable. Trusted `DeerFlowClient.stream()` accepts identity overrides. Its graph key always includes effective storage `user_id` and, when enforced, the full Principal; nested attributes are copied so mutation cannot hide stale cache state.
@@ -99,7 +99,7 @@ Before changing a later authorization phase, read the [authorization RFC](../../
 22. **MemoryMiddleware** - Queues conversations for async memory update (filters to user + final AI responses); captures the runtime-resolved user so standalone LangGraph Server reads and writes stay in the same bucket
 23. **ViewImageMiddleware** - *(optional, if the model supports vision)* Appends a hidden HumanMessage with base64 image data, identified by a reserved ID prefix plus a server-owned metadata marker, to `ModelRequest.messages` in `wrap_model_call` / `awrap_model_call`. The payload lives only in that request and is never returned as a state update, so no checkpoint carries it and an interrupted run cannot strand it in history; state keeps only the lightweight `viewed_images` metadata. It owns that context and rebuilds it per call: its own message is swept out of the request first — a thread checkpointed by the earlier `before_model`/`after_model` pair (which wrote the payload into state and took it back out with `RemoveMessage`) can carry one that reached state but was never removed, and leaving it in would resend that base64 in every later request for the life of the thread — then a freshly built one is appended when warranted. The sweep requires both the reserved ID prefix and the server-owned marker, so a client cannot get its own message dropped; unmarked leftovers predating the marker are left in place and merely not duplicated
 24. **McpRoutingMiddleware** - Auto-promotes deferred schemas matching the latest real user message before `DeferredToolFilterMiddleware`, without executing tools. New names emit `middleware:tool_promotion` with `source=routing_hint`; repeat passes emit nothing
-25. **DeferredToolPromotionAuditMiddleware** - Observes final `tool_search` `Command`s; keep it outer of `SkillToolPolicyMiddleware` so denied names are excluded. It atomically claims new names per lead run or subagent execution to dedupe parallel searches, returns the original `Command`, and omits private payloads
+25. **DeferredToolPromotionAuditMiddleware** - Observes final `tool_search` `Command`s; keep it outer of `SkillToolPolicyMiddleware` so denied names are excluded. It atomically claims new names per lead run or subagent execution to dedupe parallel searches, derives subagent attribution from the server-installed recorder, returns the original `Command`, and omits private payloads
 26. **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)
 27. **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
 28. **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.
diff --git a/backend/packages/harness/deerflow/agents/middlewares/audit_context.py b/backend/packages/harness/deerflow/agents/middlewares/audit_context.py
index 1e8bbda6c..16ceb0f02 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/audit_context.py
+++ b/backend/packages/harness/deerflow/agents/middlewares/audit_context.py
@@ -1,4 +1,29 @@
-"""Private runtime-context keys for narrowly scoped middleware audit recorders."""
+"""Private runtime-context helpers for narrowly scoped audit recorders."""
+
+from typing import Any
 
 LOOP_DETECTION_RECORDER_CONTEXT_KEY = "__run_loop_detection_recorder"
 TOOL_PROMOTION_RECORDER_CONTEXT_KEY = "__run_tool_promotion_recorder"
+TOOL_PROGRESS_RECORDER_CONTEXT_KEY = "__run_tool_progress_recorder"
+
+
+def resolve_audit_recorder(
+    context: object,
+    *,
+    recorder_key: str,
+) -> tuple[Any | None, bool, str | None]:
+    """Resolve a recorder and trusted lead/subagent attribution.
+
+    Ordinary lead runs own ``__run_journal``. Task-tool subagents receive only
+    a server-installed narrow recorder, so its presence is the authority for
+    subagent attribution; caller-supplied ``is_subagent`` is never consulted.
+    """
+    if not isinstance(context, dict):
+        return None, False, None
+
+    recorder = context.get(recorder_key)
+    if recorder is not None:
+        agent_id = context.get("agent_id")
+        return recorder, True, agent_id if isinstance(agent_id, str) else None
+
+    return context.get("__run_journal"), False, None
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 a0af866ac..f539a2b26 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py
+++ b/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py
@@ -79,7 +79,10 @@ 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.agents.middlewares.audit_context import (
+    LOOP_DETECTION_RECORDER_CONTEXT_KEY,
+    resolve_audit_recorder,
+)
 from deerflow.runtime.events.catalog import MIDDLEWARE_LOOP_DETECTION_TAG
 
 if TYPE_CHECKING:
@@ -731,13 +734,10 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
         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")
+        recorder, is_subagent, agent_id = resolve_audit_recorder(
+            getattr(runtime, "context", None),
+            recorder_key=LOOP_DETECTION_RECORDER_CONTEXT_KEY,
+        )
         if recorder is None:
             return
 
@@ -749,7 +749,7 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
                 action=decision.action,
                 changes={
                     "is_subagent": is_subagent,
-                    "agent_id": context.get("agent_id") if is_subagent else None,
+                    "agent_id": agent_id,
                     "detection_layer": decision.detection_layer,
                     "tool_names": list(decision.tool_names),
                     "count": decision.count,
diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py
index 92c921a9d..7a0c13e36 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py
+++ b/backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py
@@ -64,7 +64,19 @@ 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
+from deerflow.agents.middlewares.audit_context import (
+    TOOL_PROGRESS_RECORDER_CONTEXT_KEY,
+    resolve_audit_recorder,
+)
+from deerflow.agents.middlewares.tool_result_meta import (
+    PROGRESS_GUARD_ERROR_TYPE,
+    TOOL_META_KEY,
+    TOOL_RESULT_ERROR_TYPES,
+    TOOL_RESULT_NEXT_ACTIONS,
+    TOOL_RESULT_STATUSES,
+    ToolResultMeta,
+)
+from deerflow.runtime.events.catalog import MIDDLEWARE_TOOL_PROGRESS_TAG
 
 if TYPE_CHECKING:
     from deerflow.config.tool_progress_config import ToolProgressConfig
@@ -76,6 +88,28 @@ _MAX_PENDING_PER_RUN = 3
 _MAX_CONTENT_FOR_WORDSET = 8192
 
 
+def _audit_error_type(value: object) -> str | None:
+    """Project an untrusted tool stamp onto the bounded audit vocabulary."""
+    if value is None:
+        return None
+    return value if type(value) is str and value in TOOL_RESULT_ERROR_TYPES else "unknown"
+
+
+def _audit_next_action(value: object) -> str:
+    """Return only framework-defined recovery actions to persistence."""
+    return value if type(value) is str and value in TOOL_RESULT_NEXT_ACTIONS else "unknown"
+
+
+def _audit_status(value: object) -> str:
+    """Project the producer status onto the canonical result vocabulary."""
+    return value if type(value) is str and value in TOOL_RESULT_STATUSES else "unknown"
+
+
+def _audit_recoverable(value: object) -> bool | None:
+    """Reject truthy non-booleans from producer-supplied tool metadata."""
+    return value if type(value) is bool else None
+
+
 # ---------------------------------------------------------------------------
 # State data structures
 
@@ -93,6 +127,14 @@ class ToolPhaseState:
     recent_word_sets: tuple[frozenset[str], ...] = field(default_factory=tuple)
 
 
+@dataclass(frozen=True, slots=True)
+class ToolPhaseTransition:
+    """The exact state-machine rule that produced a durable phase change."""
+
+    action: Literal["warn", "block", "recover", "reset"]
+    threshold: int | None
+
+
 # ---------------------------------------------------------------------------
 # Content helpers
 
@@ -231,11 +273,10 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         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.
+        # threading.Lock (not asyncio.Lock) also protects embedded callers that use the
+        # synchronous wrapper from multiple threads. Recorder callbacks are deliberately
+        # invoked after this state lock is released; observability must not stall tool
+        # state updates. This matches LoopDetectionMiddleware's publication convention.
         self._lock = threading.Lock()
         # LRU-evicting store: thread_id → {tool_name → ToolPhaseState}
         self._phase_states: OrderedDict[str, dict[str, ToolPhaseState]] = OrderedDict()
@@ -270,6 +311,52 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
     def _pending_key(self, runtime: Runtime) -> tuple[str, str]:
         return self._thread_id(runtime), self._run_id(runtime)
 
+    def _record_phase_transition(
+        self,
+        *,
+        runtime: Runtime,
+        tool_name: str,
+        state: ToolPhaseState,
+        new_state: ToolPhaseState,
+        meta: ToolResultMeta | None,
+        hook: Literal["wrap_tool_call", "awrap_tool_call", "before_agent", "abefore_agent"],
+        transition: ToolPhaseTransition,
+    ) -> None:
+        """Persist one effective transition without copying tool content."""
+        recorder, is_subagent, agent_id = resolve_audit_recorder(
+            getattr(runtime, "context", None),
+            recorder_key=TOOL_PROGRESS_RECORDER_CONTEXT_KEY,
+        )
+        if recorder is None:
+            return
+
+        try:
+            recorder.record_middleware(
+                tag=MIDDLEWARE_TOOL_PROGRESS_TAG,
+                name=type(self).__name__,
+                hook=hook,
+                action=transition.action,
+                changes={
+                    "is_subagent": is_subagent,
+                    "agent_id": agent_id,
+                    "tool_name": tool_name,
+                    "from_phase": state.phase,
+                    "to_phase": new_state.phase,
+                    "consecutive_problems": new_state.consecutive_problems,
+                    "status": _audit_status(meta.status) if meta is not None else None,
+                    "error_type": _audit_error_type(meta.error_type) if meta is not None else None,
+                    "recoverable_by_model": _audit_recoverable(meta.recoverable_by_model) if meta is not None else None,
+                    "recommended_next_action": _audit_next_action(meta.recommended_next_action) if meta is not None else None,
+                    "threshold": transition.threshold,
+                },
+            )
+        except Exception:  # noqa: BLE001
+            # Observability must never alter the progress guard or agent run.
+            logger.warning(
+                "Failed to record middleware:tool_progress event",
+                exc_info=True,
+            )
+
     # ------------------------------------------------------------------
     # State store (caller holds lock)
 
@@ -308,7 +395,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
             additional_kwargs={
                 TOOL_META_KEY: {
                     "status": "error",
-                    "error_type": "blocked_by_progress_guard",
+                    "error_type": PROGRESS_GUARD_ERROR_TYPE,
                     "recoverable_by_model": True,
                     "recommended_next_action": "summarize",
                     "source": "progress_middleware",
@@ -322,6 +409,8 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         tool_name: str,
         runtime: Runtime,
         tool_call_id: str,
+        *,
+        hook: Literal["wrap_tool_call", "awrap_tool_call"],
     ) -> ToolMessage | Command:
         """Update the state machine from a tool result; queue hints if warranted."""
         message = _result_tool_message(result, tool_call_id)
@@ -337,10 +426,24 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
             return result
         content = _message_content_str(message)
         thread_id = self._thread_id(runtime)
+        phase_change: tuple[ToolPhaseState, ToolPhaseState, ToolPhaseTransition] | None = None
         with self._lock:
             state = self._get_state(thread_id, tool_name)
-            new_state, hint = self._assess_and_transition(state, meta, content)
+            new_state, hint, transition = self._assess_and_transition(state, meta, content)
             self._set_state(thread_id, tool_name, new_state)
+            if transition is not None:
+                phase_change = (state, new_state, transition)
+        if phase_change is not None:
+            old_state, changed_state, phase_transition = phase_change
+            self._record_phase_transition(
+                runtime=runtime,
+                tool_name=tool_name,
+                state=old_state,
+                new_state=changed_state,
+                meta=meta,
+                hook=hook,
+                transition=phase_transition,
+            )
         if new_state.phase != state.phase:
             if new_state.phase == "blocked":
                 logger.warning(
@@ -374,8 +477,8 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         state: ToolPhaseState,
         meta: ToolResultMeta,
         content: str,
-    ) -> tuple[ToolPhaseState, str | None]:
-        """Return (new_state, hint_text_or_None).
+    ) -> tuple[ToolPhaseState, str | None, ToolPhaseTransition | None]:
+        """Return the new state, optional hint, and rule that changed phase.
 
         The outer wrap_tool_call gate intercepts already-blocked states before
         the handler is called, so this function is normally reached only for
@@ -389,7 +492,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         # 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
+            return state, None, 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).
@@ -397,12 +500,16 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
 
         # 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
+            return (
+                replace(
+                    state,
+                    phase="blocked",
+                    consecutive_problems=new_count,
+                    block_reason=_block_reason(meta),
+                ),
+                None,
+                ToolPhaseTransition(action="block", threshold=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.
@@ -412,9 +519,11 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         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
+            transition = ToolPhaseTransition(action="recover", threshold=None) if state.phase == "warned" else None
+            return replace(state, consecutive_problems=0, phase="active", recent_word_sets=new_recent), None, transition
 
         hint: str | None = None
+        transition: ToolPhaseTransition | None = None
 
         if new_count >= self._stagnation_threshold + self._warn_escalation:
             if meta.recoverable_by_model:
@@ -422,17 +531,28 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
                 # BLOCKED would prevent a legitimate retry with different parameters.
                 hint = _format_hint(meta)
                 new_state = replace(state, consecutive_problems=new_count, phase="warned")
+                if state.phase != "warned":
+                    transition = ToolPhaseTransition(
+                        action="warn",
+                        threshold=self._stagnation_threshold + self._warn_escalation,
+                    )
             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)
+                transition = ToolPhaseTransition(
+                    action="block",
+                    threshold=self._stagnation_threshold + self._warn_escalation,
+                )
         elif new_count >= self._stagnation_threshold:
             hint = _format_hint(meta)
             new_state = replace(state, consecutive_problems=new_count, phase="warned")
+            if state.phase != "warned":
+                transition = ToolPhaseTransition(action="warn", threshold=self._stagnation_threshold)
         else:
             new_state = replace(state, consecutive_problems=new_count)
 
-        return new_state, hint
+        return new_state, hint, transition
 
     # ------------------------------------------------------------------
     # Pending queue helpers
@@ -462,7 +582,12 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
                 if key[0] == thread_id and key[1] != current_run:
                     del self._pending[key]
 
-    def _reset_run_states(self, runtime: Runtime) -> None:
+    def _reset_run_states(
+        self,
+        runtime: Runtime,
+        *,
+        hook: Literal["before_agent", "abefore_agent"],
+    ) -> 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
@@ -483,18 +608,33 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
         middlewares therefore guard different failure modes and use different lifetimes.
         """
         thread_id = self._thread_id(runtime)
+        transitions: list[tuple[str, ToolPhaseState, ToolPhaseState]] = []
         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(
+                new_state = replace(
                     tool_state,
                     phase="active",
                     consecutive_problems=0,
                     block_reason=None,
                     recent_word_sets=(),
                 )
+                thread_tools[tool_name] = new_state
+                if tool_state.phase != new_state.phase:
+                    transitions.append((tool_name, tool_state, new_state))
+
+        for tool_name, tool_state, new_state in transitions:
+            self._record_phase_transition(
+                runtime=runtime,
+                tool_name=tool_name,
+                state=tool_state,
+                new_state=new_state,
+                meta=None,
+                hook=hook,
+                transition=ToolPhaseTransition(action="reset", threshold=None),
+            )
 
     # ------------------------------------------------------------------
     # wrap_tool_call
@@ -520,7 +660,13 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
                 block_reason,
             )
             return self._make_blocked_message(request, tool_name, block_reason)
-        return self._update_state_from_result(handler(request), tool_name, runtime, str(request.tool_call.get("id") or ""))
+        return self._update_state_from_result(
+            handler(request),
+            tool_name,
+            runtime,
+            str(request.tool_call.get("id") or ""),
+            hook="wrap_tool_call",
+        )
 
     @override
     async def awrap_tool_call(
@@ -543,7 +689,13 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
                 block_reason,
             )
             return self._make_blocked_message(request, tool_name, block_reason)
-        return self._update_state_from_result(await handler(request), tool_name, runtime, str(request.tool_call.get("id") or ""))
+        return self._update_state_from_result(
+            await handler(request),
+            tool_name,
+            runtime,
+            str(request.tool_call.get("id") or ""),
+            hook="awrap_tool_call",
+        )
 
     # ------------------------------------------------------------------
     # wrap_model_call: drain pending hints and inject before model sees messages
@@ -586,11 +738,11 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
     @override
     def before_agent(self, state: AgentState, runtime: Runtime) -> dict | None:
         self._clear_stale_pending(runtime)
-        self._reset_run_states(runtime)
+        self._reset_run_states(runtime, hook="before_agent")
         return None
 
     @override
     async def abefore_agent(self, state: AgentState, runtime: Runtime) -> dict | None:
         self._clear_stale_pending(runtime)
-        self._reset_run_states(runtime)
+        self._reset_run_states(runtime, hook="abefore_agent")
         return None
diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_promotion_audit_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_promotion_audit_middleware.py
index 20b38e6cc..980c5a769 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/tool_promotion_audit_middleware.py
+++ b/backend/packages/harness/deerflow/agents/middlewares/tool_promotion_audit_middleware.py
@@ -13,7 +13,10 @@ from langgraph.prebuilt.tool_node import ToolCallRequest
 from langgraph.runtime import Runtime
 from langgraph.types import Command
 
-from deerflow.agents.middlewares.audit_context import TOOL_PROMOTION_RECORDER_CONTEXT_KEY
+from deerflow.agents.middlewares.audit_context import (
+    TOOL_PROMOTION_RECORDER_CONTEXT_KEY,
+    resolve_audit_recorder,
+)
 from deerflow.runtime.events.catalog import MIDDLEWARE_TOOL_PROMOTION_TAG
 
 logger = logging.getLogger(__name__)
@@ -34,15 +37,10 @@ def record_tool_promotion(
     if not names:
         return
 
-    context = getattr(runtime, "context", None)
-    if not isinstance(context, dict):
-        return
-    is_subagent = context.get("is_subagent") is True
-    recorder = context.get(TOOL_PROMOTION_RECORDER_CONTEXT_KEY)
-    if recorder is None:
-        # Lead runs own a RunJournal. Ordinary task-tool subagents receive only
-        # the narrow, loop-safe recorder key above.
-        recorder = context.get("__run_journal")
+    recorder, is_subagent, agent_id = resolve_audit_recorder(
+        getattr(runtime, "context", None),
+        recorder_key=TOOL_PROMOTION_RECORDER_CONTEXT_KEY,
+    )
     if recorder is None:
         return
 
@@ -62,7 +60,7 @@ def record_tool_promotion(
                 "tool_names": names,
                 "count": len(names),
                 "is_subagent": is_subagent,
-                "agent_id": context.get("agent_id") if is_subagent else None,
+                "agent_id": agent_id,
             },
         )
     except Exception:  # noqa: BLE001
diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_result_meta.py b/backend/packages/harness/deerflow/agents/middlewares/tool_result_meta.py
index 454e2456e..84c150b6a 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/tool_result_meta.py
+++ b/backend/packages/harness/deerflow/agents/middlewares/tool_result_meta.py
@@ -10,12 +10,18 @@ from __future__ import annotations
 import json
 import re
 from dataclasses import dataclass
-from typing import Literal
+from typing import Literal, get_args
 
 from langchain_core.messages import ToolMessage
 from langgraph.types import Command
 
 TOOL_META_KEY = "deerflow_tool_meta"
+PROGRESS_GUARD_ERROR_TYPE = "blocked_by_progress_guard"
+
+ToolResultStatus = Literal["success", "error", "partial_success"]
+RecommendedNextAction = Literal["continue", "rewrite_query", "try_alternative", "summarize", "stop"]
+TOOL_RESULT_STATUSES = frozenset(get_args(ToolResultStatus)) | {"unknown"}
+TOOL_RESULT_NEXT_ACTIONS = frozenset(get_args(RecommendedNextAction)) | {"unknown"}
 
 _ERROR_PREFIX = "Error:"
 _PARTIAL_MARKERS = (
@@ -33,10 +39,10 @@ _PARTIAL_MARKERS = (
 
 @dataclass(frozen=True, slots=True)
 class ToolResultMeta:
-    status: Literal["success", "error", "partial_success"]
+    status: ToolResultStatus
     error_type: str | None
     recoverable_by_model: bool
-    recommended_next_action: Literal["continue", "rewrite_query", "try_alternative", "summarize", "stop"]
+    recommended_next_action: RecommendedNextAction
     source: Literal["exception", "tool_return", "content_analysis", "progress_middleware"]
 
 
@@ -97,6 +103,7 @@ _PAGE_CONTENT_TOOL_NAMES: frozenset[str] = frozenset({"web_fetch"})
 # _ERROR_RULES already declares. Derived rather than duplicated so a shell can
 # never drift from the recoverable/next-action contract of its own category.
 _ATTRS_BY_ERROR_TYPE: dict[str, dict[str, object]] = {str(attrs["error_type"]): attrs for _keywords, attrs in _ERROR_RULES}
+TOOL_RESULT_ERROR_TYPES = frozenset(_ATTRS_BY_ERROR_TYPE) | {"unknown", PROGRESS_GUARD_ERROR_TYPE}
 
 # Reason phrases (RFC 9110 §15 plus the wording real servers ship) mapped onto the
 # error_type they already have in _ERROR_RULES. Restricted to the statuses a fetch
diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md
index b6e2ec6d6..f7e1e8cd1 100644
--- a/backend/packages/harness/deerflow/runtime/AGENTS.md
+++ b/backend/packages/harness/deerflow/runtime/AGENTS.md
@@ -135,6 +135,24 @@ their per-execution parent-loop proxy, preserving separate events when two
 different delegated agents promote the same tool. The active catalog is fixed
 for one graph execution, so the claim needs no persisted catalog hash.
 
+**Tool-progress phase events** (`agents/middlewares/tool_progress_middleware.py`):
+effective ACTIVE → WARNED, WARNED/ACTIVE → BLOCKED, WARNED → ACTIVE recovery,
+and later-invocation WARNED/BLOCKED → ACTIVE resets append
+`middleware:tool_progress` through `RunJournal`. Recorder calls happen after
+the middleware releases its state lock, matching LoopDetectionMiddleware, so a
+slow recorder cannot stall tool-state updates. Cross-thread middleware
+producers (currently slash-skill activation via `asyncio.to_thread`) schedule
+journal mutation directly onto its owning event loop; they never mutate or
+flush `RunJournal._buffer` from the worker thread. The task-tool subagent proxy
+rejects a loop that differs from the journal owner, so its close fence always
+drains the only scheduling hop.
+The persisted projection accepts
+only framework-defined error/action values and strict booleans (using null for
+invalid values) from the producer-supplied tool stamp; tool content, args,
+prompts, and derived hashes do not enter the event. Ordinary task-tool subagents use the narrow parent-loop
+recorder proxy, never the journal itself; durable batch runs have no parent
+journal and emit no such event. Recorder failures are fail-open.
+
 **JSONL record boundaries** (`runtime/events/store/jsonl.py`): thread reads,
 run reads, and sequence recovery split on physical newlines. Do not use
 `str.splitlines()`: U+0085/U+2028/U+2029 inside valid JSON strings must remain
diff --git a/backend/packages/harness/deerflow/runtime/events/catalog.py b/backend/packages/harness/deerflow/runtime/events/catalog.py
index 606bc39c4..a0e85aed5 100644
--- a/backend/packages/harness/deerflow/runtime/events/catalog.py
+++ b/backend/packages/harness/deerflow/runtime/events/catalog.py
@@ -82,6 +82,7 @@ MIDDLEWARE_SAFETY_TERMINATION_TAG = "safety_termination"
 MIDDLEWARE_SKILL_ACTIVATION_TAG = "skill_activation"
 MIDDLEWARE_SKILL_SECRETS_TAG = "skill_secrets"
 MIDDLEWARE_TOOL_PROMOTION_TAG = "tool_promotion"
+MIDDLEWARE_TOOL_PROGRESS_TAG = "tool_progress"
 MIDDLEWARE_EVENT_TAGS = (
     MIDDLEWARE_GUARDRAIL_TAG,
     MIDDLEWARE_LOOP_DETECTION_TAG,
@@ -89,6 +90,7 @@ MIDDLEWARE_EVENT_TAGS = (
     MIDDLEWARE_SKILL_ACTIVATION_TAG,
     MIDDLEWARE_SKILL_SECRETS_TAG,
     MIDDLEWARE_TOOL_PROMOTION_TAG,
+    MIDDLEWARE_TOOL_PROGRESS_TAG,
 )
 
 JOURNAL_RUN_EVENT_DEFINITIONS = (
diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py
index 3aa4ff446..9a91980cb 100644
--- a/backend/packages/harness/deerflow/runtime/journal.py
+++ b/backend/packages/harness/deerflow/runtime/journal.py
@@ -25,6 +25,7 @@ from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
 from copy import deepcopy
 from dataclasses import dataclass
 from datetime import UTC, datetime
+from functools import partial
 from typing import TYPE_CHECKING, Any, cast
 from uuid import UUID
 
@@ -250,11 +251,16 @@ class RunJournal(BaseCallbackHandler):
         self._flush_threshold = flush_threshold
         self._progress_reporter = progress_reporter
         self._progress_flush_interval = progress_flush_interval
+        try:
+            self._owner_loop: asyncio.AbstractEventLoop | None = asyncio.get_running_loop()
+        except RuntimeError:
+            self._owner_loop = None
 
         # Write buffer
         self._buffer: list[dict] = []
         self._pending_llm_response: _PendingLlmResponse | None = None
         self._pending_flush_tasks: set[asyncio.Task[None]] = set()
+        self._explicit_flush_in_progress = False
         self._pending_progress_task: asyncio.Task[None] | None = None
         self._pending_progress_delayed = False
         self._progress_dirty = False
@@ -829,7 +835,7 @@ class RunJournal(BaseCallbackHandler):
             return
         # Skip if a flush is already in flight — avoids concurrent writes
         # to the same SQLite file from multiple fire-and-forget tasks.
-        if self._pending_flush_tasks:
+        if self._pending_flush_tasks or self._explicit_flush_in_progress:
             return
         try:
             loop = asyncio.get_running_loop()
@@ -995,8 +1001,31 @@ class RunJournal(BaseCallbackHandler):
             action: Specific action performed (e.g., "generate_title").
             changes: Dict describing the state changes made.
         """
+        event_type = MIDDLEWARE_EVENT_PATTERN.event_type(tag)
+        owner_loop = self._owner_loop
+        try:
+            running_loop = asyncio.get_running_loop()
+        except RuntimeError:
+            running_loop = None
+        if owner_loop is not None and running_loop is not owner_loop:
+            if owner_loop.is_closed() or not owner_loop.is_running():
+                logger.warning("Dropping cross-thread middleware event after run loop shutdown")
+                return
+            try:
+                owner_loop.call_soon_threadsafe(
+                    partial(
+                        self._put,
+                        event_type=event_type,
+                        category=MIDDLEWARE_EVENT_PATTERN.category,
+                        content={"name": name, "hook": hook, "action": action, "changes": dict(changes)},
+                    )
+                )
+            except RuntimeError:
+                logger.warning("Dropping cross-thread middleware event after run loop shutdown")
+            return
+
         self._put(
-            event_type=MIDDLEWARE_EVENT_PATTERN.event_type(tag),
+            event_type=event_type,
             category=MIDDLEWARE_EVENT_PATTERN.category,
             content={"name": name, "hook": hook, "action": action, "changes": changes},
         )
@@ -1068,39 +1097,43 @@ class RunJournal(BaseCallbackHandler):
         """Force flush remaining buffer. Called in worker's finally block."""
         if self._closed:
             return
-        self._commit_pending_llm_response()
-        if self._pending_flush_tasks:
-            await asyncio.gather(*tuple(self._pending_flush_tasks), return_exceptions=True)
-        while self._pending_progress_task is not None:
-            pending_progress_task = self._pending_progress_task
-            if pending_progress_task.done():
-                if self._pending_progress_task is pending_progress_task:
-                    self._pending_progress_task = None
-                break
-            if self._pending_progress_delayed:
-                pending_progress_task.cancel()
+        self._explicit_flush_in_progress = True
+        try:
+            self._commit_pending_llm_response()
+            if self._pending_flush_tasks:
+                await asyncio.gather(*tuple(self._pending_flush_tasks), return_exceptions=True)
+            while self._pending_progress_task is not None:
+                pending_progress_task = self._pending_progress_task
+                if pending_progress_task.done():
+                    if self._pending_progress_task is pending_progress_task:
+                        self._pending_progress_task = None
+                    break
+                if self._pending_progress_delayed:
+                    pending_progress_task.cancel()
+                    await asyncio.gather(pending_progress_task, return_exceptions=True)
+                    if self._pending_progress_task is pending_progress_task:
+                        self._pending_progress_task = None
+                    self._progress_dirty = False
+                    self._pending_progress_delayed = False
+                    break
                 await asyncio.gather(pending_progress_task, return_exceptions=True)
                 if self._pending_progress_task is pending_progress_task:
                     self._pending_progress_task = None
-                self._progress_dirty = False
-                self._pending_progress_delayed = False
-                break
-            await asyncio.gather(pending_progress_task, return_exceptions=True)
-            if self._pending_progress_task is pending_progress_task:
-                self._pending_progress_task = None
 
-        while self._buffer:
-            batch = self._buffer[: self._flush_threshold]
-            del self._buffer[: self._flush_threshold]
-            try:
-                store = self._store
-                if store is None:
-                    return
-                await store.put_batch(batch)
-                self._feed_generation += 1
-            except Exception:
-                self._buffer = batch + self._buffer
-                raise
+            while self._buffer:
+                batch = self._buffer[: self._flush_threshold]
+                del self._buffer[: self._flush_threshold]
+                try:
+                    store = self._store
+                    if store is None:
+                        return
+                    await store.put_batch(batch)
+                    self._feed_generation += 1
+                except Exception:
+                    self._buffer = batch + self._buffer
+                    raise
+        finally:
+            self._explicit_flush_in_progress = False
 
     def _detach_runtime_dependencies(self) -> None:
         """Drop every external or potentially cyclic run-scoped reference."""
@@ -1110,6 +1143,7 @@ class RunJournal(BaseCallbackHandler):
         self._buffer.clear()
         self._pending_llm_response = None
         self._pending_flush_tasks.clear()
+        self._explicit_flush_in_progress = False
         self._pending_progress_task = None
         self._pending_progress_delayed = False
         self._progress_dirty = False
diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py
index d75431bd9..ec818bcd0 100644
--- a/backend/packages/harness/deerflow/runtime/runs/worker.py
+++ b/backend/packages/harness/deerflow/runtime/runs/worker.py
@@ -521,6 +521,11 @@ _SERVER_OWNED_RUNTIME_CONTEXT_KEYS: Final[frozenset[str]] = (
             CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY,
             DEERFLOW_TRACE_METADATA_KEY,
             CONVERSATION_READER_CONTEXT_KEY,
+            "is_subagent",
+            "agent_id",
+            "__run_loop_detection_recorder",
+            "__run_tool_promotion_recorder",
+            "__run_tool_progress_recorder",
         }
     )
     | SANDBOX_SERVER_OWNED_CONTEXT_KEYS
diff --git a/backend/packages/harness/deerflow/subagents/AGENTS.md b/backend/packages/harness/deerflow/subagents/AGENTS.md
index 5cc9ff5b5..df6060190 100644
--- a/backend/packages/harness/deerflow/subagents/AGENTS.md
+++ b/backend/packages/harness/deerflow/subagents/AGENTS.md
@@ -38,6 +38,6 @@ executions are not checked, and acceptance never changes automatic retry policy.
 
 **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 and deferred-tool-promotion audit records cross this boundary through a deliberately narrow exception: `task_tool` captures the parent loop and passes one `_ParentLoopMiddlewareRecorderProxy` to `SubagentExecutor`, which installs it under separate loop-detection and tool-promotion context keys and schedules the real journal append back onto its owner loop with `call_soon_threadsafe`. The proxy exposes only `record_middleware` plus an execution-local atomic promotion claim; that claim deduplicates overlapping parallel `tool_search` calls that read the same pre-step state. Other subagent middleware consumers still do not see `__run_journal`. The task tool closes the shared proxy once 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.
+Loop-detection, deferred-tool-promotion, and tool-progress audit records cross this boundary through one narrow exception. `task_tool` captures the parent loop and passes a `_ParentLoopMiddlewareRecorderProxy` to `SubagentExecutor`, which installs separate context keys and schedules `record_middleware` on the journal's owner loop. The proxy also owns the execution-local atomic promotion claim that deduplicates parallel `tool_search` calls. Other middleware consumers never see `__run_journal`. Close the shared proxy once before `task` returns to fence late events and drain accepted appends before completion capture. Only the proxy crosses loops; do not broaden it or call the event store from a subagent loop. Durable batch subagents have no parent journal and 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 f51c76892..0e2bd98b3 100644
--- a/backend/packages/harness/deerflow/subagents/executor.py
+++ b/backend/packages/harness/deerflow/subagents/executor.py
@@ -26,7 +26,11 @@ 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, TOOL_PROMOTION_RECORDER_CONTEXT_KEY
+from deerflow.agents.middlewares.audit_context import (
+    LOOP_DETECTION_RECORDER_CONTEXT_KEY,
+    TOOL_PROGRESS_RECORDER_CONTEXT_KEY,
+    TOOL_PROMOTION_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
@@ -791,6 +795,7 @@ class SubagentExecutor:
         acceptance_criteria: list[str] | None = None,
         loop_detection_recorder: Any | None = None,
         tool_promotion_recorder: Any | None = None,
+        tool_progress_recorder: Any | None = None,
         context_snapshot: ParentContextSnapshot | None = None,
     ):
         """Initialize the executor.
@@ -841,6 +846,8 @@ class SubagentExecutor:
                 ``RunJournal`` itself.
             tool_promotion_recorder: Optional loop-safe recorder for deferred-tool
                 promotion events. It follows the same isolated-loop boundary.
+            tool_progress_recorder: Optional loop-safe recorder for tool-progress
+                phase transitions. It follows the same isolated-loop boundary.
             context_snapshot: Optional immutable parent history captured by the
                 ordinary task tool at dispatch. Rendered as background data,
                 never as child execution evidence or inherited system authority.
@@ -894,6 +901,7 @@ class SubagentExecutor:
         self.acceptance_criteria = acceptance_criteria
         self.loop_detection_recorder = loop_detection_recorder
         self.tool_promotion_recorder = tool_promotion_recorder
+        self.tool_progress_recorder = tool_progress_recorder
 
         self._base_tools = _filter_tools(
             tools,
@@ -1511,6 +1519,8 @@ class SubagentExecutor:
                 context[LOOP_DETECTION_RECORDER_CONTEXT_KEY] = self.loop_detection_recorder
             if self.tool_promotion_recorder is not None:
                 context[TOOL_PROMOTION_RECORDER_CONTEXT_KEY] = self.tool_promotion_recorder
+            if self.tool_progress_recorder is not None:
+                context[TOOL_PROGRESS_RECORDER_CONTEXT_KEY] = self.tool_progress_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/AGENTS.md b/backend/packages/harness/deerflow/tools/AGENTS.md
index aefc0f888..7d8e822e8 100644
--- a/backend/packages/harness/deerflow/tools/AGENTS.md
+++ b/backend/packages/harness/deerflow/tools/AGENTS.md
@@ -26,7 +26,7 @@ the same host reader serves; keep reading guidance separate from permission enfo
    - `batch_task`, `batch_status`, `cancel_batch` - Explicit durable batch submission/progress/cancellation. Added only while the startup SQL-backed batch submitter is installed; large results stay in the owner-scoped API/JSONL export rather than the lead context. Items accept optional `acceptance_criteria`; item queries and exports expose the separate `acceptance_verdict`. Progress counts describe execution, not acceptance; unmet and UNVERIFIED conditions never trigger automatic retries.
    - Direct `create_deerflow_agent` integrations receive cloned tools bound to their explicit `SubagentRuntime`. The bound `task` forwards that runtime's exact execution controller and optional caller-owned `AppConfig` into registry/model/tool resolution and `SubagentExecutor`; bound batch tools use the same config snapshot and resolve only that runtime's submitter before falling back to no other application's active worker. Keep the original tool name/schema unchanged so model contracts and user-tool deduplication remain stable.
 
-The ordinary `task` boundary carries one narrow parent-loop middleware recorder into the isolated subagent runtime under separate loop-detection and tool-promotion keys. It schedules only `record_middleware` calls back onto the loop that owns `RunJournal`, keeps an execution-local atomic promotion claim so parallel searches do not double-report one new schema, is fenced and drained once before `task` returns, and never exposes the journal or event store to the child loop. Durable batch tasks have no parent run journal and do not use this bridge.
+The ordinary `task` boundary carries one narrow parent-loop middleware recorder into the isolated subagent runtime under separate loop-detection, tool-promotion, and tool-progress keys. It schedules only `record_middleware` calls back onto the loop that owns `RunJournal`, keeps an execution-local atomic promotion claim so parallel searches do not double-report one new schema, is fenced and drained once before `task` returns, and never exposes the journal or event store to the child loop. Durable batch tasks have no parent run journal and do not use this bridge.
 
 Scheduled-task runtime note:
 - Scheduled background runs set `context.non_interactive=true` and therefore exclude `ask_clarification` from the lead-agent tool list. This keeps scheduler-triggered runs from stalling on human confirmation mid-execution. `non_interactive` is an internal-only context key: it is merged from `body.context` only when the request authenticated as the process-internal user (the scheduler path), never from arbitrary HTTP/IM clients.
diff --git a/backend/packages/harness/deerflow/tools/builtins/task_tool.py b/backend/packages/harness/deerflow/tools/builtins/task_tool.py
index a9c1ea3b4..13951b40c 100644
--- a/backend/packages/harness/deerflow/tools/builtins/task_tool.py
+++ b/backend/packages/harness/deerflow/tools/builtins/task_tool.py
@@ -96,6 +96,9 @@ class _ParentLoopMiddlewareRecorderProxy:
     """
 
     def __init__(self, journal: Any, loop: asyncio.AbstractEventLoop) -> None:
+        journal_owner_loop = getattr(journal, "_owner_loop", None)
+        if isinstance(journal_owner_loop, asyncio.AbstractEventLoop) and journal_owner_loop is not loop:
+            raise ValueError("subagent middleware recorder loop must match the RunJournal owner loop")
         self._journal = journal
         self._loop = loop
         self._state_lock = threading.Lock()
@@ -941,6 +944,7 @@ async def task_tool(
         )
         executor_kwargs["loop_detection_recorder"] = middleware_recorder
         executor_kwargs["tool_promotion_recorder"] = middleware_recorder
+        executor_kwargs["tool_progress_recorder"] = middleware_recorder
     if resolved_app_config is not None:
         executor_kwargs["app_config"] = resolved_app_config
     if run_extensions is not None:
diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py
index e2ca007ea..fd3984403 100644
--- a/backend/tests/test_gateway_services.py
+++ b/backend/tests/test_gateway_services.py
@@ -3276,6 +3276,28 @@ def test_strip_internal_context_keys_scrubs_config_smuggled_context_only_keys():
     assert "disable_clarification" not in via_configurable["configurable"]
 
 
+def test_strip_internal_context_keys_scrubs_audit_attribution_and_recorders():
+    from app.gateway.services import build_run_config, strip_internal_context_keys
+
+    server_owned = {
+        "is_subagent": True,
+        "agent_id": "forged-agent",
+        "__run_loop_detection_recorder": "forged",
+        "__run_tool_promotion_recorder": "forged",
+        "__run_tool_progress_recorder": "forged",
+    }
+    config = build_run_config(
+        "thread-1",
+        {"context": dict(server_owned), "configurable": dict(server_owned)},
+        None,
+    )
+
+    strip_internal_context_keys(config)
+
+    for section in ("context", "configurable"):
+        assert not server_owned.keys() & config[section].keys()
+
+
 def test_start_run_sequence_drops_context_only_keys_for_session_caller():
     """Replay the real ``start_run`` assembly order for a session-authenticated caller
     that pushes the keys through *both* smuggling surfaces at once."""
diff --git a/backend/tests/test_loop_detection_middleware.py b/backend/tests/test_loop_detection_middleware.py
index 2d5710528..6806e6856 100644
--- a/backend/tests/test_loop_detection_middleware.py
+++ b/backend/tests/test_loop_detection_middleware.py
@@ -941,7 +941,6 @@ class TestLoopDetectionRunEvents:
         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(
@@ -960,6 +959,26 @@ class TestLoopDetectionRunEvents:
         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_lead_attribution_ignores_caller_supplied_subagent_fields(self):
+        journal = MagicMock()
+        runtime = self._runtime_with_journal(journal)
+        runtime.context["is_subagent"] = True
+        runtime.context["agent_id"] = "forged-agent"
+        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
+
+        changes = journal.record_middleware.call_args.kwargs["changes"]
+        assert changes["is_subagent"] is False
+        assert changes["agent_id"] is None
+
     def test_identical_call_hard_stop_records_event(self):
         journal = MagicMock()
         runtime = self._runtime_with_journal(journal)
diff --git a/backend/tests/test_run_event_stream_contract.py b/backend/tests/test_run_event_stream_contract.py
index 78d100ec4..e2fd2a7ff 100644
--- a/backend/tests/test_run_event_stream_contract.py
+++ b/backend/tests/test_run_event_stream_contract.py
@@ -18,6 +18,7 @@ from deerflow.runtime.events.catalog import (
     MIDDLEWARE_EVENT_PATTERN,
     MIDDLEWARE_EVENT_TAG_MAX_LENGTH,
     MIDDLEWARE_EVENT_TAGS,
+    MIDDLEWARE_TOOL_PROGRESS_TAG,
     MIDDLEWARE_TOOL_PROMOTION_TAG,
     RUN_EVENT_CATEGORY_MAX_LENGTH,
     RUN_EVENT_TYPE_MAX_LENGTH,
@@ -377,6 +378,15 @@ def test_tool_promotion_tag_is_declared_and_fits_the_persisted_event_type():
     assert len(MIDDLEWARE_EVENT_PATTERN.event_type(MIDDLEWARE_TOOL_PROMOTION_TAG)) <= RUN_EVENT_TYPE_MAX_LENGTH
 
 
+def test_tool_progress_tag_is_declared_and_fits_the_persisted_event_type():
+    pattern = _load_contract()["dynamic_event_patterns"][0]
+
+    assert MIDDLEWARE_TOOL_PROGRESS_TAG == "tool_progress"
+    assert MIDDLEWARE_TOOL_PROGRESS_TAG in MIDDLEWARE_EVENT_TAGS
+    assert MIDDLEWARE_TOOL_PROGRESS_TAG in pattern["known_tags"]
+    assert len(MIDDLEWARE_EVENT_PATTERN.event_type(MIDDLEWARE_TOOL_PROGRESS_TAG)) <= RUN_EVENT_TYPE_MAX_LENGTH
+
+
 def test_subagent_observed_events_exactly_match_its_catalog_and_payloads():
     long_result = "r" * (SUBAGENT_STEP_MAX_CHARS + 1)
     long_error = "e" * (SUBAGENT_STEP_MAX_CHARS + 1)
diff --git a/backend/tests/test_run_journal.py b/backend/tests/test_run_journal.py
index c7115e2a0..94a7092d0 100644
--- a/backend/tests/test_run_journal.py
+++ b/backend/tests/test_run_journal.py
@@ -4,6 +4,7 @@ Uses MemoryRunEventStore as the backend for direct event inspection.
 """
 
 import asyncio
+import threading
 import weakref
 from concurrent.futures import ThreadPoolExecutor
 from threading import Barrier
@@ -37,6 +38,137 @@ def test_tool_promotion_claim_is_atomic_across_parallel_sync_wrappers():
     assert sum((result for result in results), []) == ["mcp_a"]
 
 
+@pytest.mark.anyio
+async def test_cross_thread_middleware_events_are_serialized_on_owner_loop():
+    store = MemoryRunEventStore()
+    journal = RunJournal("r-thread", "t-thread", store, flush_threshold=1)
+    owner_thread_id = threading.get_ident()
+    put_thread_ids: list[int] = []
+    original_put = journal._put
+
+    def tracked_put(**kwargs) -> None:
+        put_thread_ids.append(threading.get_ident())
+        original_put(**kwargs)
+
+    journal._put = tracked_put
+
+    def record_from_tool_worker() -> None:
+        journal.record_middleware(
+            "tool_progress",
+            name="ToolProgressMiddleware",
+            hook="wrap_tool_call",
+            action="warn",
+            changes={"from_phase": "active", "to_phase": "warned"},
+        )
+
+    await asyncio.to_thread(record_from_tool_worker)
+    await journal.flush()
+
+    assert put_thread_ids == [owner_thread_id]
+    events = await store.list_events("t-thread", "r-thread")
+    assert [event["event_type"] for event in events] == ["middleware:tool_progress"]
+    assert events[0]["content"]["changes"]["to_phase"] == "warned"
+
+
+def test_middleware_event_without_owner_loop_keeps_cross_thread_append():
+    store = MemoryRunEventStore()
+    journal = RunJournal("r-sync", "t-sync", store, flush_threshold=100)
+
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        pool.submit(
+            journal.record_middleware,
+            "tool_progress",
+            name="ToolProgressMiddleware",
+            hook="wrap_tool_call",
+            action="warn",
+            changes={"from_phase": "active", "to_phase": "warned"},
+        ).result(timeout=5)
+
+    asyncio.run(journal.flush())
+    events = asyncio.run(store.list_events("t-sync", "r-sync"))
+    assert [event["event_type"] for event in events] == ["middleware:tool_progress"]
+
+
+def test_middleware_event_uses_owner_loop_identity_after_loop_moves_threads():
+    loop = asyncio.new_event_loop()
+
+    async def build_journal():
+        return RunJournal("r-moved", "t-moved", MemoryRunEventStore(), flush_threshold=100)
+
+    journal = loop.run_until_complete(build_journal())
+
+    async def record_on_current_loop() -> int:
+        journal.record_middleware(
+            "tool_progress",
+            name="ToolProgressMiddleware",
+            hook="wrap_tool_call",
+            action="warn",
+            changes={},
+        )
+        return len(journal._buffer)
+
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        buffered = pool.submit(loop.run_until_complete, record_on_current_loop()).result(timeout=5)
+
+    assert buffered == 1
+    loop.run_until_complete(journal.flush())
+    loop.close()
+
+
+@pytest.mark.anyio
+async def test_cross_thread_append_during_explicit_flush_is_not_flushed_concurrently():
+    class BlockingStore(MemoryRunEventStore):
+        def __init__(self) -> None:
+            super().__init__()
+            self.started = asyncio.Event()
+            self.release = asyncio.Event()
+            self.active_writes = 0
+            self.max_active_writes = 0
+
+        async def put_batch(self, events):
+            self.active_writes += 1
+            self.max_active_writes = max(self.max_active_writes, self.active_writes)
+            if not self.started.is_set():
+                self.started.set()
+                await self.release.wait()
+            try:
+                await super().put_batch(events)
+            finally:
+                self.active_writes -= 1
+
+    store = BlockingStore()
+    journal = RunJournal("r-flush", "t-flush", store, flush_threshold=1)
+    journal._buffer.append(
+        journal._make_event(
+            event_type="middleware:test",
+            category="middleware",
+            content={},
+        )
+    )
+    flush_task = asyncio.create_task(journal.flush())
+    await store.started.wait()
+
+    await asyncio.to_thread(
+        journal.record_middleware,
+        "tool_progress",
+        name="ToolProgressMiddleware",
+        hook="wrap_tool_call",
+        action="warn",
+        changes={},
+    )
+    await asyncio.sleep(0)
+    assert store.max_active_writes == 1
+
+    store.release.set()
+    await flush_task
+    assert store.max_active_writes == 1
+    events = await store.list_events("t-flush", "r-flush")
+    assert [event["event_type"] for event in events] == [
+        "middleware:test",
+        "middleware:tool_progress",
+    ]
+
+
 @pytest.mark.anyio
 async def test_close_flushes_and_detaches_runtime_dependencies():
     class ProgressReporter:
diff --git a/backend/tests/test_run_worker_rollback.py b/backend/tests/test_run_worker_rollback.py
index 2e6bab9a2..a29e297fd 100644
--- a/backend/tests/test_run_worker_rollback.py
+++ b/backend/tests/test_run_worker_rollback.py
@@ -2532,6 +2532,20 @@ def test_build_runtime_context_ignores_caller_sandbox_execution_identities():
     assert SANDBOX_COMMAND_SCOPE_CONTEXT_KEY not in ctx
 
 
+def test_build_runtime_context_ignores_caller_audit_attribution_and_recorders():
+    caller_context = {
+        "is_subagent": True,
+        "agent_id": "forged-agent",
+        "__run_loop_detection_recorder": object(),
+        "__run_tool_promotion_recorder": object(),
+        "__run_tool_progress_recorder": object(),
+    }
+
+    ctx = _build_runtime_context("thread-1", "run-1", caller_context)
+
+    assert set(caller_context).isdisjoint(ctx)
+
+
 def test_build_runtime_context_ignores_non_dict_caller_context():
     ctx = _build_runtime_context("thread-1", "run-1", "not-a-dict")
     assert ctx == {"thread_id": "thread-1", "run_id": "run-1"}
diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py
index 36afaf482..4ed653f5b 100644
--- a/backend/tests/test_subagent_executor.py
+++ b/backend/tests/test_subagent_executor.py
@@ -3987,6 +3987,7 @@ class TestSubagentGuardrailAttribution:
         run_id=None,
         loop_detection_recorder=None,
         tool_promotion_recorder=None,
+        tool_progress_recorder=None,
         name="general-purpose",
         parent_model="test-model",
     ):
@@ -3999,6 +4000,12 @@ class TestSubagentGuardrailAttribution:
             max_turns=5,
             timeout_seconds=30,
         )
+        recorder_kwargs = {}
+        if tool_progress_recorder is not None:
+            # Kept conditional so the pre-feature attribution cases exercise
+            # the existing constructor surface; only the new propagation case
+            # requires the additive recorder argument.
+            recorder_kwargs["tool_progress_recorder"] = tool_progress_recorder
         return SubagentExecutor(
             config=config,
             tools=[],
@@ -4012,6 +4019,7 @@ class TestSubagentGuardrailAttribution:
             run_id=run_id,
             loop_detection_recorder=loop_detection_recorder,
             tool_promotion_recorder=tool_promotion_recorder,
+            **recorder_kwargs,
         )
 
     @pytest.mark.anyio
@@ -4104,6 +4112,32 @@ class TestSubagentGuardrailAttribution:
         assert "__run_journal" not in context
         assert context.get("agent_id") == "general-purpose"
 
+    @pytest.mark.anyio
+    async def test_aexecute_propagates_narrow_tool_progress_recorder(
+        self,
+        classes,
+        executor_module,
+        monkeypatch,
+    ):
+        """Progress audit crosses the child-loop boundary without the raw journal."""
+        recorder = object()
+        executor = self._make_executor(
+            classes,
+            run_id="run-42",
+            tool_progress_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_tool_progress_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 0098b3329..16101b9f8 100644
--- a/backend/tests/test_task_tool_core_logic.py
+++ b/backend/tests/test_task_tool_core_logic.py
@@ -36,6 +36,18 @@ from deerflow.subagents.status_contract import (
 task_tool_module = importlib.import_module("deerflow.tools.builtins.task_tool")
 
 
+def test_parent_loop_middleware_recorder_requires_the_journal_owner_loop():
+    owner_loop = asyncio.new_event_loop()
+    other_loop = asyncio.new_event_loop()
+    journal = SimpleNamespace(_owner_loop=owner_loop)
+    try:
+        with pytest.raises(ValueError, match="must match"):
+            task_tool_module._ParentLoopMiddlewareRecorderProxy(journal, other_loop)
+    finally:
+        owner_loop.close()
+        other_loop.close()
+
+
 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]] = []
@@ -462,6 +474,7 @@ def test_task_tool_installs_and_closes_narrow_middleware_recorder(monkeypatch):
     kwargs = captured["executor_kwargs"]
     proxy = kwargs["loop_detection_recorder"]
     assert kwargs["tool_promotion_recorder"] is proxy
+    assert kwargs["tool_progress_recorder"] is proxy
     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()
diff --git a/backend/tests/test_tool_progress_middleware.py b/backend/tests/test_tool_progress_middleware.py
index c9addcfe8..a56859426 100644
--- a/backend/tests/test_tool_progress_middleware.py
+++ b/backend/tests/test_tool_progress_middleware.py
@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import logging
+import threading
+from concurrent.futures import ThreadPoolExecutor
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock
 
@@ -16,6 +18,8 @@ from deerflow.agents.middlewares.tool_progress_middleware import (
     word_set,
 )
 from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY
+from deerflow.runtime.events.store.memory import MemoryRunEventStore
+from deerflow.runtime.journal import RunJournal
 
 # ---------------------------------------------------------------------------
 # Helpers
@@ -593,6 +597,8 @@ def test_before_agent_resets_blocked_states_for_new_run():
     mw = _make_mw(stagnation_threshold=1, warn_escalation_count=1)
     rt_run1 = _make_runtime(thread_id="t1", run_id="run-1")
     rt_run2 = _make_runtime(thread_id="t1", run_id="run-2")
+    journal = MagicMock()
+    rt_run2.context["__run_journal"] = journal
     req = _make_tool_request(runtime=rt_run1)
 
     # Drive the tool to BLOCKED via auth error (immediate block, no WARN stage)
@@ -622,6 +628,23 @@ def test_before_agent_resets_blocked_states_for_new_run():
     assert tool_state.consecutive_problems == 0
     assert tool_state.block_reason is None
     assert tool_state.recent_word_sets == ()
+    journal.record_middleware.assert_called_once()
+    reset = journal.record_middleware.call_args.kwargs
+    assert reset["hook"] == "before_agent"
+    assert reset["action"] == "reset"
+    assert reset["changes"] == {
+        "is_subagent": False,
+        "agent_id": None,
+        "tool_name": "web_search",
+        "from_phase": "blocked",
+        "to_phase": "active",
+        "consecutive_problems": 0,
+        "status": None,
+        "error_type": None,
+        "recoverable_by_model": None,
+        "recommended_next_action": None,
+        "threshold": None,
+    }
 
 
 def test_before_agent_resets_warned_states_for_new_run():
@@ -634,6 +657,8 @@ def test_before_agent_resets_warned_states_for_new_run():
     mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5)
     rt_run1 = _make_runtime(thread_id="t1", run_id="run-1")
     rt_run2 = _make_runtime(thread_id="t1", run_id="run-2")
+    journal = MagicMock()
+    rt_run2.context["__run_journal"] = journal
     req = _make_tool_request(runtime=rt_run1)
     error_msg = _make_error_message()
 
@@ -651,6 +676,53 @@ def test_before_agent_resets_warned_states_for_new_run():
     assert tool_state.phase == "active"
     assert tool_state.consecutive_problems == 0
     assert tool_state.recent_word_sets == ()
+    assert journal.record_middleware.call_args.kwargs["action"] == "reset"
+    assert journal.record_middleware.call_args.kwargs["changes"]["from_phase"] == "warned"
+
+
+def test_before_agent_reset_recorder_does_not_hold_the_state_lock():
+    """A slow reset recorder must not stall concurrent state access."""
+
+    class DelayingResetRecorder:
+        def __init__(self):
+            self.reset_started = threading.Event()
+            self.release_reset = threading.Event()
+
+        def record_middleware(self, **kwargs):
+            if kwargs["action"] == "reset":
+                self.reset_started.set()
+                self.release_reset.wait(timeout=30)
+
+    mw = _make_mw(stagnation_threshold=1, warn_escalation_count=1)
+    run_one = _make_runtime(thread_id="t1", run_id="run-1")
+    request = _make_tool_request(runtime=run_one)
+    blocked = ToolMessage(
+        content="Error: invalid api key",
+        tool_call_id="tc-web_search",
+        name="web_search",
+        status="error",
+        additional_kwargs=_meta_kwargs(
+            status="error",
+            error_type="auth",
+            recoverable_by_model=False,
+            recommended_next_action="stop",
+        ),
+    )
+    mw.wrap_tool_call(request, lambda _request: blocked)
+
+    recorder = DelayingResetRecorder()
+    run_two = _make_runtime(thread_id="t1", run_id="run-2")
+    run_two.context["__run_journal"] = recorder
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        reset_future = pool.submit(mw.before_agent, MagicMock(), run_two)
+        assert recorder.reset_started.wait(timeout=5)
+        state_future = pool.submit(mw._get_block_reason, run_two, "web_search")
+        try:
+            assert state_future.result(timeout=5) is None
+        finally:
+            recorder.release_reset.set()
+        reset_future.result(timeout=5)
 
 
 def test_before_agent_resets_active_state_consecutive_problems_and_word_sets():
@@ -954,11 +1026,12 @@ def test_assess_and_transition_blocked_state_immediate_stop_is_idempotent():
 
     auth_meta = ToolResultMeta(**auth_meta_kwargs)
 
-    new_state, hint = mw._assess_and_transition(blocked_state, auth_meta, "")
+    new_state, hint, transition = mw._assess_and_transition(blocked_state, auth_meta, "")
 
     assert new_state.phase == "blocked"
     assert new_state.block_reason is not None
     assert hint is None  # no hint on immediate block path
+    assert transition is None
 
 
 def test_assess_and_transition_blocked_state_non_stop_increments_count():
@@ -986,12 +1059,13 @@ def test_assess_and_transition_blocked_state_non_stop_increments_count():
 
     rate_meta = ToolResultMeta(**rate_meta_kwargs)
 
-    new_state, _hint = mw._assess_and_transition(blocked_state, rate_meta, "")
+    new_state, _hint, transition = mw._assess_and_transition(blocked_state, rate_meta, "")
 
     # Must stay blocked (not regress to warned or active).
     assert new_state.phase == "blocked"
     # Counter must NOT be incremented: blocked is terminal, state returned unchanged.
     assert new_state.consecutive_problems == 3
+    assert transition is None
 
 
 def test_assess_and_transition_blocked_recoverable_does_not_regress_to_warned():
@@ -1021,11 +1095,12 @@ def test_assess_and_transition_blocked_recoverable_does_not_regress_to_warned():
 
     no_results_meta = ToolResultMeta(**no_results_meta_kwargs)
 
-    new_state, hint = mw._assess_and_transition(blocked_state, no_results_meta, "")
+    new_state, hint, transition = mw._assess_and_transition(blocked_state, no_results_meta, "")
 
     assert new_state.phase == "blocked", "blocked must not regress to warned even when the new error is recoverable"
     assert hint is None
     assert new_state is blocked_state  # exact same object returned (no copy)
+    assert transition is None
 
 
 # ---------------------------------------------------------------------------
@@ -1382,6 +1457,290 @@ def test_log_warned_to_active_reset_emits_info(caplog):
     assert "web_search" in reset_records[0].message
 
 
+class TestToolProgressRunEvents:
+    """Durable audit coverage for state-machine interventions.
+
+    The tool result feed records what a tool returned, but it cannot prove that
+    ToolProgressMiddleware crossed a phase boundary and changed later runtime
+    behavior.  Persist exactly those phase transitions, without copying tool
+    arguments or result content into the middleware event.
+    """
+
+    @staticmethod
+    def _runtime_with_journal(journal):
+        runtime = _make_runtime()
+        runtime.context["__run_journal"] = journal
+        return runtime
+
+    def test_lead_warn_and_recover_transitions_are_recorded_without_result_content(self):
+        journal = MagicMock()
+        runtime = self._runtime_with_journal(journal)
+        middleware = _make_mw(stagnation_threshold=2, warn_escalation_count=2)
+        request = _make_tool_request(runtime=runtime)
+        secret = "SENSITIVE_TOOL_RESULT_MUST_NOT_BE_PERSISTED"
+        no_results = _make_error_message(content=f"Error: no results found {secret}")
+
+        # The first problem remains ACTIVE; only the phase-changing second call
+        # is a durable intervention.
+        assert middleware.wrap_tool_call(request, lambda _request: no_results) is no_results
+        assert middleware.wrap_tool_call(request, lambda _request: no_results) is no_results
+
+        journal.record_middleware.assert_called_once()
+        warned = journal.record_middleware.call_args
+        assert warned.kwargs["tag"] == "tool_progress"
+        assert warned.kwargs["name"] == "ToolProgressMiddleware"
+        assert warned.kwargs["hook"] == "wrap_tool_call"
+        assert warned.kwargs["action"] == "warn"
+        assert warned.kwargs["changes"] == {
+            "is_subagent": False,
+            "agent_id": None,
+            "tool_name": "web_search",
+            "from_phase": "active",
+            "to_phase": "warned",
+            "consecutive_problems": 2,
+            "status": "error",
+            "error_type": "no_results",
+            "recoverable_by_model": True,
+            "recommended_next_action": "rewrite_query",
+            "threshold": 2,
+        }
+        assert secret not in repr(warned)
+        assert "content" not in warned.kwargs["changes"]
+        assert "args" not in warned.kwargs["changes"]
+
+        recovered_result = _make_tool_message(
+            "fresh evidence with enough distinct words to remain a useful result",
+        )
+        assert middleware.wrap_tool_call(request, lambda _request: recovered_result) is recovered_result
+
+        assert journal.record_middleware.call_count == 2
+        recovered = journal.record_middleware.call_args_list[-1]
+        assert recovered.kwargs["action"] == "recover"
+        assert recovered.kwargs["changes"] == {
+            "is_subagent": False,
+            "agent_id": None,
+            "tool_name": "web_search",
+            "from_phase": "warned",
+            "to_phase": "active",
+            "consecutive_problems": 0,
+            "status": "success",
+            "error_type": None,
+            "recoverable_by_model": True,
+            "recommended_next_action": "continue",
+            "threshold": None,
+        }
+
+    @pytest.mark.anyio
+    async def test_warn_transition_round_trips_through_run_journal(self):
+        store = MemoryRunEventStore()
+        journal = RunJournal("r1", "t1", store, flush_threshold=100)
+        runtime = self._runtime_with_journal(journal)
+        middleware = _make_mw(stagnation_threshold=1)
+        request = _make_tool_request(runtime=runtime)
+
+        middleware.wrap_tool_call(request, lambda _request: _make_error_message())
+        await journal.flush()
+
+        events = await store.list_events("t1", "r1")
+        assert len(events) == 1
+        assert events[0]["event_type"] == "middleware:tool_progress"
+        assert events[0]["category"] == "middleware"
+        assert events[0]["content"]["action"] == "warn"
+        assert events[0]["content"]["changes"]["to_phase"] == "warned"
+
+    def test_lead_immediate_block_transition_is_recorded(self):
+        journal = MagicMock()
+        runtime = self._runtime_with_journal(journal)
+        middleware = _make_mw(stagnation_threshold=5)
+        request = _make_tool_request(runtime=runtime)
+        auth_error = _make_error_message(
+            content="Error: invalid API key",
+            error_type="auth",
+            recoverable_by_model=False,
+            recommended_next_action="stop",
+        )
+
+        assert middleware.wrap_tool_call(request, lambda _request: auth_error) is auth_error
+
+        journal.record_middleware.assert_called_once()
+        blocked = journal.record_middleware.call_args
+        assert blocked.kwargs["tag"] == "tool_progress"
+        assert blocked.kwargs["action"] == "block"
+        assert blocked.kwargs["changes"] == {
+            "is_subagent": False,
+            "agent_id": None,
+            "tool_name": "web_search",
+            "from_phase": "active",
+            "to_phase": "blocked",
+            "consecutive_problems": 1,
+            "status": "error",
+            "error_type": "auth",
+            "recoverable_by_model": False,
+            "recommended_next_action": "stop",
+            "threshold": None,
+        }
+
+    def test_zero_warn_escalation_still_records_active_to_warned(self):
+        journal = MagicMock()
+        runtime = self._runtime_with_journal(journal)
+        middleware = _make_mw(stagnation_threshold=1, warn_escalation_count=0)
+        request = _make_tool_request(runtime=runtime)
+
+        result = _make_error_message()
+        assert middleware.wrap_tool_call(request, lambda _request: result) is result
+
+        journal.record_middleware.assert_called_once()
+        recorded = journal.record_middleware.call_args.kwargs
+        assert recorded["action"] == "warn"
+        assert recorded["changes"]["from_phase"] == "active"
+        assert recorded["changes"]["to_phase"] == "warned"
+        assert recorded["changes"]["threshold"] == 1
+
+    def test_near_duplicate_warn_records_success_status(self):
+        journal = MagicMock()
+        runtime = self._runtime_with_journal(journal)
+        middleware = _make_mw(stagnation_threshold=1)
+        request = _make_tool_request(runtime=runtime)
+        content = "alpha beta gamma delta epsilon zeta eta theta iota kappa lambda"
+        result = _make_tool_message(content)
+
+        middleware.wrap_tool_call(request, lambda _request: result)
+        middleware.wrap_tool_call(request, lambda _request: result)
+
+        recorded = journal.record_middleware.call_args.kwargs
+        assert recorded["action"] == "warn"
+        assert recorded["changes"]["status"] == "success"
+        assert recorded["changes"]["error_type"] is None
+        assert recorded["changes"]["recommended_next_action"] == "continue"
+
+    def test_producer_supplied_meta_is_projected_onto_bounded_audit_values(self):
+        journal = MagicMock()
+        runtime = self._runtime_with_journal(journal)
+        middleware = _make_mw(stagnation_threshold=1)
+        request = _make_tool_request(runtime=runtime)
+        secret = "PRIVATE_EXTENSION_VALUE_" * 50
+        result = _make_tool_message(
+            "ordinary result content",
+            meta_kwargs=_meta_kwargs(
+                status="error",
+                error_type=secret,
+                recoverable_by_model=secret,
+                recommended_next_action=secret,
+                source=secret,
+            ),
+        )
+
+        assert middleware.wrap_tool_call(request, lambda _request: result) is result
+
+        recorded = journal.record_middleware.call_args
+        assert recorded.kwargs["changes"]["error_type"] == "unknown"
+        assert recorded.kwargs["changes"]["recoverable_by_model"] is None
+        assert recorded.kwargs["changes"]["recommended_next_action"] == "unknown"
+        assert recorded.kwargs["changes"]["status"] == "error"
+        assert secret not in repr(recorded)
+
+    def test_recorder_failure_is_fail_open(self, caplog):
+        journal = MagicMock()
+        journal.record_middleware.side_effect = RuntimeError("event store unavailable")
+        runtime = self._runtime_with_journal(journal)
+        middleware = _make_mw(stagnation_threshold=1)
+        request = _make_tool_request(runtime=runtime)
+        no_results = _make_error_message()
+
+        with caplog.at_level(logging.WARNING, logger=_MW_LOGGER):
+            result = middleware.wrap_tool_call(request, lambda _request: no_results)
+
+        assert result is no_results
+        assert middleware._phase_states["t1"]["web_search"].phase == "warned"
+        assert middleware._pending[("t1", "r1")]
+        assert "Failed to record middleware:tool_progress event" in caplog.text
+
+    def test_narrow_subagent_recorder_records_without_crossing_raw_journal(self):
+        recorder = MagicMock()
+        runtime = _make_runtime()
+        runtime.context["__run_tool_progress_recorder"] = recorder
+        runtime.context["agent_id"] = "general-purpose"
+        assert "__run_journal" not in runtime.context
+        middleware = _make_mw(stagnation_threshold=1)
+        request = _make_tool_request(runtime=runtime)
+
+        middleware.wrap_tool_call(request, lambda _request: _make_error_message())
+
+        recorder.record_middleware.assert_called_once()
+        recorded = recorder.record_middleware.call_args
+        assert recorded.kwargs["tag"] == "tool_progress"
+        assert recorded.kwargs["action"] == "warn"
+        assert recorded.kwargs["changes"]["is_subagent"] is True
+        assert recorded.kwargs["changes"]["agent_id"] == "general-purpose"
+
+    def test_lead_attribution_ignores_caller_supplied_subagent_fields(self):
+        journal = MagicMock()
+        runtime = self._runtime_with_journal(journal)
+        runtime.context["is_subagent"] = True
+        runtime.context["agent_id"] = "forged-agent"
+        middleware = _make_mw(stagnation_threshold=1)
+        request = _make_tool_request(runtime=runtime)
+
+        middleware.wrap_tool_call(request, lambda _request: _make_error_message())
+
+        recorded = journal.record_middleware.call_args
+        assert recorded.kwargs["changes"]["is_subagent"] is False
+        assert recorded.kwargs["changes"]["agent_id"] is None
+
+    @pytest.mark.anyio
+    async def test_async_transition_records_actual_hook(self):
+        journal = MagicMock()
+        runtime = self._runtime_with_journal(journal)
+        middleware = _make_mw(stagnation_threshold=1)
+        request = _make_tool_request(runtime=runtime)
+
+        result = _make_error_message()
+        assert await middleware.awrap_tool_call(request, AsyncMock(return_value=result)) is result
+
+        journal.record_middleware.assert_called_once()
+        assert journal.record_middleware.call_args.kwargs["hook"] == "awrap_tool_call"
+
+    def test_slow_recorder_does_not_hold_the_state_lock(self):
+        """A custom recorder cannot stall a second state-machine transition."""
+
+        class DelayingRecorder:
+            def __init__(self):
+                self.warn_started = threading.Event()
+                self.release_warn = threading.Event()
+                self.block_seen = threading.Event()
+                self.warn_timed_out = False
+
+            def record_middleware(self, **kwargs):
+                action = kwargs["action"]
+                if action == "warn":
+                    self.warn_started.set()
+                    self.warn_timed_out = not self.release_warn.wait(timeout=5)
+                elif action == "block":
+                    self.block_seen.set()
+
+        recorder = DelayingRecorder()
+        runtime = self._runtime_with_journal(recorder)
+        middleware = _make_mw(stagnation_threshold=1, warn_escalation_count=1)
+        request = _make_tool_request(runtime=runtime)
+        result = _make_non_recoverable_error_message()
+        handlers_ready = threading.Barrier(2)
+
+        def complete_tool(_request):
+            handlers_ready.wait(timeout=5)
+            return result
+
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            futures = [pool.submit(middleware.wrap_tool_call, request, complete_tool) for _ in range(2)]
+            assert recorder.warn_started.wait(timeout=5)
+            assert recorder.block_seen.wait(timeout=5)
+            recorder.release_warn.set()
+            results = [future.result(timeout=5) for future in futures]
+
+        assert results == [result, result]
+        assert recorder.warn_timed_out is False
+        assert middleware._phase_states["t1"]["web_search"].phase == "blocked"
+
+
 def test_log_hint_injection_emits_debug(caplog):
     mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5)
     rt = _make_runtime()
diff --git a/backend/tests/test_tool_promotion_audit_middleware.py b/backend/tests/test_tool_promotion_audit_middleware.py
index 4737d0a3d..1c8638cb3 100644
--- a/backend/tests/test_tool_promotion_audit_middleware.py
+++ b/backend/tests/test_tool_promotion_audit_middleware.py
@@ -74,6 +74,39 @@ def test_records_only_new_names_from_the_final_current_catalog_command():
     assert "h1" not in persisted
 
 
+def test_narrow_recorder_identifies_subagent_without_caller_flag():
+    recorder = _Recorder()
+    request = _ToolRequest(
+        context={
+            "__run_tool_promotion_recorder": recorder,
+            "agent_id": "general-purpose",
+        }
+    )
+    result = Command(update={"promoted": {"catalog_hash": "h1", "names": ["mcp_a"]}})
+
+    assert _middleware().wrap_tool_call(request, lambda _: result) is result
+
+    assert recorder.calls[0]["changes"]["is_subagent"] is True
+    assert recorder.calls[0]["changes"]["agent_id"] == "general-purpose"
+
+
+def test_lead_attribution_ignores_caller_supplied_subagent_fields():
+    recorder = _Recorder()
+    request = _ToolRequest(
+        context={
+            "__run_journal": recorder,
+            "is_subagent": True,
+            "agent_id": "forged-agent",
+        }
+    )
+    result = Command(update={"promoted": {"catalog_hash": "h1", "names": ["mcp_a"]}})
+
+    assert _middleware().wrap_tool_call(request, lambda _: result) is result
+
+    assert recorder.calls[0]["changes"]["is_subagent"] is False
+    assert recorder.calls[0]["changes"]["agent_id"] is None
+
+
 @pytest.mark.asyncio
 async def test_async_records_promotion_but_repeated_stale_and_non_search_results_do_not():
     recorder = _Recorder()
diff --git a/backend/tests/test_tool_result_meta.py b/backend/tests/test_tool_result_meta.py
index fcb8fa61c..e8e9f4fce 100644
--- a/backend/tests/test_tool_result_meta.py
+++ b/backend/tests/test_tool_result_meta.py
@@ -10,6 +10,7 @@ from langgraph.types import Command
 
 from deerflow.agents.middlewares.tool_result_meta import (
     TOOL_META_KEY,
+    TOOL_RESULT_ERROR_TYPES,
     ToolResultMeta,
     normalize_tool_message,
     normalize_tool_result,
@@ -66,6 +67,7 @@ def test_error_prefix_classification(snippet: str, expected_type: str):
     m = _meta(result)
     assert m["status"] == "error"
     assert m["error_type"] == expected_type
+    assert m["error_type"] in TOOL_RESULT_ERROR_TYPES
     assert m["source"] == "tool_return"
 
 
diff --git a/contracts/run_event_stream_contract.json b/contracts/run_event_stream_contract.json
index fb38aafb9..b45b1c20b 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", "loop_detection", "safety_termination", "skill_activation", "skill_secrets", "tool_promotion"],
+      "known_tags": ["guardrail", "loop_detection", "safety_termination", "skill_activation", "skill_secrets", "tool_promotion", "tool_progress"],
       "event_type_schema": {
         "type": "string",
         "pattern": "^middleware:",
@@ -444,7 +444,7 @@
     {
       "id": "middleware-coverage",
       "status": "partial",
-      "notes": "Loop-detection and deferred-tool promotion events cover lead-agent and ordinary task-tool subagent runs. Durable batch subagents have no parent run journal and do not currently emit either event."
+      "notes": "Loop-detection, deferred-tool promotion, and tool-progress events cover lead-agent and ordinary task-tool subagent runs. Durable batch subagents have no parent run journal and do not currently emit these events."
     },
     {
       "id": "run-scoped-observation-context",