Sync Space agent core

README.md

@@ -44,7 +44,8 @@ These files/directories are copied from the main repo and should be refreshed
 when their corresponding upstream implementation changes:
 
 - `agent_base/`: core ReAct runtime, prompts, tool registry, provider
-  compatibility, trace/session state, image handling, and compaction logic.
+  compatibility, trace/session state, image handling, compaction logic, and
+  imported runtime helpers required by that core.
 - `agent_base/tools/`: hosted-safe tool implementations used by the frontend.
 - `frontend/static/`: shared browser UI assets, styles, and client logic.
 - `frontend/local_server.py`: WebSocket streaming frontend server base, with
@@ -89,7 +90,6 @@ The Space intentionally does not keep the full main-repo surface area:
 - benchmark adapters and benchmark documentation under `benchmarks/`
 - long-form tutorials under `docs/`
 - local placeholder directories such as `workspace/`, `api_runs/`, and `traces/`
-- CLI-only console formatting helpers
 - test fixtures and local test suites
 - `.env.example`
 

agent_base/console_utils.py

@@ -0,0 +1,223 @@
+import argparse
+import json
+import os
+from pathlib import Path
+import shutil
+import sys
+import unicodedata
+from typing import Any, Optional
+
+
+ANSI_RESET = "\033[0m"
+ANSI_COLORS = {
+    "header": "\033[36m",
+    "assistant": "\033[32m",
+    "tool": "\033[33m",
+    "runtime": "\033[34m",
+    "user": "\033[35m",
+    "error": "\033[31m",
+}
+
+
+def _char_display_width(char: str) -> int:
+    if unicodedata.combining(char):
+        return 0
+    if unicodedata.category(char) in {"Cc", "Cf"}:
+        return 0
+    return 2 if unicodedata.east_asian_width(char) in {"F", "W"} else 1
+
+
+def _display_width(text: str) -> int:
+    return sum(_char_display_width(char) for char in str(text))
+
+
+def _truncate_display(text: str, width: int) -> str:
+    if _display_width(text) <= width:
+        return text
+    suffix = "..."
+    target = max(0, width - _display_width(suffix))
+    out = []
+    used = 0
+    for char in text:
+        char_width = _char_display_width(char)
+        if used + char_width > target:
+            break
+        out.append(char)
+        used += char_width
+    return "".join(out) + suffix
+
+
+def _pad_display(text: str, width: int) -> str:
+    return text + " " * max(0, width - _display_width(text))
+
+
+def _last_soft_break(chars: list[str]) -> int:
+    for index in range(len(chars) - 1, 0, -1):
+        if chars[index].isspace() and "".join(chars[:index]).strip():
+            return index
+    return -1
+
+
+class ConsoleEventPrinter:
+    def __init__(self, *, model_name: str, workspace_root: Path, prompt: str):
+        self.model_name = model_name
+        self.workspace_root = workspace_root
+        self.prompt = prompt.strip()
+        self._printed_any = False
+        self._use_color = (
+            "NO_COLOR" not in os.environ
+            and os.environ.get("TERM") != "dumb"
+            and (sys.stdout.isatty() or bool(os.environ.get("FORCE_COLOR") or os.environ.get("CLICOLOR_FORCE")))
+        )
+
+    def print_header(self) -> None:
+        self._print_box(
+            "ResearchHarness CLI",
+            f"Model: {self.model_name}\nWorkspace Root: {self.workspace_root}\n\nPrompt:\n{self.prompt}",
+            "header",
+        )
+
+    def reset_rounds(self) -> None:
+        self._printed_any = False
+
+    def _paint(self, text: str, color_key: str) -> str:
+        if not self._use_color:
+            return text
+        return f"{ANSI_COLORS.get(color_key, '')}{text}{ANSI_RESET}"
+
+    def _terminal_width(self) -> int:
+        return max(60, min(110, shutil.get_terminal_size((100, 20)).columns))
+
+    def _wrap_line(self, line: str, width: int) -> list[str]:
+        expanded = line.expandtabs(2)
+        if expanded == "":
+            return [""]
+        chunks: list[str] = []
+        current: list[str] = []
+        current_width = 0
+        for char in expanded:
+            char_width = _char_display_width(char)
+            if current and current_width + char_width > width:
+                break_at = _last_soft_break(current)
+                if break_at > 0:
+                    chunks.append("".join(current[:break_at]).rstrip())
+                    current = list("".join(current[break_at + 1 :]).lstrip())
+                    current_width = _display_width("".join(current))
+                else:
+                    chunks.append("".join(current))
+                    current = []
+                    current_width = 0
+            current.append(char)
+            current_width += char_width
+        if current:
+            chunks.append("".join(current))
+        return chunks or [""]
+
+    def _print_box(self, title: str, body: str, color_key: str = "runtime") -> None:
+        width = self._terminal_width()
+        inner_width = width - 4
+        title_text = f" {_truncate_display(title.strip(), width - 6)} "
+        top = "+" + title_text + "-" * max(0, width - 2 - _display_width(title_text)) + "+"
+        bottom = "+" + "-" * (width - 2) + "+"
+        if self._printed_any:
+            print()
+        print(self._paint(top, color_key))
+        for raw_line in str(body or "").splitlines() or [""]:
+            for line in self._wrap_line(raw_line, inner_width):
+                padded = _pad_display(line, inner_width)
+                print(f"{self._paint('|', color_key)} {padded} {self._paint('|', color_key)}")
+        print(self._paint(bottom, color_key))
+        self._printed_any = True
+
+    def _title(self, label: str, turn_index: int) -> str:
+        return f"{label} | round {turn_index}" if turn_index > 0 else label
+
+    def _format_tool_call(self, tool_name: str, tool_args: Any) -> str:
+        try:
+            tool_args_text = json.dumps(tool_args, ensure_ascii=False, indent=2)
+        except TypeError:
+            tool_args_text = str(tool_args)
+        return f"- {tool_name}\n{tool_args_text}"
+
+    def handle_event(self, row: dict[str, Any]) -> None:
+        role = str(row.get("role", ""))
+        turn_index = int(row.get("turn_index", 0) or 0)
+        text = str(row.get("text", ""))
+        capture_type = str(row.get("capture_type", ""))
+        tool_names = row.get("tool_names") if isinstance(row.get("tool_names"), list) else []
+        tool_arguments = row.get("tool_arguments") if isinstance(row.get("tool_arguments"), list) else []
+        finish_reason = str(row.get("finish_reason", ""))
+        error = str(row.get("error", ""))
+
+        if capture_type and not text.strip():
+            return
+
+        if role == "system":
+            return
+
+        if role == "user":
+            if turn_index == 0:
+                return
+            self._print_box(self._title("Runtime Message", turn_index), text, "user")
+            return
+
+        if role == "assistant":
+            lines: list[str] = []
+            if tool_names:
+                if text.strip():
+                    lines.append(text)
+                else:
+                    suffix = f" finish_reason={finish_reason}" if finish_reason else ""
+                    lines.append(f"(no text; native tool-calls only.{suffix})")
+                lines.append("")
+                lines.append("Assistant Tool Calls:")
+                for idx, tool_name in enumerate(tool_names):
+                    tool_args = tool_arguments[idx] if idx < len(tool_arguments) else {}
+                    lines.append(self._format_tool_call(str(tool_name), tool_args))
+            elif text.strip():
+                lines.append(text)
+            else:
+                suffix = f" finish_reason={finish_reason}" if finish_reason else ""
+                lines.append(f"(empty assistant output.{suffix})")
+            if error:
+                lines.append("")
+                lines.append(f"Assistant Error: {error}")
+            self._print_box(self._title("Assistant", turn_index), "\n".join(lines), "error" if error else "assistant")
+            return
+
+        if role == "tool":
+            tool_name = str(tool_names[0]) if tool_names else "Tool"
+            lines = [text]
+            if error:
+                lines.extend(["", f"{tool_name} Error: {error}"])
+            self._print_box(self._title(f"{tool_name} Result", turn_index), "\n".join(lines), "error" if error else "tool")
+            return
+
+        if role == "runtime":
+            lines = [text]
+            if error:
+                lines.extend(["", f"Runtime Error: {error}"])
+            self._print_box(self._title("Runtime", turn_index), "\n".join(lines), "error" if error else "runtime")
+
+
+def main(argv: Optional[list[str]] = None) -> int:
+    parser = argparse.ArgumentParser(description="Show a minimal example of the CLI console event formatter.")
+    parser.parse_args(argv)
+    printer = ConsoleEventPrinter(model_name="demo-model", workspace_root=Path("."), prompt="demo question")
+    printer.print_header()
+    printer.handle_event(
+        {
+            "role": "assistant",
+            "turn_index": 1,
+            "text": "",
+            "tool_names": ["Read"],
+            "tool_arguments": [{"path": "demo.txt"}],
+            "termination": "",
+            "error": "",
+        }
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

agent_base/react_agent.py

@@ -1,15 +1,18 @@
+import argparse
 from contextlib import contextmanager
 import json
 import os
 import re
 import signal
+import sys
 import threading
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional, Sequence
+from typing import Any, Callable, Dict, List, Optional, Sequence, Type
 
 from openai import OpenAI, APIError, APIConnectionError, APITimeoutError
 import tiktoken
 from agent_base.base import BaseAgent
+from agent_base.console_utils import ConsoleEventPrinter
 from agent_base.context_compact import compact_messages, should_compact_messages
 from agent_base.model_profiles import resolve_model_profile
 from agent_base.provider_compat import apply_sampling_params
@@ -17,13 +20,22 @@ from agent_base.prompt import composed_system_prompt
 from agent_base.session_state import AgentSessionState, CompactionRecord, persist_session_state, resolve_session_state_path
 from agent_base.trace_utils import FlatTraceWriter
 from agent_base.tools.tooling import normalize_workspace_root
+from agent_base.tools.tool_extra import StrReplaceEditor
 from agent_base.tools.tool_file import Edit, Glob, Grep, Read, ReadImage, ReadPDF, Write
 from agent_base.tools.tool_runtime import Bash, TerminalInterrupt, TerminalKill, TerminalRead, TerminalStart, TerminalWrite
 from agent_base.tools.tool_user import AskUser
 from agent_base.tools.tool_web import ScholarSearch, WebFetch, WebSearch
 from agent_base.utils import (
+    PROJECT_ROOT,
+    MissingRequiredEnvError,
+    append_saved_image_paths_to_prompt,
     env_flag,
+    image_input_content_parts,
+    load_dotenv,
+    read_role_prompt_files,
+    require_required_env,
     safe_jsonable,
+    stage_image_file_for_input,
 )
 
 import datetime
@@ -50,6 +62,11 @@ AVAILABLE_TOOLS = [
     TerminalKill(),
 ]
 AVAILABLE_TOOL_MAP = {tool.name: tool for tool in AVAILABLE_TOOLS}
+OPTIONAL_TOOLS = [
+    StrReplaceEditor(),
+]
+OPTIONAL_TOOL_MAP = {tool.name: tool for tool in OPTIONAL_TOOLS}
+ALL_TOOL_MAP = {**AVAILABLE_TOOL_MAP, **OPTIONAL_TOOL_MAP}
 DEFAULT_IMAGE_TOKEN_ESTIMATE = 1536
 DEFAULT_MODEL_NAME = "gpt-5.4"
 DEFAULT_MAX_LLM_CALLS = 100
@@ -384,7 +401,32 @@ def resolved_tool_names(function_list: Optional[Sequence[str]]) -> list[str]:
 
 
 def available_tool_schemas(function_list: Optional[Sequence[str]] = None) -> list[dict[str, Any]]:
-    return [tool_schema(AVAILABLE_TOOL_MAP[name]) for name in resolved_tool_names(function_list)]
+    names = resolved_tool_names(function_list)
+    unknown_tools = [name for name in names if name not in ALL_TOOL_MAP]
+    if unknown_tools:
+        raise ValueError(f"Unknown tools requested: {unknown_tools}")
+    return [tool_schema(ALL_TOOL_MAP[name]) for name in names]
+
+
+def resolve_extra_tool_names(extra_tools: Optional[Sequence[str]]) -> list[str]:
+    resolved: list[str] = []
+    for raw_name in extra_tools or []:
+        name = str(raw_name).strip()
+        if not name:
+            continue
+        if name not in OPTIONAL_TOOL_MAP:
+            raise ValueError(f"Unknown extra tool requested: {name}")
+        if name not in resolved:
+            resolved.append(name)
+    return resolved
+
+
+def default_tool_names(*, include_ask_user: bool = True, extra_tools: Optional[Sequence[str]] = None) -> list[str]:
+    names = [name for name in AVAILABLE_TOOL_MAP if include_ask_user or name != "AskUser"]
+    for name in resolve_extra_tool_names(extra_tools):
+        if name not in names:
+            names.append(name)
+    return names
 
 
 def normalized_tool_call(tool_call: Any) -> dict[str, Any]:
@@ -587,7 +629,7 @@ class MultiTurnReactAgent(BaseAgent):
         requested_tools = self.resolve_function_list(function_list)
         if requested_tools is None:
             requested_tools = list(AVAILABLE_TOOL_MAP.keys())
-        unknown_tools = [tool for tool in requested_tools if tool not in AVAILABLE_TOOL_MAP]
+        unknown_tools = [tool for tool in requested_tools if tool not in ALL_TOOL_MAP]
         if unknown_tools:
             raise ValueError(f"Unknown tools requested: {unknown_tools}")
         if "model" not in llm or not str(llm["model"]).strip():
@@ -595,7 +637,7 @@ class MultiTurnReactAgent(BaseAgent):
         if "generate_cfg" not in llm or not isinstance(llm["generate_cfg"], dict):
             raise ValueError('llm["generate_cfg"] must be a dict.')
 
-        self.tool_map = {tool_name: AVAILABLE_TOOL_MAP[tool_name] for tool_name in requested_tools}
+        self.tool_map = {tool_name: ALL_TOOL_MAP[tool_name] for tool_name in requested_tools}
         self.tool_names = list(self.tool_map.keys())
         self.model = str(llm["model"])
         self.llm_generate_cfg = llm["generate_cfg"]
@@ -1306,3 +1348,157 @@ class MultiTurnReactAgent(BaseAgent):
 
     def custom_call_tool(self, tool_name: str, tool_args: Any, **kwargs):
         return execute_tool_by_name(self.tool_map, tool_name, tool_args, **kwargs)
+
+
+def _path_has_suffix(path: Path, suffix_parts: Sequence[str]) -> bool:
+    normalized_parts = tuple(part.casefold() for part in path.parts)
+    normalized_suffix = tuple(part.casefold() for part in suffix_parts)
+    if len(normalized_parts) < len(normalized_suffix):
+        return False
+    return normalized_parts[-len(normalized_suffix) :] == normalized_suffix
+
+
+def resolve_agent_class_for_role_prompt_files(role_prompt_files: Sequence[str]) -> Type[MultiTurnReactAgent]:
+    for raw_path in role_prompt_files:
+        path_text = str(raw_path).strip()
+        if not path_text:
+            continue
+        path = Path(path_text).expanduser().resolve(strict=False)
+        if _path_has_suffix(path, ("benchmarks", "ResearchClawBench", "role_prompt.md")):
+            from benchmarks.ResearchClawBench.adapter import ResearchClawBenchAgent
+
+            return ResearchClawBenchAgent
+    return MultiTurnReactAgent
+
+
+def _parse_cli_args(argv: list[str]) -> tuple[str, Optional[str], Optional[str], str, list[str], list[str], Optional[bool], list[str]]:
+    parser = argparse.ArgumentParser(description="Run the local agent directly from agent_base.react_agent.")
+    parser.add_argument("prompt", nargs="*", help="Prompt text.")
+    parser.add_argument("--prompt-file", help="Optional UTF-8 text file containing the prompt.")
+    parser.add_argument("--trace-dir", help="Optional directory where the run trace JSONL should be created.")
+    parser.add_argument(
+        "--workspace-root",
+        help="Optional workspace root for local file tools, Bash, and TerminalStart.",
+    )
+    parser.add_argument(
+        "--role-prompt-file",
+        action="append",
+        default=[],
+        dest="role_prompt_files",
+        metavar="PATH",
+        help="Append one role-specific prompt file to the base system prompt. May be passed multiple times.",
+    )
+    parser.add_argument(
+        "--images",
+        action="append",
+        nargs="+",
+        default=[],
+        dest="image_paths",
+        metavar="PATH",
+        help="Attach one or more local image paths to the initial user message.",
+    )
+    parser.add_argument(
+        "--chat",
+        action=argparse.BooleanOptionalAction,
+        default=None,
+        help="Continue asking for follow-up user messages after each final answer. Defaults to on only in an interactive terminal.",
+    )
+    parser.add_argument(
+        "--extra-tool",
+        action="append",
+        default=[],
+        dest="extra_tools",
+        metavar="NAME",
+        help="Enable one optional extra tool for this run. Currently supported: str_replace_editor. May be passed multiple times.",
+    )
+    args = parser.parse_args(argv)
+
+    prompt_text = ""
+    if args.prompt_file:
+        prompt_text = Path(args.prompt_file).read_text(encoding="utf-8").strip()
+    elif args.prompt:
+        prompt_text = " ".join(args.prompt).strip()
+
+    if not prompt_text:
+        raise ValueError("A non-empty prompt is required via positional args or --prompt-file.")
+    role_prompt = read_role_prompt_files(args.role_prompt_files)
+    return (
+        prompt_text,
+        args.trace_dir,
+        args.workspace_root,
+        role_prompt,
+        list(args.role_prompt_files),
+        [path for group in args.image_paths for path in group],
+        args.chat,
+        resolve_extra_tool_names(args.extra_tools),
+    )
+
+
+def main(argv: Optional[list[str]] = None) -> int:
+    load_dotenv(PROJECT_ROOT / ".env")
+    try:
+        require_required_env("ResearchHarness agent")
+        prompt_text, trace_dir, workspace_root, role_prompt, role_prompt_files, image_paths, chat_arg, extra_tools = _parse_cli_args(argv or sys.argv[1:])
+        agent_cls = resolve_agent_class_for_role_prompt_files(role_prompt_files)
+        forbidden_tools = set(getattr(agent_cls, "forbidden_tool_names", set()))
+        agent = agent_cls(
+            function_list=(
+                default_tool_names(include_ask_user="AskUser" not in forbidden_tools, extra_tools=extra_tools)
+                if extra_tools
+                else None
+            ),
+            llm=default_llm_config(),
+            trace_dir=trace_dir,
+            role_prompt=role_prompt or None,
+        )
+        resolved_workspace_root = normalize_workspace_root(workspace_root)
+        initial_content_parts: list[dict[str, Any]] = []
+        saved_image_paths: list[str] = []
+        for image_index, image_path in enumerate(image_paths):
+            saved_path, data_url = stage_image_file_for_input(
+                image_path,
+                workspace_root=resolved_workspace_root,
+                image_index=image_index,
+            )
+            saved_image_paths.append(saved_path)
+            initial_content_parts.extend(image_input_content_parts(data_url, saved_path))
+        run_prompt = append_saved_image_paths_to_prompt(prompt_text, saved_image_paths)
+        printer = ConsoleEventPrinter(
+            model_name=agent.model,
+            workspace_root=resolved_workspace_root,
+            prompt=run_prompt,
+        )
+        printer.print_header()
+        session = agent._run_session(
+            run_prompt,
+            workspace_root=str(resolved_workspace_root),
+            event_callback=printer.handle_event,
+            initial_content_parts=initial_content_parts or None,
+        )
+        chat_enabled = chat_arg if chat_arg is not None else (sys.stdin.isatty() and sys.stdout.isatty())
+        messages = session.get("messages", [])
+        while chat_enabled:
+            try:
+                followup = input("\n[ResearchHarness] Follow-up (Ctrl+C to exit): ").strip()
+            except (KeyboardInterrupt, EOFError):
+                print("\n[ResearchHarness] Chat ended.")
+                break
+            if not followup:
+                continue
+            print(f"\n[ResearchHarness] Continuing conversation: {followup}")
+            printer.reset_rounds()
+            session = agent._run_session(
+                followup,
+                workspace_root=str(resolved_workspace_root),
+                event_callback=printer.handle_event,
+                prior_messages=messages,
+            )
+            messages = session.get("messages", messages)
+        return 0
+    except (MissingRequiredEnvError, ValueError) as exc:
+        print(str(exc), file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

agent_base/tools/__init__.py

@@ -10,6 +10,7 @@ __all__ = [
     "ReadImage",
     "ReadPDF",
     "ScholarSearch",
+    "StrReplaceEditor",
     "TerminalInterrupt",
     "TerminalKill",
     "TerminalRead",
@@ -30,6 +31,7 @@ _EXPORT_TO_MODULE = {
     "ReadImage": "agent_base.tools.tool_file",
     "ReadPDF": "agent_base.tools.tool_file",
     "ScholarSearch": "agent_base.tools.tool_web",
+    "StrReplaceEditor": "agent_base.tools.tool_extra",
     "TerminalInterrupt": "agent_base.tools.tool_runtime",
     "TerminalKill": "agent_base.tools.tool_runtime",
     "TerminalRead": "agent_base.tools.tool_runtime",

agent_base/tools/tool_extra.py

@@ -0,0 +1,343 @@
+from __future__ import annotations
+
+import json
+import threading
+import zipfile
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Optional, Union
+from xml.etree import ElementTree
+
+from agent_base.tools.tool_file import ReadPDF
+from agent_base.tools.tooling import ToolBase, validate_tool_path
+
+
+DEFAULT_EDITOR_MAX_CHARS = 20000
+BINARY_MARKDOWN_SUFFIXES = {".xlsx", ".pptx", ".wav", ".mp3", ".m4a", ".flac", ".pdf", ".docx"}
+
+
+@dataclass
+class _UndoRecord:
+    existed: bool
+    previous_text: str
+
+
+class StrReplaceEditor(ToolBase):
+    name = "str_replace_editor"
+    description = (
+        "Optional compatibility editor for viewing, creating, exact string replacement, "
+        "line insertion, and undoing the last edit to a workspace text file."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "command": {
+                "type": "string",
+                "description": "Allowed commands: view, create, str_replace, insert, undo_edit.",
+                "enum": ["view", "create", "str_replace", "insert", "undo_edit"],
+            },
+            "path": {
+                "type": "string",
+                "description": "Absolute path to a file or directory inside the workspace.",
+            },
+            "file_text": {
+                "type": "string",
+                "description": "Required for create; full text content of the file to create.",
+            },
+            "old_str": {
+                "type": "string",
+                "description": "Required for str_replace; exact unique string to replace.",
+            },
+            "new_str": {
+                "type": "string",
+                "description": "Replacement text for str_replace, or text block for insert.",
+            },
+            "insert_line": {
+                "type": "integer",
+                "description": (
+                    "Required for insert; insert new_str after this 1-based line. "
+                    "Use 0 to insert at the start."
+                ),
+            },
+            "view_range": {
+                "type": "array",
+                "description": "Optional [start_line, end_line] for file view. Use [start_line, -1] to read to EOF.",
+                "items": {"type": "integer"},
+            },
+        },
+        "required": ["command", "path"],
+        "additionalProperties": False,
+    }
+
+    def __init__(self, cfg: Optional[dict] = None):
+        super().__init__(cfg)
+        self._undo_records: dict[Path, _UndoRecord] = {}
+        self._lock = threading.Lock()
+
+    def call(self, params: Union[str, dict], **kwargs: Any) -> str:
+        try:
+            parsed = self.parse_json_args(params)
+            command = str(parsed.get("command", "")).strip()
+            path = self._resolve_absolute_path(
+                str(parsed.get("path", "")),
+                workspace_root=kwargs.get("workspace_root"),
+            )
+        except ValueError as exc:
+            return f"[str_replace_editor] {exc}"
+
+        if command == "view":
+            return self._view(path, parsed, workspace_root=kwargs.get("workspace_root"))
+        if command == "create":
+            return self._create(path, parsed)
+        if command == "str_replace":
+            return self._str_replace(path, parsed)
+        if command == "insert":
+            return self._insert(path, parsed)
+        if command == "undo_edit":
+            return self._undo_edit(path)
+        return "[str_replace_editor] command must be one of: view, create, str_replace, insert, undo_edit."
+
+    def _resolve_absolute_path(self, path_value: str, *, workspace_root: Any) -> Path:
+        path_text = str(path_value or "").strip()
+        if not path_text:
+            raise ValueError("path must be a non-empty absolute path.")
+        path = Path(path_text).expanduser()
+        if not path.is_absolute():
+            raise ValueError("path must be absolute.")
+        return validate_tool_path(path, "str_replace_editor access", base_root=workspace_root)
+
+    def _view(self, path: Path, params: dict[str, Any], *, workspace_root: Any) -> str:
+        if not path.exists():
+            return f"[str_replace_editor] Path not found: {path}"
+        if path.is_dir():
+            return self._clip(self._view_directory(path))
+        if not path.is_file():
+            return f"[str_replace_editor] Path is neither a file nor a directory: {path}"
+        if path.suffix.lower() in BINARY_MARKDOWN_SUFFIXES:
+            return self._clip(self._view_binary_markdown(path, workspace_root=workspace_root))
+        try:
+            text = self._read_text(path)
+        except (OSError, UnicodeDecodeError) as exc:
+            return f"[str_replace_editor] Error reading text file: {exc}"
+        try:
+            selected, start_line, end_line = self._select_view_range(text, params.get("view_range"))
+        except ValueError as exc:
+            return f"[str_replace_editor] {exc}"
+        numbered = self._number_lines(selected, start_line=start_line)
+        header = {
+            "path": str(path),
+            "start_line": start_line,
+            "end_line": end_line,
+            "total_lines": len(text.splitlines()),
+        }
+        return self._clip(json.dumps(header, ensure_ascii=False) + "\n" + numbered)
+
+    def _create(self, path: Path, params: dict[str, Any]) -> str:
+        if "file_text" not in params:
+            return "[str_replace_editor] Missing required parameter for create: file_text"
+        if path.exists():
+            return f"[str_replace_editor] Cannot create because file already exists: {path}"
+        if not path.parent.exists() or not path.parent.is_dir():
+            return f"[str_replace_editor] Parent directory does not exist: {path.parent}"
+        file_text = str(params.get("file_text", ""))
+        try:
+            path.write_text(file_text, encoding="utf-8")
+        except OSError as exc:
+            return f"[str_replace_editor] Error creating file: {exc}"
+        with self._lock:
+            self._undo_records[path] = _UndoRecord(existed=False, previous_text="")
+        return f"[str_replace_editor] Created file: {path}"
+
+    def _str_replace(self, path: Path, params: dict[str, Any]) -> str:
+        if "old_str" not in params:
+            return "[str_replace_editor] Missing required parameter for str_replace: old_str"
+        old_str = str(params.get("old_str", ""))
+        new_str = str(params.get("new_str", ""))
+        if not old_str:
+            return "[str_replace_editor] old_str must be non-empty."
+        if old_str == new_str:
+            return "[str_replace_editor] old_str and new_str must be different."
+        try:
+            original = self._read_existing_file(path)
+        except (ValueError, OSError, UnicodeDecodeError) as exc:
+            return f"[str_replace_editor] {exc}"
+        match_count = original.count(old_str)
+        if match_count == 0:
+            return "[str_replace_editor] old_str was not found exactly once; found 0 matches."
+        if match_count > 1:
+            return f"[str_replace_editor] old_str must be unique; found {match_count} matches."
+        updated = original.replace(old_str, new_str, 1)
+        try:
+            path.write_text(updated, encoding="utf-8")
+        except OSError as exc:
+            return f"[str_replace_editor] Error writing file: {exc}"
+        with self._lock:
+            self._undo_records[path] = _UndoRecord(existed=True, previous_text=original)
+        return f"[str_replace_editor] Replaced text in file: {path}"
+
+    def _insert(self, path: Path, params: dict[str, Any]) -> str:
+        if "insert_line" not in params:
+            return "[str_replace_editor] Missing required parameter for insert: insert_line"
+        if "new_str" not in params:
+            return "[str_replace_editor] Missing required parameter for insert: new_str"
+        try:
+            insert_line = int(params.get("insert_line"))
+        except (TypeError, ValueError):
+            return "[str_replace_editor] insert_line must be an integer."
+        new_str = str(params.get("new_str", ""))
+        try:
+            original = self._read_existing_file(path)
+        except (ValueError, OSError, UnicodeDecodeError) as exc:
+            return f"[str_replace_editor] {exc}"
+        lines = original.splitlines(keepends=True)
+        if insert_line < 0 or insert_line > len(lines):
+            return f"[str_replace_editor] insert_line must be between 0 and {len(lines)}."
+        insert_text = new_str if new_str.endswith("\n") else new_str + "\n"
+        updated = "".join(lines[:insert_line]) + insert_text + "".join(lines[insert_line:])
+        try:
+            path.write_text(updated, encoding="utf-8")
+        except OSError as exc:
+            return f"[str_replace_editor] Error writing file: {exc}"
+        with self._lock:
+            self._undo_records[path] = _UndoRecord(existed=True, previous_text=original)
+        return f"[str_replace_editor] Inserted text after line {insert_line} in file: {path}"
+
+    def _undo_edit(self, path: Path) -> str:
+        with self._lock:
+            record = self._undo_records.pop(path, None)
+        if record is None:
+            return f"[str_replace_editor] No edit to undo for file: {path}"
+        try:
+            if record.existed:
+                path.write_text(record.previous_text, encoding="utf-8")
+                return f"[str_replace_editor] Reverted last edit to file: {path}"
+            if path.exists():
+                path.unlink()
+            return f"[str_replace_editor] Removed file created by last edit: {path}"
+        except OSError as exc:
+            return f"[str_replace_editor] Error undoing edit: {exc}"
+
+    def _read_existing_file(self, path: Path) -> str:
+        if not path.exists():
+            raise ValueError(f"File not found: {path}")
+        if not path.is_file():
+            raise ValueError(f"Path is not a file: {path}")
+        if path.suffix.lower() in BINARY_MARKDOWN_SUFFIXES:
+            raise ValueError(f"Editing binary/derived formats is not supported: {path}")
+        return self._read_text(path)
+
+    def _read_text(self, path: Path) -> str:
+        return path.read_text(encoding="utf-8")
+
+    def _select_view_range(self, text: str, view_range: Any) -> tuple[list[str], int, int]:
+        lines = text.splitlines()
+        if view_range is None:
+            return lines, 1, len(lines)
+        if not isinstance(view_range, list) or len(view_range) != 2:
+            raise ValueError("view_range must be [start_line, end_line].")
+        try:
+            start_line = int(view_range[0])
+            end_line = int(view_range[1])
+        except (TypeError, ValueError) as exc:
+            raise ValueError("view_range entries must be integers.") from exc
+        if start_line < 1:
+            raise ValueError("view_range start_line must be >= 1.")
+        if end_line != -1 and end_line < start_line:
+            raise ValueError("view_range end_line must be -1 or >= start_line.")
+        resolved_end = len(lines) if end_line == -1 else min(end_line, len(lines))
+        return lines[start_line - 1:resolved_end], start_line, resolved_end
+
+    def _number_lines(self, lines: list[str], *, start_line: int) -> str:
+        return "\n".join(f"{line_number:6}\t{line}" for line_number, line in enumerate(lines, start=start_line))
+
+    def _view_directory(self, path: Path) -> str:
+        rows = [str(path)]
+        for child in self._iter_non_hidden(path):
+            rel = child.relative_to(path)
+            rows.append(f"{rel}/" if child.is_dir() else str(rel))
+            if child.is_dir():
+                for grandchild in self._iter_non_hidden(child):
+                    grand_rel = grandchild.relative_to(path)
+                    rows.append(f"{grand_rel}/" if grandchild.is_dir() else str(grand_rel))
+        return "\n".join(rows)
+
+    def _iter_non_hidden(self, path: Path) -> list[Path]:
+        try:
+            return sorted(
+                (child for child in path.iterdir() if not child.name.startswith(".")),
+                key=lambda item: item.name.casefold(),
+            )
+        except OSError:
+            return []
+
+    def _view_binary_markdown(self, path: Path, *, workspace_root: Any) -> str:
+        suffix = path.suffix.lower()
+        if suffix == ".pdf":
+            return ReadPDF().call(
+                {"path": str(path), "max_chars": DEFAULT_EDITOR_MAX_CHARS},
+                workspace_root=workspace_root,
+            )
+        if suffix == ".docx":
+            return self._extract_docx_text(path)
+        if suffix == ".pptx":
+            return self._extract_pptx_text(path)
+        if suffix == ".xlsx":
+            return self._extract_xlsx_text(path)
+        stat = path.stat()
+        return f"# {path.name}\n\nBinary media file.\n\n- path: {path}\n- size_bytes: {stat.st_size}"
+
+    def _extract_docx_text(self, path: Path) -> str:
+        try:
+            with zipfile.ZipFile(path) as archive:
+                xml = archive.read("word/document.xml")
+        except (KeyError, OSError, zipfile.BadZipFile) as exc:
+            return f"[str_replace_editor] Error reading DOCX: {exc}"
+        try:
+            text = self._xml_text(xml)
+        except ElementTree.ParseError as exc:
+            return f"[str_replace_editor] Error reading DOCX XML: {exc}"
+        return f"# {path.name}\n\n" + text
+
+    def _extract_pptx_text(self, path: Path) -> str:
+        try:
+            with zipfile.ZipFile(path) as archive:
+                slide_names = sorted(
+                    name
+                    for name in archive.namelist()
+                    if name.startswith("ppt/slides/slide") and name.endswith(".xml")
+                )
+                sections = []
+                for index, name in enumerate(slide_names, start=1):
+                    text = self._xml_text(archive.read(name)).strip()
+                    if text:
+                        sections.append(f"## Slide {index}\n\n{text}")
+        except (OSError, zipfile.BadZipFile, ElementTree.ParseError) as exc:
+            return f"[str_replace_editor] Error reading PPTX: {exc}"
+        return f"# {path.name}\n\n" + "\n\n".join(sections)
+
+    def _extract_xlsx_text(self, path: Path) -> str:
+        try:
+            with zipfile.ZipFile(path) as archive:
+                names = sorted(
+                    name
+                    for name in archive.namelist()
+                    if name.startswith("xl/worksheets/") and name.endswith(".xml")
+                )
+                sections = []
+                for index, name in enumerate(names, start=1):
+                    text = self._xml_text(archive.read(name)).strip()
+                    if text:
+                        sections.append(f"## Sheet {index}\n\n{text}")
+        except (OSError, zipfile.BadZipFile, ElementTree.ParseError) as exc:
+            return f"[str_replace_editor] Error reading XLSX: {exc}"
+        return f"# {path.name}\n\n" + "\n\n".join(sections)
+
+    def _xml_text(self, xml_bytes: bytes) -> str:
+        root = ElementTree.fromstring(xml_bytes)
+        values = [text.strip() for text in root.itertext() if text and text.strip()]
+        return "\n".join(values)
+
+    def _clip(self, text: str, max_chars: int = DEFAULT_EDITOR_MAX_CHARS) -> str:
+        if len(text) <= max_chars:
+            return text
+        return text[:max_chars] + "\n<response clipped>"

agent_base/utils.py

@@ -87,6 +87,21 @@ def require_required_env(context: str = "ResearchHarness") -> None:
     )
 
 
+def read_role_prompt_files(paths: Iterable[str]) -> str:
+    blocks: list[str] = []
+    for raw_path in paths:
+        path_text = str(raw_path).strip()
+        if not path_text:
+            continue
+        path = Path(path_text).expanduser()
+        if not path.exists():
+            raise ValueError(f"Role prompt file does not exist: {path}")
+        if not path.is_file():
+            raise ValueError(f"Role prompt path is not a file: {path}")
+        blocks.append(path.read_text(encoding="utf-8").strip())
+    return "\n\n".join(block for block in blocks if block.strip())
+
+
 def _safe_image_stem(name: str, fallback: str) -> str:
     stem = re.sub(r"[^A-Za-z0-9_.-]+", "_", Path(name).stem).strip("._")
     return stem or fallback