From 85acf643c032d0b2e8262ad276af3be85e67c3a1 Mon Sep 17 00:00:00 2001 From: theater <1347507191@qq.com> Date: Mon, 28 Sep 2026 15:29:18 +0800 Subject: [PATCH] fix(subagents): terminalize batches after exhausted leases expire (#5897) * fix(subagents): terminalize batches after exhausted leases expire * docs(subagents): stay within agent guidance budget --- README.md | 2 + .../persistence/subagent_batches/sql.py | 6 +++ .../tests/test_subagent_batch_repository.py | 51 +++++++++++++++++++ 3 files changed, 59 insertions(+) diff --git a/README.md b/README.md index 1399679cf0..121628c7ce 100644 --- a/README.md +++ b/README.md @@ -1702,6 +1702,8 @@ Durable `batch_task` workers use one app-owned plugin snapshot for tool assembly Ordinary `task` delegation and explicit durable `batch_task` execution share the startup-scoped `subagent_runtime` process capacity. Batch mode keeps large independent item sets in SQL with separate total, live, and running limits, restart recovery, bounded results, and a thread-scoped Web UI panel. The panel pages through bounded previews on demand; full stored result text is available only through the owner-scoped JSONL export, while internal execution and authorization context never enters owner-facing responses. If the batch worker is later stopped or disabled, threads with persisted batches retain read-only item inspection and JSONL export; execution controls remain disabled until the worker is running again. See `config.example.yaml` and [the implementation contract](docs/plans/2026-08-24-subagent-batch-capacity-implementation.md) for limits and recovery semantics. +If an item's final lease expires after exhausting the retry budget, the batch reaches a terminal state once all items are terminal: `failed` when none succeeded, or `completed` when at least one succeeded. The item remains `failed` in both cases so partial results stay visible. + Direct `create_deerflow_agent(...)` integrations can own the same boundary explicitly instead of relying on Gateway startup. Construct one `SubagentRuntime` and share it across every graph in that application; its `max_running`, ordinary per-run total, bound `task` tool, and optional durable-batch tools then use the same caller-owned snapshot and execution controller. A runtime with a batch repository owns a worker and must be started before graph construction and stopped during application shutdown: ```python diff --git a/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py b/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py index 238b22d607..c441ea520d 100644 --- a/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py +++ b/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py @@ -264,6 +264,7 @@ class SubagentBatchRepository: ) ).scalars() ) + terminalized_expired = False for item in expired: item.lease_owner = None item.lease_expires_at = None @@ -271,14 +272,19 @@ class SubagentBatchRepository: if item.cancel_requested_at is not None: item.status = "cancelled" item.completed_at = now + terminalized_expired = True elif item.attempt >= batch.max_attempts: item.status = "failed" item.error = item.error or "Execution lease expired after the maximum retry count" item.completed_at = now + terminalized_expired = True else: item.status = "queued" item.error = "Previous worker lease expired; retrying" + if terminalized_expired: + await self._refresh_batch_status(session, batch, now=now) + counts = await self._counts(session, batch.id) live = counts["queued"] + counts["leased"] + counts["running"] promote_count = max(0, batch.max_live_items - live) diff --git a/backend/tests/test_subagent_batch_repository.py b/backend/tests/test_subagent_batch_repository.py index 32d07bc607..2c2dc7bd8c 100644 --- a/backend/tests/test_subagent_batch_repository.py +++ b/backend/tests/test_subagent_batch_repository.py @@ -161,6 +161,57 @@ async def test_expired_lease_is_recovered_with_stable_item_identity(tmp_path) -> assert reclaimed[0]["attempt"] == 2 +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("succeed_sibling", "expected_status"), + [(False, "failed"), (True, "completed")], +) +async def test_exhausted_expired_lease_terminalizes_batch(tmp_path, succeed_sibling: bool, expected_status: str) -> None: + repo = await _repo(tmp_path) + count = 2 if succeed_sibling else 1 + await _create(repo, count=count, max_live=count, max_running=count, max_attempts=1) + now = datetime.now(UTC) + claimed = await repo.claim_items(now=now, lease_owner="worker-1", lease_seconds=30, limit=count) + assert len(claimed) == count + + if succeed_sibling: + assert await repo.finalize_item( + claimed[0]["id"], + lease_owner="worker-1", + succeeded=True, + result="done", + result_preview="done", + result_truncated=False, + error=None, + stop_reason=None, + token_usage=None, + model_name=None, + completed_at=now + timedelta(seconds=1), + ) + + assert ( + await repo.claim_items( + now=now + timedelta(seconds=31), + lease_owner="worker-2", + lease_seconds=30, + limit=count, + ) + == [] + ) + batch = await repo.get_batch("batch-1", user_id="user-1") + assert batch is not None + assert batch["counts"]["failed"] == 1 + assert batch["status"] == expected_status + assert batch["completed_at"] is not None + + retried = await repo.retry_item("batch-1", claimed[-1]["id"], user_id="user-1") + assert retried is not None and retried["status"] == "pending" + reopened = await repo.get_batch("batch-1", user_id="user-1") + assert reopened is not None + assert reopened["status"] == "queued" + assert reopened["completed_at"] is None + + @pytest.mark.asyncio async def test_finalize_retries_then_terminalizes_and_completes_batch(tmp_path) -> None: repo = await _repo(tmp_path)