diff --git a/CHANGELOG.md b/CHANGELOG.md index e3b8160a7e..ec9c8608fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,15 @@ This section accumulates work toward the **2.3.0** milestone ### Fixed +- **harness:** Retrying a failed item no longer reopens a cancelled durable + subagent batch. `retry_item` kept only a paused batch as it was, so a failed + item of a cancelled batch put the batch back to `queued`, the worker claimed + and ran the item, and the batch ended `completed`. The retry route now + answers 409 for a cancelled batch, including when the cancel commits while + the retry is in flight on SQLite, and the workspace no longer offers Retry + there. Retrying in a completed, failed, running or paused batch is + unchanged. ([#7792]) + - **agents:** `PUT /api/agents/{name}` with `"description": null` now resets the description to empty. It used to store the null, which the agent config rejects, so the agent then failed every load: the update, GET and repair @@ -15980,3 +15989,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7732]: https://github.com/bytedance/deer-flow/pull/7732 [#7762]: https://github.com/bytedance/deer-flow/pull/7762 [#7772]: https://github.com/bytedance/deer-flow/pull/7772 +[#7792]: https://github.com/bytedance/deer-flow/pull/7792 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 8f22cf3393..bf8ef3c752 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -20,6 +20,13 @@ ### 修复 +- **harness:** 重试失败条目不再重新打开已取消的持久子智能体批次。`retry_item` + 此前只保留已暂停批次的状态,因此重试已取消批次中的失败条目会把批次改回 + `queued`,worker 随即认领并执行该条目,批次最终变为 `completed`。现在对已取消 + 批次的重试路由返回 409,包括 SQLite 上取消在重试进行中提交的情况;工作区也不再 + 为这些条目显示重试按钮。在已完成、失败、运行中或已暂停批次中重试的行为不变。 + ([#7792]) + - **智能体:** `PUT /api/agents/{name}` 传入 `"description": null` 时现在会把描述重置为 空。此前会把 null 原样保存,而智能体配置不接受该值,导致之后每次加载都失败:更新本身、 GET 和修复性更新都返回 500,文件后端会把它从智能体列表中漏掉,数据库后端会让该用户的 @@ -12732,3 +12739,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子 [#7732]: https://github.com/bytedance/deer-flow/pull/7732 [#7762]: https://github.com/bytedance/deer-flow/pull/7762 [#7772]: https://github.com/bytedance/deer-flow/pull/7772 +[#7792]: https://github.com/bytedance/deer-flow/pull/7792 diff --git a/backend/app/gateway/routers/subagent_batches.py b/backend/app/gateway/routers/subagent_batches.py index 350953d82e..a1d44952a8 100644 --- a/backend/app/gateway/routers/subagent_batches.py +++ b/backend/app/gateway/routers/subagent_batches.py @@ -122,7 +122,7 @@ async def retry_batch_item(thread_id: ThreadId, batch_id: str, item_id: str, req repo, user_id, _batch = await _owned_batch(request, thread_id, batch_id) item = await repo.retry_item(batch_id, item_id, user_id=user_id) if item is None: - raise HTTPException(status_code=409, detail="Only failed items can be retried") + raise HTTPException(status_code=409, detail="Only failed items of a batch that is not cancelled can be retried") return item diff --git a/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py b/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py index d847d6c1a7..1ab94b35e5 100644 --- a/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py +++ b/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py @@ -627,6 +627,14 @@ class SubagentBatchRepository: item.completed_at = None item.cancel_requested_at = None item.updated_at = now + # SQLite ignores FOR UPDATE: flush first so this transaction holds the + # write lock, then re-read the batch a cancel may have committed since. + await session.flush() + await session.refresh(batch) + # Cancellation is final; reopening the batch would run work the user + # stopped. Returning without commit discards the item reset. + if batch.status == "cancelled": + return None if batch.status != "paused": batch.status = "queued" batch.completed_at = None diff --git a/backend/tests/test_cancelled_batch_retry.py b/backend/tests/test_cancelled_batch_retry.py new file mode 100644 index 0000000000..a90b266bc4 --- /dev/null +++ b/backend/tests/test_cancelled_batch_retry.py @@ -0,0 +1,109 @@ +"""Retrying a failed item never reopens a cancelled batch or runs its work.""" + +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import httpx +import pytest +import pytest_asyncio +from _router_auth_helpers import make_authed_test_app + +from app.gateway.auth_disabled import AUTH_SOURCE_SESSION +from app.gateway.routers import subagent_batches +from deerflow.config.database_config import DatabaseConfig +from deerflow.config.subagent_batches_config import SubagentBatchesConfig +from deerflow.config.subagent_runtime_config import SubagentRuntimeConfig +from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config +from deerflow.persistence.subagent_batches import SubagentBatchRepository +from deerflow.persistence.thread_meta import ThreadMetaRepository +from deerflow.subagents.batch_service import SubagentBatchService + +_URL = "/api/threads/thread-1/subagent-batches/batch-1" + + +@pytest_asyncio.fixture +async def api(tmp_path): + # Synthetic session identity; routing, owner checks, SQL and claiming are real. + user = SimpleNamespace(id="user-1", system_role="user") + app = make_authed_test_app(user_factory=lambda: user, bind_current_user=True) + app.include_router(subagent_batches.router) + + @app.middleware("http") + async def session_source(request, call_next): + request.state.auth_source = AUTH_SOURCE_SESSION + return await call_next(request) + + try: + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) + sessions = get_session_factory() + app.state.subagent_batch_repo = SubagentBatchRepository(sessions) + app.state.thread_store = ThreadMetaRepository(sessions) + service = SubagentBatchService( + repository=app.state.subagent_batch_repo, + config=SubagentBatchesConfig(lease_seconds=10), + runtime_config=SubagentRuntimeConfig(), + ) + # No executor is needed: a cancelled batch must never dispatch its items. + service._execute_item = AsyncMock() + app.state.subagent_batch_service = service + app.state.subagent_batches_available = True + await app.state.thread_store.create("thread-1", user_id="user-1") + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client: + yield app, client + finally: + await close_engine() + + +@pytest.mark.asyncio +async def test_retry_in_cancelled_batch_is_rejected_and_never_dispatched(api): + app, client = api + repo = app.state.subagent_batch_repo + service = app.state.subagent_batch_service + await repo.create_batch( + batch_id="batch-1", + user_id="user-1", + thread_id="thread-1", + run_id="run-1", + tool_call_id="call-1", + submission_key="run-1:call-1", + title="Cancelled research", + subagent_type="general-purpose", + items=[{"key": f"item-{index}", "prompt": f"Process {index}"} for index in range(3)], + max_live_items=1, + max_running_items=1, + max_attempts=1, + execution_spec={}, + ) + now = datetime.now(UTC) + failed = (await repo.claim_items(now=now, lease_owner="worker-1", lease_seconds=10, limit=1))[0] + assert await repo.finalize_item( + failed["id"], + lease_owner="worker-1", + succeeded=False, + result=None, + result_preview=None, + result_truncated=False, + error="permanent", + stop_reason=None, + token_usage=None, + model_name=None, + completed_at=now, + ) + assert (await client.post(f"{_URL}/cancel")).json()["status"] == "cancelled" + + retried = await client.post(f"{_URL}/items/{failed['id']}/retry") + + assert retried.status_code == 409 + assert retried.json()["detail"] == "Only failed items of a batch that is not cancelled can be retried" + await service.run_once(now=now + timedelta(seconds=1)) + service._execute_item.assert_not_awaited() + batch = (await client.get(_URL)).json() + assert batch["status"] == "cancelled" + assert batch["completed_at"] is not None + items = (await client.get(f"{_URL}/items")).json() + assert [(item["status"], item["error"]) for item in items] == [ + ("failed", "permanent"), + ("cancelled", "Cancelled by user"), + ("cancelled", "Cancelled by user"), + ] diff --git a/backend/tests/test_subagent_batch_repository.py b/backend/tests/test_subagent_batch_repository.py index ba4adf9768..6cf8a46614 100644 --- a/backend/tests/test_subagent_batch_repository.py +++ b/backend/tests/test_subagent_batch_repository.py @@ -9,6 +9,7 @@ from deerflow.config.database_config import DatabaseConfig from deerflow.mcp_scope import THREAD_INCARNATION_CONTEXT_KEY 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 _MISSING = object() @@ -350,6 +351,92 @@ async def test_retry_item_in_running_batch_keeps_batch_schedulable(tmp_path) -> assert failed["id"] in {item["id"] for item in claimed} +async def _fail_first_item(repo: SubagentBatchRepository, now: datetime) -> dict: + failed = (await repo.claim_items(now=now, lease_owner="worker-1", lease_seconds=60, limit=1))[0] + assert await repo.finalize_item( + failed["id"], + lease_owner="worker-1", + succeeded=False, + result=None, + result_preview=None, + result_truncated=False, + error="permanent", + stop_reason=None, + token_usage=None, + model_name="model-a", + completed_at=now, + ) + return failed + + +async def _assert_cancel_kept(repo: SubagentBatchRepository, failed: dict, now: datetime) -> None: + items = await repo.list_items("batch-1", user_id="user-1") + retried = next(item for item in items if item["id"] == failed["id"]) + assert retried["status"] == "failed" + assert retried["error"] == "permanent" + batch = await repo.get_batch("batch-1", user_id="user-1") + assert batch is not None and batch["status"] == "cancelled" + assert batch["completed_at"] is not None + assert batch["counts"]["pending"] == 0 + assert await repo.claim_items(now=now, lease_owner="worker-2", lease_seconds=60, limit=10) == [] + + +@pytest.mark.asyncio +async def test_retry_item_does_not_reopen_cancelled_batch(tmp_path) -> None: + repo = await _repo(tmp_path) + await _create(repo, count=3, max_live=3, max_running=3, max_attempts=1) + now = datetime.now(UTC) + failed = await _fail_first_item(repo, now) + cancelled = await repo.cancel_batch("batch-1", user_id="user-1") + assert cancelled is not None and cancelled["status"] == "cancelled" + + assert await repo.retry_item("batch-1", failed["id"], user_id="user-1") is None + + await _assert_cancel_kept(repo, failed, now + timedelta(seconds=1)) + + +class _CancelAfterItemRead: + """Session proxy that commits a cancel between ``retry_item``'s reads and its write.""" + + def __init__(self, session, cancel) -> None: + self._session = session + self._cancel = cancel + + def __getattr__(self, name): + return getattr(self._session, name) + + async def __aenter__(self): + await self._session.__aenter__() + return self + + async def __aexit__(self, *exc_info): + return await self._session.__aexit__(*exc_info) + + async def get(self, entity, ident, **kwargs): + row = await self._session.get(entity, ident, **kwargs) + if entity is SubagentBatchItemRow: + await self._cancel() + return row + + +@pytest.mark.asyncio +async def test_retry_item_does_not_reopen_batch_cancelled_after_its_read(tmp_path) -> None: + # SQLite ignores FOR UPDATE, so the cancel can commit after retry_item has + # read the batch as running. (On PostgreSQL the row lock blocks the cancel.) + repo = await _repo(tmp_path) + await _create(repo, count=3, max_live=3, max_running=3, max_attempts=1) + now = datetime.now(UTC) + failed = await _fail_first_item(repo, now) + batch = await repo.get_batch("batch-1", user_id="user-1") + assert batch is not None and batch["status"] == "running" + sessions = get_session_factory() + racing = SubagentBatchRepository(lambda: _CancelAfterItemRead(sessions(), lambda: repo.cancel_batch("batch-1", user_id="user-1"))) + + assert await racing.retry_item("batch-1", failed["id"], user_id="user-1") is None + + await _assert_cancel_kept(repo, failed, now + timedelta(seconds=1)) + + @pytest.mark.asyncio async def test_cancel_terminalizes_in_flight_items_and_fences_stale_completion(tmp_path) -> None: repo = await _repo(tmp_path) diff --git a/frontend/src/components/workspace/thread-subagent-batches.tsx b/frontend/src/components/workspace/thread-subagent-batches.tsx index 2c9e8ba317..d2284460b5 100644 --- a/frontend/src/components/workspace/thread-subagent-batches.tsx +++ b/frontend/src/components/workspace/thread-subagent-batches.tsx @@ -304,6 +304,7 @@ function BatchItems({ key={item.id} item={item} workerRunning={workerRunning} + batchCancelled={batch.status === "cancelled"} retrying={retry.isPending && retry.variables === item.id} onRetry={() => retry.mutate(item.id)} /> @@ -332,6 +333,7 @@ function BatchItemRow({ batchId, item, workerRunning, + batchCancelled, retrying, onRetry, }: { @@ -339,6 +341,7 @@ function BatchItemRow({ batchId: string; item: SubagentBatchItem; workerRunning: boolean; + batchCancelled: boolean; retrying: boolean; onRetry: () => void; }) { @@ -374,7 +377,7 @@ function BatchItemRow({ > {t.subagentBatches.viewResult} - {item.status === "failed" && ( + {item.status === "failed" && !batchCancelled && (