core/config: document read_file L-range and prettier summaries

CLAUDE.md

@@ -85,13 +85,13 @@ 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`), `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).
+- **`file`**: `read_file` (`with_lineno`, `max_chars`; 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 truncated `read_file`/`fetch` via `TRUNCATE_TOKEN`).
-- **`process`**: `run_command` (argv, no shell, timeout, optional stdin/env); `run_command_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).
+- **`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.
 - Destructive confirms: `classify_danger` + `ToolRegistry(on_confirm=…)`; CLI default deny; config `confirm_destructive` / `path_denylist`. Soft confirm also for shells/REPLs and `python|bash -c` one-liners (argv + code preview); deny may include a user comment for the model. Session YOLO: `/yolo on|off|once` and `--yes` (skip soft confirms; hard denylists unchanged; `once` expires after the next user turn).
 - **`vcs`**: read-only VCS tools (`vcs_kind` / `vcs_status` / `vcs_diff` / `vcs_log` / `vcs_branch`) via `VcsBackend` protocol; **git** implemented; detectors are pluggable for other systems.
-- **`chat`**: human prompts as tools — `ask_user_line` / `ask_user_choice` / `ask_user_form` (shared `prompting` core).
+- **`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.

README.md

@@ -153,7 +153,7 @@ max_context_tokens = 200000
 # enable = ["acme"]
 ```
 
-Per-provider **`timeout`** is passed to the HTTP session for chat/completions, Responses, and `GET /models`. A single number sets one timeout; `{ connect, read }` splits TCP/TLS setup vs idle wait between response bytes (SSE can run longer than `read` while chunks keep arriving). Tool/process timeouts (`run_command`, PTY, policy confirm) are separate.
+Per-provider **`timeout`** is passed to the HTTP session for chat/completions, Responses, and `GET /models`. A single number sets one timeout; `{ connect, read }` splits TCP/TLS setup vs idle wait between response bytes (SSE can run longer than `read` while chunks keep arriving). Tool/process timeouts (`run_argv`, PTY, policy confirm) are separate.
 
 Third-party **plugins**: install a package that declares `project.entry-points."plyngent.tools"` (and later other groups), then allowlist the entry-point name under **`[plugins].enable`**. Details: [doc/plugins.md](doc/plugins.md).
 
@@ -270,7 +270,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_command` / 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`), and todo stack tools (`todo_list` / `todo_push` / `todo_pop` / `todo_update` / `todo_clear`).
+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 truncated results via `TRUNCATE_TOKEN`), and todo stack tools (`todo_list` / `todo_push` / `todo_pop` / `todo_update` / `todo_clear`).
 
 Safety defaults:
 

doc/plugins.md

@@ -159,7 +159,7 @@ Listed tools show **tags** and **catalog source** (`builtin` vs `plugin:…`).
 ## Policy checklist for authors
 
 1. Prefer **LOCAL** for host FS, shell, PTY, or arbitrary network side effects.
-2. Do not shadow builtin names (`read_file`, `run_command`, todo tools, …).
+2. Do not shadow builtin names (`read_file`, `run_argv`, todo tools, …).
 3. Do not rely on process-global workspace/todo alone when a host binds
    instance/session context — prefer the same patterns as builtins
    (`get_workspace_root` already prefers instance policy; session state via

src/plyngent/agent/budget.py

@@ -50,7 +50,7 @@ def truncate_tool_result(
         return (
             f"{text[:max_chars]}\n"
             f"...[Truncated ({max_chars} chars max.; {omitted} chars omitted). "
-            f"You may ask for user to increase the limit via configuration file.]"
+            f"Use `get_truncated` when a `TRUNCATE_TOKEN` is shown.]"
         )
     return f"{text[:max_chars]}\n...[truncated {omitted} characters]"
 

src/plyngent/agent/directive_checkpoint.py

@@ -17,12 +17,15 @@ DEFAULT_DIRECTIVE_REMINDER_TOKENS = 100_000
 
 DEFAULT_DIRECTIVE_REMINDER_TEXT = """\
 Tool playbook still applies (see system). Hard constraints:
-- Prefer file tools over shell; several `run_command` calls may run in parallel; \
+- Prefer file tools over shell; several `run_argv` calls may run in parallel; \
-use `run_command_batch` for ordered pipelines.
+use `run_argv_batch` for ordered pipelines.
 - `edit_replace`: fix match / `max_replaces`; `edit_lineno` edits only lines read via \
 `read_file(with_lineno=true)`, and line numbers reset after any file write.
+- `read_file` results start with `L{begin}-{end}` (1-based lines; offset is 0-based).
+- Truncated results carry a `TRUNCATE_TOKEN`; continue with `get_truncated`.
 - Prefer `fetch` for HTTP(S); private/LAN hosts need human policy allow (not YOLO).
 - PTY secrets only via `ask_into_pty`; denylists and confirms still apply.
+- `wait` pauses; pressing Enter disturbs early (optional reason).
 - Todo stack: open items mean unfinished work.
 """
 

src/plyngent/agent/responses_bridge.py

@@ -171,7 +171,7 @@ def response_to_assistant_message(response: Response) -> AssistantChatMessage:
             )
             for call in calls
         ]
-    reasoning = _reasoning_summary_text(response)
+    reasoning = reasoning_summary_text(response)
     return AssistantChatMessage(
         content=text or None,
         tool_calls=tool_calls,
@@ -179,23 +179,40 @@ def response_to_assistant_message(response: Response) -> AssistantChatMessage:
     )
 
 
-def _reasoning_summary_text(response: Response) -> str:
+def _reasoning_text_blocks(raw: object) -> list[str]:
+    """Collect text from a reasoning block list (``summary``/``content``)."""
+    parts: list[str] = []
+    if not isinstance(raw, list):
+        return parts
+    for block_obj in cast("list[object]", raw):
+        if not isinstance(block_obj, dict):
+            continue
+        block_map = cast("dict[str, object]", block_obj)
+        if block_map.get("type") in {"summary_text", "output_text", "reasoning_content"}:
+            text = block_map.get("text")
+            if isinstance(text, str) and text:
+                parts.append(text)
+    return parts
+
+
+def reasoning_summary_text(response: Response) -> str:
+    """Concatenate reasoning text from the response.
+
+    OpenAI puts ``reasoning`` items in ``output`` with a ``summary`` block list.
+    DeepSeek (Responses convention) additionally returns the full chain-of-thought
+    in the top-level ``response.reasoning`` (``content``/``summary`` block lists);
+    prefer it when present so the two sources are not duplicated.
+    """
+    reasoning = response.reasoning
+    if isinstance(reasoning, dict):
+        parts = _reasoning_text_blocks(reasoning.get("content")) + _reasoning_text_blocks(reasoning.get("summary"))
+        if parts:
+            return "".join(parts)
     parts: list[str] = []
     for raw in response.output:
         if raw.get("type") != "reasoning":
             continue
-        summary = raw.get("summary")
+        parts.extend(_reasoning_text_blocks(raw.get("summary")))
-        if not isinstance(summary, list):
-            continue
-        summary_items = cast("list[object]", summary)
-        for block_obj in summary_items:
-            if not isinstance(block_obj, dict):
-                continue
-            block_map = cast("dict[str, object]", block_obj)
-            if block_map.get("type") in {"summary_text", "output_text"}:
-                text = block_map.get("text")
-                if isinstance(text, str) and text:
-                    parts.append(text)
     return "".join(parts)
 
 

src/plyngent/agent/responses_dispatch.py

@@ -15,6 +15,7 @@ from msgspec import UNSET
 
 from plyngent.agent.responses_bridge import (
     chat_param_to_responses_kwargs,
+    reasoning_summary_text,
     response_to_assistant_message,
     response_to_chat_completion,
     responses_status_to_finish_reason,
@@ -45,7 +46,14 @@ if TYPE_CHECKING:
     )
 
 # Responses stream event types that carry reasoning text deltas.
-_REASONING_DELTA_TYPES = {"response.reasoning_summary_text.delta", "response.reasoning_text.delta"}
+# OpenAI streams ``response.reasoning_summary_text.delta``; DeepSeek streams
+# ``response.reasoning_text.delta`` and emits the full text once via
+# ``response.reasoning_text.done``.
+_REASONING_DELTA_TYPES = {
+    "response.reasoning_summary_text.delta",
+    "response.reasoning_text.delta",
+    "response.reasoning_text.done",
+}
 # Terminal events that carry the full Response object (status, tool calls, usage).
 # ``response.failed`` also carries a Response but is surfaced as an error.
 _FINAL_EVENT_TYPES = {"response.completed", "response.incomplete"}
@@ -90,6 +98,27 @@ def _decode_final_event(event: ResponseStreamEvent) -> ResponseModel | None:
         return None
 
 
+def _stream_reasoning_content(event: ResponseStreamEvent) -> str | None:
+    """Return reasoning text from a reasoning stream event (delta or full-text done)."""
+    if event.type not in _REASONING_DELTA_TYPES:
+        return None
+    content = event.delta if isinstance(event.delta, str) else event.text
+    return content if isinstance(content, str) and content else None
+
+
+def _terminal_chunks(final: ResponseModel, *, model: str) -> list[ChatCompletionChunk]:
+    """Build the post-stream chunks (tool calls, finish_reason, usage)."""
+    assistant = response_to_assistant_message(final)
+    has_tools = assistant.tool_calls is not UNSET and bool(assistant.tool_calls)
+    finish = responses_status_to_finish_reason(final, has_tool_calls=has_tools)
+    chunks: list[ChatCompletionChunk] = list(tool_call_chunks_from_response(final, model=model))
+    chunks.append(finish_reason_chunk(model=model, finish_reason=finish))
+    usage = usage_chunk_from_response(final, model=model)
+    if usage is not None:
+        chunks.append(usage)
+    return chunks
+
+
 async def _stream_as_chat_chunks(
     client: OpenAIClient,
     create: ResponsesCreateParam,
@@ -108,6 +137,7 @@ async def _stream_as_chat_chunks(
     """
     stream = await client.responses(create, stream=True)
     final: ResponseModel | None = None
+    saw_reasoning = False
     async for event in stream:
         etype = event.type
         if etype in {"error", "response.failed"}:
@@ -118,8 +148,10 @@ async def _stream_as_chat_chunks(
         if etype == "response.output_text.delta" and isinstance(event.delta, str) and event.delta:
             yield text_delta_chunk(model=model, content=event.delta)
             continue
-        if etype in _REASONING_DELTA_TYPES and isinstance(event.delta, str) and event.delta:
+        reasoning = _stream_reasoning_content(event)
-            yield reasoning_delta_chunk(model=model, content=event.delta)
+        if reasoning is not None:
+            saw_reasoning = True
+            yield reasoning_delta_chunk(model=model, content=reasoning)
             continue
 
     if final is None:
@@ -127,15 +159,16 @@ async def _stream_as_chat_chunks(
         # The loop will treat empty + no finish_reason as a glitch.
         return
 
-    assistant = response_to_assistant_message(final)
+    if not saw_reasoning:
-    has_tools = assistant.tool_calls is not UNSET and bool(assistant.tool_calls)
+        # DeepSeek sometimes delivers reasoning only on the terminal response
-    finish = responses_status_to_finish_reason(final, has_tool_calls=has_tools)
+        # (top-level ``reasoning``); surface it so tool-call turns keep their
-    for chunk in tool_call_chunks_from_response(final, model=model):
+        # reasoning_content when the assistant message goes back to the API.
+        terminal_reasoning = reasoning_summary_text(final)
+        if terminal_reasoning:
+            yield reasoning_delta_chunk(model=model, content=terminal_reasoning)
+
+    for chunk in _terminal_chunks(final, model=model):
         yield chunk
-    yield finish_reason_chunk(model=model, finish_reason=finish)
-    usage = usage_chunk_from_response(final, model=model)
-    if usage is not None:
-        yield usage
 
 
 async def dispatch_responses(

src/plyngent/cli/display.py

@@ -1,10 +1,11 @@
 from __future__ import annotations
 
 import contextlib
+import json
 import os
 import sys
 from contextvars import ContextVar
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Literal, cast
 
 import click
 
@@ -28,6 +29,10 @@ if TYPE_CHECKING:
 _TOOL_RESULT_PREVIEW = 120
 _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"})
+
 # Process/session display flags (set from ReplState / slash).
 _verbose_tool_results: ContextVar[bool] = ContextVar("verbose_tool_results", default=False)
 _markdown_enabled: ContextVar[bool] = ContextVar("markdown_enabled", default=True)
@@ -69,6 +74,49 @@ 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."""
+    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 None
+
+
+def _read_file_pretty(args_json: str, result: str) -> str:
+    """One-line summary for a ``read_file`` call: ``* Read 'path' L1-4 (done)``."""
+    path = _json_str_arg(args_json, "path") or "?"
+    if result.startswith("error: file not found"):
+        return f"* Read '{path}' (file not found)"
+    if result.startswith("error: not a file"):
+        return f"* Read '{path}' (not a file)"
+    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)"
+
+
+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}"
+
+
+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
+
+
 def _echo_stream(text: str) -> None:
     """Write without newline and flush so assistant text appears token-by-token."""
     click.echo(text, nl=False)
