fix(subagents): report an explicit zero batch limit instead of defaulting it (#5609)

* fix(subagents): report an explicit zero batch limit instead of defaulting it

SubagentBatchService.submit resolved max_live_items / max_running_items with
`or`, so a caller that explicitly passed 0 got the configured default
(100 / 3) persisted and the 1..N range guards never saw the value, while a
negative already failed there. Resolve the defaults on 'is None' so an
explicit 0 reaches the guard that names it.

* docs(subagents): state the >= 1 window constraint in the batch_task args

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/packages/harness/deerflow/subagents/batch_service.py

@@ -136,8 +136,11 @@ class SubagentBatchService:
         total = len(request.items)
         if total < 1 or total > self._config.max_items_per_batch:
             raise ValueError(f"Batch item count must be between 1 and {self._config.max_items_per_batch}")
-        max_live = request.max_live_items or self._config.default_max_live_items
-        max_running = request.max_running_items or self._config.default_max_running_items
+        # `or` would read an explicit 0 as "unset", substitute the configured
+        # default, and hide the caller's value from the range guards below --
+        # a negative already fails there, so 0 was the asymmetric case.
+        max_live = self._config.default_max_live_items if request.max_live_items is None else request.max_live_items
+        max_running = self._config.default_max_running_items if request.max_running_items is None else request.max_running_items
         if not 1 <= max_live <= self._config.max_live_items_per_batch:
             raise ValueError(f"max_live_items must be between 1 and {self._config.max_live_items_per_batch}")
         if not 1 <= max_running <= self._config.max_running_items_per_batch:

backend/packages/harness/deerflow/tools/builtins/batch_task_tool.py

@@ -162,8 +162,8 @@ async def batch_task(
         title: Short batch name shown to the user.
         items: Stable item keys, self-contained prompts, and optional per-item acceptance_criteria.
         subagent_type: Native subagent definition used for every item.
-        max_live_items: Optional queued-plus-running item window.
-        max_running_items: Optional per-batch real execution concurrency.
+        max_live_items: Optional queued-plus-running item window; when set it must be >= 1.
+        max_running_items: Optional per-batch real execution concurrency; when set it must be >= 1.
     """
     submitter = _batch_submitter()
     if submitter is None:

backend/tests/test_subagent_batch_service.py

@@ -69,6 +69,52 @@ async def test_submit_keeps_batch_running_limit_separate_from_one_process_capaci
     assert repository.create_batch.await_args.kwargs["max_running_items"] == 10
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("overrides", "expected"),
+    [
+        ({"max_running_items": 0}, "max_running_items must be between 1 and 64"),
+        ({"max_live_items": 1, "max_running_items": 0}, "max_running_items must be between 1 and 64"),
+        ({"max_live_items": 0}, "max_live_items must be between 1 and 1000"),
+    ],
+)
+async def test_submit_reports_an_explicit_zero_limit_by_name(overrides: dict[str, int], expected: str) -> None:
+    repository = SimpleNamespace(create_batch=AsyncMock(return_value={"id": "batch-1"}))
+    service = SubagentBatchService(
+        repository=repository,
+        config=SubagentBatchesConfig(),
+        runtime_config=SubagentRuntimeConfig(),
+    )
+
+    with pytest.raises(ValueError, match=expected):
+        await service.submit(_request(**overrides))
+
+    repository.create_batch.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("overrides", "live", "running"),
+    [
+        ({}, 100, 3),
+        ({"max_live_items": 40}, 40, 3),
+        ({"max_live_items": 40, "max_running_items": 5}, 40, 5),
+    ],
+)
+async def test_submit_defaults_only_the_limits_the_caller_omitted(overrides: dict[str, int], live: int, running: int) -> None:
+    repository = SimpleNamespace(create_batch=AsyncMock(return_value={"id": "batch-1"}))
+    service = SubagentBatchService(
+        repository=repository,
+        config=SubagentBatchesConfig(),
+        runtime_config=SubagentRuntimeConfig(),
+    )
+
+    await service.submit(_request(**overrides))
+
+    kwargs = repository.create_batch.await_args.kwargs
+    assert (kwargs["max_live_items"], kwargs["max_running_items"]) == (live, running)
+
+
 @pytest.mark.asyncio
 async def test_execute_item_marks_real_running_then_persists_terminal_result(monkeypatch) -> None:
     result = SimpleNamespace(