From 76d91fdfa149dc577df91754ce1926d671d9f0e0 Mon Sep 17 00:00:00 2001 From: tjb-tech <1193992557@qq.com> Date: Fri, 10 Apr 2026 09:50:07 +0000 Subject: [PATCH] fix(swarm): run subprocess teammates in headless worker mode --- src/openharness/cli.py | 23 ++++- src/openharness/swarm/spawn_utils.py | 20 ++-- src/openharness/swarm/subprocess_backend.py | 4 +- src/openharness/ui/app.py | 105 ++++++++++++++++++++ tests/test_commands/test_cli.py | 15 +++ tests/test_swarm/test_spawn_utils.py | 28 ++++++ tests/test_tools/test_task_tools.py | 2 + tests/test_ui/test_react_launcher.py | 58 ++++++++++- 8 files changed, 244 insertions(+), 11 deletions(-) create mode 100644 tests/test_swarm/test_spawn_utils.py diff --git a/src/openharness/cli.py b/src/openharness/cli.py index e7f0109..b92f38e 100644 --- a/src/openharness/cli.py +++ b/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, diff --git a/src/openharness/swarm/spawn_utils.py b/src/openharness/swarm/spawn_utils.py index 4b592ec..72e23fd 100644 --- a/src/openharness/swarm/spawn_utils.py +++ b/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: diff --git a/src/openharness/swarm/subprocess_backend.py b/src/openharness/swarm/subprocess_backend.py index cd025e8..4d38b80 100644 --- a/src/openharness/swarm/subprocess_backend.py +++ b/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() diff --git a/src/openharness/ui/app.py b/src/openharness/ui/app.py index eef5a18..fe3cfa8 100644 --- a/src/openharness/ui/app.py +++ b/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, diff --git a/tests/test_commands/test_cli.py b/tests/test_commands/test_cli.py index 9d9baf9..e0aaf07 100644 --- a/tests/test_commands/test_cli.py +++ b/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" diff --git a/tests/test_swarm/test_spawn_utils.py b/tests/test_swarm/test_spawn_utils.py new file mode 100644 index 0000000..9686d61 --- /dev/null +++ b/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" diff --git a/tests/test_tools/test_task_tools.py b/tests/test_tools/test_task_tools.py index ed54964..de50bfa 100644 --- a/tests/test_tools/test_task_tools.py +++ b/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 diff --git a/tests/test_ui/test_react_launcher.py b/tests/test_ui/test_react_launcher.py index e6aa4b0..b2e56a6 100644 --- a/tests/test_ui/test_react_launcher.py +++ b/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"]