@@ -144,6 +192,9 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
     assistant_buf: list[str] = []
     printed_reasoning = False
     printed_assistant = False
+    # Tool calls buffer so prettified tools can render once their result lands
+    # (the summary line needs the status/range). FIFO matches loop event order.
+    pending_tools: list[tuple[str, str, bool]] = []
 
     def flush_assistant() -> None:
         nonlocal source, assistant_buf, printed_assistant
@@ -192,14 +243,24 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
             flush_assistant()
             call = event.tool_call
             if isinstance(call, AssistantFunctionToolCall):
-                args = _preview(call.function.arguments, _TOOL_ARGS_PREVIEW)
+                name = call.function.name
-                click.secho(f"\n[tool] {call.function.name}({args})", fg="yellow")
+                args = call.function.arguments
+                pretty_tool = name in _PRETTY_TOOLS
+                pending_tools.append((name, args, pretty_tool))
+                if not pretty_tool:
+                    preview = _preview(args, _TOOL_ARGS_PREVIEW)
+                    click.secho(f"\n[tool] {name}({preview})", fg="yellow")
             else:
+                pending_tools.append(("custom", call.id, False))
                 click.secho(f"\n[tool] custom id={call.id}", fg="yellow")
         elif isinstance(event, ToolResultEvent):
             flush_assistant()
             content = event.message.content
-            if show_full:
+            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")
+            elif show_full:
                 click.secho(f"[tool ok]\n{content}", fg="magenta")
             else:
                 preview = _preview(content, _TOOL_RESULT_PREVIEW)

src/plyngent/config/models.py

@@ -32,10 +32,14 @@ raise `max_replaces` or narrow `old_string`.
 - `edit_lineno` edits only lines you read via `read_file(path, with_lineno=true)`; \
 re-read with line numbers after any `edit_*` / `write_file` / copy/move/delete, \
 since line numbers go stale.
+- `read_file` results start with a 1-based line range `L{begin}-{end}` (`offset` \
+is 0-based); `with_lineno` shows per-line numbers instead.
+- Truncated `read_file` / `fetch` results carry a `TRUNCATE_TOKEN`; use \
+`get_truncated` with it to keep reading (chains through truncations).
 
 ### Commands
-- Prefer `run_command` / `run_command_batch` (argv, no shell) over `bash -c` or similar.
+- Prefer `run_argv` / `run_argv_batch` (argv lists, no shell) over `bash -c` or similar.
-- Several `run_command` calls in one step may run in parallel; use `run_command_batch` \
+- Several `run_argv` calls in one step may run in parallel; use `run_argv_batch` \
 for ordered pipelines (`pipe_out` / `mix_stderr` as needed).
 - Prefer `vcs_*` for status/diff/log/branch when enough.
 
@@ -49,6 +53,7 @@ otherwise a small default is used. Never rely on shell to spoof identity.
 
 ### Humans & safety
 - Prefer `ask_user_line` / `ask_user_choice` / `ask_user_form` over waiting for the next free-form turn.
+- Use `wait` to pause; pressing Enter during the wait disturbs it (reason optional; turn continues immediately).
 - Overwrites, deletes, shells, and risky ops may require human confirm; hard denylists are not skipped by YOLO.
 - Temporary scratch: `new_temporary_workspace`.
 

src/plyngent/prompting.py

@@ -2,10 +2,12 @@ from __future__ import annotations
 
 import asyncio
 import contextlib
+import select
 import sys
+import time
 from contextlib import AbstractContextManager
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Protocol, runtime_checkable
+from typing import TYPE_CHECKING, Protocol, TextIO, cast, runtime_checkable
 
 import click
 
@@ -98,6 +100,87 @@ def _readline_input(prompt: str, *, completions: Sequence[str] | None = None) ->
             readline.set_completer_delims(previous_delims)
 
 
+type PollFn = Callable[
+    [list[object], list[object], list[object], float],
+    tuple[list[object], list[object], list[object]],
+]
+
+
+def _read_line_with_timeout_posix(
+    prompt: str,
+    timeout: float,
+    *,
+    stream: TextIO,
+    poll: PollFn,
+) -> str | None:
+    """Blocking line read on POSIX: poll stdin, then read one line on data.
+
+    Returns the line (without trailing newline), ``""`` for an empty Enter, or
+    ``None`` when the timeout elapses before any line is complete (EOF too).
+    """
+    _ = sys.stdout.write(prompt)
+    _ = sys.stdout.flush()
+    deadline = time.monotonic() + timeout
+    while True:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            return None
+        ready, _, _ = poll([stream], [], [], min(remaining, 0.1))
+        if not ready:
+            continue
+        line = stream.readline()
+        if line == "":
+            return None  # EOF
+        return line.rstrip("\n")
+
+
+def _read_line_with_timeout_windows(prompt: str, timeout: float) -> str | None:
+    """Blocking line read on Windows via msvcrt char polling (best-effort)."""
+    import msvcrt
+
+    _ = sys.stdout.write(prompt)
+    _ = sys.stdout.flush()
+    deadline = time.monotonic() + timeout
+    buf: list[str] = []
+    while time.monotonic() < deadline:
+        if msvcrt.kbhit():
+            ch = msvcrt.getwch()
+            if ch in ("\r", "\n"):
+                _ = sys.stdout.write("\n")
+                _ = sys.stdout.flush()
+                return "".join(buf)
+            if ch == "\x03":
+                raise KeyboardInterrupt
+            if ch in ("\x08", "\x7f") and buf:  # backspace / DEL
+                _ = buf.pop()
+                _ = sys.stdout.write("\b \b")
+                _ = sys.stdout.flush()
+                continue
+            buf.append(ch)
+            _ = sys.stdout.write(ch)
+            _ = sys.stdout.flush()
+        else:
+            time.sleep(0.05)
+    _ = sys.stdout.write("\n")
+    _ = sys.stdout.flush()
+    return None
+
+
+def read_line_with_timeout(prompt: str, timeout: float) -> str | None:
+    """Read one line, blocking at most ``timeout`` seconds.
+
+    Returns the typed line (without trailing newline), ``""`` for an empty
+    Enter, or ``None`` when the timeout elapses first (EOF too). Ctrl+C
+    propagates as :class:`KeyboardInterrupt`. POSIX polls stdin via ``select``;
+    Windows uses ``msvcrt`` char polling (best-effort).
+    """
+    if timeout < 0:
+        timeout = 0.0
+    if sys.platform == "win32":
+        return _read_line_with_timeout_windows(prompt, timeout)
+    return _read_line_with_timeout_posix(prompt, timeout, stream=sys.stdin, poll=cast("PollFn", select.select))
+
+
 class ClickPromptBackend:
     """Click/TTY backend for interactive prompts (readline + Tab when available)."""
 

src/plyngent/tools/__init__.py

@@ -8,6 +8,7 @@ from .chat import CHAT_TOOLS as CHAT_TOOLS
 from .chat import ask_user as ask_user
 from .chat import choose_user as choose_user
 from .chat import form_user as form_user
+from .chat import wait as wait
 from .context import InstanceState as InstanceState
 from .context import SessionState as SessionState
 from .context import bind_instance as bind_instance
@@ -41,8 +42,8 @@ from .process import ask_into_pty as ask_into_pty
 from .process import close_pty as close_pty
 from .process import open_pty as open_pty
 from .process import read_pty as read_pty
-from .process import run_command as run_command
+from .process import run_argv as run_argv
-from .process import run_command_batch as run_command_batch
+from .process import run_argv_batch as run_argv_batch
 from .process import write_pty as write_pty
 from .process import write_pty_keys as write_pty_keys
 from .temp_workspace import cleanup_temporary_workspaces as cleanup_temporary_workspaces

src/plyngent/tools/chat/__init__.py

@@ -1,5 +1,6 @@
 from .ask import ask_user as ask_user
 from .choose import choose_user as choose_user
 from .form import form_user as form_user
+from .wait import wait as wait
 
-CHAT_TOOLS = [ask_user, choose_user, form_user]
+CHAT_TOOLS = [ask_user, choose_user, form_user, wait]

src/plyngent/tools/chat/wait.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+import asyncio
+
+from plyngent.agent import ToolTag, tool
+from plyngent.prompting import get_prompt_backend, read_line_with_timeout, run_prompt_async
+
+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)
+async def wait(duration: int, *, reason: str | None = None) -> str:
+    """Wait ``duration`` seconds before continuing.
+
+    Interactive sessions show a line prompt: pressing Enter (optionally after
+    typing a reason) "disturbs" the wait so the turn continues immediately.
+    Non-interactive runs simply sleep the full duration.
+    """
+    if duration < 0:
+        return "error: duration must not be negative"
+    if duration == 0:
+        return "waited 0s"
+    backend = get_prompt_backend()
+    if not backend.is_interactive():
+        await asyncio.sleep(duration)
+        return f"waited {duration}s"
+    prompt = WAIT_PROMPT.format(duration=duration)
+    if reason:
+        prompt = (
+            f"Waiting for {duration}s (reason: {reason}). Press Enter to disturb. "
+            "Feel free to ship any reason before pressing Enter. "
+        )
+    line = await run_prompt_async(read_line_with_timeout, prompt, float(duration))
+    if line is None:
+        return f"waited {duration}s"
+    text = line.strip()
+    if text:
+        return f"disturbed by user: {text}"
+    return "disturbed by user (no reason)"

src/plyngent/tools/danger.py

@@ -59,8 +59,8 @@ def _basename(argv0: str) -> str:
     return name.lower()
 
 
-def _as_argv(args: Mapping[str, object]) -> list[str] | None:
+def _as_argv(args: Mapping[str, object], *, key: str = "command") -> list[str] | None:
-    command = args.get("command")
+    command = args.get(key)
     if not isinstance(command, list) or not command:
         return None
     out: list[str] = []
