mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-29 01:15:59 +00:00
fix(runtime): regenerate interrupted responses (#4524)
This commit is contained in:
Aari
GitHub
parent
9a43d8276d
commit
e47bf80122
@ -794,7 +794,7 @@ Interrupted first-turn runs still persist a fallback conversation title, so stop
|
||||
|
||||
Streaming Markdown responses animate only newly arrived words; text that is already visible is not faded out and replayed when the next chunk extends the same block.
|
||||
|
||||
In the Web UI, completed assistant turns can be branched into a new main conversation. The new thread starts from that turn's checkpoint and keeps the preceding replay checkpoint, so the branched response can be regenerated immediately. Regenerating the latest response preserves the thread's current title, including a title you renamed manually after the original response. Legacy or imported histories without checkpoint parent links use a bounded chronological fallback; if no earlier replay checkpoint exists, branching still succeeds with the legacy single-checkpoint shape, while regeneration remains unavailable for that inherited response. Existing single-checkpoint branches are left unchanged rather than attempting an unsafe checkpoint copy. Because workspace files are not checkpointed, the branch only receives a best-effort copy of the current workspace when you branch from the latest turn; branching from an older turn keeps just the restored message history so the branch never inherits files that were created in a later part of the conversation.
|
||||
In the Web UI, completed assistant turns can be branched into a new main conversation. The new thread starts from that turn's checkpoint and keeps the preceding replay checkpoint, so the branched response can be regenerated immediately. The latest response can also be regenerated after an interruption, even when its streamed partial text never reached a checkpoint. Regenerating the latest response preserves the thread's current title, including a title you renamed manually after the original response. Legacy or imported histories without checkpoint parent links use a bounded chronological fallback; if no earlier replay checkpoint exists, branching still succeeds with the legacy single-checkpoint shape, while regeneration remains unavailable for that inherited response. Existing single-checkpoint branches are left unchanged rather than attempting an unsafe checkpoint copy. Because workspace files are not checkpointed, the branch only receives a best-effort copy of the current workspace when you branch from the latest turn; branching from an older turn keeps just the restored message history so the branch never inherits files that were created in a later part of the conversation.
|
||||
|
||||
The Web UI reports completed task time once per run. This is total wall-clock time—including model reasoning, tool calls, and waiting—not a per-step or model-only thinking duration. Reasoning content remains available through its own separate disclosure.
|
||||
|
||||
|
||||
@ -441,7 +441,7 @@ Localhost persistence deliberately reads the direct request `Host` and ignores `
|
||||
| **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` - serve artifacts; active content types (`text/html`, `application/xhtml+xml`, `image/svg+xml`) are always forced as download attachments to reduce XSS risk; `?download=true` still forces download for other file types |
|
||||
| **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 (`<think>...</think>`, 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 `<think>` 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; `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest 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 /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `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 /../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 |
|
||||
| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block; `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 /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `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 /../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 |
|
||||
| **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` - stateless run + SSE; `POST /wait` - stateless run + block; `GET /{rid}/messages` - paginated messages by run_id `{data, has_more}` (cursor: `after_seq`/`before_seq`); `GET /{rid}/feedback` - list feedback by run_id |
|
||||
| **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. |
|
||||
@ -949,7 +949,7 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c
|
||||
|
||||
**Never bypass `CheckpointStateAccessor` (`runtime/checkpoint_state.py`) for thread-state access.** It is the single choke point binding graph + checkpointer + mode: it injects the mode marker into configs, runs the compatibility check before every `get`/`update`/`history`, and returns materialized state (delta checkpoints lack `channel_values.messages` — raw `get_tuple` reads see a sentinel). Gateway `services.py` builds and passes the accessor; thread-owned reads (state/history/regeneration) must use `build_thread_checkpoint_state_accessor` so the recorded assistant's middleware schema materializes every channel. `history(limit)` semantics: `0` means zero items (explicit empty), `None` means unlimited — do not pass `limit=0` through to `graph.get_state_history`. Assistant metadata lookup is fail-closed for mutation accessors so a store outage cannot silently select the default schema and discard extension channels. In `full` mode the read path degrades to a raw checkpointer read (`_RawCheckpointReadAccessor`) when the agent factory cannot build the graph (bad model config, MCP outage) — full checkpoints carry complete `channel_values`, so reads don't need the graph; degraded snapshots take `created_at` from the standard checkpoint `ts` field, falling back to metadata only for compatibility. The delta gate still applies on the degraded path; `next`/`tasks` degrade to empty and thread status falls back to the stored status because task presence is not derivable, while delta mode has no fallback (materialization needs the channel table).
|
||||
|
||||
**Replay checkpoint lookup prefers lineage and degrades only for an explicitly missing legacy parent link.** Branch and regenerate paths first walk `parent_config`, which prevents a global chronological scan from selecting a sibling created by regeneration. `CheckpointParentMissingError` alone enables the bounded newest-first history fallback in `app/gateway/checkpoint_lineage.py`; cycles, dangling/non-addressable parents, target mismatches, and depth exhaustion raise `CheckpointLineageIntegrityError` and fail closed instead of selecting a sibling. The compatibility scans request 400 raw checkpoints so up to 200 duration-only entries do not consume the effective branch-history budget; the fallback scans oldest-to-newest internally, skips duration-only checkpoints, and accepts only checkpoints with an addressable id as the replay base. A source history with no discoverable pre-user checkpoint preserves the historical single-checkpoint branch behavior instead of rejecting the branch; regeneration remains unavailable for that inherited response. Existing single-checkpoint branches are not mutated by regenerate preparation, and no raw checkpoint tuple is copied across threads because delta state depends on ancestry and pending writes. Regenerate source-run lookup uses the current thread's exact event, then the server-stamped `run_id` on the copied human message, then verified RunManager content matching; it does not read parent-thread events. Storage or checkpoint-mode failures are not treated as a missing base and still fail closed.
|
||||
**Replay checkpoint lookup prefers lineage and degrades only for an explicitly missing legacy parent link.** Branch and regenerate paths first walk `parent_config`, which prevents a global chronological scan from selecting a sibling created by regeneration. `CheckpointParentMissingError` alone enables the bounded newest-first history fallback in `app/gateway/checkpoint_lineage.py`; cycles, dangling/non-addressable parents, target mismatches, and depth exhaustion raise `CheckpointLineageIntegrityError` and fail closed instead of selecting a sibling. The compatibility scans request 400 raw checkpoints so up to 200 duration-only entries do not consume the effective branch-history budget; the fallback scans oldest-to-newest internally, skips duration-only checkpoints, and accepts only checkpoints with an addressable id as the replay base. A source history with no discoverable pre-user checkpoint preserves the historical single-checkpoint branch behavior instead of rejecting the branch; regeneration remains unavailable for that inherited response. Existing single-checkpoint branches are not mutated by regenerate preparation, and no raw checkpoint tuple is copied across threads because delta state depends on ancestry and pending writes. Regenerate source-run lookup uses the current thread's exact event, then the server-stamped `run_id` on the copied human message, then verified RunManager content matching; it does not read parent-thread events. When an interrupted response was streamed but never checkpointed, regeneration accepts only the latest visible human message's server-stamped `run_id` after verifying that it belongs to the same thread and still has `interrupted` status. Storage or checkpoint-mode failures are not treated as a missing base and still fail closed.
|
||||
|
||||
**A delta-mode run cannot fork; `runtime/runs/worker.py` linearizes the resume instead.** Resuming from an older checkpoint (regenerate, or any client-supplied `checkpoint`) forks the lineage, and delta state for a fork is not materializable: `BaseCheckpointSaver.get_delta_channel_history` — and the bespoke overrides in `InMemorySaver`/`PostgresSaver` — collect **every** `pending_writes` entry stored on each on-path ancestor, but a shared parent also carries the writes of the sibling child that was abandoned. Those writes replay into the fork, so the run starts from a message list still containing the answer it was supposed to replace (#4458: regenerating in a branched thread showed the superseded assistant message beside the new one after a reload; reproduced on postgres, sqlite, and the in-memory saver). Write-to-child ownership belongs to the upstream delta contract, so DeerFlow does not reimplement the walk: `_linearize_delta_checkpoint_resume` materializes the requested checkpoint's complete state and writes every channel onto the **current head** (which has no siblings) through the state mutation graph, using `Overwrite` for reducer channels and resetting newer head-only channels to their schema default (or `None` when no constructible default exists); it then drops the `checkpoint_id` selector and lets the run proceed linearly, while the abandoned turn stays in history as the rewritten head's ancestry. The worker holds `_checkpoint_thread_lock` across `_capture_rollback_point` and the optional linear rewrite, making the rollback snapshot and rewrite atomic with graph streaming and the preceding run's duration-metadata checkpoint write. Capture preserves the complete real pre-run state; cancel-with-rollback then linearly replaces the current delta head with that captured state rather than forking the now-shared pre-run checkpoint, so the abandoned turn is restored without replaying the resume sibling's writes. The worker also recomputes the current-run message boundary from the rewritten state and fails closed (an unreadable resume checkpoint raises rather than falling back to the corrupt fork). `full` mode keeps forking — its checkpoints carry complete `channel_values` and need no replay — so LangGraph branching semantics are unchanged there. Root namespace only; subgraph namespaces are left alone.
|
||||
|
||||
|
||||
@ -557,6 +557,33 @@ async def _require_successful_source_run(thread_id: str, run_id: str, request: R
|
||||
return record
|
||||
|
||||
|
||||
async def _find_interrupted_target_run_id(
|
||||
thread_id: str,
|
||||
source_human: Any,
|
||||
request: Request,
|
||||
) -> str | None:
|
||||
source_run_id = _message_additional_kwargs(source_human).get("run_id")
|
||||
if not isinstance(source_run_id, str) or not source_run_id:
|
||||
return None
|
||||
|
||||
run_mgr = get_run_manager(request)
|
||||
user_id = await get_current_user(request)
|
||||
record = await run_mgr.get(source_run_id, user_id=user_id)
|
||||
if record is None:
|
||||
records = await run_mgr.list_by_thread(thread_id, user_id=user_id, limit=20)
|
||||
record = next(
|
||||
(candidate for candidate in records if getattr(candidate, "run_id", None) == source_run_id),
|
||||
None,
|
||||
)
|
||||
if record is None:
|
||||
return None
|
||||
if getattr(record, "thread_id", None) != thread_id:
|
||||
return None
|
||||
if _run_status_value(record) != RunStatus.interrupted.value:
|
||||
return None
|
||||
return source_run_id
|
||||
|
||||
|
||||
async def _prepare_regenerate_payload(thread_id: str, message_id: str, request: Request) -> RegeneratePrepareResponse:
|
||||
accessor, latest_config = await build_thread_checkpoint_state_accessor(request, thread_id=thread_id)
|
||||
try:
|
||||
@ -571,18 +598,41 @@ async def _prepare_regenerate_payload(thread_id: str, message_id: str, request:
|
||||
messages = _checkpoint_messages(latest_checkpoint)
|
||||
target_index = next((i for i, message in enumerate(messages) if _message_id(message) == message_id), None)
|
||||
if target_index is None:
|
||||
raise HTTPException(status_code=404, detail=f"Message {message_id} not found")
|
||||
target_message = messages[target_index]
|
||||
if not _is_visible_ai_message(target_message):
|
||||
raise HTTPException(status_code=409, detail="Only visible assistant messages can be regenerated")
|
||||
# A response interrupted during an LLM call can be visible in the live
|
||||
# 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)),
|
||||
None,
|
||||
)
|
||||
target_run_id = await _find_interrupted_target_run_id(thread_id, previous_human, request) if previous_human is not None else None
|
||||
if target_run_id is None:
|
||||
raise HTTPException(status_code=404, detail=f"Message {message_id} not found")
|
||||
else:
|
||||
target_message = messages[target_index]
|
||||
if not _is_visible_ai_message(target_message):
|
||||
raise HTTPException(status_code=409, detail="Only visible assistant messages can be regenerated")
|
||||
|
||||
latest_visible_ai = next((message for message in reversed(messages) if _is_visible_ai_message(message)), None)
|
||||
if _message_id(latest_visible_ai) != message_id:
|
||||
raise HTTPException(status_code=409, detail="Only the latest assistant message can be regenerated")
|
||||
latest_visible_ai = next((message for message in reversed(messages) if _is_visible_ai_message(message)), None)
|
||||
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_visible_human_message(message)), None)
|
||||
target_run_id = (
|
||||
await _find_target_run_id(
|
||||
thread_id,
|
||||
message_id,
|
||||
target_message,
|
||||
previous_human,
|
||||
request,
|
||||
)
|
||||
if previous_human is not None
|
||||
else None
|
||||
)
|
||||
if previous_human is None:
|
||||
raise HTTPException(status_code=409, detail="Could not find the user message for this assistant response")
|
||||
if target_run_id is None:
|
||||
raise HTTPException(status_code=409, detail="Could not find source run for assistant message")
|
||||
previous_human_id = _message_id(previous_human)
|
||||
if not previous_human_id:
|
||||
raise HTTPException(status_code=409, detail="The source user message is missing an id")
|
||||
@ -593,13 +643,6 @@ async def _prepare_regenerate_payload(thread_id: str, message_id: str, request:
|
||||
request,
|
||||
head_checkpoint=latest_checkpoint,
|
||||
)
|
||||
target_run_id = await _find_target_run_id(
|
||||
thread_id,
|
||||
message_id,
|
||||
target_message,
|
||||
previous_human,
|
||||
request,
|
||||
)
|
||||
checkpoint = _checkpoint_response(base_checkpoint_tuple)
|
||||
metadata = {
|
||||
"regenerate_from_message_id": message_id,
|
||||
|
||||
@ -445,6 +445,100 @@ def test_prepare_regenerate_payload_preserves_latest_thread_title():
|
||||
assert response.input["title"] == "User renamed title"
|
||||
|
||||
|
||||
def test_prepare_regenerate_payload_supports_latest_interrupted_response_missing_from_checkpoint():
|
||||
from app.gateway.routers.thread_runs import _prepare_regenerate_payload
|
||||
|
||||
human = HumanMessage(
|
||||
id="human-1",
|
||||
content="question",
|
||||
additional_kwargs={"run_id": "run-interrupted"},
|
||||
)
|
||||
base = _checkpoint("ckpt-base", [])
|
||||
latest = _checkpoint("ckpt-human", [human])
|
||||
checkpointer = FakeCheckpointer([latest, base])
|
||||
run_manager = FakeRunManager(
|
||||
[
|
||||
SimpleNamespace(
|
||||
run_id="run-interrupted",
|
||||
thread_id="thread-1",
|
||||
status=RunStatus.interrupted,
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
response = asyncio.run(
|
||||
_prepare_regenerate_payload(
|
||||
"thread-1",
|
||||
"lc_run--partial-response",
|
||||
_request(
|
||||
checkpointer,
|
||||
FakeEventStore([]),
|
||||
run_manager=run_manager,
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
assert response.checkpoint["checkpoint_id"] == "ckpt-base"
|
||||
assert response.target_run_id == "run-interrupted"
|
||||
assert response.metadata == {
|
||||
"regenerate_from_message_id": "lc_run--partial-response",
|
||||
"regenerate_from_run_id": "run-interrupted",
|
||||
"regenerate_checkpoint_id": "ckpt-base",
|
||||
}
|
||||
assert response.input["messages"][0]["id"] == "human-1"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("status", "record_thread_id"),
|
||||
[
|
||||
(RunStatus.success, "thread-1"),
|
||||
(RunStatus.interrupted, "another-thread"),
|
||||
],
|
||||
)
|
||||
def test_prepare_regenerate_payload_does_not_accept_unverified_missing_response(
|
||||
status: RunStatus,
|
||||
record_thread_id: str,
|
||||
):
|
||||
from app.gateway.routers.thread_runs import _prepare_regenerate_payload
|
||||
|
||||
human = HumanMessage(
|
||||
id="human-1",
|
||||
content="question",
|
||||
additional_kwargs={"run_id": "source-run"},
|
||||
)
|
||||
checkpointer = FakeCheckpointer(
|
||||
[
|
||||
_checkpoint("ckpt-human", [human]),
|
||||
_checkpoint("ckpt-base", []),
|
||||
]
|
||||
)
|
||||
run_manager = FakeRunManager(
|
||||
[
|
||||
SimpleNamespace(
|
||||
run_id="source-run",
|
||||
thread_id=record_thread_id,
|
||||
status=status,
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
asyncio.run(
|
||||
_prepare_regenerate_payload(
|
||||
"thread-1",
|
||||
"missing-response",
|
||||
_request(
|
||||
checkpointer,
|
||||
FakeEventStore([]),
|
||||
run_manager=run_manager,
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
assert exc.value.status_code == 404
|
||||
assert exc.value.detail == "Message missing-response not found"
|
||||
|
||||
|
||||
def test_prepare_regenerate_payload_does_not_mutate_legacy_single_checkpoint_branch():
|
||||
from app.gateway.routers.thread_runs import _prepare_regenerate_payload
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user