refactor(file_io): update file I/O operations and truncation logic (#177)

* refactor(file_io): update file I/O operations and truncation logic

* refactor(memory): update file-based memory compaction logic

docs/copaw_context_design.md

@@ -0,0 +1,122 @@
+## Copaw Context Management V2
+
+> 注：不涉及长期记忆
+
+### 上下文数据结构
+
+#### 1. 上下文-内存
+
+- **compact_summary**（可选）：
+    - **历史对话原始数据引导**：存储于 `dialog/YYYY-MM-DD.jsonl`，共 N 行，按时间顺序排列；回顾时建议从后往前读。
+    - **历史对话摘要**：包含 `Goal + Constraints + Progress + KeyDecisions + NextSteps`。
+- **messages**：当前对话上下文（完整消息列表）。
+
+#### 2. 上下文-缓存到文件系统
+
+- **历史对话原始数据**：`dialog/YYYY-MM-DD.jsonl`
+- **工具调用结果原始数据**：`tool_result/{uuid}.txt`（保留 N 天）
+
+```mermaid
+flowchart TD
+    A[Context] --> B[compact_summary]
+    B --> C[dialog path guide + Goal/Constraints/Progress/KeyDecisions/NextSteps]
+    A --> E[messages: full dialogue history]
+    A --> F[File System Cache] --> G[dialog/YYYY-MM-DD.jsonl]
+    F --> H[tool_result/uuid.txt N-day TTL]
+```
+
+---
+
+### 上下文机制（Pre-Reasoning Hook）
+
+1. **工具结果 Offload** (`ToolCallResultCompact`)
+2. **上下文检查** (`ContextChecker`)
+3. **若 Token 超阈值**：
+    - 保留最近 **X%** 的 Token（保障连贯性）
+    - 其余历史对话生成摘要 (`Compactor`)
+4. **被摘要的上下文 Offload 到文件系统** (`SaveDialog`)
+
+```mermaid
+flowchart LR
+    A[Pre-Reasoning Hook] --> B[ToolCallResultCompact]
+    B --> C[ContextChecker]
+    C --> D{Token > Threshold?}
+    D -->|Yes| E[Keep recent X% tokens]
+    E --> F[Compact & Summary old context]
+    F --> G[SaveDialog: offload to file]
+    D -->|No| H[Proceed normally]
+```
+
+---
+
+### 工具结果 Offload 机制
+
+1. 所有工具调用结果先放入上下文，等待 Pre-Reasoning Hook 处理。
+2. 根据是否属于 **recent_n** 范围，决定截断策略：
+    - **recent_n 内**：近期内容 → 低截断比例
+    - **recent_n 外**：远期内容 → 高截断比例
+
+#### 示例：Browser Use 类工具
+
+| 阶段 | 行为                                                                           |
+|----|------------------------------------------------------------------------------|
+| 1  | 原始工具调用结果                                                                     |
+| 2  | 保存原始内容到文件：– 若在 recent_n 内：截断较少– 附注：“FullText saved to xxxx”– 提示：“请从第 N 行开始读” |
+| 3  | 若再次引用且超出 recent_n：– 二次截断（更激进）– 仍指向原文件路径                                      |
+
+```mermaid
+flowchart LR
+    A[Tool Call Result] --> B{Within recent_n?}
+    B -->|Yes| C[Low truncation<br>Save full text to tool_result/uuid.txt<br>Hint: 'Read from line N']
+    B -->|No| D[High truncation<br>Reference existing file<br>More aggressive truncation]
+    C --> E[Context includes snippet + file ref]
+    D --> E
+```
+
+---
+
+### ReadFile 工具调用结果变化示例
+
+| 阶段 | 行为                                          |
+|----|---------------------------------------------|
+| 1  | 原始工具调用结果                                    |
+| 2  | 若在 recent_n 内：– 不截断– 不保存文件（因内容已由用户指定）       |
+| 3  | 若超出 recent_n：– 二次截断（更小）– 保存 FullText 到文件并引用 |
+
+> 注：ReadFile 本身读取的是外部文件，因此首次调用通常无需重复保存。
+
+```mermaid
+flowchart LR
+    A[ReadFile Result] --> B{Within recent_n?}
+    B -->|Yes| C[No truncation<br>No file save needed]
+    B -->|No| D[Apply secondary truncation<br>Save FullText to tool_result/uuid.txt]
+    C --> E[Include full content in context]
+    D --> F[Include snippet + file ref]
+```
+
+---
+
+## Copaw Memory
+
+### 触发逻辑
+
+1. **主 Agent 主动写入**：
+    - `Memory.md`（长期记忆主干）
+    - `YYYY-MM-DD.md`（当日日志）
+2. **触发阈值时**，由 **Summarizer（React Agent）** 写日志：
+    - 个性化信息（如偏好、习惯）
+    - Try-error 信息（失败尝试与修正）
+3. **定时任务**（每日 00:00）：
+    - 汇总最近的 `YYYY-MM-DD.md` 文件
+    - 更新 `Memory.md`
+
+```mermaid
+flowchart TD
+    A[Main Agent] --> B[Write Memory.md]
+    A --> C[Write YYYY-MM-DD.md]
+    D[Context Threshold Reached?] -->|Yes| E[Summarizer Agent]
+    E --> F[Log: Personalization]
+    E --> G[Log: Try-Error Info]
+    H[Cron @ 00:00 daily] --> I[Aggregate recent YYYY-MM-DD.md]
+I --> J[Update Memory.md]
+```

reme/config/light.yaml

@@ -3,6 +3,8 @@ as_llms:
     backend: openai
     model_name: qwen3.5-plus
 
+thread_pool_max_workers: -1
+
 as_llm_formatters:
   default:
     backend: openai

reme/core/application.py

@@ -156,13 +156,15 @@ class Application:
             if not ray.is_initialized():
                 ray.init(num_cpus=self.service_config.ray_max_workers)
 
-        if (
+        if self.service_config.thread_pool_max_workers > 0 and (
             self.service_context.thread_pool is None
             or self.service_context.thread_pool._shutdown  # pylint: disable=protected-access
         ):
             self.service_context.thread_pool = ThreadPoolExecutor(
                 max_workers=self.service_config.thread_pool_max_workers,
             )
+        elif self.service_config.thread_pool_max_workers <= 0:
+            logger.info("Thread pool is disabled (thread_pool_max_workers <= 0)")
 
         if self.service_context.service_config.enable_logo:
             print_logo(service_config=self.service_config)
@@ -518,7 +520,7 @@ class Application:
 
     def shutdown_thread_pool(self, wait: bool = True):
         """Shutdown the thread pool executor."""
-        if self.service_context.thread_pool:
+        if self.service_context.thread_pool is not None:
             self.service_context.thread_pool.shutdown(wait=wait)
 
     def shutdown_ray(self, wait: bool = True):

reme/core/file_watcher/base_file_watcher.py

@@ -29,12 +29,13 @@ class BaseFileWatcher:
         watch_paths: list[str] | str,
         suffix_filters: list[str] | None = None,
         recursive: bool = False,
-        debounce: int = 500,  # Millisecond debounce
+        debounce: int = 2000,
         chunk_tokens: int = 400,
         chunk_overlap: int = 80,
         file_store: BaseFileStore | None = None,
         callback: Callable[[set[tuple[Change, str]]], None | Coroutine[Any, Any, None]] | None = None,
         rebuild_index_on_start: bool = True,
+        poll_delay_ms: int = 1000,
         **kwargs,
     ):
         """
@@ -51,6 +52,7 @@ class BaseFileWatcher:
             callback: Callback function for changes
             rebuild_index_on_start: If True, clear all indexed data on start and rescan existing files.
                            If False, only monitor new changes without initialization.
+            poll_delay_ms: Polling delay in milliseconds. If > 300ms, force_polling will be enabled automatically.
             **kwargs: Additional keyword arguments
         """
         self.watch_paths: list[str] = [watch_paths] if isinstance(watch_paths, str) else watch_paths
@@ -62,6 +64,7 @@ class BaseFileWatcher:
         self.file_store: BaseFileStore = file_store
         self.callback = callback
         self.rebuild_index_on_start: bool = rebuild_index_on_start
+        self.poll_delay_ms: int = poll_delay_ms
         self.kwargs: dict = kwargs
 
         self._stop_event = asyncio.Event()
@@ -178,11 +181,15 @@ class BaseFileWatcher:
 
             try:
                 logger.info(f"Starting watch on valid paths: {valid_paths}")
+                # Enable force_polling if poll_delay_ms > default 300ms to reduce CPU usage
+                force_polling = self.poll_delay_ms > 300
                 async for changes in awatch(
                     *valid_paths,
                     watch_filter=self.watch_filter,
                     recursive=self.recursive,
                     debounce=self.debounce,
+                    poll_delay_ms=self.poll_delay_ms,
+                    force_polling=force_polling,
                     stop_event=self._stop_event,
                 ):
                     if self._stop_event.is_set():

reme/core/op/base_op.py

@@ -301,7 +301,7 @@ class BaseOp(metaclass=ABCMeta):
 
     def submit_sync_task(self, fn: Callable, *args, **kwargs) -> "BaseOp":
         """Submit a task to the thread pool or local queue."""
-        if self.enable_parallel:
+        if self.enable_parallel and self.service_context.thread_pool is not None:
             task = self.service_context.thread_pool.submit(fn, *args, **kwargs)
         else:
             task = (fn, args, kwargs)

reme/core/schema/service_config.py

@@ -116,7 +116,10 @@ class ServiceConfig(BasicConfig):
     working_dir: str = Field(default=".reme")
     enable_logo: bool = Field(default=True)
     language: str = Field(default="")
-    thread_pool_max_workers: int = Field(default=16)
+    thread_pool_max_workers: int = Field(
+        default=16,
+        description="Number of thread pool workers. Set to -1 to disable thread pool.",
+    )
     ray_max_workers: int = Field(default=-1)
     log_to_console: bool = Field(default=True)
     disabled_flows: list[str] = Field(default_factory=list)

reme/memory/file_based/components/cli.py

@@ -113,9 +113,9 @@ class CliAgent(BaseOp):
 
         toolkit = Toolkit()
         file_io = FileIO(working_dir=self.working_dir)
-        toolkit.register_tool_function(file_io.read)
-        toolkit.register_tool_function(file_io.write)
-        toolkit.register_tool_function(file_io.edit)
+        toolkit.register_tool_function(file_io.read_file)
+        toolkit.register_tool_function(file_io.write_file)
+        toolkit.register_tool_function(file_io.edit_file)
 
         return toolkit
 

reme/memory/file_based/components/summarizer.yaml

@@ -1,28 +1,29 @@
 user_message: |
-   Memory Pre-compression Flush Cycle Initiated
+   Memory Pre-compression Flush Cycle.
 
    The current session is about to enter the automatic compression phase. Please capture persistent memory AND session reflections, then write them to disk.
 
    Current date: {date}
    Working directory: {working_dir}
 
+   # Task
    Immediately store persistent memory and reflections to: {memory_dir}/YYYY-MM-DD.md
 
-   Workflow:
-   1. First, `read` {memory_dir}/YYYY-MM-DD.md (if the file doesn’t exist, an error message will be returned).
-   2. Extract and synthesize content from the current session:
+   # Workflow
+   1. Extract and synthesize content from the current session:
       - Persistent Memory: Facts, user profile updates, project states, and important events.
       - Experience Reflection: Reusable thinking logic derived from user feedback, successful problem-solving strategies, mistakes made/pitfalls to avoid, and actionable insights for future interactions.
-   3. Intelligently merge new information with existing content (skip merging if the file doesn’t exist):
+   2. `read` {memory_dir}/YYYY-MM-DD.md (if the file doesn’t exist, an error message will be returned)
+      - If the file doesn’t exist, use `write` tool directly.
+      - If the file exists, intelligently merge new information with existing content, prefer using `edit` to update specific sections.
+      - Use `write` to overwrite the entire file only if substantial restructuring is required.
+
+   # Principles
+   - Intelligently merge new information with existing content:
       - Categorize clearly (e.g., separate "Factual Memory" from "Reflections & Logic").
       - Avoid duplicating already recorded information.
       - Enrich existing entries with new details where relevant.
       - Maintain chronological order wherever applicable.
-   4. Write the updated content:
-      - Prefer using `edit` to update specific sections when possible.
-      - Use `write` to overwrite the entire file only if substantial restructuring is required.
-
-   Principles:
    - Always preserve timestamps and any date/time-related context.
    - Add only genuinely new or meaningfully enriching information.
    - Reflections MUST focus on forming reusable cognitive frameworks based on user feedback, aiming to improve future task execution.
@@ -37,23 +38,23 @@ user_message_zh: |
    当前日期：{date}
    工作目录：{working_dir}
 
+   # 任务
    立即存储持久化记忆与反思（使用路径 {memory_dir}/YYYY-MM-DD.md）。
 
-   工作流程：
-   1. 先 `read` {memory_dir}/YYYY-MM-DD.md（如文件不存在，会返回错误提示）
-   2. 从当前会话中提取并综合两类内容：
+   # 工作流程
+   1. 从当前会话中提取并综合两类内容：
       - 持久化记忆：客观事实、用户信息更新、项目状态及重要事件。
       - 经验反思：基于用户反馈形成的可复用思考逻辑、成功的问题解决策略、犯下的错误/应避免的陷阱，以及对未来交互有帮助的行动指南。
-   3. 智能合并新信息与现有内容（若文件不存在则跳过合并）：
+   2. `read` {memory_dir}/YYYY-MM-DD.md（如文件不存在，会返回错误提示）
+      - 若文件不存在，直接使用 `write` 工具写入。
+      - 若文件已存在，智能合并新信息与现有内容，尽可能使用 `edit` 更新特定部分，仅在需要大幅重构时使用 `write` 覆盖整个文件。
+
+   # 原则
+   - 智能合并新信息与现有内容：
       - 将内容进行清晰的分类（例如明确区分“事实记忆”与“反思与逻辑”）。
       - 避免重复已记录的信息。
       - 在相关时丰富现有条目的新细节。
       - 在适用时保持时间顺序。
-   4. 写入更新后的内容：
-      - 尽可能使用 `edit` 更新特定部分。
-      - 如需大幅重构则使用 `write` 覆盖整个文件。
-
-   原则：
    - 始终保留时间戳、日期和时间相关上下文。
    - 仅添加真正新的或有丰富价值的信息。
    - 反思内容必须侧重于根据用户反馈构建可复用的思维逻辑，以改善未来的任务执行。

reme/memory/file_based/components/tool_result_compactor.py

@@ -1,33 +1,19 @@
 """Tool Result Compactor: truncate large tool results and save full content to files."""
 
+import os
+import sys
 import uuid
 from datetime import datetime, timedelta
 from pathlib import Path
 
 from agentscope.message import Msg
 
+from ..utils import truncate_text_output, DEFAULT_MAX_BYTES, TRUNCATION_NOTICE_MARKER
 from ....core.op import BaseOp
 from ....core.utils import get_logger
-from ....core.utils import truncate_text_head, TRUNCATION_MARKER_START
 
 logger = get_logger()
 
-MAX_LINE_LENGTH = 10000
-
-
-def _split_long_lines(text: str, max_len: int = MAX_LINE_LENGTH) -> str:
-    """Split lines that exceed max_len by inserting newlines."""
-    lines = text.split("\n")
-    result = []
-    for line in lines:
-        if len(line) <= max_len:
-            result.append(line)
-        else:
-            # Split line into chunks of max_len
-            for i in range(0, len(line), max_len):
-                result.append(line[i : i + max_len])
-    return "\n".join(result)
-
 
 class ToolResultCompactor(BaseOp):
     """Truncate large tool_result outputs and save full content to files."""
@@ -35,62 +21,50 @@ class ToolResultCompactor(BaseOp):
     def __init__(
         self,
         tool_result_dir: str | Path,
-        retention_days: int = 7,
+        retention_days: int = 3,
+        old_max_bytes: int = 3000,
+        recent_max_bytes: int = DEFAULT_MAX_BYTES,
         recent_n: int = 1,
-        old_threshold: int = 500,
-        recent_threshold: int = 30000,
+        encoding: str = "utf-8",
         **kwargs,
     ):
         super().__init__(**kwargs)
         self.tool_result_dir = Path(tool_result_dir)
         self.retention_days = retention_days
+        self.old_max_bytes = old_max_bytes
+        self.recent_max_bytes = recent_max_bytes
         self.recent_n = recent_n
-        self.old_threshold = old_threshold
-        self.recent_threshold = recent_threshold
-
-    def _save_and_truncate(self, content: str, tool_name: str, threshold: int) -> str:
-        """Save full content to file and return truncated version with file reference."""
-        if not content:
-            return content
-
-        # Check if content was previously truncated
-        if TRUNCATION_MARKER_START in content:
-            parts = content.split(TRUNCATION_MARKER_START, 1)
-            if len(parts[0]) <= threshold:
-                return content
-            return f"{truncate_text_head(parts[0], threshold)}{parts[1]}"
-
-        # Not truncated before
-        if len(content) <= threshold:
-            return content
-
-        # Save full content with long lines split
+        self.encoding = encoding
         self.tool_result_dir.mkdir(parents=True, exist_ok=True)
-        file_path = self.tool_result_dir / f"{uuid.uuid4().hex}.txt"
-        created_at = datetime.now().isoformat()
 
-        processed_content = _split_long_lines(content)
-        file_path.write_text(
-            f"# tool_name: {tool_name}\n# created_at: {created_at}\n# ---\n{processed_content}",
-            encoding="utf-8",
-        )
-        logger.debug("Saved tool result to %s (len=%d)", file_path, len(content))
+    def _compact(self, output: str | list[dict], max_bytes: int) -> str | list[dict]:
+        """Truncate output to max_bytes, saving full content to file if needed."""
 
-        # Return truncated with file reference
-        return f"{truncate_text_head(content, threshold)}\n\n[Full content saved to: {file_path}]"
+        def _truncate(content: str) -> str:
+            if not content:
+                return content
+
+            if TRUNCATION_NOTICE_MARKER in content:
+                return truncate_text_output(content, max_bytes=max_bytes)
+
+            if len(content.encode(self.encoding)) <= max_bytes + 100:
+                return content
+
+            saved_path: str | None = None
+            try:
+                fp = self.tool_result_dir / f"{uuid.uuid4().hex}.txt"
+                fp.write_text(content, encoding=self.encoding)
+                saved_path = str(fp)
+            except Exception as e:
+                logger.warning("Failed to save full tool result to file: %s", e)
+
+            return truncate_text_output(content, 1, content.count("\n") + 1, max_bytes, file_path=saved_path)
 
-    def _process_output(self, output: str | list[dict], tool_name: str, threshold: int) -> str | list[dict]:
-        """Process tool result output, truncating if necessary."""
         if isinstance(output, str):
-            return self._save_and_truncate(output, tool_name, threshold)
-
+            return _truncate(output)
         if isinstance(output, list):
             return [
-                (
-                    {**b, "text": self._save_and_truncate(b.get("text", ""), tool_name, threshold)}
-                    if isinstance(b, dict) and b.get("type") == "text"
-                    else b
-                )
+                {**b, "text": _truncate(b.get("text", ""))} if isinstance(b, dict) and b.get("type") == "text" else b
                 for b in output
             ]
         return output
@@ -101,43 +75,54 @@ class ToolResultCompactor(BaseOp):
         if not messages:
             return messages
 
-        # Split messages into old and recent parts
-        split_index = max(0, len(messages) - self.recent_n)
+        recent_n = 0
+        for msg in reversed(messages):
+            if not isinstance(msg.content, list) or not any(
+                isinstance(b, dict) and b.get("type") == "tool_result" for b in msg.content
+            ):
+                break
+            recent_n += 1
+        split_index = max(0, len(messages) - max(recent_n, self.recent_n))
 
         for idx, msg in enumerate(messages):
             if not isinstance(msg.content, list):
                 continue
-
-            # Determine threshold based on message position
-            threshold = self.recent_threshold if idx >= split_index else self.old_threshold
-
+            is_recent = idx >= split_index
+            max_bytes = self.recent_max_bytes if is_recent else self.old_max_bytes
             for block in msg.content:
-                if isinstance(block, dict) and block.get("type") == "tool_result":
-                    output = block.get("output")
-                    if output:
-                        block["output"] = self._process_output(output, block.get("name", "unknown"), threshold)
+                if isinstance(block, dict) and block.get("type") == "tool_result" and block.get("output"):
+                    block["output"] = self._compact(block["output"], max_bytes)
 
         return messages
 
     def cleanup_expired_files(self) -> int:
-        """Clean up files older than retention_days."""
+        """Clean up files older than retention_days.
+
+        Returns:
+            Number of files successfully deleted.
+        """
         if not self.tool_result_dir.exists():
             return 0
 
         cutoff = datetime.now() - timedelta(days=self.retention_days)
-        deleted = 0
+        deleted = failed = 0
 
         for fp in self.tool_result_dir.glob("*.txt"):
             try:
-                for line in fp.read_text(encoding="utf-8").splitlines()[:3]:
-                    if line.startswith("# created_at:"):
-                        if datetime.fromisoformat(line.split(":", 1)[1].strip()) < cutoff:
-                            fp.unlink()
-                            deleted += 1
-                        break
+                stat = os.stat(fp)
+                if sys.platform == "win32":
+                    ts = stat.st_ctime  # creation time on Windows
+                else:
+                    ts = getattr(stat, "st_birthtime", stat.st_mtime)  # macOS/BSD; Linux fallback to mtime
+                if datetime.fromtimestamp(ts) < cutoff:
+                    fp.unlink()
+                    deleted += 1
+            except FileNotFoundError:
+                pass  # deleted by another process between glob and stat/unlink
             except Exception as e:
-                logger.warning("Failed to process %s: %s", fp, e)
+                failed += 1
+                logger.warning("Failed to delete %s: %s", fp, e)
 
-        if deleted:
-            logger.info("Cleaned up %d expired files", deleted)
+        if deleted or failed:
+            logger.info("Cleaned up %d expired files (%d failed)", deleted, failed)
         return deleted

reme/memory/file_based/reme_in_memory_memory.py

@@ -126,9 +126,10 @@ class ReMeInMemoryMemory(InMemoryMemory):
 
         if prepend_summary and self._compressed_summary:
             previous_summary = f"""
+Raw conversation logs are in dialog/YYYY-MM-DD.jsonl (or nearby date files).
+Entries are chronological; read from the end for recent history.
 {self._compressed_summary}
-The above is a summary of previous conversation, use it as context to maintain continuity.
-                    """.strip()
+The above is a summary of previous conversation, use it as context to maintain continuity.""".strip()
 
             return [
                 Msg(

reme/memory/file_based/tools/file_io.py

@@ -7,7 +7,7 @@ from typing import Optional
 from agentscope.message import TextBlock
 from agentscope.tool import ToolResponse
 
-from ..utils import DEFAULT_MAX_BYTES, read_file_safe, truncate_output
+from ..utils import read_file_safe, truncate_text_output
 
 
 class FileIO:
@@ -38,13 +38,13 @@ class FileIO:
         else:
             return str(self.working_dir / file_path)
 
-    async def read(  # pylint: disable=too-many-return-statements
+    async def read_file(  # pylint: disable=too-many-return-statements
         self,
         file_path: str,
         start_line: Optional[int] = None,
         end_line: Optional[int] = None,
     ) -> ToolResponse:
-        """Read a file. Relative paths resolve from working_dir.
+        """Read a file. Relative paths resolve from WORKING_DIR.
 
         Use start_line/end_line to read a specific line range (output includes
         line numbers). Omit both to read the full file.
@@ -57,6 +57,34 @@ class FileIO:
             end_line (`int`, optional):
                 Last line to read (1-based, inclusive).
         """
+
+        # Convert start_line/end_line to int if they are strings
+        if start_line is not None:
+            try:
+                start_line = int(start_line)
+            except (ValueError, TypeError):
+                return ToolResponse(
+                    content=[
+                        TextBlock(
+                            type="text",
+                            text=f"Error: start_line must be an integer, got {start_line!r}.",
+                        ),
+                    ],
+                )
+
+        if end_line is not None:
+            try:
+                end_line = int(end_line)
+            except (ValueError, TypeError):
+                return ToolResponse(
+                    content=[
+                        TextBlock(
+                            type="text",
+                            text=f"Error: end_line must be an integer, got {end_line!r}.",
+                        ),
+                    ],
+                )
+
         file_path = self._resolve_file_path(file_path)
 
         if not os.path.exists(file_path):
@@ -111,29 +139,21 @@ class FileIO:
             # Extract selected lines
             selected_content = "\n".join(all_lines[s - 1 : e])
 
-            # Apply smart truncation (keep head for file reading)
-            truncated, was_truncated, output_lines, reason = truncate_output(selected_content, keep="head")
+            # Apply smart truncation (consistent with shell output format)
+            text = truncate_text_output(
+                selected_content,
+                start_line=s,
+                total_lines=total,
+                file_path=file_path,
+            )
 
-            # Build response with truncation hints
-            if was_truncated:
-                end_display = s + output_lines - 1
-                next_line = end_display + 1
-                if reason == "lines":
-                    hint = f"\n\n[Lines {s}-{end_display} of {total}. Use start_line={next_line} to continue.]"
-                else:
-                    hint = (
-                        f"\n\n[Lines {s}-{end_display} of {total} ({DEFAULT_MAX_BYTES // 1024}KB limit). "
-                        f"Use start_line={next_line} to continue.]"
-                    )
-                text = truncated + hint
-            elif e < total:
+            # Add continuation hint if partial read without truncation
+            if text == selected_content and e < total:
                 remaining = total - e
                 text = (
-                    f"{file_path}  (lines {s}-{e} of {total})\n{truncated}\n\n[{remaining} more lines. "
-                    f"Use start_line={e + 1} to continue.]"
+                    f"{file_path}  (lines {s}-{e} of {total})\n{text}\n\n"
+                    f"[{remaining} more lines. Use start_line={e + 1} to continue.]"
                 )
-            else:
-                text = truncated
 
             return ToolResponse(
                 content=[TextBlock(type="text", text=text)],
@@ -149,7 +169,7 @@ class FileIO:
                 ],
             )
 
