feat(flows): support runtime agent fan-out

docs/reference/flows.md

@@ -172,6 +172,50 @@ a time, which is what most of them want.
 Write the flow as a plain `def run` unless it has something to wait for. Both are flows; neither
 is the newer one.
 
+## Agents whose count is learned at runtime
+
+A flow still declares a fixed number of agents, because that is what somebody can configure
+before it starts. One declared agent may be a **template** for a fan-out whose size becomes known
+only after a turn. `spawn` clones that template under names the flow supplies and registers the
+new agents with the running flow:
+
+```python
+import asyncio
+from typing import NamedTuple
+
+from hmz.flows import Agent, flow, spawn
+
+
+class Agents(NamedTuple):
+    triage: Agent
+    expert: Agent
+
+
+@flow
+async def run(agents: Agents, task: str) -> None:
+    specialties = await agents.triage.aturn(task, schema=Specialties)
+    experts = spawn(
+        agents.expert,
+        (f"expert-{at:02d}-{role.slug}" for at, role in enumerate(specialties.roles, 1)),
+    )
+    opinions = await asyncio.gather(
+        *(expert.aturn(role.prompt) for expert, role in zip(experts, specialties.roles, strict=True))
+    )
+```
+
+Each spawned agent inherits the template's backend, model, effort, provider, permission,
+machine and flow Skills. It starts with no sessions, hooks or watchers, and its name must be
+non-empty and unique across the whole run, including flows this one called. Passing no names is
+a valid fan-out of zero.
+
+When called inside a running flow, the agents join that run before `spawn` returns. They appear
+in the terminal, `Runner.agents`, the SDK's `Run.agents`, and the run's cycle; stopping the run
+stops them too. A spawned agent belongs to the flow record that made it and is stopped when that
+record ends, while the agents selected before the run keep their existing reusable lifetime.
+
+Called outside a run, `spawn` is simply a named group of `clone` calls. There is then no run to
+register or clean them up, so whoever drives those agents owns their lifetime.
+
 ## How many agents, and what they are for
 
 The count is checked before the first turn:

docs/reference/sdk.md

@@ -53,7 +53,7 @@ holding.
 
 | | |
 | --- | --- |
-| `agents` | Every agent it drives, the person the flow talks to among them. |
+| `agents` | Every agent it drives, the person the flow talks to and any runtime-spawned agents among them. The tuple grows as the flow adds agents and remains inspectable after it ends. |
 | `running` | Whether the flow is still going. `False` before it is started. |
 | `raised` | Whatever the flow raised, for a run started on a thread and now over. |
 | `run()` | Runs the flow here, until it returns. |

docs/reference/tracing.md

@@ -109,6 +109,7 @@ as it goes — a run that died is a run whose cycle still says what it got to.
 | `event` | Written | Carries |
 | --- | --- | --- |
 | `began` | when the flow starts | `flow`, `task`, `workspace`, whether the flow is `resumable`, the run it was `picked_up` from where there was one, and one entry per agent with its `agent` id, `backend`, `model`, `effort`, `permission`, `provider`, `goals` and whether it was the `person` at the prompt |
+| `spawned` | when a flow adds an agent after it began | the template in `parent` and the new agent's backend, model and configuration; one made by a called flow also names that flow and its record |
 | `opened` | each time an agent opens a session | `agent`, `backend`, `provider`, `session`, the `name` the run gives it and `where` its links are |
 | `called` | when the flow calls another flow | `flow`, `task`, and the `cycle` — the record that call was written to |
 | `returned` | when that call returns, however it ended | `flow` and the same `cycle` |

src/hmz/cycle.py

@@ -51,7 +51,7 @@
 from hmz import backends, home
 
 if TYPE_CHECKING:
-    from collections.abc import Mapping, Sequence
+    from collections.abc import Callable, Mapping, Sequence
 
     from .agents import AgentBase
     from .tracing.profile import Profiler
