fix(harness): keep a cancelled batch cancelled when retrying a failed item (#7792)

* fix(harness): keep a cancelled batch cancelled when retrying a failed item

retry_item only kept a paused batch as it was, so retrying a failed item
of a cancelled batch reset the batch to queued and the worker claimed and
ran the item. A cancelled batch now rejects the retry (409), re-checked
under the write lock because SQLite ignores FOR UPDATE, and the workspace
no longer offers Retry there.

* docs(changelog): reference #7792 for the cancelled batch retry fix

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

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

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
 
 

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

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"),
+    ]

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)

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({
         >
           <EyeIcon /> {t.subagentBatches.viewResult}
         </Button>
-        {item.status === "failed" && (
+        {item.status === "failed" && !batchCancelled && (
           <Button
             type="button"
             size="sm"

frontend/src/content/en/harness/subagents/observability.mdx

@@ -111,7 +111,7 @@ Durable batches have no SSE; the workspace UI polls, every 2 seconds while a bat
 | `POST /{batch_id}/pause`                   | Pause                                                                            |
 | `POST /{batch_id}/resume`                  | Resume                                                                           |
 | `POST /{batch_id}/cancel`                  | Cancel; 503 when the worker is not running                                       |
-| `POST /{batch_id}/items/{item_id}/retry`   | Retry one item; only `failed` items, otherwise 409                               |
+| `POST /{batch_id}/items/{item_id}/retry`   | Retry one `failed` item; 409 otherwise or when the batch is cancelled            |
 | `GET /{batch_id}/items/{position}`         | Read one owner/thread-scoped report and captured evidence snapshot                |
 | `GET /{batch_id}/results.jsonl`            | Stream every item as NDJSON, including full results and acceptance verdicts      |
 

frontend/src/content/zh/harness/subagents/observability.mdx

@@ -111,7 +111,7 @@ GET /api/threads/{thread_id}/token-usage?include_active=false
 | `POST /{batch_id}/pause`                  | 暂停                                                                  |
 | `POST /{batch_id}/resume`                 | 恢复                                                                  |
 | `POST /{batch_id}/cancel`                 | 取消；worker 未运行时返回 503                                         |
-| `POST /{batch_id}/items/{item_id}/retry`  | 重试单个条目，只接受 `failed` 状态，否则 409                          |
+| `POST /{batch_id}/items/{item_id}/retry`  | 重试单个 `failed` 条目；其他状态或批次已取消时返回 409                |
 | `GET /{batch_id}/items/{position}`        | 读取一个按所有者/线程授权的报告和已捕获证据快照                  |
 | `GET /{batch_id}/results.jsonl`           | 以 NDJSON 流式导出全部条目，含完整结果与验收结论                      |
 

frontend/tests/unit/components/workspace/thread-subagent-batches.dom.test.tsx

@@ -197,6 +197,38 @@ describe("ThreadSubagentBatches capability gating", () => {
     expect(batchState.fetchNextPage).toHaveBeenCalledTimes(1);
   });
 
+  it.each([
+    ["running", true],
+    ["completed", true],
+    ["cancelled", false],
+  ])(
+    "offers retry for a failed item of a %s batch: %s",
+    async (status, offered) => {
+      featureState.workerRunning = true;
+      batchState.batches = [{ ...HISTORICAL_BATCH, status }];
+      batchState.itemPages = [
+        [
+          {
+            id: "item-1",
+            item_key: "record-1",
+            position: 0,
+            status: "failed",
+            error: "permanent",
+          },
+        ],
+      ];
+      render(<ThreadSubagentBatches threadId="thread-1" />);
+
+      fireEvent.click(screen.getByRole("button", { name: "Batches" }));
+      fireEvent.click(screen.getByRole("button", { name: "View items" }));
+
+      expect(await screen.findByText("permanent")).toBeDefined();
+      expect(screen.queryByRole("button", { name: "Retry" }) !== null).toBe(
+        offered,
+      );
+    },
+  );
+
   it("reviews a stored report and captured evidence", async () => {
     batchState.batches = [HISTORICAL_BATCH];
     batchState.itemPages = [