@@ -110,7 +110,7 @@ def _shell_or_dash_c_reason(argv: Sequence[str], *, via: str) -> str | None:
     """Confirm interactive shells and ``*-c`` one-liners so the user can inspect argv.
 
     Multi-line reason (shown inside the CLI confirm box). ``via`` is a short
-    label such as ``run_command`` or ``open_pty``.
+    label such as ``run_argv`` or ``open_pty``.
 
     For ``-c`` scripts, argv shows ``$(command)`` instead of inlining the body;
     the full script is printed below untruncated, with every line indented.
@@ -181,31 +181,31 @@ def _delete_path_reason(args: Mapping[str, object]) -> str | None:
     return f"delete path {path!r}{extra}"
 
 
-def _run_command_reason(args: Mapping[str, object]) -> str | None:
+def _run_argv_reason(args: Mapping[str, object]) -> str | None:
-    argv = _as_argv(args)
+    argv = _as_argv(args, key="argv")
     if argv is None:
         return None
-    return _shell_or_dash_c_reason(argv, via="run_command")
+    return _shell_or_dash_c_reason(argv, via="run_argv")
 
 
 def _batch_step_argv(item: object) -> list[str] | None:
     if not isinstance(item, dict):
         return None
     step = cast("dict[str, object]", item)
-    command = step.get("command")
+    argv = step.get("argv")
-    if not isinstance(command, list) or not command:
+    if not isinstance(argv, list) or not argv:
         return None
-    argv: list[str] = []
+    parts: list[str] = []
-    for part in cast("list[object]", command):
+    for part in cast("list[object]", argv):
         if not isinstance(part, str):
             return None
-        argv.append(part)
+        parts.append(part)
-    return argv or None
+    return parts or None
 
 
-def _run_command_batch_reason(args: Mapping[str, object]) -> str | None:
+def _run_argv_batch_reason(args: Mapping[str, object]) -> str | None:
     """One confirm for the whole batch if any step is shell/REPL/-c."""
-    raw = args.get("commands")
+    raw = args.get("steps")
     if isinstance(raw, str):
         try:
             loaded: object = json.loads(raw)
@@ -218,11 +218,11 @@ def _run_command_batch_reason(args: Mapping[str, object]) -> str | None:
         reason
         for index, item in enumerate(cast("list[object]", raw))
         if (argv := _batch_step_argv(item)) is not None
-        and (reason := _shell_or_dash_c_reason(argv, via=f"run_command_batch[{index}]")) is not None
+        and (reason := _shell_or_dash_c_reason(argv, via=f"run_argv_batch[{index}]")) is not None
     ]
     if not risky:
         return None
-    header = f"run_command_batch: {len(risky)} risky step(s) (review before allow)"
+    header = f"run_argv_batch: {len(risky)} risky step(s) (review before allow)"
     return header + "\n" + "\n".join(risky)
 
 
@@ -262,10 +262,10 @@ def classify_danger(name: str, args: Mapping[str, object]) -> str | None:  # noq
         return _copy_path_reason(args)
     if name == "write_file":
         return _write_file_reason(args)
-    if name == "run_command":
+    if name == "run_argv":
-        return _run_command_reason(args)
+        return _run_argv_reason(args)
-    if name == "run_command_batch":
+    if name == "run_argv_batch":
-        return _run_command_batch_reason(args)
+        return _run_argv_batch_reason(args)
     if name == "open_pty":
         return _open_pty_reason(args)
     if name == "fetch":

src/plyngent/tools/file/__init__.py

@@ -5,6 +5,7 @@ from .edit_replace import edit_replace as edit_replace
 from .fs_ops import copy_path as copy_path
 from .fs_ops import delete_path as delete_path
 from .fs_ops import move_path as move_path
+from .get_truncated import get_truncated as get_truncated
 from .glob_paths import glob_paths as glob_paths
 from .grep_files import grep_files as grep_files
 from .listdir import listdir as listdir
@@ -25,4 +26,5 @@ FILE_TOOLS = [
     move_path,
     delete_path,
     new_temporary_workspace,
+    get_truncated,
 ]

src/plyngent/tools/file/get_truncated.py

@@ -0,0 +1,46 @@
+"""``get_truncated`` tool: resume truncated tool results via a truncate token."""
+
+from __future__ import annotations
+
+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.net.fetch import fetch
+from plyngent.tools.truncate_token import decode_truncate_token, truncate_with_token
+
+
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
+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``, 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``.
+    """
+    parsed = decode_truncate_token(token)
+    if parsed is None:
+        return "error: invalid truncate token"
+    if parsed.kind == "http":
+        return await fetch.handler(parsed.location, offset=parsed.offset, max_chars=parsed.limit)
+    text, err = read_raw_text(parsed.location)
+    if err:
+        return err
+    text = cast("str", text)
+    start = parsed.offset
+    segment = text[start : start + max_chars]
+    chunk, _ = truncate_with_token(
+        segment,
+        max_chars,
+        kind="file",
+        location=parsed.location,
+        offset=start,
+        limit=max_chars,
+        total_len=len(text),
+    )
+    content_len = len(chunk.split("\n...[", 1)[0]) if "\n...[" in chunk else len(chunk)
+    header = line_range_label(text, start, start + content_len)
+    return f"{header}\n{chunk}"

src/plyngent/tools/file/read.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+from typing import cast
+
 from plyngent.agent import ToolTag, mark_lineno_read, tool
+from plyngent.tools.truncate_token import truncate_with_token
 from plyngent.tools.workspace import resolve_path
 
 _LINENO_WIDTH = 6
@@ -17,6 +20,34 @@ def _format_with_lineno(lines: list[str], *, start_lineno: int) -> str:
     return "".join(out)
 
 
