From a9e5b6a8537484f280dc42b17856eed0f35a8951 Mon Sep 17 00:00:00 2001 From: Yufeng He <40085740+he-yufeng@users.noreply.github.com> Date: Wed, 7 Oct 2026 17:24:12 +0800 Subject: [PATCH] fix(harness): keep a paused batch paused when retrying one failed item (#6417) retry_item unconditionally flipped batch.status to queued, and claim_items only schedules queued or running batches. Retrying one failed item in a paused batch therefore lifted the pause barrier and let the scheduler take every unrelated pending item without an explicit resume. Leave the batch row untouched when it is paused; the retried item is still reset to pending and waits for resume_batch like its siblings. Non paused batches keep the existing requeue behavior. Fixes #6411 Signed-off-by: Yufeng He <40085740+he-yufeng@users.noreply.github.com> --- .../persistence/subagent_batches/sql.py | 7 +- .../tests/test_subagent_batch_repository.py | 79 +++++++++++++++++++ 2 files changed, 83 insertions(+), 3 deletions(-) diff --git a/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py b/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py index c441ea520d..2b95f78381 100644 --- a/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py +++ b/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py @@ -585,8 +585,9 @@ class SubagentBatchRepository: item.completed_at = None item.cancel_requested_at = None item.updated_at = now - batch.status = "queued" - batch.completed_at = None - batch.updated_at = now + if batch.status != "paused": + batch.status = "queued" + batch.completed_at = None + batch.updated_at = now await session.commit() return self._item_dict(item) diff --git a/backend/tests/test_subagent_batch_repository.py b/backend/tests/test_subagent_batch_repository.py index 2c2dc7bd8c..4880211b31 100644 --- a/backend/tests/test_subagent_batch_repository.py +++ b/backend/tests/test_subagent_batch_repository.py @@ -269,6 +269,85 @@ async def test_pause_resume_cancel_and_owner_scope(tmp_path) -> None: assert await repo.get_batch("batch-1", user_id="other") is None +@pytest.mark.asyncio +async def test_retry_item_keeps_paused_batch_paused_until_resume(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 repo.claim_items(now=now, lease_owner="worker-1", lease_seconds=60, limit=1))[0] + 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, + ) + paused = await repo.pause_batch("batch-1", user_id="user-1") + assert paused is not None and paused["status"] == "paused" + assert await repo.claim_items(now=now + timedelta(seconds=1), lease_owner="worker-2", lease_seconds=60, limit=10) == [] + + retried = await repo.retry_item("batch-1", failed["id"], user_id="user-1") + + assert retried is not None + assert retried["status"] == "pending" + assert retried["attempt"] == 0 + assert retried["error"] is None + batch = await repo.get_batch("batch-1", user_id="user-1") + assert batch is not None and batch["status"] == "paused" + assert batch["counts"]["pending"] == 1 + assert batch["counts"]["queued"] == 2 + assert await repo.claim_items(now=now + timedelta(seconds=2), lease_owner="worker-2", lease_seconds=60, limit=10) == [] + + await close_engine() + repo = await _repo(tmp_path) + batch = await repo.get_batch("batch-1", user_id="user-1") + assert batch is not None and batch["status"] == "paused" + + resumed = await repo.resume_batch("batch-1", user_id="user-1") + assert resumed is not None and resumed["status"] == "queued" + claimed = await repo.claim_items(now=now + timedelta(seconds=3), lease_owner="worker-3", lease_seconds=60, limit=10) + assert len(claimed) == 3 + assert failed["id"] in {item["id"] for item in claimed} + + +@pytest.mark.asyncio +async def test_retry_item_in_running_batch_keeps_batch_schedulable(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 repo.claim_items(now=now, lease_owner="worker-1", lease_seconds=60, limit=1))[0] + 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, + ) + batch = await repo.get_batch("batch-1", user_id="user-1") + assert batch is not None and batch["status"] == "running" + + retried = await repo.retry_item("batch-1", failed["id"], user_id="user-1") + + assert retried is not None and retried["status"] == "pending" + batch = await repo.get_batch("batch-1", user_id="user-1") + assert batch is not None and batch["status"] == "queued" + claimed = await repo.claim_items(now=now + timedelta(seconds=1), lease_owner="worker-2", lease_seconds=60, limit=10) + assert len(claimed) == 3 + assert failed["id"] in {item["id"] for item in claimed} + + @pytest.mark.asyncio async def test_cancel_terminalizes_in_flight_items_and_fences_stale_completion(tmp_path) -> None: repo = await _repo(tmp_path)