mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
fix(subagents): leave full results out of the polled batch item list (#6781)
* fix(subagents): leave full results out of the polled batch item list The item list endpoint returns only previews, but list_items selected whole rows, so every page loaded and decoded each item's full report, evidence JSON and prompt before discarding them. The workspace polls this list every 2 seconds while a batch runs; with the default max_result_chars a 100-item page held about 34 MB and blocked the event loop for about 70 ms (170 MB and 300 ms at limit=500). Defer those columns unless the caller asks for them, with raiseload so an accidental read fails loudly instead of lazy-loading. A 100-item page now peaks under 1 MB; responses are unchanged. * test(subagents): pin the item list route to compact columns Drive list_batch_items against a real SQLite repository so a route that starts requesting full results fails, not only a repository change. * docs(changelog): reference #6781 * test(subagents): scope the route test's seed update to its batch
This commit is contained in:
parent
ec81b4d8dd
commit
75245e4b73
@ -5856,6 +5856,13 @@ This release closes that milestone with **621 merged pull requests**.
|
||||
default limit (roughly 1.7 GB at the 1,000,000 ceiling) and stalled the event
|
||||
loop for over half a second per page. The exported lines and their order are
|
||||
unchanged. ([#6706])
|
||||
- **gateway:** The subagent batch item list
|
||||
(`GET /api/threads/{thread_id}/subagent-batches/{batch_id}/items`), which the
|
||||
workspace polls every 2 seconds while a batch runs, no longer loads each
|
||||
item's full report, evidence JSON and prompt only to return its preview.
|
||||
With the default `max_result_chars`, one 100-item page held about 34 MB and
|
||||
blocked the event loop for about 70 ms (170 MB and 300 ms at `limit=500`);
|
||||
it now holds under 1 MB. The response is unchanged. ([#6781])
|
||||
|
||||
### Security
|
||||
|
||||
@ -11633,4 +11640,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
|
||||
[#6702]: https://github.com/bytedance/deer-flow/pull/6702
|
||||
[#6706]: https://github.com/bytedance/deer-flow/pull/6706
|
||||
[#6746]: https://github.com/bytedance/deer-flow/pull/6746
|
||||
[#6781]: https://github.com/bytedance/deer-flow/pull/6781
|
||||
[#6798]: https://github.com/bytedance/deer-flow/pull/6798
|
||||
|
||||
@ -4458,6 +4458,11 @@
|
||||
的报告和同等长度的证据 JSON,而导出会先把整页载入内存再开始输出,因此默认上限下
|
||||
单个请求约占用 280 MB(上限 1,000,000 时约 1.7 GB),并且每页会阻塞事件循环
|
||||
半秒以上。导出的行及其顺序不变。([#6706])
|
||||
- **网关:** 子代理批次条目列表
|
||||
(`GET /api/threads/{thread_id}/subagent-batches/{batch_id}/items`,批次运行时工作区
|
||||
每 2 秒轮询一次)不再为了只返回预览而载入每个条目的完整报告、证据 JSON 和提示词。
|
||||
默认 `max_result_chars` 下,一页 100 个条目约占用 34 MB,并阻塞事件循环约 70 ms
|
||||
(`limit=500` 时为 170 MB 和 300 ms);现在不到 1 MB。响应内容不变。([#6781])
|
||||
|
||||
### 安全
|
||||
|
||||
@ -9296,3 +9301,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子
|
||||
[#6702]: https://github.com/bytedance/deer-flow/pull/6702
|
||||
[#6706]: https://github.com/bytedance/deer-flow/pull/6706
|
||||
[#6746]: https://github.com/bytedance/deer-flow/pull/6746
|
||||
[#6781]: https://github.com/bytedance/deer-flow/pull/6781
|
||||
|
||||
@ -8,6 +8,7 @@ from typing import Any
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||
from sqlalchemy.orm import defer
|
||||
|
||||
from deerflow.persistence.subagent_batches.model import SubagentBatchItemRow, SubagentBatchRow
|
||||
from deerflow.subagents.acceptance_checks import AcceptanceVerdict, validate_acceptance_verdict
|
||||
@ -222,6 +223,12 @@ class SubagentBatchRepository:
|
||||
if batch is None or batch.user_id != user_id:
|
||||
return None
|
||||
stmt = select(SubagentBatchItemRow).where(SubagentBatchItemRow.batch_id == batch_id)
|
||||
# Full reports and evidence reach max_result_chars each; the polled
|
||||
# item list returns only previews, so leave them out of the SELECT.
|
||||
if not include_result:
|
||||
stmt = stmt.options(defer(SubagentBatchItemRow.result, raiseload=True), defer(SubagentBatchItemRow.result_artifact, raiseload=True))
|
||||
if not include_prompt:
|
||||
stmt = stmt.options(defer(SubagentBatchItemRow.prompt, raiseload=True))
|
||||
if status is not None:
|
||||
stmt = stmt.where(SubagentBatchItemRow.status == status)
|
||||
stmt = stmt.order_by(SubagentBatchItemRow.position).offset(offset).limit(limit)
|
||||
|
||||
@ -16,7 +16,7 @@ Durable batch specs store overlays as JSON and restore them before execution.
|
||||
|
||||
Batch evidence uses nullable `result_artifact` (migration 0033), persisted with
|
||||
successful results under the same lease fence. Only full-result exports expose
|
||||
bounded, cited RAG snapshots; compact item projections omit them. Cancelled,
|
||||
bounded, cited RAG snapshots; compact `list_items` defers them. Cancelled,
|
||||
failed and stale attempts cannot publish evidence; retry clears it. Cover the
|
||||
formatter/step capture -> worker -> reopened SQLite -> authorized export chain
|
||||
in `tests/test_batch_rag_evidence.py`; no provider fetch or preview UI is added.
|
||||
|
||||
@ -1,11 +1,13 @@
|
||||
import re
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sqlalchemy import event
|
||||
|
||||
from deerflow.config.database_config import DatabaseConfig
|
||||
from deerflow.mcp_scope import THREAD_INCARNATION_CONTEXT_KEY
|
||||
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
|
||||
from deerflow.persistence.engine import close_engine, get_engine, get_session_factory, init_engine_from_config
|
||||
from deerflow.persistence.subagent_batches import SubagentBatchRepository
|
||||
|
||||
_MISSING = object()
|
||||
@ -461,6 +463,32 @@ async def test_public_projections_omit_execution_context_and_full_results(tmp_pa
|
||||
assert export_item["result"] == "full private result"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_item_list_selects_only_the_heavy_columns_it_returns(tmp_path) -> None:
|
||||
# The workspace polls this list every 2 s. Selecting whole rows decoded
|
||||
# every full report and evidence artifact only to return their previews.
|
||||
repo = await _repo(tmp_path)
|
||||
await _create(repo, count=1)
|
||||
statements = []
|
||||
sync_engine = get_engine().sync_engine
|
||||
listener = lambda *args, **kwargs: statements.append(args[2]) # noqa: E731
|
||||
event.listen(sync_engine, "before_cursor_execute", listener)
|
||||
try:
|
||||
public_items = await repo.list_items("batch-1", user_id="user-1")
|
||||
full_items = await repo.list_items("batch-1", user_id="user-1", include_result=True, include_prompt=True)
|
||||
finally:
|
||||
event.remove(sync_engine, "before_cursor_execute", listener)
|
||||
|
||||
item_columns = [set(re.findall(r"subagent_batch_items\.(\w+)", statement)) for statement in statements if "FROM subagent_batch_items" in statement]
|
||||
heavy_columns = {"prompt", "result", "result_artifact"}
|
||||
assert len(item_columns) == 2
|
||||
assert not heavy_columns & item_columns[0]
|
||||
assert heavy_columns <= item_columns[1]
|
||||
assert "result_preview" in public_items[0]
|
||||
assert full_items[0]["prompt"] == "Process 0"
|
||||
assert full_items[0]["result"] is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_duplicate_submission_key_returns_original_batch(tmp_path) -> None:
|
||||
repo = await _repo(tmp_path)
|
||||
|
||||
@ -1,12 +1,18 @@
|
||||
import json
|
||||
import re
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from sqlalchemy import event, update
|
||||
|
||||
from app.gateway.app import create_app
|
||||
from app.gateway.routers import subagent_batches
|
||||
from deerflow.config.database_config import DatabaseConfig
|
||||
from deerflow.persistence.engine import close_engine, get_engine, get_session_factory, init_engine_from_config
|
||||
from deerflow.persistence.subagent_batches import SubagentBatchRepository
|
||||
from deerflow.persistence.subagent_batches.model import SubagentBatchItemRow
|
||||
|
||||
|
||||
def _batch(**overrides):
|
||||
@ -86,6 +92,50 @@ async def test_list_and_detail_are_owner_scoped(monkeypatch) -> None:
|
||||
assert cross_thread.value.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_item_list_route_never_reads_full_results(monkeypatch, tmp_path) -> None:
|
||||
# The workspace polls this route every 2 s; only the JSONL export may pay
|
||||
# for full reports, evidence JSON and prompts.
|
||||
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
|
||||
try:
|
||||
repo = SubagentBatchRepository(get_session_factory())
|
||||
await repo.create_batch(
|
||||
batch_id="batch-1",
|
||||
user_id="user-1",
|
||||
thread_id="thread-1",
|
||||
run_id=None,
|
||||
tool_call_id=None,
|
||||
submission_key="submission-1",
|
||||
title="Research",
|
||||
subagent_type="general-purpose",
|
||||
items=[{"key": "one", "prompt": "private prompt"}],
|
||||
max_live_items=1,
|
||||
max_running_items=1,
|
||||
max_attempts=1,
|
||||
execution_spec={},
|
||||
)
|
||||
async with get_session_factory()() as session:
|
||||
await session.execute(update(SubagentBatchItemRow).where(SubagentBatchItemRow.batch_id == "batch-1").values(status="succeeded", result="full report", result_artifact={"knowledge_sources": {}}, result_preview="preview"))
|
||||
await session.commit()
|
||||
monkeypatch.setattr(subagent_batches, "get_current_user", AsyncMock(return_value="user-1"))
|
||||
statements = []
|
||||
sync_engine = get_engine().sync_engine
|
||||
listener = lambda *args, **kwargs: statements.append(args[2]) # noqa: E731
|
||||
event.listen(sync_engine, "before_cursor_execute", listener)
|
||||
try:
|
||||
items = await subagent_batches.list_batch_items.__wrapped__(thread_id="thread-1", batch_id="batch-1", request=_request(repo), offset=0, limit=100, status=None)
|
||||
finally:
|
||||
event.remove(sync_engine, "before_cursor_execute", listener)
|
||||
finally:
|
||||
await close_engine()
|
||||
|
||||
assert [item["result_preview"] for item in items] == ["preview"]
|
||||
assert not {"prompt", "result", "result_artifact"} & items[0].keys()
|
||||
item_columns = set().union(*(re.findall(r"subagent_batch_items\.(\w+)", statement) for statement in statements if "FROM subagent_batch_items" in statement))
|
||||
assert "result_preview" in item_columns
|
||||
assert not {"prompt", "result", "result_artifact"} & item_columns
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cancel_requires_running_worker_and_exact_owner(monkeypatch) -> None:
|
||||
repo = Repository()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user