mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-11-13 06:31:47 +00:00
3d826c6d95
* feat(hitl): add human-in-the-loop approval for high-risk tool calls * fix(hitl): address approval review findings * docs(hitl): keep agent guidance under size limit * fix(hitl): guard approval pause recovery * docs(hitl): shrink inherited guidance * fix(hitl): harden approval admission and recovery * fix(approvals): address review safety and reconciliation gaps * ci: retry approval checks after runner timeout --------- Co-authored-by: wangzeren <1004695029@qq.com>
94 lines
4.6 KiB
Python
94 lines
4.6 KiB
Python
"""Unaddressable calls must never become approval pauses or tool results."""
|
|
|
|
import json
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from _agent_e2e_helpers import FakeToolCallingModel
|
|
from langchain.agents import create_agent
|
|
from langchain_core.messages import AIMessage, ToolMessage
|
|
from langchain_core.tools import tool
|
|
from langgraph.checkpoint.memory import InMemorySaver
|
|
|
|
from deerflow.approvals.middleware import ToolApprovalMiddleware
|
|
from deerflow.config.app_config import AppConfig
|
|
from deerflow.config.hitl_approval_config import HitlApprovalConfig
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"])
|
|
@pytest.mark.parametrize("mode", ["ask", "auto", "full"])
|
|
@pytest.mark.parametrize("unattended", [False, True], ids=["interactive", "unattended"])
|
|
@pytest.mark.parametrize("call_ids", [[None], [""], ["shared", "shared"], ["shared", "shared", "shared"]], ids=["null", "empty", "duplicate", "mixed-collision"])
|
|
async def test_unusable_call_ids_are_removed_before_review_or_execution(asynchronous, mode, unattended, call_ids):
|
|
executed = []
|
|
|
|
@tool
|
|
def external_write(value: str) -> str:
|
|
"""Record a write."""
|
|
executed.append(value)
|
|
return "written"
|
|
|
|
@tool
|
|
def ordinary_read() -> str:
|
|
"""Record an ordinary read."""
|
|
executed.append("read")
|
|
return "read"
|
|
|
|
calls = [{"name": "external_write", "args": {"value": str(i)}, "id": call_id} for i, call_id in enumerate(call_ids)]
|
|
if len(calls) == 3:
|
|
calls[-1].update(name="ordinary_read", args={})
|
|
raw = [{"id": call["id"], "type": "function", "function": {"name": call["name"], "arguments": json.dumps(call["args"])}} for call in calls]
|
|
content = [{"type": "text", "text": "Original explanation"}]
|
|
# Exercise every provider surface the shared cloning helper must trim.
|
|
for call in calls:
|
|
content.extend(
|
|
[
|
|
{"type": "tool_use", "id": call["id"], "name": call["name"], "input": call["args"]},
|
|
{"type": "function_call", "id": "fc_item", "call_id": call["id"], "name": call["name"], "arguments": json.dumps(call["args"])},
|
|
{"type": "custom_tool_call", "call_id": call["id"], "name": call["name"], "input": "value"},
|
|
{"type": "tool_call", **call},
|
|
{"type": "tool_call_chunk", **call},
|
|
{"type": "function_call", "name": call["name"], "args": call["args"]},
|
|
]
|
|
)
|
|
initial = AIMessage(content=content, tool_calls=calls, additional_kwargs={"tool_calls": raw, "function_call": raw[0]["function"]}, response_metadata={"finish_reason": "tool_calls"})
|
|
reviewer = AsyncMock(return_value={"outcome": "allow", "risk": "low", "authorization": "explicit", "reason": "authorized"})
|
|
policy = HitlApprovalConfig(enabled=True, tools=["external_write"], allow_tools=["ordinary_read"], default_mode=mode, allowed_modes=["ask", "auto", "full"])
|
|
app = AppConfig(sandbox={"use": "deerflow.sandbox.local:LocalSandboxProvider"}, hitl_approval=policy)
|
|
graph = create_agent(
|
|
FakeToolCallingModel(responses=[initial, AIMessage(content="Next turn works")]),
|
|
tools=[external_write, ordinary_read],
|
|
middleware=[ToolApprovalMiddleware(policy, app_config=app, reviewer=reviewer)],
|
|
checkpointer=InMemorySaver(),
|
|
)
|
|
config = {"configurable": {"thread_id": "unusable-calls"}}
|
|
context = {"non_interactive": unattended}
|
|
|
|
async def invoke(input):
|
|
if asynchronous:
|
|
return await graph.ainvoke(input, config, context=context)
|
|
return graph.invoke(input, config, context=context)
|
|
|
|
result = await invoke({"messages": [("user", "Perform the operation")]})
|
|
|
|
assert not result.get("__interrupt__")
|
|
assert executed == []
|
|
reviewer.assert_not_called()
|
|
assert not any(isinstance(message, ToolMessage) for message in result["messages"])
|
|
answer = result["messages"][-1]
|
|
assert answer.tool_calls == []
|
|
assert "tool_calls" not in answer.additional_kwargs
|
|
assert "function_call" not in answer.additional_kwargs
|
|
assert answer.response_metadata["finish_reason"] == "stop"
|
|
assert all(block["type"] == "text" for block in answer.content)
|
|
visible_text = "\n".join(block["text"] for block in answer.content)
|
|
assert "Original explanation" in visible_text
|
|
assert "no usable id" in visible_text
|
|
|
|
# A real second invocation on the same checkpoint must remain usable.
|
|
following = await invoke({"messages": [("user", "Continue with the explanation")]})
|
|
assert not following.get("__interrupt__")
|
|
assert following["messages"][-1].content == "Next turn works"
|
|
assert executed == []
|