fix(swarm): run subprocess teammates in headless worker mode

src/openharness/cli.py

@@ -1257,6 +1257,12 @@ def main(
         help="Run the structured backend host for the React terminal UI",
         hidden=True,
     ),
+    task_worker: bool = typer.Option(
+        False,
+        "--task-worker",
+        help="Run the stdin-driven headless worker loop used for background agent tasks",
+        hidden=True,
+    ),
 ) -> None:
     """Start an interactive session or run a single prompt."""
     if ctx.invoked_subcommand is not None:
@@ -1287,7 +1293,7 @@ def main(
         settings.theme = theme
         save_settings(settings)
 
-    from openharness.ui.app import run_print_mode, run_repl
+    from openharness.ui.app import run_print_mode, run_repl, run_task_worker
 
     # Handle --continue and --resume flags
     if continue_session or resume is not None:
@@ -1371,6 +1377,21 @@ def main(
         )
         return
 
+    if task_worker:
+        asyncio.run(
+            run_task_worker(
+                cwd=cwd,
+                model=model,
+                max_turns=max_turns,
+                base_url=base_url,
+                system_prompt=system_prompt,
+                api_key=api_key,
+                api_format=api_format,
+                permission_mode=permission_mode,
+            )
+        )
+        return
+
     asyncio.run(
         run_repl(
             prompt=None,

src/openharness/swarm/spawn_utils.py

@@ -73,21 +73,24 @@ def get_teammate_command() -> str:
     Resolution order:
     1. ``OPENHARNESS_TEAMMATE_COMMAND`` environment variable — allows the
        operator to point at a specific binary or wrapper script.
-    2. The ``openharness`` entry-point on PATH (installed package mode).
-    3. The current Python interpreter running the ``openharness`` module
-       (development / editable-install fallback).
+    2. The current Python interpreter running the ``openharness`` module.
+       This keeps spawned teammates on the same venv/source tree as the
+       leader process.
+    3. The ``openharness`` entry-point on PATH (installed package fallback).
     """
     override = os.environ.get(TEAMMATE_COMMAND_ENV_VAR)
     if override:
         return override
 
-    # Check if we are running as an installed package with an entry-point.
+    # Prefer the current interpreter so teammates inherit the same runtime and
+    # editable-install source tree as the parent process.
+    if sys.executable:
+        return sys.executable
+
     entry_point = shutil.which("openharness")
     if entry_point:
         return entry_point
-
-    # Fall back to the Python interpreter that is currently running this code.
-    return sys.executable
+    return "python"
 
 
 def build_inherited_cli_flags(
@@ -176,6 +179,9 @@ def build_inherited_env_vars() -> dict[str, str]:
     """
     env: dict[str, str] = {
         "OPENHARNESS_AGENT_TEAMS": "1",
+        # Spawned workers should behave like workers, not recursively re-enter
+        # coordinator mode just because the parent leader had the flag set.
+        "CLAUDE_CODE_COORDINATOR_MODE": "0",
     }
 
     for key in _TEAMMATE_ENV_VARS:

src/openharness/swarm/subprocess_backend.py

@@ -63,9 +63,9 @@ class SubprocessBackend:
 
         teammate_cmd = get_teammate_command()
         if teammate_cmd.endswith("python") or teammate_cmd.endswith("python3") or "/python" in teammate_cmd:
-            cmd_parts = [teammate_cmd, "-m", "openharness"] + flags
+            cmd_parts = [teammate_cmd, "-m", "openharness", "--task-worker"] + flags
         else:
-            cmd_parts = [teammate_cmd] + flags
+            cmd_parts = [teammate_cmd, "--task-worker"] + flags
         command = f"{env_prefix} {' '.join(cmd_parts)}" if env_prefix else " ".join(cmd_parts)
 
         manager = get_task_manager()

src/openharness/ui/app.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import sys
 
@@ -12,6 +13,27 @@ from openharness.ui.react_launcher import launch_react_tui
 from openharness.ui.runtime import build_runtime, close_runtime, handle_line, start_runtime
 
 
+def _decode_task_worker_line(raw: str) -> str:
+    """Normalize one stdin line for the headless task worker.
+
+    Task-manager driven agent workers receive either:
+    - a plain text line (initial prompt or simple follow-up), or
+    - a JSON object from ``send_message`` / teammate backends with a ``text`` field.
+    """
+    stripped = raw.strip()
+    if not stripped:
+        return ""
+    try:
+        payload = json.loads(stripped)
+    except json.JSONDecodeError:
+        return stripped
+    if isinstance(payload, dict):
+        text = payload.get("text")
+        if isinstance(text, str):
+            return text.strip()
+    return stripped
+
+
 async def run_repl(
     *,
     prompt: str | None = None,
@@ -59,6 +81,89 @@ async def run_repl(
         raise SystemExit(exit_code)
 
 
+async def run_task_worker(
+    *,
+    cwd: str | None = None,
+    model: str | None = None,
+    max_turns: int | None = None,
+    base_url: str | None = None,
+    system_prompt: str | None = None,
+    api_key: str | None = None,
+    api_format: str | None = None,
+    api_client: SupportsStreamingMessages | None = None,
+    permission_mode: str | None = None,
+) -> None:
+    """Run a stdin-driven headless worker for background agent tasks.
+
+    This mode exists for subprocess teammates and other task-manager managed
+    agent processes. It intentionally avoids the React TUI / Ink path so it
+    can run without a controlling TTY.
+    """
+
+    async def _noop_permission(_tool_name: str, _reason: str) -> bool:
+        return True
+
+    async def _noop_ask(_question: str) -> str:
+        return ""
+
+    async def _print_system(message: str) -> None:
+        print(message, flush=True)
+
+    async def _render_event(event: StreamEvent) -> None:
+        from openharness.engine.stream_events import AssistantTextDelta, AssistantTurnComplete, ErrorEvent, StatusEvent
+
+        if isinstance(event, AssistantTextDelta):
+            sys.stdout.write(event.text)
+            sys.stdout.flush()
+        elif isinstance(event, AssistantTurnComplete):
+            sys.stdout.write("\n")
+            sys.stdout.flush()
+        elif isinstance(event, ErrorEvent):
+            print(event.message, flush=True)
+        elif isinstance(event, StatusEvent) and event.message:
+            print(event.message, flush=True)
+
+    async def _clear_output() -> None:
+        return None
+
+    bundle = await build_runtime(
+        cwd=cwd,
+        model=model,
+        max_turns=max_turns,
+        base_url=base_url,
+        system_prompt=system_prompt,
+        api_key=api_key,
+        api_format=api_format,
+        api_client=api_client,
+        permission_prompt=_noop_permission,
+        ask_user_prompt=_noop_ask,
+        enforce_max_turns=max_turns is not None,
+        permission_mode=permission_mode,
+    )
+    await start_runtime(bundle)
+    try:
+        while True:
+            raw = await asyncio.to_thread(sys.stdin.readline)
+            if raw == "":
+                break
+            line = _decode_task_worker_line(raw)
+            if not line:
+                continue
+            await handle_line(
+                bundle,
+                line,
+                print_system=_print_system,
+                render_event=_render_event,
+                clear_output=_clear_output,
+            )
+            # Background agent tasks are one-shot workers. If the coordinator
+            # needs to send a follow-up later, BackgroundTaskManager already
+            # knows how to restart the task and write the next stdin payload.
+            break
+    finally:
+        await close_runtime(bundle)
+
+
 async def run_print_mode(
     *,
     prompt: str,

tests/test_commands/test_cli.py

@@ -129,3 +129,18 @@ def test_dangerously_skip_permissions_passes_full_auto_to_run_repl(monkeypatch):
 
     assert result.exit_code == 0
     assert captured["permission_mode"] == "full_auto"
+
+
+def test_task_worker_flag_routes_to_run_task_worker(monkeypatch):
+    runner = CliRunner()
+    captured = {}
+
+    async def fake_run_task_worker(**kwargs):
+        captured.update(kwargs)
+
+    monkeypatch.setattr("openharness.ui.app.run_task_worker", fake_run_task_worker)
+
+    result = runner.invoke(app, ["--task-worker", "--model", "kimi-k2.5"])
+
+    assert result.exit_code == 0
+    assert captured["model"] == "kimi-k2.5"

tests/test_swarm/test_spawn_utils.py

@@ -0,0 +1,28 @@
+"""Tests for teammate spawn helper behavior."""
+
+from __future__ import annotations
+
+import sys
+
+from openharness.swarm.spawn_utils import (
+    TEAMMATE_COMMAND_ENV_VAR,
+    build_inherited_env_vars,
+    get_teammate_command,
+)
+
+
+def test_get_teammate_command_prefers_current_interpreter(monkeypatch):
+    monkeypatch.delenv(TEAMMATE_COMMAND_ENV_VAR, raising=False)
+    monkeypatch.setattr(sys, "executable", "/tmp/current-python")
+
+    command = get_teammate_command()
+
+    assert command == "/tmp/current-python"
+
+
+def test_build_inherited_env_vars_disables_coordinator_mode(monkeypatch):
+    monkeypatch.setenv("CLAUDE_CODE_COORDINATOR_MODE", "1")
+
+    env = build_inherited_env_vars()
+
+    assert env["CLAUDE_CODE_COORDINATOR_MODE"] == "0"

tests/test_tools/test_task_tools.py

@@ -142,6 +142,8 @@ async def test_agent_tool_uses_subprocess_backend_and_task_is_pollable(
         f"task_id {task_id!r} not found in BackgroundTaskManager — "
         "task tools (TaskGet, TaskOutput, etc.) would have failed"
     )
+    assert record.command is not None
+    assert "--task-worker" in record.command
 
 
 @pytest.mark.asyncio

tests/test_ui/test_react_launcher.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import pytest
 from types import SimpleNamespace
 
-from openharness.ui.app import run_print_mode, run_repl
+from openharness.ui.app import run_print_mode, run_repl, run_task_worker
 from openharness.ui.react_launcher import build_backend_command
 
 
@@ -80,3 +80,59 @@ async def test_run_print_mode_passes_cwd_to_build_runtime(monkeypatch):
     await run_print_mode(prompt="hi", cwd="/tmp/demo")
 
     assert seen["cwd"] == "/tmp/demo"
+
+
+@pytest.mark.asyncio
+async def test_run_task_worker_reads_one_shot_json_line(monkeypatch):
+    seen = []
+
+    class _FakeStdin:
+        def __init__(self):
+            self._lines = iter([
+                '{"text":"follow up from coordinator","from":"coordinator"}\n',
+            ])
+
+        def readline(self):
+            return next(self._lines, "")
+
+    async def _build_runtime(**kwargs):
+        return SimpleNamespace(
+            cwd=kwargs.get("cwd"),
+            engine=SimpleNamespace(),
+            external_api_client=False,
+            extra_skill_dirs=(),
+            extra_plugin_roots=(),
+            current_settings=lambda: None,
+            current_plugins=lambda: [],
+            hook_summary=lambda: "",
+            plugin_summary=lambda: "",
+            mcp_summary=lambda: "",
+            app_state=SimpleNamespace(set=lambda **_kwargs: None),
+            mcp_manager=SimpleNamespace(close=lambda: None, list_statuses=lambda: []),
+            hook_executor=SimpleNamespace(execute=lambda *_args, **_kwargs: None, update_registry=lambda *_a, **_k: None),
+            commands=SimpleNamespace(lookup=lambda _line: None),
+            session_backend=SimpleNamespace(save_snapshot=lambda **_kwargs: None),
+            enforce_max_turns=False,
+            session_id="s1",
+        )
+
+    async def _start_runtime(_bundle):
+        return None
+
+    async def _handle_line(bundle, line, **kwargs):
+        del bundle, kwargs
+        seen.append(line)
+        return True
+
+    async def _close_runtime(_bundle):
+        return None
+
+    monkeypatch.setattr("openharness.ui.app.build_runtime", _build_runtime)
+    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.sys.stdin", _FakeStdin())
+
+    await run_task_worker(cwd="/tmp/demo")
+
+    assert seen == ["follow up from coordinator"]