mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(subagents): terminalize batches after exhausted leases expire (#5897)
* fix(subagents): terminalize batches after exhausted leases expire * docs(subagents): stay within agent guidance budget
This commit is contained in:
parent
1333c440ef
commit
85acf643c0
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user