+def line_range_label(text: str, start_char: int, end_char: int) -> str:
+    """1-based inclusive line range covering ``text[start_char:end_char]``.
+
+    Char offsets do not map 1:1 to lines; this converts a contiguous char range
+    into the 1-based file line numbers it spans (offset is 0-based, so line 1
+    starts at char 0).
+    """
+    begin = text.count("\n", 0, start_char) + 1
+    last = max(start_char, end_char - 1)
+    end = text.count("\n", 0, last) + 1
+    return f"L{begin}-{end}"
+
+
+def read_raw_text(path: str) -> tuple[str | None, str]:
+    """(text, error) for a workspace file read; ``error`` is ``""`` on success.
+
+    Shared by ``read_file`` and ``get_truncated`` so truncate-token char offsets
+    always refer to the raw file text (never the ``L{begin}-{end}`` header).
+    Missing paths and directories are distinguished for the caller.
+    """
+    target = resolve_path(path)
+    if not target.exists():
+        return None, f"error: file not found: {path}"
+    if not target.is_file():
+        return None, f"error: not a file: {path}"
+    return target.read_text(encoding="utf-8", errors="replace"), ""
+
+
 @tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
 async def read_file(
     path: str,
@@ -24,18 +55,26 @@ async def read_file(
     offset: int = 0,
     limit: int | None = None,
     with_lineno: bool = False,
+    max_chars: int | None = None,
 ) -> str:
     """Read a text file under the workspace.
 
-    ``offset`` is 0-based line start; ``limit`` is max lines (None = rest of file).
+    ``offset`` is 0-based line start (0 = first line); ``limit`` is max lines
-    When ``with_lineno`` is true, each line is prefixed with its 1-based file line
+    (None = rest of file). When ``with_lineno`` is true, each line is prefixed
-    number (``     N|…``), matching ``edit_lineno`` numbering, and those lines are
+    with its 1-based file line number (``     N|…``), matching ``edit_lineno``
-    marked readable for ``edit_lineno`` this turn.
+    numbering, and those lines are marked readable for ``edit_lineno`` this turn.
+
+    ``max_chars`` caps the returned slice; a ``TRUNCATE_TOKEN`` is appended so
+    ``get_truncated`` can continue reading the rest without a new request.
+
+    The result starts with a 1-based inclusive line range ``L{begin}-{end}`` so
+    the caller knows exactly which file lines were read (offset is 0-based).
     """
     target = resolve_path(path)
-    if not target.is_file():
+    text, err = read_raw_text(path)
-        return f"error: not a file: {path}"
+    if err:
-    text = target.read_text(encoding="utf-8", errors="replace")
+        return err
+    text = cast("str", text)
     lines = text.splitlines(keepends=True)
     if offset < 0:
         return "error: offset must be >= 0"
@@ -46,5 +85,24 @@ async def read_file(
     slice_lines = lines[start:end]
     if with_lineno:
         mark_lineno_read(str(target), set(range(start + 1, end + 1)))
-        return _format_with_lineno(slice_lines, start_lineno=start + 1)
+        body = _format_with_lineno(slice_lines, start_lineno=start + 1)
-    return "".join(slice_lines)
+    else:
+        body = "".join(slice_lines)
+    if max_chars is not None and max_chars >= 1:
+        char_start = len("".join(lines[:start]))
+        body, _ = truncate_with_token(
+            body,
+            max_chars,
+            kind="file",
+            location=path,  # arg form: short token, resolves to the same file
+            offset=char_start,
+            limit=max_chars,
+            total_len=len(text),
+        )
+    if not body:
+        return ""
+    if with_lineno:
+        return body  # per-line numbers already show the range
+    begin = start + 1
+    end_line = start + len(slice_lines)
+    return f"L{begin}-{end_line}\n{body}"

src/plyngent/tools/net/fetch.py

@@ -16,6 +16,7 @@ from plyngent.tools.net.policy import (
     parse_fetch_url,
     soft_confirm_reason,
 )
+from plyngent.tools.truncate_token import truncate_with_token
 
 
 def format_fetch_result(
@@ -32,13 +33,20 @@ def format_fetch_result(
     warnings: list[str],
     body_kind: str,
     max_chars: int,
+    offset: int = 0,
 ) -> str:
-    text = body_text
+    start = max(0, offset)
-    char_truncated = False
+    body = body_text[start:]
-    if max_chars >= 1 and len(text) > max_chars:
+    body, token = truncate_with_token(
-        omitted = len(text) - max_chars
+        body,
-        text = text[:max_chars] + f"\n...[truncated {omitted} characters]"
+        max_chars,
-        char_truncated = True
+        kind="http",
+        location=final_url,
+        offset=start,
+        limit=max_chars,
+        total_len=len(body_text),
+    )
+    char_truncated = token is not None
     warn_line = "; ".join(warnings) if warnings else ""
     parts = [
         f"status={status}",
@@ -54,7 +62,7 @@ def format_fetch_result(
     if warn_line:
         parts.append(f"warnings={warn_line}")
     parts.append("--- body ---")
-    parts.append(text)
+    parts.append(body)
     return "\n".join(parts)
 
 
@@ -72,6 +80,7 @@ async def fetch(
     max_redirects: int = DEFAULT_MAX_REDIRECTS,
     follow_redirects: bool = True,
     allow_http_downgrade: bool = False,
+    offset: int = 0,
 ) -> str:
     """HTTP request (GET/POST/PUT/DELETE); return status, metadata, and truncated body.
 
@@ -122,6 +131,7 @@ async def fetch(
         warnings=result.warnings,
         body_kind=result.body_kind,
         max_chars=max_chars,
+        offset=offset,
     )
 
 

src/plyngent/tools/process/__init__.py

@@ -4,14 +4,14 @@ from .open_pty import open_pty as open_pty
 from .pty_terminal import decode_write_data as decode_write_data
 from .pty_terminal import sanitize_pty_output_for_tool as sanitize_pty_output_for_tool
 from .read_pty import read_pty as read_pty
-from .run_command import run_command as run_command
+from .run_argv import run_argv as run_argv
-from .run_command_batch import run_command_batch as run_command_batch
+from .run_argv_batch import run_argv_batch as run_argv_batch
 from .write_pty import write_pty as write_pty
 from .write_pty_keys import write_pty_keys as write_pty_keys
 
 PROCESS_TOOLS = [
-    run_command,
+    run_argv,
-    run_command_batch,
+    run_argv_batch,
     open_pty,
     read_pty,
     write_pty,

src/plyngent/tools/process/command_exec.py

@@ -83,7 +83,7 @@ async def execute_argv(
     timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
     mix_stderr: bool = False,
 ) -> CommandStepResult:
-    """Run one argv command; no shell. Used by run_command and run_command_batch."""
+    """Run one argv command; no shell. Used by run_argv and run_argv_batch."""
     if not command:
         msg = "command argv must not be empty"
         raise WorkspaceError(msg)

src/plyngent/tools/process/run_argv.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from plyngent.agent import ToolTag, tool
-from plyngent.tools.workspace import WorkspaceError
+from plyngent.tools.workspace import WorkspaceError, argv_shape_error
 
 from .command_exec import (
     DEFAULT_TIMEOUT_SECONDS,
@@ -11,18 +11,21 @@ from .command_exec import (
 
 
 @tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO)
-async def run_command(
+async def run_argv(
-    command: list[str],
+    argv: list[str],
     *,
     cwd: str = ".",
     timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
     stdin: str | None = None,
     env: dict[str, str] | None = None,
 ) -> str:
-    """Run ``command`` (argv, no shell) under the workspace; capture stdout/stderr."""
+    """Run ``argv`` (argv, no shell) under the workspace; capture stdout/stderr."""
+    shape_error = argv_shape_error(argv)
+    if shape_error is not None:
+        return f"error: `run_argv` {shape_error}"
     try:
         result = await execute_argv(
-            command,
+            argv,
             cwd=cwd,
             env=env,
             stdin=stdin,

src/plyngent/tools/process/run_argv_batch.py

@@ -50,23 +50,23 @@ def _as_str_dict(value: object) -> dict[str, str] | None:
 
 def _parse_step(raw: object, index: int) -> dict[str, Any]:
     if not isinstance(raw, dict):
-        msg = f"commands[{index}] must be an object"
+        msg = f"steps[{index}] must be an object"
         raise WorkspaceError(msg)
     data = cast("dict[str, object]", raw)
-    command = data.get("command")
+    argv = data.get("argv")
-    if not isinstance(command, list) or not command:
+    if not isinstance(argv, list) or not argv:
-        msg = f"commands[{index}].command must be a non-empty argv list"
+        msg = f"steps[{index}].argv must be a non-empty argv list"
         raise WorkspaceError(msg)
-    command_parts = cast("list[object]", command)
+    argv_parts = cast("list[object]", argv)
-    argv = [part for part in command_parts if isinstance(part, str)]
+    argv_str = [part for part in argv_parts if isinstance(part, str)]
-    if len(argv) != len(command_parts):
+    if len(argv_str) != len(argv_parts):
-        msg = f"commands[{index}].command must be a list of strings"
+        msg = f"steps[{index}].argv must be a list of strings"
         raise WorkspaceError(msg)
     cwd = data.get("cwd")
     stdin = data.get("stdin")
     stop = data.get("stop_on_error", None)
     return {
-        "command": argv,
+        "argv": argv_str,
         "cwd": str(cwd) if isinstance(cwd, str) and cwd else None,
         "env": _as_str_dict(data.get("env")),
         "stdin": None if stdin is None else str(stdin),
@@ -77,18 +77,18 @@ def _parse_step(raw: object, index: int) -> dict[str, Any]:
     }
 
 
-def _normalize_commands(commands: list[dict[str, object]] | str) -> list[object]:
+def _normalize_steps(steps: list[dict[str, object]] | str) -> list[object]:
-    if isinstance(commands, str):
+    if isinstance(steps, str):
         try:
-            loaded: object = json.loads(commands)
+            loaded: object = json.loads(steps)
         except json.JSONDecodeError as exc:
-            msg = f"commands must be a JSON array: {exc}"
+            msg = f"steps must be a JSON array: {exc}"
             raise WorkspaceError(msg) from exc
         if not isinstance(loaded, list):
-            msg = "commands must be a JSON array of step objects"
+            msg = "steps must be a JSON array of step objects"
             raise WorkspaceError(msg)
         return cast("list[object]", loaded)
-    return cast("list[object]", commands)
+    return cast("list[object]", steps)
 
 
 def _format_step(index: int, result: CommandStepResult, *, pipe_out: bool) -> str:
@@ -131,7 +131,7 @@ async def _run_batch_steps(
     for index, step in enumerate(steps):
         stdin_text = prev_capture if prev_piped else step["stdin"]
         result = await execute_argv(
-            step["command"],
+            step["argv"],
             cwd=step["cwd"] if step["cwd"] is not None else cwd,
             env=_merge_env(env, step["env"]),
             stdin=stdin_text,
@@ -155,8 +155,8 @@ async def _run_batch_steps(
 
 
 @tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO)
-async def run_command_batch(
+async def run_argv_batch(
-    commands: list[dict[str, object]] | str,
+    steps: list[dict[str, object]] | str,
     *,
     cwd: str = ".",
     env: dict[str, str] | None = None,
@@ -164,10 +164,10 @@ async def run_command_batch(
 ) -> str:
     """Run a serial batch of argv commands (no shell).
 
-    Each element of ``commands`` is an object::
+    Each element of ``steps`` is an object::
 
         {
-          "command": ["git", "status"],   # required argv
+          "argv": ["git", "status"],     # required argv
           "cwd": ".",                     # optional, else batch cwd
           "env": {"FOO": "1"},            # optional overlay
           "stdin": "...",                 # optional; unused if previous pipe_out
@@ -181,15 +181,15 @@ async def run_command_batch(
     Default ``stop_on_error`` is true. Max 20 steps; total output budget 32k chars.
     """
     try:
-        raw_steps = _normalize_commands(commands)
+        raw_steps = _normalize_steps(steps)
         if not raw_steps:
-            return "error: commands must not be empty"
+            return "error: steps must not be empty"
         if len(raw_steps) > DEFAULT_MAX_BATCH_STEPS:
-            return f"error: at most {DEFAULT_MAX_BATCH_STEPS} commands per batch"
+            return f"error: at most {DEFAULT_MAX_BATCH_STEPS} steps per batch"
 
         steps = [_parse_step(item, i) for i, item in enumerate(raw_steps)]
         if steps[-1]["pipe_out"]:
-            return "error: last command has pipe_out=true but there is no next step"
+            return "error: last step has pipe_out=true but there is no next step"
 
         results, stopped_early = await _run_batch_steps(steps, cwd=cwd, env=env, stop_on_error=stop_on_error)
         body = "\n".join(

src/plyngent/tools/truncate_token.py

@@ -0,0 +1,96 @@
+"""Truncate-token cursors for resuming truncated tool results.
+
+``fetch`` and ``read_file`` append a ``[TRUNCATE_TOKEN: ...]`` marker when they
+cut output short; the model passes the token to ``get_truncated`` to read the
+next chunk without re-requesting the whole source or raising limits.
+"""
+
+from __future__ import annotations
+
+import base64
+from typing import Literal
+
+import msgspec
+
+type TruncateKind = Literal["file", "http"]
+
+
+class TruncateToken(msgspec.Struct, frozen=True):
+    """Opaque cursor into a truncated source.
+
+    ``offset``/``limit`` are in characters (http bodies) or lines (files is
+    line-based via ``read_file``); ``get_truncated`` interprets by ``kind``.
+    """
+
+    kind: TruncateKind
+    location: str
+    offset: int
+    limit: int
+
+
+def encode_truncate_token(token: TruncateToken) -> str:
+    """Base64url-encode a token so it is safe inside tool-result text."""
+    payload = msgspec.json.encode(token)
+    return base64.urlsafe_b64encode(payload).decode("ascii")
+
+
+def decode_truncate_token(text: str) -> TruncateToken | None:
+    """Decode a token; None when malformed (model garbage or tampering)."""
+    try:
+        raw = base64.urlsafe_b64decode(text.encode("ascii"))
+        token = msgspec.json.decode(raw, type=TruncateToken)
+    except ValueError, msgspec.DecodeError:
+        return None
+    if token.offset < 0 or token.limit < 1:
+        return None
+    return token
+
+
+def _truncated_marker(omitted: int, token: TruncateToken) -> str:
+    return (
+        f"\n...[truncated {omitted} chars; more via get_truncated token] "
+        f"[TRUNCATE_TOKEN: {encode_truncate_token(token)}]"
+    )
+
+
+def truncate_with_token(
+    text: str,
+    max_chars: int,
+    *,
+    kind: TruncateKind,
+    location: str,
+    offset: int,
+    limit: int,
+    total_len: int,
+) -> tuple[str, TruncateToken | None]:
+    """Bound ``text`` to ``max_chars``; return (bounded_text, next_token).
+
+    ``text`` is a slice of the full source starting at ``offset``; ``total_len``
+    is the full source length. A token is embedded whenever more content remains
+    after the returned chunk, so ``get_truncated`` chains through truncations
+    and each chunk may itself carry a fresh token. The token advances by the
+    length actually returned (no content is skipped), and the marker is budgeted
+    so the total returned length stays near ``max_chars`` (the agent loop's own
+    cap never re-cuts it and never strips the token).
+    """
+    if max_chars < 1:
+        return text, None
+    if len(text) > max_chars:
+        omitted = len(text) - max_chars
+        # First pass sizes the marker with an offset+max_chars token; re-encode
+        # with the real cut so the cursor lands exactly where content stops.
+        first = TruncateToken(kind=kind, location=location, offset=offset + max_chars, limit=limit)
+        cut = max_chars - len(_truncated_marker(omitted, first))
+        if cut < 1:
+            return text[:max_chars] + _truncated_marker(omitted, first), first
+        token = TruncateToken(kind=kind, location=location, offset=offset + cut, limit=limit)
+        return text[:cut] + _truncated_marker(omitted, token), token
+    more_after = offset + len(text) < total_len
+    if not more_after:
+        return text, None
+    token = TruncateToken(kind=kind, location=location, offset=offset + len(text), limit=limit)
+    marker = (
+        f"\n...[more content available; use get_truncated with the token] "
+        f"[TRUNCATE_TOKEN: {encode_truncate_token(token)}]"
+    )
+    return text + marker, token

src/plyngent/tools/workspace.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Sequence
@@ -268,6 +268,23 @@ def resolve_path(path: str | Path) -> Path:
     return resolved
 
 
+def argv_shape_error(argv: object) -> str | None:
+    """Return a model-facing error when ``argv`` is not a non-empty argv list.
+
+    Models occasionally pass a shell-style string or a JSON-encoded array as a
+    string; duck-typing would otherwise let it reach ``create_subprocess_exec``
+    and splat into single characters. Declared as ``object`` so the isinstance
+    checks are not flagged as unnecessary.
+    """
+    if isinstance(argv, str):
+        return 'only accepts an array of args, not a string; pass e.g. ["ls", "-la"]'
+    if not isinstance(argv, list):
+        return f"only accepts an array of args, not a {type(argv).__name__}"
+    if not all(isinstance(part, str) for part in cast("list[object]", argv)):
+        return "only accepts an array of string args"
+    return None
+
+
 def check_command_allowed(argv: list[str]) -> None:
     """Raise if argv is empty or the executable basename is denylisted.
 
@@ -275,6 +292,10 @@ def check_command_allowed(argv: list[str]) -> None:
     installed: the human is asked (with a timeout; default deny). Session grants
     skip re-prompting for the same basename. Independent of YOLO soft-confirm.
     """
+    shape_error = argv_shape_error(argv)
+    if shape_error is not None:
+        msg = f"command {shape_error}"
+        raise WorkspaceError(msg)
     if not argv:
         msg = "command argv must not be empty"
         raise WorkspaceError(msg)

tests/test_agent/test_budget.py

@@ -15,7 +15,8 @@ def test_truncate_tool_result_long() -> None:
     assert "Truncated" in out
     assert "20" in out
     assert "30" in out
-    assert "configuration file" in out
+    assert "get_truncated" in out
+    assert "configuration file" not in out
 
 
 def test_truncate_tool_result_no_hint() -> None:
@@ -23,7 +24,7 @@ def test_truncate_tool_result_no_hint() -> None:
     out = truncate_tool_result(text, 20, hint=False)
     assert out.startswith("a" * 20)
     assert "truncated" in out
-    assert "configuration file" not in out
+    assert "get_truncated" not in out
 
 
 def test_estimate_message_chars() -> None:

tests/test_agent/test_compact.py

@@ -44,7 +44,8 @@ def test_truncate_tool_result_long() -> None:
     assert "Truncated" in out
     assert "20" in out
     assert "30" in out
-    assert "configuration file" in out
+    assert "get_truncated" in out
+    assert "configuration file" not in out
 
 
 def test_truncate_tool_result_no_hint() -> None:
@@ -52,7 +53,7 @@ def test_truncate_tool_result_no_hint() -> None:
     out = truncate_tool_result(text, 20, hint=False)
     assert out.startswith("a" * 20)
     assert "truncated" in out
-    assert "configuration file" not in out
+    assert "get_truncated" not in out
 
 
 def test_compact_shrinks_old_tool_results() -> None:

tests/test_agent/test_responses_bridge.py

@@ -1,8 +1,13 @@
+from __future__ import annotations
+
+from typing import Any
+
 import msgspec
 
 from plyngent.agent.responses_bridge import (
     chat_messages_to_responses_input,
     chat_param_to_responses_kwargs,
+    reasoning_summary_text,
     response_to_assistant_message,
     response_to_chat_completion,
     responses_status_to_finish_reason,
@@ -145,6 +150,54 @@ def test_response_to_assistant_with_tools() -> None:
     assert completion.choices[0].finish_reason == "tool_calls"
 
 
+def _reasoning_response(
+    *,
+    output_reasoning: bool = False,
+    top_level: dict[str, Any] | None = None,
+) -> Response:
+    output: list[dict[str, Any]] = []
+    if output_reasoning:
+        output.append(
+            {
+                "id": "reason_1",
+                "type": "reasoning",
+                "summary": [{"type": "summary_text", "text": "openai-style reasoning"}],
+            }
+        )
+    return Response(
+        id="resp_r",
+        created_at=1,
+        model="deepseek-v4-flash",
+        output=output,
+        reasoning=top_level,
+    )
+
+
+def test_reasoning_summary_text_deepseek_top_level() -> None:
+    """DeepSeek Responses returns reasoning in the top-level ``reasoning`` field."""
+    response = _reasoning_response(top_level={"content": [{"type": "reasoning_content", "text": "think step by step"}]})
+    assert reasoning_summary_text(response) == "think step by step"
+    assistant = response_to_assistant_message(response)
+    assert assistant.reasoning_content == "think step by step"
+
+
+def test_reasoning_summary_text_prefers_top_level() -> None:
+    response = _reasoning_response(
+        output_reasoning=True,
+        top_level={"summary": [{"type": "summary_text", "text": "top-level reasoning"}]},
+    )
+    assert reasoning_summary_text(response) == "top-level reasoning"
+
+
+def test_reasoning_summary_text_falls_back_to_output_items() -> None:
+    response = _reasoning_response(output_reasoning=True)
+    assert reasoning_summary_text(response) == "openai-style reasoning"
+
+
+def test_reasoning_summary_text_empty_when_absent() -> None:
+    assert reasoning_summary_text(_reasoning_response()) == ""
+
+
 def test_responses_status_to_finish_reason_incomplete() -> None:
     body = {
         "id": "resp_i",

tests/test_agent/test_responses_dispatch.py

@@ -177,13 +177,67 @@ async def test_dispatch_responses_stream_text_and_usage() -> None:
     )
     stream = await dispatch_responses(cast("Any", client), param, stream=True)
     chunks = [chunk async for chunk in cast("Any", stream)]
-    # text deltas + finish + usage
     texts = [c.choices[0].delta.content for c in chunks if c.choices and isinstance(c.choices[0].delta.content, str)]
     assert texts == ["stream", "ed"]
     assert any(c.choices and c.choices[0].finish_reason == "stop" for c in chunks)
     assert any(token_usage_present(c) for c in chunks)
 
 
+async def _streamed_reasoning(
+    events: list[ResponseStreamEvent],
+) -> str:
+    """Run one Responses stream and join all streamed ``reasoning_content`` deltas."""
+    import msgspec
+
+    final = _completed_response(text="ok")
+    final.reasoning = {"content": [{"type": "reasoning_content", "text": "terminal reasoning"}]}
+    events.append(ResponseStreamEvent(type="response.completed", response=msgspec.to_builtins(final)))
+    client = ScriptedResponsesClient(stream_events=[events])
+    param = ChatCompletionsParam(model="gpt-test", messages=[UserChatMessage(content="hi")])
+    stream = await dispatch_responses(cast("Any", client), param, stream=True)
+    chunks = [chunk async for chunk in cast("Any", stream)]
+    parts: list[str] = []
+    for chunk in chunks:
+        if not chunk.choices:
+            continue
+        delta = chunk.choices[0].delta
+        if isinstance(delta.reasoning_content, str) and delta.reasoning_content:
+            parts.append(delta.reasoning_content)
+    return "".join(parts)
+
+
+async def test_dispatch_responses_stream_reasoning_text_done() -> None:
+    """DeepSeek ``response.reasoning_text.done`` carries the full reasoning text."""
+    reasoning = await _streamed_reasoning(
+        [ResponseStreamEvent(type="response.reasoning_text.done", delta="full chain of thought")]
+    )
+    assert reasoning == "full chain of thought"
+
+
+async def test_dispatch_responses_stream_reasoning_deltas() -> None:
+    reasoning = await _streamed_reasoning(
+        [
+            ResponseStreamEvent(type="response.reasoning_text.delta", delta="step one "),
+            ResponseStreamEvent(type="response.reasoning_text.delta", delta="step two"),
+        ]
+    )
+    assert reasoning == "step one step two"
+
+
+async def test_dispatch_responses_stream_terminal_only_reasoning() -> None:
+    """DeepSeek sometimes delivers reasoning only on the terminal response."""
+    reasoning = await _streamed_reasoning([])
+    assert reasoning == "terminal reasoning"
+
+
+async def test_dispatch_responses_stream_reasoning_no_duplicate() -> None:
+    """Streamed deltas win; the terminal reasoning is not re-emitted."""
+    reasoning = await _streamed_reasoning(
+        [ResponseStreamEvent(type="response.reasoning_text.delta", delta="streamed only")]
+    )
+    assert reasoning == "streamed only"
+
+
 def token_usage_present(chunk: object) -> bool:
     usage = getattr(chunk, "usage", UNSET)
     return usage is not UNSET and usage is not None

tests/test_cli/test_display.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from plyngent.agent import ReasoningDeltaEvent, TextDeltaEvent, ToolResultEvent
+from plyngent.agent import ReasoningDeltaEvent, TextDeltaEvent, ToolCallEvent, ToolResultEvent
 from plyngent.cli.display import (
     get_markdown_enabled,
     markdown_render_available,
@@ -11,7 +11,11 @@ from plyngent.cli.display import (
     set_markdown_enabled,
     set_verbose_tool_results,
 )
-from plyngent.lmproto.openai_compatible.model import ToolChatMessage
+from plyngent.lmproto.openai_compatible.model import (
+    AssistantFunctionTool,
+    AssistantFunctionToolCall,
+    ToolChatMessage,
+)
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator
@@ -61,7 +65,7 @@ async def test_flush_markdown_on_source_change(
 
     call = AssistantFunctionToolCall(
         id="1",
-        function=AssistantFunctionTool(name="read_file", arguments="{}"),
+        function=AssistantFunctionTool(name="run_argv", arguments='{"argv": ["echo", "x"]}'),
     )
     await render_events(
         _aiter(
@@ -78,6 +82,73 @@ async def test_flush_markdown_on_source_change(
     assert "after" in out
 
 
+def _read_call(args_json: str) -> ToolCallEvent:
+    return ToolCallEvent(
+        tool_call=AssistantFunctionToolCall(
+            id="1",
+            function=AssistantFunctionTool(name="read_file", arguments=args_json),
+        )
+    )
+
+
+def _result(content: str) -> ToolResultEvent:
+    return ToolResultEvent(message=ToolChatMessage(content=content, tool_call_id="1"))
+
+
+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 "[tool]" not in out
+    assert "[tool ok]" not in out
+
+
+async def test_pretty_read_file_statuses(capsys: pytest.CaptureFixture[str]) -> None:
+    cases = [
+        ("error: file not found: a.txt", "(file not found)"),
+        ("error: not a file: a.txt", "(not a file)"),
+        ("error: something else", "(error)"),
+        ("no header content", "(done)"),
+    ]
+    for content, status in cases:
+        await render_events(_aiter([_read_call('{"path": "a.txt"}'), _result(content)]))
+        out = capsys.readouterr().out
+        assert f"* Read 'a.txt' {status}" in out, f"status {status} not rendered for {content!r}"
+
+
+async def test_pretty_todo_push(capsys: pytest.CaptureFixture[str]) -> None:
+    call = ToolCallEvent(
+        tool_call=AssistantFunctionToolCall(
+            id="2",
+            function=AssistantFunctionTool(name="todo_push", arguments='{"titles": ["T1"]}'),
+        )
+    )
+    result = (
+        "pushed group (depth=1) items=[a1]\n"
+        "(LIFO of groups: depth=1; TOP group = current breakdown level)\n"
+        "group d=0 TOP:\n"
+        "  [ ] a1: T1"
+    )
+    await render_events(_aiter([call, _result(result)]))
+    out = capsys.readouterr().out
+    assert "* Todo Push:" in out
+    assert "  [ ] a1: T1" in out
+    assert "[tool]" not in out
+
+
+async def test_non_pretty_tool_keeps_old_style(capsys: pytest.CaptureFixture[str]) -> None:
+    call = ToolCallEvent(
+        tool_call=AssistantFunctionToolCall(
+            id="3",
+            function=AssistantFunctionTool(name="run_argv", arguments='{"argv": ["ls"]}'),
+        )
+    )
+    await render_events(_aiter([call, _result("exit_code=0")]))
+    out = capsys.readouterr().out
+    assert "[tool] run_argv(" in out
+    assert "[tool ok]" in out
+
+
 async def test_tool_result_preview_vs_verbose(capsys: pytest.CaptureFixture[str]) -> None:
     long = "x" * 200
     msg = ToolChatMessage(content=long, tool_call_id="1")

tests/test_cli/test_limits.py

@@ -30,10 +30,10 @@ def test_prompt_confirm_tool_default_deny() -> None:
 
 
 def test_format_tool_confirm_box_multiline() -> None:
-    reason = "run_command: python3 -c (review code before allow)\nargv:\n  python3 -c print(1)\n-c code:\n  print(1)"
+    reason = "run_argv: python3 -c (review code before allow)\nargv:\n  python3 -c print(1)\n-c code:\n  print(1)"
-    box = format_tool_confirm_box("run_command", reason)
+    box = format_tool_confirm_box("run_argv", reason)
     assert "┌" in box and "└" in box
-    assert "confirm · tool 'run_command'" in box
+    assert "confirm · tool 'run_argv'" in box
     assert "python3 -c" in box
     assert "print(1)" in box
     # Distinct lines, not one jammed row

tests/test_prompting.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import io
+
 import pytest
 
 from plyngent.prompting import (
@@ -7,14 +9,48 @@ from plyngent.prompting import (
     FormField,
     NonInteractiveBackend,
     NonInteractiveError,
+    _read_line_with_timeout_posix,
     ask,
     choose,
     confirm,
     form,
+    read_line_with_timeout,
     temporary_backend,
 )
 
 
+def _ready_poll(r, w, x, t):
+    return (r, w, x)
+
+
+def _never_poll(r, w, x, t):
+    return ([], [], [])
+
+
+def test_read_line_with_timeout_reads_line() -> None:
+    stream = io.StringIO("hello\n")
+    assert _read_line_with_timeout_posix("p> ", 5.0, stream=stream, poll=_ready_poll) == "hello"
+
+
+def test_read_line_with_timeout_empty_enter() -> None:
+    stream = io.StringIO("\n")
+    assert _read_line_with_timeout_posix("p> ", 5.0, stream=stream, poll=_ready_poll) == ""
+
+
+def test_read_line_with_timeout_timeout_returns_none() -> None:
+    stream = io.StringIO()
+    assert _read_line_with_timeout_posix("p> ", 0.01, stream=stream, poll=_never_poll) is None
+
+
+def test_read_line_with_timeout_eof_returns_none() -> None:
+    stream = io.StringIO("")
+    assert _read_line_with_timeout_posix("p> ", 5.0, stream=stream, poll=_ready_poll) is None
+
+
+def test_read_line_with_timeout_negative_timeout() -> None:
+    assert read_line_with_timeout("p> ", -1) is None
+
+
 class ScriptedBackend:
     """Deterministic backend for unit tests."""
 

tests/test_tools/test_chat_tools.py

@@ -1,12 +1,15 @@
 from __future__ import annotations
 
+import importlib
 import json
 
 from plyngent.agent import ToolRegistry
 from plyngent.prompting import NonInteractiveBackend, temporary_backend
-from plyngent.tools.chat import CHAT_TOOLS, ask_user, choose_user, form_user
+from plyngent.tools.chat import CHAT_TOOLS, ask_user, choose_user, form_user, wait
 from tests.test_prompting import ScriptedBackend
 
+wait_module = importlib.import_module("plyngent.tools.chat.wait")
+
 
 async def test_ask_user_tool() -> None:
     backend = ScriptedBackend(["42"])
@@ -77,7 +80,7 @@ async def test_form_user_tool() -> None:
 
 async def test_chat_tools_in_default_list() -> None:
     names = {t.name for t in CHAT_TOOLS}
-    assert names == {"ask_user_line", "ask_user_choice", "ask_user_form"}
+    assert names == {"ask_user_line", "ask_user_choice", "ask_user_form", "wait"}
 
 
 async def test_ask_user_non_interactive_error() -> None:
@@ -85,3 +88,59 @@ async def test_ask_user_non_interactive_error() -> None:
         registry = ToolRegistry([ask_user])
         out = await registry.execute("ask_user_line", '{"question": "hi"}')
     assert out.startswith("error:")
+
+
+async def test_wait_tool_zero_seconds(monkeypatch) -> None:
+    with temporary_backend(NonInteractiveBackend()):
+        registry = ToolRegistry([wait])
+        out = await registry.execute("wait", '{"duration": 0}')
+    assert out == "waited 0s"
+
+
+async def test_wait_tool_negative_duration() -> None:
+    with temporary_backend(NonInteractiveBackend()):
+        registry = ToolRegistry([wait])
+        out = await registry.execute("wait", '{"duration": -1}')
+    assert out.startswith("error:")
+
+
+async def test_wait_tool_noninteractive_sleeps(monkeypatch) -> None:
+    slept: list[float] = []
+
+    async def fake_sleep(seconds: float) -> None:
+        slept.append(seconds)
+
+    monkeypatch.setattr(wait_module.asyncio, "sleep", fake_sleep)
+    with temporary_backend(NonInteractiveBackend()):
+        registry = ToolRegistry([wait])
+        out = await registry.execute("wait", '{"duration": 5}')
+    assert out == "waited 5s"
+    assert slept == [5]
+
+
+async def test_wait_tool_disturbed_with_reason(monkeypatch) -> None:
+    monkeypatch.setattr(
+        wait_module,
+        "read_line_with_timeout",
+        lambda prompt, timeout: "tests are slow",
+    )
+    with temporary_backend(ScriptedBackend([])):
+        registry = ToolRegistry([wait])
+        out = await registry.execute("wait", '{"duration": 3}')
+    assert out == "disturbed by user: tests are slow"
+
+
+async def test_wait_tool_disturbed_empty_enter(monkeypatch) -> None:
+    monkeypatch.setattr(wait_module, "read_line_with_timeout", lambda prompt, timeout: "")
+    with temporary_backend(ScriptedBackend([])):
+        registry = ToolRegistry([wait])
+        out = await registry.execute("wait", '{"duration": 3}')
+    assert out == "disturbed by user (no reason)"
+
+
+async def test_wait_tool_times_out(monkeypatch) -> None:
+    monkeypatch.setattr(wait_module, "read_line_with_timeout", lambda prompt, timeout: None)
+    with temporary_backend(ScriptedBackend([])):
+        registry = ToolRegistry([wait])
+        out = await registry.execute("wait", '{"duration": 3}')
+    assert out == "waited 3s"

tests/test_tools/test_danger.py

@@ -43,46 +43,46 @@ def test_classify_write_overwrite_only(tmp_path: Path) -> None:
 def test_classify_safe_tools() -> None:
     assert classify_danger("read_file", {"path": "a"}) is None
     assert classify_danger("listdir", {"path": "."}) is None
-    assert classify_danger("run_command", {"command": ["echo", "hi"]}) is None
+    assert classify_danger("run_argv", {"argv": ["echo", "hi"]}) is None
     assert classify_danger("open_pty", {"command": ["true"]}) is None
-    assert classify_danger("run_command", {"command": ["ls", "-la"]}) is None
+    assert classify_danger("run_argv", {"argv": ["ls", "-la"]}) is None
 
 
-def test_classify_run_command_batch_risky() -> None:
+def test_classify_run_argv_batch_risky() -> None:
     reason = classify_danger(
-        "run_command_batch",
+        "run_argv_batch",
         {
-            "commands": [
+            "steps": [
-                {"command": ["echo", "ok"]},
+                {"argv": ["echo", "ok"]},
-                {"command": ["bash", "-c", "echo risky"]},
+                {"argv": ["bash", "-c", "echo risky"]},
             ]
         },
     )
     assert reason is not None
-    assert "run_command_batch" in reason
+    assert "run_argv_batch" in reason
     assert "risky" in reason or "bash" in reason
     assert (
         classify_danger(
-            "run_command_batch",
+            "run_argv_batch",
-            {"commands": [{"command": ["echo", "ok"]}]},
+            {"steps": [{"argv": ["echo", "ok"]}]},
         )
         is None
     )
 
 
 def test_classify_shell_and_dash_c() -> None:
-    r = classify_danger("run_command", {"command": ["bash", "-c", "rm -rf /"]})
+    r = classify_danger("run_argv", {"argv": ["bash", "-c", "rm -rf /"]})
     assert r is not None
     assert "bash -c" in r
     assert "rm -rf" in r
 
-    r2 = classify_danger("run_command", {"command": ["python3", "-c", "print(1)"]})
+    r2 = classify_danger("run_argv", {"argv": ["python3", "-c", "print(1)"]})
     assert r2 is not None
     assert "python" in r2
     assert "-c" in r2
     assert "print(1)" in r2
 
-    r_py = classify_danger("run_command", {"command": ["python", "-c", "import os"]})
+    r_py = classify_danger("run_argv", {"argv": ["python", "-c", "import os"]})
     assert r_py is not None and "python -c" in r_py
 
     r3 = classify_danger("open_pty", {"command": ["bash"]})
@@ -118,8 +118,8 @@ async def test_confirm_deny_with_comment() -> None:
 def test_shell_confirm_formats_command_placeholder() -> None:
     script = chr(10).join(["line1", "line2", "line3"])
     reason = classify_danger(
-        "run_command",
+        "run_argv",
-        {"command": ["bash", "-c", script]},
+        {"argv": ["bash", "-c", script]},
     )
     assert reason is not None
     assert "$(command)" in reason

tests/test_tools/test_edit_lineno.py

@@ -19,7 +19,7 @@ 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 call_sync(read_file, "a.txt") == "one\nTWO\nTHREE\nfour\n"
+    assert call_sync(read_file, "a.txt") == "L1-4\none\nTWO\nTHREE\nfour\n"
 
 
 def test_edit_lineno_requires_all_lines_read(workspace: object) -> None:
@@ -49,7 +49,7 @@ def test_edit_lineno_append_requires_last_line_read(workspace: object) -> None:
     _ = call_sync(read_file, "ap.txt", offset=2, limit=1, with_lineno=True)
     out = call_sync(edit_lineno, "ap.txt", 4, 4, "d\n")
     assert "appended" in out
-    assert call_sync(read_file, "ap.txt") == "a\nb\nc\nd\n"
+    assert call_sync(read_file, "ap.txt") == "L1-4\na\nb\nc\nd\n"
 
 
 def test_edit_lineno_delete_range(workspace: object) -> None:
@@ -58,7 +58,7 @@ def test_edit_lineno_delete_range(workspace: object) -> None:
     _ = _read_lineno("b.txt")
     out = call_sync(edit_lineno, "b.txt", 2, 2, "")
     assert "replaced" in out
-    assert call_sync(read_file, "b.txt") == "a\nc\n"
+    assert call_sync(read_file, "b.txt") == "L1-2\na\nc\n"
 
 
 def test_edit_lineno_append(workspace: object) -> None:
@@ -67,7 +67,7 @@ def test_edit_lineno_append(workspace: object) -> None:
     _ = _read_lineno("c.txt")
     out = call_sync(edit_lineno, "c.txt", 2, 2, "more\n")
     assert "appended" in out
-    assert call_sync(read_file, "c.txt") == "only\nmore\n"
+    assert call_sync(read_file, "c.txt") == "L1-2\nonly\nmore\n"
 
 
 def test_edit_lineno_invalid(workspace: object) -> None:

tests/test_tools/test_fetch.py

@@ -20,6 +20,7 @@ from plyngent.tools.net.policy import (
     parse_fetch_url,
     soft_confirm_reason,
 )
+from plyngent.tools.truncate_token import decode_truncate_token
 from tests.test_tools.helpers import call_async
 
 if TYPE_CHECKING:
@@ -237,6 +238,40 @@ async def test_fetch_truncation_and_binary(workspace: Path, http_server: str) ->
     assert "binary body omitted" in out_bin
 
 
+async def test_fetch_char_truncation_embeds_token(workspace: Path, http_server: str) -> None:
+    del workspace
+    port = int(http_server.rsplit(":", 1)[1])
+    grant_private_host("127.0.0.1", port)
+
+    out = await call_async(fetch, f"{http_server}/big", max_chars=1000)
+    assert "truncated=true" in out
+    body = out.split("--- body ---", 1)[-1]
+    assert "TRUNCATE_TOKEN:" in body
+    token_line = next(ln for ln in body.splitlines() if "TRUNCATE_TOKEN:" in ln)
+    token_str = token_line.split("[TRUNCATE_TOKEN: ", 1)[1].rstrip("]")
+    token = decode_truncate_token(token_str)
+    assert token is not None
+    assert token.kind == "http"
+    assert token.limit == 1000
+    assert 0 < token.offset < 1000
+    # Continuation resumes exactly where the chunk stopped (no gap).
+    out2 = await call_async(fetch, f"{http_server}/big", max_chars=1000, offset=token.offset)
+    body2 = out2.split("--- body ---", 1)[-1].strip()
+    assert body2.startswith("x")
+
+
+async def test_fetch_offset_skips_body(workspace: Path, http_server: str) -> None:
+    del workspace
+    port = int(http_server.rsplit(":", 1)[1])
+    grant_private_host("127.0.0.1", port)
+
+    # /big body is 5000 chars; offset 4900 leaves 100 chars, no token needed.
+    out = await call_async(fetch, f"{http_server}/big", max_chars=200, offset=4900)
+    body = out.split("--- body ---", 1)[-1]
+    assert body.strip() == "x" * 100
+    assert "TRUNCATE_TOKEN:" not in out
+
+
 async def test_fetch_private_denied_without_grant(workspace: Path, http_server: str) -> None:
     del workspace
     # No grant, no policy hook → hard deny (even though server is up).

tests/test_tools/test_file.py

@@ -1,26 +1,62 @@
 from __future__ import annotations
 
+import importlib
+from pathlib import Path
+
 from plyngent.tools.file import edit_replace, listdir, read_file, write_file
 from tests.test_tools.helpers import call_sync
 
+get_truncated_module = importlib.import_module("plyngent.tools.file.get_truncated")
+
+
+def _extract_token(out: str) -> str | None:
+    for line in out.splitlines():
+        if "TRUNCATE_TOKEN:" in line:
+            return line.split("[TRUNCATE_TOKEN: ", 1)[1].rstrip("]")
+    return None
+
 
 def test_write_read_listdir_edit(workspace: object) -> None:
     del workspace
     assert "wrote" in call_sync(write_file, "notes/a.txt", "hello world")
-    assert call_sync(read_file, "notes/a.txt") == "hello world"
+    assert call_sync(read_file, "notes/a.txt") == "L1-1\nhello world"
     listing = call_sync(listdir, "notes")
     assert "a.txt" in listing
     assert "file" in listing
     result = call_sync(edit_replace, "notes/a.txt", "world", "there")
     assert "replaced" in result
-    assert call_sync(read_file, "notes/a.txt") == "hello there"
+    assert call_sync(read_file, "notes/a.txt") == "L1-1\nhello there"
+
+
+def test_read_file_max_chars_embeds_token(workspace: Path) -> None:
+    (workspace / "big.txt").write_text("a" * 1000, encoding="utf-8")
+    out = call_sync(read_file, "big.txt", max_chars=200)
+    header, _, rest = out.partition("\n")
+    assert header.startswith("L1-")  # 1-based range of the raw slice read
+    content, _, marker = rest.partition("\n...[")
+    assert content == "a" * len(content)
+    assert len(content) < 200
+    assert "TRUNCATE_TOKEN:" in marker
+    token = _extract_token(out)
+    assert token is not None
+    parsed = get_truncated_module.decode_truncate_token(token)
+    assert parsed is not None
+    assert parsed.kind == "file"
+    assert parsed.offset == len(content)  # cursor lands exactly where content stops
+
+
+def test_read_file_max_chars_small_file_untouched(workspace: Path) -> None:
+    (workspace / "small.txt").write_text("tiny", encoding="utf-8")
+    out = call_sync(read_file, "small.txt", max_chars=200)
+    assert out == "L1-1\ntiny"
+    assert "TRUNCATE_TOKEN:" not in out
 
 
 def test_edit_replace_first_only(workspace: object) -> None:
     del workspace
     _ = call_sync(write_file, "t.txt", "aa aa")
     result = call_sync(edit_replace, "t.txt", "aa", "bb")
-    assert call_sync(read_file, "t.txt") == "bb aa"
+    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
 
@@ -29,7 +65,7 @@ def test_edit_replace_max_replaces(workspace: object) -> None:
     del workspace
     _ = call_sync(write_file, "t.txt", "aa aa aa")
     result = call_sync(edit_replace, "t.txt", "aa", "bb", max_replaces=2)
-    assert call_sync(read_file, "t.txt") == "bb bb aa"
+    assert call_sync(read_file, "t.txt") == "L1-1\nbb bb aa"
     assert "2 of 3" in result
     assert "1 remain" in result
 
@@ -38,7 +74,7 @@ def test_edit_replace_all_matches(workspace: object) -> None:
     del workspace
     _ = call_sync(write_file, "t.txt", "aa aa")
     result = call_sync(edit_replace, "t.txt", "aa", "bb", max_replaces=10)
-    assert call_sync(read_file, "t.txt") == "bb bb"
+    assert call_sync(read_file, "t.txt") == "L1-1\nbb bb"
     assert "all 2 matches" in result
 
 
@@ -57,7 +93,14 @@ def test_edit_missing_old_string(workspace: object) -> None:
 def test_read_offset_limit(workspace: object) -> None:
     del workspace
     _ = call_sync(write_file, "lines.txt", "a\nb\nc\nd\n")
-    assert call_sync(read_file, "lines.txt", offset=1, limit=2) == "b\nc\n"
+    # offset is 0-based: offset=1 skips line 1, so the read starts at 1-based line 2.
+    assert call_sync(read_file, "lines.txt", offset=1, limit=2) == "L2-3\nb\nc\n"
+
+
+def test_read_file_missing_and_directory(workspace: Path) -> None:
+    (workspace / "sub").mkdir()
+    assert call_sync(read_file, "nope.txt") == "error: file not found: nope.txt"
+    assert call_sync(read_file, "sub") == "error: not a file: sub"
 
 
 def test_read_with_lineno(workspace: object) -> None:

tests/test_tools/test_fs_ops.py

@@ -11,7 +11,7 @@ def test_copy_file(workspace: object) -> None:
     _ = call_sync(write_file, "a.txt", "hello")
     out = call_sync(copy_path, "a.txt", "b.txt")
     assert "copied" in out
-    assert call_sync(read_file, "b.txt") == "hello"
+    assert call_sync(read_file, "b.txt") == "L1-1\nhello"
     assert (workspace / "a.txt").is_file()
 
 
@@ -21,8 +21,8 @@ def test_copy_directory(workspace: object) -> None:
     _ = call_sync(write_file, "src/nested/y.txt", "y")
     out = call_sync(copy_path, "src", "dst")
     assert "directory" in out
-    assert call_sync(read_file, "dst/x.txt") == "x"
+    assert call_sync(read_file, "dst/x.txt") == "L1-1\nx"
-    assert call_sync(read_file, "dst/nested/y.txt") == "y"
+    assert call_sync(read_file, "dst/nested/y.txt") == "L1-1\ny"
 
 
 def test_copy_overwrite(workspace: object) -> None:
@@ -32,7 +32,7 @@ def test_copy_overwrite(workspace: object) -> None:
     assert "exists" in call_sync(copy_path, "b.txt", "a.txt")
     out = call_sync(copy_path, "b.txt", "a.txt", overwrite=True)
     assert "copied" in out
-    assert call_sync(read_file, "a.txt") == "new"
+    assert call_sync(read_file, "a.txt") == "L1-1\nnew"
 
 
 def test_move_file(workspace: object) -> None:
@@ -41,7 +41,7 @@ def test_move_file(workspace: object) -> None:
     out = call_sync(move_path, "old.txt", "renamed.txt")
     assert "moved" in out
     assert not (workspace / "old.txt").exists()
-    assert call_sync(read_file, "renamed.txt") == "data"
+    assert call_sync(read_file, "renamed.txt") == "L1-1\ndata"
 
 
 def test_move_directory(workspace: object) -> None:
@@ -50,7 +50,7 @@ def test_move_directory(workspace: object) -> None:
     out = call_sync(move_path, "d", "e")
     assert "moved" in out
     assert not (workspace / "d").exists()
-    assert call_sync(read_file, "e/a.txt") == "a"
+    assert call_sync(read_file, "e/a.txt") == "L1-1\na"
 
 
 def test_delete_file(workspace: object) -> None:

tests/test_tools/test_get_truncated.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import importlib
+from pathlib import Path
+
+from plyngent.tools.truncate_token import TruncateToken, decode_truncate_token, encode_truncate_token
+from tests.test_tools.helpers import call_async
+
+get_truncated_module = importlib.import_module("plyngent.tools.file.get_truncated")
+get_truncated = get_truncated_module.get_truncated
+
+
+def _extract_token(out: str) -> str | None:
+    for line in out.splitlines():
+        if "TRUNCATE_TOKEN:" in line:
+            return line.split("[TRUNCATE_TOKEN: ", 1)[1].rstrip("]")
+    return None
+
+
+async def test_get_truncated_invalid_token() -> None:
+    out = await call_async(get_truncated, "garbage-token")
+    assert out.startswith("error:")
+    assert "invalid" in out
+
+
+async def test_get_truncated_file_chains_without_gap(workspace: Path) -> None:
+    (workspace / "big.txt").write_text("a" * 1000, encoding="utf-8")
+    token = encode_truncate_token(TruncateToken(kind="file", location="big.txt", offset=0, limit=200))
+    total_read = 0
+    for _ in range(20):
+        out = await call_async(get_truncated, token, max_chars=200)
+        body, _, _ = out.partition("\n...[")
+        header, _, content = body.partition("\n")
+        assert header.startswith("L") and "-" in header
+        assert content == "a" * len(content)
+        total_read += len(content)
+        next_token = _extract_token(out)
+        if next_token is None:
+            break
+        parsed = decode_truncate_token(next_token)
+        assert parsed is not None
+        assert parsed.offset == total_read  # no gap, no overlap
+        token = next_token
+    assert total_read == 1000
+
+
+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)
+    assert out.startswith("error:")
+
+
+async def test_get_truncated_http_passthrough(monkeypatch) -> None:
+    calls: list[tuple[str, int, int]] = []
+
+    async def fake_fetch(url: str, *, offset: int, max_chars: int) -> str:
+        calls.append((url, offset, max_chars))
+        return f"body-from-{offset}"
+
+    monkeypatch.setattr(get_truncated_module.fetch, "handler", fake_fetch)
+    token = encode_truncate_token(TruncateToken(kind="http", location="https://example.com/x", offset=500, limit=400))
+    out = await call_async(get_truncated, token, max_chars=200)
+    assert out == "body-from-500"
+    assert calls == [("https://example.com/x", 500, 400)]

tests/test_tools/test_process.py

@@ -9,8 +9,8 @@ from plyngent.tools.process import (
     close_pty,
     open_pty,
     read_pty,
-    run_command,
+    run_argv,
-    run_command_batch,
+    run_argv_batch,
     write_pty,
     write_pty_keys,
 )
@@ -41,20 +41,50 @@ def _py(code: str) -> list[str]:
     return [sys.executable, "-c", code]
 
 
-async def test_run_command_echo(workspace: object) -> None:
+async def test_run_argv_echo(workspace: object) -> None:
     del workspace
-    out = await call_async(run_command, _py("print('hi')"))
+    out = await call_async(run_argv, _py("print('hi')"))
     assert "exit_code=0" in out
     assert "hi" in out
 
 
-async def test_run_command_batch_serial(workspace: object) -> None:
+async def test_run_argv_string_command_error(workspace: object) -> None:
+    del workspace
+    out = await call_async(run_argv, "ls -la")
+    assert "error:" in out
+    assert "only accepts an array" in out
+    assert "executable not found" not in out
+
+
+async def test_run_argv_stringified_json_array_error(workspace: object) -> None:
+    del workspace
+    out = await call_async(run_argv, '["ls", "-la"]')
+    assert "error:" in out
+    assert "only accepts an array" in out
+    assert "executable not found" not in out
+
+
+async def test_run_argv_non_string_args_error(workspace: object) -> None:
+    del workspace
+    out = await call_async(run_argv, ["ls", 1])
+    assert "error:" in out
+    assert "array of string args" in out
+
+
+async def test_open_pty_string_command_error(workspace: object) -> None:
+    del workspace
+    out = await call_async(open_pty, "echo hi")
+    assert "error:" in out
+    assert "only accepts an array" in out
+
+
+async def test_run_argv_batch_serial(workspace: object) -> None:
     del workspace
     out = await call_async(
-        run_command_batch,
+        run_argv_batch,
         [
-            {"command": _py("print('a')")},
+            {"argv": _py("print('a')")},
-            {"command": _py("print('b')")},
+            {"argv": _py("print('b')")},
         ],
     )
     assert "steps=2" in out
@@ -65,13 +95,13 @@ async def test_run_command_batch_serial(workspace: object) -> None:
     assert "a" in out and "b" in out
 
 
-async def test_run_command_batch_stop_on_error(workspace: object) -> None:
+async def test_run_argv_batch_stop_on_error(workspace: object) -> None:
     del workspace
     out = await call_async(
-        run_command_batch,
+        run_argv_batch,
         [
-            {"command": _py("import sys; sys.exit(2)")},
+            {"argv": _py("import sys; sys.exit(2)")},
-            {"command": _py("print('should-not-run')")},
+            {"argv": _py("print('should-not-run')")},
         ],
         stop_on_error=True,
     )
@@ -80,17 +110,17 @@ async def test_run_command_batch_stop_on_error(workspace: object) -> None:
     assert "should-not-run" not in out
 
 
-async def test_run_command_batch_pipe_out(workspace: object) -> None:
+async def test_run_argv_batch_pipe_out(workspace: object) -> None:
     del workspace
     out = await call_async(
-        run_command_batch,
+        run_argv_batch,
         [
             {
-                "command": _py("print('piped-payload', end='')"),
+                "argv": _py("print('piped-payload', end='')"),
                 "pipe_out": True,
             },
             {
-                "command": _py("import sys; print(sys.stdin.read())"),
+                "argv": _py("import sys; print(sys.stdin.read())"),
             },
         ],
     )
@@ -99,27 +129,27 @@ async def test_run_command_batch_pipe_out(workspace: object) -> None:
     assert "pipe_out=true" in out
 
 
-async def test_run_command_batch_last_pipe_out_error(workspace: object) -> None:
+async def test_run_argv_batch_last_pipe_out_error(workspace: object) -> None:
     del workspace
     out = await call_async(
-        run_command_batch,
+        run_argv_batch,
-        [{"command": _py("print(1)"), "pipe_out": True}],
+        [{"argv": _py("print(1)"), "pipe_out": True}],
     )
     assert "error:" in out
     assert "pipe_out" in out
 
 
-async def test_run_command_batch_mix_stderr(workspace: object) -> None:
+async def test_run_argv_batch_mix_stderr(workspace: object) -> None:
     del workspace
     out = await call_async(
-        run_command_batch,
+        run_argv_batch,
         [
             {
-                "command": _py("import sys; sys.stdout.write('out'); sys.stderr.write('err')"),
+                "argv": _py("import sys; sys.stdout.write('out'); sys.stderr.write('err')"),
                 "mix_stderr": True,
                 "pipe_out": True,
             },
-            {"command": _py("import sys; print(repr(sys.stdin.read()))")},
+            {"argv": _py("import sys; print(repr(sys.stdin.read()))")},
         ],
     )
     assert "ran=2" in out
@@ -127,32 +157,32 @@ async def test_run_command_batch_mix_stderr(workspace: object) -> None:
     assert "out" in out and "err" in out
 
 
-async def test_run_command_denied(workspace: object) -> None:
+async def test_run_argv_denied(workspace: object) -> None:
     del workspace
-    out = await call_async(run_command, ["rm", "-rf", "."])
+    out = await call_async(run_argv, ["rm", "-rf", "."])
     assert "denied" in out
 
 
-async def test_run_command_cwd(workspace: object) -> None:
+async def test_run_argv_cwd(workspace: object) -> None:
     assert isinstance(workspace, Path)
     sub = workspace / "sub"
     sub.mkdir()
     _ = (sub / "f.txt").write_text("z", encoding="utf-8")
-    out = await call_async(run_command, _py("import os; print('\\n'.join(os.listdir()))"), cwd="sub")
+    out = await call_async(run_argv, _py("import os; print('\\n'.join(os.listdir()))"), cwd="sub")
     assert "f.txt" in out
 
 
-async def test_run_command_timeout(workspace: object) -> None:
+async def test_run_argv_timeout(workspace: object) -> None:
     del workspace
-    out = await call_async(run_command, _py("import time; time.sleep(5)"), timeout_seconds=0.2)
+    out = await call_async(run_argv, _py("import time; time.sleep(5)"), timeout_seconds=0.2)
     assert "timed_out=true" in out
 
 
-async def test_run_command_timeout_keeps_partial_output(workspace: object) -> None:
+async def test_run_argv_timeout_keeps_partial_output(workspace: object) -> None:
     del workspace
     # Print then sleep past the timeout so communicate has partial stdout after kill.
     out = await call_async(
-        run_command,
+        run_argv,
         _py("import sys, time; sys.stdout.write('partial-out'); sys.stdout.flush(); time.sleep(5)"),
         timeout_seconds=0.3,
     )
@@ -160,10 +190,10 @@ async def test_run_command_timeout_keeps_partial_output(workspace: object) -> No
     assert "partial-out" in out
 
 
-async def test_run_command_stdin(workspace: object) -> None:
+async def test_run_argv_stdin(workspace: object) -> None:
     del workspace
     out = await call_async(
-        run_command,
+        run_argv,
         _py("import sys; print(sys.stdin.read(), end='')"),
         stdin="hello-stdin\n",
     )
@@ -171,10 +201,10 @@ async def test_run_command_stdin(workspace: object) -> None:
     assert "hello-stdin" in out
 
 
-async def test_run_command_env(workspace: object) -> None:
+async def test_run_argv_env(workspace: object) -> None:
     del workspace
     out = await call_async(
-        run_command,
+        run_argv,
         _py("import os; print(os.environ.get('PLYNGENT_TEST_VAR', ''), end='')"),
         env={"PLYNGENT_TEST_VAR": "from-env"},
     )

tests/test_tools/test_temp_workspace.py

@@ -33,10 +33,10 @@ def test_new_temporary_workspace_allowlist(workspace: object) -> None:
     target = temp / "scratch.txt"
     _ = target.write_text("hello-temp", encoding="utf-8")
     assert resolve_path(str(target)) == target.resolve()
-    assert call_sync(read_file, str(target)) == "hello-temp"
+    assert call_sync(read_file, str(target)) == "L1-1\nhello-temp"
 
     _ = call_sync(write_file, "project.txt", "proj")
-    assert call_sync(read_file, "project.txt") == "proj"
+    assert call_sync(read_file, "project.txt") == "L1-1\nproj"
 
     # Sibling under system temp that we did not allowlist still fails.
     outside = temp.parent / "not-ours-should-fail"

tests/test_tools/test_truncate_token.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+from plyngent.tools.truncate_token import (
+    TruncateToken,
+    decode_truncate_token,
+    encode_truncate_token,
+    truncate_with_token,
+)
+
+
+def test_token_roundtrip() -> None:
+    token = TruncateToken(kind="file", location="src/a.py", offset=100, limit=32_000)
+    encoded = encode_truncate_token(token)
+    assert decode_truncate_token(encoded) == token
+
+
+def test_decode_garbage_returns_none() -> None:
+    assert decode_truncate_token("not-a-token") is None
+    assert decode_truncate_token("") is None
+    assert decode_truncate_token("!!!") is None
+
+
+def test_decode_rejects_invalid_fields() -> None:
+    bad = TruncateToken(kind="file", location="x", offset=-1, limit=10)
+    assert decode_truncate_token(encode_truncate_token(bad)) is None
+    bad_limit = TruncateToken(kind="http", location="u", offset=0, limit=0)
+    assert decode_truncate_token(encode_truncate_token(bad_limit)) is None
+
+
+def test_truncate_with_token_short_text_untouched() -> None:
+    out, token = truncate_with_token(
+        "hello",
+        32_000,
+        kind="file",
+        location="a.txt",
+        offset=0,
+        limit=32_000,
+        total_len=5,
+    )
+    assert out == "hello"
+    assert token is None
+
+
+def test_truncate_with_token_cuts_and_advances() -> None:
+    text = "x" * 500
+    out, token = truncate_with_token(
+        text,
+        200,
+        kind="file",
+        location="a.txt",
+        offset=0,
+        limit=200,
+        total_len=500,
+    )
+    assert token is not None
+    content, _, _ = out.partition("\n...[")
+    assert content.startswith("x")
+    assert "truncated 300 chars" in out
+    assert f"[TRUNCATE_TOKEN: {encode_truncate_token(token)}]" in out
+    assert token.offset == len(content)  # cursor lands exactly where content stops
+    assert len(out) <= 200  # marker fits inside the char budget
+
+
+def test_truncate_with_token_more_after_nonoverflow() -> None:
+    # Segment fits the chunk, but the source continues: still emit a token.
+    text = "y" * 30
+    out, token = truncate_with_token(
+        text,
+        40,
+        kind="file",
+        location="a.txt",
+        offset=0,
+        limit=40,
+        total_len=200,
+    )
+    assert token is not None
+    assert token.offset == 30  # advanced by returned length only
+    assert "more content available" in out
+    assert out.startswith("y" * 30)
+
+
+def test_truncate_with_token_end_reached_no_token() -> None:
+    text = "z" * 30
+    out, token = truncate_with_token(
+        text,
+        40,
+        kind="http",
+        location="https://example.com/x",
+        offset=0,
+        limit=40,
+        total_len=30,
+    )
+    assert token is None
+    assert out == text
+
+
+def test_truncate_with_token_nonzero_start() -> None:
+    text = "y" * 500
+    out, token = truncate_with_token(
+        text,
+        200,
+        kind="http",
+        location="https://example.com/x",
+        offset=150,
+        limit=200,
+        total_len=700,
+    )
+    assert token is not None
+    assert token.kind == "http"
+    content, _, _ = out.partition("\n...[")
+    assert token.offset == 150 + len(content)
+    assert "truncated 300 chars" in out
+    assert len(out) <= 200