fix(delegation): one completion per call by default; queued units no longer stalled; tell the model results land between turns

Three orchestrator failures traced through the Sep 7 gpt-6-astra campaign sessions:

1. delegation.independent_completions (new, default false). #104299 made every
   ungrouped task its own completion message, so a 15-task call woke the
   orchestrator up to 15 times; one chain received 132 notices and answered
   130 of them with "already incorporated". A multi-task call now returns as
   ONE consolidated message unless the flag is on; `group` is inert until then.

2. Queued units were killed before they started. Units of one call share a
   pool slot but the executor was still sized by slots, so with 15 units live
   a new unit queued behind a full pool; the stale monitor's clock ran from
   dispatch, interrupted it at 450 s, and the child exited `interrupted 0.02s`
   when its thread finally came up (13 such lanes in one session). The
   executor now grows to the number of live units and the stall clock arms
   when the runner actually starts.

3. The tool text said "do not wait or poll — just continue" without saying
   that completions are delivered only BETWEEN turns. A model that never ends
   its turn (one 203-minute turn, 717 API calls) never received 40 finished
   results. Tool description, dispatch note and completion header now say to
   finish independent work, give a one-line status, and end the turn.

hermes_cli/config_defaults.py

@@ -1246,6 +1246,9 @@ DEFAULT_CONFIG = {
         # Max parallel children per batch AND max concurrent background delegation units; async
         # dispatches beyond it run synchronously. Floor 1, no ceiling.
         "max_concurrent_children": 10,
+        # Background fan-outs return as ONE message when the whole call finishes. true = each task
+        # (or `group`) returns on its own as it finishes — more new turns for the orchestrator.
+        "independent_completions": False,
         # Orchestrator role controls. Depth floored at 1, no ceiling; each level multiplies cost.
         "max_spawn_depth": 1,  # 1 = flat, 2 = orchestrator→leaf, 3+ = deeper
         "orchestrator_enabled": True,  # kill switch for role="orchestrator"

tests/tools/test_async_delegation.py

@@ -985,7 +985,10 @@ def _grouped_fanout(monkeypatch, tasks, gates):
 
 
 def test_ungrouped_task_completes_alone_and_group_completes_together(monkeypatch):
-    """A finished ungrouped task must not wait for its siblings; tasks sharing a `group` must."""
+    """With delegation.independent_completions on, a finished ungrouped task must not wait for its siblings;
+    tasks sharing a `group` must."""
+    import tools.delegate_tool as dt
+    monkeypatch.setattr(dt, "_load_config", lambda: {"independent_completions": True})
     gates = [threading.Event() for _ in range(4)]
     tasks = [
         {"goal": "review PR 1 thoroughly and report"},
@@ -1032,6 +1035,57 @@ def test_units_of_one_call_share_a_single_capacity_slot():
     gate.set()
 
 
+def test_multi_task_call_is_one_completion_unless_independent_completions(monkeypatch):
+    """Default: a background fan-out returns as ONE message when every task is done, so an orchestrator
+    is not woken N times per call; `group` is inert until delegation.independent_completions is on."""
+    import tools.delegate_tool as dt
+    monkeypatch.setattr(dt, "_load_config", lambda: {})
+    gates = [threading.Event() for _ in range(3)]
+    tasks = [{"goal": "review PR 1 thoroughly and report"}, {"goal": "review PR 2 thoroughly and report", "group": "g"},
+             {"goal": "review PR 3 thoroughly and report", "group": "g"}]
+    handle = _grouped_fanout(monkeypatch, tasks, gates)
+    assert handle["status"] == "dispatched" and "units" not in handle
+    gates[0].set()
+    gates[1].set()
+    assert _drain_one(timeout=0.5) is None  # two of three done: no message yet
+    gates[2].set()
+    evt = _drain_one()
+    assert sorted(r["task_index"] for r in evt["results"]) == [0, 1, 2]
+
+
+def test_units_beyond_slot_count_still_start_and_are_not_stalled_while_queued(monkeypatch):
+    """Units of one call share a slot, so live units can exceed the slot cap; every unit must still get a worker,
+    and a unit must not be judged stalled for time it spent waiting to start."""
+    _fast_stale_monitor(monkeypatch, idle=0.3, grace=0.2)
+    started, release = [], threading.Event()
+    frozen = lambda: (((0, None, None),), False)  # noqa: E731 - child never progresses => token never changes
+
+    def blocker(uid):
+        def run():
+            started.append(uid)
+            release.wait(timeout=10)
+            return {"results": [{"task_index": 0, "status": "completed"}], "total_duration_seconds": 0}
+        return run
+
+    common = dict(goals=["x"], context=None, toolsets=None, role="leaf", model="m", session_key="", max_async_children=1)
+    ad.dispatch_async_delegation_batch(delegation_id="deleg_c-1", runner=blocker("c-1"), progress_fn=frozen, **common)
+    ad.dispatch_async_delegation_batch(delegation_id="deleg_c-2", runner=blocker("c-2"), slot_key="deleg_c-1", **common)
+    deadline = time.monotonic() + 2.0
+    while len(started) < 2 and time.monotonic() < deadline:
+        time.sleep(0.02)
+    assert sorted(started) == ["c-1", "c-2"]  # second unit started despite a 1-slot pool
+
+    # A unit whose runner has NOT started yet must not accrue stall time: pin the pool so it stays queued.
+    monkeypatch.setattr(ad, "_get_executor", lambda n: ad._executor)
+    ad.dispatch_async_delegation_batch(delegation_id="deleg_q", runner=blocker("q"), progress_fn=frozen,
+                                       **{**common, "max_async_children": 3})
+    time.sleep(0.7)  # > idle + grace with the unit still queued
+    with ad._records_lock:
+        assert ad._records["deleg_q"]["status"] == "running"
+    assert "q" not in started
+    release.set()
+
+
 def test_child_finished_before_crash_is_recovered_with_its_result(tmp_path):
     """Real-import E2E: a 2-task group unit whose owner dies mid-run replays the finished child's real result
     and marks only the unfinished sibling unknown — a crash costs the stragglers, never the finished work."""

tools/AGENTS.md

@@ -79,14 +79,16 @@ fixed at the mount, not by adding a tool.
 Spawns a subagent with isolated context + terminal session; the parent waits for the summary unless
 `background=true`, which returns a delegation id and re-enters the result via the async-delegation
 completion queue. Shapes: single (`goal` + optional `context`, `toolsets`) or batch (`tasks: [...]`,
-concurrency capped by `delegation.max_concurrent_children`, default 3). A background batch is split
-into completion **units** (`delegate_tool_dispatch._units_of`): tasks sharing a `group` join and
-report together; each ungrouped task reports alone as it finishes. Units of one call share ONE pool
-slot (`slot_key` in `async_delegation._dispatch`) — never count units against capacity. Roles: `leaf` (default;
+concurrency capped by `delegation.max_concurrent_children`, default 3). A background batch returns as ONE
+completion by default; with `delegation.independent_completions` it is split into completion **units**
+(`delegate_tool_dispatch._units_of`): tasks sharing a `group` join and report together; each ungrouped
+task reports alone as it finishes. Units of one call share ONE pool slot (`slot_key` in
+`async_delegation._dispatch`) — never count units against capacity; the executor is sized by live UNITS
+and the stall clock arms when the runner starts, so a queued unit is never judged stalled. Roles: `leaf` (default;
 no `delegate_task`, `clarify`, `memory`, `send_message`, `cronjob`; keeps `execute_code`) and
 `orchestrator` (keeps `delegate_task`; gated by `delegation.orchestrator_enabled`, bounded by
 `delegation.max_spawn_depth`, default 2). Config knobs under `delegation:`:
-`max_concurrent_children, max_spawn_depth, child_timeout_seconds, orchestrator_enabled,
+`max_concurrent_children, independent_completions, max_spawn_depth, child_timeout_seconds, orchestrator_enabled,
 subagent_auto_approve, inherit_mcp_toolsets, max_iterations`. **Durability:** background
 delegation is process-local; work that must survive restart uses `cronjob` or
 `terminal(background=True, notify_on_complete=True)`. API: `website/docs/developer-guide/subagent-lifecycle-api.md`.

tools/async_delegation.py

@@ -451,13 +451,16 @@ def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]:
 
 # ── In-memory registry queries ──────────────────────────────────────────────
 def _get_executor(max_workers: int) -> ThreadPoolExecutor:
-    """Lazily create (or grow, never shrink) the shared daemon executor; in-flight
-    futures keep running on a replaced pool until it is collected."""
+    """Lazily create (or grow in place, never shrink) the shared daemon executor. Raising
+    ``_max_workers`` is enough: the next ``submit`` spawns threads up to the new cap."""
     global _executor, _executor_max_workers
     with _executor_lock:
-        if _executor is None or max_workers > _executor_max_workers:
+        if _executor is None:
             _executor = DaemonThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="async-delegate")
             _executor_max_workers = max_workers
+        elif max_workers > _executor_max_workers:
+            _executor._max_workers = max_workers
+            _executor_max_workers = max_workers
         return _executor
 
 
@@ -576,12 +579,20 @@ def _dispatch(
         if record["slot_key"] not in active_slots and len(active_slots) >= max_async_children:
             return {"status": "rejected", "error": capacity_error}
         _records[delegation_id] = record
+        live_units = sum(1 for r in _records.values() if r.get("status") in _LIVE_STATES)
     _persist_dispatch(record)
-    executor = _get_executor(max_async_children)
+    # Units of one call share a slot, so live units can exceed slots: size the pool by units or a
+    # unit queues behind a full pool and the stale monitor kills it before its child ever starts.
+    executor = _get_executor(max(max_async_children, live_units))
 
     def _worker() -> None:
         result: Dict[str, Any] = {}
         status = "error"
+        with _records_lock:
+            rec = _records.get(delegation_id)
+            if rec is not None:
+                # The stall clock starts when the runner starts; a unit queued behind a full pool is not stalled.
+                rec.update(_started=True, _progress_ts=time.time())
         try:
             result = runner() or {}
             status = classify(result)
@@ -804,6 +815,8 @@ def _sweep_stale_locked(now: float):
         if status != "running" or progress_fn is None:
             continue
         any_monitorable = True
+        if not record.get("_started"):
+            continue  # queued behind a full pool: not stalled, but keep the monitor alive for when it starts
         try:
             token, in_tool = progress_fn()
         except Exception:

tools/delegate_tool.py

@@ -507,10 +507,11 @@ _DESCRIPTION_HEAD = (
     "Spawn subagents in isolated contexts; each gets its own conversation, terminal session, and toolset, and only its "
     "final summary returns to you. Pass every task in `tasks` — one entry spawns one subagent, several run in parallel "
     "(limit in the tasks description).\n\n"
-    "Runs in the background: dispatch returns immediately with live transcript paths, and each completed unit "
-    "re-enters the conversation on its own — an ungrouped task as soon as IT finishes, tasks sharing a `group` "
-    "together once all of them finish. Choose groups by how you want to use the results (see `group`). "
-    "No need to wait or poll; continue other work while results are pending. "
+    "Runs in the background: dispatch returns immediately with live transcript paths, and the call's results re-enter "
+    "the conversation as a new message when its subagents finish (one message per call by default; with "
+    "delegation.independent_completions each ungrouped task / `group` returns on its own). Results are delivered only "
+    "BETWEEN your turns: finish whatever does not depend on them, then give a one-line status and END YOUR TURN. Never "
+    "poll transcripts, artifact files, or CI to wait for a child. "
     "While children run, `action` (list/steer/stop) controls them live — steer when a transcript shows a "
     "child drifting.\n\n"
     "USE FOR: reasoning-heavy subtasks, work that would flood your context with intermediate data, or independent "
@@ -600,12 +601,10 @@ DELEGATE_TASK_SCHEMA = {
                         ),
                         "group": _p(
                             "string",
-                            "Optional result-delivery bucket within this call; all tasks still run in parallel. "
-                            "Use the same group when you want to review results together (comparison, synthesis, "
-                            "or one coordinated decision): ONE message arrives after all tasks in that group finish. "
-                            "Omit when each result is useful to act on separately: it arrives as soon as that task "
-                            "finishes. Different groups report independently. This does not order execution; if B "
-                            "needs A's output, dispatch B after A returns.",
+                            "Optional result-delivery bucket within this call (only when delegation.independent_completions "
+                            "is enabled; otherwise the whole call returns as one message). Tasks sharing a group return "
+                            "together in ONE message; ungrouped tasks return individually as each finishes. This does not "
+                            "order execution; if B needs A's output, dispatch B after A returns.",
                         ),
                     },
                     "required": ["goal"],

tools/delegate_tool_config.py

@@ -98,6 +98,11 @@ def _get_max_concurrent_children() -> int:
         )
     return result
 
+def _get_independent_completions() -> bool:
+    """delegation.independent_completions (bool, default False): split a background call into per-task / per-group
+    completion messages that land as each finishes. Off = one consolidated message when the whole call is done."""
+    return is_truthy_value(_cfg().get("independent_completions", False))
+
 def _get_worktree_isolation() -> bool:
     """delegation.worktree_isolation (bool, default False): each child gets its own
     git worktree off the parent's HEAD so parallel children never contend for one

tools/delegate_tool_dispatch.py

@@ -274,14 +274,15 @@ def _batch_progress_token(child_agents: List[Any]) -> tuple:
 
 _BACKGROUND_NOTES = {
     "one": (
-        "Subagent is running in the background. You and the user can keep working; its full result re-enters the "
-        "conversation as a new message when it finishes. Do not wait or poll — just continue."
+        "Subagent is running in the background; its full result re-enters the conversation as a new message when it "
+        "finishes. Results are delivered only after you END YOUR TURN: do anything that does not depend on it, then "
+        "stop with a one-line status. Do not poll its transcript or artifacts to wait for it."
     ),
     "many": (
-        "{n} subagents are running in parallel in the background as {k} independent unit(s). You and the user can keep "
-        "working; each unit's results re-enter the conversation as their own message as soon as THAT unit finishes "
-        "(tasks sharing a `group` finish together; ungrouped tasks report individually), so act on each as it lands. "
-        "Do not wait or poll — just continue."
+        "{n} subagents are running in parallel in the background as {k} completion unit(s); each unit's results "
+        "re-enter the conversation as their own new message when THAT unit finishes. Results are delivered only "
+        "after you END YOUR TURN: do anything that does not depend on them, then stop with a one-line status. Do not "
+        "poll transcripts or artifacts to wait for them."
     ),
     "control_hint": (
         "While a child runs you can orchestrate it live with this same tool: delegate_task(action='list') to see live "
@@ -320,7 +321,13 @@ def _dispatched_payload(batch: _Batch, units: List[tuple[_Batch, str]]) -> dict:
 def _units_of(batch: _Batch) -> List[_Batch]:
     """Partition the call's children into async units: one per distinct task ``group`` (first-appearance order) and
     one per ungrouped task. Each unit is a ``_Batch`` sharing the call's task_list/transcripts but owning a subset of
-    ``children``, so a unit joins only on itself and its completion re-enters the conversation on its own."""
+    ``children``, so a unit joins only on itself and its completion re-enters the conversation on its own.
+
+    Off by default (``delegation.independent_completions``): the whole call is ONE unit and returns as one message.
+    A per-task flurry of completions (one new turn each) fragmented orchestrators that had no plan for it."""
+    from tools.delegate_tool_config import _get_independent_completions
+    if not _get_independent_completions():
+        return [batch]
     members: Dict[Any, List[tuple]] = {}
     for i, t, c in batch.children:
         g = t.get("group")

tools/process_registry_notifications.py

@@ -143,8 +143,9 @@ def _format_batch_delegation(evt: dict, deleg_id: str, completed_at: float) -> s
         evt,
         f"[ASYNC DELEGATION BATCH COMPLETE — {deleg_id}]",
         f"A background fan-out unit you dispatched earlier — {unit} — has finished; its consolidated results are "
-        "below. Other units from the same delegate_task call (other groups / ungrouped tasks) report separately as "
-        "they finish. You may have moved on since dispatching — act on these or re-dispatch if things have changed.",
+        "below. Any other units from the same delegate_task call report separately as they finish. You may have "
+        "moved on since dispatching — act on these or re-dispatch if things have changed. If you are still waiting "
+        "on siblings, end your turn after acting on this one.",
         completed_at, with_goal=False)
     lines[-1] += f"   Total duration: {evt.get('total_duration_seconds', evt.get('duration_seconds', '?'))}s"
     if evt.get("error") and not results:

website/docs/user-guide/features/delegation.md

@@ -123,13 +123,17 @@ delegate_task(
 
 ## Batch Mode Details
 
-When a top-level agent provides a `tasks` array, Hermes returns one background handle and runs the subagents in parallel. Results come back **per completion unit**, not once at the end:
+When a top-level agent provides a `tasks` array, Hermes returns one background handle and runs the subagents in parallel. By default the call returns **one** consolidated message once every task has finished. Results are delivered only between the parent's turns: the parent should finish anything that does not depend on the children, then end its turn rather than polling transcripts, artifacts, or CI while it waits.
+
+### Independent completions (opt-in)
+
+Set `delegation.independent_completions: true` to have results land **per completion unit** as each finishes instead:
 
 - Omit `group` when each result is useful to act on separately. Each task reports as soon as it finishes.
 - Use the same `group` string when you want to review outputs together: comparison, synthesis, or one coordinated decision. The group returns **one** consolidated message after all its tasks finish. Even independently executable tasks can belong in one group when their results inform the same decision.
 - Different groups report independently; grouped and ungrouped tasks can share one call.
 
-Grouping controls **result delivery, not execution order**: all tasks still run in parallel. If task B needs task A's output to do its work, dispatch A first, then dispatch B with that output after A returns.
+This is off by default because every unit is a new turn for the orchestrator: a 15-task call becomes up to 15 wake-ups, which fragmented long campaigns. Grouping controls **result delivery, not execution order**: all tasks still run in parallel. If task B needs task A's output to do its work, dispatch A first, then dispatch B with that output after A returns.
 
 ```json
 {"tasks": [
@@ -140,7 +144,7 @@ Grouping controls **result delivery, not execution order**: all tasks still run
 ]}
 ```
 
-The dispatch handle lists each unit (`units[].delegation_id`, `group`, `task_indexes`); unit ids are the call's id suffixed `-1`, `-2`, …, and every unit of one call shares a single slot of `delegation.max_concurrent_children`, so grouping never changes capacity accounting. An orchestrator subagent waits for its whole batch in the current turn so it can synthesize the results.
+The dispatch handle lists each unit (`units[].delegation_id`, `group`, `task_indexes`); unit ids are the call's id suffixed `-1`, `-2`, …, and every unit of one call shares a single slot of `delegation.max_concurrent_children`, so grouping never changes capacity accounting (the worker pool grows to the number of live units so no unit waits behind a full pool). An orchestrator subagent waits for its whole batch in the current turn so it can synthesize the results.
 
 - **Maximum concurrency:** 3 tasks by default (configurable via `delegation.max_concurrent_children` or the `DELEGATION_MAX_CONCURRENT_CHILDREN` env var; floor of 1, no hard ceiling). Batches larger than the limit return a tool error rather than being silently truncated.
 - **Thread pool:** Uses `ThreadPoolExecutor` with the configured concurrency limit as max workers
@@ -532,6 +536,7 @@ error.
 delegation:
   max_iterations: 50                        # Max turns per child (default: 50)
   # max_concurrent_children: 3              # Parallel children per batch (default: 3)
+  # independent_completions: false          # true = each task/group returns as it finishes (default: one message per call)
   # worktree_isolation: false               # Give each child its own git worktree (see Worktree Isolation above)
   # max_spawn_depth: 1                      # Tree depth (floor 1, no ceiling, default 1 = flat). Raise to 2 to allow orchestrator children to spawn leaves; 3+ for deeper trees.
   # orchestrator_enabled: true              # Disable to force all children to leaf role.