From 226d1bc5ce2447b423e93e4bf67f5dc5186b27c7 Mon Sep 17 00:00:00 2001 From: Evan <1178327328@qq.com> Date: Tue, 10 Nov 2026 02:11:09 -0500 Subject: [PATCH] fix(runtime): persist a rejected HITL call in the resume run's history (#8073) * fix(runtime): persist a rejected HITL call in the resume run's history An approval pause and its resolution are two Runs. The paused Run dispatches the gated call; the resume Run produces its result. An approved call executes and reaches on_tool_end, but a rejection is written by HITL after_model with no tool callback. Journal reconciliation only persists such middleware-written results for calls dispatched by the current Run, so the rejection stayed in checkpoint state and never reached the run-event history: /messages and the resume run's messages had no llm.tool.result for it. The worker already reads the durable interrupt payload to recover the gated tool names. It now also returns each paused tool_call_id with its name, and the resume journal adopts those calls, so the existing reconciliation persists the rejection under the resume Run. Approved results keep their tool-callback event and are deduplicated by message id; results from other earlier Runs are still not re-attributed. * fix(runtime): keep HITL replay names independent of call adoption Derive the replay-gate tool names and the adopted call ids separately from the matched pause's actions, so an action without a usable tool_call_id still gates replay. Pin that the all-reject escape paths (approval disabled, gated tools changed or cleared while paused) reach the history feed under the resume run, check resume-run attribution in the mixed-decision test, and add a CHANGELOG entry. * refactor(runtime): drop unused _pending_hitl_resume_calls wrapper --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 5 + backend/docs/RUN_INTERACTION_POLICY.md | 6 + .../harness/deerflow/runtime/journal.py | 12 ++ .../harness/deerflow/runtime/runs/worker.py | 42 +++++-- backend/tests/test_hitl_approval_api.py | 20 +++- backend/tests/test_hitl_resolution_history.py | 105 ++++++++++++++++++ backend/tests/test_run_journal.py | 23 ++++ 7 files changed, 203 insertions(+), 10 deletions(-) create mode 100644 backend/tests/test_hitl_resolution_history.py 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