fix(ui): drain coordinator async agents in interactive backends

Issue #195: in coordinator mode, the React TUI and Textual app never
injected `<task-notification>` envelopes after a worker finished.
`run_print_mode` had `_drain_coordinator_async_agents` wired in, but
`ReactBackendHost._process_line` and `TextualApp._process_line` only
called `handle_line` and returned, leaving the contract documented in
the coordinator system prompt (results arrive between turns) unmet.

Extract the drain helpers into `openharness.ui.coordinator_drain` and
invoke `drain_coordinator_async_agents` from both interactive hosts
after `handle_line` when coordinator mode is active. The shared module
also avoids a circular import between `ui.app` and `ui.backend_host`.

src/openharness/ui/app.py

@@ -6,25 +6,16 @@ import asyncio
 import json
 import sys
 
-from openharness.coordinator.coordinator_mode import (
-    TaskNotification,
-    format_task_notification,
-    is_coordinator_mode,
-)
-from openharness.engine.query import MaxTurnsExceeded
-from openharness.prompts.context import build_runtime_system_prompt
-from openharness.tasks.manager import get_task_manager
+from openharness.coordinator.coordinator_mode import is_coordinator_mode
 
 from openharness.api.client import SupportsStreamingMessages
 from openharness.engine.stream_events import StreamEvent
 from openharness.ui.backend_host import run_backend_host
+from openharness.ui.coordinator_drain import drain_coordinator_async_agents
 from openharness.ui.react_launcher import launch_react_tui
 from openharness.ui.runtime import build_runtime, close_runtime, handle_line, start_runtime
 
 
