fix(runtime): persist consumed middleware tool results (#5671)

* fix(runtime): persist consumed middleware tool results

* docs(agents): shorten journal reconciliation guidance

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/packages/harness/deerflow/agents/middlewares/AGENTS.md

@@ -127,4 +127,4 @@ Before changing a later authorization phase, read the [authorization RFC](../../
 34. **TerminalResponseMiddleware** - When a provider returns an empty terminal `AIMessage` after tool execution, injects a hidden recovery prompt and retries the model once; a second empty response is replaced in checkpoint state by a visible error fallback marked for the run worker, so the run finishes as an error instead of a silent success
 35. **ModelLengthFinishReasonMiddleware** - Match stamps `stop_reason=model_length_capped` and ends tool loop. Suppresses calls, appends notice even with partial text, and stamps `model_length_termination` so downstream guards stand down. Preserves content blocks.
 36. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after terminal-response/custom/configured middlewares so LangChain's reverse-order `after_model` dispatch runs it first
-37. **ClarificationMiddleware** - Intercepts `ask_clarification`, writes a readable `ToolMessage.content` fallback plus a structured `ToolMessage.artifact.human_input` payload, and interrupts via `Command(goto=END)` (must be last). `after_model` drops same-turn sibling tool calls so they cannot run before the user answers; a malformed `ask_clarification` parked on `invalid_tool_calls` is the same stop signal. `disable_clarification` runs keep the siblings. Payloads are versioned — legacy `free_text`/`choice_with_other` stay `version: 1`; the v2 `form` mode (from `fields`) is `version: 2` so older frontends reject it and fall back to plain text. Field normalization is deterministic and lives in the middleware (it short-circuits before tool execution, so tool-arg typing gives no runtime validation), and it is atomic: any structurally broken entry — non-dict, bad/duplicate name, a name colliding with a JS `Object.prototype` member (`__proto__`/`constructor`), or exceeding the caps (16 fields / 24 options per field / 200 chars per text / `MAX_FORM_SERIALIZED_BYTES` = 16KB UTF-8, the per-item caps alone admitting forms whose IM text fallback overruns channel limits) — degrades the whole form to the legacy option/free-text modes, so a card never renders "complete" while missing a field. Benign issues degrade locally (unknown types — incl. unhashable JSON like `type: []`, which must not raise from the membership probe — and option-less selects become `text`); options are trimmed/deduped with blanks dropped (form- and top-level) since the frontend rejects blank labels. XML-to-dict option payloads are recursively flattened from dict/list containers in source order, scalar leaves kept, residual XML tags stripped before that trimming. Checkboxes are booleans defaulting to "no"; `required` on one means consent semantics. The response protocol is unchanged (v1 `text`/`option`): form cards submit a text summary as `response_kind: "text"`, so journal persistence needs no new allowlist entries. Because this middleware can short-circuit before `on_tool_end`, `RunJournal` does a root-run reconciliation for `ToolMessage`s whose `tool_call_id` came from the current run, so cards survive checkpoint compaction. That reconciliation is **not** `ask_clarification`-only — any middleware that answers a tool call has the same gap, and a result the user saw must not vanish on reload (#4666 — `ReadBeforeWriteMiddleware` blocked-write errors reached the UI but not the event store). It is bounded by three conditions, not a name allowlist: the message is user-visible, the call belongs to this run's **lead agent** (`_remember_current_run_tool_calls` records lead-agent calls only; subagent results stay in `subagent.step`), and it is not already persisted. Human Input Card replies are `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden sources (currently `ask_clarification`) as `llm.human.input`.
+37. **ClarificationMiddleware** - Intercepts `ask_clarification`, writes a readable `ToolMessage.content` fallback plus a structured `ToolMessage.artifact.human_input` payload, and interrupts via `Command(goto=END)` (must be last). `after_model` drops same-turn sibling tool calls so they cannot run before the user answers; a malformed `ask_clarification` parked on `invalid_tool_calls` is the same stop signal. `disable_clarification` runs keep the siblings. Payloads are versioned — legacy `free_text`/`choice_with_other` stay `version: 1`; the v2 `form` mode (from `fields`) is `version: 2` so older frontends reject it and fall back to plain text. Field normalization is deterministic and lives in the middleware (it short-circuits before tool execution, so tool-arg typing gives no runtime validation), and it is atomic: any structurally broken entry — non-dict, bad/duplicate name, a name colliding with a JS `Object.prototype` member (`__proto__`/`constructor`), or exceeding the caps (16 fields / 24 options per field / 200 chars per text / `MAX_FORM_SERIALIZED_BYTES` = 16KB UTF-8, the per-item caps alone admitting forms whose IM text fallback overruns channel limits) — degrades the whole form to the legacy option/free-text modes, so a card never renders "complete" while missing a field. Benign issues degrade locally (unknown types — incl. unhashable JSON like `type: []`, which must not raise from the membership probe — and option-less selects become `text`); options are trimmed/deduped with blanks dropped (form- and top-level) since the frontend rejects blank labels. XML-to-dict option payloads are recursively flattened from dict/list containers in source order, scalar leaves kept, residual XML tags stripped before that trimming. Checkboxes are booleans defaulting to "no"; `required` on one means consent semantics. The response protocol is unchanged (v1 `text`/`option`): form cards submit a text summary as `response_kind: "text"`, so journal persistence needs no new allowlist entries. `RunJournal` reconciles visible, unpersisted `ToolMessage`s for current-run lead-agent calls at the next lead-agent `on_chat_model_start` after consumption, or at successful root `on_chain_end`. This preserves any middleware short-circuit, including blocked writes, across later model errors and checkpoint compaction (#4666). `_remember_current_run_tool_calls` excludes subagents; their results stay in `subagent.step`. Human Input Card replies are `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden sources (currently `ask_clarification`) as `llm.human.input`.

backend/packages/harness/deerflow/runtime/journal.py

@@ -6,9 +6,10 @@ handles token usage accumulation.
 
 Key design decisions:
 - on_llm_new_token is NOT implemented -- only complete messages via on_llm_end
-- on_chat_model_start captures the first user-visible prompt as llm.human.input and
-  extracts the first human message for run.input, because it is more reliable than
-  on_chain_start (fires on every node) — messages here are fully structured.
+- on_chat_model_start reconciles consumed middleware tool results and captures the
+  first user-visible prompt as llm.human.input. It extracts the first human message
+  for run.input because it is more reliable than on_chain_start (fires on every
+  node) — messages here are fully structured.
 - on_chain_start with parent_run_id=None emits a run.start trace marking root invocation.
 - on_llm_end emits llm.ai.response in checkpoint-aligned AIMessage.model_dump() format
 - Token usage accumulated in memory, written to RunRow on run completion
@@ -412,7 +413,7 @@ class RunJournal(BaseCallbackHandler):
         tags: list[str] | None = None,
         **kwargs: Any,
     ) -> None:
-        """Capture the first user-visible prompt as llm.human.input.
+        """Capture consumed tool results and the first user-visible prompt.
 
         This is also the canonical place to extract the first human message:
         messages are fully structured here, it fires only on real LLM calls,
@@ -433,6 +434,9 @@ class RunJournal(BaseCallbackHandler):
 
         # Capture the first user message sent to the lead agent in this run.
         caller = self._identify_caller(tags)
+        if caller == "lead_agent":
+            for batch in messages:
+                self._reconcile_consumed_tool_messages(batch)
         if caller == "lead_agent" and not self._first_human_msg and messages:
             for batch in reversed(messages):
                 for m in reversed(batch):
@@ -673,11 +677,23 @@ class RunJournal(BaseCallbackHandler):
             return tool_call.get(key)
         return getattr(tool_call, key, None)
 
+    @staticmethod
+    def _is_ai_message(message: Any) -> bool:
+        return isinstance(message, AIMessage) or getattr(message, "type", None) == "ai"
+
+    def _has_current_run_tool_call(self, message: Any) -> bool:
+        if not self._is_ai_message(message):
+            return False
+        for tool_call in getattr(message, "tool_calls", None) or []:
+            tool_call_id = self._tool_call_value(tool_call, "id")
+            if isinstance(tool_call_id, str) and tool_call_id in self._current_run_tool_call_names:
+                return True
+        return False
+
     def _remember_current_run_tool_calls(self, message: AnyMessage, *, caller: str) -> None:
         if caller != "lead_agent":
             return
-        is_ai_message = isinstance(message, AIMessage) or getattr(message, "type", None) == "ai"
-        if not is_ai_message:
+        if not self._is_ai_message(message):
             return
         tool_calls = getattr(message, "tool_calls", None) or []
         if not isinstance(tool_calls, list):
@@ -732,13 +748,42 @@ class RunJournal(BaseCallbackHandler):
         identity = self._message_identity(message)
         return identity is not None and identity not in self._persisted_tool_message_identities
 
-    def _reconcile_final_tool_messages(self, outputs: Any) -> None:
-        for message in self._final_output_messages(outputs):
+    def _reconcile_tool_messages(self, messages: Iterable[Any]) -> None:
+        for message in messages:
             if not isinstance(message, ToolMessage):
                 continue
             if self._should_reconcile_tool_message(message):
                 self._persist_tool_result_message(message)
 
+    def _reconcile_consumed_tool_messages(self, messages: Iterable[Any]) -> None:
+        """Reconcile results after the latest current-run tool call."""
+        suffix = self._current_run_tool_call_suffix(messages)
+        if suffix is not None:
+            self._reconcile_tool_messages(suffix)
+
+    def _current_run_tool_call_suffix(self, messages: Iterable[Any]) -> list[Any] | None:
+        """Return messages after the latest current-run AI tool-call boundary."""
+        batch = list(messages)
+        boundary = None
+        has_ai_message = False
+        for index, message in enumerate(batch):
+            if not self._is_ai_message(message):
+                continue
+            has_ai_message = True
+            if self._has_current_run_tool_call(message):
+                boundary = index
+        if boundary is not None:
+            return batch[boundary + 1 :]
+        # Some callback fixtures provide only the tool result. Preserve that
+        # fallback, but never scan a history containing AI messages without a
+        # current-run boundary.
+        return batch if not has_ai_message else None
+
+    def _reconcile_final_tool_messages(self, outputs: Any) -> None:
+        suffix = self._current_run_tool_call_suffix(self._final_output_messages(outputs))
+        if suffix is not None:
+            self._reconcile_tool_messages(suffix)
+
     def _make_event(self, *, event_type: str, category: str, content: str | dict = "", metadata: dict | None = None) -> dict:
         return {
             "thread_id": self.thread_id,

backend/tests/test_run_journal.py

@@ -773,6 +773,182 @@ class TestFinalToolMessageReconciliation:
         assert len(tool_results) == 1
         assert tool_results[0]["content"]["name"] == "write_file"
 
+    @pytest.mark.anyio
+    async def test_root_chain_end_does_not_reconcile_old_result_when_tool_call_id_is_reused(self, journal_setup):
+        from langchain_core.messages import AIMessage, ToolMessage
+
+        j, store = journal_setup
+        j.on_llm_end(
+            _make_llm_response("", tool_calls=[{"id": "call_shared", "name": "write_file", "args": {}}]),
+            run_id=uuid4(),
+            parent_run_id=None,
+            tags=["lead_agent"],
+        )
+        old_ai = AIMessage(content="", tool_calls=[{"id": "call_shared", "name": "write_file", "args": {}}])
+        old = ToolMessage(content="Old", tool_call_id="call_shared", name="write_file")
+        current_ai = AIMessage(content="", tool_calls=[{"id": "call_shared", "name": "write_file", "args": {}}])
+        current = ToolMessage(content="Current", tool_call_id="call_shared", name="write_file")
+
+        j.on_chain_end({"messages": [old_ai, old, current_ai, current]}, run_id=uuid4())
+        await j.flush()
+
+        messages = await store.list_messages("t1")
+        tool_results = [event for event in messages if event["event_type"] == "llm.tool.result"]
+        assert len(tool_results) == 1
+        assert tool_results[0]["content"]["content"] == "Current"
+
+    @pytest.mark.anyio
+    async def test_model_start_persists_consumed_short_circuited_tool_result_before_later_run_error(self, journal_setup):
+        from langchain_core.messages import AIMessage, ToolMessage
+
+        j, store = journal_setup
+        j.on_llm_end(
+            _make_llm_response("", tool_calls=[{"id": "call_write", "name": "write_file", "args": {"path": "/tmp/a", "content": "x"}}]),
+            run_id=uuid4(),
+            parent_run_id=None,
+            tags=["lead_agent"],
+        )
+        current_ai = AIMessage(content="", tool_calls=[{"id": "call_write", "name": "write_file", "args": {}}])
+        blocked = ToolMessage(
+            id="blocked-write",
+            content="Read the file before modifying it.",
+            tool_call_id="call_write",
+            name="write_file",
+            status="error",
+        )
+
+        j.on_chat_model_start({}, [[current_ai, blocked]], run_id=uuid4(), tags=["lead_agent"])
+        j.on_chain_error(RuntimeError("provider failed"), run_id=uuid4())
+        await j.flush()
+
+        messages = await store.list_messages("t1")
+        tool_results = [event for event in messages if event["event_type"] == "llm.tool.result"]
+        assert len(tool_results) == 1
+        assert tool_results[0]["content"]["id"] == "blocked-write"
+
+    @pytest.mark.anyio
+    async def test_model_start_does_not_duplicate_tool_result_captured_by_on_tool_end(self, journal_setup):
+        from langchain_core.messages import AIMessage, ToolMessage
+
+        j, store = journal_setup
+        j.on_llm_end(
+            _make_llm_response("", tool_calls=[{"id": "call_write", "name": "write_file", "args": {}}]),
+            run_id=uuid4(),
+            parent_run_id=None,
+            tags=["lead_agent"],
+        )
+        current_ai = AIMessage(content="", tool_calls=[{"id": "call_write", "name": "write_file", "args": {}}])
+        blocked = ToolMessage(content="Blocked", tool_call_id="call_write", name="write_file")
+
+        j.on_tool_end(blocked, run_id=uuid4())
+        j.on_chat_model_start({}, [[current_ai, blocked]], run_id=uuid4(), tags=["lead_agent"])
+        await j.flush()
+
+        messages = await store.list_messages("t1")
+        assert len([event for event in messages if event["event_type"] == "llm.tool.result"]) == 1
+
+    @pytest.mark.anyio
+    async def test_model_start_ignores_retained_old_run_tool_result(self, journal_setup):
+        from langchain_core.messages import ToolMessage
+
+        j, store = journal_setup
+        retained = ToolMessage(content="Old result", tool_call_id="call_old", name="write_file")
+
+        j.on_chat_model_start({}, [[retained]], run_id=uuid4(), tags=["lead_agent"])
+        await j.flush()
+
+        messages = await store.list_messages("t1")
+        assert not any(event["event_type"] == "llm.tool.result" for event in messages)
+
+    @pytest.mark.anyio
+    async def test_model_start_ignores_reused_tool_call_id_without_current_run_boundary(self, journal_setup):
+        from langchain_core.messages import AIMessage, ToolMessage
+
+        j, store = journal_setup
+        j.on_llm_end(
+            _make_llm_response("", tool_calls=[{"id": "call_shared", "name": "write_file", "args": {}}]),
+            run_id=uuid4(),
+            parent_run_id=None,
+            tags=["lead_agent"],
+        )
+        unrelated_ai = AIMessage(content="Old response")
+        retained = ToolMessage(content="Old result", tool_call_id="call_shared", name="write_file")
+
+        j.on_chat_model_start({}, [[unrelated_ai, retained]], run_id=uuid4(), tags=["lead_agent"])
+        await j.flush()
+
+        messages = await store.list_messages("t1")
+        assert not any(event["event_type"] == "llm.tool.result" for event in messages)
+
+    @pytest.mark.anyio
+    async def test_model_start_ignores_hidden_tool_result(self, journal_setup):
+        from langchain_core.messages import AIMessage, ToolMessage
+
+        j, store = journal_setup
+        j.on_llm_end(
+            _make_llm_response("", tool_calls=[{"id": "call_hidden", "name": "write_file", "args": {}}]),
+            run_id=uuid4(),
+            parent_run_id=None,
+            tags=["lead_agent"],
+        )
+        hidden = ToolMessage(
+            content="Hidden result",
+            tool_call_id="call_hidden",
+            name="write_file",
+            additional_kwargs={"hide_from_ui": True},
+        )
+
+        current_ai = AIMessage(content="", tool_calls=[{"id": "call_hidden", "name": "write_file", "args": {}}])
+        j.on_chat_model_start({}, [[current_ai, hidden]], run_id=uuid4(), tags=["lead_agent"])
+        await j.flush()
+
+        messages = await store.list_messages("t1")
+        assert not any(event["event_type"] == "llm.tool.result" for event in messages)
+
+    @pytest.mark.anyio
+    async def test_subagent_model_start_ignores_tool_result_with_colliding_call_id(self, journal_setup):
+        from langchain_core.messages import AIMessage, ToolMessage
+
+        j, store = journal_setup
+        j.on_llm_end(
+            _make_llm_response("", tool_calls=[{"id": "call_shared", "name": "write_file", "args": {}}]),
+            run_id=uuid4(),
+            parent_run_id=None,
+            tags=["lead_agent"],
+        )
+        current_ai = AIMessage(content="", tool_calls=[{"id": "call_shared", "name": "write_file", "args": {}}])
+        subagent_result = ToolMessage(content="Subagent result", tool_call_id="call_shared", name="write_file")
+
+        j.on_chat_model_start({}, [[current_ai, subagent_result]], run_id=uuid4(), tags=["subagent:general-purpose"])
+        await j.flush()
+
+        messages = await store.list_messages("t1")
+        assert not any(event["event_type"] == "llm.tool.result" for event in messages)
+
+    @pytest.mark.anyio
+    async def test_model_start_does_not_reconcile_old_result_when_tool_call_id_is_reused(self, journal_setup):
+        from langchain_core.messages import AIMessage, ToolMessage
+
+        j, store = journal_setup
+        j.on_llm_end(
+            _make_llm_response("", tool_calls=[{"id": "call_shared", "name": "write_file", "args": {}}]),
+            run_id=uuid4(),
+            parent_run_id=None,
+            tags=["lead_agent"],
+        )
+        old_ai = AIMessage(content="", tool_calls=[{"id": "call_shared", "name": "write_file", "args": {}}])
+        old = ToolMessage(content="Old", tool_call_id="call_shared", name="write_file")
+        current_ai = AIMessage(content="", tool_calls=[{"id": "call_shared", "name": "write_file", "args": {}}])
+        current = ToolMessage(content="Current blocked", tool_call_id="call_shared", name="write_file")
+
+        j.on_chat_model_start({}, [[old_ai, old, current_ai, current]], run_id=uuid4(), tags=["lead_agent"])
+        await j.flush()
+
+        messages = await store.list_messages("t1")
+        tool_results = [event for event in messages if event["event_type"] == "llm.tool.result"]
+        assert len(tool_results) == 1
+        assert tool_results[0]["content"]["content"] == "Current blocked"
+
 
 class TestCustomEvents:
     @pytest.mark.anyio