@@ -578,6 +578,7 @@ def __init__(
         resumable: bool = False,
         picked_up: str = "",
         profile: bool = False,
+        joined: Callable[[tuple[AgentBase, ...]], None] | None = None,
     ) -> None:
         """Opens a cycle, and writes down what it is a run of.
 
@@ -594,6 +595,8 @@ def __init__(
           profile: Whether to sample the programs the agents start while the run goes, so
             that what a turn spent its minutes on is in the run's trace beside the turn. A
             setting of the workspace, asked of it by whoever opens the cycle.
+          joined: What to tell when the flow adds runtime agents, for whoever is watching or
+            controlling the run. The agents have joined the cycle before this is called.
         """
         self._begin(
             home()
@@ -609,6 +612,7 @@ def __init__(
             (workspace or Path.cwd()).resolve(),
             flow,
             agents,
+            joined=joined,
         )
         #: The programs this run starts, sampled while it runs, or None for a run nobody
         #: asked to profile -- which is every run until somebody says otherwise.
@@ -630,6 +634,9 @@ def _begin(
         workspace: Path,
         flow: str,
         agents: Sequence[AgentBase],
+        *,
+        under: Cycle | None = None,
+        joined: Callable[[tuple[AgentBase, ...]], None] | None = None,
     ) -> None:
         """Settles what is written down, and where.
 
@@ -643,13 +650,27 @@ def _begin(
           workspace: Where the run is happening.
           flow: The flow this is a record of, as it was named.
           agents: The agents it is being run with, in the order it takes them.
+          under: The record this one is nested under, or None for the root run.
+          joined: What the root run tells when runtime agents join it.
         """
         self._at = at
         self._journal = journal
         self._writing = (
             threading.Lock()
         )  # sessions open on whichever thread a turn runs on
         self._agents = list(agents)
+        self._agents_lock = threading.Lock()
+        self._spawned: list[AgentBase] = []
+        if under is None:
+            # A called flow has a record of its own, but agents spawned inside it still belong
+            # to the run as a whole. This root inventory is what makes names unique across the
+            # whole tree and lets whoever started the run stop every agent it owns.
+            self._root = self
+            self._run_agents = list(agents)
+            self._run_agents_lock = threading.Lock()
+            self._joined = joined
+        else:
+            self._root = under.root
         #: Every session this run has opened, by the name it was written down under, so that
         #: the links can be made again as the backends go on writing to them.
         self._sessions: dict[str, tuple[str, str]] = {}
@@ -680,6 +701,73 @@ def workspace(self) -> Path:
         """Where this run is happening, which is what its cycles are kept under."""
         return self._where
 
+    @property
+    def agents(self) -> tuple[AgentBase, ...]:
+        """Every agent in this record, including ones the flow spawned at runtime."""
+        with self._agents_lock:
+            return tuple(self._agents)
+
+    @property
+    def run_agents(self) -> tuple[AgentBase, ...]:
+        """Every agent owned by the whole run, including ones spawned in called flows."""
+        root = self._root
+        return root._run_inventory()  # noqa: SLF001 -- root inventory is an internal registry
+
+    @property
+    def root(self) -> Cycle:
+        """The root record shared by this cycle and any nested called flow."""
+        return self._root
+
+    def _run_inventory(self) -> tuple[AgentBase, ...]:
+        with self._run_agents_lock:
+            return tuple(self._run_agents)
+
+    def spawned(self, parent: AgentBase, agents: Sequence[AgentBase]) -> None:
+        """Adds agents made from ``parent`` after the flow's runtime fan-out is known.
+
+        The whole group is admitted before any event is written, so a duplicate name cannot
+        leave half a requested fan-out attached to the run.
+        """
+        made = tuple(agents)
+        if not made:
+            return
+        records = _drove(made)
+        root = self._root
+        with root._run_agents_lock:  # noqa: SLF001 -- root owns the run inventory
+            occupied = {agent.id for agent in root._run_agents}  # noqa: SLF001
+            names = [agent.id for agent in made]
+            collision = next((name for name in names if name in occupied), None)
+            if collision is not None:
+                raise ValueError(f"agent name {collision!r} is already in this run")
+            if len(set(names)) != len(names):
+                raise ValueError("spawned agent names must be unique")
+            root._run_agents.extend(made)  # noqa: SLF001
+            with self._agents_lock:
+                self._agents.extend(made)
+                self._spawned.extend(made)
+                for agent in made:
+                    agent.cycle = self
+
+        # Whoever is driving the run learns about these before spawn() hands them to the
+        # flow, so a stop or a UI watcher cannot race their first turn.
+        if root.joined is not None:
+            root.joined(made)
+        for record in records:
+            self.write("spawned", parent=parent.id, agent=record)
+            if self is not root:
+                root.write(
+                    "spawned",
+                    parent=parent.id,
+                    agent=record,
+                    flow=self._flow,
+                    cycle=self._journal,
+                )
+
+    @property
+    def joined(self) -> Callable[[tuple[AgentBase, ...]], None] | None:
+        """The callback that observes agents joining the root run, if one was supplied."""
+        return self._joined if self is self._root else self._root.joined
+
     def _profiling(self) -> Profiler | None:
         """The sampler this run is profiled by, started, or None where there is none.
 
@@ -732,9 +820,23 @@ def __exit__(
           why: The exception itself, unread.
           traceback: Where it was raised, unread.
         """
-        self._close(kind)
-        for agent in self._agents:
-            agent.cycle = None
+        try:
+            self._close(kind)
+        finally:
+            self._finish_spawned()
+            for agent in self.agents:
+                if agent.cycle is self:
+                    agent.cycle = None
+
+    def _finish_spawned(self) -> None:
+        """Stops and detaches the runtime agents this record created."""
+        with self._agents_lock:
+            spawned = tuple(self._spawned)
+        for agent in spawned:
+            with contextlib.suppress(Exception):
+                agent.stop()
+            if agent.cycle is self:
+                agent.cycle = None
 
     def _close(self, kind: type[BaseException] | None) -> None:
         """Writes down that what this is a record of has ended, and how it ended.
@@ -763,7 +865,7 @@ def _close(self, kind: type[BaseException] | None) -> None:
         # way made of it: the process goes out from under that turn, and from inside one that
         # reads as a turn that could not finish.
         stopped = kind is not None and (
-            issubclass(kind, Stopped) or any(agent.stopped for agent in self._agents)
+            issubclass(kind, Stopped) or any(agent.stopped for agent in self.agents)
         )
         self.write(
             "ended",
@@ -893,7 +995,14 @@ def __init__(
           resumable: Whether it says it can be picked up again.
         """
         self._under = under
-        self._begin(under.path, record, under.workspace, flow, agents)
+        self._begin(
+            under.path,
+            record,
+            under.workspace,
+            flow,
+            agents,
+            under=under,
+        )
         self.write(
             "began",
             flow=flow,
@@ -912,7 +1021,10 @@ def ended(self, kind: type[BaseException] | None = None) -> None:
         Args:
           kind: What was raised out of the called flow, if anything.
         """
-        self._close(kind)
+        try:
+            self._close(kind)
+        finally:
+            self._finish_spawned()
         self._under.write("returned", flow=self._flow, cycle=self._journal)
 
 
@@ -1087,7 +1199,9 @@ def read(cycle: Path) -> Ran | None:
         return None
     ended = next((one for one in reversed(events) if one.get("event") == "ended"), None)
     agents: list[Drove] = []
-    for one in began.get("agents") or ():
+    declared: list[object] = list(cast("list[object]", began.get("agents") or []))
+    declared.extend(one.get("agent") for one in events if one.get("event") == "spawned")
+    for one in declared:
         if not isinstance(one, dict):
             continue
         said = cast("dict[str, Any]", one)

src/hmz/flows/__init__.py

@@ -79,6 +79,7 @@ def run(agents: tuple[Agent, Agent], task: str) -> None:
     load,
     resumes,
     running,
+    spawn,
     wanted,
 )
 from .verses import (
@@ -244,6 +245,7 @@ def run(agents: tuple[Agent, Agent], task: str) -> None:
     "resumes",
     "running",
     "shipped",
+    "spawn",
     "sub",
     "told",
     "wanted",

src/hmz/flows/driving.py

@@ -39,7 +39,7 @@
 from hmz import telemetry
 
 if TYPE_CHECKING:
-    from collections.abc import Awaitable, Callable, Generator, Sequence
+    from collections.abc import Awaitable, Callable, Generator, Iterable, Sequence
 
     from pydantic import BaseModel
 
@@ -72,6 +72,7 @@
     "resumes",
     "running",
     "set_up",
+    "spawn",
     "wanted",
 ]
 
@@ -447,6 +448,68 @@ def wanted(flow: str | os.PathLike[str]) -> tuple[Place, ...]:
     return tuple(place for place in declares(flow)[1] if not place.person)
 
 
+def spawn(template: Agent, names: Iterable[str]) -> tuple[Agent, ...]:
+    """Makes runtime agents from one configured template and joins them to its run.
+
+    A flow declares the agents it needs before it starts. Some work only reveals its fan-out
+    after a turn has landed, such as a triage agent choosing which specialists are needed. The
+    flow declares one template for that role, then expands it once the work is known::
+
+        experts = spawn(agents.expert, (f"expert-{at + 1}" for at in range(len(tasks))))
+        await asyncio.gather(
+            *(expert.aturn(task) for expert, task in zip(experts, tasks, strict=True))
+        )
+
+    Each result is a distinct agent with the template's backend, configuration, machine and
+    skills. When the template belongs to a running flow, the new agents join that same cycle:
+    their sessions are traced under their own names and stopping the run stops them too.
+
+    Args:
+      template: The agent whose settled configuration each new agent inherits.
+      names: One non-empty, unique name per agent to create. The iterable's length is the
+        runtime fan-out.
+
+    Returns:
+      The new agents, in the same order as ``names``.
+
+    Raises:
+      ValueError: If a name is empty or repeated, or collides with an agent already in the run.
+    """
+    requested = tuple(str(name).strip() for name in names)
+    if any(not name for name in requested):
+        raise ValueError("spawned agent names must not be empty")
+    if len(set(requested)) != len(requested):
+        raise ValueError("spawned agent names must be unique")
+
+    creating: list[Agent] = []
+    try:
+        for name in requested:
+            creating.append(template.clone(name=name))  # noqa: PERF401 -- cleanup needs partial clones
+    except BaseException:
+        for agent in creating:
+            with contextlib.suppress(Exception):
+                agent.stop()
+        raise
+    made = tuple(creating)
+    if template.cycle is None:
+        return made
+    joins = getattr(template.cycle, "spawned", None)
+    if callable(joins):
+        try:
+            joins(template, made)
+        except BaseException:
+            for agent in made:
+                with contextlib.suppress(Exception):
+                    agent.stop()
+            raise
+    else:
+        # A flow may be driven under a journal of its own. It can still trace the sessions even
+        # when that journal predates runtime agent registration.
+        for agent in made:
+            agent.cycle = template.cycle
+    return made
+
+
 def declares(
     flow: str | os.PathLike[str],
 ) -> tuple[