feat(monitoring): count background_work task-granular (expand delegate batches)

A delegate_task fan-out batch occupies ONE async-pool slot by design, so
active_count() (unit/slot count) reports a 3-task batch as 1 — which
undercounts real concurrent subagent load on the background_work metric.
Add active_task_count() that expands a running batch to its child count
(N-task batch -> N, single -> 1) and switch the background_work reader to
it. active_count() is unchanged (capacity semantics preserved). Adds a
contract test for the unit-vs-task distinction and documents both counts.

agent/monitoring/gateway_health_export.py

@@ -312,12 +312,17 @@ def _read_background_work_count() -> int:
     metric a peer churning through delegated subagents shows ``active_agents=0``
     on the fleet dashboard. Best-effort and content-free: a single integer,
     no job/task identity. Returns 0 if a source can't be imported.
+
+    Delegation is counted TASK-granular (``active_task_count``): a fan-out batch
+    of N subagents contributes N, not 1, so the metric reflects real concurrent
+    subagent load rather than dispatch-unit/pool-slot count. This intentionally
+    differs from the async pool's capacity accounting (one batch = one slot).
     """
     total = 0
     try:
-        from tools.async_delegation import active_count
+        from tools.async_delegation import active_task_count
 
-        total += max(0, int(active_count()))
+        total += max(0, int(active_task_count()))
     except Exception:
         logger.debug("background-work async-delegation count failed", exc_info=True)
     try:

docs/observability/monitoring.md

@@ -26,6 +26,17 @@ Signals carry `service.name`, version, supervision mode, and a stable one-way
 hash of the install id so an operator can distinguish instances without
 exporting account/profile identity or the raw install identifier.
 
+`hermes.gateway.active_agents` and `hermes.gateway.background_work` are
+complementary and deliberately disjoint. `active_agents` counts foreground
+message turns plus in-flight cron jobs plus API runs — the work the gateway
+drains on shutdown. `background_work` counts detached work that `active_agents`
+never includes: backgrounded `delegate_task` subagents, `terminal(background=true)`
+processes, and kanban workers. `background_work` is **task-granular** — a
+fan-out batch of N subagents counts as N, not as one dispatch unit — so it
+reflects real concurrent subagent load rather than async-pool slot usage (a
+batch occupies one slot but runs N children). Sum both for total live work per
+instance.
+
 ## Enabling
 
 ```yaml

tests/tools/test_async_delegation.py

@@ -244,6 +244,34 @@ def test_completed_records_pruned_to_cap():
     assert len(ad.list_async_delegations()) <= ad._MAX_RETAINED_COMPLETED
 
 
+def test_active_task_count_expands_batches_while_active_count_stays_unit(monkeypatch):
+    """active_count() counts dispatch UNITS (batch=1); active_task_count()
+    expands a batch to its child count. This is the batch-vs-single distinction
+    the background_work metric relies on so a 3-task fan-out isn't undercounted
+    as 1 running subagent.
+    """
+    # Deterministic: install synthetic running records directly, no real spawn.
+    with ad._records_lock:
+        saved = dict(ad._records)
+        ad._records.clear()
+        ad._records["single_a"] = {"status": "running"}  # single subagent
+        ad._records["batch_3"] = {"status": "running", "is_batch": True,
+                                   "goals": ["g1", "g2", "g3"]}  # 3-task batch
+        ad._records["batch_missing"] = {"status": "running", "is_batch": True}  # goals absent -> 1
+        ad._records["done"] = {"status": "completed", "is_batch": True,
+                                "goals": ["x", "y"]}  # not running -> ignored
+    try:
+        # 3 running UNITS (single + 2 batches); the completed one is excluded.
+        assert ad.active_count() == 3
+        # TASKS: single(1) + batch_3(3) + batch_missing(1, fallback) = 5.
+        assert ad.active_task_count() == 5
+    finally:
+        with ad._records_lock:
+            ad._records.clear()
+            ad._records.update(saved)
+
+
+
 def test_completion_is_persisted_and_delivery_can_be_acknowledged(tmp_path, monkeypatch):
     """A finished child remains pending on disk until its queue consumer acks it."""
     monkeypatch.setenv("HERMES_HOME", str(tmp_path))

tools/async_delegation.py

@@ -473,11 +473,41 @@ def _get_executor(max_workers: int) -> ThreadPoolExecutor:
 
 
 def active_count() -> int:
-    """Number of async delegations currently running."""
+    """Number of async delegation UNITS currently running.
+
+    A unit is one dispatch: a single subagent OR a whole fan-out batch. A batch
+    counts as ONE here because it occupies one async-pool slot (the capacity
+    semantics ``dispatch_async_delegation_batch`` relies on). For the count of
+    actual concurrent child subagents (batch expanded), use
+    ``active_task_count()``.
+    """
     with _records_lock:
         return sum(1 for r in _records.values() if r.get("status") in {"running", "finalizing"})
 
 
+def active_task_count() -> int:
+    """Number of async delegation TASKS (child subagents) currently running.
+
+    Unlike ``active_count()`` (units/slots), this expands a batch to its child
+    count: a running batch of N tasks contributes N, a single subagent
+    contributes 1. This is the truthful "how many subagents are actually
+    working right now" figure for observability, where a 3-task batch shown as
+    "1" undercounts real concurrent work. Falls back to counting a batch as 1
+    if its goal list is missing.
+    """
+    with _records_lock:
+        total = 0
+        for r in _records.values():
+            if r.get("status") not in {"running", "finalizing"}:
+                continue
+            if r.get("is_batch"):
+                goals = r.get("goals")
+                total += len(goals) if isinstance(goals, (list, tuple)) and goals else 1
+            else:
+                total += 1
+        return total
+
+
 def _new_delegation_id() -> str:
     return f"deleg_{uuid.uuid4().hex[:8]}"