-_TERMINAL_TASK_STATUSES = frozenset({"completed", "failed", "killed"})
-
-
 def _decode_task_worker_line(raw: str) -> str:
     """Normalize one stdin line for the headless task worker.
 
@@ -46,166 +37,6 @@ def _decode_task_worker_line(raw: str) -> str:
     return stripped
 
 
-def _async_agent_task_entries(tool_metadata: dict[str, object] | None) -> list[dict[str, object]]:
-    if not isinstance(tool_metadata, dict):
-        return []
-    value = tool_metadata.get("async_agent_tasks")
-    if not isinstance(value, list):
-        return []
-    return [entry for entry in value if isinstance(entry, dict)]
-
-
-def _pending_async_agent_entries(tool_metadata: dict[str, object] | None) -> list[dict[str, object]]:
-    pending: list[dict[str, object]] = []
-    for entry in _async_agent_task_entries(tool_metadata):
-        task_id = str(entry.get("task_id") or "").strip()
-        if not task_id:
-            continue
-        if bool(entry.get("notification_sent")):
-            continue
-        pending.append(entry)
-    return pending
-
-
-def _build_async_task_summary(entry: dict[str, object], *, task_status: str, return_code: int | None) -> str:
-    description = str(entry.get("description") or entry.get("agent_id") or "background task").strip()
-    if task_status == "completed":
-        return f'Agent "{description}" completed'
-    if task_status == "killed":
-        return f'Agent "{description}" was stopped'
-    if return_code is not None:
-        return f'Agent "{description}" failed with exit code {return_code}'
-    return f'Agent "{description}" failed'
-
-
-async def _wait_for_completed_async_agent_entries(
-    tool_metadata: dict[str, object] | None,
-    *,
-    poll_interval_seconds: float = 0.1,
-) -> list[dict[str, object]]:
-    manager = get_task_manager()
-    while True:
-        pending = _pending_async_agent_entries(tool_metadata)
-        if not pending:
-            return []
-        completed: list[dict[str, object]] = []
-        for entry in pending:
-            task_id = str(entry.get("task_id") or "").strip()
-            task = manager.get_task(task_id)
-            if task is None:
-                entry["notification_sent"] = True
-                entry["status"] = "missing"
-                continue
-            entry["status"] = task.status
-            if task.status in _TERMINAL_TASK_STATUSES:
-                entry["return_code"] = task.return_code
-                completed.append(entry)
-        if completed:
-            return completed
-        await asyncio.sleep(poll_interval_seconds)
-
-
-def _format_completed_task_notifications(completed: list[dict[str, object]]) -> str:
-    manager = get_task_manager()
-    notifications: list[str] = []
-    for entry in completed:
-        task_id = str(entry.get("task_id") or "").strip()
-        agent_id = str(entry.get("agent_id") or task_id).strip()
-        task = manager.get_task(task_id)
-        if task is None:
-            continue
-        output = manager.read_task_output(task_id, max_bytes=8000).strip()
-        notifications.append(
-            format_task_notification(
-                TaskNotification(
-                    task_id=agent_id,
-                    status=task.status,
-                    summary=_build_async_task_summary(
-                        entry,
-                        task_status=task.status,
-                        return_code=task.return_code,
-                    ),
-                    result=output or None,
-                )
-            )
-        )
-        entry["notification_sent"] = True
-        entry["notified_status"] = task.status
-    return "\n\n".join(notifications)
-
-
-async def _submit_print_follow_up(
-    bundle,
-    message: str,
-    *,
-    prompt_seed: str,
-    print_system,
-    render_event,
-) -> None:
-    from openharness.ui.runtime import _format_pending_tool_results
-
-    settings = bundle.current_settings()
-    if bundle.enforce_max_turns:
-        bundle.engine.set_max_turns(settings.max_turns)
-    system_prompt = build_runtime_system_prompt(
-        settings,
-        cwd=bundle.cwd,
-        latest_user_prompt=prompt_seed,
-        extra_skill_dirs=bundle.extra_skill_dirs,
-        extra_plugin_roots=bundle.extra_plugin_roots,
-    )
-    bundle.engine.set_system_prompt(system_prompt)
-    try:
-        async for event in bundle.engine.submit_message(message):
-            await render_event(event)
-    except MaxTurnsExceeded as exc:
-        await print_system(f"Stopped after {exc.max_turns} turns (max_turns).")
-        pending = _format_pending_tool_results(bundle.engine.messages)
-        if pending:
-            await print_system(pending)
-    bundle.session_backend.save_snapshot(
-        cwd=bundle.cwd,
-        model=settings.model,
-        system_prompt=system_prompt,
-        messages=bundle.engine.messages,
-        usage=bundle.engine.total_usage,
-        session_id=bundle.session_id,
-        tool_metadata=bundle.engine.tool_metadata,
-    )
-
-
-async def _drain_coordinator_async_agents(
-    bundle,
-    *,
-    prompt_seed: str,
-    output_format: str,
-    print_system,
-    render_event,
-) -> None:
-    engine = getattr(bundle, "engine", None)
-    if engine is None:
-        return
-    while True:
-        pending = _pending_async_agent_entries(getattr(engine, "tool_metadata", None))
-        if not pending:
-            return
-        if output_format == "text":
-            await print_system(
-                f"Waiting for {len(pending)} background agent task(s) to finish..."
-            )
-        completed = await _wait_for_completed_async_agent_entries(getattr(engine, "tool_metadata", None))
-        notification_payload = _format_completed_task_notifications(completed)
-        if not notification_payload.strip():
-            return
-        await _submit_print_follow_up(
-            bundle,
-            notification_payload,
-            prompt_seed=prompt_seed,
-            print_system=print_system,
-            render_event=render_event,
-        )
-
-
 async def run_repl(
     *,
     prompt: str | None = None,
@@ -467,12 +298,12 @@ async def run_print_mode(
             clear_output=_clear_output,
         )
         if is_coordinator_mode():
-            await _drain_coordinator_async_agents(
+            await drain_coordinator_async_agents(
                 bundle,
                 prompt_seed=prompt,
-                output_format=output_format,
                 print_system=_print_system,
                 render_event=_render_event,
+                announce_waiting=output_format == "text",
             )
 
         if output_format == "json":

src/openharness/ui/backend_host.py

@@ -16,6 +16,7 @@ from openharness.api.client import SupportsStreamingMessages
 from openharness.auth.manager import AuthManager
 from openharness.config.settings import CLAUDE_MODEL_ALIAS_OPTIONS, resolve_model_setting
 from openharness.bridge import get_bridge_manager
+from openharness.coordinator.coordinator_mode import is_coordinator_mode
 from openharness.themes import list_themes
 from openharness.engine.stream_events import (
     AssistantTextDelta,
@@ -29,6 +30,7 @@ from openharness.engine.stream_events import (
 )
 from openharness.output_styles import load_output_styles
 from openharness.tasks import get_task_manager
+from openharness.ui.coordinator_drain import drain_coordinator_async_agents
 from openharness.ui.protocol import BackendEvent, FrontendRequest, TranscriptItem
 from openharness.ui.runtime import build_runtime, close_runtime, handle_line, start_runtime
 from openharness.services.session_backend import SessionBackend
@@ -306,6 +308,13 @@ class ReactBackendHost:
             render_event=_render_event,
             clear_output=_clear_output,
         )
+        if is_coordinator_mode():
+            await drain_coordinator_async_agents(
+                self._bundle,
+                prompt_seed=line,
+                print_system=_print_system,
+                render_event=_render_event,
+            )
         await self._emit(self._status_snapshot())
         await self._emit(BackendEvent.tasks_snapshot(get_task_manager().list_tasks()))
         await self._emit(BackendEvent(type="line_complete"))

src/openharness/ui/coordinator_drain.py

@@ -0,0 +1,197 @@
+"""Coordinator-mode helpers for draining background agent tasks between turns.
+
+When coordinator mode dispatches workers via the ``agent`` tool, the system
+prompt promises that worker results arrive as user-role ``<task-notification>``
+messages between coordinator turns. The harness has to honor that contract by
+polling the task manager for completion, formatting the notification, and
+submitting it as a follow-up to the coordinator. These helpers implement that
+behavior independently of the UI host so both print mode and interactive
+backends can share the same logic.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from openharness.coordinator.coordinator_mode import (
+    TaskNotification,
+    format_task_notification,
+)
+from openharness.engine.query import MaxTurnsExceeded
+from openharness.prompts.context import build_runtime_system_prompt
+from openharness.tasks.manager import get_task_manager
+
+
+_TERMINAL_TASK_STATUSES = frozenset({"completed", "failed", "killed"})
+
+
+def _async_agent_task_entries(tool_metadata: dict[str, object] | None) -> list[dict[str, object]]:
+    if not isinstance(tool_metadata, dict):
+        return []
+    value = tool_metadata.get("async_agent_tasks")
+    if not isinstance(value, list):
+        return []
+    return [entry for entry in value if isinstance(entry, dict)]
+
+
+def pending_async_agent_entries(tool_metadata: dict[str, object] | None) -> list[dict[str, object]]:
+    pending: list[dict[str, object]] = []
+    for entry in _async_agent_task_entries(tool_metadata):
+        task_id = str(entry.get("task_id") or "").strip()
+        if not task_id:
+            continue
+        if bool(entry.get("notification_sent")):
+            continue
+        pending.append(entry)
+    return pending
+
+
+def _build_async_task_summary(
+    entry: dict[str, object], *, task_status: str, return_code: int | None
+) -> str:
+    description = str(entry.get("description") or entry.get("agent_id") or "background task").strip()
+    if task_status == "completed":
+        return f'Agent "{description}" completed'
+    if task_status == "killed":
+        return f'Agent "{description}" was stopped'
+    if return_code is not None:
+        return f'Agent "{description}" failed with exit code {return_code}'
+    return f'Agent "{description}" failed'
+
+
+async def wait_for_completed_async_agent_entries(
+    tool_metadata: dict[str, object] | None,
+    *,
+    poll_interval_seconds: float = 0.1,
+) -> list[dict[str, object]]:
+    manager = get_task_manager()
+    while True:
+        pending = pending_async_agent_entries(tool_metadata)
+        if not pending:
+            return []
+        completed: list[dict[str, object]] = []
+        for entry in pending:
+            task_id = str(entry.get("task_id") or "").strip()
+            task = manager.get_task(task_id)
+            if task is None:
+                entry["notification_sent"] = True
+                entry["status"] = "missing"
+                continue
+            entry["status"] = task.status
+            if task.status in _TERMINAL_TASK_STATUSES:
+                entry["return_code"] = task.return_code
+                completed.append(entry)
+        if completed:
+            return completed
+        await asyncio.sleep(poll_interval_seconds)
+
+
+def format_completed_task_notifications(completed: list[dict[str, object]]) -> str:
+    manager = get_task_manager()
+    notifications: list[str] = []
+    for entry in completed:
+        task_id = str(entry.get("task_id") or "").strip()
+        agent_id = str(entry.get("agent_id") or task_id).strip()
+        task = manager.get_task(task_id)
+        if task is None:
+            continue
+        output = manager.read_task_output(task_id, max_bytes=8000).strip()
+        notifications.append(
+            format_task_notification(
+                TaskNotification(
+                    task_id=agent_id,
+                    status=task.status,
+                    summary=_build_async_task_summary(
+                        entry,
+                        task_status=task.status,
+                        return_code=task.return_code,
+                    ),
+                    result=output or None,
+                )
+            )
+        )
+        entry["notification_sent"] = True
+        entry["notified_status"] = task.status
+    return "\n\n".join(notifications)
+
+
+async def submit_follow_up(
+    bundle,
+    message: str,
+    *,
+    prompt_seed: str,
+    print_system,
+    render_event,
+) -> None:
+    from openharness.ui.runtime import _format_pending_tool_results
+
+    settings = bundle.current_settings()
+    if bundle.enforce_max_turns:
+        bundle.engine.set_max_turns(settings.max_turns)
+    system_prompt = build_runtime_system_prompt(
+        settings,
+        cwd=bundle.cwd,
+        latest_user_prompt=prompt_seed,
+        extra_skill_dirs=bundle.extra_skill_dirs,
+        extra_plugin_roots=bundle.extra_plugin_roots,
+    )
+    bundle.engine.set_system_prompt(system_prompt)
+    try:
+        async for event in bundle.engine.submit_message(message):
+            await render_event(event)
+    except MaxTurnsExceeded as exc:
+        await print_system(f"Stopped after {exc.max_turns} turns (max_turns).")
+        pending = _format_pending_tool_results(bundle.engine.messages)
+        if pending:
+            await print_system(pending)
+    bundle.session_backend.save_snapshot(
+        cwd=bundle.cwd,
+        model=settings.model,
+        system_prompt=system_prompt,
+        messages=bundle.engine.messages,
+        usage=bundle.engine.total_usage,
+        session_id=bundle.session_id,
+        tool_metadata=bundle.engine.tool_metadata,
+    )
+
+
+async def drain_coordinator_async_agents(
+    bundle,
+    *,
+    prompt_seed: str,
+    print_system,
+    render_event,
+    announce_waiting: bool = True,
+) -> None:
+    """Block until pending async-agent tasks finish, then submit notifications.
+
+    Submits one follow-up turn per batch of completed workers so the coordinator
+    sees ``<task-notification>`` envelopes between its own turns, matching the
+    contract documented in the coordinator system prompt.
+
+    Returns immediately when there are no pending async-agent entries.
+    """
+    engine = getattr(bundle, "engine", None)
+    if engine is None:
+        return
+    while True:
+        pending = pending_async_agent_entries(getattr(engine, "tool_metadata", None))
+        if not pending:
+            return
+        if announce_waiting:
+            await print_system(
+                f"Waiting for {len(pending)} background agent task(s) to finish..."
+            )
+        completed = await wait_for_completed_async_agent_entries(
+            getattr(engine, "tool_metadata", None)
+        )
+        notification_payload = format_completed_task_notifications(completed)
+        if not notification_payload.strip():
+            return
+        await submit_follow_up(
+            bundle,
+            notification_payload,
+            prompt_seed=prompt_seed,
+            print_system=print_system,
+            render_event=render_event,
+        )

