diff --git a/README.md b/README.md
index dece0d6..d485713 100644
--- a/README.md
+++ b/README.md
@@ -110,6 +110,22 @@ 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
+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..cd4a430 100644
--- a/examples/README.md
+++ b/examples/README.md
@@ -5,9 +5,19 @@ 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. 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`.
+
 ## 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 <h5i repo>`),
@@ -19,24 +29,28 @@ 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
 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
@@ -45,9 +59,16 @@ 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.
+`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 9586054..a5bf70e 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 ["<task>"]   # default: implement quicksort with pytest
 """
 
 import asyncio
@@ -15,23 +15,25 @@
 
 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") 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"),
+            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
         # 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,
             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 9502f6b..c2022c5 100644
--- a/examples/custom_control_flow.py
+++ b/examples/custom_control_flow.py
@@ -3,15 +3,20 @@
 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 ["<task>"]   # 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 with Conductor(".", "triage-and-fix") as c:
+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 2003da5..42b642a 100644
--- a/examples/debate_then_build.py
+++ b/examples/debate_then_build.py
@@ -5,29 +5,36 @@
 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 ["<task>"]   # 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:
-    async with Conductor(".", "debate-demo", launcher="resident") as c:
+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")
+        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"
         )
 
         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 8ed67d7..79cb51d 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 ["<task>"]   # default: implement quicksort with pytest
 """
 
 import asyncio
@@ -14,18 +14,20 @@
 
 from h5i.orchestra import Conductor
 
+DEMO_TASK = "implement quicksort with pytest"
+
 
 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"
         )
-        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
         # 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(
@@ -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 efe44a0..15ffa65 100644
--- a/examples/judge_panel_score.py
+++ b/examples/judge_panel_score.py
@@ -9,25 +9,26 @@
 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 ["<task>"]   # 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 with Conductor(".", "panel-demo", launcher="resident") as c:
+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"),
@@ -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 b0b1843..fae70d5 100644
--- a/examples/pipeline_score.py
+++ b/examples/pipeline_score.py
@@ -6,22 +6,23 @@
 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 ["<task>"]   # 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 with Conductor(".", "pipeline-demo", launcher="resident") as c:
+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"
         )
-        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"
         )
@@ -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 0cc39a9..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 ["<task>"]   # default: implement quicksort with pytest
 """
 
 import asyncio
@@ -14,11 +14,12 @@
 
 from h5i.orchestra import Conductor
 
+DEMO_TASK = "implement quicksort with pytest"
 MAX_REVISE_CYCLES = 2
 
 
 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"
@@ -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 fa6db42..806468e 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 ["<task>"]   # 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,22 @@
 ]
 
 
-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") as c:
+    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)
+        outcome = await patterns.arena(c, task, list(agents.values()), verify=VERIFY)
         verdict = outcome.verdict
         assert verdict is not None
         winner_artifact = next(
@@ -45,16 +54,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))
diff --git a/src/h5i/orchestra/_conductor.py b/src/h5i/orchestra/_conductor.py
index 6dddf20..70ac9b3 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 (
@@ -79,10 +81,24 @@ 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
       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__(
@@ -95,11 +111,13 @@ 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,
         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)")
@@ -114,11 +132,15 @@ 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
         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._session_watcher: SessionWatcher | None = None
         self._bridge: Bridge | None = None
         self._run_id: str | None = None
         self._actor_resolved: str | None = None
@@ -195,11 +217,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:
+            self._session_watcher = SessionWatcher(
+                self._run_id,
+                template=self._watch if isinstance(self._watch, str) else None,
+            )
+            self._watch_task = asyncio.ensure_future(self._session_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()
@@ -227,18 +262,39 @@ 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,
     ) -> "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.
+
+        ``effort`` records a reasoning-effort override on the roster seat
+        (codex sessions launch with ``-c model_reasoning_effort=<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
+        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
         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:
+            params["isolation"] = isolation
         if env is not None:
             params["env"] = env
         seat = await self._request("agent.hire", params)
@@ -424,6 +480,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})
@@ -493,7 +564,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(
@@ -518,9 +591,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
@@ -540,20 +615,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,
@@ -561,5 +638,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
new file mode 100644
index 0000000..1a5a62f
--- /dev/null
+++ b/src/h5i/orchestra/_watch.py
@@ -0,0 +1,311 @@
+"""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-<run>-<agent>``, 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 <session>`` 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 Any, 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,
+        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).
+        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 = (
+            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)
+
+    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):
+            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 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
+
+    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
+        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 '{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 '{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] agent '{agent}' session up, but its viewer failed ({e}) — "
+                f"attach with: tmux attach -t {session}"
+            )
+            return
+        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(
+            *argv,
+            stdin=asyncio.subprocess.DEVNULL,
+            stdout=asyncio.subprocess.DEVNULL,
+            stderr=asyncio.subprocess.DEVNULL,
+            start_new_session=True,
+        )
+        # 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(
+            "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_conductor.py b/tests/test_conductor.py
index 10dc65b..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
@@ -285,6 +288,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"})
diff --git a/tests/test_watch.py b/tests/test_watch.py
new file mode 100644
index 0000000..311a34e
--- /dev/null
+++ b/tests/test_watch.py
@@ -0,0 +1,305 @@
+"""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] = []
+        kwargs.setdefault("spawn_gap", 0.0)
+        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]
+    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
+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_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 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 line for line in bad.echoed), bad.echoed
+    assert not any("opened in false" in line for line in bad.echoed)
+
+
+@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)
+
+
+# ── 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 = [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]
+
+
+@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 line for line 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 = [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 line for line 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 line for line 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 ─────────────────────────────────────────────────────────
+
+
+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