From 683a85b4ba7b66e73ea544e5a239f33eac4dfca5 Mon Sep 17 00:00:00 2001 From: Koukyosyumei Date: Mon, 13 Jul 2026 16:56:23 -0400 Subject: [PATCH 1/8] feat(orchestra): auto-open viewers on resident tmux sessions (watch=) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit launcher="resident" brings agent sessions up detached, forcing a manual tmux attach -t hunt per agent. Conductor now defaults watch=True for that launcher: a background SessionWatcher polls for h5i-orch-- sessions and surfaces each as it appears — linked window when already inside tmux, Windows Terminal tab on WSL, GUI terminal on a display, or a printed attach hint headless. watch=False disables; a string (or $H5I_TERMINAL) is a custom command template with {session}. Viewer failures degrade to the hint and never fail the score. --- README.md | 8 ++ examples/README.md | 8 ++ src/h5i/orchestra/_conductor.py | 26 ++++ src/h5i/orchestra/_watch.py | 225 ++++++++++++++++++++++++++++++++ tests/test_watch.py | 176 +++++++++++++++++++++++++ 5 files changed, 443 insertions(+) create mode 100644 src/h5i/orchestra/_watch.py create mode 100644 tests/test_watch.py diff --git a/README.md b/README.md index dece0d6..2d2c35a 100644 --- a/README.md +++ b/README.md @@ -110,6 +110,14 @@ brings up tmux sessions itself), or `on_turn=my_callback` (every turn is delivered to your Python function — script deterministic agents in tests, or spawn your own runtimes). +**Watching resident sessions.** With `launcher="resident"` the score also +auto-opens a viewer on each agent's tmux session as it comes up — a window +linked into your current tmux session, a Windows Terminal tab under WSL, or a +new GUI terminal — so you never hunt for `tmux attach -t …` by hand. Headless +environments just get the attach command printed. Tune it with +`watch="wezterm start -- tmux attach -t {session}"` (or `$H5I_TERMINAL`), or +turn it off with `watch=False`. + **Discipline the journal asks of you** (same as the Rust eDSL): steps that run concurrently need distinct labels (`c.scope(f"item/{i}").step("fetch", …)` in parallel loops), and one agent's turns run sequentially — one resident diff --git a/examples/README.md b/examples/README.md index 7e20cbb..b406236 100644 --- a/examples/README.md +++ b/examples/README.md @@ -5,6 +5,14 @@ again — journaled steps replay and the run continues where it stopped. They assume agent runtimes are available (`launcher="resident"` brings tmux sessions up itself; drop it if you park resident sessions yourself). +You don't need to `tmux attach` by hand to watch the agents: a resident-launcher +score auto-opens a viewer on each agent session as it comes up — a window +linked into your current tmux session if you run the score inside tmux, a +Windows Terminal tab under WSL, or a new GUI terminal window otherwise. In a +headless shell it prints the exact `tmux attach -t …` command instead. Pass +`watch=False` to `Conductor` to disable it, or set a custom terminal with +`watch="kitty -e tmux attach -t {session}"` / `$H5I_TERMINAL`. + ## Running the examples You need four things: diff --git a/src/h5i/orchestra/_conductor.py b/src/h5i/orchestra/_conductor.py index 6dddf20..4cea4a6 100644 --- a/src/h5i/orchestra/_conductor.py +++ b/src/h5i/orchestra/_conductor.py @@ -19,6 +19,7 @@ from __future__ import annotations +import asyncio import hashlib import inspect import sys @@ -26,6 +27,7 @@ from typing import Any, Awaitable, Callable, Iterable, Sequence from . import policy as _policy +from ._watch import SessionWatcher from ._errors import BridgeClosedError, OrchestraError, ProtocolError from ._rpc import Bridge, resolve_h5i_bin from ._types import ( @@ -83,6 +85,14 @@ class Conductor: - ``score_digest``: provenance digest recorded at launch. Defaults to the sha256 of ``sys.argv[0]``; pass ``None`` to record nothing, or your own string. + - ``watch``: auto-open a viewer on each agent's resident tmux session as + it comes up, instead of hunting for ``tmux attach -t …`` by hand. + ``True`` picks the best available surface (a window linked into your + current tmux session, a Windows Terminal tab under WSL, or a GUI + terminal); a string is a command template, e.g. + ``watch="kitty -e tmux attach -t {session}"``. Defaults to on for + ``launcher="resident"`` (set ``watch=False`` to silence); viewer + failures never fail the score — they degrade to a printed attach hint. """ def __init__( @@ -100,6 +110,7 @@ def __init__( turn_timeout: float | None = None, score_digest: Any = _AUTO, h5i_bin: str | None = None, + watch: bool | str | None = None, ): if not run: raise TypeError("Conductor(...) requires a run id: Conductor(repo, run)") @@ -119,6 +130,8 @@ def __init__( self._turn_timeout = turn_timeout self._score_digest = score_digest self._h5i_bin = h5i_bin + self._watch = self._launcher == "resident" if watch is None else watch + self._watch_task: asyncio.Task | None = None self._bridge: Bridge | None = None self._run_id: str | None = None self._actor_resolved: str | None = None @@ -195,11 +208,24 @@ async def launch(self) -> "Conductor": self._run_id = launched.get("run_id") self._actor_resolved = launched.get("actor") self._replayed_steps = int(launched.get("replayed_steps") or 0) + if self._watch and self._run_id: + watcher = SessionWatcher( + self._run_id, + template=self._watch if isinstance(self._watch, str) else None, + ) + self._watch_task = asyncio.ensure_future(watcher.run()) return self async def close(self) -> None: """Shut the bridge down. The run's journal stays durable — re-running the score resumes it.""" + watch_task, self._watch_task = self._watch_task, None + if watch_task is not None: + watch_task.cancel() + try: + await watch_task + except (asyncio.CancelledError, Exception): + pass # a viewer must never mask the real shutdown path bridge, self._bridge = self._bridge, None if bridge is not None: await bridge.notify_close() diff --git a/src/h5i/orchestra/_watch.py b/src/h5i/orchestra/_watch.py new file mode 100644 index 0000000..30df6d5 --- /dev/null +++ b/src/h5i/orchestra/_watch.py @@ -0,0 +1,225 @@ +"""Auto-attach viewers for resident agent sessions — ``Conductor(watch=True)``. + +The ``"resident"`` launcher brings each agent up in a *detached* tmux session +named ``h5i-orch--``, which normally means hunting for the right +``tmux attach -t …`` in a second terminal. Watching removes the hunt: a +background task polls for those sessions and opens an interactive viewer on +each one as it appears (and again if it dies and comes back). + +How a viewer is opened, in order of preference: + +1. an explicit command template — ``watch="kitty --title {session} tmux + attach -t {session}"`` or ``$H5I_TERMINAL``. ``{session}`` is replaced + with the session name; a template without the placeholder gets + ``tmux attach-session -t `` appended as trailing argv (the + ``terminal -e``-style convention); +2. already inside tmux (``$TMUX``): the agent's window is **linked** into + your current session — no nested clients, switch to it like any other + window (it is removed when the agent session ends); +3. WSL with Windows Terminal on PATH: a new ``wt.exe`` tab per agent; +4. a GUI terminal on ``$DISPLAY``/``$WAYLAND_DISPLAY`` (wezterm, kitty, + alacritty, gnome-terminal, konsole, foot, x-terminal-emulator, xterm); +5. otherwise: a hint line on stderr with the exact attach command. + +A broken viewer never fails the score — every opener error degrades to the +stderr hint. +""" + +from __future__ import annotations + +import asyncio +import os +import shlex +import shutil +import sys +from typing import Callable, Mapping, NamedTuple + +__all__ = ["Opener", "SessionWatcher", "resolve_opener", "session_prefix"] + + +def session_prefix(run_id: str) -> str: + """The resident launcher's session-name prefix for a run (mirrors the + Rust side's ``h5i-orch-{run_id}-{agent_id}``).""" + return f"h5i-orch-{run_id}-" + + +def _attach_argv(session: str) -> list[str]: + return ["tmux", "attach-session", "-t", session] + + +def template_argv(template: str, session: str) -> list[str]: + """Expand a user command template into argv for one session.""" + words = shlex.split(template) + if any("{session}" in w for w in words): + return [w.replace("{session}", session) for w in words] + return words + _attach_argv(session) + + +def wt_argv(session: str, distro: str | None = None) -> list[str]: + """A new Windows Terminal tab attached to ``session`` (WSL interop).""" + wsl = ["wsl.exe"] + (["-d", distro] if distro else []) + return ["wt.exe", "-w", "0", "new-tab", "--title", session, *wsl, "-e", *_attach_argv(session)] + + +#: GUI terminals we know how to hand an attach command, most specific first. +_GUI_TERMINALS: tuple[tuple[str, Callable[[str], list[str]]], ...] = ( + ("wezterm", lambda s: ["wezterm", "start", "--", *_attach_argv(s)]), + ("kitty", lambda s: ["kitty", "--title", s, *_attach_argv(s)]), + ("alacritty", lambda s: ["alacritty", "--title", s, "-e", *_attach_argv(s)]), + ("gnome-terminal", lambda s: ["gnome-terminal", "--title", s, "--", *_attach_argv(s)]), + ("konsole", lambda s: ["konsole", "-p", f"tabtitle={s}", "-e", *_attach_argv(s)]), + ("foot", lambda s: ["foot", "--title", s, *_attach_argv(s)]), + ("x-terminal-emulator", lambda s: ["x-terminal-emulator", "-e", *_attach_argv(s)]), + ("xterm", lambda s: ["xterm", "-T", s, "-e", *_attach_argv(s)]), +) + + +class Opener(NamedTuple): + """How viewers are opened: spawn argv, link a tmux window, or just hint.""" + + name: str + kind: str # "spawn" | "tmux-link" | "hint" + argv: Callable[[str], list[str]] | None = None + + +def resolve_opener( + template: str | None = None, + *, + env: Mapping[str, str] = os.environ, + which: Callable[[str], str | None] = shutil.which, +) -> Opener: + """Pick the best way to surface agent sessions in this environment.""" + template = template or env.get("H5I_TERMINAL") + if template: + name = shlex.split(template)[0] + return Opener(name, "spawn", lambda s: template_argv(template, s)) + if env.get("TMUX"): + return Opener("tmux link-window", "tmux-link") + if which("wt.exe") and which("wsl.exe"): + distro = env.get("WSL_DISTRO_NAME") + return Opener("wt.exe", "spawn", lambda s: wt_argv(s, distro)) + if env.get("DISPLAY") or env.get("WAYLAND_DISPLAY"): + for name, build in _GUI_TERMINALS: + if which(name): + return Opener(name, "spawn", build) + return Opener("hint", "hint") + + +class SessionWatcher: + """Polls tmux for a run's agent sessions and opens a viewer on each. + + Sessions appear lazily (the resident launcher brings one up on an + agent's *first turn*), so the watcher runs for the whole score. A + session that vanishes and comes back gets a fresh viewer. + """ + + def __init__( + self, + run_id: str, + *, + template: str | None = None, + opener: Opener | None = None, + poll_interval: float = 1.0, + echo: Callable[[str], None] | None = None, + ): + self._prefix = session_prefix(run_id) + self._opener = opener or resolve_opener(template) + self._poll_interval = poll_interval + self._echo = echo or (lambda line: print(line, file=sys.stderr, flush=True)) + self._open_now: set[str] = set() + + async def run(self) -> None: + """Poll until cancelled (or until tmux turns out not to exist).""" + while await self.poll_once(): + await asyncio.sleep(self._poll_interval) + + async def poll_once(self) -> bool: + """One poll step; returns False when polling can never succeed.""" + names = await self._list_sessions() + if names is None: + return False # no tmux binary — sessions will never appear + current = {n for n in names if n.startswith(self._prefix)} + self._open_now &= current # a vanished session may come back + for session in sorted(current - self._open_now): + self._open_now.add(session) + await self._open(session) + return True + + async def _list_sessions(self) -> list[str] | None: + try: + proc = await asyncio.create_subprocess_exec( + "tmux", + "list-sessions", + "-F", + "#{session_name}", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.DEVNULL, + ) + except FileNotFoundError: + return None + out, _ = await proc.communicate() + if proc.returncode != 0: + return [] # no tmux server yet — keep polling + return out.decode(errors="replace").splitlines() + + async def _open(self, session: str) -> None: + opener = self._opener + try: + if opener.kind == "spawn": + assert opener.argv is not None + await self._spawn(opener.argv(session)) + self._echo(f"[h5i] agent session '{session}' up — opened in {opener.name}") + return + if opener.kind == "tmux-link": + await self._link_window(session) + self._echo( + f"[h5i] agent session '{session}' up — linked as a window " + "in your current tmux session" + ) + return + except Exception as e: # a broken viewer must not fail the score + self._echo( + f"[h5i] could not open a viewer for '{session}' ({e}) — " + f"attach with: tmux attach -t {session}" + ) + return + self._echo(f"[h5i] agent session up — view it with: tmux attach -t {session}") + + async def _spawn(self, argv: list[str]) -> None: + proc = await asyncio.create_subprocess_exec( + *argv, + stdin=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + start_new_session=True, + ) + # Terminals live as long as the viewer stays open; reap in background. + asyncio.ensure_future(proc.wait()) + + async def _link_window(self, session: str) -> None: + proc = await asyncio.create_subprocess_exec( + "tmux", + "list-windows", + "-t", + session, + "-F", + "#{window_id}", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.DEVNULL, + ) + out, _ = await proc.communicate() + window_ids = out.decode(errors="replace").split() + if proc.returncode != 0 or not window_ids: + raise RuntimeError(f"no windows found in tmux session '{session}'") + link = await asyncio.create_subprocess_exec( + "tmux", + "link-window", + "-d", + "-a", + "-s", + window_ids[0], + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + ) + if await link.wait() != 0: + raise RuntimeError("tmux link-window failed") diff --git a/tests/test_watch.py b/tests/test_watch.py new file mode 100644 index 0000000..4e15a0f --- /dev/null +++ b/tests/test_watch.py @@ -0,0 +1,176 @@ +"""Unit tests for the resident-session auto-viewer (``_watch``).""" + +import asyncio + +import pytest + +from h5i.orchestra._conductor import Conductor +from h5i.orchestra._watch import ( + Opener, + SessionWatcher, + resolve_opener, + session_prefix, + template_argv, + wt_argv, +) + +# ── opener resolution ──────────────────────────────────────────────────────── + + +def _no_binary(_name: str) -> None: + return None + + +def test_template_wins_over_everything(): + opener = resolve_opener( + "kitty -e tmux attach -t {session}", + env={"TMUX": "/tmp/tmux-1000/default,42,0"}, + which=_no_binary, + ) + assert opener.kind == "spawn" + assert opener.argv("h5i-orch-r-a") == [ + "kitty", "-e", "tmux", "attach", "-t", "h5i-orch-r-a" + ] + + +def test_template_without_placeholder_gets_attach_appended(): + assert template_argv("foot --hold", "s1") == [ + "foot", "--hold", "tmux", "attach-session", "-t", "s1" + ] + + +def test_env_template_is_honored(): + opener = resolve_opener(env={"H5I_TERMINAL": "myterm -e"}, which=_no_binary) + assert opener.name == "myterm" + assert opener.argv("s")[:2] == ["myterm", "-e"] + + +def test_inside_tmux_links_windows(): + opener = resolve_opener(env={"TMUX": "sock,1,0"}, which=_no_binary) + assert opener.kind == "tmux-link" + + +def test_wsl_uses_windows_terminal(): + opener = resolve_opener( + env={"WSL_DISTRO_NAME": "Ubuntu"}, + which=lambda name: name if name in ("wt.exe", "wsl.exe") else None, + ) + assert opener.name == "wt.exe" + argv = opener.argv("s1") + assert argv[:2] == ["wt.exe", "-w"] + assert ["wsl.exe", "-d", "Ubuntu"] == argv[argv.index("wsl.exe"):][:3] + assert argv[-4:] == ["tmux", "attach-session", "-t", "s1"] + + +def test_wt_argv_without_distro(): + assert "-d" not in wt_argv("s") + + +def test_gui_terminal_on_display(): + opener = resolve_opener( + env={"DISPLAY": ":0"}, + which=lambda name: name if name == "alacritty" else None, + ) + assert opener.name == "alacritty" + assert opener.argv("s1")[0] == "alacritty" + + +def test_headless_falls_back_to_hint(): + assert resolve_opener(env={}, which=_no_binary).kind == "hint" + + +# ── the watcher loop ───────────────────────────────────────────────────────── + + +class ScriptedWatcher(SessionWatcher): + """A watcher over a scripted tmux session list — no subprocesses.""" + + def __init__(self, run_id: str, listings, **kwargs): + self.opened: list[str] = [] + self.echoed: list[str] = [] + super().__init__( + run_id, + opener=Opener("fake", "spawn", lambda s: ["true", s]), + echo=self.echoed.append, + **kwargs, + ) + self._listings = iter(listings) + + async def _list_sessions(self): + return next(self._listings) + + async def _spawn(self, argv): + self.opened.append(argv[1]) + + +@pytest.mark.asyncio +async def test_opens_each_new_session_once(): + prefix = session_prefix("run1") + w = ScriptedWatcher( + "run1", + [ + ["unrelated"], + [f"{prefix}claude", "unrelated"], + [f"{prefix}claude", f"{prefix}codex"], + [f"{prefix}claude", f"{prefix}codex"], + ], + ) + for _ in range(4): + assert await w.poll_once() + assert w.opened == [f"{prefix}claude", f"{prefix}codex"] + + +@pytest.mark.asyncio +async def test_reopens_a_session_that_died_and_came_back(): + session = session_prefix("run1") + "claude" + w = ScriptedWatcher("run1", [[session], [], [session]]) + for _ in range(3): + await w.poll_once() + assert w.opened == [session, session] + + +@pytest.mark.asyncio +async def test_no_tmux_binary_stops_the_loop(): + class NoTmux(ScriptedWatcher): + async def _list_sessions(self): + return None + + assert not await NoTmux("run1", []).poll_once() + + +@pytest.mark.asyncio +async def test_broken_viewer_degrades_to_hint(): + session = session_prefix("run1") + "claude" + + class Broken(ScriptedWatcher): + async def _spawn(self, argv): + raise OSError("boom") + + w = Broken("run1", [[session]]) + assert await w.poll_once() + assert any(f"tmux attach -t {session}" in line for line in w.echoed) + + +# ── Conductor wiring ───────────────────────────────────────────────────────── + + +def test_watch_defaults_follow_the_launcher(): + assert Conductor(".", "r", launcher="resident")._watch is True + assert Conductor(".", "r")._watch is False + assert Conductor(".", "r", launcher="resident", watch=False)._watch is False + c = Conductor(".", "r", watch="kitty -e tmux attach -t {session}") + assert c._watch == "kitty -e tmux attach -t {session}" + + +@pytest.mark.asyncio +async def test_close_cancels_the_watch_task(): + c = Conductor(".", "r", launcher="resident") + + async def forever(): + await asyncio.Event().wait() + + c._watch_task = asyncio.ensure_future(forever()) + task = c._watch_task + await c.close() + assert task.cancelled() + assert c._watch_task is None From 0469b9de497dc54d4c1211e7e4f137015d349d9f Mon Sep 17 00:00:00 2001 From: Koukyosyumei Date: Mon, 13 Jul 2026 19:02:47 -0400 Subject: [PATCH 2/8] feat(orchestra): log agent tmux session lifecycle in the watcher Every resident-launcher score now narrates session status on stderr: a watch-start line naming the viewer that will open, per-agent 'session up' lines that always include the manual tmux attach command, 'session ended' lines when an agent's tmux session goes away, and an explicit notice when tmux is missing entirely. --- src/h5i/orchestra/_watch.py | 33 ++++++++++++++++++++++++++++----- tests/test_watch.py | 9 +++++++++ 2 files changed, 37 insertions(+), 5 deletions(-) diff --git a/src/h5i/orchestra/_watch.py b/src/h5i/orchestra/_watch.py index 30df6d5..f754bc0 100644 --- a/src/h5i/orchestra/_watch.py +++ b/src/h5i/orchestra/_watch.py @@ -128,8 +128,17 @@ def __init__( self._echo = echo or (lambda line: print(line, file=sys.stderr, flush=True)) self._open_now: set[str] = set() + def _agent(self, session: str) -> str: + return session[len(self._prefix):] or session + async def run(self) -> None: """Poll until cancelled (or until tmux turns out not to exist).""" + how = ( + f"each opens in {self._opener.name} as its first turn starts" + if self._opener.kind != "hint" + else "attach commands are printed as each comes up" + ) + self._echo(f"[h5i] watching for agent sessions ({self._prefix}*) — {how}") while await self.poll_once(): await asyncio.sleep(self._poll_interval) @@ -137,8 +146,16 @@ async def poll_once(self) -> bool: """One poll step; returns False when polling can never succeed.""" names = await self._list_sessions() if names is None: + self._echo( + "[h5i] tmux not found — agent sessions cannot be watched " + "(the resident launcher needs tmux)" + ) return False # no tmux binary — sessions will never appear current = {n for n in names if n.startswith(self._prefix)} + for session in sorted(self._open_now - current): + self._echo( + f"[h5i] agent '{self._agent(session)}' session ended ({session})" + ) self._open_now &= current # a vanished session may come back for session in sorted(current - self._open_now): self._open_now.add(session) @@ -164,26 +181,32 @@ async def _list_sessions(self) -> list[str] | None: async def _open(self, session: str) -> None: opener = self._opener + agent = self._agent(session) try: if opener.kind == "spawn": assert opener.argv is not None await self._spawn(opener.argv(session)) - self._echo(f"[h5i] agent session '{session}' up — opened in {opener.name}") + self._echo( + f"[h5i] agent '{agent}' session up — opened in {opener.name} " + f"(or: tmux attach -t {session})" + ) return if opener.kind == "tmux-link": await self._link_window(session) self._echo( - f"[h5i] agent session '{session}' up — linked as a window " - "in your current tmux session" + f"[h5i] agent '{agent}' session up — linked as window " + f"'{session}' in your current tmux session" ) return except Exception as e: # a broken viewer must not fail the score self._echo( - f"[h5i] could not open a viewer for '{session}' ({e}) — " + f"[h5i] agent '{agent}' session up, but its viewer failed ({e}) — " f"attach with: tmux attach -t {session}" ) return - self._echo(f"[h5i] agent session up — view it with: tmux attach -t {session}") + self._echo( + f"[h5i] agent '{agent}' session up — view it with: tmux attach -t {session}" + ) async def _spawn(self, argv: list[str]) -> None: proc = await asyncio.create_subprocess_exec( diff --git a/tests/test_watch.py b/tests/test_watch.py index 4e15a0f..25f37b3 100644 --- a/tests/test_watch.py +++ b/tests/test_watch.py @@ -127,6 +127,15 @@ async def test_reopens_a_session_that_died_and_came_back(): for _ in range(3): await w.poll_once() assert w.opened == [session, session] + assert any("'claude' session ended" in line for line in w.echoed) + + +@pytest.mark.asyncio +async def test_run_logs_watch_start_and_missing_tmux(): + w = ScriptedWatcher("run1", [None], poll_interval=0) + await w.run() + assert any("watching for agent sessions" in line for line in w.echoed) + assert any("tmux not found" in line for line in w.echoed) @pytest.mark.asyncio From e3dded1f822ee95da946ad96b47f23e08b50139e Mon Sep 17 00:00:00 2001 From: Koukyosyumei Date: Mon, 13 Jul 2026 19:21:06 -0400 Subject: [PATCH 3/8] feat(orchestra): warn when a pending turn's tmux session is missing or dies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Agent.work/ask/review/revise now tell the session watcher a turn is in flight (resident launcher only). The watcher warns once if the agent's session hasn't appeared within a grace period (default 15s) or ended while the turn is still pending — both point at 'h5i env shell -- true' to diagnose — instead of the score hanging with no output. This is the SDK-side complement to h5i's new fail-closed env checks in hire and LaunchResident. --- examples/README.md | 4 +- src/h5i/orchestra/_conductor.py | 36 ++++++++++--- src/h5i/orchestra/_watch.py | 54 ++++++++++++++++++-- tests/test_watch.py | 89 +++++++++++++++++++++++++++++++++ 4 files changed, 173 insertions(+), 10 deletions(-) diff --git a/examples/README.md b/examples/README.md index b406236..81cfbf5 100644 --- a/examples/README.md +++ b/examples/README.md @@ -9,7 +9,9 @@ You don't need to `tmux attach` by hand to watch the agents: a resident-launcher score auto-opens a viewer on each agent session as it comes up — a window linked into your current tmux session if you run the score inside tmux, a Windows Terminal tab under WSL, or a new GUI terminal window otherwise. In a -headless shell it prints the exact `tmux attach -t …` command instead. Pass +headless shell it prints the exact `tmux attach -t …` command instead. It also +warns when an agent has a turn in flight but its session never appeared or died +mid-turn (with the `h5i env shell … -- true` command to diagnose why). Pass `watch=False` to `Conductor` to disable it, or set a custom terminal with `watch="kitty -e tmux attach -t {session}"` / `$H5I_TERMINAL`. diff --git a/src/h5i/orchestra/_conductor.py b/src/h5i/orchestra/_conductor.py index 4cea4a6..eed6652 100644 --- a/src/h5i/orchestra/_conductor.py +++ b/src/h5i/orchestra/_conductor.py @@ -132,6 +132,7 @@ def __init__( self._h5i_bin = h5i_bin self._watch = self._launcher == "resident" if watch is None else watch self._watch_task: asyncio.Task | None = None + self._session_watcher: SessionWatcher | None = None self._bridge: Bridge | None = None self._run_id: str | None = None self._actor_resolved: str | None = None @@ -209,11 +210,11 @@ async def launch(self) -> "Conductor": self._actor_resolved = launched.get("actor") self._replayed_steps = int(launched.get("replayed_steps") or 0) if self._watch and self._run_id: - watcher = SessionWatcher( + self._session_watcher = SessionWatcher( self._run_id, template=self._watch if isinstance(self._watch, str) else None, ) - self._watch_task = asyncio.ensure_future(watcher.run()) + self._watch_task = asyncio.ensure_future(self._session_watcher.run()) return self async def close(self) -> None: @@ -450,6 +451,21 @@ async def _request(self, method: str, params: dict) -> Any: ) return await self._bridge.request(method, params) + async def _turn_request( + self, method: str, params: dict, agent_id: str, env_id: str + ) -> Any: + """A `_request` that tells the session watcher a turn is in flight for + ``agent_id`` — so a resident session that dies on startup (or mid-turn) + is warned about instead of the score hanging silently.""" + watcher = self._session_watcher + if watcher is None or self._launcher != "resident": + return await self._request(method, params) + watcher.expect(agent_id, env_id) + try: + return await self._request(method, params) + finally: + watcher.unexpect(agent_id) + async def _abort(self, method: str, token: str) -> None: try: await self._request(method, {"token": token}) @@ -519,7 +535,9 @@ async def work( } if materials: params["materials"] = [m.to_payload() for m in materials] - raw = await self._conductor._request("agent.work", params) + raw = await self._conductor._turn_request( + "agent.work", params, self.id, self.env_id + ) return Artifact.from_raw(raw) async def ask( @@ -544,9 +562,11 @@ async def ask( value: Any = None last_error: Exception | None = None for _ in range(max(1, attempts)): - value = await self._conductor._request( + value = await self._conductor._turn_request( "agent.ask", {"agent": self.id, "env_id": self.env_id, "prompt": current}, + self.id, + self.env_id, ) if parse is None: return value @@ -566,20 +586,22 @@ async def ask( async def review(self, artifact: Artifact) -> Review: """Review a teammate's artifact (scoped read grant → posted review).""" - raw = await self._conductor._request( + raw = await self._conductor._turn_request( "agent.review", { "reviewer": self.id, "env_id": self.env_id, "artifact": artifact.to_payload(), }, + self.id, + self.env_id, ) return Review.from_raw(raw) async def revise(self, artifact: Artifact, review: Review) -> Artifact: """Address a review and re-submit. Completes on any new submission — an agent that finds nothing to fix re-submits as-is.""" - raw = await self._conductor._request( + raw = await self._conductor._turn_request( "agent.revise", { "agent": self.id, @@ -587,5 +609,7 @@ async def revise(self, artifact: Artifact, review: Review) -> Artifact: "artifact": artifact.to_payload(), "review": review.to_payload(), }, + self.id, + self.env_id, ) return Artifact.from_raw(raw) diff --git a/src/h5i/orchestra/_watch.py b/src/h5i/orchestra/_watch.py index f754bc0..a91ceb6 100644 --- a/src/h5i/orchestra/_watch.py +++ b/src/h5i/orchestra/_watch.py @@ -120,17 +120,54 @@ def __init__( template: str | None = None, opener: Opener | None = None, poll_interval: float = 1.0, + grace: float = 15.0, echo: Callable[[str], None] | None = None, ): self._prefix = session_prefix(run_id) self._opener = opener or resolve_opener(template) self._poll_interval = poll_interval + self._grace = grace self._echo = echo or (lambda line: print(line, file=sys.stderr, flush=True)) self._open_now: set[str] = set() + #: session name → pending-turn record (env id, started-at, warned yet). + self._expected: dict[str, dict[str, Any]] = {} def _agent(self, session: str) -> str: return session[len(self._prefix):] or session + # ── pending-turn tracking (fed by Agent turn calls) ───────────────────── + + def expect(self, agent_id: str, env_id: str) -> None: + """A turn for ``agent_id`` is in flight — its session should exist + (or appear within the grace period). Warns loudly otherwise: the + classic silent failure is a session that dies on startup because the + env behind it is gone.""" + self._expected[f"{self._prefix}{agent_id}"] = { + "env": env_id, + "since": asyncio.get_running_loop().time(), + "warned": False, + } + + def unexpect(self, agent_id: str) -> None: + """The turn finished (either way) — stop holding it to the deadline.""" + self._expected.pop(f"{self._prefix}{agent_id}", None) + + def _check_expected(self, current: set[str]) -> None: + now = asyncio.get_running_loop().time() + for session, pending in self._expected.items(): + if session in current: + pending["warned"] = False # it's up; re-arm for a later death + continue + if pending["warned"] or now - pending["since"] <= self._grace: + continue + pending["warned"] = True + self._echo( + f"[h5i] WARNING: agent '{self._agent(session)}' has a turn in flight " + f"but its tmux session ({session}) has not appeared in {int(self._grace)}s " + f"— it may be dying on startup. Check the env is alive: " + f"h5i env shell {pending['env']} -- true" + ) + async def run(self) -> None: """Poll until cancelled (or until tmux turns out not to exist).""" how = ( @@ -153,13 +190,24 @@ async def poll_once(self) -> bool: return False # no tmux binary — sessions will never appear current = {n for n in names if n.startswith(self._prefix)} for session in sorted(self._open_now - current): - self._echo( - f"[h5i] agent '{self._agent(session)}' session ended ({session})" - ) + pending = self._expected.get(session) + if pending is not None: + pending["warned"] = True # this line already says it all + self._echo( + f"[h5i] WARNING: agent '{self._agent(session)}' session ended " + f"while its turn is still pending ({session}) — the runtime may " + f"have crashed. Check the env is alive: " + f"h5i env shell {pending['env']} -- true" + ) + else: + self._echo( + f"[h5i] agent '{self._agent(session)}' session ended ({session})" + ) self._open_now &= current # a vanished session may come back for session in sorted(current - self._open_now): self._open_now.add(session) await self._open(session) + self._check_expected(current) return True async def _list_sessions(self) -> list[str] | None: diff --git a/tests/test_watch.py b/tests/test_watch.py index 25f37b3..6f9fc69 100644 --- a/tests/test_watch.py +++ b/tests/test_watch.py @@ -160,6 +160,95 @@ async def _spawn(self, argv): assert any(f"tmux attach -t {session}" in line for line in w.echoed) +# ── pending-turn warnings ──────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_warns_once_when_an_expected_session_never_appears(): + w = ScriptedWatcher("run1", [[], [], []], grace=0.0) + w.expect("claude", "env/claude/run1-claude") + await asyncio.sleep(0.01) # move past the zero grace period + for _ in range(3): + await w.poll_once() + warnings = [l for l in w.echoed if "has not appeared" in l] + assert len(warnings) == 1, w.echoed + assert "h5i env shell env/claude/run1-claude -- true" in warnings[0] + + +@pytest.mark.asyncio +async def test_no_warning_while_the_expected_session_is_up(): + session = session_prefix("run1") + "claude" + w = ScriptedWatcher("run1", [[session], [session]], grace=0.0) + w.expect("claude", "env/x") + for _ in range(2): + await w.poll_once() + assert not any("WARNING" in l for l in w.echoed) + + +@pytest.mark.asyncio +async def test_session_dying_mid_turn_is_a_warning(): + session = session_prefix("run1") + "claude" + w = ScriptedWatcher("run1", [[session], [], []], grace=0.0) + w.expect("claude", "env/x") + for _ in range(3): + await w.poll_once() + died = [l for l in w.echoed if "ended while its turn is still pending" in l] + assert len(died) == 1, w.echoed + # ...and the death warning is not doubled by the never-appeared check. + assert not any("has not appeared" in l for l in w.echoed) + + +@pytest.mark.asyncio +async def test_unexpect_silences_the_deadline(): + w = ScriptedWatcher("run1", [[], []], grace=0.0) + w.expect("claude", "env/x") + w.unexpect("claude") + await asyncio.sleep(0.01) + for _ in range(2): + await w.poll_once() + assert not any("WARNING" in l for l in w.echoed) + + +@pytest.mark.asyncio +async def test_turn_request_registers_expectation_with_the_watcher(): + from mock_server import MockOrchestra, launch_conductor + + class FakeWatcher: + def __init__(self): + self.calls = [] + + def expect(self, agent_id, env_id): + self.calls.append(("expect", agent_id, env_id)) + + def unexpect(self, agent_id): + self.calls.append(("unexpect", agent_id)) + + mock = MockOrchestra() + mock.on("agent.hire", lambda p: {"agent_id": p["name"], "env_id": f"env/{p['name']}/r"}) + mock.on( + "agent.work", + lambda p: { + "id": "sha:1", "owner_agent": p["agent"], "round": 1, + "env_id": p["env_id"], "commit_oid": "c", "tree_oid": "t", + "capture_ids": [], "files_changed": 1, "insertions": 1, + "deletions": 0, "submitted_at": "2026-01-01T00:00:00Z", + "independent": True, + }, + ) + c = await launch_conductor(mock, launcher="resident", watch=False) + try: + fake = FakeWatcher() + c._session_watcher = fake + agent = await c.hire("claude", runtime="claude") + await agent.work("task") + assert fake.calls == [ + ("expect", "claude", "env/claude/r"), + ("unexpect", "claude"), + ] + finally: + await c.close() + + # ── Conductor wiring ───────────────────────────────────────────────────────── From 0831ace3472d32b9c423500858a10326395c6d92 Mon Sep 17 00:00:00 2001 From: Koukyosyumei Date: Mon, 13 Jul 2026 21:20:20 -0400 Subject: [PATCH 4/8] fix(orchestra): serialize viewer spawns so every agent gets its own tab MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Agents' first turns are gathered, so their tmux sessions surface in the same watcher poll and the viewers spawned back-to-back — Windows Terminal drops tabs dispatched near-simultaneously, leaving only one agent visible. The watcher now waits for each viewer command to finish dispatching (up to 2s; long-lived terminals keep reaping in the background), treats a fast non-zero exit as a failed viewer (degrading to the printed attach hint), and pauses spawn_gap (0.5s) between opens in one batch. --- src/h5i/orchestra/_watch.py | 21 ++++++++++++++++++--- tests/test_watch.py | 31 +++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 3 deletions(-) diff --git a/src/h5i/orchestra/_watch.py b/src/h5i/orchestra/_watch.py index a91ceb6..4383f84 100644 --- a/src/h5i/orchestra/_watch.py +++ b/src/h5i/orchestra/_watch.py @@ -121,12 +121,14 @@ def __init__( opener: Opener | None = None, poll_interval: float = 1.0, grace: float = 15.0, + spawn_gap: float = 0.5, echo: Callable[[str], None] | None = None, ): self._prefix = session_prefix(run_id) self._opener = opener or resolve_opener(template) self._poll_interval = poll_interval self._grace = grace + self._spawn_gap = spawn_gap self._echo = echo or (lambda line: print(line, file=sys.stderr, flush=True)) self._open_now: set[str] = set() #: session name → pending-turn record (env id, started-at, warned yet). @@ -204,8 +206,14 @@ async def poll_once(self) -> bool: f"[h5i] agent '{self._agent(session)}' session ended ({session})" ) self._open_now &= current # a vanished session may come back - for session in sorted(current - self._open_now): + for i, session in enumerate(sorted(current - self._open_now)): self._open_now.add(session) + if i: + # Several agents often come up in one poll (gathered first + # turns). Rapid-fire viewer spawns race — Windows Terminal + # drops tabs dispatched near-simultaneously — so give each + # viewer a beat to register before the next. + await asyncio.sleep(self._spawn_gap) await self._open(session) self._check_expected(current) return True @@ -264,8 +272,15 @@ async def _spawn(self, argv: list[str]) -> None: stderr=asyncio.subprocess.DEVNULL, start_new_session=True, ) - # Terminals live as long as the viewer stays open; reap in background. - asyncio.ensure_future(proc.wait()) + # Two kinds of viewer command: dispatchers (wt.exe, wezterm cli) exit + # as soon as the tab is registered — wait for that, so consecutive + # opens don't race, and a fast non-zero exit means the viewer failed + # (degrade to the attach hint). Long-lived terminals (kitty, xterm) + # outlive the timeout; leave them reaping in the background. + waiter = asyncio.ensure_future(proc.wait()) + done, _ = await asyncio.wait({waiter}, timeout=2.0) + if waiter in done and waiter.result() != 0: + raise RuntimeError(f"viewer command exited with status {waiter.result()}") async def _link_window(self, session: str) -> None: proc = await asyncio.create_subprocess_exec( diff --git a/tests/test_watch.py b/tests/test_watch.py index 6f9fc69..d2288ba 100644 --- a/tests/test_watch.py +++ b/tests/test_watch.py @@ -88,6 +88,7 @@ class ScriptedWatcher(SessionWatcher): def __init__(self, run_id: str, listings, **kwargs): self.opened: list[str] = [] self.echoed: list[str] = [] + kwargs.setdefault("spawn_gap", 0.0) super().__init__( run_id, opener=Opener("fake", "spawn", lambda s: ["true", s]), @@ -147,6 +148,36 @@ async def _list_sessions(self): assert not await NoTmux("run1", []).poll_once() +@pytest.mark.asyncio +async def test_two_sessions_in_one_poll_both_open(): + prefix = session_prefix("run1") + w = ScriptedWatcher("run1", [[f"{prefix}claude", f"{prefix}codex"]]) + assert await w.poll_once() + assert w.opened == [f"{prefix}claude", f"{prefix}codex"] + + +@pytest.mark.asyncio +async def test_spawn_success_and_fast_failure(): + session = session_prefix("run1") + "claude" + + class RealSpawn(ScriptedWatcher): + async def _spawn(self, argv): + await SessionWatcher._spawn(self, argv) + + # `true` exits 0 fast → treated as a successful dispatch. + ok = RealSpawn("run1", [[session]]) + ok._opener = Opener("true", "spawn", lambda s: ["true"]) + await ok.poll_once() + assert any("opened in true" in l for l in ok.echoed) + + # `false` exits non-zero fast → the viewer failed; degrade to the hint. + bad = RealSpawn("run1", [[session]]) + bad._opener = Opener("false", "spawn", lambda s: ["false"]) + await bad.poll_once() + assert any(f"tmux attach -t {session}" in l for l in bad.echoed), bad.echoed + assert not any("opened in false" in l for l in bad.echoed) + + @pytest.mark.asyncio async def test_broken_viewer_degrades_to_hint(): session = session_prefix("run1") + "claude" From 9ad4767798d4c7cab07384bcc97288252df310f9 Mon Sep 17 00:00:00 2001 From: Koukyosyumei Date: Mon, 13 Jul 2026 21:30:38 -0400 Subject: [PATCH 5/8] feat(orchestra): run-level isolation default; examples pin supervised MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Conductor(isolation=...) sets the run's default sandbox tier: every hire inherits it unless it passes its own (or 'auto' to re-enable auto-picking). Explicit tiers are fail-closed on hosts that cannot enforce them — never silently downgraded. Every example now pins isolation="supervised" so hired agents run seccomp-gated and network-jailed instead of whatever tier the host auto-picks; arena and ensemble also assert the floor across the roster with preflight(min_isolation="supervised"), which catches resumed runs whose envs predate the tier. --- README.md | 8 ++++++++ examples/README.md | 10 +++++++++- examples/arena_score.py | 4 ++-- examples/custom_control_flow.py | 2 +- examples/debate_then_build.py | 2 +- examples/ensemble_score.py | 4 ++-- examples/judge_panel_score.py | 2 +- examples/pipeline_score.py | 2 +- examples/review_escalation.py | 2 +- examples/tournament.py | 2 +- src/h5i/orchestra/_conductor.py | 23 ++++++++++++++++++++++- tests/test_conductor.py | 25 +++++++++++++++++++++++++ 12 files changed, 74 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 2d2c35a..d485713 100644 --- a/README.md +++ b/README.md @@ -110,6 +110,14 @@ brings up tmux sessions itself), or `on_turn=my_callback` (every turn is delivered to your Python function — script deterministic agents in tests, or spawn your own runtimes). +**Sandboxing.** `Conductor(..., isolation="supervised")` sets the run's +default sandbox tier: every `hire` creates its env at that tier unless it +passes its own (`isolation="container"`, or `"auto"` to re-enable +auto-picking). Explicit tiers are fail-closed — hire errors if the host +cannot enforce them, never silently downgrades — and +`preflight(min_isolation=...)` verifies the floor across the whole roster, +which also catches a *resumed* run whose envs were created at a weaker tier. + **Watching resident sessions.** With `launcher="resident"` the score also auto-opens a viewer on each agent's tmux session as it comes up — a window linked into your current tmux session, a Windows Terminal tab under WSL, or a diff --git a/examples/README.md b/examples/README.md index 81cfbf5..f2535cc 100644 --- a/examples/README.md +++ b/examples/README.md @@ -17,7 +17,7 @@ mid-turn (with the `h5i env shell … -- true` command to diagnose why). Pass ## Running the examples -You need four things: +You need five things: 1. **The SDK** — from the repo root: `pip install -e .` (Python ≥ 3.10). 2. **The engine** — the `h5i` binary on `PATH` (`cargo install --path `), @@ -29,6 +29,14 @@ You need four things: the repository the agents should modify, with a **clean worktree** (the arena and ensemble scores call `preflight(clean_worktree=True)` and fail fast otherwise). +5. **A host that can enforce the `supervised` sandbox tier** — every example + pins `isolation="supervised"` on the `Conductor`, so hired agents run in + supervised envs (seccomp-gated, network-jailed) rather than whatever tier + the host happens to auto-pick. The tier is fail-closed: on a host that + cannot enforce it, hire errors instead of silently downgrading — drop the + `isolation=` argument (or set `"auto"`) to fall back to auto-picking. + Note: a *resumed* run keeps the envs it was created with; changing the + tier (like changing a model) needs a fresh run id. `preflight(live=...)` is intended for the default `"attach"` launcher, where sessions must already be parked on their inboxes. Do not use that check before diff --git a/examples/arena_score.py b/examples/arena_score.py index 9586054..1f4e8d8 100644 --- a/examples/arena_score.py +++ b/examples/arena_score.py @@ -17,7 +17,7 @@ async def main(task: str) -> None: - async with Conductor(".", "arena-demo", launcher="resident") as c: + async with Conductor(".", "arena-demo", launcher="resident", isolation="supervised") as c: agents = await asyncio.gather( c.hire("claude", runtime="claude", model="claude-haiku-4-5"), c.hire("codex", runtime="codex", model="gpt-5.4-mini"), @@ -25,7 +25,7 @@ async def main(task: str) -> None: ) # LaunchResident starts each session on its first turn, so there is no # live session to check yet. Still fail fast on repository hygiene. - await c.preflight(clean_worktree=True) + await c.preflight(clean_worktree=True, min_isolation="supervised") outcome = await patterns.arena( c, diff --git a/examples/custom_control_flow.py b/examples/custom_control_flow.py index 9502f6b..e4fa2a0 100644 --- a/examples/custom_control_flow.py +++ b/examples/custom_control_flow.py @@ -11,7 +11,7 @@ async def main() -> None: - async with Conductor(".", "triage-and-fix") as c: + async with Conductor(".", "triage-and-fix", isolation="supervised") as c: lead = await c.hire( "lead", runtime="claude", model="claude-haiku-4-5" ) diff --git a/examples/debate_then_build.py b/examples/debate_then_build.py index 2003da5..4b380f3 100644 --- a/examples/debate_then_build.py +++ b/examples/debate_then_build.py @@ -19,7 +19,7 @@ async def main() -> None: - async with Conductor(".", "debate-demo", launcher="resident") as c: + async with Conductor(".", "debate-demo", launcher="resident", isolation="supervised") as c: pro = await c.hire("pro", runtime="claude", model="claude-haiku-4-5") con = await c.hire("con", runtime="codex", model="gpt-5.4-mini") moderator = await c.hire( diff --git a/examples/ensemble_score.py b/examples/ensemble_score.py index 8ed67d7..16892df 100644 --- a/examples/ensemble_score.py +++ b/examples/ensemble_score.py @@ -16,7 +16,7 @@ async def main(task: str) -> None: - async with Conductor(".", "ensemble-demo", launcher="resident") as c: + async with Conductor(".", "ensemble-demo", launcher="resident", isolation="supervised") as c: claude = await c.hire( "claude", runtime="claude", model="claude-haiku-4-5" ) @@ -25,7 +25,7 @@ async def main(task: str) -> None: # Fail the predictable ways now, not at minute 30. # LaunchResident starts each session on its first turn, so there is no # live session to check yet. Still fail fast on repository hygiene. - await c.preflight(clean_worktree=True) + await c.preflight(clean_worktree=True, min_isolation="supervised") # Independent attempts, in parallel — then seal the round. a, b = await asyncio.gather( diff --git a/examples/judge_panel_score.py b/examples/judge_panel_score.py index efe44a0..460b9da 100644 --- a/examples/judge_panel_score.py +++ b/examples/judge_panel_score.py @@ -24,7 +24,7 @@ async def main() -> None: - async with Conductor(".", "panel-demo", launcher="resident") as c: + async with Conductor(".", "panel-demo", launcher="resident", isolation="supervised") as c: workers = await asyncio.gather( c.hire("claude", runtime="claude", model="claude-haiku-4-5"), c.hire("codex", runtime="codex", model="gpt-5.4-mini"), diff --git a/examples/pipeline_score.py b/examples/pipeline_score.py index b0b1843..863d956 100644 --- a/examples/pipeline_score.py +++ b/examples/pipeline_score.py @@ -17,7 +17,7 @@ async def main() -> None: - async with Conductor(".", "pipeline-demo", launcher="resident") as c: + async with Conductor(".", "pipeline-demo", launcher="resident", isolation="supervised") as c: architect = await c.hire( "architect", runtime="claude", model="claude-haiku-4-5" ) diff --git a/examples/review_escalation.py b/examples/review_escalation.py index 0cc39a9..231495d 100644 --- a/examples/review_escalation.py +++ b/examples/review_escalation.py @@ -18,7 +18,7 @@ async def main(task: str) -> None: - async with Conductor(".", "escalation-demo", launcher="resident") as c: + async with Conductor(".", "escalation-demo", launcher="resident", isolation="supervised") as c: junior = await c.hire("junior", runtime="claude", model="claude-haiku-4-5") senior = await c.hire( "senior", runtime="claude", model="claude-haiku-4-5" diff --git a/examples/tournament.py b/examples/tournament.py index fa6db42..ecd1b38 100644 --- a/examples/tournament.py +++ b/examples/tournament.py @@ -27,7 +27,7 @@ async def match(run_id: str, contenders: list[tuple[str, str, str | None]]) -> tuple[str, str, str | None]: """One arena run; returns the winning seed.""" - async with Conductor(".", run_id, launcher="resident") as c: + async with Conductor(".", run_id, launcher="resident", isolation="supervised") as c: agents = { name: await c.hire(name, runtime=runtime, model=model) for name, runtime, model in contenders diff --git a/src/h5i/orchestra/_conductor.py b/src/h5i/orchestra/_conductor.py index eed6652..4bd98bb 100644 --- a/src/h5i/orchestra/_conductor.py +++ b/src/h5i/orchestra/_conductor.py @@ -81,6 +81,12 @@ class Conductor: itself), or ``"client"`` (every turn is delivered to ``on_turn`` in *this* process — how tests script agents, and how a score can spawn its own runtimes). Passing ``on_turn`` implies ``"client"``. + - ``isolation``: the run's default sandbox tier for hired agents' envs + (``"workspace"``, ``"process"``, ``"supervised"``, ``"container"``, …). + Every :meth:`hire` inherits it unless it passes its own. An explicit + tier is fail-closed — hire errors if the host cannot enforce it, never + silently downgrades; ``None`` auto-picks per env, like + ``h5i env create``. - ``turn_timeout``/``poll_interval``: seconds (floats fine). - ``score_digest``: provenance digest recorded at launch. Defaults to the sha256 of ``sys.argv[0]``; pass ``None`` to record nothing, or your own @@ -105,6 +111,7 @@ def __init__( max_rounds: int | None = None, actor: str | None = None, launcher: str | None = None, + isolation: str | None = None, on_turn: OnTurn | None = None, poll_interval: float | None = None, turn_timeout: float | None = None, @@ -125,6 +132,7 @@ def __init__( self._max_rounds = max_rounds self._actor = actor self._launcher = "client" if on_turn is not None else (launcher or "attach") + self._isolation = isolation self._on_turn = on_turn self._poll_interval = poll_interval self._turn_timeout = turn_timeout @@ -255,10 +263,21 @@ async def hire( runtime: str | None = None, model: str | None = None, profile: str | None = None, + isolation: str | None = None, env: str | None = None, ) -> "Agent": """Hire an agent into the run: create (or bind ``env``) its sandboxed - env and enroll it on the roster. Journaled — a resume rebinds.""" + env and enroll it on the roster. Journaled — a resume rebinds. + + ``isolation`` requests a sandbox tier for the created env + (``"workspace"``, ``"process"``, ``"supervised"``, ``"container"``, …), + defaulting to the conductor's run-level ``isolation``. An explicit + tier is fail-closed — hire errors if the host cannot enforce it, + never silently downgrades; pass ``"auto"`` to override a run-level + tier back to auto-picking. + """ + if isolation is None: + isolation = self._isolation params: dict[str, Any] = {"name": name} if runtime is not None: params["runtime"] = runtime @@ -266,6 +285,8 @@ async def hire( params["model"] = model if profile is not None: params["profile"] = profile + if isolation is not None: + params["isolation"] = isolation if env is not None: params["env"] = env seat = await self._request("agent.hire", params) diff --git a/tests/test_conductor.py b/tests/test_conductor.py index 10dc65b..bc638da 100644 --- a/tests/test_conductor.py +++ b/tests/test_conductor.py @@ -285,6 +285,31 @@ async def on_turn(_turn): await c.close() +async def test_hire_isolation_run_default_and_override(): + mock = MockOrchestra() + mock.on("agent.hire", lambda p: {"agent_id": p["name"], "env_id": "e"}) + c = await launch_conductor(mock, isolation="supervised") + try: + await c.hire("a", runtime="claude") # inherits the run tier + await c.hire("b", isolation="container") # explicit override + await c.hire("c", isolation="auto") # back to auto-picking + tiers = [p.get("isolation") for p in mock.calls_to("agent.hire")] + assert tiers == ["supervised", "container", "auto"] + finally: + await c.close() + + # Without a run-level tier, hire sends nothing (server auto-picks). + mock = MockOrchestra() + mock.on("agent.hire", lambda p: {"agent_id": p["name"], "env_id": "e"}) + c = await launch_conductor(mock) + try: + await c.hire("a") + (hire,) = mock.calls_to("agent.hire") + assert "isolation" not in hire + finally: + await c.close() + + async def test_misc_surface_marshaling(): mock = MockOrchestra() mock.on("conductor.freeze", lambda p: {"id": "testrun", "phase": "sealed_submit"}) From a263d4ec66feb5768d6aec6434f2453c3d96ef70 Mon Sep 17 00:00:00 2001 From: Koukyosyumei Date: Mon, 13 Jul 2026 22:01:14 -0400 Subject: [PATCH 6/8] feat(examples): every example takes a task arg, defaulting to the quicksort demo All eight examples now accept the task as an optional CLI argument and fall back to the same demo task, 'implement quicksort with pytest', so a bare invocation always works. The previously hard-coded h5i-specific tasks (judge_panel, tournament, pipeline, debate, custom_control_flow) are parameterized: pipeline's architect designs the given task, debate derives its question from it (simplest-vs-robust), and custom_control_flow decomposes it into subtasks instead of triaging test hotspots. Verify commands switch from 'cargo test --quiet' to 'pytest -q' to match the demo task (submissions carry their own tests); the README notes to change verify= when passing a task from another ecosystem. --- examples/README.md | 21 +++++++++++---------- examples/arena_score.py | 8 +++++--- examples/custom_control_flow.py | 19 ++++++++++++------- examples/debate_then_build.py | 27 +++++++++++++++++---------- examples/ensemble_score.py | 10 ++++++---- examples/judge_panel_score.py | 13 +++++++------ examples/pipeline_score.py | 15 ++++++++------- examples/review_escalation.py | 7 ++++--- examples/tournament.py | 23 +++++++++++++---------- 9 files changed, 83 insertions(+), 60 deletions(-) diff --git a/examples/README.md b/examples/README.md index f2535cc..853ecb6 100644 --- a/examples/README.md +++ b/examples/README.md @@ -43,18 +43,14 @@ sessions must already be parked on their inboxes. Do not use that check before the first turn with `launcher="resident"`: resident sessions are started lazily when a turn is dispatched. -Then run any example as a plain Python script. Three take the task as an -optional CLI argument (falling back to a demo task): - -```bash -python examples/ensemble_score.py "implement quicksort" -python examples/arena_score.py "implement quicksort" -python examples/review_escalation.py "implement quicksort" -``` - -The rest are self-contained — the task is written into the score: +Then run any example as a plain Python script. Every example takes the task +as an optional CLI argument and falls back to the same demo task, +**`implement quicksort with pytest`** — so a bare invocation always works: ```bash +python examples/ensemble_score.py # demo task +python examples/arena_score.py "implement quicksort with pytest" +python examples/review_escalation.py "fix the flaky msg_integration test" python examples/pipeline_score.py python examples/judge_panel_score.py python examples/debate_then_build.py @@ -63,6 +59,11 @@ python examples/custom_control_flow.py # uses the default "attach" launcher: # park resident sessions yourself first ``` +To match the demo task, the scores verify candidates with `pytest -q` — the +quicksort submissions carry their own tests. If you pass a task from another +ecosystem, change the `verify` command in the score too (e.g. back to +`["cargo", "test", "--quiet"]`). + Every example pins inexpensive models instead of inheriting a potentially costly CLI default: Claude seats use `claude-haiku-4-5` and Codex seats use `gpt-5.4-mini`. Edit the `model=` arguments if your account lacks either model. diff --git a/examples/arena_score.py b/examples/arena_score.py index 1f4e8d8..18c472c 100644 --- a/examples/arena_score.py +++ b/examples/arena_score.py @@ -7,7 +7,7 @@ the smallest green diff. The compare rows are the same arena view `h5i team compare` renders. - python examples/arena_score.py "make `h5i doctor` exit non-zero on repair failures" + python examples/arena_score.py [""] # default: implement quicksort with pytest """ import asyncio @@ -15,6 +15,8 @@ from h5i.orchestra import Conductor, patterns +DEMO_TASK = "implement quicksort with pytest" + async def main(task: str) -> None: async with Conductor(".", "arena-demo", launcher="resident", isolation="supervised") as c: @@ -31,7 +33,7 @@ async def main(task: str) -> None: c, task, agents, - verify=["cargo", "test", "--quiet"], + verify=["pytest", "-q"], isolation="process", ) @@ -54,4 +56,4 @@ async def main(task: str) -> None: if __name__ == "__main__": - asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else "demo task")) + asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else DEMO_TASK)) diff --git a/examples/custom_control_flow.py b/examples/custom_control_flow.py index e4fa2a0..c2022c5 100644 --- a/examples/custom_control_flow.py +++ b/examples/custom_control_flow.py @@ -3,14 +3,19 @@ This score shows the pieces no manifest could express: data turns feeding `if`, a journaled fan-out over a dynamic work list, a custom (LLM-assisted) verdict policy, and a mid-run score migration marker. + + python examples/custom_control_flow.py [""] # default: implement quicksort with pytest """ import asyncio +import sys from h5i.orchestra import Conductor, Verdict, patterns +DEMO_TASK = "implement quicksort with pytest" + -async def main() -> None: +async def main(task: str) -> None: async with Conductor(".", "triage-and-fix", isolation="supervised") as c: lead = await c.hire( "lead", runtime="claude", model="claude-haiku-4-5" @@ -25,26 +30,26 @@ async def main() -> None: # A data turn: the agent replies with JSON, not code. The reply is # journaled — on resume this list comes back without re-asking. hotspots: list = await lead.ask( - "List up to 4 modules with failing or flaky tests as a JSON " - 'array: [{"path": "...", "symptom": "..."}]', + f"We want to: {task}. Break it into at most 4 independent " + 'subtasks as a JSON array: [{"item": "...", "detail": "..."}]', parse=lambda v: list(v), ) if not hotspots: - print("nothing to fix") + print("nothing to do") return # Journal any host-side effect exactly-once; distinct labels for # parallel loops come from scopes. for i, spot in enumerate(hotspots): await c.scope(f"triage/{i}").step( - "log", lambda spot=spot: f"queued {spot['path']}" + "log", lambda spot=spot: f"queued {spot['item']}" ) # Dynamic fan-out: assignments follow the data, round-robin. outcome = await patterns.map_reduce( c, [ - (crew[i % len(crew)], f"fix `{s['path']}`: {s['symptom']}") + (crew[i % len(crew)], f"{s['item']}: {s['detail']} (part of: {task})") for i, s in enumerate(hotspots) ], reduce=(lead, "merge every fix into one coherent candidate"), @@ -82,4 +87,4 @@ def only_if_green(run) -> Verdict: if __name__ == "__main__": - asyncio.run(main()) + asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else DEMO_TASK)) diff --git a/examples/debate_then_build.py b/examples/debate_then_build.py index 4b380f3..48b2298 100644 --- a/examples/debate_then_build.py +++ b/examples/debate_then_build.py @@ -5,20 +5,27 @@ Debate is pure `ask` (no artifacts, no freeze), so the winning side can go straight into a work turn afterwards, in the same run and journal. - python examples/debate_then_build.py + python examples/debate_then_build.py [""] # default: implement quicksort with pytest """ import asyncio +import sys from h5i.orchestra import Conductor, patterns -QUESTION = ( - "Should `h5i msg wait` grow a --push webhook mode, or stay poll-only? " - "pro argues for the webhook; con argues for polling." -) +DEMO_TASK = "implement quicksort with pytest" -async def main() -> None: +def question_for(task: str) -> str: + return ( + f"What is the right way to '{task}'? pro argues for the simplest " + "implementation that could work; con argues for the most robust, " + "test-heavy one." + ) + + +async def main(task: str) -> None: + question = question_for(task) async with Conductor(".", "debate-demo", launcher="resident", isolation="supervised") as c: pro = await c.hire("pro", runtime="claude", model="claude-haiku-4-5") con = await c.hire("con", runtime="codex", model="gpt-5.4-mini") @@ -27,7 +34,7 @@ async def main() -> None: ) outcome = await patterns.debate( - c, QUESTION, [pro, con], moderator=moderator, rounds=2 + c, question, [pro, con], moderator=moderator, rounds=2 ) for who, argument in outcome.transcript: print(f"[{who}] {argument[:120]}…") @@ -42,8 +49,8 @@ async def main() -> None: artifact, risks = await asyncio.gather( winner.work( - f"You won this debate: {QUESTION}\nYour side prevailed with: " - f"{conclusion.rationale}\nImplement your position." + f"You won this debate: {question}\nYour side prevailed with: " + f"{conclusion.rationale}\nNow do the task your way: {task}" ), loser.ask( "You lost the debate. List the 3 biggest risks of the winning " @@ -57,4 +64,4 @@ async def main() -> None: if __name__ == "__main__": - asyncio.run(main()) + asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else DEMO_TASK)) diff --git a/examples/ensemble_score.py b/examples/ensemble_score.py index 16892df..9ee9e05 100644 --- a/examples/ensemble_score.py +++ b/examples/ensemble_score.py @@ -6,7 +6,7 @@ attached (bring them up with `launcher="resident"` to let the score spawn tmux sessions itself). - python examples/ensemble_score.py "implement `h5i pull` mirroring `h5i push`" + python examples/ensemble_score.py [""] # default: implement quicksort with pytest """ import asyncio @@ -14,6 +14,8 @@ from h5i.orchestra import Conductor +DEMO_TASK = "implement quicksort with pytest" + async def main(task: str) -> None: async with Conductor(".", "ensemble-demo", launcher="resident", isolation="supervised") as c: @@ -45,8 +47,8 @@ async def main(task: str) -> None: # Neutral verification in fresh sandboxed worktrees (never the # author's box), then the built-in verdict rule. - await c.verify(a, ["cargo", "test", "--quiet"]) - await c.verify(b, ["cargo", "test", "--quiet"]) + await c.verify(a, ["pytest", "-q"]) + await c.verify(b, ["pytest", "-q"]) verdict = await c.judge() print(await c.trace()) @@ -68,4 +70,4 @@ async def main(task: str) -> None: if __name__ == "__main__": - asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else "demo task")) + asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else DEMO_TASK)) diff --git a/examples/judge_panel_score.py b/examples/judge_panel_score.py index 460b9da..5ef8b19 100644 --- a/examples/judge_panel_score.py +++ b/examples/judge_panel_score.py @@ -9,21 +9,22 @@ Judges are read-only seats that must be hired before the round seals — alongside the workers, exactly like the roster note in patterns.py says. - python examples/judge_panel_score.py + python examples/judge_panel_score.py [""] # default: implement quicksort with pytest """ import asyncio +import sys from h5i.orchestra import Conductor, patterns -TASK = "reduce `h5i team status` latency without changing its output" +DEMO_TASK = "implement quicksort with pytest" RUBRIC = ( "prefer the smallest change that demonstrably keeps behavior identical; " "penalize speculative refactors and untested hot paths" ) -async def main() -> None: +async def main(task: str) -> None: async with Conductor(".", "panel-demo", launcher="resident", isolation="supervised") as c: workers = await asyncio.gather( c.hire("claude", runtime="claude", model="claude-haiku-4-5"), @@ -36,11 +37,11 @@ async def main() -> None: # Independent attempts → seal → neutral evidence for the panel. artifacts = await asyncio.gather( - *(w.work(TASK, expect_independent=True) for w in workers) + *(w.work(task, expect_independent=True) for w in workers) ) await c.freeze() for artifact in artifacts: - await c.verify(artifact, ["cargo", "test", "--quiet"]) + await c.verify(artifact, ["pytest", "-q"]) outcome = await patterns.judge_panel(c, RUBRIC, judges) @@ -56,4 +57,4 @@ async def main() -> None: if __name__ == "__main__": - asyncio.run(main()) + asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else DEMO_TASK)) diff --git a/examples/pipeline_score.py b/examples/pipeline_score.py index 863d956..3d9fcb5 100644 --- a/examples/pipeline_score.py +++ b/examples/pipeline_score.py @@ -6,17 +6,18 @@ Stage 1 works pre-freeze; the round seals right after it, because materials ride the sealed-phase-only discuss channel. - python examples/pipeline_score.py + python examples/pipeline_score.py [""] # default: implement quicksort with pytest """ import asyncio +import sys from h5i.orchestra import Conductor, patterns -FEATURE = "add `--json` output to `h5i vibe`" +DEMO_TASK = "implement quicksort with pytest" -async def main() -> None: +async def main(task: str) -> None: async with Conductor(".", "pipeline-demo", launcher="resident", isolation="supervised") as c: architect = await c.hire( "architect", runtime="claude", model="claude-haiku-4-5" @@ -31,8 +32,8 @@ async def main() -> None: [ ( architect, - f"Design {FEATURE}: write docs/design-vibe-json.md with the " - "schema, flag semantics, and test plan. Submit only the doc.", + f"Design this task: {task}. Write docs/design.md with the " + "approach, interface, and test plan. Submit only the doc.", ), ( builder, @@ -51,11 +52,11 @@ async def main() -> None: print("stages:", [(a.owner_agent, a.id) for a in (design, impl, hardened)]) # The final artifact carries the whole chain; verify and gate that one. - verification = await c.verify(hardened, ["cargo", "test", "--quiet"]) + verification = await c.verify(hardened, ["pytest", "-q"]) print("tests passed:", verification.tests_passed) if verification.tests_passed and (await c.gate("apply the pipeline result?")).approved: await c.apply(hardened, force=True) # force: our gate is the verdict if __name__ == "__main__": - asyncio.run(main()) + asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else DEMO_TASK)) diff --git a/examples/review_escalation.py b/examples/review_escalation.py index 231495d..d0a9c3d 100644 --- a/examples/review_escalation.py +++ b/examples/review_escalation.py @@ -6,7 +6,7 @@ for. Both seats are hired up front (enrollment is open-round-only), so the escalation path exists from the start whether or not it's taken. - python examples/review_escalation.py "fix the flaky msg_integration test" + python examples/review_escalation.py [""] # default: implement quicksort with pytest """ import asyncio @@ -14,6 +14,7 @@ from h5i.orchestra import Conductor +DEMO_TASK = "implement quicksort with pytest" MAX_REVISE_CYCLES = 2 @@ -51,7 +52,7 @@ async def main(task: str) -> None: materials=[artifact], ) - verification = await c.verify(artifact, ["cargo", "test", "--quiet"]) + verification = await c.verify(artifact, ["pytest", "-q"]) print( f"final candidate by {artifact.owner_agent}: " f"tests_passed={verification.tests_passed}" @@ -59,4 +60,4 @@ async def main(task: str) -> None: if __name__ == "__main__": - asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else "demo task")) + asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else DEMO_TASK)) diff --git a/examples/tournament.py b/examples/tournament.py index ecd1b38..877628a 100644 --- a/examples/tournament.py +++ b/examples/tournament.py @@ -6,15 +6,16 @@ means a killed bracket resumes mid-tournament — finished matches replay their recorded verdicts instantly. - python examples/tournament.py + python examples/tournament.py [""] # default: implement quicksort with pytest """ import asyncio +import sys from h5i.orchestra import Conductor, patterns -TASK = "make `h5i log --limit 0` print every record instead of none" -VERIFY = ["cargo", "test", "--quiet"] +DEMO_TASK = "implement quicksort with pytest" +VERIFY = ["pytest", "-q"] #: (name, runtime, model) seeds, bracket order. SEEDS = [ @@ -25,14 +26,16 @@ ] -async def match(run_id: str, contenders: list[tuple[str, str, str | None]]) -> tuple[str, str, str | None]: +async def match( + run_id: str, task: str, contenders: list[tuple[str, str, str | None]] +) -> tuple[str, str, str | None]: """One arena run; returns the winning seed.""" async with Conductor(".", run_id, launcher="resident", isolation="supervised") as c: agents = { name: await c.hire(name, runtime=runtime, model=model) for name, runtime, model in contenders } - outcome = await patterns.arena(c, TASK, list(agents.values()), verify=VERIFY) + outcome = await patterns.arena(c, task, list(agents.values()), verify=VERIFY) verdict = outcome.verdict assert verdict is not None winner_artifact = next( @@ -45,16 +48,16 @@ async def match(run_id: str, contenders: list[tuple[str, str, str | None]]) -> t return winner -async def main() -> None: +async def main(task: str) -> None: # Semifinals run concurrently — separate runs, separate envs, no shared # journal labels to collide. finalist_a, finalist_b = await asyncio.gather( - match("bracket-semi-1", [SEEDS[0], SEEDS[3]]), - match("bracket-semi-2", [SEEDS[1], SEEDS[2]]), + match("bracket-semi-1", task, [SEEDS[0], SEEDS[3]]), + match("bracket-semi-2", task, [SEEDS[1], SEEDS[2]]), ) - champion = await match("bracket-final", [finalist_a, finalist_b]) + champion = await match("bracket-final", task, [finalist_a, finalist_b]) print("champion:", champion[0]) if __name__ == "__main__": - asyncio.run(main()) + asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else DEMO_TASK)) From a6bf5a55ba2e10c6a0301e1d04d1f439f172c51d Mon Sep 17 00:00:00 2001 From: Koukyosyumei Date: Mon, 13 Jul 2026 22:23:56 -0400 Subject: [PATCH 7/8] feat(examples): codex seats run at medium reasoning effort Conductor.hire grows an effort= param (recorded on the roster seat; codex sessions launch with -c model_reasoning_effort=, overriding the box's config.toml). Every codex hire in the examples pins effort="medium" so a user-level 'high' default doesn't slow the demos down; claude seats stay effort-free since that adapter has no effort knob and fails closed. --- examples/README.md | 4 +++- examples/arena_score.py | 2 +- examples/debate_then_build.py | 2 +- examples/ensemble_score.py | 2 +- examples/judge_panel_score.py | 2 +- examples/pipeline_score.py | 2 +- examples/tournament.py | 8 +++++++- src/h5i/orchestra/_conductor.py | 8 ++++++++ tests/test_conductor.py | 3 +++ 9 files changed, 26 insertions(+), 7 deletions(-) diff --git a/examples/README.md b/examples/README.md index 853ecb6..cd4a430 100644 --- a/examples/README.md +++ b/examples/README.md @@ -66,7 +66,9 @@ ecosystem, change the `verify` command in the score too (e.g. back to Every example pins inexpensive models instead of inheriting a potentially costly CLI default: Claude seats use `claude-haiku-4-5` and Codex seats use -`gpt-5.4-mini`. Edit the `model=` arguments if your account lacks either model. +`gpt-5.4-mini` with `effort="medium"` (launched as +`-c model_reasoning_effort=medium`, which wins over your `~/.codex/config.toml` +— so a `high` default there won't slow the demos down). Edit the `model=` arguments if your account lacks either model. Changing a model on an existing run does not rewrite a journaled hire, so also change the run id (for example, `"ensemble-demo-v2"`) when experimenting with another model. To resume an interrupted run without configuration changes, diff --git a/examples/arena_score.py b/examples/arena_score.py index 18c472c..a5bf70e 100644 --- a/examples/arena_score.py +++ b/examples/arena_score.py @@ -22,7 +22,7 @@ async def main(task: str) -> None: async with Conductor(".", "arena-demo", launcher="resident", isolation="supervised") as c: agents = await asyncio.gather( c.hire("claude", runtime="claude", model="claude-haiku-4-5"), - c.hire("codex", runtime="codex", model="gpt-5.4-mini"), + c.hire("codex", runtime="codex", model="gpt-5.4-mini", effort="medium"), c.hire("haiku", runtime="claude", model="claude-haiku-4-5"), ) # LaunchResident starts each session on its first turn, so there is no diff --git a/examples/debate_then_build.py b/examples/debate_then_build.py index 48b2298..42b642a 100644 --- a/examples/debate_then_build.py +++ b/examples/debate_then_build.py @@ -28,7 +28,7 @@ async def main(task: str) -> None: question = question_for(task) async with Conductor(".", "debate-demo", launcher="resident", isolation="supervised") as c: pro = await c.hire("pro", runtime="claude", model="claude-haiku-4-5") - con = await c.hire("con", runtime="codex", model="gpt-5.4-mini") + con = await c.hire("con", runtime="codex", model="gpt-5.4-mini", effort="medium") moderator = await c.hire( "moderator", runtime="claude", model="claude-haiku-4-5" ) diff --git a/examples/ensemble_score.py b/examples/ensemble_score.py index 9ee9e05..79cb51d 100644 --- a/examples/ensemble_score.py +++ b/examples/ensemble_score.py @@ -22,7 +22,7 @@ async def main(task: str) -> None: claude = await c.hire( "claude", runtime="claude", model="claude-haiku-4-5" ) - codex = await c.hire("codex", runtime="codex", model="gpt-5.4-mini") + codex = await c.hire("codex", runtime="codex", model="gpt-5.4-mini", effort="medium") # Fail the predictable ways now, not at minute 30. # LaunchResident starts each session on its first turn, so there is no diff --git a/examples/judge_panel_score.py b/examples/judge_panel_score.py index 5ef8b19..15ffa65 100644 --- a/examples/judge_panel_score.py +++ b/examples/judge_panel_score.py @@ -28,7 +28,7 @@ async def main(task: str) -> None: async with Conductor(".", "panel-demo", launcher="resident", isolation="supervised") as c: workers = await asyncio.gather( c.hire("claude", runtime="claude", model="claude-haiku-4-5"), - c.hire("codex", runtime="codex", model="gpt-5.4-mini"), + c.hire("codex", runtime="codex", model="gpt-5.4-mini", effort="medium"), ) judges = await asyncio.gather( c.hire("judge-a", runtime="claude", model="claude-haiku-4-5"), diff --git a/examples/pipeline_score.py b/examples/pipeline_score.py index 3d9fcb5..fae70d5 100644 --- a/examples/pipeline_score.py +++ b/examples/pipeline_score.py @@ -22,7 +22,7 @@ async def main(task: str) -> None: architect = await c.hire( "architect", runtime="claude", model="claude-haiku-4-5" ) - builder = await c.hire("builder", runtime="codex", model="gpt-5.4-mini") + builder = await c.hire("builder", runtime="codex", model="gpt-5.4-mini", effort="medium") hardener = await c.hire( "hardener", runtime="claude", model="claude-haiku-4-5" ) diff --git a/examples/tournament.py b/examples/tournament.py index 877628a..806468e 100644 --- a/examples/tournament.py +++ b/examples/tournament.py @@ -32,7 +32,13 @@ async def match( """One arena run; returns the winning seed.""" async with Conductor(".", run_id, launcher="resident", isolation="supervised") as c: agents = { - name: await c.hire(name, runtime=runtime, model=model) + name: await c.hire( + name, + runtime=runtime, + model=model, + # Keep codex seats cheap; claude has no effort knob. + effort="medium" if runtime == "codex" else None, + ) for name, runtime, model in contenders } outcome = await patterns.arena(c, task, list(agents.values()), verify=VERIFY) diff --git a/src/h5i/orchestra/_conductor.py b/src/h5i/orchestra/_conductor.py index 4bd98bb..70ac9b3 100644 --- a/src/h5i/orchestra/_conductor.py +++ b/src/h5i/orchestra/_conductor.py @@ -262,6 +262,7 @@ async def hire( *, runtime: str | None = None, model: str | None = None, + effort: str | None = None, profile: str | None = None, isolation: str | None = None, env: str | None = None, @@ -269,6 +270,11 @@ async def hire( """Hire an agent into the run: create (or bind ``env``) its sandboxed env and enroll it on the roster. Journaled — a resume rebinds. + ``effort`` records a reasoning-effort override on the roster seat + (codex sessions launch with ``-c model_reasoning_effort=``, + which wins over the box's ``config.toml``). Runtimes without an + effort knob fail closed at launch rather than silently ignoring it. + ``isolation`` requests a sandbox tier for the created env (``"workspace"``, ``"process"``, ``"supervised"``, ``"container"``, …), defaulting to the conductor's run-level ``isolation``. An explicit @@ -283,6 +289,8 @@ async def hire( params["runtime"] = runtime if model is not None: params["model"] = model + if effort is not None: + params["effort"] = effort if profile is not None: params["profile"] = profile if isolation is not None: diff --git a/tests/test_conductor.py b/tests/test_conductor.py index bc638da..e29a478 100644 --- a/tests/test_conductor.py +++ b/tests/test_conductor.py @@ -78,6 +78,9 @@ async def test_work_round_trip_preserves_unknown_fields(): (hire,) = mock.calls_to("agent.hire") assert hire == {"name": "claude", "runtime": "claude", "model": "opus"} + await c.hire("codex", runtime="codex", model="gpt-5.4-mini", effort="medium") + assert mock.calls_to("agent.hire")[-1]["effort"] == "medium" + artifact = await claude.work("do it", expect_independent=True) (work,) = mock.calls_to("agent.work") assert work["expect_independent"] is True From 74bf718f52cbd8f34025f043eeaad69a51a8b2f2 Mon Sep 17 00:00:00 2001 From: Koukyosyumei Date: Tue, 14 Jul 2026 16:38:57 -0400 Subject: [PATCH 8/8] fix(orchestra): import Any in _watch, rename ambiguous 'l' in test_watch --- src/h5i/orchestra/_watch.py | 2 +- tests/test_watch.py | 16 ++++++++-------- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/src/h5i/orchestra/_watch.py b/src/h5i/orchestra/_watch.py index 4383f84..1a5a62f 100644 --- a/src/h5i/orchestra/_watch.py +++ b/src/h5i/orchestra/_watch.py @@ -32,7 +32,7 @@ import shlex import shutil import sys -from typing import Callable, Mapping, NamedTuple +from typing import Any, Callable, Mapping, NamedTuple __all__ = ["Opener", "SessionWatcher", "resolve_opener", "session_prefix"] diff --git a/tests/test_watch.py b/tests/test_watch.py index d2288ba..311a34e 100644 --- a/tests/test_watch.py +++ b/tests/test_watch.py @@ -168,14 +168,14 @@ async def _spawn(self, argv): ok = RealSpawn("run1", [[session]]) ok._opener = Opener("true", "spawn", lambda s: ["true"]) await ok.poll_once() - assert any("opened in true" in l for l in ok.echoed) + assert any("opened in true" in line for line in ok.echoed) # `false` exits non-zero fast → the viewer failed; degrade to the hint. bad = RealSpawn("run1", [[session]]) bad._opener = Opener("false", "spawn", lambda s: ["false"]) await bad.poll_once() - assert any(f"tmux attach -t {session}" in l for l in bad.echoed), bad.echoed - assert not any("opened in false" in l for l in bad.echoed) + assert any(f"tmux attach -t {session}" in line for line in bad.echoed), bad.echoed + assert not any("opened in false" in line for line in bad.echoed) @pytest.mark.asyncio @@ -201,7 +201,7 @@ async def test_warns_once_when_an_expected_session_never_appears(): await asyncio.sleep(0.01) # move past the zero grace period for _ in range(3): await w.poll_once() - warnings = [l for l in w.echoed if "has not appeared" in l] + warnings = [line for line in w.echoed if "has not appeared" in line] assert len(warnings) == 1, w.echoed assert "h5i env shell env/claude/run1-claude -- true" in warnings[0] @@ -213,7 +213,7 @@ async def test_no_warning_while_the_expected_session_is_up(): w.expect("claude", "env/x") for _ in range(2): await w.poll_once() - assert not any("WARNING" in l for l in w.echoed) + assert not any("WARNING" in line for line in w.echoed) @pytest.mark.asyncio @@ -223,10 +223,10 @@ async def test_session_dying_mid_turn_is_a_warning(): w.expect("claude", "env/x") for _ in range(3): await w.poll_once() - died = [l for l in w.echoed if "ended while its turn is still pending" in l] + died = [line for line in w.echoed if "ended while its turn is still pending" in line] assert len(died) == 1, w.echoed # ...and the death warning is not doubled by the never-appeared check. - assert not any("has not appeared" in l for l in w.echoed) + assert not any("has not appeared" in line for line in w.echoed) @pytest.mark.asyncio @@ -237,7 +237,7 @@ async def test_unexpect_silences_the_deadline(): await asyncio.sleep(0.01) for _ in range(2): await w.poll_once() - assert not any("WARNING" in l for l in w.echoed) + assert not any("WARNING" in line for line in w.echoed) @pytest.mark.asyncio