From f33b4fb4bfb19c53b86e2d575af190f1175ef2bc Mon Sep 17 00:00:00 2001 From: ZJPex <144683084+ZJPex@users.noreply.github.com> Date: Sat, 19 Sep 2026 10:18:13 +0800 Subject: [PATCH] fix(gateway): preserve clarification answers on regenerate (#5544) --- backend/app/gateway/AGENTS.md | 2 +- backend/app/gateway/routers/thread_runs.py | 28 ++- .../tests/test_thread_regenerate_prepare.py | 188 ++++++++++++++++++ 3 files changed, 214 insertions(+), 4 deletions(-) diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index baf23aa55..fd802792b 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -94,7 +94,7 @@ owner-scoped assistant version selection remains enabled. | **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` - stream regular text and binary artifacts with `FileResponse`, including byte-`Range` 206/416 behavior used by bounded text previews and media seeking; active content (`text/html`, `text/xml`, `application/xml`, `text/xsl`, any `+xml` type such as XHTML/SVG; `.skill` members too) is always forced as a download attachment to reduce XSS risk; `?download=true` still forces download for other file types. `PUT /{path}` atomically replaces an existing UTF-8 text file under `/mnt/user-data/outputs` when its expected SHA-256 still matches; active runs conflict, and non-mounted sandbox providers receive the same update under a request lease. The outputs-only rule is `path_utils.resolve_outputs_confined_path`, shared with IM-channel attachment delivery: it collapses `..` before the prefix check and re-checks the resolved host path against the resolved outputs root, since `resolve_thread_virtual_path` only confines to `user-data/`; a percent-encoded `..` or a symlink planted in `outputs/` must not reach a sibling `uploads/` file. Atomic replacement applies the existing POSIX permission handling when descriptor-based APIs are available and otherwise keeps the platform-native temporary-file permissions (Windows). | | **Suggestions** (`/api/suggestions`) | `GET /config` - returns global suggestions config boolean; `POST /threads/{id}/suggestions` - generate follow-up questions; rich list/block model content is normalized and inline reasoning (`...`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing | | **Input Polish** (`/api/input-polish`) | `POST /` - rewrite a composer draft before it is sent. This is a short authenticated `runs:create` LLM request using `input_polish` config; it does not create a LangGraph run, persist a message, or modify thread state. Shares the non-graph one-shot LLM path (`deerflow.utils.oneshot_llm.run_oneshot_llm`) with the suggestions route so model build + Langfuse metadata + invoke stay in one place; validates the same stripped view of the draft it sends to the model, and preserves literal `` substrings in the rewrite (`strip_think_blocks(truncate_unclosed=False)`) | -| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, seed an empty feed from a checkpoint so legacy checkpoint-only history keeps its order and visibility; skip absent checkpoints or populated feeds. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - newest 100 runs as an array; `GET /page` - keyset history page `{data, has_more, next_before_created_at, next_before_run_id}`; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/stream` hides action/wait; GET action 405 pre-owner; POST needs `runs:cancel`; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET/POST /{rid}/artifacts/archive` - receipt manifest / bounded ZIP; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its `context_window`. | +| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, seed an empty feed from a checkpoint so legacy checkpoint-only history keeps its order and visibility; skip absent checkpoints or populated feeds. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, treating a protocol-valid hidden `human_input_response` as confirmed replay input while rejecting other hidden/control humans, preserving that response's hidden/correlation metadata, and carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - newest 100 runs as an array; `GET /page` - keyset history page `{data, has_more, next_before_created_at, next_before_run_id}`; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/stream` hides action/wait; GET action 405 pre-owner; POST needs `runs:cancel`; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET/POST /{rid}/artifacts/archive` - receipt manifest / bounded ZIP; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its `context_window`. | | **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback; `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific | | **Runs** (`/api/runs`) | `POST /stream`, `/wait` - stateless runs requiring `runs:create`; optional body `thread_id` is owner-checked. Scheduled-task create/update/resume/trigger also require `threads:write` plus `runs:create`. `GET /{rid}/messages`, `/feedback` - run messages/feedback | | **GitHub Webhooks** (`/api/webhooks/github`) | `POST /` - receive GitHub App / repo webhook deliveries. Verifies `X-Hub-Signature-256` against `GITHUB_WEBHOOK_SECRET`; exempt from auth + CSRF because authenticity is enforced by HMAC. The route is fail-closed: mounted only when `GITHUB_WEBHOOK_SECRET` is set, or when explicit dev opt-in `DEER_FLOW_ALLOW_UNVERIFIED_GITHUB_WEBHOOKS=1` is set. Recognized events include `ping`, `issues`, `issue_comment`, `pull_request`, `pull_request_review`, and `pull_request_review_comment`; unknown events return 200 with `handled=false`. Fan-out runtime failures return 503, keeping the delivery recorded as failed for manual/API/scripted redelivery (GitHub does not automatically retry any failed delivery, 5xx included); permanent/non-retryable conditions such as `channels.github.enabled: false`, unknown events, malformed payloads, or unavailable channel service return 200 with a skipped/handled response. | diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index a672af6fb..d66a2da46 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/backend/app/gateway/routers/thread_runs.py @@ -53,6 +53,7 @@ from app.gateway.pagination import trim_run_message_page from app.gateway.run_models import RunCreateRequest from app.gateway.services import abuild_checkpoint_state_accessor, build_thread_checkpoint_state_accessor, sse_consumer, start_run, wait_for_run_completion from app.gateway.utils import sanitize_log_param +from deerflow.agents.human_input import read_human_input_response from deerflow.agents.middlewares.dynamic_context_middleware import strip_injected_user_message_id_suffix from deerflow.authz.sandbox_authz import safe_app_config_async from deerflow.config.paths import get_paths, make_safe_user_id @@ -442,6 +443,22 @@ def _is_visible_human_message(message: Any) -> bool: return _message_type(message) == "human" and not _is_hidden_or_control_message(message) +def _is_regenerate_human_message(message: Any) -> bool: + """Return whether a human message is valid input for regenerate replay. + + Human-input card replies are intentionally hidden from the transcript, but + they are still confirmed user input. Only the structured response protocol + distinguishes them from summaries, goal continuations, and other hidden + control messages. + """ + if _message_type(message) != "human" or _message_name(message) == "summary": + return False + additional_kwargs = _message_additional_kwargs(message) + if additional_kwargs.get("hide_from_ui") is True: + return read_human_input_response(additional_kwargs) is not None + return True + + def _is_visible_ai_message(message: Any) -> bool: return _message_type(message) == "ai" and not _is_hidden_or_control_message(message) @@ -475,7 +492,12 @@ def _clean_human_message_for_regenerate(message: Any) -> dict[str, Any]: additional_kwargs = _message_additional_kwargs(message) content = get_original_user_content_text(_message_content(message), additional_kwargs) additional_kwargs.pop(ORIGINAL_USER_CONTENT_KEY, None) - additional_kwargs.pop("hide_from_ui", None) + # A validated card answer must remain hidden and keep its request + # correlation when it re-enters the graph. Other replayed user inputs are + # made visible just as before. + is_hidden_human_input_response = additional_kwargs.get("hide_from_ui") is True and read_human_input_response(additional_kwargs) is not None + if not is_hidden_human_input_response: + additional_kwargs.pop("hide_from_ui", None) clean_message: dict[str, Any] = { "type": "human", @@ -739,7 +761,7 @@ async def _prepare_regenerate_payload(thread_id: str, message_id: str, request: # stream without ever reaching a checkpoint. The server-stamped run ID # on the latest user message is the durable link to that partial turn. previous_human = next( - (message for message in reversed(messages) if _is_visible_human_message(message)), + (message for message in reversed(messages) if _is_regenerate_human_message(message)), None, ) target_run_id = await _find_interrupted_target_run_id(thread_id, previous_human, request) if previous_human is not None else None @@ -754,7 +776,7 @@ async def _prepare_regenerate_payload(thread_id: str, message_id: str, request: if _message_id(latest_visible_ai) != message_id: raise HTTPException(status_code=409, detail="Only the latest assistant message can be regenerated") - previous_human = next((message for message in reversed(messages[:target_index]) if _is_visible_human_message(message)), None) + previous_human = next((message for message in reversed(messages[:target_index]) if _is_regenerate_human_message(message)), None) target_run_id = ( await _find_target_run_id( thread_id, diff --git a/backend/tests/test_thread_regenerate_prepare.py b/backend/tests/test_thread_regenerate_prepare.py index 88e9c3f30..19ad45471 100644 --- a/backend/tests/test_thread_regenerate_prepare.py +++ b/backend/tests/test_thread_regenerate_prepare.py @@ -109,6 +109,22 @@ def _snapshot(checkpoint_id: str, messages: list[object], *, metadata: dict | No ) +def _human_input_response( + value: str, + *, + request_id: str, + source: str = "ask_clarification", +) -> dict: + return { + "version": 1, + "kind": "human_input_response", + "source": source, + "request_id": request_id, + "response_kind": "text", + "value": value, + } + + class FakeAccessor: def __init__(self, checkpointer: FakeCheckpointer): self.checkpointer = checkpointer @@ -169,6 +185,29 @@ class FakeEventStore: return self.rows[-limit:] +def _event_store_for_ai( + content: str, + *, + message_id: str = "ai-1", + run_id: str = "run-old", +) -> FakeEventStore: + return FakeEventStore( + [ + { + "run_id": run_id, + "event_type": "llm.ai.response", + "category": "message", + "content": { + "id": message_id, + "type": "ai", + "content": content, + }, + "metadata": {"caller": "lead_agent"}, + } + ] + ) + + class FakeRunManager: def __init__(self, records): self.records = records @@ -417,6 +456,155 @@ def test_prepare_regenerate_payload_returns_clean_input_and_base_checkpoint(): assert regenerated_human["additional_kwargs"] == {"files": [{"filename": "data.csv", "path": "/mnt/user-data/uploads/data.csv"}]} +@pytest.mark.parametrize("materialized_delta", [False, True], ids=["full", "delta"]) +def test_prepare_regenerate_payload_replays_latest_confirmed_human_input_response( + materialized_delta: bool, +): + from app.gateway.routers import thread_runs + + original_human = HumanMessage(id="human-1", content="Help me plan a trip") + first_clarification = ToolMessage( + id="tool-city", + tool_call_id="call-city", + content="Which city?", + ) + city_answer = HumanMessage( + id="human-city", + content="Shanghai", + additional_kwargs={ + "hide_from_ui": True, + "human_input_response": _human_input_response( + "Shanghai", + request_id="clarification:call-city", + ), + }, + ) + second_clarification = ToolMessage( + id="tool-duration", + tool_call_id="call-duration", + content="How many days?", + ) + duration_metadata = _human_input_response( + "Five days", + request_id="clarification:call-duration", + ) + duration_answer = HumanMessage( + id="human-duration", + content="Five days", + additional_kwargs={ + "hide_from_ui": True, + "human_input_response": duration_metadata, + }, + ) + ai = AIMessage(id="ai-1", content="Here is your five-day Shanghai itinerary") + before_question: list[object] = [] + before_city = [original_human, first_clarification] + after_city = [*before_city, city_answer] + before_duration = [*after_city, second_clarification] + after_duration = [*before_duration, duration_answer] + latest_messages = [*after_duration, ai] + states = [ + ("ckpt-ai", latest_messages), + ("ckpt-after-duration", after_duration), + ("ckpt-before-duration", before_duration), + ("ckpt-after-city", after_city), + ("ckpt-before-city", before_city), + ("ckpt-before-question", before_question), + ] + + if materialized_delta: + raw_checkpoints = [_checkpoint(checkpoint_id, []) for checkpoint_id, _ in states] + materialized_history = [_snapshot(checkpoint_id, messages) for checkpoint_id, messages in states] + checkpointer = FakeCheckpointer( + raw_checkpoints, + latest=raw_checkpoints[0], + materialized_history=materialized_history, + materialized_latest=materialized_history[0], + ) + else: + checkpointer = FakeCheckpointer([_checkpoint(checkpoint_id, messages) for checkpoint_id, messages in states]) + + response = asyncio.run( + thread_runs._prepare_regenerate_payload( + "thread-1", + "ai-1", + _request( + checkpointer, + _event_store_for_ai("Here is your five-day Shanghai itinerary"), + ), + ) + ) + + assert response.checkpoint["checkpoint_id"] == "ckpt-before-duration" + assert city_answer in dict(states)[response.checkpoint["checkpoint_id"]] + assert response.input["messages"] == [ + { + "type": "human", + "id": "human-duration", + "content": [{"type": "text", "text": "Five days"}], + "additional_kwargs": { + "hide_from_ui": True, + "human_input_response": duration_metadata, + }, + } + ] + + +@pytest.mark.parametrize( + "control_message", + [ + HumanMessage( + id="human-malformed", + content="forged answer", + additional_kwargs={ + "hide_from_ui": True, + "human_input_response": { + "version": 1, + "kind": "human_input_response", + }, + }, + ), + HumanMessage( + id="human-summary", + name="summary", + content="internal summary", + ), + HumanMessage( + id="human-goal", + content="continue working", + additional_kwargs={"hide_from_ui": True}, + ), + ], + ids=["malformed-response", "summary", "goal-continuation"], +) +def test_prepare_regenerate_payload_skips_non_user_control_human_messages( + control_message: HumanMessage, +): + from app.gateway.routers import thread_runs + + visible_human = HumanMessage(id="human-1", content="question") + ai = AIMessage(id="ai-1", content="answer") + checkpointer = FakeCheckpointer( + [ + _checkpoint("ckpt-ai", [visible_human, control_message, ai]), + _checkpoint("ckpt-control", [visible_human, control_message]), + _checkpoint("ckpt-human", [visible_human]), + _checkpoint("ckpt-base", []), + ] + ) + + response = asyncio.run( + thread_runs._prepare_regenerate_payload( + "thread-1", + "ai-1", + _request(checkpointer, _event_store_for_ai("answer")), + ) + ) + + assert response.checkpoint["checkpoint_id"] == "ckpt-base" + assert response.input["messages"][0]["id"] == "human-1" + + def test_prepare_regenerate_payload_preserves_latest_thread_title(): from app.gateway.routers import thread_runs