Fix/card tool message bug (#3976)

* fix(runtime): add final reconciliation for missed tool messages

* fix(gateway): persist hidden human input card responses

Persist allowlisted hidden human_input_response messages in RunJournal so
Human Input Cards can recover answered state from run_events after checkpoint
compaction. Keep generic internal hidden messages filtered and add regression
coverage for ask_clarification responses.

backend/AGENTS.md

@@ -249,7 +249,7 @@ Lead-agent middlewares are assembled in strict order across three functions: the
 26. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits
 27. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the safety/clarification tail
 28. **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 custom middlewares so LangChain's reverse-order `after_model` dispatch runs it first
-29. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, writes a readable `ToolMessage.content` fallback plus structured `ToolMessage.artifact.human_input` request payload, and interrupts via `Command(goto=END)` (must be last)
+29. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, writes a readable `ToolMessage.content` fallback plus structured `ToolMessage.artifact.human_input` request payload, and interrupts via `Command(goto=END)` (must be last). Because this middleware can short-circuit tool execution before LangChain emits `on_tool_end`, `RunJournal` performs a root-run final reconciliation for allowlisted clarification `ToolMessage`s whose `tool_call_id` was produced by the current run, so human-input request cards remain recoverable from `run_events` after checkpoint compaction. Human Input Card replies are submitted as `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden response sources (currently `ask_clarification`) as `llm.human.input`, which preserves answered-card state after compaction without exposing generic internal hidden context.
 
 ### Configuration System
 

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

@@ -29,6 +29,7 @@ from langchain_core.callbacks import BaseCallbackHandler
 from langchain_core.messages import AIMessage, AnyMessage, BaseMessage, HumanMessage, ToolMessage
 from langgraph.types import Command
 
+from deerflow.agents.human_input import read_human_input_response
 from deerflow.utils.messages import message_to_text
 
 if TYPE_CHECKING:
@@ -37,10 +38,19 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 _LEGACY_SUMMARY_MESSAGE_NAME = "summary"
+_RECONCILED_TOOL_MESSAGE_NAMES = frozenset({"ask_clarification"})
+_PERSISTED_HIDDEN_HUMAN_INPUT_RESPONSE_SOURCES = frozenset({"ask_clarification"})
 
 
-def _is_user_visible_human_message(message: BaseMessage) -> bool:
-    return isinstance(message, HumanMessage) and message.name != _LEGACY_SUMMARY_MESSAGE_NAME and message.additional_kwargs.get("hide_from_ui") is not True
+def _should_persist_human_input_message(message: BaseMessage) -> bool:
+    if not isinstance(message, HumanMessage):
+        return False
+    if message.name == _LEGACY_SUMMARY_MESSAGE_NAME:
+        return False
+    if message.additional_kwargs.get("hide_from_ui") is not True:
+        return True
+    response = read_human_input_response(message.additional_kwargs)
+    return response is not None and response["source"] in _PERSISTED_HIDDEN_HUMAN_INPUT_RESPONSE_SOURCES
 
 
 class RunJournal(BaseCallbackHandler):
@@ -106,6 +116,8 @@ class RunJournal(BaseCallbackHandler):
         # LLM request/response tracking
         self._llm_call_index = 0
         self._seen_llm_starts: set[str] = set()  # langchain run_ids that fired on_chat_model_start
+        self._current_run_tool_call_names: dict[str, str] = {}
+        self._persisted_tool_message_identities: set[str] = set()
 
     # -- Lifecycle callbacks --
 
@@ -161,6 +173,7 @@ class RunJournal(BaseCallbackHandler):
         # represents the user-visible run lifecycle.
         if parent_run_id is not None:
             return
+        self._reconcile_final_tool_messages(outputs)
         self._put(event_type="run.end", category="outputs", content=outputs, metadata={"status": "success"})
         self._flush_sync()
 
@@ -208,7 +221,7 @@ class RunJournal(BaseCallbackHandler):
         if caller == "lead_agent" and not self._first_human_msg and messages:
             for batch in reversed(messages):
                 for m in reversed(batch):
-                    if _is_user_visible_human_message(m):
+                    if _should_persist_human_input_message(m):
                         self.set_first_human_message(m.text)
                         self._put(
                             event_type="llm.human.input",
@@ -245,6 +258,7 @@ class RunJournal(BaseCallbackHandler):
 
         for message in messages:
             caller = self._identify_caller(tags)
+            self._remember_current_run_tool_calls(message, caller=caller)
 
             # Latency
             rid = str(run_id)
@@ -338,15 +352,13 @@ class RunJournal(BaseCallbackHandler):
         try:
             if isinstance(output, ToolMessage):
                 msg = cast(ToolMessage, output)
-                self._put(event_type="llm.tool.result", category="message", content=msg.model_dump())
-                self._record_message_summary(msg)
+                self._persist_tool_result_message(msg)
             elif isinstance(output, Command):
                 cmd = cast(Command, output)
                 messages = cmd.update.get("messages", [])
                 for message in messages:
                     if isinstance(message, BaseMessage):
-                        self._put(event_type="llm.tool.result", category="message", content=message.model_dump())
-                        self._record_message_summary(message)
+                        self._persist_tool_result_message(message)
                     else:
                         logger.warning(f"on_tool_end {run_id}: command update message is not BaseMessage: {type(message)}")
             else:
@@ -356,6 +368,73 @@ class RunJournal(BaseCallbackHandler):
 
     # -- Internal methods --
 
+    @staticmethod
+    def _message_identity(message: BaseMessage) -> str | None:
+        tool_call_id = getattr(message, "tool_call_id", None)
+        if isinstance(tool_call_id, str) and tool_call_id:
+            return f"tool:{tool_call_id}"
+        message_id = getattr(message, "id", None)
+        if isinstance(message_id, str) and message_id:
+            return f"message:{message_id}"
+        return None
+
+    @staticmethod
+    def _tool_call_value(tool_call: Any, key: str) -> Any:
+        if isinstance(tool_call, Mapping):
+            return tool_call.get(key)
+        return getattr(tool_call, key, None)
+
+    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:
+            return
+        tool_calls = getattr(message, "tool_calls", None) or []
+        if not isinstance(tool_calls, list):
+            return
+        for tool_call in tool_calls:
+            tool_call_id = self._tool_call_value(tool_call, "id")
+            if not isinstance(tool_call_id, str) or not tool_call_id:
+                continue
+            name = self._tool_call_value(tool_call, "name")
+            self._current_run_tool_call_names[tool_call_id] = str(name or "")
+
+    def _persist_tool_result_message(self, message: BaseMessage) -> None:
+        self._put(event_type="llm.tool.result", category="message", content=message.model_dump())
+        identity = self._message_identity(message)
+        if identity:
+            self._persisted_tool_message_identities.add(identity)
+        self._record_message_summary(message)
+
+    def _final_output_messages(self, outputs: Any) -> list[Any]:
+        if isinstance(outputs, Mapping):
+            messages = outputs.get("messages", [])
+            return messages if isinstance(messages, list) else []
+        return []
+
+    def _should_reconcile_tool_message(self, message: ToolMessage) -> bool:
+        if message.additional_kwargs.get("hide_from_ui") is True:
+            return False
+        tool_call_id = getattr(message, "tool_call_id", None)
+        if not isinstance(tool_call_id, str) or not tool_call_id:
+            return False
+        tool_call_name = self._current_run_tool_call_names.get(tool_call_id)
+        if tool_call_name is None:
+            return False
+        message_name = getattr(message, "name", None)
+        if message_name not in _RECONCILED_TOOL_MESSAGE_NAMES and tool_call_name not in _RECONCILED_TOOL_MESSAGE_NAMES:
+            return False
+        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):
+            if not isinstance(message, ToolMessage):
+                continue
+            if self._should_reconcile_tool_message(message):
+                self._persist_tool_result_message(message)
+
     def _put(self, *, event_type: str, category: str, content: str | dict = "", metadata: dict | None = None) -> None:
         self._buffer.append(
             {

backend/tests/test_run_journal.py

@@ -233,6 +233,118 @@ class TestToolCallbacks:
         assert isinstance(events, list)
 
 
+class TestFinalToolMessageReconciliation:
+    @pytest.mark.anyio
+    async def test_root_chain_end_reconciles_missing_ask_clarification_tool_message(self, journal_setup):
+        from langchain_core.messages import ToolMessage
+
+        j, store = journal_setup
+        j.on_llm_end(
+            _make_llm_response("", tool_calls=[{"id": "call_clarify", "name": "ask_clarification", "args": {"question": "Which format?"}}]),
+            run_id=uuid4(),
+            parent_run_id=None,
+            tags=["lead_agent"],
+        )
+        tool_msg = ToolMessage(
+            content="Which format?",
+            tool_call_id="call_clarify",
+            name="ask_clarification",
+            artifact={"human_input": {"kind": "human_input_request", "request_id": "clarification:call_clarify"}},
+        )
+
+        j.on_chain_end({"messages": [tool_msg]}, run_id=uuid4())
+        await j.flush()
+
+        messages = await store.list_messages("t1")
+        tool_results = [m for m in messages if m["event_type"] == "llm.tool.result"]
+        assert len(tool_results) == 1
+        assert tool_results[0]["content"]["name"] == "ask_clarification"
+        assert tool_results[0]["content"]["artifact"]["human_input"]["request_id"] == "clarification:call_clarify"
+
+    @pytest.mark.anyio
+    async def test_root_chain_end_does_not_duplicate_tool_message_captured_by_on_tool_end(self, journal_setup):
+        from langchain_core.messages import ToolMessage
+
+        j, store = journal_setup
+        j.on_llm_end(
+            _make_llm_response("", tool_calls=[{"id": "call_clarify", "name": "ask_clarification", "args": {"question": "Which format?"}}]),
+            run_id=uuid4(),
+            parent_run_id=None,
+            tags=["lead_agent"],
+        )
+        tool_msg = ToolMessage(content="Which format?", tool_call_id="call_clarify", name="ask_clarification")
+
+        j.on_tool_end(tool_msg, run_id=uuid4())
+        j.on_chain_end({"messages": [tool_msg]}, run_id=uuid4())
+        await j.flush()
+
+        messages = await store.list_messages("t1")
+        tool_results = [m for m in messages if m["event_type"] == "llm.tool.result"]
+        assert len(tool_results) == 1
+
+    @pytest.mark.anyio
+    async def test_root_chain_end_ignores_retained_old_tool_message_from_previous_run(self, journal_setup):
+        from langchain_core.messages import ToolMessage
+
+        j, store = journal_setup
+        j.on_llm_end(
+            _make_llm_response("", tool_calls=[{"id": "call_current", "name": "ask_clarification", "args": {"question": "Current?"}}]),
+            run_id=uuid4(),
+            parent_run_id=None,
+            tags=["lead_agent"],
+        )
+        retained_old_tool_msg = ToolMessage(content="Old question", tool_call_id="call_old", name="ask_clarification")
+
+        j.on_chain_end({"messages": [retained_old_tool_msg]}, run_id=uuid4())
+        await j.flush()
+
+        messages = await store.list_messages("t1")
+        assert not any(m["event_type"] == "llm.tool.result" for m in messages)
+
+    @pytest.mark.anyio
+    async def test_root_chain_end_ignores_non_allowlisted_tool_message(self, journal_setup):
+        from langchain_core.messages import ToolMessage
+
+        j, store = journal_setup
+        j.on_llm_end(
+            _make_llm_response("", tool_calls=[{"id": "call_search", "name": "web_search", "args": {"query": "deerflow"}}]),
+            run_id=uuid4(),
+            parent_run_id=None,
+            tags=["lead_agent"],
+        )
+        tool_msg = ToolMessage(content="Search result", tool_call_id="call_search", name="web_search")
+
+        j.on_chain_end({"messages": [tool_msg]}, run_id=uuid4())
+        await j.flush()
+
+        messages = await store.list_messages("t1")
+        assert not any(m["event_type"] == "llm.tool.result" for m in messages)
+
+    @pytest.mark.anyio
+    async def test_root_chain_end_ignores_hidden_ask_clarification_tool_message(self, journal_setup):
+        from langchain_core.messages import ToolMessage
+
+        j, store = journal_setup
+        j.on_llm_end(
+            _make_llm_response("", tool_calls=[{"id": "call_clarify", "name": "ask_clarification", "args": {"question": "Hidden?"}}]),
+            run_id=uuid4(),
+            parent_run_id=None,
+            tags=["lead_agent"],
+        )
+        tool_msg = ToolMessage(
+            content="Hidden?",
+            tool_call_id="call_clarify",
+            name="ask_clarification",
+            additional_kwargs={"hide_from_ui": True},
+        )
+
+        j.on_chain_end({"messages": [tool_msg]}, run_id=uuid4())
+        await j.flush()
+
+        messages = await store.list_messages("t1")
+        assert not any(m["event_type"] == "llm.tool.result" for m in messages)
+
+
 class TestCustomEvents:
     @pytest.mark.anyio
     async def test_on_custom_event_not_implemented(self, journal_setup):
@@ -822,6 +934,18 @@ class TestProgressSnapshots:
 class TestChatModelStartHumanMessage:
     """Tests for on_chat_model_start extracting the first human message."""
 
+    @staticmethod
+    def _human_input_response(source: str = "ask_clarification") -> dict:
+        return {
+            "version": 1,
+            "kind": "human_input_response",
+            "source": source,
+            "request_id": "clarification:call-abc",
+            "response_kind": "option",
+            "option_id": "option-2",
+            "value": "staging",
+        }
+
     @pytest.mark.anyio
     async def test_extracts_first_human_message(self, journal_setup):
         """on_chat_model_start captures the first HumanMessage from prompts."""
@@ -884,6 +1008,74 @@ class TestChatModelStartHumanMessage:
         events = await store.list_events("t1", "r1")
         assert not any(e["event_type"] == "llm.human.input" for e in events)
 
+    @pytest.mark.anyio
+    async def test_hidden_human_input_response_is_captured(self, journal_setup):
+        """Hidden HumanInputCard replies are user-authored and must survive compaction."""
+        from langchain_core.messages import HumanMessage
+
+        j, store = journal_setup
+        hidden_response = HumanMessage(
+            content='For your clarification "Which environment?", my answer is: staging',
+            additional_kwargs={
+                "hide_from_ui": True,
+                "human_input_response": self._human_input_response(),
+            },
+        )
+        j.on_chat_model_start({}, [[hidden_response]], run_id=uuid4(), tags=["lead_agent"])
+        await j.flush()
+
+        assert j._first_human_msg == 'For your clarification "Which environment?", my answer is: staging'
+        assert j.get_completion_data()["message_count"] == 1
+        events = await store.list_events("t1", "r1")
+        human_events = [e for e in events if e["event_type"] == "llm.human.input"]
+        assert len(human_events) == 1
+        assert human_events[0]["content"]["additional_kwargs"]["hide_from_ui"] is True
+        assert human_events[0]["content"]["additional_kwargs"]["human_input_response"]["request_id"] == "clarification:call-abc"
+
+    @pytest.mark.anyio
+    async def test_hidden_human_input_response_wins_over_older_visible_prompt(self, journal_setup):
+        """The latest hidden card reply is the run input, not an older visible prompt."""
+        from langchain_core.messages import HumanMessage
+
+        j, store = journal_setup
+        older_prompt = HumanMessage(content="Write a quicksort PDF")
+        hidden_response = HumanMessage(
+            content='For your clarification "Which format?", my answer is: tutorial',
+            additional_kwargs={
+                "hide_from_ui": True,
+                "human_input_response": self._human_input_response(),
+            },
+        )
+        j.on_chat_model_start({}, [[older_prompt, hidden_response]], run_id=uuid4(), tags=["lead_agent"])
+        await j.flush()
+
+        assert j._first_human_msg == 'For your clarification "Which format?", my answer is: tutorial'
+        events = await store.list_events("t1", "r1")
+        human_events = [e for e in events if e["event_type"] == "llm.human.input"]
+        assert len(human_events) == 1
+        assert human_events[0]["content"]["content"] == 'For your clarification "Which format?", my answer is: tutorial'
+
+    @pytest.mark.anyio
+    async def test_hidden_human_input_response_ignores_non_allowlisted_source(self, journal_setup):
+        """Only explicit HumanInputCard sources are persisted while hidden."""
+        from langchain_core.messages import HumanMessage
+
+        j, store = journal_setup
+        hidden_response = HumanMessage(
+            content="Internal approval response",
+            additional_kwargs={
+                "hide_from_ui": True,
+                "human_input_response": self._human_input_response(source="future_approval"),
+            },
+        )
+        j.on_chat_model_start({}, [[hidden_response]], run_id=uuid4(), tags=["lead_agent"])
+        await j.flush()
+
+        assert j._first_human_msg is None
+        assert j.get_completion_data()["message_count"] == 0
+        events = await store.list_events("t1", "r1")
+        assert not any(e["event_type"] == "llm.human.input" for e in events)
+
     @pytest.mark.anyio
     async def test_legacy_summary_message_is_not_captured_as_user_input(self, journal_setup):
         """Legacy synthetic summaries are internal context even if hide_from_ui is absent."""