src/openharness/ui/textual_app.py

@@ -16,6 +16,7 @@ from textual.widgets import Button, Footer, Header, Input, RichLog, Static
 
 from openharness.api.client import SupportsStreamingMessages
 from openharness.config.settings import load_settings, save_settings
+from openharness.coordinator.coordinator_mode import is_coordinator_mode
 from openharness.engine.stream_events import (
     AssistantTextDelta,
     AssistantTurnComplete,
@@ -27,6 +28,7 @@ from openharness.engine.stream_events import (
     ToolExecutionStarted,
 )
 from openharness.tasks import get_task_manager
+from openharness.ui.coordinator_drain import drain_coordinator_async_agents
 from openharness.ui.runtime import build_runtime, close_runtime, handle_line, start_runtime
 
 
@@ -303,6 +305,13 @@ class OpenHarnessTerminalApp(App[None]):
                 render_event=self._render_event,
                 clear_output=self._clear_transcript,
             )
+            if is_coordinator_mode():
+                await drain_coordinator_async_agents(
+                    self._bundle,
+                    prompt_seed=line,
+                    print_system=self._print_system,
+                    render_event=self._render_event,
+                )
             self._refresh_sidebars()
             if not should_continue:
                 self.exit()

tests/test_ui/test_coordinator_drain.py

