From 3d5cc1efae84e60a0f182fbe64993959bc0bb5c0 Mon Sep 17 00:00:00 2001 From: dong <2970752254@qq.com> Date: Fri, 25 Sep 2026 10:06:31 +0800 Subject: [PATCH] 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 --- .../deerflow/agents/middlewares/AGENTS.md | 2 +- .../harness/deerflow/runtime/journal.py | 61 +++++- backend/tests/test_run_journal.py | 176 ++++++++++++++++++ 3 files changed, 230 insertions(+), 9 deletions(-) diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md index f7576b2632..14f24c2ad6 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md +++ b/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`. diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index d88f03a966..fb49a5ffdd 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/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, diff --git a/backend/tests/test_run_journal.py b/backend/tests/test_run_journal.py index d04cab04cc..91ba063d5e 100644 --- a/backend/tests/test_run_journal.py +++ b/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