mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
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>
This commit is contained in:
parent
734f9e0e92
commit
a9e5b6a853
@ -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)
|
||||
|
||||
@ -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)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user