@@ -0,0 +1,183 @@
+"""Tests for the coordinator-mode async-agent drain helper and its integration
+into the interactive UI hosts (React backend and Textual app).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from openharness.ui import coordinator_drain
+from openharness.ui.backend_host import BackendHostConfig, ReactBackendHost
+from openharness.ui.coordinator_drain import (
+    drain_coordinator_async_agents,
+    pending_async_agent_entries,
+)
+from openharness.ui.runtime import build_runtime, close_runtime, start_runtime
+
+from .test_react_backend import StaticApiClient
+
+
+def test_pending_async_agent_entries_skips_notified_and_missing_id():
+    metadata = {
+        "async_agent_tasks": [
+            {"task_id": "t1", "agent_id": "a1"},
+            {"task_id": "t2", "agent_id": "a2", "notification_sent": True},
+            {"task_id": "", "agent_id": "a3"},
+            "not-a-dict",
+        ]
+    }
+    pending = pending_async_agent_entries(metadata)
+    assert [entry["task_id"] for entry in pending] == ["t1"]
+
+
+def test_pending_async_agent_entries_handles_missing_metadata():
+    assert pending_async_agent_entries(None) == []
+    assert pending_async_agent_entries({}) == []
+    assert pending_async_agent_entries({"async_agent_tasks": "not a list"}) == []
+
+
+@pytest.mark.asyncio
+async def test_drain_returns_immediately_when_no_pending_entries():
+    """No pending entries = no follow-up turn, no `Waiting for...` message."""
+
+    class _FakeEngine:
+        tool_metadata: dict[str, object] = {}
+
+    class _FakeBundle:
+        engine = _FakeEngine()
+
+    announcements: list[str] = []
+
+    async def _print(message: str) -> None:
+        announcements.append(message)
+
+    async def _render(_event):  # pragma: no cover - never called in this scenario
+        raise AssertionError("render_event must not be invoked when no work is pending")
+
+    await drain_coordinator_async_agents(
+        _FakeBundle(),
+        prompt_seed="hi",
+        print_system=_print,
+        render_event=_render,
+    )
+    assert announcements == []
+
+
+@pytest.mark.asyncio
+async def test_drain_returns_when_bundle_has_no_engine():
+    class _NoEngineBundle:
+        pass
+
+    async def _print(_message: str) -> None:  # pragma: no cover
+        raise AssertionError("print_system must not be invoked")
+
+    async def _render(_event):  # pragma: no cover
+        raise AssertionError("render_event must not be invoked")
+
+    await drain_coordinator_async_agents(
+        _NoEngineBundle(),
+        prompt_seed="hi",
+        print_system=_print,
+        render_event=_render,
+    )
+
+
+@pytest.mark.asyncio
+async def test_react_backend_drains_async_agents_in_coordinator_mode(tmp_path, monkeypatch):
+    """Regression: React TUI's `_process_line` must invoke the coordinator drain.
+
+    Without this, `<task-notification>` envelopes never reach the coordinator
+    after a worker finishes, so either the user is left holding stale state or
+    the coordinator polls in-turn (locking the UI).
+    """
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+
+    host = ReactBackendHost(BackendHostConfig(api_client=StaticApiClient("unused")))
+    host._bundle = await build_runtime(api_client=StaticApiClient("unused"))
+
+    async def _fake_handle_line(bundle, line, print_system, render_event, clear_output):
+        del bundle, line, print_system, render_event, clear_output
+        return True
+
+    monkeypatch.setattr("openharness.ui.backend_host.handle_line", _fake_handle_line)
+    monkeypatch.setattr("openharness.ui.backend_host.is_coordinator_mode", lambda: True)
+
+    drain_calls: list[dict[str, object]] = []
+
+    async def _fake_drain(bundle, *, prompt_seed, print_system, render_event):
+        drain_calls.append(
+            {
+                "bundle_is_host_bundle": bundle is host._bundle,
+                "prompt_seed": prompt_seed,
+            }
+        )
+
+    monkeypatch.setattr(
+        "openharness.ui.backend_host.drain_coordinator_async_agents",
+        _fake_drain,
+    )
+
+    async def _emit(_event):
+        return None
+
+    host._emit = _emit  # type: ignore[method-assign]
+    await start_runtime(host._bundle)
+    try:
+        should_continue = await host._process_line("dispatch a worker")
+    finally:
+        await close_runtime(host._bundle)
+
+    assert should_continue is True
+    assert drain_calls == [{"bundle_is_host_bundle": True, "prompt_seed": "dispatch a worker"}]
+
+
+@pytest.mark.asyncio
+async def test_react_backend_skips_drain_when_not_coordinator(tmp_path, monkeypatch):
+    """When coordinator mode is off, the drain must not run — it would needlessly
+    poll the task manager and submit follow-up turns for unrelated background tasks.
+    """
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+
+    host = ReactBackendHost(BackendHostConfig(api_client=StaticApiClient("unused")))
+    host._bundle = await build_runtime(api_client=StaticApiClient("unused"))
+
+    async def _fake_handle_line(bundle, line, print_system, render_event, clear_output):
+        del bundle, line, print_system, render_event, clear_output
+        return True
+
+    monkeypatch.setattr("openharness.ui.backend_host.handle_line", _fake_handle_line)
+    monkeypatch.setattr("openharness.ui.backend_host.is_coordinator_mode", lambda: False)
+
+    async def _fake_drain(*args, **kwargs):  # pragma: no cover
+        del args, kwargs
+        raise AssertionError("drain must not be called outside coordinator mode")
+
+    monkeypatch.setattr(
+        "openharness.ui.backend_host.drain_coordinator_async_agents",
+        _fake_drain,
+    )
+
+    async def _emit(_event):
+        return None
+
+    host._emit = _emit  # type: ignore[method-assign]
+    await start_runtime(host._bundle)
+    try:
+        should_continue = await host._process_line("hi")
+    finally:
+        await close_runtime(host._bundle)
+
+    assert should_continue is True
+
+
+def test_drain_module_exposes_public_api():
+    """The drain helpers must keep the public names other modules import."""
+    assert callable(coordinator_drain.drain_coordinator_async_agents)
+    assert callable(coordinator_drain.pending_async_agent_entries)
+    assert callable(coordinator_drain.wait_for_completed_async_agent_entries)
+    assert callable(coordinator_drain.format_completed_task_notifications)
+    assert callable(coordinator_drain.submit_follow_up)

tests/test_ui/test_react_launcher.py

@@ -231,14 +231,17 @@ async def test_run_print_mode_waits_for_coordinator_async_agents(monkeypatch):
     monkeypatch.setattr("openharness.ui.app.start_runtime", _start_runtime)
     monkeypatch.setattr("openharness.ui.app.handle_line", _handle_line)
     monkeypatch.setattr("openharness.ui.app.close_runtime", _close_runtime)
-    monkeypatch.setattr("openharness.ui.app.get_task_manager", lambda: fake_manager)
+    monkeypatch.setattr("openharness.ui.coordinator_drain.get_task_manager", lambda: fake_manager)
     monkeypatch.setattr("openharness.ui.app.is_coordinator_mode", lambda: True)
-    monkeypatch.setattr("openharness.ui.app.build_runtime_system_prompt", lambda *args, **kwargs: "coordinator")
+    monkeypatch.setattr(
+        "openharness.ui.coordinator_drain.build_runtime_system_prompt",
+        lambda *args, **kwargs: "coordinator",
+    )
 
     async def _sleep(_seconds):
         return None
 
-    monkeypatch.setattr("openharness.ui.app.asyncio.sleep", _sleep)
+    monkeypatch.setattr("openharness.ui.coordinator_drain.asyncio.sleep", _sleep)
 
     await run_print_mode(prompt="research this", cwd="/tmp/demo")