diff --git a/backend/AGENTS.md b/backend/AGENTS.md index a6116df4b..9fedb8c53 100644 --- a/backend/AGENTS.md +++ b/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 diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index 7d7b7d4d6..8680f7653 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/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( { diff --git a/backend/tests/test_run_journal.py b/backend/tests/test_run_journal.py index 38d2de84d..495341fd7 100644 --- a/backend/tests/test_run_journal.py +++ b/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."""