From 145773c564a554dd2941d9ad1e90b4dc5a4d5fb4 Mon Sep 17 00:00:00 2001 From: antoinegg1 <3528942762@qq.com> Date: Tue, 1 Sep 2026 09:39:44 +0800 Subject: [PATCH 1/2] feat(flows): support runtime agent fan-out --- docs/reference/flows.md | 44 +++++++ docs/reference/sdk.md | 2 +- docs/reference/tracing.md | 1 + src/hmz/cycle.py | 130 ++++++++++++++++++-- src/hmz/flows/__init__.py | 2 + src/hmz/flows/driving.py | 65 +++++++++- src/hmz/runner.py | 52 +++++++- src/hmz/sdk/running.py | 3 +- src/hmz/tui/app.py | 42 ++++++- src/hmz/tui/tally.py | 10 +- tests/test_dynamic_agents.py | 223 +++++++++++++++++++++++++++++++++++ 11 files changed, 554 insertions(+), 20 deletions(-) create mode 100644 tests/test_dynamic_agents.py diff --git a/docs/reference/flows.md b/docs/reference/flows.md index 3b786ef..5de6513 100644 --- a/docs/reference/flows.md +++ b/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: diff --git a/docs/reference/sdk.md b/docs/reference/sdk.md index 6f13d2b..43e35dd 100644 --- a/docs/reference/sdk.md +++ b/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. | diff --git a/docs/reference/tracing.md b/docs/reference/tracing.md index 00ef13e..2388b68 100644 --- a/docs/reference/tracing.md +++ b/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` | diff --git a/src/hmz/cycle.py b/src/hmz/cycle.py index ab8742c..62bb7c4 100644 --- a/src/hmz/cycle.py +++ b/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,6 +650,8 @@ 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 @@ -650,6 +659,18 @@ def _begin( 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) diff --git a/src/hmz/flows/__init__.py b/src/hmz/flows/__init__.py index ef84811..40a7ad3 100644 --- a/src/hmz/flows/__init__.py +++ b/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", diff --git a/src/hmz/flows/driving.py b/src/hmz/flows/driving.py index 05b6482..0f174d3 100644 --- a/src/hmz/flows/driving.py +++ b/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[ diff --git a/src/hmz/runner.py b/src/hmz/runner.py index c1cfe2a..73b6b7d 100644 --- a/src/hmz/runner.py +++ b/src/hmz/runner.py @@ -13,6 +13,8 @@ from __future__ import annotations +import contextlib +import threading from pathlib import Path from typing import TYPE_CHECKING, Any, cast @@ -20,7 +22,7 @@ if TYPE_CHECKING: import os - from collections.abc import Awaitable, Sequence + from collections.abc import Awaitable, Callable, Sequence from pydantic import BaseModel @@ -167,6 +169,10 @@ def __init__( # whoever started the flow reaches for: the person the flow talks to is among them, # having been made here rather than chosen. self._driven = tuple(driven) + self._agents_lock = threading.Lock() + self._all_agents = list(self._driven) + self._agent_watchers: list[Callable[[AgentBase], None]] = [] + self._stopping = False # And the same agents as the flow declared them: a flow whose agents are a NamedTuple # reaches them by name, and one that unpacks a plain tuple sees no difference. self._agents = make(driven) @@ -195,7 +201,42 @@ def agents(self) -> tuple[AgentBase, ...]: one more agent than anybody chose, and whatever is driving the flow has to reach that one too -- it is the one thing here that answers with what was typed. """ - return self._driven + with self._agents_lock: + return tuple(self._all_agents) + + def watch_agents(self, listener: Callable[[AgentBase], None]) -> None: + """Calls ``listener`` for each agent the flow adds while it is running. + + The agents declared by the flow are already available through :attr:`agents`; this + watches only agents made later with :func:`hmz.flows.spawn`. + """ + with self._agents_lock: + self._agent_watchers.append(listener) + + def _joined(self, agents: tuple[AgentBase, ...]) -> None: + """Makes newly spawned agents visible to controllers of this runner.""" + with self._agents_lock: + self._all_agents.extend(agents) + listeners = tuple(self._agent_watchers) + stopping = self._stopping + for agent in agents: + for listener in listeners: + # Watching a run is observational. A broken display callback must not turn a + # successfully admitted agent into a failed flow. + with contextlib.suppress(Exception): + listener(agent) + if stopping: + with contextlib.suppress(Exception): + agent.stop() + + def stop(self) -> None: + """Stops every agent in this run, including ones it adds concurrently.""" + with self._agents_lock: + self._stopping = True + agents = tuple(self._all_agents) + for agent in agents: + with contextlib.suppress(Exception): + agent.stop() def run(self, task: str) -> None: """Runs the flow in this directory, for as long as it keeps running. @@ -223,6 +264,8 @@ def run(self, task: str) -> None: # rather than a flow it was told about and a flow it was not. started = entered(self._flow, self._driven) picked_up = self._picked_up + with self._agents_lock: + self._all_agents = list(self._driven) try: # One container for the run, started here rather than where the runner was made: # reading a flow must not pull an image, and a run that never starts must not @@ -242,6 +285,7 @@ def run(self, task: str) -> None: # minute. Read here rather than in the cycle, which is the run written down # rather than the settings under it. profile=Settings().profiling, + joined=self._joined, ) as cycle, ): for agent in self._driven: @@ -273,6 +317,10 @@ def run(self, task: str) -> None: _finished(running_now) finally: left(started) + with self._agents_lock: + # A Runner may be used for another run after this one unwinds. Keep the + # stop request scoped to this run; the next run can admit fresh clones. + self._stopping = False def read_agent( diff --git a/src/hmz/sdk/running.py b/src/hmz/sdk/running.py index 2f9dad9..88b2d97 100644 --- a/src/hmz/sdk/running.py +++ b/src/hmz/sdk/running.py @@ -110,8 +110,7 @@ def stop(self) -> None: The turn running now is closed out first: a flow told to stop unwinds in its own time. :meth:`close` is what does not wait for it. """ - for agent in self.agents: - agent.stop() + self._runner.stop() def close(self) -> None: """Closes every conversation still open, which is the backend's process going. diff --git a/src/hmz/tui/app.py b/src/hmz/tui/app.py index f43fd97..64cad97 100644 --- a/src/hmz/tui/app.py +++ b/src/hmz/tui/app.py @@ -103,6 +103,7 @@ from hmz.agents import AgentBase, Board, Event, Question, SessionBase from hmz.flows import Place + from hmz.runner import Runner from hmz.sdk import Session #: What the editor understands, named as opencode names them, one step along: what answers @@ -844,6 +845,9 @@ def __init__( self._session = session #: The agents of the flow running now, which is who a typed line is said to. self._agents: list[AgentBase] = [] + #: The controller of that run. Unlike an agent list snapshot, this also catches an + #: agent registered at the same moment the run is being stopped. + self._runner: Runner | None = None #: What the flow has done so far, which is what the right-hand column shows, and who #: reads the agents' own logs into it while it runs. self._monitor = Monitor() @@ -2251,8 +2255,11 @@ def action_stop_flow(self) -> None: kept as the ones stopping, since a flow unwinds in its own time and the press after this one is the one that does not wait for it. Silent when nothing is running. """ - for agent in self._agents: - agent.stop() + if self._runner is not None: + self._runner.stop() + else: + for agent in self._agents: + agent.stop() if self._agents: self.show("[dim]— stopping the flow —[/dim]") # Held by identity, so that the run's own thread can say when it has finished @@ -2269,8 +2276,12 @@ def on_unmount(self) -> None: waiting on a prompt that is not there, holding a backend open behind it. Said to nobody rather than to the transcript, which has gone with everything else. """ - for agent in self._agents: - agent.stop() + if self._runner is not None: + self._runner.stop() + else: + for agent in self._agents: + agent.stop() + self._runner = None self._agents, self._stopping = [], [] self._spoke.set() self._close_btw() @@ -2890,6 +2901,7 @@ def _flow(self, argv: list[str], resume: Path | None = None) -> None: self.show(f"hmz: {why}", "red") return agents = list(runner.agents) + self._runner = runner self._agents = self._ran = agents with self._btw_lock: old_side_sessions = [session for _, session in self._btw_active.values()] @@ -2912,9 +2924,27 @@ def _flow(self, argv: list[str], resume: Path | None = None) -> None: # go: a backend only says what a turn cost once the turn is over, and a turn is long. self._tally = Tally(agents, self._monitor) self._tally.watch() + tally = self._tally with self._saying: self._queued, self._given, self._handed = [], [], False + def joined(agent: AgentBase) -> None: + # This callback runs on the flow's thread, before the new agent can take its first + # turn. Keep the same list object: the run-ending and stop paths use its identity + # to avoid clearing whichever flow may have started since. + if self._agents is not agents and self._stopping is not agents: + agent.stop() + return + agents.append(agent) + tally.add(agent) + agent.watch(self._heard) + agent.waiting = self._at_turn_start + agent.ask = functools.partial(self._ask, agent) + agent.prompting = functools.partial(self._listen, agent) + if self._stopping is agents: + agent.stop() + + runner.watch_agents(joined) for agent in agents: agent.watch(self._heard) # Whichever turn starts next takes the oldest line that was held. @@ -2926,7 +2956,7 @@ def _flow(self, argv: list[str], resume: Path | None = None) -> None: self._draw() # This run's, whatever is being watched by the time it ends. - watching, tally = self._monitor, self._tally + watching = self._monitor def drive() -> int: try: @@ -2953,6 +2983,8 @@ def drive() -> int: # nowhere: a flow that ends of its own accord strands the pin exactly as # one that is stopped does. self._on_screen(self._never_sent, "the flow ended first") + if self._runner is runner: + self._runner = None return 0 self._background(drive) diff --git a/src/hmz/tui/tally.py b/src/hmz/tui/tally.py index 9e8f109..70d142f 100644 --- a/src/hmz/tui/tally.py +++ b/src/hmz/tui/tally.py @@ -126,6 +126,7 @@ def __init__(self, agents: Sequence[AgentBase], monitor: Monitor) -> None: monitor: What to tell, as the total each model has cost. """ self._agents = list(agents) + self._agents_lock = threading.Lock() self._monitor = monitor self._read: dict[Path, _Reading] = {} self._stop = threading.Event() @@ -148,6 +149,11 @@ def stops(self) -> None: """Stops reading, once the run this was watching is over.""" self._stop.set() + def add(self, agent: AgentBase) -> None: + """Includes an agent the flow added after this tally started watching.""" + with self._agents_lock: + self._agents.append(agent) + def read(self) -> None: """Reads whatever has been appended since the last read, and says what it comes to. @@ -155,7 +161,9 @@ def read(self) -> None: business reading, a row half written. What a run costs is worth nothing at the price of the run, so anything that goes wrong is left for the next read to find gone. """ - for agent in self._agents: + with self._agents_lock: + agents = tuple(self._agents) + for agent in agents: profile = backends.named(agent.backend) if profile is None: continue diff --git a/tests/test_dynamic_agents.py b/tests/test_dynamic_agents.py new file mode 100644 index 0000000..bb18435 --- /dev/null +++ b/tests/test_dynamic_agents.py @@ -0,0 +1,223 @@ +"""Agents whose count is learned while a flow is already running.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from hmz.agents import AgentConfig +from hmz.cycle import cycles, read +from hmz.runner import Runner +from tests.stubs import ShellAgent, events, written + +if TYPE_CHECKING: + from pathlib import Path + + +CONFIG = AgentConfig(model="m", effort="high") + +DYNAMIC = """ +from hmz.flows import Agent, flow, spawn + + +@flow +def run(agents: tuple[Agent], task: str) -> None: + experts = spawn(agents[0], (f"expert-{at + 1}" for at in range(int(task)))) + for at, expert in enumerate(experts): + expert.new()(f"echo specialist-{at + 1}") +""" + + +def test_a_flow_spawns_the_number_of_agents_it_learns_at_runtime( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + written(tmp_path, "dynamic", DYNAMIC) + + template = ShellAgent(CONFIG, name="expert-template") + runner = Runner(tmp_path / "dynamic", [template]) + runner.run("3") + + (cycle,) = cycles() + ran = read(cycle) + assert ran is not None + assert [agent.agent for agent in ran.agents] == [ + "expert-template", + "expert-1", + "expert-2", + "expert-3", + ] + assert [session.agent for session in ran.sessions] == [ + "expert-1", + "expert-2", + "expert-3", + ] + spawned = [event for event in events(cycle) if event["event"] == "spawned"] + assert [event["parent"] for event in spawned] == ["expert-template"] * 3 + assert [event["agent"]["agent"] for event in spawned] == [ + "expert-1", + "expert-2", + "expert-3", + ] + assert [agent.id for agent in runner.agents] == [ + "expert-template", + "expert-1", + "expert-2", + "expert-3", + ] + assert not template.stopped + assert all(agent.stopped for agent in runner.agents[1:]) + + +def test_zero_runtime_agents_is_a_valid_fan_out( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + written(tmp_path, "dynamic", DYNAMIC) + runner = Runner(tmp_path / "dynamic", [ShellAgent(CONFIG, name="expert-template")]) + + runner.run("0") + + assert [agent.id for agent in runner.agents] == ["expert-template"] + (cycle,) = cycles() + assert not any(event["event"] == "spawned" for event in events(cycle)) + + +def test_a_runner_can_be_reused_for_another_runtime_fan_out( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + written(tmp_path, "dynamic", DYNAMIC) + runner = Runner(tmp_path / "dynamic", [ShellAgent(CONFIG, name="expert-template")]) + + runner.run("0") + runner.run("1") + + assert [agent.id for agent in runner.agents] == ["expert-template", "expert-1"] + assert runner.agents[1].stopped + + +def test_a_runtime_name_cannot_collide_with_a_declared_agent( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + written( + tmp_path, + "collision", + """ +from hmz.flows import Agent, flow, spawn + + +@flow +def run(agents: tuple[Agent], task: str) -> None: + spawn(agents[0], ("expert-template", "another")) +""", + ) + runner = Runner( + tmp_path / "collision", [ShellAgent(CONFIG, name="expert-template")] + ) + + with pytest.raises(ValueError, match="already in this run"): + runner.run("go") + + assert [agent.id for agent in runner.agents] == ["expert-template"] + + +def test_agents_spawned_by_a_called_flow_belong_to_the_root_run( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + local = tmp_path / ".humanize" / "flows" + written( + local, + "inner", + """ +from hmz.flows import Agent, flow, spawn + + +@flow +def run(agents: tuple[Agent], task: str) -> None: + spawn(agents[0], ("nested-expert",))[0].new()("echo nested") +""", + ) + written( + local, + "outer", + """ +from hmz.flows import Agent, flow, load + + +@flow +def run(agents: tuple[Agent], task: str) -> None: + load("inner")(agents, task) +""", + ) + runner = Runner("outer", [ShellAgent(CONFIG, name="expert-template")]) + + runner.run("go") + + assert [agent.id for agent in runner.agents] == [ + "expert-template", + "nested-expert", + ] + assert runner.agents[1].stopped + (cycle,) = cycles() + ran = read(cycle) + assert ran is not None + assert [agent.agent for agent in ran.agents] == [ + "expert-template", + "nested-expert", + ] + assert [session.agent for session in ran.sessions] == ["nested-expert"] + + +def test_an_agent_spawned_after_stop_is_stopped_before_the_flow_can_run_it( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + written( + tmp_path, + "late", + """ +import time +from pathlib import Path + +from hmz.flows import Agent, flow, spawn + + +@flow +def run(agents: tuple[Agent], task: str) -> None: + while not Path("spawn-now").exists(): + time.sleep(0.01) + late = spawn(agents[0], ("late-expert",))[0] + Path("late-stopped").write_text(str(late.stopped)) +""", + ) + from hmz.sdk.running import Run + + running = Run( + Runner(tmp_path / "late", [ShellAgent(CONFIG, name="expert-template")]), + "go", + ) + running.start() + running.stop() + (tmp_path / "spawn-now").touch() + + assert running.wait(5) + assert running.raised is None + assert (tmp_path / "late-stopped").read_text() == "True" + assert [agent.id for agent in running.agents] == [ + "expert-template", + "late-expert", + ] + + +def test_spawn_refuses_ambiguous_agent_names() -> None: + from hmz.flows import spawn + + template = ShellAgent(CONFIG, name="expert-template") + with pytest.raises(ValueError, match="must be unique"): + spawn(template, ("expert", "expert")) + with pytest.raises(ValueError, match="must not be empty"): + spawn(template, ("",)) From 15c3e9bf617edfeb8b3712f978dd79351b040733 Mon Sep 17 00:00:00 2001 From: antoinegg1 <3528942762@qq.com> Date: Tue, 1 Sep 2026 11:19:53 +0800 Subject: [PATCH 2/2] fix(tui): keep runner type out of tui layer --- src/hmz/tui/app.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/src/hmz/tui/app.py b/src/hmz/tui/app.py index 64cad97..86e6566 100644 --- a/src/hmz/tui/app.py +++ b/src/hmz/tui/app.py @@ -98,14 +98,22 @@ if TYPE_CHECKING: from collections.abc import Callable, Sequence + from typing import Protocol from pydantic import BaseModel from hmz.agents import AgentBase, Board, Event, Question, SessionBase from hmz.flows import Place - from hmz.runner import Runner from hmz.sdk import Session + class _Runner(Protocol): + """The run controls the TUI needs without naming the runner layer.""" + + def stop(self) -> None: ... + + def watch_agents(self, listener: Callable[[AgentBase], None]) -> None: ... + + #: What the editor understands, named as opencode names them, one step along: what answers #: here is a flow rather than an agent, so opencode's `/agents` is `/flow`, and what a flow #: runs on is an agent apiece rather than one model, so its `/models` is the page along from @@ -847,7 +855,7 @@ def __init__( self._agents: list[AgentBase] = [] #: The controller of that run. Unlike an agent list snapshot, this also catches an #: agent registered at the same moment the run is being stopped. - self._runner: Runner | None = None + self._runner: _Runner | None = None #: What the flow has done so far, which is what the right-hand column shows, and who #: reads the agents' own logs into it while it runs. self._monitor = Monitor()