-    async def write(
+    async def write_file(
         self,
         file_path: str,
         content: str,
@@ -195,7 +215,7 @@ class FileIO:
                 ],
             )
 
-    async def edit(
+    async def edit_file(
         self,
         file_path: str,
         old_text: str,
@@ -212,7 +232,7 @@ class FileIO:
             new_text (`str`):
                 Replacement text.
         """
-        response = await self.read(file_path=file_path)
+        response = await self.read_file(file_path=file_path)
         if response.content and len(response.content) > 0:
             error_text = response.content[0].get("text", "")
             if error_text.startswith("Error:"):
@@ -239,7 +259,7 @@ class FileIO:
             )
 
         new_content = content.replace(old_text, new_text)
-        write_response = await self.write(file_path=file_path, content=new_content)
+        write_response = await self.write_file(file_path=file_path, content=new_content)
 
         if write_response.content and len(write_response.content) > 0:
             write_text = write_response.content[0].get("text", "")

reme/memory/file_based/tools/shell.py

@@ -12,8 +12,6 @@ from pathlib import Path
 from agentscope.message import TextBlock
 from agentscope.tool import ToolResponse
 
-from ..utils import truncate_shell_output
-
 
 def _execute_subprocess_sync(
     cmd: str,
@@ -189,10 +187,6 @@ class Shell:
                         stdout_str = ""
                         stderr_str = stderr_suffix
 
-            # Apply output truncation
-            stdout_str = truncate_shell_output(stdout_str)
-            stderr_str = truncate_shell_output(stderr_str)
-
             # Format the response in a human-friendly way
             if returncode == 0:
                 # Success case: just show the output

reme/memory/file_based/utils/__init__.py

@@ -1,13 +1,12 @@
 """utils"""
 
 from .as_msg_handler import AsMsgHandler
-from .file_utils import truncate_output, truncate_shell_output, read_file_safe, DEFAULT_MAX_BYTES, DEFAULT_MAX_LINES
+from .file_utils import truncate_text_output, read_file_safe, DEFAULT_MAX_BYTES, TRUNCATION_NOTICE_MARKER
 
 __all__ = [
     "AsMsgHandler",
-    "truncate_output",
-    "truncate_shell_output",
+    "truncate_text_output",
     "read_file_safe",
     "DEFAULT_MAX_BYTES",
-    "DEFAULT_MAX_LINES",
+    "TRUNCATION_NOTICE_MARKER",
 ]

reme/memory/file_based/utils/file_utils.py

@@ -1,112 +1,141 @@
+# -*- coding: utf-8 -*-
 """Shared utilities for file and shell tools."""
 
-# Default truncation limits
-DEFAULT_MAX_LINES = 1000
-DEFAULT_MAX_BYTES = 30 * 1024  # 30KB
+import re
+
+from ....core.utils import get_logger
+
+logger = get_logger()
+
+# Default truncation limit
+DEFAULT_MAX_BYTES = 100 * 1024
+
+# Maximum file size to read into memory (1GB)
+MAX_FILE_READ_BYTES = 1024 * 1024 * 1024
+
+# Marker prepended to every truncation notice.
+# Format: <<<TRUNCATED>>>
+#         File: file_path
+#         Content from start_line=X, next N bytes.
+#         total_lines=Z
+#         Use start_line=Y to continue.
+# Split on this to recover the original (un-truncated) portion:
+#   original = output.split(TRUNCATION_NOTICE_MARKER)[0]
+TRUNCATION_NOTICE_MARKER = "<<<TRUNCATED>>>"
 
 
-def truncate_output(
+# pylint: disable=too-many-return-statements
+def truncate_text_output(
     text: str,
-    max_lines: int = DEFAULT_MAX_LINES,
+    start_line: int = 0,
+    total_lines: int = 0,
     max_bytes: int = DEFAULT_MAX_BYTES,
-    keep: str = "head",
-) -> tuple[str, bool, int, str]:
-    """Smart truncation for large content.
+    file_path: str | None = None,
+) -> str:
+    """Truncate file output by bytes with line integrity.
 
-    Args:
-        text: Text content to truncate.
-        max_lines: Maximum number of lines.
-        max_bytes: Maximum size in bytes.
-        keep: Which part to keep - "head" (first lines) or "tail" (last lines).
+    If text is under byte limit, return as-is.
+    If over limit, truncate at the last complete line that fits,
+    allowing the next read to start from a fresh line.
 
-    Returns:
-        (truncated_content, was_truncated, output_line_count, truncate_reason)
-    """
-    if not text:
-        return text, False, 0, ""
-
-    lines = text.split("\n")
-    total_lines = len(lines)
-
-    # No truncation needed
-    if total_lines <= max_lines and len(text.encode("utf-8")) <= max_bytes:
-        return text, False, total_lines, ""
-
-    # Apply line limit
-    if total_lines > max_lines:
-        if keep == "tail":
-            lines = lines[-max_lines:]
-        else:
-            lines = lines[:max_lines]
-        reason = "lines"
-    else:
-        reason = ""
-
-    # Apply byte limit
-    if len("\n".join(lines).encode("utf-8")) > max_bytes:
-        if keep == "tail":
-            while lines and len("\n".join(lines).encode("utf-8")) > max_bytes:
-                lines.pop(0)
-        else:
-            truncated = []
-            current_bytes = 0
-            for line in lines:
-                line_bytes = len(line.encode("utf-8")) + 1
-                if current_bytes + line_bytes > max_bytes:
-                    break
-                truncated.append(line)
-                current_bytes += line_bytes
-            lines = truncated
-        reason = "bytes"
-
-    return "\n".join(lines), True, len(lines), reason
-
-
-def truncate_shell_output(text: str) -> str:
-    """Truncate shell output to last N lines or M bytes, with truncation notice.
+    If TRUNCATION_NOTICE_MARKER is already in text (previously truncated),
+    extract the original content, re-truncate it, and update the
+    truncation notice using regex.
 
     Args:
         text: The output text to truncate.
+        start_line: The starting line number (1-based). Ignored when text already
+            contains a truncation notice (values are parsed from the notice instead).
+        total_lines: Total lines in the original file. Ignored when text already
+            contains a truncation notice (values are parsed from the notice instead).
+        max_bytes: Maximum size in bytes.
+        file_path: Optional file path to include in the truncation notice.
 
     Returns:
         Truncated text with notice if truncated.
     """
     if not text:
         return text
+    if max_bytes <= 0:
+        return text
 
     try:
-        total_lines = len(text.split("\n"))
-        truncated, was_truncated, output_lines, reason = truncate_output(text, keep="tail")
+        if TRUNCATION_NOTICE_MARKER in text:
+            parts = text.split(TRUNCATION_NOTICE_MARKER, 1)
+            original_content = parts[0]
+            old_notice = parts[1]
 
-        if not was_truncated:
-            return text
+            text_bytes = original_content.encode("utf-8")
+
+            # Allow a small slack to avoid re-truncating near-limit content
+            if len(text_bytes) <= max_bytes + 100:
+                return text
+
+            # Parse start_line and total_lines from notice; return text unchanged if not found
+            start_match = re.search(r"start_line=(\d+),", old_notice)
+            total_match = re.search(r"total_lines=(\d+)", old_notice)
+            if not start_match or not total_match:
+                return text
+            start_line_parsed = int(start_match.group(1))
+            total_lines_parsed = int(total_match.group(1))
+
+            truncated_bytes = text_bytes[:max_bytes]
+            result = truncated_bytes.decode("utf-8", errors="ignore")
+            newline_count = result.count("\n")
+
+            next_line = start_line_parsed + max(1, newline_count)
+
+            if not re.search(r"next \d+ bytes", old_notice):
+                return text
+            has_continuation = bool(re.search(r"Use start_line=\d+", old_notice))
+            new_notice = re.sub(r"next \d+ bytes", f"next {max_bytes} bytes", old_notice)
+            if has_continuation:
+                new_notice = re.sub(r"Use start_line=\d+", f"Use start_line={next_line}", new_notice)
+            elif next_line <= total_lines_parsed:
+                new_notice = re.sub(r"(total_lines=\d+)", f"\\1\nUse start_line={next_line} to continue.", new_notice)
+
+            return result + TRUNCATION_NOTICE_MARKER + new_notice
 
-        start_line = total_lines - output_lines + 1
-        if reason == "lines":
-            notice = f"\n\n[Output truncated: showing lines {start_line}-{total_lines} of {total_lines} total]"
         else:
+            text_bytes = text.encode("utf-8")
+
+            if len(text_bytes) <= max_bytes:
+                return text
+
+            truncated = text_bytes[:max_bytes]
+            result = truncated.decode("utf-8", errors="ignore")
+
+            newline_count = result.count("\n")
+
+            next_line = start_line + max(1, newline_count)
+
+            continuation = f"\nUse start_line={next_line} to continue." if next_line <= total_lines else ""
             notice = (
-                f"\n\n[Output truncated: showing lines {start_line}-{total_lines} of {total_lines} "
-                f"({DEFAULT_MAX_BYTES // 1024}KB limit)]"
+                TRUNCATION_NOTICE_MARKER
+                + f"\n\nFile: {file_path or ''}\nContent from start_line={start_line}, next {max_bytes} bytes."
+                f"\ntotal_lines={total_lines}{continuation}"
             )
 
-        return truncated + notice
+            return result + notice
+
     except Exception:
+        logger.warning("truncate_text_output failed, returning original text", exc_info=True)
         return text
 
 
-def read_file_safe(file_path: str) -> str:
-    """Read file with Unicode error handling.
+def read_file_safe(file_path: str, max_bytes: int = MAX_FILE_READ_BYTES) -> str:
+    """Read file with Unicode error handling and memory protection.
 
     Args:
         file_path: Path to the file.
+        max_bytes: Maximum bytes to read into memory (default 1GB).
 
     Returns:
-        File content as string.
+        File content as string (up to max_bytes).
     """
     try:
         with open(file_path, "r", encoding="utf-8") as f:
-            return f.read()
+            return f.read(max_bytes)
     except UnicodeDecodeError:
         with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
-            return f.read()
+            return f.read(max_bytes)

reme/reme_light.py

@@ -168,7 +168,7 @@ class ReMeLight(Application):
         Returns:
             Computed compaction threshold as an integer.
         """
-        return int(max_input_length * compact_ratio * 0.9)
+        return int(max_input_length * compact_ratio * 0.95)
 
     def _cleanup_tool_results(self) -> int:
         """
@@ -231,10 +231,10 @@ class ReMeLight(Application):
     async def compact_tool_result(
         self,
         messages: list[Msg],
+        old_max_bytes: int = 3000,
+        recent_max_bytes: int = 100 * 1024,
+        retention_days: int = 3,
         recent_n: int = 1,
-        old_threshold: int = 500,
-        recent_threshold: int = 30000,
-        retention_days: int = 7,
     ) -> list[Msg]:
         """
         Compact tool results by truncating large outputs and saving full content to files.
@@ -247,30 +247,37 @@ class ReMeLight(Application):
         Args:
             messages (list[Msg]): List of messages potentially containing tool results
                 that may need compaction.
-            recent_n (int): Number of recent messages to use recent_threshold for.
-                Default 1.
-            old_threshold (int): Character threshold for old messages. Default 500.
-            recent_threshold (int): Character threshold for recent messages. Default 30000.
+            old_max_bytes (int): Byte threshold for old (non-recent) messages. Default 3000.
+            recent_max_bytes (int): Byte threshold for recent messages (trailing consecutive
+                tool-result messages). Default 100KB (102400 bytes). Content exceeding this
+                limit is saved to disk; the message retains the first 100KB with a
+                read_file-style truncation notice and the saved file path.
             retention_days (int): Number of days to retain tool result files.
-                Default 7.
+                Default 3.
+            recent_n (int): Minimum number of most-recent tool-result messages to treat
+                as "recent" (using recent_max_bytes). The actual recent window is the
+                larger of this value and the trailing consecutive tool-result run.
+                Default 1.
 
         Returns:
             list[Msg]: The processed list of messages with large tool results compacted.
                 If an error occurs, returns the original unmodified messages.
 
         Note:
-            - Tool results are truncated based on old_threshold/recent_threshold
-            - Full content of truncated results is saved to tool_result_path
-            - Expired files are automatically cleaned up during this operation
+            - Recent tool results (trailing consecutive tool-result messages) are truncated
+              to recent_max_bytes using read_file-style output with a file path hint.
+            - Old tool results are truncated to old_max_bytes bytes.
+            - Full content of truncated results is saved to tool_result_path.
+            - Expired files are automatically cleaned up during this operation.
         """
         try:
             # Create compactor with instance configuration
             compactor = ToolResultCompactor(
                 tool_result_dir=self.tool_result_path,
                 retention_days=retention_days,
+                old_max_bytes=old_max_bytes,
+                recent_max_bytes=recent_max_bytes,
                 recent_n=recent_n,
-                old_threshold=old_threshold,
-                recent_threshold=recent_threshold,
             )
 
             # Execute compaction and get processed messages
@@ -455,9 +462,9 @@ class ReMeLight(Application):
             if toolkit is None:
                 toolkit = Toolkit()
                 file_io = FileIO(working_dir=str(self.working_path))
-                toolkit.register_tool_function(file_io.read)
-                toolkit.register_tool_function(file_io.write)
-                toolkit.register_tool_function(file_io.edit)
+                toolkit.register_tool_function(file_io.read_file)
+                toolkit.register_tool_function(file_io.write_file)
+                toolkit.register_tool_function(file_io.edit_file)
 
             summarizer = Summarizer(
                 working_dir=str(self.working_path),

tests/light/test_summarizer.py

@@ -101,9 +101,9 @@ def create_toolkit(working_dir: str) -> Toolkit:
     """Create a default Toolkit with FileIO tools for testing."""
     toolkit = Toolkit()
     file_io = FileIO(working_dir=working_dir)
-    toolkit.register_tool_function(file_io.read)
-    toolkit.register_tool_function(file_io.write)
-    toolkit.register_tool_function(file_io.edit)
+    toolkit.register_tool_function(file_io.read_file)
+    toolkit.register_tool_function(file_io.write_file)
+    toolkit.register_tool_function(file_io.edit_file)
     return toolkit
 
 

tests/light/test_tool_result_compactor.py

@@ -33,7 +33,7 @@ class TestToolResultCompactor:
     def test_no_truncation_when_under_threshold(self):
         """Test that short content is not truncated."""
         with tempfile.TemporaryDirectory() as tmpdir:
-            op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=1000)
+            op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=1000)
             messages = [create_tool_result_msg("short content")]
 
             result = asyncio.run(op.call(messages=messages))
@@ -45,7 +45,7 @@ class TestToolResultCompactor:
     def test_truncation_when_over_threshold(self):
         """Test that long content is truncated and saved to file."""
         with tempfile.TemporaryDirectory() as tmpdir:
-            op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=100)
+            op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=100)
             long_content = "x" * 500
             messages = [create_tool_result_msg(long_content)]
 
@@ -68,7 +68,7 @@ class TestToolResultCompactor:
     def test_skip_already_truncated(self):
         """Test that already truncated content is not re-truncated."""
         with tempfile.TemporaryDirectory() as tmpdir:
-            op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=100)
+            op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=100)
             truncated_content = "head<<<TRUNCATED>>>(100 chars omitted)<<<END_TRUNCATED>>>tail"
             messages = [create_tool_result_msg(truncated_content)]
 
@@ -80,7 +80,7 @@ class TestToolResultCompactor:
     def test_truncation_list_output(self):
         """Test truncation of list output with text blocks."""
         with tempfile.TemporaryDirectory() as tmpdir:
-            op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=100)
+            op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=100)
             list_output = [{"type": "text", "text": "y" * 500}]
             messages = [create_tool_result_msg(list_output)]
 
@@ -93,7 +93,7 @@ class TestToolResultCompactor:
     def test_list_output_no_truncation_when_short(self):
         """Test that short list output is not truncated."""
         with tempfile.TemporaryDirectory() as tmpdir:
-            op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=1000)
+            op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=1000)
             list_output = [{"type": "text", "text": "short"}]
             messages = [create_tool_result_msg(list_output)]
 
@@ -105,7 +105,7 @@ class TestToolResultCompactor:
     def test_list_output_multiple_text_blocks(self):
         """Test truncation of multiple text blocks in list output."""
         with tempfile.TemporaryDirectory() as tmpdir:
-            op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=100)
+            op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=100)
             list_output = [
                 {"type": "text", "text": "a" * 500},
                 {"type": "text", "text": "short"},
@@ -124,7 +124,7 @@ class TestToolResultCompactor:
     def test_list_output_mixed_block_types(self):
         """Test that non-text blocks in list output are unchanged."""
         with tempfile.TemporaryDirectory() as tmpdir:
-            op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=100)
+            op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=100)
             list_output = [
                 {"type": "text", "text": "c" * 500},
                 {"type": "image", "source": {"type": "url", "url": "http://example.com/img.png"}},
@@ -141,7 +141,7 @@ class TestToolResultCompactor:
     def test_cleanup_expired_files(self):
         """Test cleanup of expired files."""
         with tempfile.TemporaryDirectory() as tmpdir:
-            op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=100, retention_days=1)
+            op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=100, retention_days=1)
 
             # Create an old file
             old_time = (datetime.now() - timedelta(days=2)).isoformat()
@@ -162,7 +162,7 @@ class TestToolResultCompactor:
     def test_string_content_msg_unchanged(self):
         """Test that messages with string content are unchanged."""
         with tempfile.TemporaryDirectory() as tmpdir:
-            op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=100)
+            op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=100)
             messages = [Msg(name="user", role="user", content="hello world")]
 
             asyncio.run(op.call(messages=messages))

tests/light/test_tools.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import os
+import re
 import shutil
 import tempfile
 
@@ -11,7 +12,7 @@ import pytest
 
 from reme.memory.file_based.tools.file_io import FileIO
 from reme.memory.file_based.tools.shell import Shell
-from reme.memory.file_based.utils import DEFAULT_MAX_LINES, DEFAULT_MAX_BYTES
+from reme.memory.file_based.utils import DEFAULT_MAX_BYTES
 
 
 # ============ Shell Tests ============
@@ -76,23 +77,6 @@ def test_shell_multiline_output(shell_env):
     assert "line3" in text
 
 
-def test_shell_truncated_output(shell_env):
-    """Test output truncation for large output."""
-    lines_to_generate = DEFAULT_MAX_LINES + 500
-    cmd = f"seq 1 {lines_to_generate}"
-    result = asyncio.run(shell_env["shell"].execute_shell_command(cmd))
-    text = result.content[0].get("text", "")
-
-    # Should contain truncation notice
-    assert "truncated" in text.lower()
-    # Should contain the last line (tail is kept)
-    assert str(lines_to_generate) in text
-    # Verify first numeric line is > 1 (truncated from head)
-    numeric_lines = [tl for tl in text.strip().split("\n") if tl.isdigit()]
-    if numeric_lines:
-        assert int(numeric_lines[0]) > 1
-
-
 def test_shell_timeout(shell_env):
     """Test command timeout handling."""
     result = asyncio.run(
@@ -116,13 +100,17 @@ def fileio_env():
     with open(simple_file, "w", encoding="utf-8") as f:
         f.write("line1\nline2\nline3\nline4\nline5")
 
-    # Create large file (exceeds DEFAULT_MAX_LINES)
+    # Create large file (exceeds DEFAULT_MAX_BYTES)
     large_file = os.path.join(test_dir, "large.txt")
     with open(large_file, "w", encoding="utf-8") as f:
-        for i in range(1, DEFAULT_MAX_LINES + 500):
+        # Each line is ~7-10 bytes ("line N\n"); generate enough to exceed limit.
+        # Line 1 is literally "line 1" so the head-kept assertion can match it.
+        line_count = (DEFAULT_MAX_BYTES // 7) + 1000
+        for i in range(1, line_count + 1):
             f.write(f"line {i}\n")
 
     # Create large bytes file (exceeds DEFAULT_MAX_BYTES)
+    # Lines are 101 bytes each; at DEFAULT_MAX_BYTES the cut lands mid-line → else branch
     large_bytes_file = os.path.join(test_dir, "large_bytes.txt")
     with open(large_bytes_file, "w", encoding="utf-8") as f:
         content = "x" * 100 + "\n"
@@ -130,19 +118,31 @@ def fileio_env():
         for _ in range(lines_needed):
             f.write(content)
 
+    # Single line larger than DEFAULT_MAX_BYTES → newline_count==0 branch in truncate
+    huge_line_file = os.path.join(test_dir, "huge_line.txt")
+    with open(huge_line_file, "w", encoding="utf-8") as f:
+        f.write("A" * (DEFAULT_MAX_BYTES + 1000) + "\nline2\n")
+
+    # Empty file
+    empty_file = os.path.join(test_dir, "empty.txt")
+    with open(empty_file, "w", encoding="utf-8") as f:
+        f.write("")
+
     yield {
         "dir": test_dir,
         "file_io": file_io,
         "simple_file": simple_file,
         "large_file": large_file,
         "large_bytes_file": large_bytes_file,
+        "huge_line_file": huge_line_file,
+        "empty_file": empty_file,
     }
     shutil.rmtree(test_dir, ignore_errors=True)
 
 
 def test_read_file_success(fileio_env):
     """Test successful file reading."""
-    result = asyncio.run(fileio_env["file_io"].read(fileio_env["simple_file"]))
+    result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["simple_file"]))
     text = result.content[0].get("text", "")
     assert "line1" in text
     assert "line5" in text
@@ -150,14 +150,14 @@ def test_read_file_success(fileio_env):
 
 def test_read_file_relative_path(fileio_env):
     """Test reading file with relative path."""
-    result = asyncio.run(fileio_env["file_io"].read("simple.txt"))
+    result = asyncio.run(fileio_env["file_io"].read_file("simple.txt"))
     text = result.content[0].get("text", "")
     assert "line1" in text
 
 
 def test_read_file_not_exists(fileio_env):
     """Test reading non-existent file."""
-    result = asyncio.run(fileio_env["file_io"].read("nonexistent.txt"))
+    result = asyncio.run(fileio_env["file_io"].read_file("nonexistent.txt"))
     text = result.content[0].get("text", "")
     assert "Error" in text
     assert "does not exist" in text
@@ -166,7 +166,7 @@ def test_read_file_not_exists(fileio_env):
 def test_read_file_with_line_range(fileio_env):
     """Test reading specific line range."""
     result = asyncio.run(
-        fileio_env["file_io"].read(fileio_env["simple_file"], start_line=2, end_line=4),
+        fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line=2, end_line=4),
     )
     text = result.content[0].get("text", "")
     assert "line2" in text
@@ -177,7 +177,7 @@ def test_read_file_with_line_range(fileio_env):
 def test_read_file_start_line_exceeds(fileio_env):
     """Test start_line exceeding file length."""
     result = asyncio.run(
-        fileio_env["file_io"].read(fileio_env["simple_file"], start_line=100),
+        fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line=100),
     )
     text = result.content[0].get("text", "")
     assert "Error" in text
@@ -187,15 +187,15 @@ def test_read_file_start_line_exceeds(fileio_env):
 def test_read_file_invalid_range(fileio_env):
     """Test invalid line range (start > end)."""
     result = asyncio.run(
-        fileio_env["file_io"].read(fileio_env["simple_file"], start_line=4, end_line=2),
+        fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line=4, end_line=2),
     )
     text = result.content[0].get("text", "")
     assert "Error" in text
 
 
-def test_read_file_truncated_by_lines(fileio_env):
-    """Test file truncation by line limit."""
-    result = asyncio.run(fileio_env["file_io"].read(fileio_env["large_file"]))
+def test_read_file_truncated(fileio_env):
+    """Test file truncation by byte limit."""
+    result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["large_file"]))
     text = result.content[0].get("text", "")
     assert "line 1" in text  # Head is kept
     assert "continue" in text.lower()
@@ -203,19 +203,140 @@ def test_read_file_truncated_by_lines(fileio_env):
 
 def test_read_file_truncated_by_bytes(fileio_env):
     """Test file truncation by byte limit."""
-    result = asyncio.run(fileio_env["file_io"].read(fileio_env["large_bytes_file"]))
+    result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["large_bytes_file"]))
     text = result.content[0].get("text", "")
-    assert "continue" in text.lower() or "KB" in text
+    assert "continue" in text.lower()
+    assert "KB limit" in text
 
 
 def test_read_directory_error(fileio_env):
     """Test reading a directory returns error."""
-    result = asyncio.run(fileio_env["file_io"].read(fileio_env["dir"]))
+    result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["dir"]))
     text = result.content[0].get("text", "")
     assert "Error" in text
     assert "not a file" in text
 
 
+def test_read_file_single_line_range(fileio_env):
+    """Test reading exactly one line (start_line == end_line)."""
+    result = asyncio.run(
+        fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line=3, end_line=3),
+    )
+    text = result.content[0].get("text", "")
+    assert "line3" in text
+    assert "line2" not in text
+    assert "line4" not in text
+
+
+def test_read_file_only_start_line(fileio_env):
+    """Test reading from start_line to end of file (no end_line)."""
+    result = asyncio.run(
+        fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line=4),
+    )
+    text = result.content[0].get("text", "")
+    assert "line4" in text
+    assert "line5" in text
+    assert "line1" not in text
+    assert "line3" not in text
+
+
+def test_read_file_only_end_line(fileio_env):
+    """Test reading from beginning to end_line (no start_line)."""
+    result = asyncio.run(
+        fileio_env["file_io"].read_file(fileio_env["simple_file"], end_line=2),
+    )
+    text = result.content[0].get("text", "")
+    assert "line1" in text
+    assert "line2" in text
+    assert "line4" not in text
+    assert "line5" not in text
+
+
+def test_read_file_end_line_clamped(fileio_env):
+    """Test end_line beyond total lines is silently clamped to file end."""
+    result = asyncio.run(
+        fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line=1, end_line=999),
+    )
+    text = result.content[0].get("text", "")
+    assert "Error" not in text
+    assert "line1" in text
+    assert "line5" in text
+
+
+def test_read_file_continuation_hint(fileio_env):
+    """Partial range read without truncation shows remaining-lines continuation hint."""
+    # simple.txt has 5 lines; reading 1-3 leaves 2 more
+    result = asyncio.run(
+        fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line=1, end_line=3),
+    )
+    text = result.content[0].get("text", "")
+    assert "more lines" in text
+    assert "start_line=4" in text
+
+
+def test_read_file_truncated_next_line_hint(fileio_env):
+    """Truncated large file provides a valid start_line > 1 to continue."""
+    result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["large_file"]))
+    text = result.content[0].get("text", "")
+    match = re.search(r"start_line=(\d+)", text)
+    assert match is not None, "Expected start_line hint in truncated output"
+    assert int(match.group(1)) > 1
+
+
+def test_read_file_truncated_mid_line_message(fileio_env):
+    """Truncation mid-line reports which line is truncated (else branch)."""
+    # large_bytes_file lines are 101 bytes; truncation lands mid-line
+    result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["large_bytes_file"]))
+    text = result.content[0].get("text", "")
+    assert "is truncated" in text.lower()
+
+
+def test_read_file_huge_single_line(fileio_env):
+    """Single line exceeding byte limit triggers 'partially shown' notice (newline_count==0 branch)."""
+    result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["huge_line_file"]))
+    text = result.content[0].get("text", "")
+    assert "partially shown" in text.lower()
+    assert "start_line=2" in text
+
+
+def test_read_file_invalid_start_line_type(fileio_env):
+    """Non-integer start_line returns a descriptive error."""
+    result = asyncio.run(
+        fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line="abc"),
+    )
+    text = result.content[0].get("text", "")
+    assert "Error" in text
+    assert "start_line" in text
+
+
+def test_read_file_invalid_end_line_type(fileio_env):
+    """Non-integer end_line returns a descriptive error."""
+    result = asyncio.run(
+        fileio_env["file_io"].read_file(fileio_env["simple_file"], end_line="xyz"),
+    )
+    text = result.content[0].get("text", "")
+    assert "Error" in text
+    assert "end_line" in text
+
+
+def test_read_file_start_line_as_string(fileio_env):
+    """Numeric-string start_line/end_line are coerced to int successfully."""
+    result = asyncio.run(
+        fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line="2", end_line="4"),
+    )
+    text = result.content[0].get("text", "")
+    assert "Error" not in text
+    assert "line2" in text
+    assert "line4" in text
+
+
+def test_read_file_empty(fileio_env):
+    """Reading an empty file returns without error."""
+    result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["empty_file"]))
+    text = result.content[0].get("text", "")
+    assert "Error" not in text
+
+
 # ============ FileIO Write Tests ============
 
 
@@ -231,7 +352,7 @@ def write_env():
 def test_write_new_file(write_env):
     """Test writing a new file."""
     file_path = os.path.join(write_env["dir"], "new_file.txt")
-    result = asyncio.run(write_env["file_io"].write(file_path, "test content"))
+    result = asyncio.run(write_env["file_io"].write_file(file_path, "test content"))
     text = result.content[0].get("text", "")
     assert "Wrote" in text
 
@@ -245,7 +366,7 @@ def test_write_overwrite_file(write_env):
     with open(file_path, "w", encoding="utf-8") as f:
         f.write("old content")
 
-    result = asyncio.run(write_env["file_io"].write(file_path, "new content"))
+    result = asyncio.run(write_env["file_io"].write_file(file_path, "new content"))
     text = result.content[0].get("text", "")
     assert "Wrote" in text
 
@@ -255,14 +376,14 @@ def test_write_overwrite_file(write_env):
 
 def test_write_empty_path(write_env):
     """Test writing with empty path."""
-    result = asyncio.run(write_env["file_io"].write("", "content"))
+    result = asyncio.run(write_env["file_io"].write_file("", "content"))
     text = result.content[0].get("text", "")
     assert "Error" in text
 
 
 def test_write_relative_path(write_env):
     """Test writing file with relative path."""
-    result = asyncio.run(write_env["file_io"].write("relative.txt", "relative content"))
+    result = asyncio.run(write_env["file_io"].write_file("relative.txt", "relative content"))
     text = result.content[0].get("text", "")
     assert "Wrote" in text
 
@@ -290,7 +411,7 @@ def edit_env():
 def test_edit_replace_text(edit_env):
     """Test replacing text in file."""
     result = asyncio.run(
-        edit_env["file_io"].edit(edit_env["edit_file"], "Hello", "Hi"),
+        edit_env["file_io"].edit_file(edit_env["edit_file"], "Hello", "Hi"),
     )
     text = result.content[0].get("text", "")
     assert "Successfully" in text
@@ -305,7 +426,7 @@ def test_edit_replace_text(edit_env):
 def test_edit_text_not_found(edit_env):
     """Test editing when text not found."""
     result = asyncio.run(
-        edit_env["file_io"].edit(edit_env["edit_file"], "NotExists", "Replacement"),
+        edit_env["file_io"].edit_file(edit_env["edit_file"], "NotExists", "Replacement"),
     )
     text = result.content[0].get("text", "")
     assert "Error" in text
@@ -315,7 +436,7 @@ def test_edit_text_not_found(edit_env):
 def test_edit_nonexistent_file(edit_env):
     """Test editing non-existent file."""
     result = asyncio.run(
-        edit_env["file_io"].edit("nonexistent.txt", "old", "new"),
+        edit_env["file_io"].edit_file("nonexistent.txt", "old", "new"),
     )
     text = result.content[0].get("text", "")
     assert "Error" in text