diff --git a/CHANGELOG.md b/CHANGELOG.md index e4067b7bc7..e9d9b18af7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,21 @@ This section accumulates work toward the **2.3.0** milestone ### Fixed +- **agents:** A high-risk tool call without a usable id no longer wedges the + thread when `hitl_approval` is enabled. A provider that omits a tool-call id + (`None`) or repeats one within a response made the middleware pause with an + approval payload the Gateway can never validate: every resume answered 422 + and every new run stayed 409-blocked, all-reject included. An empty id was + technically resumable but is denied as well, since strict providers reject it + when the conversation is replayed. Such calls are now removed from the model + response with a visible rejection note instead of pausing, and deny paths no + longer build an invalid `ToolMessage` for a null id. A call whose id an + ordinary call also carries is stripped before it can be answered by result, + so the next request cannot advertise the id twice with two results. + A pause stored before + this change stays unresolvable; in-place recovery is tracked in [#8071]. + ([#8069]) + - **runtime:** A HITL rejection, including the all-reject escape after approval is disabled or the gated tools change during a pause, now reaches the thread's message history under the resume run instead of existing only in checkpoint @@ -16257,5 +16272,6 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#8043]: https://github.com/bytedance/deer-flow/pull/8043 [#8057]: https://github.com/bytedance/deer-flow/pull/8057 [#8064]: https://github.com/bytedance/deer-flow/pull/8064 +[#8069]: https://github.com/bytedance/deer-flow/issues/8069 +[#8071]: https://github.com/bytedance/deer-flow/issues/8071 [#8072]: https://github.com/bytedance/deer-flow/pull/8072 - diff --git a/backend/docs/RUN_INTERACTION_POLICY.md b/backend/docs/RUN_INTERACTION_POLICY.md index b0f994fdf3..1b786db7f7 100644 --- a/backend/docs/RUN_INTERACTION_POLICY.md +++ b/backend/docs/RUN_INTERACTION_POLICY.md @@ -59,6 +59,20 @@ tool callback; rejections have none, so the journal reconciles them under the resume run like other middleware-written results (#4666). Results from other earlier runs are not re-attributed. +A call that cannot carry a stable approval decision is never paused. A null or +duplicated call id can never match the resume validator and would block new runs +forever; an empty id is technically matchable, but it is denied as well because +strict providers reject an empty tool-call id when the conversation is replayed. +The middleware removes those calls from the model response with a visible +rejection note instead of pausing, so the tool is never executed and the thread +stays usable. The same uniqueness rule applies when a call is denied by result: +answering a denial `ToolMessage` also answers any other call that shares its id, +so a shared id is stripped rather than answered, and the next request cannot +carry the id twice. A pause stored by an older release with a null or duplicated +id is **not** cleared by this: every resume, including all-reject, is still +rejected, so continue the work in a new thread. In-place recovery is tracked in +[#8071](https://github.com/bytedance/deer-flow/issues/8071). + `waiting_for_input` is not a terminal success or failure. Goal checks, final artifact delivery checks, and batch continuation cancellation must not treat a pending approval as a completed source run. Pending approvals do not expire and diff --git a/backend/packages/harness/deerflow/agents/middlewares/hitl_approval_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/hitl_approval_middleware.py index f1822ef95e..ad26dd2446 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/hitl_approval_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/hitl_approval_middleware.py @@ -9,6 +9,7 @@ execution chain (validated by the design spike). from __future__ import annotations import logging +from collections import Counter from collections.abc import Mapping from typing import Any @@ -19,6 +20,7 @@ from langgraph.types import interrupt from deerflow.agents.interaction_policy import resolve_run_interaction_policy from deerflow.agents.middlewares.pii_redaction_middleware import redact_text +from deerflow.agents.middlewares.tool_call_metadata import clone_ai_message_with_tool_calls from deerflow.runtime.run_origin import origin_kind_of logger = logging.getLogger(__name__) @@ -64,6 +66,36 @@ def _bounded_text(value: Any) -> str: return f""[:MAX_REASON_CHARS] +def _addressable(calls: list[dict[str, Any]]) -> bool: + """Whether every call can carry a stable approval decision. + + ``ToolMessage`` requires a non-empty string ``tool_call_id``. A pause whose + payload carries a null or duplicated call id can never be validated: the + Gateway resume validator answers 422 (all-reject included) while a pending + pause blocks new runs with 409, so pausing would wedge the thread forever. + An empty id is technically resumable end to end, but strict providers reject + it when the conversation is replayed, so it is denied by the same rule. + These calls are denied by removing them from the model response instead of + by result. + """ + ids = [call.get("id") for call in calls] + return all(isinstance(call_id, str) and call_id for call_id in ids) and len(set(ids)) == len(ids) + + +def _unambiguous_ids(last: AIMessage, calls: list[dict[str, Any]]) -> bool: + """Whether every targeted call has an id no other call in the response shares. + + ``_addressable`` only asks whether a ``ToolMessage`` can answer the targeted + calls. An id shared with an ordinary call makes the approval ambiguous in + both directions: approving, rejecting or denying one call would also answer + the other by id. Such a group is denied before pausing or answering by result. + """ + if not _addressable(calls): + return False + response_ids = Counter(tool_call.get("id") for tool_call in last.tool_calls) + return all(response_ids[call.get("id")] == 1 for call in calls) + + def _deny_message(tool_call: Mapping[str, Any], reason: str) -> ToolMessage: return ToolMessage( content=reason, @@ -73,6 +105,29 @@ def _deny_message(tool_call: Mapping[str, Any], reason: str) -> ToolMessage: ) +def _append_visible_note(content: Any, note: str) -> Any: + if isinstance(content, list): + return [*content, {"type": "text", "text": note}] + text = content if isinstance(content, str) else "" + return f"{text}\n\n{note}" if text else note + + +def _strip_calls(last: AIMessage, calls: list[dict[str, Any]], reason: str) -> dict[str, Any]: + """Deny calls that cannot carry a result by removing them from the response. + + A non-target call can share an id with a removed one. ``clone_ai_message_with_tool_calls`` + filters the raw provider payload and content tool-call blocks by id, so such + a survivor would keep the removed call's raw entry, and the next request + would advertise a call that has no result. Drop the whole colliding id group + instead: an id shared with a denied approval call cannot be answered + unambiguously either. + """ + dropped_ids = {call["id"] for call in calls if isinstance(call.get("id"), str) and call["id"]} + dropped = {id(call) for call in calls} + kept = [tool_call for tool_call in last.tool_calls if id(tool_call) not in dropped and not (isinstance(tool_call.get("id"), str) and tool_call["id"] in dropped_ids)] + return {"messages": [clone_ai_message_with_tool_calls(last, kept, content=_append_visible_note(last.content, reason))]} + + class HitlApprovalMiddleware(AgentMiddleware): """Pause high-risk tool calls for human approval. @@ -102,7 +157,13 @@ class HitlApprovalMiddleware(AgentMiddleware): def _target_calls(self, last: AIMessage) -> list[dict[str, Any]]: return [tc for tc in last.tool_calls if tc["name"] in self.tool_names] - def _deny_all(self, calls: list[dict[str, Any]], reason: str) -> dict[str, Any]: + def _deny_all(self, last: AIMessage, calls: list[dict[str, Any]], reason: str) -> dict[str, Any]: + # A denial ``ToolMessage`` answers a call by id, so it needs an id no + # other call in the response shares: otherwise the ordinary call is + # answered too while staying serialized, and the next request carries + # the id twice with two results. Strip the whole colliding group instead. + if not _unambiguous_ids(last, calls): + return _strip_calls(last, calls, reason) return {"messages": [_deny_message(tc, reason) for tc in calls]} def _is_subagent(self, runtime: Any) -> bool: @@ -147,18 +208,25 @@ class HitlApprovalMiddleware(AgentMiddleware): if not targets: return None if self.max_actions and len(targets) > self.max_actions: - return self._deny_all(targets, _too_many_reason(len(targets), self.max_actions)) + return self._deny_all(last, targets, _too_many_reason(len(targets), self.max_actions)) if not self._can_resume(runtime): if self.non_interactive == "allow": # allow is meaningful for genuine non-interactive entries (e.g. IM). # Missing resume support or subagent is always a hard fail-closed. if not self.supports_resume or self._is_subagent(runtime): - return self._deny_all(targets, _no_resume_reason()) + return self._deny_all(last, targets, _no_resume_reason()) return None - return self._deny_all(targets, _no_resume_reason()) + return self._deny_all(last, targets, _no_resume_reason()) ai_message_id = last.id or "" + # A pause is only resolvable when every targeted call has a unique, + # non-empty string id: the Gateway resume validator rejects a payload + # with a null or duplicate id with 422, and a pending pause blocks new + # runs with 409, so pausing here would wedge the thread permanently. + # Fail closed instead of interrupting. + if not _unambiguous_ids(last, targets): + return _strip_calls(last, targets, _unaddressable_reason()) actions = [ { "tool_call_id": tc["id"], @@ -177,10 +245,10 @@ class HitlApprovalMiddleware(AgentMiddleware): "pending_tool_call_ids": [tc["id"] for tc in targets], } if len(str(payload)) > MAX_PAYLOAD_BYTES: - return self._deny_all(targets, _too_large_reason()) + return self._deny_all(last, targets, _too_large_reason()) responses = interrupt(payload) if any(tc["name"] in resume_tool_names and tc["name"] not in live_tool_names for tc in targets): - return self._deny_all(targets, _invalid_response_reason()) + return self._deny_all(last, targets, _invalid_response_reason()) return self._process_responses(last, targets, responses) async def aafter_model(self, state: Any, runtime: Any) -> dict[str, Any] | None: @@ -195,24 +263,24 @@ class HitlApprovalMiddleware(AgentMiddleware): resume_value: Any, ) -> dict[str, Any]: target_ids = [tc["id"] for tc in targets] - if any(not isinstance(call_id, str) for call_id in target_ids) or len(set(target_ids)) != len(target_ids): - return self._deny_all(targets, _invalid_response_reason()) + if not _unambiguous_ids(last, targets): + return _strip_calls(last, targets, _invalid_response_reason()) ai_message_id = last.id or "" expected_request_ids = {call_id: f"hitl_approval:{ai_message_id}:{call_id}" for call_id in target_ids} responses = resume_value.get("responses") if isinstance(resume_value, Mapping) else None if not isinstance(responses, list) or len(responses) != len(target_ids): - return self._deny_all(targets, _invalid_response_reason()) + return self._deny_all(last, targets, _invalid_response_reason()) by_id = {tc["id"]: tc for tc in targets} seen_ids: set[str] = set() artificial: list[ToolMessage] = [] for response in responses: if not isinstance(response, Mapping): - return self._deny_all(targets, _invalid_response_reason()) + return self._deny_all(last, targets, _invalid_response_reason()) response_id = response.get("tool_call_id") if not isinstance(response_id, str) or response_id in seen_ids or response_id not in by_id: - return self._deny_all(targets, _invalid_response_reason()) + return self._deny_all(last, targets, _invalid_response_reason()) if response.get("request_id") != expected_request_ids[response_id]: - return self._deny_all(targets, _invalid_response_reason()) + return self._deny_all(last, targets, _invalid_response_reason()) seen_ids.add(response_id) call = by_id[response_id] decision = response.get("option_id") @@ -235,6 +303,10 @@ def _no_resume_reason() -> str: return "Approval is required for this tool, but this run cannot collect approval (non-interactive entry). The tool was not executed." +def _unaddressable_reason() -> str: + return "Approval could not be requested because this tool call has no usable id (missing, empty, or duplicated). The tool was not executed. Do not retry unless the user explicitly asks." + + def _invalid_response_reason() -> str: return "The approval response was invalid, mismatched, or the approval configuration changed while paused. The tool calls were rejected for safety. The tool was not executed. Do not retry unless the user explicitly asks." diff --git a/backend/tests/test_hitl_approval_middleware.py b/backend/tests/test_hitl_approval_middleware.py index 1769eeaa1c..4d8b3e38e8 100644 --- a/backend/tests/test_hitl_approval_middleware.py +++ b/backend/tests/test_hitl_approval_middleware.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json from typing import Any import pytest @@ -433,3 +434,188 @@ def test_summarize_args_preserves_high_semantic_fields(): assert "chars omitted" in summary["content"] assert summary["content"].startswith("xxxx") assert summary["content"].endswith("xxxx") + + +def _visible_text(content): + if isinstance(content, str): + return content + return chr(10).join(block.get("text", "") for block in content if isinstance(block, dict)) + + +def _assert_stripped_without_pausing(out): + assert "__interrupt__" not in out + messages = messages_of(out) + # No result can be paired with an unaddressable call, so the call is removed + # instead of answered: nothing may execute and nothing may pause. + assert not any(isinstance(message, ToolMessage) for message in messages) + assert not any("OK" in _visible_text(message.content) for message in messages) + answers = [message for message in messages if isinstance(message, AIMessage)] + assert all(not message.tool_calls for message in answers) + assert any("no usable id" in _visible_text(message.content) for message in answers) + + +def test_null_tool_call_id_is_denied_instead_of_pausing(): + """A provider can emit a tool call without an id; pausing would wedge the thread. + + The Gateway resume validator rejects a payload whose ``pending_tool_call_ids`` + contains a non-string with 422, while a pending pause blocks new runs with + 409, so an interrupt created here could never be cleared. + """ + calls = [{"name": "write_file", "args": {"path": "/tmp/a.txt", "content": "a"}, "id": None}] + + _, _, out = run_pause(HitlApprovalMiddleware(tool_names={"write_file"}), tool_calls=calls) + + _assert_stripped_without_pausing(out) + + +def test_duplicate_tool_call_ids_are_denied_instead_of_pausing(): + calls = [ + {"name": "write_file", "args": {"path": "/tmp/a.txt", "content": "a"}, "id": "call_dup"}, + {"name": "write_file", "args": {"path": "/tmp/b.txt", "content": "b"}, "id": "call_dup"}, + ] + + _, _, out = run_pause(HitlApprovalMiddleware(tool_names={"write_file"}), tool_calls=calls) + + _assert_stripped_without_pausing(out) + + +def test_empty_tool_call_id_is_denied_instead_of_pausing(): + calls = [{"name": "write_file", "args": {"path": "/tmp/a.txt", "content": "a"}, "id": ""}] + + _, _, out = run_pause(HitlApprovalMiddleware(tool_names={"write_file"}), tool_calls=calls) + + _assert_stripped_without_pausing(out) + + +class _RawPayloadApprovalModel(FakeMessagesListChatModel): + """Emit structured tool calls plus the raw OpenAI-style provider payload.""" + + calls: list[dict[str, Any]] + + def bind_tools(self, tools: Any, **kwargs: Any) -> Any: + return self + + def _generate(self, messages: Any, stop: Any = None, run_manager: Any = None, **kwargs: Any) -> ChatResult: + if any(isinstance(message, ToolMessage) for message in messages): + return ChatResult(generations=[ChatGeneration(message=AIMessage(content="final answer"))]) + raw = [{"id": call["id"], "type": "function", "function": {"name": call["name"], "arguments": json.dumps(call.get("args", {}))}} for call in self.calls] + return ChatResult(generations=[ChatGeneration(message=AIMessage(content="", tool_calls=self.calls, additional_kwargs={"tool_calls": raw}))]) + + +def test_stripping_a_shared_id_clears_the_raw_provider_payload_too(): + """A non-target call sharing the removed id must not keep the removed call serialized. + + ``clone_ai_message_with_tool_calls`` filters ``additional_kwargs["tool_calls"]`` + and content blocks by id, so a surviving ordinary call with the same id would + keep the denied approval calls in the raw provider payload; the next request + would advertise calls that have no result. + """ + executed: list[str] = [] + + @tool + def echo(text: str = "") -> str: + """Echo text.""" + executed.append("echo") + return "echoed" + + calls = [ + {"name": "write_file", "args": {"path": "/tmp/a.txt", "content": "a"}, "id": "call_shared"}, + {"name": "write_file", "args": {"path": "/tmp/b.txt", "content": "b"}, "id": "call_shared"}, + {"name": "echo", "args": {"text": "hi"}, "id": "call_shared"}, + ] + model = _RawPayloadApprovalModel(responses=[], calls=calls) + agent = create_agent( + model, + tools=[write_file, echo], + middleware=[HitlApprovalMiddleware(tool_names={"write_file"})], + checkpointer=InMemorySaver(), + ) + + out = agent.invoke({"messages": [{"role": "user", "content": "go"}]}, {"configurable": {"thread_id": "shared-raw"}}) + + assert "__interrupt__" not in out + messages = messages_of(out) + assert not any(isinstance(message, ToolMessage) for message in messages) + assert executed == [] + answer = messages[-1] + assert answer.tool_calls == [] + assert "tool_calls" not in answer.additional_kwargs + assert "no usable id" in _visible_text(answer.content) + + +def test_shared_id_with_an_ordinary_call_is_denied_instead_of_pausing(): + """An id shared with a non-target call makes the approval ambiguous. + + Approving or rejecting the targeted call would also answer the ordinary one + by id, so the colliding group is rejected before any pause is created. + """ + executed: list[str] = [] + + @tool + def echo(text: str = "") -> str: + """Echo text.""" + executed.append("echo") + return "echoed" + + calls = [ + {"name": "write_file", "args": {"path": "/tmp/a.txt", "content": "a"}, "id": "call_shared"}, + {"name": "echo", "args": {"text": "hi"}, "id": "call_shared"}, + ] + model = _RawPayloadApprovalModel(responses=[], calls=calls) + agent = create_agent( + model, + tools=[write_file, echo], + middleware=[HitlApprovalMiddleware(tool_names={"write_file"})], + checkpointer=InMemorySaver(), + ) + + out = agent.invoke({"messages": [{"role": "user", "content": "go"}]}, {"configurable": {"thread_id": "shared-mixed"}}) + + assert "__interrupt__" not in out + messages = messages_of(out) + assert not any(isinstance(message, ToolMessage) for message in messages) + assert executed == [] + answer = messages[-1] + assert answer.tool_calls == [] + assert "tool_calls" not in answer.additional_kwargs + assert "no usable id" in _visible_text(answer.content) + + +def test_deny_path_strips_a_target_sharing_an_id_with_an_ordinary_call(): + """Denying by result answers a call by id, so a shared id must be stripped. + + ``_deny_all`` routed on ``_addressable`` (uniqueness within the targets), so + a target whose id an ordinary call also carried got a deny ``ToolMessage`` + while the ordinary call stayed serialized. The next request then advertised + the id twice with two results, which strict providers reject. + """ + executed: list[str] = [] + + @tool + def echo(text: str = "") -> str: + """Echo text.""" + executed.append("echo") + return "echoed" + + calls = [ + {"name": "write_file", "args": {"path": "/tmp/a.txt", "content": "a"}, "id": "call_shared"}, + {"name": "echo", "args": {"text": "hi"}, "id": "call_shared"}, + ] + model = _RawPayloadApprovalModel(responses=[], calls=calls) + agent = create_agent( + model, + tools=[write_file, echo], + middleware=[HitlApprovalMiddleware(tool_names={"write_file"}, supports_resume=False)], + checkpointer=InMemorySaver(), + ) + + out = agent.invoke({"messages": [{"role": "user", "content": "go"}]}, {"configurable": {"thread_id": "shared-deny"}}) + + assert "__interrupt__" not in out + messages = messages_of(out) + assert not any(isinstance(message, ToolMessage) for message in messages) + assert executed == [] + answer = messages[-1] + assert answer.tool_calls == [] + assert "tool_calls" not in answer.additional_kwargs + assert "cannot collect approval" in _visible_text(answer.content)