diff --git a/src/openharness/tasks/manager.py b/src/openharness/tasks/manager.py index 031e0e9..b298efe 100644 --- a/src/openharness/tasks/manager.py +++ b/src/openharness/tasks/manager.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import json import logging import os import shlex @@ -17,6 +18,31 @@ from openharness.tasks.types import TaskRecord, TaskStatus, TaskType from openharness.utils.shell import create_shell_subprocess log = logging.getLogger(__name__) +_TASK_RESTART_NOTICE = "[OpenHarness] Agent task restarted; prior interactive context was not preserved.\n" + + +def _encode_task_worker_payload(data: str) -> bytes: + """Serialize one worker input as a single JSON line. + + Plain-text prompts may contain embedded newlines, so they cannot be written + directly to a readline()-based worker protocol. We wrap them in a JSON + object with a ``text`` field, while preserving already-structured payloads + emitted by teammate backends. + """ + + stripped = data.rstrip("\n") + try: + payload = json.loads(stripped) + except json.JSONDecodeError: + payload = None + + if isinstance(payload, dict) and isinstance(payload.get("text"), str): + framed = stripped + elif "\n" not in stripped and "\r" not in stripped: + framed = stripped + else: + framed = json.dumps({"text": stripped}, ensure_ascii=False) + return (framed + "\n").encode("utf-8") CompletionListener = Callable[[TaskRecord], Awaitable[None] | None] @@ -155,16 +181,17 @@ class BackgroundTaskManager: async def write_to_task(self, task_id: str, data: str) -> None: """Write one line to task stdin, auto-resuming local agents when needed.""" task = self._require_task(task_id) + payload = _encode_task_worker_payload(data) async with self._input_locks[task_id]: process = await self._ensure_writable_process(task) - process.stdin.write((data.rstrip("\n") + "\n").encode("utf-8")) + process.stdin.write(payload) try: await process.stdin.drain() except (BrokenPipeError, ConnectionResetError): if task.type not in {"local_agent", "remote_agent", "in_process_teammate"}: raise ValueError(f"Task {task_id} does not accept input") from None process = await self._restart_agent_task(task) - process.stdin.write((data.rstrip("\n") + "\n").encode("utf-8")) + process.stdin.write(payload) await process.stdin.drain() def read_task_output(self, task_id: str, *, max_bytes: int = 12000) -> str: @@ -267,10 +294,13 @@ class BackgroundTaskManager: restart_count = int(task.metadata.get("restart_count", "0")) + 1 task.metadata["restart_count"] = str(restart_count) + task.metadata["status_note"] = "Task restarted; prior interactive context was not preserved." task.status = "running" task.started_at = time.time() task.ended_at = None task.return_code = None + with task.output_file.open("ab") as handle: + handle.write(_TASK_RESTART_NOTICE.encode("utf-8")) return await self._start_process(task.id) async def _notify_completion_listeners(self, task: TaskRecord) -> None: diff --git a/tests/test_tasks/test_manager.py b/tests/test_tasks/test_manager.py index 1bcb1dd..76d7ec9 100644 --- a/tests/test_tasks/test_manager.py +++ b/tests/test_tasks/test_manager.py @@ -3,11 +3,12 @@ from __future__ import annotations import asyncio +import json from pathlib import Path import pytest -from openharness.tasks.manager import BackgroundTaskManager +from openharness.tasks.manager import BackgroundTaskManager, _encode_task_worker_payload @pytest.mark.asyncio @@ -44,6 +45,25 @@ async def test_create_agent_task_with_command_override_and_write(tmp_path: Path, assert "got:first" in manager.read_task_output(task.id) +@pytest.mark.asyncio +async def test_create_agent_task_preserves_multiline_prompt(tmp_path: Path, monkeypatch): + monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) + manager = BackgroundTaskManager() + + task = await manager.create_agent_task( + prompt="line 1\nline 2\nline 3", + description="agent", + cwd=tmp_path, + command=( + "python -u -c \"import sys, json; " + "print(json.loads(sys.stdin.readline())['text'].replace(chr(10), '|'))\"" + ), + ) + + await asyncio.wait_for(manager._waiters[task.id], timeout=5) # type: ignore[attr-defined] + assert "line 1|line 2|line 3" in manager.read_task_output(task.id) + + @pytest.mark.asyncio async def test_write_to_stopped_agent_task_restarts_process(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) @@ -62,10 +82,23 @@ async def test_write_to_stopped_agent_task_restarts_process(tmp_path: Path, monk output = manager.read_task_output(task.id) assert "got:ready" in output + assert "[OpenHarness] Agent task restarted; prior interactive context was not preserved." in output assert "got:follow-up" in output updated = manager.get_task(task.id) assert updated is not None assert updated.metadata["restart_count"] == "1" + assert updated.metadata["status_note"] == "Task restarted; prior interactive context was not preserved." + + +def test_encode_task_worker_payload_wraps_multiline_text() -> None: + payload = _encode_task_worker_payload("alpha\nbeta\n") + assert json.loads(payload.decode("utf-8")) == {"text": "alpha\nbeta"} + + +def test_encode_task_worker_payload_preserves_structured_messages() -> None: + raw = '{"text":"follow up","from":"coordinator"}' + payload = _encode_task_worker_payload(raw) + assert payload.decode("utf-8") == raw + "\n" @pytest.mark.asyncio diff --git a/tests/test_ui/test_react_launcher.py b/tests/test_ui/test_react_launcher.py index 62e7900..0b7f199 100644 --- a/tests/test_ui/test_react_launcher.py +++ b/tests/test_ui/test_react_launcher.py @@ -140,6 +140,62 @@ async def test_run_task_worker_reads_one_shot_json_line(monkeypatch): assert seen == ["follow up from coordinator"] +@pytest.mark.asyncio +async def test_run_task_worker_decodes_multiline_json_payload(monkeypatch): + seen = [] + + class _FakeStdin: + def __init__(self): + self._lines = iter([ + '{"text":"line 1\\nline 2\\nline 3","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 == ["line 1\nline 2\nline 3"] + + @pytest.mark.asyncio async def test_run_print_mode_waits_for_coordinator_async_agents(monkeypatch): class _FakeEngine: