diff --git a/CHANGELOG.md b/CHANGELOG.md index f1e7715009..48a4b57c5a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,11 @@ This section accumulates work toward the **2.3.0** milestone ### Fixed +- **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 + state. ([#8073](https://github.com/bytedance/deer-flow/pull/8073)) + - **summarization:** vLLM compaction counts reasoning tokens replayed to the model when checking the trigger and estimating how much history to keep. ([#8077](https://github.com/bytedance/deer-flow/pull/8077)) diff --git a/backend/docs/RUN_INTERACTION_POLICY.md b/backend/docs/RUN_INTERACTION_POLICY.md index b3f4bedf7d..b0f994fdf3 100644 --- a/backend/docs/RUN_INTERACTION_POLICY.md +++ b/backend/docs/RUN_INTERACTION_POLICY.md @@ -53,6 +53,12 @@ AI message before applying any decision. If worker attachment fails after consumption, the Gateway conditionally restores that exact answered run to `waiting_for_input` so a valid response or all-reject can retry safely. +The resume run's journal adopts the paused calls it answers, recovered from the +durable interrupt payload. Approved calls reach the history feed through their +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. + `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/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index 6be3911822..8095d0ebb1 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/backend/packages/harness/deerflow/runtime/journal.py @@ -903,6 +903,18 @@ class RunJournal(BaseCallbackHandler): self._current_tool_result_ids.pop(tool_call_id, None) self._terminal_task_results.discard(tool_call_id) + def adopt_resumed_tool_calls(self, calls: Mapping[str, str]) -> None: + """Own lead-agent calls an earlier Run paused and this Run resolves. + + A HITL resume answers calls the paused Run dispatched. An approved call + executes here and reaches ``on_tool_end``, but a rejection is written by + the middleware without any tool callback, so final-output + reconciliation must treat these calls as this Run's to persist it. + """ + for tool_call_id, name in calls.items(): + if isinstance(tool_call_id, str) and tool_call_id: + self._current_run_tool_call_names[tool_call_id] = str(name or "") + def _persist_tool_result_message(self, message: BaseMessage) -> None: # Tool callbacks run before the graph reducer assigns missing ids. # Assign on the returned object so the feed and checkpoint share the diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index 03cd8a0c09..1d5eb1feef 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -1355,7 +1355,11 @@ async def run_agent( hitl_app_config = get_app_config() hitl_config = getattr(hitl_app_config, "hitl_approval", None) hitl_resume_escape = hitl_config is not None and not hitl_config.enabled and all_reject_hitl_resume - hitl_resume_tool_names = await _pending_hitl_resume_tool_names(checkpointer, checkpoint_config, graph_input) + hitl_resume_actions = await _pending_hitl_resume_actions(checkpointer, checkpoint_config, graph_input) + hitl_resume_tool_names = _hitl_action_tool_names(hitl_resume_actions) + hitl_resume_calls = _hitl_action_calls(hitl_resume_actions) + if journal is not None and hitl_resume_calls: + journal.adopt_resumed_tool_calls(hitl_resume_calls) agent_app_config = ctx.app_config if hitl_resume_escape: agent_app_config = hitl_app_config.model_copy(deep=True) @@ -3183,25 +3187,45 @@ def _is_resume_command_input(graph_input: Any) -> bool: async def _pending_hitl_resume_tool_names(checkpointer: Any, checkpoint_config: dict[str, Any], graph_input: Any) -> set[str]: """Recover gated names from the durable pause matched by this resume.""" + return _hitl_action_tool_names(await _pending_hitl_resume_actions(checkpointer, checkpoint_config, graph_input)) + + +def _hitl_action_tool_names(actions: list[Mapping[str, Any]]) -> set[str]: + """Names for the replay gate: every action naming a tool, with or without a usable call id.""" + return {action["tool_name"] for action in actions if isinstance(action.get("tool_name"), str)} + + +def _hitl_action_calls(actions: list[Mapping[str, Any]]) -> dict[str, str]: + """Calls for journal adoption: only actions that also carry a usable ``tool_call_id``. + + The calls were dispatched by the paused Run, but the resume Run produces + their results: a rejection is written by HITL ``after_model`` with no tool + callback, so the resume journal must adopt these calls to persist it. + """ + return {action["tool_call_id"]: action["tool_name"] for action in actions if isinstance(action.get("tool_call_id"), str) and action["tool_call_id"] and isinstance(action.get("tool_name"), str)} + + +async def _pending_hitl_resume_actions(checkpointer: Any, checkpoint_config: dict[str, Any], graph_input: Any) -> list[Mapping[str, Any]]: + """Return the actions of the live durable pause this resume answers.""" if checkpointer is None or not _is_resume_command_input(graph_input) or not isinstance(graph_input.resume, Mapping): - return set() + return [] responses = graph_input.resume.get("responses") if not isinstance(responses, list): - return set() + return [] response_ids = {response.get("tool_call_id") for response in responses if isinstance(response, Mapping) and isinstance(response.get("tool_call_id"), str)} if not response_ids: - return set() + return [] async for checkpoint_tuple in recent_checkpoint_history(checkpointer, checkpoint_config): writes = list(getattr(checkpoint_tuple, "pending_writes", ()) or ()) # The newest checkpoint carrying either marker decides whether the pause # is still live. A newer head-moving write must not hide an interrupt; # a newer resume must not resurrect an older one. if any(isinstance(write, (list, tuple)) and len(write) >= 2 and write[1] == "__resume__" for write in writes): - return set() + return [] interrupt_writes = [write for write in writes if isinstance(write, (list, tuple)) and len(write) >= 3 and write[1] == "__interrupt__"] if not interrupt_writes: continue - names: set[str] = set() + matched: list[Mapping[str, Any]] = [] for write in interrupt_writes: interrupts = write[2] if isinstance(write[2], (list, tuple)) else (write[2],) for pending_interrupt in interrupts: @@ -3212,9 +3236,9 @@ async def _pending_hitl_resume_tool_names(checkpointer: Any, checkpoint_config: actions = payload.get("actions") if not isinstance(pending_ids, list) or set(pending_ids) != response_ids or not isinstance(actions, list): continue - names.update(action.get("tool_name") for action in actions if isinstance(action, Mapping) and isinstance(action.get("tool_name"), str)) - return names - return set() + matched.extend(action for action in actions if isinstance(action, Mapping)) + return matched + return [] def _is_all_reject_hitl_resume(graph_input: Any) -> bool: diff --git a/backend/tests/test_hitl_approval_api.py b/backend/tests/test_hitl_approval_api.py index 5f5393d75a..1f37f5971d 100644 --- a/backend/tests/test_hitl_approval_api.py +++ b/backend/tests/test_hitl_approval_api.py @@ -189,6 +189,17 @@ def _resume(client, csrf, tid, option_id, value=None, *, tool_call_id="call_hitl return _run(client, csrf, tid, {"command": {"resume": {"responses": [response]}}}) +def _feed_tool_result_ids(client, tid): + rows = client.get(f"/api/threads/{tid}/messages").json() + return [row["content"]["id"] for row in rows if row.get("event_type") == "llm.tool.result"] + + +def _run_tool_result_call_ids(client, tid, run_id): + rows = client.get(f"/api/threads/{tid}/runs/{run_id}/messages").json() + rows = rows["data"] if isinstance(rows, dict) else rows + return [row["content"]["tool_call_id"] for row in rows if row.get("event_type") == "llm.tool.result"] + + def test_approval_pause_then_approve_executes(isolated_app): _enable_approval() with patch(_MODEL_PATCH, new=_fake_model), TestClient(isolated_app) as client: @@ -428,9 +439,13 @@ def test_disabled_resume_409_and_all_reject_escape(isolated_app): terminal = _wait_run_terminal(client, tid, escape.json()["run_id"]) assert terminal["status"] == "success" assert _wait_for_status(client, tid, "idle")["status"] == "idle" - contents = [m.get("content") for m in client.get(f"/api/threads/{tid}").json()["values"]["messages"] if m.get("type") == "tool"] + state_tools = [m for m in client.get(f"/api/threads/{tid}").json()["values"]["messages"] if m.get("type") == "tool"] + contents = [m.get("content") for m in state_tools] assert "OK" not in contents assert len(contents) == 2 + # The escape run adopts the paused call, so its denial reaches the history feed. + assert _feed_tool_result_ids(client, tid) == [m["id"] for m in state_tools] + assert "call_hitl_1" in _run_tool_result_call_ids(client, tid, escape.json()["run_id"]) @pytest.mark.parametrize("current_tools", [["bash"], []], ids=["changed", "cleared"]) @@ -456,6 +471,7 @@ def test_changed_tools_all_reject_escape_preserves_later_gate(isolated_app, curr assert tool_messages[0]["tool_call_id"] == "call_hitl_1" assert tool_messages[0]["status"] == "error" assert "OK" not in [message.get("content") for message in tool_messages] + assert _run_tool_result_call_ids(client, tid, second_run_id) == ["call_hitl_1"] second_escape = _resume(client, csrf, tid, "reject", tool_call_id="call_hitl_2") assert second_escape.status_code == 200, second_escape.text @@ -466,6 +482,8 @@ def test_changed_tools_all_reject_escape_preserves_later_gate(isolated_app, curr assert len(tool_messages) == 2 assert all(message.get("status") == "error" for message in tool_messages) assert "OK" not in [message.get("content") for message in tool_messages] + assert _feed_tool_result_ids(client, tid) == [message["id"] for message in tool_messages] + assert _run_tool_result_call_ids(client, tid, second_escape.json()["run_id"]) == ["call_hitl_2"] def test_stale_all_reject_cannot_dismiss_newer_pause(isolated_app): diff --git a/backend/tests/test_hitl_resolution_history.py b/backend/tests/test_hitl_resolution_history.py new file mode 100644 index 0000000000..216eaf00ee --- /dev/null +++ b/backend/tests/test_hitl_resolution_history.py @@ -0,0 +1,105 @@ +"""A resumed HITL decision reaches the persisted history feed. + +The approval pause and its resolution run as two Runs. A rejected call never +executes, so HITL answers it in ``after_model`` of the resume run and no +``on_tool_end`` fires. The resume run's journal must still own that call, or the +rejection exists only in checkpoint state and disappears from +``/messages`` (and from the reloaded conversation once compaction drops it). +""" + +from __future__ import annotations + +from unittest.mock import patch + +import pytest +from starlette.testclient import TestClient +from test_hitl_approval_api import ( # noqa: F401 + _MODEL_PATCH, + _approval_request_id, + _create_thread, + _enable_approval, + _fake_model, + _fake_two_target_model, + _register_user, + _resume, + _run, + _run_body, + _run_tool_result_call_ids, + _wait_for_approval, + _wait_run_status, + _wait_run_terminal, + isolated_app, + isolated_deer_flow_home, +) + +from deerflow.runtime.runs.worker import _hitl_action_calls, _hitl_action_tool_names + +pytestmark = pytest.mark.no_auto_user + + +def _feed_tool_results(client, tid): + rows = client.get(f"/api/threads/{tid}/messages").json() + return [row["content"] for row in rows if row.get("event_type") == "llm.tool.result"] + + +def _state_tool_messages(client, tid): + state = client.get(f"/api/threads/{tid}").json() + return [message for message in state["values"]["messages"] if message.get("type") == "tool"] + + +@pytest.mark.parametrize("decision", ["reject", "approve"]) +def test_resolved_call_reaches_the_history_feed_once(isolated_app, decision): # noqa: F811 + _enable_approval() + with patch(_MODEL_PATCH, new=_fake_model), TestClient(isolated_app) as client: + csrf = _register_user(client) + tid = _create_thread(client, csrf) + _run(client, csrf, tid, _run_body(input={"messages": [{"role": "user", "content": "write"}]})) + _wait_for_approval(client, tid) + resumed = _resume(client, csrf, tid, decision, value="do not write" if decision == "reject" else None) + assert resumed.status_code == 200, resumed.text + resume_run_id = resumed.json()["run_id"] + _wait_run_terminal(client, tid, resume_run_id) + + state_results = _state_tool_messages(client, tid) + feed_results = _feed_tool_results(client, tid) + assert [message["id"] for message in feed_results] == [message["id"] for message in state_results] + assert [message["tool_call_id"] for message in feed_results] == ["call_hitl_1"] + if decision == "reject": + assert "do not write" in feed_results[0]["content"] + run_rows = client.get(f"/api/threads/{tid}/runs/{resume_run_id}/messages").json() + run_rows = run_rows["data"] if isinstance(run_rows, dict) else run_rows + assert any(row.get("event_type") == "llm.tool.result" for row in run_rows) + + +def test_mixed_decision_keeps_both_results_in_the_history_feed(isolated_app): # noqa: F811 + _enable_approval() + with patch(_MODEL_PATCH, new=_fake_two_target_model), TestClient(isolated_app) as client: + csrf = _register_user(client) + tid = _create_thread(client, csrf) + rid = _run(client, csrf, tid, _run_body(input={"messages": [{"role": "user", "content": "write twice"}]})).json()["run_id"] + _wait_run_status(client, tid, rid, "waiting_for_input") + responses = [ + {"option_id": "approve", "tool_call_id": "call_hitl_1", "request_id": _approval_request_id(client, tid, "call_hitl_1")}, + {"option_id": "reject", "tool_call_id": "call_hitl_2", "request_id": _approval_request_id(client, tid, "call_hitl_2"), "value": "skip the second"}, + ] + resumed = _run(client, csrf, tid, {"command": {"resume": {"responses": responses}}}) + assert resumed.status_code == 200, resumed.text + resume_run_id = resumed.json()["run_id"] + _wait_run_terminal(client, tid, resume_run_id) + + state_ids = sorted(message["id"] for message in _state_tool_messages(client, tid)) + feed_results = _feed_tool_results(client, tid) + assert sorted(message["id"] for message in feed_results) == state_ids + assert sorted(message["tool_call_id"] for message in feed_results) == ["call_hitl_1", "call_hitl_2"] + assert sorted(_run_tool_result_call_ids(client, tid, resume_run_id)) == ["call_hitl_1", "call_hitl_2"] + + +def test_replay_gate_names_do_not_depend_on_an_adoptable_call_id(): + actions = [ + {"tool_call_id": "call_1", "tool_name": "write_file"}, + {"tool_name": "bash"}, + {"tool_call_id": "", "tool_name": "str_replace"}, + ] + + assert _hitl_action_tool_names(actions) == {"write_file", "bash", "str_replace"} + assert _hitl_action_calls(actions) == {"call_1": "write_file"} diff --git a/backend/tests/test_run_journal.py b/backend/tests/test_run_journal.py index 0efbbb1e46..85ee2279ea 100644 --- a/backend/tests/test_run_journal.py +++ b/backend/tests/test_run_journal.py @@ -1657,6 +1657,29 @@ class TestFinalToolMessageReconciliation: assert len(tool_results) == 1 assert tool_results[0]["content"]["name"] == "write_file" + @pytest.mark.anyio + async def test_root_chain_end_reconciles_a_rejection_for_a_call_adopted_from_the_paused_run(self, journal_setup): + """A HITL resume answers calls an earlier Run dispatched. + + The rejection is written in ``after_model`` with no tool callback, so the + resume journal only persists it for the calls the paused Run handed over; + an earlier result outside that handover stays with its own Run. + """ + from langchain_core.messages import AIMessage, ToolMessage + + j, store = journal_setup + earlier_ai = AIMessage(content="", tool_calls=[{"id": "call_old", "name": "bash", "args": {}}]) + earlier = ToolMessage(content="ran earlier", tool_call_id="call_old", name="bash") + paused_ai = AIMessage(content="", tool_calls=[{"id": "call_write", "name": "write_file", "args": {}}]) + rejected = ToolMessage(content="The tool was not executed.", tool_call_id="call_write", name="write_file", status="error") + + j.adopt_resumed_tool_calls({"call_write": "write_file"}) + j.on_chain_end({"messages": [earlier_ai, earlier, paused_ai, rejected]}, run_id=uuid4()) + await j.flush() + + tool_results = [m for m in await store.list_messages("t1") if m["event_type"] == "llm.tool.result"] + assert [result["content"]["tool_call_id"] for result in tool_results] == ["call_write"] + @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