mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-25 07:28:07 +00:00
fix(thread-history): keep completed subtask cards stable after reload (#4432)
* fix(thread-history): hide subagent AI responses * refactor(thread-runs): remove unused _is_middleware_message_row helper --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Nan Gao
GitHub
parent
16b612cfcf
commit
58befaf248
@ -765,7 +765,7 @@ Use `/compact` in the Web UI composer to summarize older context for the current
|
||||
|
||||
Complex tasks rarely fit in a single pass. DeerFlow decomposes them.
|
||||
|
||||
The lead agent can spawn sub-agents on the fly — each with its own scoped context, tools, and termination conditions. Sub-agents run in parallel when possible, report back structured results, and the lead agent synthesizes everything into a coherent output. Their configured skills are resolved from the same user-scoped catalog as the lead agent, so user-owned custom skills remain available without exposing another user's version. Their internal AI and tool messages stay scoped to the delegated graph instead of entering the parent chat stream. Long-running sub-agents compact older history when summarization is enabled and re-inject the summary as guarded, hidden durable context before continuing, so recent assistant/tool activity remains grounded in the task. Provider/model request failures are reported as failed sub-agent tasks rather than successful results, so the lead agent and Web UI can react to them correctly. Collapsed sub-agent cards show the effective model and, when the provider returns usage metadata, a cumulative token total that updates after each completed sub-agent LLM call and persists after a reload. When token usage tracking is enabled, completed sub-agent usage is also attributed back to the dispatching step.
|
||||
The lead agent can spawn sub-agents on the fly — each with its own scoped context, tools, and termination conditions. Sub-agents run in parallel when possible, report back structured results, and the lead agent synthesizes everything into a coherent output. Their configured skills are resolved from the same user-scoped catalog as the lead agent, so user-owned custom skills remain available without exposing another user's version. Their internal AI and tool messages stay scoped to the delegated graph instead of entering the parent chat stream. Reloaded thread history enforces the same boundary: callback-captured sub-agent AI responses remain available in run-event diagnostics but are excluded from the parent transcript, while the parent `task` result remains attached to its subtask card. Long-running sub-agents compact older history when summarization is enabled and re-inject the summary as guarded, hidden durable context before continuing, so recent assistant/tool activity remains grounded in the task. Provider/model request failures are reported as failed sub-agent tasks rather than successful results, so the lead agent and Web UI can react to them correctly. Collapsed sub-agent cards show the effective model and, when the provider returns usage metadata, a cumulative token total that updates after each completed sub-agent LLM call and persists after a reload. When token usage tracking is enabled, completed sub-agent usage is also attributed back to the dispatching step.
|
||||
|
||||
This is how DeerFlow handles tasks that take minutes to hours: a research task might fan out into a dozen sub-agents, each exploring a different angle, then converge into a single report — or a website — or a slide deck with generated visuals. One harness, many hands.
|
||||
|
||||
|
||||
@ -422,7 +422,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; `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/successful-regenerate filtering and page-run-scoped feedback enrichment; `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 assistant answer; `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 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. |
|
||||
|
||||
@ -277,8 +277,9 @@ def _is_visible_ai_message(message: Any) -> bool:
|
||||
return _message_type(message) == "ai" and not _is_hidden_or_control_message(message)
|
||||
|
||||
|
||||
def _is_middleware_message_row(row: dict[str, Any]) -> bool:
|
||||
return str((row.get("metadata") or {}).get("caller", "")).startswith("middleware:")
|
||||
def _is_thread_history_hidden_message_row(row: dict[str, Any]) -> bool:
|
||||
caller = str((row.get("metadata") or {}).get("caller", ""))
|
||||
return caller.startswith("middleware:") or (caller.startswith("subagent:") and _message_type(row.get("content")) == "ai")
|
||||
|
||||
|
||||
def _checkpoint_messages(snapshot: Any) -> list[Any]:
|
||||
@ -837,7 +838,7 @@ async def _scan_thread_message_page(
|
||||
raise RuntimeError("Run event message rows are missing sequence values")
|
||||
|
||||
for row in reversed(raw):
|
||||
if _is_middleware_message_row(row) or row.get("run_id") in superseded_run_ids:
|
||||
if _is_thread_history_hidden_message_row(row) or row.get("run_id") in superseded_run_ids:
|
||||
continue
|
||||
visible_desc.append(row)
|
||||
if len(visible_desc) == limit + 1:
|
||||
|
||||
@ -41,8 +41,11 @@ database store.
|
||||
|
||||
`category="message"` means an event is eligible for a message projection; it
|
||||
does not guarantee that the row is visible in the UI. Thread-history APIs also
|
||||
filter middleware model calls and superseded regenerate runs, and the frontend
|
||||
honors message-level visibility markers such as `hide_from_ui`.
|
||||
filter middleware model calls, subagent AI responses, and superseded regenerate
|
||||
runs, and the frontend honors message-level visibility markers such as
|
||||
`hide_from_ui`. Subagent events remain available through the run-events
|
||||
endpoint; parent `task` ToolMessages remain in thread history so subtask cards
|
||||
can restore their terminal status after reload.
|
||||
|
||||
All other categories are excluded from message projections and are available
|
||||
through run-event or specialized APIs:
|
||||
@ -122,7 +125,7 @@ Schema. It is the authoritative field-level reference.
|
||||
|
||||
| Consumer | Read path and behavior |
|
||||
| --- | --- |
|
||||
| Frontend thread history | `GET /api/threads/{thread_id}/messages/page` scans `list_messages()`, removes middleware rows and superseded regenerate runs, then applies frontend message visibility rules. |
|
||||
| Frontend thread history | `GET /api/threads/{thread_id}/messages/page` scans `list_messages()`, removes middleware rows, subagent AI responses, and superseded regenerate runs, then applies frontend message visibility rules. |
|
||||
| Per-run message clients | Thread-scoped and stateless run message endpoints call `list_messages_by_run()`. |
|
||||
| Run debug/audit | `GET /api/threads/{thread_id}/runs/{run_id}/events` calls `list_events()` and supports `event_types`, `task_id`, `limit`, and `after_seq`. |
|
||||
| Historical subtask cards | Fetch `subagent.step` through the run-events endpoint, filtered and paginated by `task_id`. |
|
||||
|
||||
@ -62,14 +62,15 @@ def test_thread_page_orders_across_runs_and_paginates_without_gaps():
|
||||
assert older.json()["next_before_seq"] is None
|
||||
|
||||
|
||||
def test_thread_page_scans_past_middleware_chunks_to_fill_visible_page(monkeypatch):
|
||||
@pytest.mark.parametrize("caller", ["middleware:title", "subagent:general-purpose"])
|
||||
def test_thread_page_scans_past_internal_ai_chunks_to_fill_visible_page(monkeypatch, caller):
|
||||
monkeypatch.setattr(thread_runs, "THREAD_MESSAGE_PAGE_SCAN_BATCH", 3)
|
||||
store = MemoryRunEventStore()
|
||||
|
||||
async def seed():
|
||||
await _put_message(store, "run-1", "human", "visible-old")
|
||||
for index in range(3):
|
||||
await _put_message(store, "run-1", "ai", f"middleware-{index}", caller="middleware:title")
|
||||
await _put_message(store, "run-1", "ai", f"internal-{index}", caller=caller)
|
||||
await _put_message(store, "run-2", "human", "visible-new-human")
|
||||
await _put_message(store, "run-2", "ai", "visible-new-ai")
|
||||
|
||||
@ -84,6 +85,62 @@ def test_thread_page_scans_past_middleware_chunks_to_fill_visible_page(monkeypat
|
||||
assert body["next_before_seq"] == 5
|
||||
|
||||
|
||||
def test_thread_page_hides_subagent_ai_but_keeps_root_task_result():
|
||||
store = MemoryRunEventStore()
|
||||
|
||||
async def seed():
|
||||
await _put_message(store, "run-1", "human", "user-prompt")
|
||||
await store.put(
|
||||
thread_id="thread-1",
|
||||
run_id="run-1",
|
||||
event_type="llm.ai.response",
|
||||
category="message",
|
||||
content={
|
||||
"type": "ai",
|
||||
"id": "lead-task-call",
|
||||
"content": "",
|
||||
"tool_calls": [{"id": "task-0", "name": "task", "args": {}}],
|
||||
"additional_kwargs": {},
|
||||
},
|
||||
metadata={"caller": "lead_agent"},
|
||||
)
|
||||
await _put_message(
|
||||
store,
|
||||
"run-1",
|
||||
"ai",
|
||||
"internal-subagent-answer",
|
||||
caller="subagent:general-purpose",
|
||||
)
|
||||
await store.put(
|
||||
thread_id="thread-1",
|
||||
run_id="run-1",
|
||||
event_type="llm.tool.result",
|
||||
category="message",
|
||||
content={
|
||||
"type": "tool",
|
||||
"id": "root-task-result",
|
||||
"tool_call_id": "task-0",
|
||||
"content": "Task Succeeded. Result: poem",
|
||||
"additional_kwargs": {"subagent_status": "completed"},
|
||||
},
|
||||
metadata={},
|
||||
)
|
||||
await _put_message(store, "run-1", "ai", "lead-final-answer")
|
||||
|
||||
asyncio.run(seed())
|
||||
app = _make_app(store)
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/api/threads/thread-1/messages/page")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert [row["content"]["id"] for row in response.json()["data"]] == [
|
||||
"user-prompt",
|
||||
"lead-task-call",
|
||||
"root-task-result",
|
||||
"lead-final-answer",
|
||||
]
|
||||
|
||||
|
||||
def test_thread_page_scans_large_middleware_only_region_with_production_batch_size():
|
||||
store = MemoryRunEventStore()
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user