release: plyngent 0.4.1

Appending via edit_lineno required the last line to be read, which made
writing to an empty file impossible (line 0 does not exist). An empty
file has nothing to clobber, so the append needs no prior read.

CLAUDE.md

@@ -85,7 +85,7 @@ Async SQLAlchemy + aiosqlite. `MemoryStore`: schema init (+ lightweight SQLite `
 Module-level `@tool` handlers. Call `set_workspace_root()` before use.
 
 - **`workspace`**: path resolve under primary root **or** temporary allowlist roots; path substring denylist; command basename denylist (CLI: timed human allow override, default deny on timeout; **not** skipped by YOLO); clearer deny messages.
-- **`file`**: `read_file` (`with_lineno`, `max_chars` default 32k; results start with a 1-based `L{begin}-{end}` range), `write_file`, `listdir`, `tree` (VCS + default noise dirs + optional `skip_dirs` / denylist walk), `glob_paths`, `grep_files` (regex, skip VCS/binary), `edit_replace` (`max_replaces`, reports remaining matches), `edit_lineno` (1-based range; only lines read via `with_lineno`; read state resets on any file write/edit/copy/move/delete), `copy_path` / `move_path` / `delete_path`, `new_temporary_workspace` (system temp allowlist; cleanup on chat exit; no session rebind), `get_truncated` (resume any truncated result via its `truncate_token=...`).
+- **`file`**: `read_file` (`with_lineno`, `max_chars` default 32k; results start with a 1-based `L{begin}-{end}` range), `write_file`, `listdir`, `tree` (model picks `format=markdown` default / `flat` paths / `decorated`; VCS + default noise dirs + optional `skip_dirs` / denylist walk), `glob_paths`, `grep_files` (regex, skip VCS/binary), `edit_replace` (`max_replaces`, reports remaining matches), `edit_lineno` (1-based range; only lines read via `with_lineno`; read state resets on any file write/edit/copy/move/delete), `copy_path` / `move_path` / `delete_path`, `new_temporary_workspace` (system temp allowlist; cleanup on chat exit; no session rebind), `get_truncated` (resume any truncated result via its `truncate_token=...`).
 - **`process`**: `run_argv` (argv, no shell, timeout, optional stdin/env); `run_argv_batch` (serial steps, `pipe_out` on provider, `mix_stderr`, `stop_on_error`); PTY `open_pty` / `read_pty` / `write_pty` (literal) / `write_pty_keys` (escapes) / `ask_into_pty` (human→PTY only; `secret` no-echo; answer never in tool result) / `close_pty` (POSIX: `pty`+`fork`; Windows: ConPTY via `pywinpty` env marker dep).
 - PTY: backend in `pty_backend.py`; structured status (`alive`/`exit_code`/`data`); `read_pty(..., until=)` (**CSI sanitized** so host TTY is never reset); keys via `write_pty_keys` only (`\xHH`, `ctrl+x`, `key=esc`); secrets via `ask_into_pty` (not `write_pty` data); session limit/idle TTL/output budget; close terminate→kill.
 - CLI limit hooks: interactive confirm to raise tool-loop rounds, PTY session cap, or PTY output budget.
@@ -94,7 +94,7 @@ Module-level `@tool` handlers. Call `set_workspace_root()` before use.
 - **`chat`**: human prompts as tools — `ask_user_line` / `ask_user_choice` / `ask_user_form` (shared `prompting` core); `wait` (line prompt with timeout; Enter disturbs).
 - **`todo`**: LIFO stack of **task groups** (not a queue of tasks) — `todo_push` creates one group of siblings; `todo_pop` removes the whole top group; `todo_update` by id; DFS breakdown push[T1,T2]→push[T1.1…]→pop→push[T2.1…]; stored on session row; open items = unfinished work (`[TODO OPEN WORK]`); all-terminal non-empty = hygiene (`[TODO HYGIENE]`); nag channel via `[agent] todo_nag_strategy` (`developer`/`user` = prose nag; `synthetic_tool` = forged `todo_list` call + **real** `stack.render()` result + ToolCall/Result events; `none`).
 - **`net` / `fetch`**: HTTP GET/POST/PUT/DELETE via niquests (isolated session, not the LLM client). Manual redirects with per-hop SSRF checks (`asyncio` `getaddrinfo`). Private/loopback hosts need instance policy grant (CLI timed confirm; **not** YOLO). Public cleartext HTTP and mutating methods soft-confirm (`YOLO`/`TRUSTABLE`). HTTPS→HTTP redirects blocked unless `allow_http_downgrade`. `user_agent` tool arg (or headers) is never replaced by the default when provided. Caps: body bytes/chars, request body size.
-- **`DEFAULT_TOOLS` / catalog**: file + process + vcs + chat + todo + net (`fetch`) for local surface.
+- **`DEFAULT_TOOLS` / catalog**: file + process + vcs + chat + todo + net (`fetch`) for local surface. `ToolTag.READ_ONLY` marks non-mutating tools (`read_file`, `get_truncated`, `listdir`, `glob_paths`, `grep_files`, `tree`, `todo_list`, `wait`, `ask_user_*`, `vcs_*`, `fetch`); `ToolRegistry.clone(read_only_only=True)` selects just those for safe side turns / planning, and `read_only_context()` (in `tools/context.py`) makes `fetch` accept GET only.
 
 ### Prompting (`prompting.py`)
 
@@ -106,6 +106,7 @@ Click app + readline REPL. Entry: `plyngent` / `python -m plyngent`.
 
 - **`plyngent chat`**: provider/model (flags or interactive; Tab via readline in `prompting`); sessions store `provider_name`/`model` and restore on resume; SQLite via `[database]` (file DB under user data when url unset; explicit `url = ":memory:"` kept + warn); workspace-bound; resume latest for cwd/`--workspace` by default (`--new` / `--session`). One-shot: `-p/--prompt` and non-TTY stdin; exit codes 0/1/2/3; `--yes` (YOLO on), `--stream/--no-stream`, `--quiet`. Root `--log-level`.
 - Slash: Click group in `cli/slash.py` + `awaitlet` for async work; Tab completer from registry + ParamType `shell_complete`. Multiline `"""` … `"""`; `/edit` via `$VISUAL`/`$EDITOR` (blocking only). `/yolo on|off|once` for soft destructive confirms. `/model --persist` / `/models --persist` write model catalog entries into TOML. `/todos` for human show/push/pop/clear of the todo stack.
+- `/btw` side questions: forked session; main transcript/DB untouched. `--tools=read` (default) exposes only `READ_ONLY` tools under `read_only_context()` (fetch=GET) with the main session view; `--tools=no` off; `--tools=full` whole-registry clone with fresh session. Tools off: default falls back to no tools; explicit `read`/`full` errors.
 - Explicit `/resume` or `--session` from another workspace prompts: **keep** / **update** / **abort**.
 - Failed/cancelled turns: user kept; **committed tool rounds kept** (side effects not re-run on `/retry`); only unfinished assistant rolled back; Ctrl+C cancels; TTY confirms off-loop; auto-retry 10s/20s/30s then `/retry`.
 - **`plyngent providers`**, **`config path|edit`**. Config open: `$VISUAL`/`$EDITOR` (wait), else system default (`xdg-open` / `open` / `startfile`, non-blocking). `/edit` stays blocking-only. No providers → optional edit then reload when waited.

README.md

@@ -252,6 +252,7 @@ Type `/help` in the REPL for the live list. Common ones:
 | `/yolo [on\|off\|once]` | Soft destructive confirms: sticky skip, off, or next turn only |
 
 | `/retry` | Re-run incomplete last user turn (after error/cancel) |
+| `/btw [--tools read\|no\|full] [--fresh]` | Side question without changing the main session (read-only tools by default) |
 | `/provider` `/model` | Switch without restarting |
 | `/model --persist` | Save current model id into `plyngent.toml` catalog |
 | `/models` | List config + remote `GET /models` (always re-fetches) |
@@ -270,7 +271,7 @@ User messages are saved immediately. On API error or Ctrl+C, partial assistant/t
 
 ## Tools (when enabled)
 
-Default registry: file ops (including `tree` with default noise-dir skips), `run_argv` / `run_argv_batch` / PTY (POSIX openpty; Windows ConPTY via pywinpty), read-only VCS (git), HTTP `fetch` (GET/POST/PUT/DELETE via niquests; private/loopback hosts need a human policy grant, not YOLO), human prompts (`ask_user_line` / `ask_user_choice` / `ask_user_form`), `wait` (line prompt with timeout; Enter disturbs), `get_truncated` (resume any truncated result via its `truncate_token=...`), and todo stack tools (`todo_list` / `todo_push` / `todo_pop` / `todo_update` / `todo_clear`).
+Default registry: file ops (including `tree` with a markdown bullet-list default, `flat` paths or classic `decorated` on demand; default noise-dir skips), `run_argv` / `run_argv_batch` / PTY (POSIX openpty; Windows ConPTY via pywinpty), read-only VCS (git), HTTP `fetch` (GET/POST/PUT/DELETE via niquests; private/loopback hosts need a human policy grant, not YOLO), human prompts (`ask_user_line` / `ask_user_choice` / `ask_user_form`), `wait` (line prompt with timeout; Enter disturbs), `get_truncated` (resume any truncated result via its `truncate_token=...`), and todo stack tools (`todo_list` / `todo_push` / `todo_pop` / `todo_update` / `todo_clear`).
 
 Safety defaults:
 

doc/plugins.md

@@ -50,6 +50,11 @@ Notes:
 - Optional soft-confirm bits: **`YOLO`** (eligible for YOLO auto-approve),
   **`TRUSTABLE`** (grant once per session after approve). Hard denylists are
   never YOLO-skipped.
+- Optional host-policy bit: **`READ_ONLY`** marks tools that never mutate host
+  state — they become eligible for read-only contexts such as
+  `/btw --tools=read` side turns or planning. Tools that only sometimes mutate
+  (e.g. `fetch` with a non-GET method) may still opt in; the handler must
+  enforce the constraint via the read-only context flag.
 - Docstring becomes the model-facing description; parameter schemas come from
   type hints.
 - Use `register=False` only for unit tests that build a private registry.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "plyngent"
-version = "0.4.0"
+version = "0.4.1"
 description = "LLM chat and agent toolkit"
 authors = [
   { name = "HivertMoZara", email = "worldmozara@163.com" },

src/plyngent/agent/__init__.py

@@ -2,6 +2,7 @@ from .chat import ChatAgent as ChatAgent
 from .chat import get_lineno_read_files as get_lineno_read_files
 from .chat import invalidate_lineno_read as invalidate_lineno_read
 from .chat import lineno_read_lines as lineno_read_lines
+from .chat import lineno_read_mtime as lineno_read_mtime
 from .chat import mark_lineno_read as mark_lineno_read
 from .chat import reset_lineno_tracker as reset_lineno_tracker
 from .chat import was_lineno_read as was_lineno_read

src/plyngent/agent/chat.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from contextvars import ContextVar, Token
+from pathlib import Path
 from typing import TYPE_CHECKING
 
 from msgspec import UNSET
@@ -47,14 +48,18 @@ if TYPE_CHECKING:
 
 
 # Per-conversation tracking of lines read with ``with_lineno=True``, keyed by
-# resolved file path. Set to a fresh ``dict`` at the start of each user turn via
-# :func:`reset_lineno_tracker`; cleared after the assistant finishes.
-# Tool handlers read/write via :func:`get_lineno_read_files` /
-# :func:`lineno_read_lines` / :func:`mark_lineno_read`.
-_lineno_read_files: ContextVar[dict[str, set[int]] | None] = ContextVar("_lineno_read_files", default=None)
+# resolved file path, with the file's mtime at read time (so stale line numbers
+# after external/process mutations are detected). Set to a fresh ``dict`` at the
+# start of each user turn via :func:`reset_lineno_tracker`; cleared after the
+# assistant finishes. Tool handlers read/write via :func:`get_lineno_read_files` /
+# :func:`lineno_read_lines` / :func:`lineno_read_mtime` / :func:`mark_lineno_read`.
+_lineno_read_files: ContextVar[dict[str, tuple[set[int], int]] | None] = ContextVar(
+    "_lineno_read_files",
+    default=None,
+)
 
 
-def reset_lineno_tracker() -> Token[dict[str, set[int]] | None]:
+def reset_lineno_tracker() -> Token[dict[str, tuple[set[int], int]] | None]:
     """Replace the tracker with a fresh empty mapping; return the previous token.
 
     Call at the start of a new user turn. Restore with
@@ -80,17 +85,38 @@ def lineno_read_lines(path: str) -> set[int]:
     val = _lineno_read_files.get()
     if val is None:
         return set()
-    return set(val.get(path, ()))
+    entry = val.get(path)
+    return set(entry[0]) if entry is not None else set()
+
+
+def lineno_read_mtime(path: str) -> int | None:
+    """Return the file mtime (ns) recorded when *path* was read, or None."""
+    val = _lineno_read_files.get()
+    if val is None:
+        return None
+    entry = val.get(path)
+    return entry[1] if entry is not None else None
 
 
 def mark_lineno_read(path: str, lines: set[int]) -> None:
     """Record that *path* was read with line numbers this turn.
 
     ``lines`` are 1-based absolute file line numbers (``read_file`` slice).
+    The file's current mtime is recorded too, so edits can detect that the
+    file changed since the read (external/process mutations included).
     """
     val = _lineno_read_files.get()
-    if val is not None:
-        val.setdefault(path, set()).update(lines)
+    if val is None:
+        return
+    try:
+        mtime_ns = Path(path).stat().st_mtime_ns
+    except OSError:
+        mtime_ns = 0
+    entry = val.get(path)
+    if entry is None:
+        val[path] = (set(lines), mtime_ns)
+    else:
+        val[path] = (entry[0] | lines, mtime_ns)
 
 
 def invalidate_lineno_read(path: str) -> None:
@@ -569,7 +595,8 @@ class ChatAgent:
         - Tools default **off**. ``tools=True`` clones this agent's registry with
           a **fresh session** bag (unless *session_state* is passed) and the
           given *instance_state* (CLI typically shares the host instance for
-          workspace identity).
+          workspace identity). A caller-built :class:`ToolRegistry` (e.g. a
+          ``clone(read_only_only=True)`` subset) is used as-is.
         """
         text = user_text.strip()
         if not text:

src/plyngent/agent/tools.py

@@ -42,6 +42,7 @@ class ToolTag(Flag):
     YOLO = auto()  # soft-confirm: eligible for YOLO auto-approve
     INSTANCE_STATE = auto()  # needs instance-scoped state
     SESSION_STATE = auto()  # needs session-scoped state
+    READ_ONLY = auto()  # never mutates host state; safe for read-only contexts
 
 
 class ToolDefinition:
@@ -266,14 +267,22 @@ class ToolRegistry:
         session_state: object | None = None,
         yolo: bool | None = None,
         auto_bind_state: bool | None = None,
+        read_only_only: bool = False,
     ) -> ToolRegistry:
         """New registry with the same tools/hooks and rebound host state.
 
         *instance_state* / *session_state* are applied as given (use the parent
         registry's handles only when the caller passes them through).
+
+        With *read_only_only*, only :attr:`ToolTag.READ_ONLY` tools are kept
+        (e.g. for safe side turns / planning); hooks and policy flags are
+        preserved.
         """
+        definitions = self.definitions()
+        if read_only_only:
+            definitions = [d for d in definitions if d.tags & ToolTag.READ_ONLY]
         return ToolRegistry(
-            self.definitions(),
+            definitions,
             danger=self._danger,
             on_confirm=self._on_confirm,
             yolo=self._yolo if yolo is None else yolo,

src/plyngent/cli/display.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 import contextlib
 import json
 import os
+import re
+import shlex
 import sys
 from contextvars import ContextVar
 from typing import TYPE_CHECKING, Literal, cast
@@ -22,7 +24,7 @@ from plyngent.agent import (
 from plyngent.lmproto.openai_compatible.model import AssistantFunctionToolCall
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterator
+    from collections.abc import AsyncIterator, Callable
 
     from plyngent.agent import AgentEvent
 
@@ -31,7 +33,29 @@ _TOOL_ARGS_PREVIEW = 80
 
 # Tool calls that render as a single pretty summary line instead of
 # ``[tool]`` / ``[tool result]``.
-_PRETTY_TOOLS = frozenset({"read_file", "todo_push", "todo_update"})
+_PRETTY_TOOLS = frozenset(
+    {
+        "read_file",
+        "todo_push",
+        "todo_update",
+        "tree",
+        "listdir",
+        "glob_paths",
+        "grep_files",
+        "run_argv",
+        "run_argv_batch",
+        "fetch",
+        "edit_replace",
+        "edit_lineno",
+        "write_file",
+        "copy_path",
+        "move_path",
+        "delete_path",
+        "vcs_status",
+        "vcs_log",
+        "wait",
+    }
+)
 
 # Process/session display flags (set from ReplState / slash).
 _verbose_tool_results: ContextVar[bool] = ContextVar("verbose_tool_results", default=False)
@@ -74,47 +98,327 @@ def _preview(text: str, limit: int) -> str:
     return text[:limit] + "…"
 
 
-def _json_str_arg(args_json: str, key: str) -> str | None:
-    """Extract a string argument from a tool-call arguments JSON blob."""
+def _preview_result(content: str, limit: int) -> str:
+    """Non-verbose tool-result preview: first line + a ``(…N more lines)`` tail."""
+    lines = content.splitlines()
+    if not lines:
+        return "(no output)"
+    first = _preview(lines[0], limit)
+    more = len(lines) - 1
+    if more == 0:
+        return first
+    unit = "line" if more == 1 else "lines"
+    return f"{first} (…{more} more {unit})"
+
+
+def _json_arg(args_json: str, key: str) -> object | None:
+    """Extract a single argument value (any type) from a tool-call args JSON blob."""
     try:
         raw = json.loads(args_json)
     except ValueError:
         return None
     if isinstance(raw, dict):
-        value = cast("dict[str, object]", raw).get(key)
-        if isinstance(value, str):
-            return value
+        return cast("dict[str, object]", raw).get(key)
     return None
 
 
+def _json_str_arg(args_json: str, key: str) -> str | None:
+    value = _json_arg(args_json, key)
+    return value if isinstance(value, str) else None
+
+
+def _json_list_arg(args_json: str, key: str) -> list[str] | None:
+    value = _json_arg(args_json, key)
+    if isinstance(value, list):
+        raw_items = cast("list[object]", value)
+        items = [item for item in raw_items if isinstance(item, str)]
+        if len(items) == len(raw_items):
+            return items
+    return None
+
+
+def _pretty_line(prefix: str, *segments: tuple[str, str | None]) -> str:
+    """Style a pretty tool summary: yellow ``* Verb`` prefix + colored detail segments.
+
+    ``segments`` are ``(text, fg)`` pairs; ``fg=None`` leaves a segment uncolored
+    and ``"dim"`` renders it dim. click.style emits ANSI codes only on a TTY,
+    so non-TTY output stays plain.
+    """
+    parts = [click.style(prefix, fg="yellow")]
+    for text, fg in segments:
+        if fg == "dim":
+            parts.append(click.style(text, dim=True))
+        else:
+            parts.append(click.style(text, fg=fg))
+    return "".join(parts)
+
+
 def _read_file_pretty(args_json: str, result: str) -> str:
-    """One-line summary for a ``read_file`` call: ``* Read 'path' L1-4 (done)``."""
+    """One-line summary for a ``read_file`` call: ``* Read 'path' (done)``."""
     path = _json_str_arg(args_json, "path") or "?"
     if result.startswith("error: file not found"):
-        return f"* Read '{path}' (file not found)"
+        return _pretty_line(f"* Read '{path}' ", ("(file not found)", "red"))
     if result.startswith("error: not a file"):
-        return f"* Read '{path}' (not a file)"
+        return _pretty_line(f"* Read '{path}' ", ("(not a file)", "red"))
     if result.startswith("error:"):
-        return f"* Read '{path}' (error)"
-    first = result.splitlines()[0] if result.splitlines() else ""
-    if first.startswith("L") and "-" in first:
-        return f"* Read '{path}' {first} (done)"
-    return f"* Read '{path}' (done)"
+        return _pretty_line(f"* Read '{path}' ", ("(error)", "red"))
+    return _pretty_line(f"* Read '{path}' ", ("(done)", "green"))
 
 
 def _todo_pretty(name: str, result: str) -> str:
     """Summary for todo push/update: header + the rendered todo stack."""
     label = "Todo Push" if name == "todo_push" else "Todo Update"
-    return f"* {label}:\n{result}"
+    return _pretty_line(f"* {label}:\n{result}")
+
+
+def _tree_line_stats(fmt: str, line: str) -> tuple[str, int] | None:
+    """Classify one ``tree`` result line as ``(kind, depth)`` or None (not an entry).
+
+    ``kind`` is ``"dir"`` or ``"file"``; depth is 0 for ``flat`` (no hierarchy).
+    """
+    stripped = line.strip()
+    if not stripped:
+        return None
+    kind: str | None = None
+    depth = 0
+    if fmt == "flat":
+        if not stripped.startswith("…"):
+            kind = "dir" if stripped.endswith("/") else "file"
+    elif fmt == "markdown":
+        if stripped.startswith("- ") and not stripped.startswith(("- …", "- error")):
+            kind = "dir" if stripped[2:].endswith("/") else "file"
+            depth = (len(line) - len(line.lstrip(" "))) // 2 + 1
+    elif "… (" not in stripped:
+        branch = "├── " if "├── " in line else ("└── " if "└── " in line else None)
+        if branch is not None:
+            name = line[line.index(branch) + len(branch) :].rstrip()
+            kind = "dir" if name.endswith("/") else "file"
+            depth = line.index(branch) // 4 + 1
+    if kind is None:
+        return None
+    return kind, depth
+
+
+def _tree_pretty(args_json: str, result: str) -> str:
+    """One-line summary for a ``tree`` call: ``* Tree 'src' (3 dirs, 24 files, depth 3)``.
+
+    Flat trees have no depth, so the depth part is omitted for ``format=flat``.
+    """
+    path = _json_str_arg(args_json, "path") or "."
+    fmt = _json_str_arg(args_json, "format") or "markdown"
+    if result.startswith("error:"):
+        return f"* Tree '{path}' (error)"
+    dirs = 0
+    files = 0
+    depth = 0
+    for line in result.splitlines():
+        stats = _tree_line_stats(fmt, line)
+        if stats is None:
+            continue
+        kind, line_depth = stats
+        if kind == "dir":
+            dirs += 1
+        else:
+            files += 1
+        depth = max(depth, line_depth)
+    if depth:
+        return _pretty_line(f"* Tree '{path}' ", (f"({dirs} dirs, {files} files, depth {depth})", None))
+    return _pretty_line(f"* Tree '{path}' ", (f"({dirs} dirs, {files} files)", None))
+
+
+def _listdir_pretty(args_json: str, result: str) -> str:
+    """One-line summary for a ``listdir`` call: ``* List 'path' (3 dirs, 5 files)``."""
+    path = _json_str_arg(args_json, "path") or "."
+    if result.startswith("error:"):
+        return _pretty_line(f"* List '{path}' ", (f"({result})", "red"))
+    if result == "(empty)":
+        return _pretty_line(f"* List '{path}' ", ("(empty)", "dim"))
+    dirs = 0
+    files = 0
+    for line in result.splitlines():
+        if line.startswith("dir\t"):
+            dirs += 1
+        elif line.startswith("file\t"):
+            files += 1
+    return _pretty_line(f"* List '{path}' ", (f"({dirs} dirs, {files} files)", None))
+
+
+def _glob_pretty(args_json: str, result: str) -> str:
+    """One-line summary for a ``glob_paths`` call: ``* Glob '**/*.py' in '.' (12 paths)``."""
+    pattern = _json_str_arg(args_json, "pattern") or "?"
+    path = _json_str_arg(args_json, "path") or "."
+    if result.startswith("error:"):
+        return _pretty_line(f"* Glob '{pattern}' in '{path}' ", (f"({result})", "red"))
+    if result == "(no matches)":
+        return _pretty_line(f"* Glob '{pattern}' in '{path}' ", ("(no matches)", "dim"))
+    count = sum(1 for line in result.splitlines() if line and not line.startswith("...[truncated"))
+    unit = "path" if count == 1 else "paths"
+    return _pretty_line(f"* Glob '{pattern}' in '{path}' ", (f"({count} {unit})", None))
+
+
+def _grep_pretty(args_json: str, result: str) -> str:
+    """One-line summary for a ``grep_files`` call: ``* Grep 'pat' in '.' (12 matches in 3 files)``."""
+    pattern = _json_str_arg(args_json, "pattern") or "?"
+    path = _json_str_arg(args_json, "path") or "."
+    if result.startswith("error:"):
+        return _pretty_line(f"* Grep '{pattern}' in '{path}' ", (f"({result})", "red"))
+    if result == "(no matches)":
+        return _pretty_line(f"* Grep '{pattern}' in '{path}' ", ("(no matches)", "dim"))
+    matches = [line for line in result.splitlines() if line and not line.startswith("...[truncated")]
+    files = len({line.split(":", 1)[0] for line in matches})
+    match_unit = "match" if len(matches) == 1 else "matches"
+    file_unit = "file" if files == 1 else "files"
+    return _pretty_line(
+        f"* Grep '{pattern}' in '{path}' ",
+        (f"({len(matches)} {match_unit} in {files} {file_unit})", None),
+    )
+
+
+def _run_argv_pretty(args_json: str, result: str) -> str:
+    """One-line summary for a ``run_argv`` call: ``* Run $ git status --short (exit code 0)``."""
+    argv = _json_list_arg(args_json, "argv")
+    cmd = shlex.join(argv) if argv else "?"
+    if result.startswith("error:"):
+        return _pretty_line(f"* Run $ {cmd} ", (f"({result})", "red"))
+    fields: dict[str, str] = {}
+    for line in result.splitlines():
+        if line.startswith("--- "):
+            break
+        if "=" in line:
+            key, _, value = line.partition("=")
+            fields[key] = value
+    if fields.get("timed_out") == "true":
+        return _pretty_line(f"* Run $ {cmd} ", ("(timed out)", "red"))
+    code = fields.get("exit_code", "")
+    fg = "green" if code == "0" else "red"
+    return _pretty_line(f"* Run $ {cmd} ", (f"(exit code {code or 'killed'})", fg))
+
+
+def _run_argv_batch_pretty(_args_json: str, result: str) -> str:
+    """One-line summary for a ``run_argv_batch`` call: ``* Run batch (done)``."""
+    if result.startswith("error:"):
+        return _pretty_line("* Run batch ", (f"({result})", "red"))
+    lines = result.splitlines()
+    head = lines[0] if lines else ""
+    stopped_early = False
+    for part in head.split():
+        if part == "stopped_early=true":
+            stopped_early = True
+    if stopped_early:
+        return _pretty_line("* Run batch ", ("(stopped early)", "yellow"))
+    return _pretty_line("* Run batch ", ("(done)", "green"))
+
+
+def _http_status_fg(status: str) -> str | None:
+    """Foreground color by HTTP status class (2xx green, 3xx yellow, 4xx/5xx red)."""
+    if status and status[0] in {"2", "3", "4", "5"}:
+        return {"2": "green", "3": "yellow", "4": "red", "5": "red"}[status[0]]
+    return None
+
+
+def _fetch_pretty(args_json: str, result: str) -> str:
+    """One-line summary for a ``fetch`` call: ``* Fetch GET https://… (200)``."""
+    method = _json_str_arg(args_json, "method") or "GET"
+    url = _json_str_arg(args_json, "url") or "?"
+    if result.startswith("error:"):
+        return _pretty_line(f"* Fetch {method} {url} ", (f"({result})", "red"))
+    status = ""
+    for line in result.splitlines():
+        if line.startswith("--- "):
+            break
+        if line.startswith("status="):
+            status = line.partition("=")[2]
+            break
+    return _pretty_line(f"* Fetch {method} {url} ", (f"({status or 'done'})", _http_status_fg(status)))
+
+
+_MUTATOR_VERBS: dict[str, str] = {
+    "edit_replace": "Edit",
+    "edit_lineno": "Edit",
+    "write_file": "Write",
+    "copy_path": "Copy",
+    "move_path": "Move",
+    "delete_path": "Delete",
+}
+
+
+def _mutator_pretty(name: str, args_json: str, result: str) -> str:
+    """One-line summary for a file mutation: ``* Wrote 'path' (120 chars)`` / ``* Edit 'path' (done)``."""
+    verb = _MUTATOR_VERBS[name]
+    if name in {"copy_path", "move_path"}:
+        src = _json_str_arg(args_json, "src") or "?"
+        dst = _json_str_arg(args_json, "dst") or "?"
+        target = f"'{src}' → '{dst}'"
+    else:
+        path = _json_str_arg(args_json, "path") or "?"
+        target = f"'{path}'"
+    if result.startswith("error:"):
+        return _pretty_line(f"* {verb} {target} ", (f"({result})", "red"))
+    if name == "write_file":
+        # write_file keeps its brief detail: ``wrote 120 characters to path``.
+        match = re.match(r"wrote (\d+) characters to (.+)$", result)
+        if match:
+            chars, path = match.groups()
+            return _pretty_line(f"* Write '{path}' ", (f"({chars} chars)", None))
+    return _pretty_line(f"* {verb} {target} ", ("(done)", "green"))
+
+
+def _vcs_status_pretty(_args_json: str, result: str) -> str:
+    """One-line summary for ``vcs_status``: ``* VCS Status (done)``."""
+    if result.startswith("error:"):
+        return _pretty_line("* VCS Status ", (f"({result})", "red"))
+    return _pretty_line("* VCS Status ", ("(done)", "green"))
+
+
+def _vcs_log_pretty(_args_json: str, result: str) -> str:
+    """One-line summary for ``vcs_log``: ``* VCS Log (5 commits)``."""
+    if result.startswith("error:"):
+        return _pretty_line("* VCS Log ", (f"({result})", "red"))
+    if result == "(no commits)":
+        return _pretty_line("* VCS Log ", ("(no commits)", "dim"))
+    commits = [line for line in result.splitlines() if line.strip()]
+    unit = "commit" if len(commits) == 1 else "commits"
+    return _pretty_line("* VCS Log ", (f"({len(commits)} {unit})", None))
+
+
+def _wait_pretty(_args_json: str, result: str) -> str:
+    """One-line summary for a ``wait`` call: ``* Wait (5s)`` / ``* Wait (disturbed)``."""
+    if result.startswith("error:"):
+        return _pretty_line("* Wait ", (f"({result})", "red"))
+    if result.startswith("waited "):
+        return _pretty_line("* Wait ", (f"({result.removeprefix('waited ')})", "green"))
+    return _pretty_line("* Wait ", ("(disturbed)", "yellow"))
+
+
+_PRETTY_BUILDERS: dict[str, Callable[[str, str], str]] = {
+    "read_file": _read_file_pretty,
+    "tree": _tree_pretty,
+    "todo_push": lambda _args, result: _todo_pretty("todo_push", result),
+    "todo_update": lambda _args, result: _todo_pretty("todo_update", result),
+    "listdir": _listdir_pretty,
+    "glob_paths": _glob_pretty,
+    "grep_files": _grep_pretty,
+    "run_argv": _run_argv_pretty,
+    "run_argv_batch": _run_argv_batch_pretty,
+    "fetch": _fetch_pretty,
+    "edit_replace": lambda args, result: _mutator_pretty("edit_replace", args, result),
+    "edit_lineno": lambda args, result: _mutator_pretty("edit_lineno", args, result),
+    "write_file": lambda args, result: _mutator_pretty("write_file", args, result),
+    "copy_path": lambda args, result: _mutator_pretty("copy_path", args, result),
+    "move_path": lambda args, result: _mutator_pretty("move_path", args, result),
+    "delete_path": lambda args, result: _mutator_pretty("delete_path", args, result),
+    "vcs_status": _vcs_status_pretty,
+    "vcs_log": _vcs_log_pretty,
+    "wait": _wait_pretty,
+}
 
 
 def _pretty_tool_result(name: str, args_json: str, result: str) -> str | None:
     """Pretty summary for a known tool call + result; None keeps the old style."""
-    if name == "read_file":
-        return _read_file_pretty(args_json, result)
-    if name in {"todo_push", "todo_update"}:
-        return _todo_pretty(name, result)
-    return None
+    builder = _PRETTY_BUILDERS.get(name)
+    if builder is None:
+        return None
+    return builder(args_json, result)
 
 
 def _echo_stream(text: str) -> None:
@@ -258,14 +562,13 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
             content = event.message.content
             name, args, pretty_tool = pending_tools.pop(0) if pending_tools else ("", "", False)
             pretty_line = _pretty_tool_result(name, args, content) if pretty_tool else None
-            if pretty_line is not None:
-                click.secho(f"\n{pretty_line}", fg="yellow")
+            if pretty_line is not None and not show_full:
+                click.echo(f"\n{pretty_line}")
             elif show_full:
                 click.secho(f"[tool ok]\n{content}", fg="magenta")
             else:
-                preview = _preview(content, _TOOL_RESULT_PREVIEW)
-                one_line = preview.replace("\n", " ")
-                click.secho(f"[tool ok] {one_line}", fg="magenta")
+                preview = _preview_result(content, _TOOL_RESULT_PREVIEW)
+                click.secho(f"[tool ok] {preview}", fg="magenta")
         elif isinstance(event, ErrorEvent):
             flush_assistant()
             suffix = ""

src/plyngent/cli/slash.py

@@ -24,6 +24,7 @@ from plyngent.runtime import ProviderNotSupportedError
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Sequence
 
+    from plyngent.agent import ToolRegistry
     from plyngent.cli.state import ReplState, YoloMode
     from plyngent.lmproto.openai_compatible.model import AnyChatMessage
 
@@ -83,7 +84,8 @@ HELP_FOOTER = (
     "works after resume, not only via readline history). Auto-retry: 10s/20s/30s.\n"
     "\n"
     "Side questions: /btw … answers without changing the main session\n"
-    "(optional --tools / --fresh). Tab completes slash commands and arguments.\n"
+    "(--tools=read by default: read-only tools; --tools no/full opt-in; --fresh).\n"
+    "Tab completes slash commands and arguments.\n"
     "Use --session ID or /resume to continue a prior chat after restart.\n"
     "\n"
     'Multiline: start a message with """ then end a later line with """.\n'
@@ -928,11 +930,12 @@ def _format_tool_tags(tags: object) -> str:
 @slash.command("btw")
 @click.argument("question", nargs=-1, required=False)
 @click.option(
-    "--tools/--no-tools",
-    "with_tools",
-    default=False,
+    "--tools",
+    "tools_mode",
+    type=click.Choice(["no", "read", "full"]),
+    default="read",
     show_default=True,
-    help="Allow tools on the side turn (shared workspace; forked session grants/todo).",
+    help=("Tool access on the side turn: read = read-only tools only (default), no = no tools, full = all tools."),
 )
 @click.option(
     "--fresh",
@@ -945,48 +948,73 @@ def btw_cmd(
     state: ReplState,
     question: tuple[str, ...],
     *,
-    with_tools: bool,
+    tools_mode: str,
     fresh: bool,
 ) -> None:
     """Ask a side question without changing the main session transcript or DB.
 
     Uses the current model and (unless ``--fresh``) a copy of main history.
-    Tools are **off** by default; ``--tools`` clones the main tool registry with
-    a fresh session bag and the shared instance workspace.
+    Tools default to **read-only** (``--tools=read``): file/vcs/ask/fetch-GET
+    reads with no state mutation. ``--tools=no`` turns tools off and
+    ``--tools=full`` clones the whole main registry with a fresh session bag
+    and the shared instance workspace.
     """
     text = " ".join(question).strip()
     if not text:
-        click.echo("usage: /btw [--tools] [--fresh] <question>")
+        click.echo("usage: /btw [--tools read|no|full] [--fresh] <question>")
         return
     if state.agent.pending_retry_text is not None:
         click.echo("error: finish or /retry the main turn before /btw")
         return
-    if with_tools and (not state.tools_enabled or state.agent.tools is None):
-        click.echo("error: --tools requires tools on (/tools on)")
-        return
+
+    tools_on = state.tools_enabled and state.agent.tools is not None
+    explicit = click.get_current_context().get_parameter_source("tools_mode") is not click.core.ParameterSource.DEFAULT
+    if tools_mode != "no" and not tools_on:
+        # Explicitly requesting tools while they are off is an error; the
+        # default (read) silently falls back to no tools so bare /btw still
+        # works under /tools off.
+        if tools_mode == "full" or explicit:
+            click.echo("error: --tools requires tools on (/tools on)")
+            return
+        tools_mode = "no"
 
     from plyngent.cli.display import render_events
     from plyngent.cli.retry import run_cancellable
-    from plyngent.tools.context import SessionState
+    from plyngent.tools.context import SessionState, read_only_context
 
     click.secho("btw: ", fg="magenta", nl=False)
     click.echo(text)
-    tools_arg: bool = with_tools
-    aside_session = SessionState() if with_tools else None
-    aside_instance = state.instance_state if with_tools else None
+    if tools_mode == "no":
+        tools_arg: ToolRegistry | bool = False
+        aside_session = None
+        aside_instance = None
+    elif tools_mode == "full":
+        tools_arg = True
+        aside_session = SessionState()
+        aside_instance = state.instance_state
+    else:  # read
+        assert state.agent.tools is not None
+        # Read-only clone: keep the main session view so read-only session
+        # reads (e.g. todo_list) see the real state (they cannot mutate it).
+        tools_arg = state.agent.tools.clone(read_only_only=True)
+        aside_session = state.session_state
+        aside_instance = state.instance_state
 
     async def _run() -> None:
-        await run_cancellable(
-            render_events(
-                state.agent.run_aside(
-                    text,
-                    include_history=not fresh,
-                    tools=tools_arg,
-                    instance_state=aside_instance,
-                    session_state=aside_session,
-                )
+        side = render_events(
+            state.agent.run_aside(
+                text,
+                include_history=not fresh,
+                tools=tools_arg,
+                instance_state=aside_instance,
+                session_state=aside_session,
             )
         )
+        if tools_mode == "read":
+            with read_only_context():
+                await run_cancellable(side)
+        else:
+            await run_cancellable(side)
 
     try:
         _await(_run())

src/plyngent/tools/chat/ask.py

@@ -4,7 +4,7 @@ from plyngent.agent import ToolTag, tool
 from plyngent.prompting import NonInteractiveError, ask_async
 
 
-@tool(name="ask_user_line", tags=ToolTag.LOCAL)
+@tool(name="ask_user_line", tags=ToolTag.LOCAL | ToolTag.READ_ONLY)
 async def ask_user(question: str, default: str = "") -> str:
     """Ask the human a free-form one-line question and return their answer.
 

src/plyngent/tools/chat/choose.py

@@ -36,7 +36,7 @@ def parse_options(options: list[object]) -> list[ChoiceOption]:
     return out
 
 
-@tool(name="ask_user_choice", tags=ToolTag.LOCAL)
+@tool(name="ask_user_choice", tags=ToolTag.LOCAL | ToolTag.READ_ONLY)
 async def choose_user(
     question: str,
     options: list[str] | list[dict[str, object]],

src/plyngent/tools/chat/form.py

@@ -46,7 +46,7 @@ def parse_fields(fields: list[dict[str, object]]) -> list[FormField]:
     return out
 
 
-@tool(name="ask_user_form", tags=ToolTag.LOCAL)
+@tool(name="ask_user_form", tags=ToolTag.LOCAL | ToolTag.READ_ONLY)
 async def form_user(
     title: str,
     fields: list[dict[str, object]],

src/plyngent/tools/chat/wait.py

@@ -8,7 +8,7 @@ from plyngent.prompting import get_prompt_backend, read_line_with_timeout, run_p
 WAIT_PROMPT = "Waiting for {duration}s. Press Enter to disturb. Feel free to ship any reason before pressing Enter. "
 
 
-@tool(name="wait", tags=ToolTag.LOCAL)
+@tool(name="wait", tags=ToolTag.LOCAL | ToolTag.READ_ONLY)
 async def wait(duration: int, *, reason: str | None = None) -> str:
     """Wait ``duration`` seconds before continuing.
 

src/plyngent/tools/context.py

@@ -137,6 +137,29 @@ def bind_session(state: SessionState | None) -> Generator[SessionState | None]:
         _session.reset(token)
 
 
+_read_only: ContextVar[bool] = ContextVar("plyngent_read_only_context", default=False)
+
+
+@contextmanager
+def read_only_context() -> Generator[None]:
+    """Mark the current context as read-only for tool execution.
+
+    Tools that may mutate depending on arguments (e.g. ``fetch`` with a
+    non-GET method) check :func:`is_read_only_context` and refuse to mutate
+    while it is set.
+    """
+    token = _read_only.set(True)
+    try:
+        yield
+    finally:
+        _read_only.reset(token)
+
+
+def is_read_only_context() -> bool:
+    """Whether tools currently run inside a read-only context."""
+    return _read_only.get()
+
+
 @contextmanager
 def bind_tool_context(
     *,

src/plyngent/tools/file/edit_lineno.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from plyngent.agent import ToolTag, invalidate_lineno_read, lineno_read_lines, tool
+from plyngent.agent import ToolTag, lineno_read_lines, lineno_read_mtime, tool
 from plyngent.tools.workspace import resolve_path
 
 if TYPE_CHECKING:
@@ -60,10 +60,11 @@ def _unread_lines_in_range(
     """Lines in ``start_line..end_line`` that were not read with line numbers.
 
     Appending (``start_line == n + 1``) requires the last line to be read, so
-    line numbers stay truthful; the append point itself is not a line.
+    line numbers stay truthful; the append point itself is not a line. An empty
+    file (``n == 0``) has nothing to clobber, so appending needs no read.
     """
     if start_line == n + 1:
-        return [] if n in read_lines else [n]
+        return [] if n == 0 or n in read_lines else [n]
     return sorted(set(range(start_line, end_line + 1)) - read_lines)
 
 
@@ -77,6 +78,14 @@ def _append_after(target: Path, path: str, text: str, lines: list[str], new_cont
     return f"appended content after line {n} in {path}"
 
 
+def _excerpt(text: str, limit: int = 40) -> str:
+    """Collapse newlines and truncate for a one-line message excerpt."""
+    one = text.replace("\r\n", " ").replace("\n", " ")
+    if len(one) > limit:
+        return one[:limit] + "…"
+    return one
+
+
 def _replace_range(
     target: Path,
     path: str,
@@ -92,7 +101,11 @@ def _replace_range(
     new_lines = lines[: start_line - 1] + replacement + lines[end:]
     _ = target.write_text("".join(new_lines), encoding="utf-8")
     removed = end - start_line + 1
-    return f"replaced lines {start_line}-{end} ({removed} lines) with {len(replacement)} lines in {path}"
+    old_first = _excerpt(lines[start_line - 1].rstrip("\r\n"))
+    return (
+        f"replaced lines {start_line}-{end} ({removed} lines; first: {old_first!r}) "
+        f"with {len(replacement)} lines in {path}"
+    )
 
 
 @tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO)
@@ -111,18 +124,37 @@ async def edit_lineno(path: str, start_line: int, end_line: int, new_content: st
     if err is not None:
         return err
 
-    unread = _unread_lines_in_range(lineno_read_lines(str(target)), start_line, end_line, len(lines))
+    read_lines = lineno_read_lines(str(target))
+    unread = _unread_lines_in_range(read_lines, start_line, end_line, len(lines))
     if unread:
         preview = unread[:_UNREAD_PREVIEW]
         head = ", ".join(str(i) for i in preview)
         if len(unread) > _UNREAD_PREVIEW:
             head += f", … ({len(unread)} total)"
-        return f"error: lines {head} not read; call read_file(path, with_lineno=true) first to see current line numbers"
+        window = ""
+        if read_lines:
+            window = f" (this turn read lines {min(read_lines)}-{max(read_lines)})"
+        return (
+            f"error: lines {head} not read{window}; call read_file(path, with_lineno=true) "
+            "first — read with a small margin around the lines you plan to edit"
+        )
+
+    recorded_mtime = lineno_read_mtime(str(target))
+    if recorded_mtime is not None:
+        try:
+            current_mtime = target.stat().st_mtime_ns
+        except OSError:
+            current_mtime = None
+        if current_mtime is not None and current_mtime != recorded_mtime:
+            return (
+                f"error: {path} changed since it was read (line numbers may be stale); "
+                "re-read with read_file(path, with_lineno=true)"
+            )
 
     if start_line == len(lines) + 1:
         result = _append_after(target, path, text, lines, new_content)
     else:
         result = _replace_range(target, path, lines, start_line, end_line, new_content)
-    # Line numbers are stale after a write; force a re-read before next edit.
-    invalidate_lineno_read(str(target))
+    # Line numbers are stale after a write; the fresh mtime makes the next
+    # edit_lineno reject until the file is re-read.
     return result

src/plyngent/tools/file/edit_replace.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from plyngent.agent import ToolTag, invalidate_lineno_read, tool
+from plyngent.agent import ToolTag, tool
 from plyngent.tools.workspace import resolve_path
 
 
@@ -18,15 +18,31 @@ def _count_non_overlapping(text: str, needle: str) -> int:
         start = index + len(needle)
 
 
-def _success_message(path: str, *, replaced: int, found: int) -> str:
+def _excerpt(text: str, limit: int = 40) -> str:
+    """Collapse newlines and truncate for a one-line message excerpt."""
+    one = text.replace("\r\n", " ").replace("\n", " ")
+    if len(one) > limit:
+        return one[:limit] + "…"
+    return one
+
+
+def _success_message(
+    path: str,
+    *,
+    replaced: int,
+    found: int,
+    old_string: str,
+    new_string: str,
+) -> str:
     remaining = found - replaced
     unit = "occurrence" if replaced == 1 else "occurrences"
+    change = f" ('{_excerpt(old_string)}' → '{_excerpt(new_string)}')"
     if remaining == 0:
         if found == 1:
-            return f"replaced 1 occurrence in {path}"
-        return f"replaced {replaced} {unit} in {path} (all {found} matches)"
+            return f"replaced 1 occurrence in {path}{change}"
+        return f"replaced {replaced} {unit} in {path}{change} (all {found} matches)"
     return (
-        f"replaced {replaced} {unit} in {path} "
+        f"replaced {replaced} {unit} in {path}{change} "
         f"({replaced} of {found} matches; {remaining} remain — "
         f"raise max_replaces or use a more specific old_string)"
     )
@@ -54,6 +70,12 @@ async def edit_replace(path: str, old_string: str, new_string: str, max_replaces
     n = min(max_replaces, found)
     updated = text.replace(old_string, new_string, n)
     _ = target.write_text(updated, encoding="utf-8")
-    # Any lineno read state is stale after a write; force a re-read.
-    invalidate_lineno_read(str(target))
-    return _success_message(path, replaced=n, found=found)
+    # The write changes the file's mtime, so edit_lineno's freshness check
+    # rejects stale line numbers until the model re-reads with_lineno.
+    return _success_message(
+        path,
+        replaced=n,
+        found=found,
+        old_string=old_string,
+        new_string=new_string,
+    )

src/plyngent/tools/file/fs_ops.py

@@ -68,10 +68,13 @@ def _copy_or_move_validated(
             else:
                 _ = shutil.copy2(source, dest, follow_symlinks=False)
                 label = _kind(source)
-            # A copy does not change the source, but the destination content is new.
+            # A copy does not change the source, but the destination content is
+            # new. copy2/copytree preserve the source mtime, so the freshness
+            # check alone could miss the change — invalidate explicitly.
             invalidate_lineno_read(str(dest))
             return f"copied {label} {src} -> {dest.relative_to(root)}"
         moved = Path(shutil.move(str(source), str(dest))).resolve()
+        # shutil.move preserves mtime too; drop read state for both paths.
         invalidate_lineno_read(str(source))
         invalidate_lineno_read(str(moved))
         return f"moved {_kind(moved)} {src} -> {moved.relative_to(root)}"

src/plyngent/tools/file/get_truncated.py

@@ -6,26 +6,68 @@ from typing import cast
 
 from plyngent.agent import ToolTag, tool
 from plyngent.agent.budget import DEFAULT_TOOL_RESULT_MAX_CHARS
-from plyngent.tools.file.read import line_range_label, read_raw_text
+from plyngent.tools.file.read import line_range_label, read_file, read_raw_text
 from plyngent.tools.net.fetch import fetch
 from plyngent.tools.truncate_token import (
+    TruncateToken,
     decode_truncate_token,
     get_remainder,
     truncate_with_token,
 )
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
+def _memory_chunk(parsed: TruncateToken, max_chars: int) -> str:
+    """Serve one chunk from the in-memory remainder store (or an error)."""
+    remainder = get_remainder(parsed.location)
+    if remainder is None:
+        return "error: truncate token expired (no longer in memory); re-run the tool"
+    start = parsed.offset
+    segment = remainder[start : start + max_chars]
+    chunk, _ = truncate_with_token(
+        segment,
+        max_chars,
+        kind="memory",
+        location=parsed.location,
+        offset=start,
+        limit=max_chars,
+        total_len=len(remainder),
+    )
+    return chunk
+
+
+async def _numbered_continuation(text: str, location: str, char_offset: int, max_chars: int) -> str:
+    """Resume a numbered read at the line containing *char_offset*.
+
+    Delegates to ``read_file(with_lineno=True)`` so the continuation carries
+    ``N|`` line numbers and marks the resumed lines readable for ``edit_lineno``.
+    """
+    pos = 0
+    line = 0
+    for ln in text.splitlines(keepends=True):
+        if pos + len(ln) > char_offset:
+            break
+        pos += len(ln)
+        line += 1
+    return await read_file.handler(
+        location,
+        offset=line,
+        with_lineno=True,
+        max_chars=max_chars,
+    )
+
+
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
 async def get_truncated(token: str, *, max_chars: int = DEFAULT_TOOL_RESULT_MAX_CHARS) -> str:
     """Fetch the next chunk of a truncated result using its ``truncate_token``.
 
     Pass the token shown at the end of a truncated ``read_file``, ``fetch``,
     ``run_command``, or earlier ``get_truncated`` result to keep reading without
     re-requesting the whole source or raising limits. Truncated output carries a
-    fresh token, so chunks chain indefinitely. File chunks start with a 1-based
-    ``L{begin}-{end}`` line range like ``read_file``; memory chunks (generic
-    tool output) are served from a short-lived in-memory store that is forgotten
-    when the agent exits.
+    fresh token, so chunks chain indefinitely. File chunks resume in the same
+    view as the original read: numbered reads continue as numbered lines
+    (editable via ``edit_lineno``), plain reads as raw text with a 1-based
+    ``L{begin}-{end}`` range. Memory chunks (generic tool output) are served
+    from a short-lived in-memory store that is forgotten when the agent exits.
     """
     parsed = decode_truncate_token(token)
     if parsed is None:
@@ -33,26 +75,14 @@ async def get_truncated(token: str, *, max_chars: int = DEFAULT_TOOL_RESULT_MAX_
     if parsed.kind == "http":
         return await fetch.handler(parsed.location, offset=parsed.offset, max_chars=parsed.limit)
     if parsed.kind == "memory":
-        remainder = get_remainder(parsed.location)
-        if remainder is None:
-            return "error: truncate token expired (no longer in memory); re-run the tool"
-        start = parsed.offset
-        segment = remainder[start : start + max_chars]
-        chunk, _ = truncate_with_token(
-            segment,
-            max_chars,
-            kind="memory",
-            location=parsed.location,
-            offset=start,
-            limit=max_chars,
-            total_len=len(remainder),
-        )
-        return chunk
+        return _memory_chunk(parsed, max_chars)
     text, err = read_raw_text(parsed.location)
     if err:
         return err
     text = cast("str", text)
     start = parsed.offset
+    if parsed.numbered:
+        return await _numbered_continuation(text, parsed.location, start, max_chars)
     segment = text[start : start + max_chars]
     chunk, _ = truncate_with_token(
         segment,

src/plyngent/tools/file/glob_paths.py

@@ -65,7 +65,7 @@ def _resolve_glob_base(path: str) -> tuple[Path, Path] | str:
     return root, base
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
 async def glob_paths(
     pattern: str,
     path: str = ".",

src/plyngent/tools/file/grep_files.py

@@ -127,7 +127,7 @@ def _compile_pattern(pattern: str, *, case_insensitive: bool) -> re.Pattern[str]
         return f"error: invalid regex: {exc}"
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
 async def grep_files(
     pattern: str,
     path: str = ".",

src/plyngent/tools/file/listdir.py

@@ -4,7 +4,7 @@ from plyngent.agent import ToolTag, tool
 from plyngent.tools.workspace import resolve_path
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
 async def listdir(path: str = ".") -> str:
     """List entries in a directory under the workspace (name and type)."""
     target = resolve_path(path)

src/plyngent/tools/file/read.py

@@ -4,7 +4,7 @@ from typing import cast
 
 from plyngent.agent import ToolTag, mark_lineno_read, tool
 from plyngent.agent.budget import DEFAULT_TOOL_RESULT_MAX_CHARS
-from plyngent.tools.truncate_token import truncate_with_token
+from plyngent.tools.truncate_token import TruncateToken, truncate_with_token, truncation_marker
 from plyngent.tools.workspace import resolve_path
 
 _LINENO_WIDTH = 6
@@ -49,7 +49,7 @@ def read_raw_text(path: str) -> tuple[str | None, str]:
     return target.read_text(encoding="utf-8", errors="replace"), ""
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
 async def read_file(
     path: str,
     *,
@@ -64,6 +64,8 @@ async def read_file(
     (None = rest of file). When ``with_lineno`` is true, each line is prefixed
     with its 1-based file line number (``     N|…``), matching ``edit_lineno``
     numbering, and those lines are marked readable for ``edit_lineno`` this turn.
+    Only the displayed lines become editable, so read with a small margin around
+    any range you plan to edit with ``edit_lineno``.
 
     ``max_chars`` caps the returned slice (default: the agent's tool-result cap,
     ``DEFAULT_TOOL_RESULT_MAX_CHARS``); a ``truncate_token`` is appended when cut
@@ -83,14 +85,10 @@ async def read_file(
     if start >= len(lines):
         return ""
     slice_lines = lines[start:end]
-    if with_lineno:
-        mark_lineno_read(str(target), set(range(start + 1, end + 1)))
-        body = _format_with_lineno(slice_lines, start_lineno=start + 1)
-    else:
-        body = "".join(slice_lines)
+    body = _format_with_lineno(slice_lines, start_lineno=start + 1) if with_lineno else "".join(slice_lines)
+    char_start = len("".join(lines[:start]))
     if max_chars >= 1:
-        char_start = len("".join(lines[:start]))
-        body, _ = truncate_with_token(
+        body, token = truncate_with_token(
             body,
             max_chars,
             kind="file",
@@ -102,10 +100,39 @@ async def read_file(
             # returned slice itself was cut by max_chars (not just "file continues").
             emit_more_after=False,
         )
+    else:
+        token = None
     if not body:
         return ""
     if with_lineno:
-        return body  # per-line numbers already show the range
+        # Mark only the lines actually displayed: complete numbered lines in the
+        # returned body (the [Truncated] marker, when present, is not a file line).
+        # The max_chars-truncated tail was never shown, so it stays unread —
+        # edit_lineno must not trust line numbers the model never saw.
+        counted = body
+        if token is not None:
+            counted = body.split("\n[Truncated", 1)[0]
+            shown = counted.count("\n")
+            # The truncate cursor counts *numbered* chars (each line carries an
+            # "N|" prefix), which would over-advance into raw text and skip
+            # lines on continuation. Rebase the token onto raw characters of the
+            # complete lines shown (at least one, so the cursor always moves)
+            # and mark it as the numbered view for get_truncated.
+            consumed_lines = max(shown, 1)
+            raw_consumed = sum(len(ln) for ln in slice_lines[:consumed_lines])
+            corrected = TruncateToken(
+                kind="file",
+                location=path,
+                offset=char_start + raw_consumed,
+                limit=max_chars,
+                numbered=True,
+            )
+            omitted = len(text) - (char_start + raw_consumed)
+            body = counted + truncation_marker(max_chars, omitted, corrected)
+        else:
+            shown = counted.count("\n")
+        mark_lineno_read(str(target), set(range(start + 1, start + 1 + shown)))
+        return body
     begin = start + 1
     end_line = start + len(slice_lines)
     return f"L{begin}-{end_line}\n{body}"

src/plyngent/tools/file/tree.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Literal
 
 from plyngent.agent import ToolTag, tool
 from plyngent.tools.workspace import WorkspaceError, get_path_denylist, resolve_path
@@ -10,6 +10,11 @@ if TYPE_CHECKING:
     from collections.abc import Sequence
     from pathlib import Path
 
+# Model-facing renderings. ``markdown`` (default) is the cheapest, most
+# reliable representation for the agent; ``flat`` gives directly-usable paths;
+# ``decorated`` keeps the classic box-drawing tree for human readers.
+type TreeFormat = Literal["markdown", "flat", "decorated"]
+
 DEFAULT_MAX_DEPTH = 4
 DEFAULT_MAX_ENTRIES = 50
 
@@ -144,6 +149,80 @@ def _render_tree(
         lines.append(f"{prefix}└── … ({more} more entries not shown)")
 
 
+def _render_tree_markdown(
+    directory: Path,
+    *,
+    prefix: str,
+    depth: int,
+    limits: _TreeLimits,
+    lines: list[str],
+) -> None:
+    """Nested ``- `` bullet list (2-space indent per level)."""
+    if depth >= limits.max_depth:
+        return
+
+    children = _list_children(directory, limits=limits)
+    if isinstance(children, str):
+        lines.append(f"{prefix}- {children}")
+        return
+
+    truncated = len(children) > limits.max_entries
+    shown = children[: limits.max_entries]
+    for child in shown:
+        try:
+            is_dir = child.is_dir()
+        except OSError:
+            lines.append(f"{prefix}- {child.name} [error: stat failed]")
+            continue
+        if is_dir:
+            lines.append(f"{prefix}- {child.name}/")
+            _render_tree_markdown(
+                child,
+                prefix=prefix + "  ",
+                depth=depth + 1,
+                limits=limits,
+                lines=lines,
+            )
+        else:
+            lines.append(f"{prefix}- {child.name}")
+
+    if truncated:
+        more = len(children) - limits.max_entries
+        lines.append(f"{prefix}- … ({more} more entries not shown)")
+
+
+def _flat_paths(directory: Path, *, limits: _TreeLimits) -> list[str]:
+    """All paths (dirs with trailing ``/``, files plain) in depth-first order.
+
+    Includes empty directories; bounded by *max_depth* (``max_entries`` is
+    applied to the total by the caller).
+    """
+    out: list[str] = []
+
+    def visit(child_dir: Path, rel: str, depth: int) -> None:
+        if depth >= limits.max_depth:
+            return
+        children = _list_children(child_dir, limits=limits)
+        if isinstance(children, str):
+            out.append(f"{rel or child_dir.name} [error: cannot list]")
+            return
+        for child in children:
+            try:
+                is_dir = child.is_dir()
+            except OSError:
+                out.append(f"{child.name} [error: stat failed]")
+                continue
+            child_rel = child.name if not rel else f"{rel}/{child.name}"
+            if is_dir:
+                out.append(f"{child_rel}/")
+                visit(child, child_rel, depth + 1)
+            else:
+                out.append(child_rel)
+
+    visit(directory, "", 0)
+    return out
+
+
 def _resolve_skip_basenames(skip_dirs: Sequence[str] | None) -> frozenset[str]:
     """None → default noise set; explicit list (including empty) replaces defaults."""
     if skip_dirs is None:
@@ -151,10 +230,11 @@ def _resolve_skip_basenames(skip_dirs: Sequence[str] | None) -> frozenset[str]:
     return frozenset(name for name in skip_dirs if name)
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
 async def tree(
     path: str = ".",
     *,
+    format: TreeFormat = "markdown",  # noqa: A002 — model-facing param name
     max_depth: int = DEFAULT_MAX_DEPTH,
     max_entries: int = DEFAULT_MAX_ENTRIES,
     skip_hidden_dirs: bool = True,
@@ -174,8 +254,15 @@ async def tree(
     ``apply_path_denylist`` (default true) hides entries whose full path matches
     the agent ``path_denylist`` policy.
 
+    ``format`` selects the representation (the model chooses per call):
+    ``markdown`` (default) renders a nested ``- `` bullet list — cheapest and
+    most reliable for the agent; ``flat`` lists one relative path per line
+    (directories with a trailing ``/``) so every path is directly usable
+    without reconstructing it; ``decorated`` uses the classic box-drawing tree.
+
     ``max_depth`` limits how deep directories are expanded (1 = origin + children).
-    ``max_entries`` caps how many entries are listed per directory.
+    ``max_entries`` caps entries per directory (``markdown``/``decorated``) or
+    the total number of lines (``flat``).
     """
     if max_depth < 1:
         return "error: max_depth must be >= 1"
@@ -189,8 +276,6 @@ async def tree(
     if not origin.is_dir():
         return f"error: not a directory: {path}"
 
-    root_label = path.rstrip("/\\") or "."
-    lines = [f"{root_label}/"]
     limits = _TreeLimits(
         max_depth=max_depth,
         max_entries=max_entries,
@@ -198,7 +283,18 @@ async def tree(
         skip_basenames=_resolve_skip_basenames(skip_dirs),
         apply_path_denylist=apply_path_denylist,
     )
-    _render_tree(
+
+    if format == "flat":
+        paths = _flat_paths(origin, limits=limits)
+        lines = paths[:max_entries]
+        if len(paths) > max_entries:
+            lines.append(f"… ({len(paths) - max_entries} more paths not shown)")
+        return "\n".join(lines)
+
+    root_label = path.rstrip("/\\") or "."
+    lines: list[str] = [] if format == "markdown" else [f"{root_label}/"]
+    renderer = _render_tree_markdown if format == "markdown" else _render_tree
+    renderer(
         origin,
         prefix="",
         depth=0,

src/plyngent/tools/file/write.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from plyngent.agent import ToolTag, invalidate_lineno_read, tool
+from plyngent.agent import ToolTag, tool
 from plyngent.tools.workspace import resolve_path
 
 
@@ -10,5 +10,6 @@ async def write_file(path: str, content: str) -> str:
     target = resolve_path(path)
     target.parent.mkdir(parents=True, exist_ok=True)
     _ = target.write_text(content, encoding="utf-8")
-    invalidate_lineno_read(str(target))
+    # The write changes the file's mtime, so edit_lineno's freshness check
+    # rejects stale line numbers until the model re-reads with_lineno.
     return f"wrote {len(content)} characters to {path}"

src/plyngent/tools/net/fetch.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from plyngent.agent import ToolTag, tool
+from plyngent.tools.context import is_read_only_context
 from plyngent.tools.net.client import http_fetch
 from plyngent.tools.net.policy import (
     DEFAULT_MAX_BODY_CHARS_IN,
@@ -66,7 +67,7 @@ def format_fetch_result(
     return "\n".join(parts)
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO | ToolTag.TRUSTABLE)
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO | ToolTag.TRUSTABLE | ToolTag.READ_ONLY)
 async def fetch(
     url: str,
     *,
@@ -92,9 +93,13 @@ async def fetch(
     Private/loopback/link-local hosts require an explicit human policy grant
     (not skipped by YOLO). HTTPS→HTTP redirects are blocked unless
     ``allow_http_downgrade`` is true.
+
+    Read-only contexts (e.g. ``/btw --tools=read``) accept only ``GET``.
     """
     try:
         verb = normalize_method(method)
+        if is_read_only_context() and verb != "GET":
+            return "error: fetch method must be GET in a read-only context (POST/PUT/DELETE mutate)"
         parsed = parse_fetch_url(url)
         hdrs = normalize_request_headers(headers, user_agent=user_agent)
         body_bytes: bytes | None = None

src/plyngent/tools/todo.py

@@ -48,7 +48,7 @@ async def _with_todo_stack(mutator: Callable[[TodoStack], str]) -> str:
     return result
 
 
-@tool(name="todo_list", tags=ToolTag.LOCAL | ToolTag.PUBLIC | ToolTag.SESSION_STATE)
+@tool(name="todo_list", tags=ToolTag.LOCAL | ToolTag.PUBLIC | ToolTag.SESSION_STATE | ToolTag.READ_ONLY)
 async def todo_list() -> str:
     """Show the LIFO stack of **task groups** (TOP group = current breakdown level).
 

src/plyngent/tools/truncate_token.py

@@ -37,13 +37,15 @@ class TruncateToken(msgspec.Struct, frozen=True):
     lines (files is line-based via ``read_file``); ``get_truncated`` interprets
     by ``kind``. For ``memory``, ``location`` is a key into the in-memory
     remainder store and ``offset``/``limit`` index characters of that stored
-    remainder.
+    remainder. ``numbered`` marks file tokens from ``read_file(with_lineno)``:
+    the continuation resumes the numbered view instead of raw text.
     """
 
     kind: TruncateKind
     location: str
     offset: int
     limit: int
+    numbered: bool = False
 
 
 # In-memory remainders for generic (non-resumable) tool output. Process-lifetime

src/plyngent/tools/vcs/tools.py

@@ -22,7 +22,7 @@ def _backend_or_error() -> VcsBackend | str:
     return backend
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
 async def vcs_kind() -> str:
     """Return the detected VCS kind under the workspace (e.g. ``git``), or an error."""
     backend = _backend_or_error()
@@ -31,7 +31,7 @@ async def vcs_kind() -> str:
     return backend.kind
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
 async def vcs_status() -> str:
     """Show working-tree status for the detected VCS (read-only)."""
     backend = _backend_or_error()
@@ -40,7 +40,7 @@ async def vcs_status() -> str:
     return backend.status()
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
 async def vcs_diff(path: str = "", *, staged: bool = False) -> str:
     """Show a unified diff for the detected VCS (read-only).
 
@@ -54,7 +54,7 @@ async def vcs_diff(path: str = "", *, staged: bool = False) -> str:
     return backend.diff(staged=staged, path=rel)
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
 async def vcs_log(limit: int = 10) -> str:
     """Show recent commits for the detected VCS (read-only)."""
     if limit < 1:
@@ -65,7 +65,7 @@ async def vcs_log(limit: int = 10) -> str:
     return backend.log(limit=limit)
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
 async def vcs_branch() -> str:
     """Show the current branch / named head for the detected VCS (read-only)."""
     backend = _backend_or_error()

tests/test_agent/test_run_aside.py

@@ -209,3 +209,91 @@ async def test_run_aside_tools_clones_registry_with_fresh_session() -> None:
         pass
     assert hits == ["ok"]
     assert "aside" not in main_session.extras
+
+
+async def test_run_aside_read_only_clone_excludes_write_tools() -> None:
+    """A read-only-cloned registry passed to run_aside exposes no write tools."""
+    hits: list[str] = []
+
+    @tool(tags=ToolTag.LOCAL | ToolTag.READ_ONLY, register=False)
+    async def peek() -> str:
+        hits.append("peek")
+        return "peeked"
+
+    @tool(tags=ToolTag.LOCAL, register=False)
+    async def write_thing() -> str:
+        hits.append("write")
+        return "wrote"
+
+    registry = ToolRegistry([peek, write_thing], auto_bind_state=True)
+
+    def make_client(first_tool: str | None) -> Any:
+        calls = 0
+
+        class _Client:
+            @overload
+            async def chat_completions(
+                self, param: ChatCompletionsParam, *, stream: Literal[False] = False
+            ) -> ChatCompletionResponse: ...
+
+            @overload
+            async def chat_completions(
+                self, param: ChatCompletionsParam, *, stream: Literal[True]
+            ) -> AsyncIterator[ChatCompletionChunk]: ...
+
+            async def chat_completions(
+                self, param: ChatCompletionsParam, *, stream: bool = False
+            ) -> ChatCompletionResponse | AsyncIterator[ChatCompletionChunk]:
+                del stream, param
+                nonlocal calls
+                calls += 1
+                if calls == 1 and first_tool is not None:
+                    from plyngent.lmproto.openai_compatible.model import (
+                        AssistantFunctionTool,
+                        AssistantFunctionToolCall,
+                    )
+
+                    message = AssistantChatMessage(
+                        content="",
+                        tool_calls=[
+                            AssistantFunctionToolCall(
+                                id="c1",
+                                function=AssistantFunctionTool(name=first_tool, arguments="{}"),
+                            )
+                        ],
+                    )
+                    finish = "tool_calls"
+                else:
+                    message = AssistantChatMessage(content="done")
+                    finish = "stop"
+                return ChatCompletionResponse(
+                    id="1",
+                    object="chat.completion",
+                    created=0,
+                    model="m",
+                    choices=[
+                        ChatCompletionChoice(
+                            index=0,
+                            message=message,
+                            logprobs={},
+                            finish_reason=finish,
+                        )
+                    ],
+                    system_fingerprint="",
+                    usage={},
+                )
+
+        return _Client()
+
+    # Read-only tool is available to the side turn.
+    agent_peek = ChatAgent(make_client("peek"), model="m", tools=registry, stream=False)
+    async for _ in agent_peek.run_aside("go", tools=registry.clone(read_only_only=True)):
+        pass
+    assert hits == ["peek"]
+
+    # Write tool is not in the read-only clone: the call errors, never runs.
+    hits.clear()
+    agent_write = ChatAgent(make_client("write_thing"), model="m", tools=registry, stream=False)
+    async for _ in agent_write.run_aside("go", tools=registry.clone(read_only_only=True)):
+        pass
+    assert hits == []

tests/test_cli/test_display.py

@@ -65,7 +65,7 @@ async def test_flush_markdown_on_source_change(
 
     call = AssistantFunctionToolCall(
         id="1",
-        function=AssistantFunctionTool(name="run_argv", arguments='{"argv": ["echo", "x"]}'),
+        function=AssistantFunctionTool(name="vcs_diff", arguments='{"path": ""}'),
     )
     await render_events(
         _aiter(
@@ -98,7 +98,7 @@ def _result(content: str) -> ToolResultEvent:
 async def test_pretty_read_file_done(capsys: pytest.CaptureFixture[str]) -> None:
     await render_events(_aiter([_read_call('{"path": "a.txt"}'), _result("L1-4\none\ntwo\n")]))
     out = capsys.readouterr().out
-    assert "* Read 'a.txt' L1-4 (done)" in out
+    assert "* Read 'a.txt' (done)" in out
     assert "[tool]" not in out
     assert "[tool ok]" not in out
 
@@ -116,6 +116,44 @@ async def test_pretty_read_file_statuses(capsys: pytest.CaptureFixture[str]) ->
         assert f"* Read 'a.txt' {status}" in out, f"status {status} not rendered for {content!r}"
 
 
+def _tree_call(args_json: str) -> ToolCallEvent:
+    return ToolCallEvent(
+        tool_call=AssistantFunctionToolCall(
+            id="1",
+            function=AssistantFunctionTool(name="tree", arguments=args_json),
+        )
+    )
+
+
+async def test_pretty_tree_markdown(capsys: pytest.CaptureFixture[str]) -> None:
+    result = "- src/\n  - main.py\n  - nested/\n    - deep.txt\n- a.txt\n"
+    await render_events(_aiter([_tree_call('{"path": "."}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Tree '.' (2 dirs, 3 files, depth 3)" in out
+    assert "[tool]" not in out
+
+
+async def test_pretty_tree_flat_omits_depth(capsys: pytest.CaptureFixture[str]) -> None:
+    result = "src/\nsrc/main.py\na.txt\n… (5 more paths not shown)"
+    await render_events(_aiter([_tree_call('{"path": ".", "format": "flat"}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Tree '.' (1 dirs, 2 files)" in out
+    assert "depth" not in out
+
+
+async def test_pretty_tree_decorated(capsys: pytest.CaptureFixture[str]) -> None:
+    result = "src/\n├── main.py\n└── nested/\n    └── deep.txt\n"
+    await render_events(_aiter([_tree_call('{"path": "src", "format": "decorated"}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Tree 'src' (1 dirs, 2 files, depth 2)" in out
+
+
+async def test_pretty_tree_error(capsys: pytest.CaptureFixture[str]) -> None:
+    await render_events(_aiter([_tree_call('{"path": "nope"}'), _result("error: not a directory: nope")]))
+    out = capsys.readouterr().out
+    assert "* Tree 'nope' (error)" in out
+
+
 async def test_pretty_todo_push(capsys: pytest.CaptureFixture[str]) -> None:
     call = ToolCallEvent(
         tool_call=AssistantFunctionToolCall(
@@ -140,12 +178,12 @@ async def test_non_pretty_tool_keeps_old_style(capsys: pytest.CaptureFixture[str
     call = ToolCallEvent(
         tool_call=AssistantFunctionToolCall(
             id="3",
-            function=AssistantFunctionTool(name="run_argv", arguments='{"argv": ["ls"]}'),
+            function=AssistantFunctionTool(name="vcs_diff", arguments='{"path": ""}'),
         )
     )
-    await render_events(_aiter([call, _result("exit_code=0")]))
+    await render_events(_aiter([call, _result("diff --git a/x b/x")]))
     out = capsys.readouterr().out
-    assert "[tool] run_argv(" in out
+    assert "[tool] vcs_diff(" in out
     assert "[tool ok]" in out
 
 
@@ -210,3 +248,305 @@ def test_markdown_render_available_respects_plain_env(monkeypatch: pytest.Monkey
     monkeypatch.setenv("PLYNGENT_PLAIN", "1")
     assert markdown_render_available() is False
     monkeypatch.delenv("PLYNGENT_PLAIN", raising=False)
+
+
+async def test_pretty_verbose_shows_full_result(capsys: pytest.CaptureFixture[str]) -> None:
+    """Verbose mode prints the full tool result even for pretty tools."""
+    await render_events(
+        _aiter([_read_call('{"path": "a.txt"}'), _result("L1-4\none\ntwo\n")]),
+        verbose=True,
+    )
+    out = capsys.readouterr().out
+    assert "[tool ok]" in out
+    assert "L1-4\none\ntwo" in out
+    assert "* Read" not in out
+
+
+async def test_pretty_plain_output_has_no_ansi(capsys: pytest.CaptureFixture[str]) -> None:
+    """Pretty summaries are plain text when stdout is not a TTY."""
+    await render_events(_aiter([_read_call('{"path": "a.txt"}'), _result("L1-4\none\n")]))
+    out = capsys.readouterr().out
+    assert "\x1b[" not in out
+
+
+async def test_tool_result_preview_first_line_and_count(capsys: pytest.CaptureFixture[str]) -> None:
+    """Non-verbose preview shows the first line plus a line-count tail."""
+    set_markdown_enabled(False)
+    content = "first line\nsecond\nthird"
+    await render_events(_aiter([ToolResultEvent(message=ToolChatMessage(content=content, tool_call_id="1"))]))
+    out = capsys.readouterr().out
+    assert "[tool ok] first line (…2 more lines)" in out
+
+    await render_events(_aiter([ToolResultEvent(message=ToolChatMessage(content="only line", tool_call_id="2"))]))
+    out = capsys.readouterr().out
+    assert "[tool ok] only line" in out
+    assert "more" not in out
+
+    await render_events(_aiter([ToolResultEvent(message=ToolChatMessage(content="", tool_call_id="3"))]))
+    out = capsys.readouterr().out
+    assert "[tool ok] (no output)" in out
+    set_markdown_enabled(True)
+
+
+def _pretty_call(name: str, args_json: str) -> ToolCallEvent:
+    return ToolCallEvent(
+        tool_call=AssistantFunctionToolCall(
+            id="1",
+            function=AssistantFunctionTool(name=name, arguments=args_json),
+        )
+    )
+
+
+async def test_pretty_listdir(capsys: pytest.CaptureFixture[str]) -> None:
+    result = "dir\tsrc\nfile\tREADME.md\nfile\tpyproject.toml\n"
+    await render_events(_aiter([_pretty_call("listdir", '{"path": "src"}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* List 'src' (1 dirs, 2 files)" in out
+    assert "[tool]" not in out
+    assert "[tool ok]" not in out
+
+
+async def test_pretty_listdir_empty(capsys: pytest.CaptureFixture[str]) -> None:
+    await render_events(_aiter([_pretty_call("listdir", '{"path": "src"}'), _result("(empty)")]))
+    out = capsys.readouterr().out
+    assert "* List 'src' (empty)" in out
+
+
+async def test_pretty_listdir_error(capsys: pytest.CaptureFixture[str]) -> None:
+    await render_events(_aiter([_pretty_call("listdir", '{"path": "nope"}'), _result("error: not a directory: nope")]))
+    out = capsys.readouterr().out
+    assert "* List 'nope' (error: not a directory: nope)" in out
+
+
+async def test_pretty_glob(capsys: pytest.CaptureFixture[str]) -> None:
+    result = "src/a.py\nsrc/b.py\n"
+    await render_events(_aiter([_pretty_call("glob_paths", '{"pattern": "**/*.py"}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Glob '**/*.py' in '.' (2 paths)" in out
+
+
+async def test_pretty_glob_no_matches(capsys: pytest.CaptureFixture[str]) -> None:
+    await render_events(_aiter([_pretty_call("glob_paths", '{"pattern": "**/*.rs"}'), _result("(no matches)")]))
+    out = capsys.readouterr().out
+    assert "* Glob '**/*.rs' in '.' (no matches)" in out
+
+
+async def test_pretty_glob_truncated(capsys: pytest.CaptureFixture[str]) -> None:
+    result = "a\nb\n...[truncated at 200 matches]"
+    await render_events(_aiter([_pretty_call("glob_paths", '{"pattern": "*"}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Glob '*' in '.' (2 paths)" in out
+
+
+async def test_pretty_grep(capsys: pytest.CaptureFixture[str]) -> None:
+    result = "src/a.py:3: x = 1\nsrc/a.py:9: x = 2\nsrc/b.py:1: y = 3\n"
+    await render_events(_aiter([_pretty_call("grep_files", '{"pattern": "x"}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Grep 'x' in '.' (3 matches in 2 files)" in out
+
+
+async def test_pretty_grep_no_matches(capsys: pytest.CaptureFixture[str]) -> None:
+    await render_events(_aiter([_pretty_call("grep_files", '{"pattern": "zzz"}'), _result("(no matches)")]))
+    out = capsys.readouterr().out
+    assert "* Grep 'zzz' in '.' (no matches)" in out
+
+
+async def test_pretty_run_argv_success(capsys: pytest.CaptureFixture[str]) -> None:
+    result = (
+        "exit_code=0\ntimed_out=false\ncwd=.\ncmd=git status --short\n--- stdout ---\n M src/foo.py\n--- stderr ---\n"
+    )
+    await render_events(_aiter([_pretty_call("run_argv", '{"argv": ["git", "status", "--short"]}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Run $ git status --short (exit code 0)" in out
+    assert "[tool]" not in out
+    assert "[tool ok]" not in out
+
+
+async def test_pretty_run_argv_failure(capsys: pytest.CaptureFixture[str]) -> None:
+    result = (
+        "exit_code=1\ntimed_out=false\ncwd=.\ncmd=ls /nope\n--- stdout ---\n--- stderr ---\nls: cannot access '/nope'\n"
+    )
+    await render_events(_aiter([_pretty_call("run_argv", '{"argv": ["ls", "/nope"]}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Run $ ls /nope (exit code 1)" in out
+
+
+async def test_pretty_run_argv_timed_out(capsys: pytest.CaptureFixture[str]) -> None:
+    result = "exit_code=\ntimed_out=true\ncwd=.\ncmd=sleep 10\n--- stdout ---\n--- stderr ---\n"
+    await render_events(_aiter([_pretty_call("run_argv", '{"argv": ["sleep", "10"]}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Run $ sleep 10 (timed out)" in out
+
+
+async def test_pretty_run_argv_error(capsys: pytest.CaptureFixture[str]) -> None:
+    await render_events(
+        _aiter([_pretty_call("run_argv", '{"argv": ["nope"]}'), _result("error: executable not found: 'nope'")])
+    )
+    out = capsys.readouterr().out
+    assert "* Run $ nope (error: executable not found: 'nope')" in out
+
+
+async def test_pretty_run_argv_batch_stopped(capsys: pytest.CaptureFixture[str]) -> None:
+    result = (
+        "steps=3 ran=2 stop_on_error=true stopped_early=true\n"
+        "--- step 1 ---\nexit_code=1\ntimed_out=false\ncwd=.\ncmd=git status\n"
+        "pipe_out=false\nmix_stderr=false\n--- stdout ---\n--- stderr ---\n"
+        "--- step 2 ---\nexit_code=0\ntimed_out=false\ncwd=.\ncmd=git diff\n"
+        "pipe_out=false\nmix_stderr=false\n--- stdout ---\n--- stderr ---\n"
+        "--- summary ---\nlast_exit=1\n"
+    )
+    await render_events(_aiter([_pretty_call("run_argv_batch", '{"steps": []}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Run batch (stopped early)" in out
+
+
+async def test_pretty_run_argv_batch_complete(capsys: pytest.CaptureFixture[str]) -> None:
+    result = (
+        "steps=2 ran=2 stop_on_error=true stopped_early=false\n"
+        "--- step 1 ---\nexit_code=0\n--- stdout ---\n--- stderr ---\n"
+        "--- step 2 ---\nexit_code=0\n--- stdout ---\n--- stderr ---\n"
+        "--- summary ---\nlast_exit=0\n"
+    )
+    await render_events(_aiter([_pretty_call("run_argv_batch", '{"steps": []}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Run batch (done)" in out
+
+
+async def test_pretty_fetch(capsys: pytest.CaptureFixture[str]) -> None:
+    result = (
+        "status=200\nmethod=GET\nfinal_url=https://example.com\ncontent_type=application/json\n"
+        "body_kind=json\nbytes=1229\ntruncated=false\nredirects=0\nsecurity=public\n--- body ---\n{...}\n"
+    )
+    await render_events(_aiter([_pretty_call("fetch", '{"url": "https://example.com"}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Fetch GET https://example.com (200)" in out
+    assert "[tool]" not in out
+
+
+async def test_pretty_fetch_error_status(capsys: pytest.CaptureFixture[str]) -> None:
+    result = (
+        "status=404\nmethod=GET\nfinal_url=https://example.com/nope\ncontent_type=text/html\n"
+        "body_kind=html\nbytes=45\ntruncated=false\nredirects=0\nsecurity=public\n--- body ---\nNot found\n"
+    )
+    await render_events(_aiter([_pretty_call("fetch", '{"url": "https://example.com/nope"}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Fetch GET https://example.com/nope (404)" in out
+
+
+async def test_pretty_fetch_truncated(capsys: pytest.CaptureFixture[str]) -> None:
+    result = (
+        "status=200\nmethod=GET\nfinal_url=https://example.com\ncontent_type=text/plain\n"
+        "body_kind=text\nbytes=50000\ntruncated=true\nredirects=0\nsecurity=public\n--- body ---\n..."
+    )
+    await render_events(_aiter([_pretty_call("fetch", '{"url": "https://example.com"}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Fetch GET https://example.com (200)" in out
+
+
+async def test_pretty_fetch_error(capsys: pytest.CaptureFixture[str]) -> None:
+    await render_events(
+        _aiter([_pretty_call("fetch", '{"url": "https://example.com"}'), _result("error: fetch failed: boom")])
+    )
+    out = capsys.readouterr().out
+    assert "* Fetch GET https://example.com (error: fetch failed: boom)" in out
+
+
+async def test_pretty_mutators(capsys: pytest.CaptureFixture[str]) -> None:
+    cases = [
+        ("write_file", '{"path": "src/x.py"}', "wrote 120 characters to src/x.py", "* Write 'src/x.py' (120 chars)"),
+        (
+            "edit_replace",
+            '{"path": "src/a.py"}',
+            "replaced 1 occurrence in src/a.py ('x' → 'y')",
+            "* Edit 'src/a.py' (done)",
+        ),
+        (
+            "edit_lineno",
+            '{"path": "src/a.py"}',
+            "replaced lines 3-5 (3 lines; first: 'pass') with 2 lines in src/a.py",
+            "* Edit 'src/a.py' (done)",
+        ),
+        (
+            "copy_path",
+            '{"src": "a.txt", "dst": "b.txt"}',
+            "copied file a.txt -> b.txt",
+            "* Copy 'a.txt' → 'b.txt' (done)",
+        ),
+        ("move_path", '{"src": "src", "dst": "dst"}', "moved directory src -> dst", "* Move 'src' → 'dst' (done)"),
+        ("delete_path", '{"path": "tmp/x.txt"}', "deleted file tmp/x.txt", "* Delete 'tmp/x.txt' (done)"),
+    ]
+    for name, args, content, expected in cases:
+        await render_events(_aiter([_pretty_call(name, args), _result(content)]))
+        out = capsys.readouterr().out
+        assert expected in out, f"{name}: expected {expected!r} in {out!r}"
+        assert "[tool ok]" not in out
+
+
+async def test_pretty_mutator_error(capsys: pytest.CaptureFixture[str]) -> None:
+    await render_events(
+        _aiter([_pretty_call("copy_path", '{"src": "x", "dst": "y"}'), _result("error: source does not exist: x")])
+    )
+    out = capsys.readouterr().out
+    assert "* Copy 'x' → 'y' (error: source does not exist: x)" in out
+
+
+async def test_pretty_vcs_status_done(capsys: pytest.CaptureFixture[str]) -> None:
+    await render_events(_aiter([_pretty_call("vcs_status", "{}"), _result("(clean)")]))
+    out = capsys.readouterr().out
+    assert "* VCS Status (done)" in out
+    assert "[tool]" not in out
+
+
+async def test_pretty_vcs_status_error(capsys: pytest.CaptureFixture[str]) -> None:
+    await render_events(
+        _aiter(
+            [
+                _pretty_call("vcs_status", "{}"),
+                _result("error: no supported VCS detected under workspace (currently: git)"),
+            ]
+        )
+    )
+    out = capsys.readouterr().out
+    assert "* VCS Status (error: no supported VCS detected under workspace (currently: git))" in out
+
+
+async def test_pretty_vcs_log_count(capsys: pytest.CaptureFixture[str]) -> None:
+    result = (
+        "5436457 2026-08-15 worldmozara core/tools: allow edit_lineno append\nabc1234 2026-08-14 worldmozara cli: fix\n"
+    )
+    await render_events(_aiter([_pretty_call("vcs_log", '{"limit": 2}'), _result(result)]))
+    out = capsys.readouterr().out
+    assert "* VCS Log (2 commits)" in out
+
+
+async def test_pretty_vcs_log_single(capsys: pytest.CaptureFixture[str]) -> None:
+    content = "abc1234 2026-08-14 worldmozara x\n"
+    await render_events(_aiter([_pretty_call("vcs_log", '{"limit": 1}'), _result(content)]))
+    out = capsys.readouterr().out
+    assert "* VCS Log (1 commit)" in out
+
+
+async def test_pretty_vcs_log_no_commits(capsys: pytest.CaptureFixture[str]) -> None:
+    await render_events(_aiter([_pretty_call("vcs_log", "{}"), _result("(no commits)")]))
+    out = capsys.readouterr().out
+    assert "* VCS Log (no commits)" in out
+
+
+async def test_pretty_wait_waited(capsys: pytest.CaptureFixture[str]) -> None:
+    await render_events(_aiter([_pretty_call("wait", '{"duration": 5}'), _result("waited 5s")]))
+    out = capsys.readouterr().out
+    assert "* Wait (5s)" in out
+    assert "[tool]" not in out
+
+
+async def test_pretty_wait_disturbed(capsys: pytest.CaptureFixture[str]) -> None:
+    await render_events(_aiter([_pretty_call("wait", '{"duration": 30}'), _result("disturbed by user: ship it")]))
+    out = capsys.readouterr().out
+    assert "* Wait (disturbed)" in out
+
+
+async def test_pretty_wait_error(capsys: pytest.CaptureFixture[str]) -> None:
+    content = "error: duration must not be negative"
+    await render_events(_aiter([_pretty_call("wait", '{"duration": -1}'), _result(content)]))
+    out = capsys.readouterr().out
+    assert "* Wait (error: duration must not be negative)" in out

tests/test_cli/test_repl_commands.py

@@ -633,3 +633,108 @@ async def test_repl_second_ctrl_c_after_cancel_returns_to_prompt(
     assert calls["reads"] == 4  # three entries + EOF
     assert calls["echoes"] == 3  # the interrupting echo still counted, loop survived
     assert "interrupted" in capsys.readouterr().out
+
+
+async def test_btw_default_read_falls_back_when_tools_off(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
+    """Bare /btw (default --tools=read) still works under /tools off: no tools."""
+    assert state.tools_enabled is False
+    state.agent.stream = False  # DummyClient only completes via non-stream path
+    assert await handle_slash(state, "/btw hello") is True
+    out = capsys.readouterr().out
+    assert "btw: hello" in out
+    assert "btw done" in out
+    assert "error" not in out
+
+
+async def test_btw_explicit_tools_errors_when_off(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
+    """Explicitly requesting read/full tools while they are off is an error."""
+    state.agent.stream = False  # DummyClient only completes via non-stream path
+    assert await handle_slash(state, "/btw --tools read q") is True
+    assert "error: --tools requires tools on" in capsys.readouterr().out
+    assert await handle_slash(state, "/btw --tools full q") is True
+    assert "error: --tools requires tools on" in capsys.readouterr().out
+    # Explicit no is fine.
+    assert await handle_slash(state, "/btw --tools no q") is True
+    out = capsys.readouterr().out
+    assert "btw done" in out
+    assert "error" not in out
+
+
+async def test_btw_read_mode_wires_read_only_clone(
+    state: ReplState,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """--tools=read passes a read-only clone + read-only context + main session."""
+
+    from plyngent.agent import ToolRegistry, ToolTag, tool
+    from plyngent.tools.context import is_read_only_context
+
+    @tool(tags=ToolTag.LOCAL | ToolTag.READ_ONLY, register=False)
+    async def ro_tool() -> str:
+        return "ro"
+
+    @tool(tags=ToolTag.LOCAL, register=False)
+    async def w_tool() -> str:
+        return "w"
+
+    state.tools_enabled = True
+    state.agent.tools = ToolRegistry([ro_tool, w_tool], auto_bind_state=True)
+    captured: dict[str, object] = {}
+
+    async def fake_run_aside(_text: str, **kwargs: object) -> AsyncIterator[object]:
+        captured["tools"] = kwargs.get("tools")
+        captured["session_state"] = kwargs.get("session_state")
+        captured["read_only"] = is_read_only_context()
+        if False:
+            yield None
+
+    monkeypatch.setattr(state.agent, "run_aside", fake_run_aside)
+
+    assert await handle_slash(state, "/btw --tools read hi") is True
+    assert "btw done" in capsys.readouterr().out
+
+    tools = captured["tools"]
+    assert isinstance(tools, ToolRegistry)
+    assert {t.name for t in tools.definitions()} == {"ro_tool"}
+    assert captured["session_state"] is state.session_state
+    assert captured["read_only"] is True
+
+
+async def test_btw_full_mode_clones_full_registry(
+    state: ReplState,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """--tools=full keeps the whole registry with a fresh session, no context."""
+
+    from plyngent.agent import ToolRegistry, ToolTag, tool
+    from plyngent.tools.context import is_read_only_context
+
+    @tool(tags=ToolTag.LOCAL | ToolTag.READ_ONLY, register=False)
+    async def ro_tool() -> str:
+        return "ro"
+
+    @tool(tags=ToolTag.LOCAL, register=False)
+    async def w_tool() -> str:
+        return "w"
+
+    state.tools_enabled = True
+    state.agent.tools = ToolRegistry([ro_tool, w_tool], auto_bind_state=True)
+    captured: dict[str, object] = {}
+
+    async def fake_run_aside(_text: str, **kwargs: object) -> AsyncIterator[object]:
+        captured["tools"] = kwargs.get("tools")
+        captured["session_state"] = kwargs.get("session_state")
+        captured["read_only"] = is_read_only_context()
+        if False:
+            yield None
+
+    monkeypatch.setattr(state.agent, "run_aside", fake_run_aside)
+
+    assert await handle_slash(state, "/btw --tools full hi") is True
+    assert "btw done" in capsys.readouterr().out
+
+    assert captured["tools"] is True
+    assert captured["session_state"] is not state.session_state
+    assert captured["read_only"] is False

tests/test_tools/test_catalog.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import inspect
 
-from plyngent.agent import ToolTag, tool
+from plyngent.agent import ToolRegistry, ToolTag, tool
 from plyngent.tools import default_tool_definitions, register_builtin_tools
 from plyngent.tools.catalog import ToolCatalog, ToolSource, catalog_scope, get_catalog, registration_source
 from plyngent.tools.chat import CHAT_TOOLS
@@ -42,6 +42,45 @@ def test_public_surface_is_subset() -> None:
     assert "read_file" not in public
 
 
+def test_read_only_classification() -> None:
+    """READ_ONLY marks exactly the non-mutating builtins."""
+    register_builtin_tools()
+    read_only = {t.name for t in default_tool_definitions(surface="local") if t.tags & ToolTag.READ_ONLY}
+    assert read_only == {
+        "read_file",
+        "get_truncated",
+        "glob_paths",
+        "grep_files",
+        "listdir",
+        "tree",
+        "todo_list",
+        "wait",
+        "ask_user_line",
+        "ask_user_choice",
+        "ask_user_form",
+        "vcs_kind",
+        "vcs_status",
+        "vcs_diff",
+        "vcs_log",
+        "vcs_branch",
+        "fetch",
+    }
+
+
+def test_registry_clone_read_only_only() -> None:
+    """clone(read_only_only=True) keeps only READ_ONLY tools, nothing else."""
+    register_builtin_tools()
+    full = ToolRegistry(default_tool_definitions(surface="local"))
+    ro = full.clone(read_only_only=True)
+    assert 0 < len(ro) < len(full)
+    assert {t.name for t in ro.definitions()} == {t.name for t in full.definitions() if t.tags & ToolTag.READ_ONLY}
+    assert all(t.tags & ToolTag.READ_ONLY for t in ro.definitions())
+    assert "run_argv" not in ro
+    assert "edit_replace" not in ro
+    # Plain clone keeps every tool.
+    assert len(full.clone()) == len(full)
+
+
 def test_catalog_scope_empty_isolates_registration() -> None:
     with catalog_scope(empty=True) as catalog:
 

tests/test_tools/test_edit_lineno.py

@@ -19,9 +19,38 @@ def test_edit_lineno_replace_middle(workspace: object) -> None:
     _ = _read_lineno("a.txt")
     out = call_sync(edit_lineno, "a.txt", 2, 3, "TWO\nTHREE\n")
     assert "replaced lines 2-3" in out
+    assert "first: 'two'" in out  # old-content excerpt for self-correction
     assert call_sync(read_file, "a.txt") == "L1-4\none\nTWO\nTHREE\nfour\n"
 
 
+def test_edit_lineno_read_truncated_marks_only_displayed_lines(workspace: object) -> None:
+    """A max_chars-truncated numbered read must not mark the unseen tail."""
+    del workspace
+    _ = call_sync(write_file, "t.txt", "".join(f"line {i}\n" for i in range(1, 101)))
+    # Displayed complete lines are editable...
+    _ = reset_lineno_tracker()
+    _ = call_sync(read_file, "t.txt", with_lineno=True, max_chars=80)
+    out = call_sync(edit_lineno, "t.txt", 5, 5, "FIVE\n")
+    assert "replaced lines 5-5" in out
+    # ...but the truncated tail (never shown) stays unread.
+    _ = reset_lineno_tracker()
+    _ = call_sync(read_file, "t.txt", with_lineno=True, max_chars=80)
+    out = call_sync(edit_lineno, "t.txt", 90, 90, "X\n")
+    assert "not read" in out and "90" in out
+
+
+def test_edit_lineno_error_shows_read_window(workspace: object) -> None:
+    """The unread error names the read window so the model can re-read precisely."""
+    del workspace
+    _ = call_sync(write_file, "j.txt", "a\nb\nc\nd\n")
+    _ = reset_lineno_tracker()
+    _ = call_sync(read_file, "j.txt", offset=1, limit=2, with_lineno=True)  # lines 2-3
+    out = call_sync(edit_lineno, "j.txt", 1, 1, "X\n")
+    assert "not read" in out
+    assert "read lines 2-3" in out
+    assert "margin" in out
+
+
 def test_edit_lineno_requires_all_lines_read(workspace: object) -> None:
     del workspace
     _ = call_sync(write_file, "r.txt", "a\nb\nc\nd\ne\n")
@@ -37,6 +66,16 @@ def test_edit_lineno_requires_all_lines_read(workspace: object) -> None:
     assert "replaced lines 2-3" in out
 
 
+def test_edit_lineno_append_to_empty_file(workspace: object) -> None:
+    """An empty file has nothing to clobber, so appending needs no read."""
+    del workspace
+    _ = call_sync(write_file, "empty.txt", "")
+    _ = reset_lineno_tracker()
+    out = call_sync(edit_lineno, "empty.txt", 1, 1, "first\n")
+    assert "appended" in out
+    assert call_sync(read_file, "empty.txt") == "L1-1\nfirst\n"
+
+
 def test_edit_lineno_append_requires_last_line_read(workspace: object) -> None:
     del workspace
     _ = call_sync(write_file, "ap.txt", "a\nb\nc\n")
@@ -92,7 +131,7 @@ def test_edit_lineno_requires_rerun_after_write(workspace: object) -> None:
     _ = _read_lineno("f.txt")
     _ = call_sync(write_file, "f.txt", "changed\n")
     out = call_sync(edit_lineno, "f.txt", 1, 1, "x\n")
-    assert "not read" in out
+    assert "changed" in out and "re-read" in out
 
 
 def test_edit_lineno_requires_rerun_after_edit_replace(workspace: object) -> None:
@@ -101,7 +140,7 @@ def test_edit_lineno_requires_rerun_after_edit_replace(workspace: object) -> Non
     _ = _read_lineno("g.txt")
     _ = call_sync(edit_replace, "g.txt", "two", "TWO")
     out = call_sync(edit_lineno, "g.txt", 2, 2, "x\n")
-    assert "not read" in out
+    assert "changed" in out and "re-read" in out
 
 
 def test_edit_lineno_invalidates_after_successful_edit(workspace: object) -> None:
@@ -111,4 +150,19 @@ def test_edit_lineno_invalidates_after_successful_edit(workspace: object) -> Non
     _ = call_sync(edit_lineno, "h.txt", 2, 2, "B\n")
     # After the edit, line numbers are stale — must re-read before editing again.
     out = call_sync(edit_lineno, "h.txt", 2, 2, "C\n")
-    assert "not read" in out
+    assert "changed" in out and "re-read" in out
+
+
+def test_edit_lineno_detects_external_change(workspace: object) -> None:
+    """A mutation outside the file tools (run_argv, another process) is caught."""
+    import os
+    import time
+
+    assert isinstance(workspace, Path)
+    _ = call_sync(write_file, "i.txt", "one\ntwo\nthree\n")
+    _ = _read_lineno("i.txt")
+    # Simulate an external/process-tool mutation: rewrite content + new mtime.
+    _ = call_sync(write_file, "i.txt", "one\ntwo\nthree\nCHANGED\n")
+    os.utime(workspace / "i.txt", (time.time() + 1, time.time() + 1))
+    out = call_sync(edit_lineno, "i.txt", 1, 1, "x\n")
+    assert "changed since it was read" in out

tests/test_tools/test_fetch.py

@@ -8,6 +8,7 @@ from typing import TYPE_CHECKING, override
 
 import pytest
 
+from plyngent.tools.context import read_only_context
 from plyngent.tools.danger import classify_danger
 from plyngent.tools.net import fetch, grant_private_host
 from plyngent.tools.net.policy import (
@@ -199,6 +200,23 @@ async def test_fetch_get_post_put_delete(workspace: Path, http_server: str) -> N
     assert "status=204" in out_del
 
 
+async def test_fetch_read_only_context_blocks_non_get(workspace: Path, http_server: str) -> None:
+    """In a read-only context fetch only allows GET; outside it POST works."""
+    del workspace
+    port = int(http_server.rsplit(":", 1)[1])
+    grant_private_host("127.0.0.1", port)
+
+    with read_only_context():
+        out = await call_async(fetch, f"{http_server}/hello")
+        assert "status=200" in out  # GET is fine in a read-only context
+        blocked = await call_async(fetch, f"{http_server}/echo", method="POST", body="hi")
+        assert blocked.startswith("error:")
+        assert "GET" in blocked
+
+    out_post = await call_async(fetch, f"{http_server}/echo", method="POST", body="hi")
+    assert "POST:hi" in out_post
+
+
 async def test_fetch_user_agent_passthrough(workspace: Path, http_server: str) -> None:
     del workspace
     port = int(http_server.rsplit(":", 1)[1])

tests/test_tools/test_file.py

@@ -74,6 +74,7 @@ def test_edit_replace_first_only(workspace: object) -> None:
     assert call_sync(read_file, "t.txt") == "L1-1\nbb aa"
     assert "1 of 2" in result or "1 of 2 matches" in result
     assert "remain" in result
+    assert "'aa' → 'bb'" in result  # old→new excerpt for self-correction
 
 
 def test_edit_replace_max_replaces(workspace: object) -> None:

tests/test_tools/test_get_truncated.py

@@ -44,6 +44,40 @@ async def test_get_truncated_file_chains_without_gap(workspace: Path) -> None:
     assert total_read == 1000
 
 
+async def test_get_truncated_numbered_continues_and_marks(workspace: Path) -> None:
+    """A numbered read's token resumes the numbered view, gap-free and editable."""
+    from plyngent.agent import reset_lineno_tracker
+    from plyngent.tools.file import edit_lineno, read_file
+
+    (workspace / "big.txt").write_text("".join(f"line {i}\n" for i in range(1, 101)), encoding="utf-8")
+    reset_lineno_tracker()
+    first = await call_async(read_file, "big.txt", with_lineno=True, max_chars=120)
+    token = _extract_token(first)
+    assert token is not None
+    parsed = decode_truncate_token(token)
+    assert parsed is not None
+    assert parsed.kind == "file"
+    assert parsed.numbered is True
+    # The token points at the raw start of the next line (no skip, no overlap).
+    body = first.split("\n[Truncated", 1)[0]
+    shown = body.count("\n")
+    expected_raw = sum(len(f"line {i}\n") for i in range(1, shown + 1))
+    assert parsed.offset == expected_raw
+
+    # Continuation is numbered and starts right after the displayed lines.
+    cont = await call_async(get_truncated, token, max_chars=200)
+    numbered = [ln for ln in cont.splitlines() if "|" in ln and ln.strip().split("|", 1)[0].strip().isdigit()]
+    assert numbered
+    first_num = int(numbered[0].split("|", 1)[0].strip())
+    assert first_num == shown + 1
+
+    # Resumed lines are editable via edit_lineno; unseen lines stay unread.
+    out = await call_async(edit_lineno, "big.txt", first_num, first_num, "EDITED\n")
+    assert "replaced lines" in out
+    out2 = await call_async(edit_lineno, "big.txt", 90, 90, "X\n")
+    assert "not read" in out2
+
+
 async def test_get_truncated_file_missing(workspace: Path) -> None:
     token = encode_truncate_token(TruncateToken(kind="file", location="nope.txt", offset=0, limit=200))
     out = await call_async(get_truncated, token, max_chars=200)

tests/test_tools/test_tree.py

@@ -67,7 +67,60 @@ def test_tree_origin_subdir(workspace: object) -> None:
     assert isinstance(workspace, Path)
     _build_sample(workspace)
     out = call_sync(tree, "src")
+    assert out.startswith("- ")
+    assert "- nested/" in out
+    assert "main.py" in out
+    assert "deep.txt" in out
+
+
+def test_tree_default_is_markdown(workspace: object) -> None:
+    assert isinstance(workspace, Path)
+    _build_sample(workspace)
+    out = call_sync(tree, ".")
+    assert out.startswith("- ")
+    assert "- src/" in out
+    assert "├──" not in out
+
+
+def test_tree_markdown_format(workspace: object) -> None:
+    assert isinstance(workspace, Path)
+    _build_sample(workspace)
+    out = call_sync(tree, ".", format="markdown", max_depth=3, max_entries=10)
+    assert "- src/" in out
+    # nested bullets are indented by 2 spaces per level
+    assert "  - main.py" in out
+    assert "  - nested/" in out
+    assert "    - deep.txt" in out
+    assert "more entries not shown" in out
+
+
+def test_tree_flat_format(workspace: object) -> None:
+    assert isinstance(workspace, Path)
+    _build_sample(workspace)
+    out = call_sync(tree, ".", format="flat", max_entries=200)
+    assert "src/" in out  # dirs keep a trailing slash
+    assert "src/main.py" in out
+    assert "src/nested/deep.txt" in out
+    assert "a.txt" in out
+    assert "- " not in out
+    assert "├──" not in out
+
+
+def test_tree_flat_total_cap(workspace: object) -> None:
+    assert isinstance(workspace, Path)
+    _build_sample(workspace)
+    out = call_sync(tree, "pkg", format="flat", max_depth=2, max_entries=10)
+    paths = [line for line in out.splitlines() if not line.startswith("…")]
+    assert len(paths) == 10
+    assert "more paths not shown" in out
+
+
+def test_tree_decorated_format(workspace: object) -> None:
+    assert isinstance(workspace, Path)
+    _build_sample(workspace)
+    out = call_sync(tree, "src", format="decorated")
     assert out.startswith("src/")
+    assert "├──" in out
     assert "main.py" in out
     assert "deep.txt" in out