diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 2178a019a9..151e08dc32 100644 --- a/hermes_cli/config_defaults.py +++ b/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" diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py index 648fc0e260..361e5df93f 100644 --- a/tests/tools/test_async_delegation.py +++ b/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.""" diff --git a/tools/AGENTS.md b/tools/AGENTS.md index fa624f2bb7..83e891dddb 100644 --- a/tools/AGENTS.md +++ b/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`. diff --git a/tools/async_delegation.py b/tools/async_delegation.py index 99330e84f7..7a46d0ca7c 100644 --- a/tools/async_delegation.py +++ b/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: diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 8ecb8d539b..718373be0d 100644 --- a/tools/delegate_tool.py +++ b/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"], diff --git a/tools/delegate_tool_config.py b/tools/delegate_tool_config.py index 0d1eb1b2de..98603368d0 100644 --- a/tools/delegate_tool_config.py +++ b/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 diff --git a/tools/delegate_tool_dispatch.py b/tools/delegate_tool_dispatch.py index 09f8a42da4..81f4c30412 100644 --- a/tools/delegate_tool_dispatch.py +++ b/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") diff --git a/tools/process_registry_notifications.py b/tools/process_registry_notifications.py index 0e7f82f5e4..867155080b 100644 --- a/tools/process_registry_notifications.py +++ b/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: diff --git a/website/docs/user-guide/features/delegation.md b/website/docs/user-guide/features/delegation.md index c4cca38970..9e66ca8f77 100644 --- a/website/docs/user-guide/features/delegation.md +++ b/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.