fix(tasks): preserve multiline prompts for task workers

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:

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

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: