docs: document host notices, restart resume, and /btw scope

AGENTS.md

@@ -110,9 +110,11 @@ Click app + readline REPL. Entry: `plyngent` / `python -m plyngent`.
 
 - **`plyngent chat`**: provider/model (flags or interactive; Tab via readline in `prompting`); sessions store `provider_name`/`model` and restore on resume; SQLite via `[database]` (file DB under user data when url unset; explicit `url = ":memory:"` kept + warn); workspace-bound; resume latest for cwd/`--workspace` by default (`--new` / `--session`). One-shot: `-p/--prompt` and non-TTY stdin; exit codes 0/1/2/3; `--yes` (YOLO on), `--auto-continue` (auto-raise limits), `--stream/--no-stream`, `--quiet`. Root `--log-level`. Enabled `[mcp]` servers start before the REPL (`_start_mcp_manager`, per-server failures warn on stderr) and are closed on exit (`McpManager.aclose`).
 - Slash: Click group in `cli/slash.py` + `awaitlet` for async work; Tab completer from registry + ParamType `shell_complete`. Multiline `"""` … `"""`; `/edit` via `$VISUAL`/`$EDITOR` (blocking only). `/yolo on|off|once` for soft destructive confirms. `/model --persist` / `/models --persist` write model catalog entries into TOML. `/todos` for human show/push/pop/clear of the todo stack. `/mcp` (`list` default) prints per-server status/tool counts + an `instructions` preview when the server sent usage guidance, + last stderr lines; `/mcp reconnect` re-reads config, restarts every enabled server, and rebuilds the tool registry. `/grants` lists directory-access grants (`revoke <index|all>`; config grants point at the TOML); `/status` includes the grant count.
-- `/btw` side questions: forked session; main transcript/DB untouched. `--tools=read` (default) exposes only `READ_ONLY` tools under `read_only_context()` (fetch=GET) with the main session view; `--tools=no` off; `--tools=full` whole-registry clone with fresh session. Tools off: default falls back to no tools; explicit `read`/`full` errors.
-- Explicit `/resume` or `--session` from another workspace prompts: **keep** / **update** / **abort**.
-- Failed/cancelled turns: user kept; **committed tool rounds kept** (side effects not re-run on `/retry`); only unfinished assistant rolled back; Ctrl+C cancels; TTY confirms off-loop; auto-retry 10s/20s/30s then `/retry`.
+- `/btw` side questions: forked session; main transcript/DB untouched. `--tools=read` (default) exposes only `READ_ONLY` tools under `read_only_context()` (fetch=GET) with the main session view; `--tools=no` off; `--tools=full` whole-registry clone with fresh session. Tools off: default falls back to no tools; explicit `read`/`full` errors. The model gets an `aside` notice as its last message before the question (unsaved exchange + tool scope), shown to the user as `[notice] aside: …`.
+- Explicit `/resume` or `--session` from another workspace prompts: **keep** / **update** / **abort**. A session resumed by a **freshly started process** (`--session` / default latest) also gets a `resume` notice about what a restart drops (PTY sessions, running commands, temp workspaces) — resuming inside a live process pushes nothing.
+- Host notices (`agent/notices.py`): system-level messages appended as trailing `developer` rows, never system-prompt edits — `Notice.to_message()` for the model, `.summary` for the user via `cli.display.echo_notice` (`--quiet` hides the user line only). Durable hosts persist them (`ChatAgent.push_notice`); a side agent without memory keeps its notice local. Trailing developer messages never end a turn, so `/retry` still sees an interrupted turn behind a notice or checkpoint.
+- Failed/cancelled turns: user kept; **committed tool rounds kept** (side effects not re-run on `/retry`); only unfinished assistant rolled back; Ctrl+C cancels; TTY confirms off-loop; auto-retry waits 5s/10s/15s/20s then +10s (10 attempts) and the budget **resets when an attempt completes a model round** (connection recovered); then `/retry`.
+- Pretty tool lines: known tools (`read_file`, `run_argv`, `grep_files`, …) print their `* Verb 'target' ` prefix as the call starts and append the outcome (`L1-4 (done)`, `(exit code 0)`) when the result lands, so a blocking tool stays visible while it runs. Interactive TTY only (`interactive_terminal()`): piped/plain output and `--verbose` keep whole lines / `[tool ok]`. A second call before the first result (parallel batch) erases the open prefix and falls back to whole lines.
 - **`plyngent providers`**, **`config path|edit`**. Config open: `$VISUAL`/`$EDITOR` (wait), else system default (`xdg-open` / `open` / `startfile`, non-blocking). `/edit` stays blocking-only. No providers → optional edit then reload when waited.
 - Tools default on; workspace defaults to cwd; `--max-rounds` default 32. Readline history under platformdirs (`repl_history`). PTY: `close_all` on chat exit.
 

README.md

@@ -221,10 +221,12 @@ plyngent chat --session 3
 | `--auto-continue` | Auto-raise tool/PTY limits without prompting (like answering `yyy`) |
 | `--log-level` | On the root CLI: `DEBUG`, `INFO`, `WARNING`, … |
 
-Sessions resume the **most recently updated** session for the current workspace unless you pass `--new` or `--session`. Each session remembers the last **provider** and **model** (restored on resume so you are not re-prompted).
+Sessions resume the **most recently updated** session for the current workspace unless you pass `--new` or `--session`. Each session remembers the last **provider** and **model** (restored on resume so you are not re-prompted). A session resumed by a fresh `plyngent` process also gets a `[notice] resume: …` line: the model is told that process-scoped state — PTY sessions, running commands, temporary workspaces — did not survive the restart, and that an unfinished turn may have been cut off. The notice is kept with the transcript; `--quiet` hides only the user-facing line.
 
 When a tool-loop or PTY limit is hit, the prompt accepts `y` (continue once), `n` (stop), or `yyy` — **stop asking for the rest of this turn** (the next user turn prompts again). `--auto-continue` / `[agent] auto_continue_limits = true` skip the prompt for every turn.
 
+Tool calls show progress as they run: known tools (`read_file`, `run_argv`, `grep_files`, …) print a `* Read 'src/x.py' ` prefix the moment the call starts, and the outcome (`L1-80 (done)`, `(exit code 0)`) is appended once the result lands. Piped output and `--verbose` print the finished line as before.
+
 ### One-shot (scripts / CI)
 
 ```bash
@@ -264,7 +266,7 @@ Type `/help` in the REPL for the live list. Common ones:
 | `/yolo [on\|off\|once]` | Soft destructive confirms: sticky skip, off, or next turn only |
 
 | `/retry` | Re-run incomplete last user turn (after error/cancel) |
-| `/btw [--tools read\|no\|full] [--fresh]` | Side question without changing the main session (read-only tools by default) |
+| `/btw [--tools read\|no\|full] [--fresh]` | Side question without changing the main session (read-only tools by default; the model is told the exchange is unsaved and how its tools are scoped) |
 | `/provider` `/model` | Switch without restarting |
 | `/model --persist` | Save current model id into `plyngent.toml` catalog |
 | `/models` | List config + remote `GET /models` (always re-fetches) |
@@ -274,7 +276,7 @@ Type `/help` in the REPL for the live list. Common ones:
 | `/config` | Edit `plyngent.toml` ($VISUAL/$EDITOR or system open); reload after blocking editor |
 | `/quit` | Leave the REPL |
 
-User messages are saved immediately. On API error or Ctrl+C, partial assistant/tool output is discarded but the user message stays so `/retry` works after resume. Interactive auto-retry uses 10s / 20s / 30s delays.
+User messages are saved immediately. On API error or Ctrl+C, partial assistant/tool output is discarded but the user message stays so `/retry` works after resume. Interactive auto-retry waits 5s, 10s, 15s, 20s, then +10s each step (10 attempts); an attempt that gets a full model round back **resets that budget**, so a connection that recovers and drops again starts over from 5s.
 
 ## Workspace model
 

src/plyngent/agent/__init__.py

@@ -20,6 +20,10 @@ from .events import ToolResultEvent as ToolResultEvent
 from .events import UsageEvent as UsageEvent
 from .loop import DEFAULT_MAX_ROUNDS as DEFAULT_MAX_ROUNDS
 from .loop import run_chat_loop as run_chat_loop
+from .notices import Notice as Notice
+from .notices import aside_notice as aside_notice
+from .notices import format_moment as format_moment
+from .notices import resume_notice as resume_notice
 from .tools import DangerClassifier as DangerClassifier
 from .tools import ToolConfirmHook as ToolConfirmHook
 from .tools import ToolDefinition as ToolDefinition

src/plyngent/agent/chat.py

@@ -9,6 +9,7 @@ from msgspec import UNSET
 from plyngent.lmproto.openai_compatible.model import (
     AssistantChatMessage,
     AssistantFunctionToolCall,
+    DeveloperChatMessage,
     SystemChatMessage,
     ToolChatMessage,
     UserChatMessage,
@@ -40,6 +41,7 @@ if TYPE_CHECKING:
     from plyngent.memory import MemoryStore
 
     from .events import AgentEvent
+    from .notices import Notice
     from .todo_nag import TodoNagStrategy
     from .todo_stack import TodoStack
     from .types import AnyLLMClient
@@ -138,12 +140,16 @@ def incomplete_turn_user_text(messages: Sequence[AnyChatMessage]) -> str | None:
     Incomplete when the last non-system message is a user message (failed before
     any commit), or tool results after a complete tool batch (failed on a later
     model round — keep tools so retry does not re-execute side effects).
+    Trailing developer messages (host notices, directive checkpoints) belong to
+    the turn but never end it, so they are skipped when reading the tail.
     """
     # Skip leading system prompt only.
     start = 0
     if messages and isinstance(messages[0], SystemChatMessage):
         start = 1
     body = list(messages[start:])
+    while body and isinstance(body[-1], DeveloperChatMessage):
+        _ = body.pop()
     if not body:
         return None
     last = body[-1]
@@ -420,6 +426,21 @@ class ChatAgent:
         if load:
             await self.load_history()
 
+    async def push_notice(self, notice: Notice) -> DeveloperChatMessage:
+        """Append a host notice for the model and persist it with the transcript.
+
+        Notices are trailing ``developer`` messages, never system-prompt edits:
+        the model reads the newest host state last. Hosts with a durable session
+        store them as normal history (a side agent without memory keeps its
+        notice local to that exchange). The checkpoint cursor stays ahead of the
+        new message, so it is never re-persisted by a later tool batch.
+        """
+        message = notice.to_message()
+        self.messages.append(message)
+        await self._persist(message)
+        self._persist_from = len(self.messages)
+        return message
+
     async def _persist(self, message: AnyChatMessage) -> None:
         if self.memory is not None and self.session_id is not None:
             _ = await self.memory.append_message(self.session_id, message)
@@ -461,8 +482,6 @@ class ChatAgent:
 
     def _developer_tail_end(self, start: int) -> int:
         """Extend *start* through trailing developer checkpoint messages."""
-        from plyngent.lmproto.openai_compatible.model import DeveloperChatMessage
-
         end = start
         while end < len(self.messages) and isinstance(self.messages[end], DeveloperChatMessage):
             end += 1
@@ -584,19 +603,24 @@ class ChatAgent:
         tools: ToolRegistry | bool | None = False,
         max_rounds: int | None = None,
         system_prompt: str | None = None,
+        notice: Notice | None = None,
         instance_state: object | None = None,
         session_state: object | None = None,
     ) -> AsyncIterator[AgentEvent]:
         """Run a side turn that does not mutate this agent or its memory.
 
         - Message list is forked (optional history copy); never written back.
-        - ``memory`` / ``session_id`` are unset on the side agent (no DB).
+        - ``memory`` / ``session_id`` are unset on the side agent (no DB), so a
+          *notice* stays local to this exchange.
         - ``todo_stack`` is unset (no nags / main stack thrash).
         - Tools default **off**. ``tools=True`` clones this agent's registry with
           a **fresh session** bag (unless *session_state* is passed) and the
           given *instance_state* (CLI typically shares the host instance for
           workspace identity). A caller-built :class:`ToolRegistry` (e.g. a
           ``clone(read_only_only=True)`` subset) is used as-is.
+        - *notice* is appended after the forked history, i.e. right before the
+          side question, so the model knows this exchange's rules (unsaved,
+          read-only tools, …).
         """
         text = user_text.strip()
         if not text:
@@ -621,6 +645,8 @@ class ChatAgent:
             session_state=aside_session,
         )
         history = list(self.messages) if include_history else []
+        if notice is not None:
+            history.append(notice.to_message())
         rounds = self.max_rounds if max_rounds is None else max_rounds
         if aside_tools is None and max_rounds is None:
             rounds = 1

src/plyngent/agent/notices.py

@@ -0,0 +1,92 @@
+"""Host notices: short system-level messages appended after existing history.
+
+A notice never edits the system prompt. It is written as a trailing
+``developer`` message — the mid-history system channel, coerced to ``system``
+for providers without that role (see ``lmproto.openai_compatible.compat``) — so
+the model reads the newest host state last, right before the next user turn.
+
+Hosts show :attr:`Notice.summary` to the user and persist
+:meth:`Notice.to_message` with the rest of the transcript when they have a
+durable session; side agents without memory keep it local to that exchange.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import UTC, datetime
+
+from plyngent.lmproto.openai_compatible.model import DeveloperChatMessage
+
+# Marker prefix so hosts and the model can tell notices apart from other
+# developer messages (durable directive checkpoints share the role).
+NOTICE_MARKER = "[plyngent notice"
+
+
+@dataclass(frozen=True, slots=True)
+class Notice:
+    """One host notice: model-facing :attr:`body`, user-facing :attr:`summary`."""
+
+    kind: str
+    body: str
+    summary: str = ""
+
+    def to_message(self) -> DeveloperChatMessage:
+        """Render as a trailing ``developer`` message (system-level channel)."""
+        return DeveloperChatMessage(content=f"{NOTICE_MARKER}: {self.kind}]\n{self.body.strip()}")
+
+
+def format_moment(moment: datetime | None) -> str:
+    """Local ``YYYY-MM-DD HH:MM`` stamp for notices; ``unknown`` when absent."""
+    if moment is None:
+        return "unknown"
+    if moment.tzinfo is None:
+        # SQLite hands back naive datetimes for tz-aware columns.
+        moment = moment.replace(tzinfo=UTC)
+    return moment.astimezone().strftime("%Y-%m-%d %H:%M")
+
+
+def resume_notice(*, last_active: datetime | None) -> Notice:
+    """Notice for a session resumed in a freshly started process.
+
+    Only process-scoped state is gone after a restart (PTY sessions, running
+    commands, temporary workspaces); durable session state — transcript, todo
+    stack, workspace binding, access grants — is restored before this lands.
+    """
+    stamp = format_moment(last_active)
+    body = (
+        f"This session was resumed in a new plyngent process (previous run's last "
+        f"activity: {stamp}). Process-scoped state did not survive the restart: PTY "
+        "sessions, running commands and temporary workspaces are gone, and any shell "
+        "working directory you changed no longer applies. The transcript, todo stack, "
+        "workspace binding and directory-access grants were restored. A turn that looks "
+        "unfinished was cut off mid-way: re-read files or re-run commands before relying "
+        "on their earlier output."
+    )
+    summary = f"resumed in a new process (previous run's last activity: {stamp})"
+    return Notice(kind="resume", body=body, summary=summary)
+
+
+_ASIDE_BODIES: dict[str, str] = {
+    "read": (
+        "Side question: this exchange is not saved to the session transcript and does not "
+        "change the main conversation. Tools here are read-only (file/VCS reads, ask, fetch "
+        "GET): you cannot edit files or run state-changing commands, so describe a change "
+        "instead of applying it."
+    ),
+    "no": (
+        "Side question: this exchange is not saved to the session transcript and does not "
+        "change the main conversation. Tools are disabled here — answer from this "
+        "conversation and your own knowledge, and say so when something cannot be verified."
+    ),
+    "full": (
+        "Side question: this exchange is not saved to the session transcript and does not "
+        "change the main conversation. Full tools are available and their side effects are "
+        "real, but nothing here is kept: re-check anything you would need later."
+    ),
+}
+
+
+def aside_notice(*, tools_mode: str) -> Notice:
+    """Notice for a ``/btw`` side question: unsaved exchange plus tool scope."""
+    body = _ASIDE_BODIES.get(tools_mode, _ASIDE_BODIES["read"])
+    return Notice(kind="aside", body=body, summary=f"{tools_mode} tools; not saved to the transcript")

src/plyngent/cli/app.py

@@ -167,7 +167,8 @@ async def _bind_session(
 ) -> None:
     if session_id is not None:
         try:
-            await state.resume_session(session_id)
+            # Startup resume: this process did not run the session's last turn.
+            await state.resume_session(session_id, restarted=True)
         except ValueError as exc:
             raise click.ClickException(str(exc)) from exc
         if not quiet and not oneshot:
@@ -184,7 +185,7 @@ async def _bind_session(
                 err=True,
             )
         return
-    mode = await state.resume_latest_or_new()
+    mode = await state.resume_latest_or_new(restarted=True)
     if quiet:
         return
     if mode == "resume":
@@ -324,6 +325,7 @@ async def _run_chat(  # noqa: C901, PLR0912, PLR0915 — chat orchestration
             interactive_limits=interactive,
             auto_continue_limits=auto_effective,
             yolo=yolo,
+            quiet=quiet,
             mcp_manager=mcp_manager,
         )
         # Path denylist and policy confirm live on instance.workspace (no process bag).

src/plyngent/cli/display.py

@@ -7,6 +7,7 @@ import re
 import shlex
 import sys
 from contextvars import ContextVar
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Literal, cast
 
 import click
@@ -26,37 +27,11 @@ from plyngent.lmproto.openai_compatible.model import AssistantFunctionToolCall
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator, Callable
 
-    from plyngent.agent import AgentEvent
+    from plyngent.agent import AgentEvent, Notice
 
 _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",
-        "tree",
-        "listdir",
-        "glob_paths",
-        "grep_files",
-        "run_argv",
-        "run_argv_batch",
-        "fetch",
-        "edit_replace",
-        "edit_lineno",
-        "write_file",
-        "copy_path",
-        "move_path",
-        "delete_path",
-        "vcs_status",
-        "vcs_log",
-        "wait",
-    }
-)
-
 # Process/session display flags (set from ReplState / slash).
 _verbose_tool_results: ContextVar[bool] = ContextVar("verbose_tool_results", default=False)
 _markdown_enabled: ContextVar[bool] = ContextVar("markdown_enabled", default=True)
@@ -82,8 +57,8 @@ def get_markdown_enabled() -> bool:
     return _markdown_enabled.get()
 
 
-def markdown_render_available() -> bool:
-    """True when stdout is a TTY and plain mode is not forced via env."""
+def interactive_terminal() -> bool:
+    """True when stdout is a terminal we may write mid-line and erase on."""
     if os.environ.get("PLYNGENT_PLAIN", "").strip() in {"1", "true", "yes", "on"}:
         return False
     try:
@@ -92,6 +67,17 @@ def markdown_render_available() -> bool:
         return False
 
 
+def markdown_render_available() -> bool:
+    """True when stdout is a TTY and plain mode is not forced via env."""
+    return interactive_terminal()
+
+
+def echo_notice(notice: Notice, *, err: bool = False) -> None:
+    """Show a host notice to the user; the model gets the full body."""
+    text = f"[notice] {notice.kind}: {notice.summary}" if notice.summary else f"[notice] {notice.kind}"
+    click.secho(text, fg="yellow", err=err)
+
+
 def _preview(text: str, limit: int) -> str:
     if len(text) <= limit:
         return text
@@ -137,14 +123,19 @@ def _json_list_arg(args_json: str, key: str) -> list[str] | None:
     return None
 
 
-def _pretty_line(prefix: str, *segments: tuple[str, str | None]) -> str:
-    """Style a pretty tool summary: yellow ``* Verb`` prefix + colored detail segments.
+def _pretty_prefix(text: str) -> str:
+    """Style the leading ``* Verb 'target' `` segment of a pretty tool line."""
+    return click.style(text, fg="yellow")
+
+
+def _pretty_segments(*segments: tuple[str, str | None]) -> str:
+    """Style pretty-line detail segments.
 
     ``segments`` are ``(text, fg)`` pairs; ``fg=None`` leaves a segment uncolored
     and ``"dim"`` renders it dim. click.style emits ANSI codes only on a TTY,
     so non-TTY output stays plain.
     """
-    parts = [click.style(prefix, fg="yellow")]
+    parts: list[str] = []
     for text, fg in segments:
         if fg == "dim":
             parts.append(click.style(text, dim=True))
@@ -153,25 +144,51 @@ def _pretty_line(prefix: str, *segments: tuple[str, str | None]) -> str:
     return "".join(parts)
 
 
-def _read_file_pretty(args_json: str, result: str) -> str:
-    """One-line summary for a ``read_file`` call: ``* Read 'path' L1-4 (done)``."""
+@dataclass(frozen=True, slots=True)
+class PrettyLine:
+    """Split renderer for one pretty tool call.
+
+    ``prefix`` is written as soon as the call starts — a blocking tool call is
+    then visible while it runs — and ``detail`` is appended once its result
+    lands. Together they equal the single line the whole-line path prints.
+    """
+
+    prefix: Callable[[str], str]
+    detail: Callable[[str, str], str]
+
+
+def _read_file_prefix(args_json: str) -> str:
+    """``* Read 'path' ``: known before the result, so it can lead the line."""
     path = _json_str_arg(args_json, "path") or "?"
+    return _pretty_prefix(f"* Read '{path}' ")
+
+
+def _read_file_detail(_args_json: str, result: str) -> str:
+    """``read_file`` outcome: range/done, or which failure kind occurred."""
     if result.startswith("error: file not found"):
-        return _pretty_line(f"* Read '{path}' ", ("(file not found)", "red"))
+        return _pretty_segments(("(file not found)", "red"))
     if result.startswith("error: not a file"):
-        return _pretty_line(f"* Read '{path}' ", ("(not a file)", "red"))
+        return _pretty_segments(("(not a file)", "red"))
     if result.startswith("error:"):
-        return _pretty_line(f"* Read '{path}' ", ("(error)", "red"))
-    first = result.splitlines()[0] if result.splitlines() else ""
+        return _pretty_segments(("(error)", "red"))
+    lines = result.splitlines()
+    first = lines[0] if lines else ""
     if first.startswith("L") and "-" in first:
-        return _pretty_line(f"* Read '{path}' ", (f"{first} ", "dim"), ("(done)", "green"))
-    return _pretty_line(f"* Read '{path}' ", ("(done)", "green"))
+        return _pretty_segments((f"{first} ", "dim"), ("(done)", "green"))
+    return _pretty_segments(("(done)", "green"))
 
 
-def _todo_pretty(name: str, result: str) -> str:
-    """Summary for todo push/update: header + the rendered todo stack."""
+def _todo_line(name: str) -> PrettyLine:
+    """``todo_push`` / ``todo_update``: header first, rendered stack as the detail."""
     label = "Todo Push" if name == "todo_push" else "Todo Update"
-    return _pretty_line(f"* {label}:\n{result}")
+
+    def prefix(_args_json: str) -> str:
+        return click.style(f"* {label}:", fg="yellow")
+
+    def detail(_args_json: str, result: str) -> str:
+        return click.style(f"\n{result}", fg="yellow")
+
+    return PrettyLine(prefix=prefix, detail=detail)
 
 
 def _tree_line_stats(fmt: str, line: str) -> tuple[str, int] | None:
@@ -202,15 +219,19 @@ def _tree_line_stats(fmt: str, line: str) -> tuple[str, int] | None:
     return kind, depth
 
 
-def _tree_pretty(args_json: str, result: str) -> str:
-    """One-line summary for a ``tree`` call: ``* Tree 'src' (3 dirs, 24 files, depth 3)``.
+def _tree_prefix(args_json: str) -> str:
+    path = _json_str_arg(args_json, "path") or "."
+    return _pretty_prefix(f"* Tree '{path}' ")
+
+
+def _tree_detail(args_json: str, result: str) -> str:
+    """``tree`` outcome: dir/file counts, plus depth when the format has one.
 
     Flat trees have no depth, so the depth part is omitted for ``format=flat``.
     """
-    path = _json_str_arg(args_json, "path") or "."
-    fmt = _json_str_arg(args_json, "format") or "markdown"
     if result.startswith("error:"):
-        return f"* Tree '{path}' (error)"
+        return _pretty_segments(("(error)", "red"))
+    fmt = _json_str_arg(args_json, "format") or "markdown"
     dirs = 0
     files = 0
     depth = 0
@@ -225,17 +246,21 @@ def _tree_pretty(args_json: str, result: str) -> str:
             files += 1
         depth = max(depth, line_depth)
     if depth:
-        return _pretty_line(f"* Tree '{path}' ", (f"({dirs} dirs, {files} files, depth {depth})", None))
-    return _pretty_line(f"* Tree '{path}' ", (f"({dirs} dirs, {files} files)", None))
+        return _pretty_segments((f"({dirs} dirs, {files} files, depth {depth})", None))
+    return _pretty_segments((f"({dirs} dirs, {files} files)", None))
 
 
-def _listdir_pretty(args_json: str, result: str) -> str:
-    """One-line summary for a ``listdir`` call: ``* List 'path' (3 dirs, 5 files)``."""
+def _listdir_prefix(args_json: str) -> str:
     path = _json_str_arg(args_json, "path") or "."
+    return _pretty_prefix(f"* List '{path}' ")
+
+
+def _listdir_detail(_args_json: str, result: str) -> str:
+    """``listdir`` outcome: dir/file counts, or empty/error."""
     if result.startswith("error:"):
-        return _pretty_line(f"* List '{path}' ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     if result == "(empty)":
-        return _pretty_line(f"* List '{path}' ", ("(empty)", "dim"))
+        return _pretty_segments(("(empty)", "dim"))
     dirs = 0
     files = 0
     for line in result.splitlines():
@@ -243,46 +268,55 @@ def _listdir_pretty(args_json: str, result: str) -> str:
             dirs += 1
         elif line.startswith("file\t"):
             files += 1
-    return _pretty_line(f"* List '{path}' ", (f"({dirs} dirs, {files} files)", None))
+    return _pretty_segments((f"({dirs} dirs, {files} files)", None))
 
 
-def _glob_pretty(args_json: str, result: str) -> str:
-    """One-line summary for a ``glob_paths`` call: ``* Glob '**/*.py' in '.' (12 paths)``."""
+def _glob_prefix(args_json: str) -> str:
     pattern = _json_str_arg(args_json, "pattern") or "?"
     path = _json_str_arg(args_json, "path") or "."
+    return _pretty_prefix(f"* Glob '{pattern}' in '{path}' ")
+
+
+def _glob_detail(_args_json: str, result: str) -> str:
+    """``glob_paths`` outcome: match count, or none/error."""
     if result.startswith("error:"):
-        return _pretty_line(f"* Glob '{pattern}' in '{path}' ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     if result == "(no matches)":
-        return _pretty_line(f"* Glob '{pattern}' in '{path}' ", ("(no matches)", "dim"))
+        return _pretty_segments(("(no matches)", "dim"))
     count = sum(1 for line in result.splitlines() if line and not line.startswith("...[truncated"))
     unit = "path" if count == 1 else "paths"
-    return _pretty_line(f"* Glob '{pattern}' in '{path}' ", (f"({count} {unit})", None))
+    return _pretty_segments((f"({count} {unit})", None))
 
 
-def _grep_pretty(args_json: str, result: str) -> str:
-    """One-line summary for a ``grep_files`` call: ``* Grep 'pat' in '.' (12 matches in 3 files)``."""
+def _grep_prefix(args_json: str) -> str:
     pattern = _json_str_arg(args_json, "pattern") or "?"
     path = _json_str_arg(args_json, "path") or "."
+    return _pretty_prefix(f"* Grep '{pattern}' in '{path}' ")
+
+
+def _grep_detail(_args_json: str, result: str) -> str:
+    """``grep_files`` outcome: matches and file count, or none/error."""
     if result.startswith("error:"):
-        return _pretty_line(f"* Grep '{pattern}' in '{path}' ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     if result == "(no matches)":
-        return _pretty_line(f"* Grep '{pattern}' in '{path}' ", ("(no matches)", "dim"))
+        return _pretty_segments(("(no matches)", "dim"))
     matches = [line for line in result.splitlines() if line and not line.startswith("...[truncated")]
     files = len({line.split(":", 1)[0] for line in matches})
     match_unit = "match" if len(matches) == 1 else "matches"
     file_unit = "file" if files == 1 else "files"
-    return _pretty_line(
-        f"* Grep '{pattern}' in '{path}' ",
-        (f"({len(matches)} {match_unit} in {files} {file_unit})", None),
-    )
+    return _pretty_segments((f"({len(matches)} {match_unit} in {files} {file_unit})", None))
 
 
-def _run_argv_pretty(args_json: str, result: str) -> str:
-    """One-line summary for a ``run_argv`` call: ``* Run $ git status --short (exit code 0)``."""
+def _run_argv_prefix(args_json: str) -> str:
     argv = _json_list_arg(args_json, "argv")
     cmd = shlex.join(argv) if argv else "?"
+    return _pretty_prefix(f"* Run $ {cmd} ")
+
+
+def _run_argv_detail(_args_json: str, result: str) -> str:
+    """``run_argv`` outcome: exit code, timeout, or error."""
     if result.startswith("error:"):
-        return _pretty_line(f"* Run $ {cmd} ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     fields: dict[str, str] = {}
     for line in result.splitlines():
         if line.startswith("--- "):
@@ -291,25 +325,25 @@ def _run_argv_pretty(args_json: str, result: str) -> str:
             key, _, value = line.partition("=")
             fields[key] = value
     if fields.get("timed_out") == "true":
-        return _pretty_line(f"* Run $ {cmd} ", ("(timed out)", "red"))
+        return _pretty_segments(("(timed out)", "red"))
     code = fields.get("exit_code", "")
     fg = "green" if code == "0" else "red"
-    return _pretty_line(f"* Run $ {cmd} ", (f"(exit code {code or 'killed'})", fg))
+    return _pretty_segments((f"(exit code {code or 'killed'})", fg))
 
 
-def _run_argv_batch_pretty(_args_json: str, result: str) -> str:
-    """One-line summary for a ``run_argv_batch`` call: ``* Run batch (done)``."""
+def _run_argv_batch_prefix(_args_json: str) -> str:
+    return _pretty_prefix("* Run batch ")
+
+
+def _run_argv_batch_detail(_args_json: str, result: str) -> str:
+    """``run_argv_batch`` outcome: done, stopped early, or error."""
     if result.startswith("error:"):
-        return _pretty_line("* Run batch ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     lines = result.splitlines()
     head = lines[0] if lines else ""
-    stopped_early = False
-    for part in head.split():
-        if part == "stopped_early=true":
-            stopped_early = True
-    if stopped_early:
-        return _pretty_line("* Run batch ", ("(stopped early)", "yellow"))
-    return _pretty_line("* Run batch ", ("(done)", "green"))
+    if any(part == "stopped_early=true" for part in head.split()):
+        return _pretty_segments(("(stopped early)", "yellow"))
+    return _pretty_segments(("(done)", "green"))
 
 
 def _http_status_fg(status: str) -> str | None:
@@ -319,12 +353,16 @@ def _http_status_fg(status: str) -> str | None:
     return None
 
 
-def _fetch_pretty(args_json: str, result: str) -> str:
-    """One-line summary for a ``fetch`` call: ``* Fetch GET https://… (200)``."""
+def _fetch_prefix(args_json: str) -> str:
     method = _json_str_arg(args_json, "method") or "GET"
     url = _json_str_arg(args_json, "url") or "?"
+    return _pretty_prefix(f"* Fetch {method} {url} ")
+
+
+def _fetch_detail(_args_json: str, result: str) -> str:
+    """``fetch`` outcome: HTTP status, or error."""
     if result.startswith("error:"):
-        return _pretty_line(f"* Fetch {method} {url} ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     status = ""
     for line in result.splitlines():
         if line.startswith("--- "):
@@ -332,7 +370,7 @@ def _fetch_pretty(args_json: str, result: str) -> str:
         if line.startswith("status="):
             status = line.partition("=")[2]
             break
-    return _pretty_line(f"* Fetch {method} {url} ", (f"({status or 'done'})", _http_status_fg(status)))
+    return _pretty_segments((f"({status or 'done'})", _http_status_fg(status)))
 
 
 _MUTATOR_VERBS: dict[str, str] = {
@@ -345,83 +383,107 @@ _MUTATOR_VERBS: dict[str, str] = {
 }
 
 
-def _mutator_pretty(name: str, args_json: str, result: str) -> str:
-    """One-line summary for a file mutation: ``* Wrote 'path' (120 chars)`` / ``* Edit 'path' (done)``."""
-    verb = _MUTATOR_VERBS[name]
+def _mutator_prefix(name: str, args_json: str) -> str:
     if name in {"copy_path", "move_path"}:
         src = _json_str_arg(args_json, "src") or "?"
         dst = _json_str_arg(args_json, "dst") or "?"
         target = f"'{src}' → '{dst}'"
     else:
-        path = _json_str_arg(args_json, "path") or "?"
-        target = f"'{path}'"
+        target = f"'{_json_str_arg(args_json, 'path') or '?'}'"
+    return _pretty_prefix(f"* {_MUTATOR_VERBS[name]} {target} ")
+
+
+def _mutator_detail(name: str, result: str) -> str:
+    """File mutation outcome: written size, done, or error."""
     if result.startswith("error:"):
-        return _pretty_line(f"* {verb} {target} ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     if name == "write_file":
         # write_file keeps its brief detail: ``wrote 120 characters to path``.
         match = re.match(r"wrote (\d+) characters to (.+)$", result)
         if match:
-            chars, path = match.groups()
-            return _pretty_line(f"* Write '{path}' ", (f"({chars} chars)", None))
-    return _pretty_line(f"* {verb} {target} ", ("(done)", "green"))
+            chars, _path = match.groups()
+            return _pretty_segments((f"({chars} chars)", None))
+    return _pretty_segments(("(done)", "green"))
 
 
-def _vcs_status_pretty(_args_json: str, result: str) -> str:
-    """One-line summary for ``vcs_status``: ``* VCS Status (done)``."""
+def _mutator_line(name: str) -> PrettyLine:
+    return PrettyLine(
+        prefix=lambda args: _mutator_prefix(name, args),
+        detail=lambda _args, result: _mutator_detail(name, result),
+    )
+
+
+def _vcs_status_prefix(_args_json: str) -> str:
+    return _pretty_prefix("* VCS Status ")
+
+
+def _vcs_status_detail(_args_json: str, result: str) -> str:
+    """``vcs_status`` outcome: done, or error."""
     if result.startswith("error:"):
-        return _pretty_line("* VCS Status ", (f"({result})", "red"))
-    return _pretty_line("* VCS Status ", ("(done)", "green"))
+        return _pretty_segments((f"({result})", "red"))
+    return _pretty_segments(("(done)", "green"))
 
 
-def _vcs_log_pretty(_args_json: str, result: str) -> str:
-    """One-line summary for ``vcs_log``: ``* VCS Log (5 commits)``."""
+def _vcs_log_prefix(_args_json: str) -> str:
+    return _pretty_prefix("* VCS Log ")
+
+
+def _vcs_log_detail(_args_json: str, result: str) -> str:
+    """``vcs_log`` outcome: commit count, none, or error."""
     if result.startswith("error:"):
-        return _pretty_line("* VCS Log ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     if result == "(no commits)":
-        return _pretty_line("* VCS Log ", ("(no commits)", "dim"))
+        return _pretty_segments(("(no commits)", "dim"))
     commits = [line for line in result.splitlines() if line.strip()]
     unit = "commit" if len(commits) == 1 else "commits"
-    return _pretty_line("* VCS Log ", (f"({len(commits)} {unit})", None))
+    return _pretty_segments((f"({len(commits)} {unit})", None))
 
 
-def _wait_pretty(_args_json: str, result: str) -> str:
-    """One-line summary for a ``wait`` call: ``* Wait (5s)`` / ``* Wait (disturbed)``."""
+def _wait_prefix(_args_json: str) -> str:
+    return _pretty_prefix("* Wait ")
+
+
+def _wait_detail(_args_json: str, result: str) -> str:
+    """``wait`` outcome: slept duration, disturbed, or error."""
     if result.startswith("error:"):
-        return _pretty_line("* Wait ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     if result.startswith("waited "):
-        return _pretty_line("* Wait ", (f"({result.removeprefix('waited ')})", "green"))
-    return _pretty_line("* Wait ", ("(disturbed)", "yellow"))
+        return _pretty_segments((f"({result.removeprefix('waited ')})", "green"))
+    return _pretty_segments(("(disturbed)", "yellow"))
 
 
-_PRETTY_BUILDERS: dict[str, Callable[[str, str], str]] = {
-    "read_file": _read_file_pretty,
-    "tree": _tree_pretty,
-    "todo_push": lambda _args, result: _todo_pretty("todo_push", result),
-    "todo_update": lambda _args, result: _todo_pretty("todo_update", result),
-    "listdir": _listdir_pretty,
-    "glob_paths": _glob_pretty,
-    "grep_files": _grep_pretty,
-    "run_argv": _run_argv_pretty,
-    "run_argv_batch": _run_argv_batch_pretty,
-    "fetch": _fetch_pretty,
-    "edit_replace": lambda args, result: _mutator_pretty("edit_replace", args, result),
-    "edit_lineno": lambda args, result: _mutator_pretty("edit_lineno", args, result),
-    "write_file": lambda args, result: _mutator_pretty("write_file", args, result),
-    "copy_path": lambda args, result: _mutator_pretty("copy_path", args, result),
-    "move_path": lambda args, result: _mutator_pretty("move_path", args, result),
-    "delete_path": lambda args, result: _mutator_pretty("delete_path", args, result),
-    "vcs_status": _vcs_status_pretty,
-    "vcs_log": _vcs_log_pretty,
-    "wait": _wait_pretty,
+# Pretty tools: ``prefix`` prints as the call starts, ``detail`` when it lands.
+_PRETTY_LINES: dict[str, PrettyLine] = {
+    "read_file": PrettyLine(prefix=_read_file_prefix, detail=_read_file_detail),
+    "tree": PrettyLine(prefix=_tree_prefix, detail=_tree_detail),
+    "todo_push": _todo_line("todo_push"),
+    "todo_update": _todo_line("todo_update"),
+    "listdir": PrettyLine(prefix=_listdir_prefix, detail=_listdir_detail),
+    "glob_paths": PrettyLine(prefix=_glob_prefix, detail=_glob_detail),
+    "grep_files": PrettyLine(prefix=_grep_prefix, detail=_grep_detail),
+    "run_argv": PrettyLine(prefix=_run_argv_prefix, detail=_run_argv_detail),
+    "run_argv_batch": PrettyLine(prefix=_run_argv_batch_prefix, detail=_run_argv_batch_detail),
+    "fetch": PrettyLine(prefix=_fetch_prefix, detail=_fetch_detail),
+    "edit_replace": _mutator_line("edit_replace"),
+    "edit_lineno": _mutator_line("edit_lineno"),
+    "write_file": _mutator_line("write_file"),
+    "copy_path": _mutator_line("copy_path"),
+    "move_path": _mutator_line("move_path"),
+    "delete_path": _mutator_line("delete_path"),
+    "vcs_status": PrettyLine(prefix=_vcs_status_prefix, detail=_vcs_status_detail),
+    "vcs_log": PrettyLine(prefix=_vcs_log_prefix, detail=_vcs_log_detail),
+    "wait": PrettyLine(prefix=_wait_prefix, detail=_wait_detail),
 }
 
+_PRETTY_TOOLS = frozenset(_PRETTY_LINES)
 
-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."""
-    builder = _PRETTY_BUILDERS.get(name)
-    if builder is None:
+
+def _pretty_parts(name: str, args_json: str, result: str) -> tuple[str, str] | None:
+    """Prefix + detail for a known tool call; None keeps the ``[tool]`` style."""
+    line = _PRETTY_LINES.get(name)
+    if line is None:
         return None
-    return builder(args_json, result)
+    return line.prefix(args_json), line.detail(args_json, result)
 
 
 def _echo_stream(text: str) -> None:
@@ -452,6 +514,63 @@ def _line_count_for_clear(label: str, body: str) -> int:
     return text.count("\n") + 1
 
 
+class _PrettyToolStream:
+    """Pretty tool lines: prefix at call time, detail appended when it lands.
+
+    Writing the prefix immediately keeps a blocking tool call visible; appending
+    the detail later reproduces the single whole-line write byte for byte, so
+    non-interactive output is unchanged. The calls of one batch are yielded
+    before their results, so a second call while a prefix is open (parallel
+    tools) cannot append its detail in order: that prefix is erased and the
+    batch falls back to whole lines.
+    """
+
+    _interactive: bool
+    _open: bool
+    _whole_lines: bool
+
+    def __init__(self, *, interactive: bool) -> None:
+        self._interactive = interactive
+        self._open = False
+        self._whole_lines = False
+
+    def start_call(self, prefix: str) -> None:
+        """Show *prefix* for a call that is about to run (cursor left mid-line)."""
+        if not (self._interactive and prefix) or self._whole_lines:
+            return
+        if self._open:
+            # Parallel batch: the open prefix could never receive its detail.
+            _clear_streamed_lines(1)
+            self._open = False
+            self._whole_lines = True
+            return
+        click.echo(f"\n{prefix}", nl=False)
+        with contextlib.suppress(OSError):
+            _ = sys.stdout.flush()
+        self._open = True
+
+    def finish_call(self, detail: str) -> bool:
+        """Append *detail* to the open prefix; False when a whole line is needed."""
+        if not (self._interactive and self._open):
+            return False
+        click.echo(f"{detail}\n", nl=False)
+        self._open = False
+        return True
+
+    def close_line(self) -> None:
+        """End a dangling prefix line before other output takes the next line."""
+        if not self._open:
+            return
+        click.echo()
+        self._open = False
+        self._whole_lines = True
+
+    def end_batch(self) -> None:
+        """Reset batch-scoped state once every call of the batch has a result."""
+        self.close_line()
+        self._whole_lines = False
+
+
 def print_markdown(text: str, *, label: str = "assistant:") -> None:
     """Render *text* as markdown via Rich; *label* on its own line when set."""
     from rich.console import Console
@@ -490,10 +609,15 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
     content source changes (reasoning ↔ assistant, or tools/errors), the
     assistant markdown buffer is flushed so streams do not mix and Rich can
     re-render completed assistant segments.
+
+    Pretty tool calls print their ``* Verb 'target' `` prefix as soon as the call
+    starts, so a blocking tool is visible while it runs; the outcome is appended
+    when the result lands (same bytes as the whole-line fallback).
     """
     show_full = get_verbose_tool_results() if verbose is None else verbose
     use_markdown = get_markdown_enabled() if markdown is None else markdown
     pretty = bool(use_markdown and markdown_render_available())
+    tool_lines = _PrettyToolStream(interactive=interactive_terminal() and not show_full)
 
     source: StreamSource | None = None
     assistant_buf: list[str] = []
@@ -521,6 +645,7 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
             return
         if source == "assistant":
             flush_assistant()
+        tool_lines.close_line()
         click.echo()
         click.secho("reasoning:", fg="bright_black")
         source = "reasoning"
@@ -533,6 +658,7 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
         if source == "reasoning":
             click.echo()  # end reasoning stream line
             source = None
+        tool_lines.close_line()
         click.echo()
         click.secho("assistant:", fg="cyan")
         source = "assistant"
@@ -554,26 +680,37 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
                 args = call.function.arguments
                 pretty_tool = name in _PRETTY_TOOLS
                 pending_tools.append((name, args, pretty_tool))
-                if not pretty_tool:
+                if pretty_tool:
+                    tool_lines.start_call(_PRETTY_LINES[name].prefix(args))
+                else:
+                    tool_lines.close_line()
                     preview = _preview(args, _TOOL_ARGS_PREVIEW)
                     click.secho(f"\n[tool] {name}({preview})", fg="yellow")
             else:
+                tool_lines.close_line()
                 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
             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 and not show_full:
-                click.echo(f"\n{pretty_line}")
+            parts = _pretty_parts(name, args, content) if pretty_tool else None
+            if parts is not None and not show_full:
+                prefix, detail = parts
+                if not tool_lines.finish_call(detail):
+                    click.echo(f"\n{prefix}{detail}")
             elif show_full:
+                tool_lines.close_line()
                 click.secho(f"[tool ok]\n{content}", fg="magenta")
             else:
+                tool_lines.close_line()
                 preview = _preview_result(content, _TOOL_RESULT_PREVIEW)
                 click.secho(f"[tool ok] {preview}", fg="magenta")
+            if not pending_tools:
+                tool_lines.end_batch()
         elif isinstance(event, ErrorEvent):
             flush_assistant()
+            tool_lines.close_line()
             suffix = ""
             if event.source:
                 suffix += f" source={event.source}"
@@ -582,12 +719,14 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
             click.secho(f"\n[error]{suffix} {event.message}", fg="bright_red")
         elif isinstance(event, CancelledEvent):
             flush_assistant()
+            tool_lines.close_line()
             if event.reason:
                 click.secho(f"\n[cancelled] {event.reason}", fg="yellow")
             else:
                 click.secho("\n[cancelled]", fg="yellow")
         elif isinstance(event, MaxRoundsEvent):
             flush_assistant()
+            tool_lines.close_line()
             if event.continued:
                 click.secho(
                     f"\n[max rounds {event.rounds} reached — continuing with a higher allowance]",
@@ -602,6 +741,7 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
             _ = event
 
     # End-of-turn: flush any open assistant segment; close reasoning stream.
+    tool_lines.close_line()
     if assistant_buf or printed_assistant:
         flush_assistant()
     elif printed_reasoning:

src/plyngent/cli/retry.py

@@ -8,6 +8,7 @@ from typing import TYPE_CHECKING
 
 import click
 
+from plyngent.agent import AssistantMessageEvent
 from plyngent.cli.display import render_events
 from plyngent.cli.interrupt import allow_task_cancel, set_sigint_reinstall
 from plyngent.cli.limits import reset_auto_continue_turn
@@ -186,6 +187,10 @@ async def run_turn_with_retries(
 
     Ctrl+C cancels the in-flight task; user message stays in DB for ``/retry``.
 
+    The retry budget counts *consecutive* failures: an attempt that completed a
+    model round proves the connection answered, so the next failure starts a
+    fresh budget (and the short first delay) again.
+
     The ``yyy`` limit auto-continue is turn-scoped: cleared before and after the
     turn, so a retry of the same turn keeps it but the next user turn prompts.
     """
@@ -196,7 +201,7 @@ async def run_turn_with_retries(
         reset_auto_continue_turn()
 
 
-async def _run_turn_with_retries(
+async def _run_turn_with_retries(  # noqa: C901 — attempt / cancel / error state machine
     agent: ChatAgent,
     *,
     starter: Callable[[], AsyncIterator[AgentEvent]],
@@ -204,10 +209,21 @@ async def _run_turn_with_retries(
 ) -> bool:
     max_retries = len(delays)
     attempt = 0
+    recovered = False
     current: Callable[[], AsyncIterator[AgentEvent]] = starter
+
+    async def note_completed_round(events: AsyncIterator[AgentEvent]) -> AsyncIterator[AgentEvent]:
+        """Yield *events*, recording that a full model round came back."""
+        nonlocal recovered
+        async for event in events:
+            if isinstance(event, AssistantMessageEvent):
+                recovered = True
+            yield event
+
     while True:
+        recovered = False
         try:
-            await run_cancellable(render_events(current()))
+            await run_cancellable(render_events(note_completed_round(current())))
         except asyncio.CancelledError:
             # Do not auto-retry cancelled turns — user intent was stop, not retry.
             _echo_cancel_lines("", "cancelled; user message kept — use /retry to try again", "")
@@ -219,6 +235,10 @@ async def _run_turn_with_retries(
             click.secho(f"error: {exc}", fg="red")
             if agent.pending_retry_text is not None:
                 current = agent.retry
+            if recovered:
+                # The connection answered a round before failing again, so this
+                # is a fresh outage: restart the budget and the delay schedule.
+                attempt = 0
             if attempt >= max_retries:
                 if agent.pending_retry_text is not None:
                     click.secho(

src/plyngent/cli/slash.py

@@ -980,10 +980,13 @@ def btw_cmd(
             return
         tools_mode = "no"
 
-    from plyngent.cli.display import render_events
+    from plyngent.agent import aside_notice
+    from plyngent.cli.display import echo_notice, render_events
     from plyngent.cli.retry import run_cancellable
     from plyngent.tools.context import SessionState, read_only_context
 
+    aside = aside_notice(tools_mode=tools_mode)
+    echo_notice(aside)
     click.secho("btw: ", fg="magenta", nl=False)
     click.echo(text)
     if tools_mode == "no":
@@ -1008,6 +1011,7 @@ def btw_cmd(
                 text,
                 include_history=not fresh,
                 tools=tools_arg,
+                notice=aside,
                 instance_state=aside_instance,
                 session_state=aside_session,
             )

src/plyngent/cli/state.py

@@ -11,6 +11,8 @@ from plyngent.agent.loop import DEFAULT_MAX_ROUNDS
 from plyngent.agent.todo_stack import TodoStack
 
 if TYPE_CHECKING:
+    from datetime import datetime
+
     from plyngent.agent.types import AnyLLMClient
 from plyngent.cli.models_source import (
     DEFAULT_MODELS_CACHE_TTL,
@@ -65,6 +67,8 @@ class ReplState:
     max_rounds: int = DEFAULT_MAX_ROUNDS
     stream_enabled: bool = True
     verbose: bool = False
+    # Less status on stderr (--quiet); host notices are still pushed to the model.
+    quiet: bool = False
     # End-of-turn Rich markdown for assistant text (TTY only).
     markdown_enabled: bool = True
     # One-shot / scripts: never prompt to raise tool-loop limits.
@@ -704,8 +708,12 @@ class ReplState:
             await self.new_session()
         return was_current
 
-    async def resume_session(self, session_id: int) -> None:
-        """Load a session; on workspace mismatch, prompt keep / rebind / abort."""
+    async def resume_session(self, session_id: int, *, restarted: bool = False) -> None:
+        """Load a session; on workspace mismatch, prompt keep / rebind / abort.
+
+        With *restarted* (a session resumed by a freshly started process) the
+        model gets a notice about the process-scoped state that a restart drops.
+        """
         from plyngent.cli.limits import prompt_workspace_mismatch
 
         self._reset_limit_auto_continue()
@@ -743,9 +751,15 @@ class ReplState:
         await self.agent.load_history()
         await self.load_todo_stack()
         await self.load_access_grants()
+        if restarted:
+            await self._notice_restart(last_active=row.updated_at)
 
-    async def resume_latest_or_new(self, name: str = "chat") -> str:
-        """Resume most recently updated session for this workspace, or create one."""
+    async def resume_latest_or_new(self, name: str = "chat", *, restarted: bool = False) -> str:
+        """Resume most recently updated session for this workspace, or create one.
+
+        With *restarted* the resumed session gets the new-process notice; a
+        freshly created session has nothing to explain, so it gets none.
+        """
         self._reset_limit_auto_continue()
         latest = await self.memory.get_latest_session(workspace=self.workspace)
         if latest is None:
@@ -760,9 +774,22 @@ class ReplState:
         await self.agent.load_history()
         await self.load_todo_stack()
         await self.load_access_grants()
+        if restarted:
+            # Before touch_session: the stamp should be the previous run's activity.
+            await self._notice_restart(last_active=latest.updated_at)
         _ = await self.memory.touch_session(latest.sid)
         return "resume"
 
+    async def _notice_restart(self, *, last_active: datetime | None) -> None:
+        """Tell the model — and the user — that this session outlived a restart."""
+        from plyngent.agent import resume_notice
+        from plyngent.cli.display import echo_notice
+
+        notice = resume_notice(last_active=last_active)
+        _ = await self.agent.push_notice(notice)
+        if not self.quiet:
+            echo_notice(notice, err=True)
+
     async def compact_to_new_session(self, *, name: str | None = None) -> tuple[int, int, str]:
         """Soft-compact + model-summarize current history into a new workspace session.
 

tests/test_agent/test_notices.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta, timezone
+
+from plyngent.agent import ChatAgent, Notice, aside_notice, format_moment, resume_notice
+from plyngent.agent.chat import incomplete_turn_user_text
+from plyngent.config.models import DatabaseConfig
+from plyngent.lmproto.openai_compatible.model import (
+    AssistantChatMessage,
+    AssistantFunctionTool,
+    AssistantFunctionToolCall,
+    DeveloperChatMessage,
+    SystemChatMessage,
+    ToolChatMessage,
+    UserChatMessage,
+)
+from plyngent.memory import MemoryStore
+from tests.test_agent.test_loop import ScriptedClient
+
+
+def test_notice_message_carries_marker_and_kind() -> None:
+    message = Notice(kind="resume", body="  hello there  ").to_message()
+    assert isinstance(message, DeveloperChatMessage)
+    assert message.content == "[plyngent notice: resume]\nhello there"
+
+
+def test_format_moment_localizes_and_labels_missing() -> None:
+    assert format_moment(None) == "unknown"
+    aware = datetime(2026, 9, 14, 22, 9, tzinfo=UTC)
+    assert format_moment(aware) == aware.astimezone().strftime("%Y-%m-%d %H:%M")
+    # SQLite hands back naive datetimes for tz-aware columns; read them as UTC.
+    assert format_moment(aware.replace(tzinfo=None)) == format_moment(aware)
+    offset = datetime(2026, 9, 14, 22, 9, tzinfo=timezone(timedelta(hours=8)))
+    assert format_moment(offset) == offset.astimezone().strftime("%Y-%m-%d %H:%M")
+
+
+def test_resume_notice_states_what_is_gone_and_restored() -> None:
+    notice = resume_notice(last_active=datetime(2026, 9, 14, 22, 9, tzinfo=UTC))
+    assert notice.kind == "resume"
+    assert "resumed in a new plyngent process" in notice.body
+    assert "PTY sessions" in notice.body
+    assert "grants were restored" in notice.body
+    assert notice.summary.startswith("resumed in a new process")
+    assert format_moment(datetime(2026, 9, 14, 22, 9, tzinfo=UTC)) in notice.summary
+
+
+def test_aside_notice_matches_tool_scope() -> None:
+    read = aside_notice(tools_mode="read")
+    assert read.kind == "aside"
+    assert "read-only" in read.body
+    assert "not saved" in read.body
+    assert "read tools" in read.summary
+
+    assert "Tools are disabled" in aside_notice(tools_mode="no").body
+    assert "Full tools are available" in aside_notice(tools_mode="full").body
+    # Unknown mode degrades to the read-only wording rather than an empty notice.
+    assert aside_notice(tools_mode="weird").body == read.body
+
+
+async def test_push_notice_appends_and_persists() -> None:
+    store = await MemoryStore.open(DatabaseConfig())
+    try:
+        session = await store.create_session(name="t")
+        _ = await store.append_message(session.sid, SystemChatMessage(content="sys"))
+        _ = await store.append_message(session.sid, UserChatMessage(content="hi"))
+        agent = ChatAgent(ScriptedClient([]), model="m", memory=store, session_id=session.sid)
+        await agent.load_history()
+
+        message = await agent.push_notice(resume_notice(last_active=None))
+
+        assert agent.messages[-1] is message
+        assert agent.persist_from == len(agent.messages)  # never re-persisted by a batch
+        loaded = await store.list_messages(session.sid)
+        assert isinstance(loaded[-1], DeveloperChatMessage)
+        assert loaded[-1].content.startswith("[plyngent notice: resume]")
+    finally:
+        await store.close()
+
+
+def test_trailing_developer_message_keeps_incomplete_turn_detectable() -> None:
+    """A pushed notice must not hide a turn that can still be retried."""
+    notice = resume_notice(last_active=None).to_message()
+    assert incomplete_turn_user_text([UserChatMessage(content="do it"), notice]) == "do it"
+    batch = [
+        UserChatMessage(content="do it"),
+        AssistantChatMessage(
+            content="",
+            tool_calls=[
+                AssistantFunctionToolCall(
+                    id="1",
+                    function=AssistantFunctionTool(name="ping", arguments="{}"),
+                )
+            ],
+        ),
+        ToolChatMessage(content="pong", tool_call_id="1"),
+        notice,
+    ]
+    assert incomplete_turn_user_text(batch) == "do it"
+    # A finished turn (trailing assistant) stays non-retryable past a notice.
+    done = [UserChatMessage(content="do it"), AssistantChatMessage(content="ok"), notice]
+    assert incomplete_turn_user_text(done) is None

tests/test_agent/test_run_aside.py

@@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, Any, Literal, cast, overload
 
 import pytest
 
-from plyngent.agent import ChatAgent, ToolRegistry, ToolTag, tool
+from plyngent.agent import ChatAgent, ToolRegistry, ToolTag, aside_notice, tool
 from plyngent.agent.todo_stack import TodoStack
 from plyngent.config.models import DatabaseConfig
 from plyngent.lmproto.openai_compatible.model import (
@@ -16,6 +16,7 @@ from plyngent.lmproto.openai_compatible.model import (
     ChatCompletionChunk,
     ChatCompletionResponse,
     ChatCompletionsParam,
+    DeveloperChatMessage,
     UserChatMessage,
 )
 from plyngent.memory import MemoryStore
@@ -120,6 +121,47 @@ async def test_run_aside_fresh_skips_history() -> None:
     assert users == ["q"]
 
 
+@pytest.mark.asyncio
+async def test_run_aside_notice_reaches_model_but_not_history() -> None:
+    """The aside notice is the model's last instruction; the host keeps none."""
+    client = ScriptedClient(["ok"])
+    agent = ChatAgent(cast("Any", client), model="m", stream=False)
+    agent.messages.append(UserChatMessage(content="prior main"))
+    before = list(agent.messages)
+    notice = aside_notice(tools_mode="read")
+
+    async for _ in agent.run_aside("side?", include_history=True, tools=False, notice=notice):
+        pass
+
+    payload = client.payloads[0]
+    assert isinstance(payload[-2], DeveloperChatMessage)
+    assert payload[-2].content == notice.to_message().content
+    assert "read-only" in payload[-2].content
+    assert isinstance(payload[-1], UserChatMessage)
+    assert payload[-1].content == "side?"
+    assert agent.messages == before
+
+
+@pytest.mark.asyncio
+async def test_run_aside_fresh_history_still_carries_notice() -> None:
+    client = ScriptedClient(["ok"])
+    agent = ChatAgent(cast("Any", client), model="m", stream=False)
+
+    async for _ in agent.run_aside(
+        "q",
+        include_history=False,
+        tools=False,
+        notice=aside_notice(tools_mode="no"),
+    ):
+        pass
+
+    payload = client.payloads[0]
+    assert isinstance(payload[0], DeveloperChatMessage)
+    assert "Tools are disabled" in payload[0].content
+    assert isinstance(payload[-1], UserChatMessage)
+    assert payload[-1].content == "q"
+
+
 @pytest.mark.asyncio
 async def test_run_aside_tools_clones_registry_with_fresh_session() -> None:
     hits: list[str] = []

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, ToolCallEvent, ToolResultEvent
+from plyngent.agent import ErrorEvent, ReasoningDeltaEvent, TextDeltaEvent, ToolCallEvent, ToolResultEvent
 from plyngent.cli.display import (
     get_markdown_enabled,
     markdown_render_available,
@@ -558,3 +558,94 @@ async def test_pretty_wait_error(capsys: pytest.CaptureFixture[str]) -> None:
     await render_events(_aiter([_pretty_call("wait", '{"duration": -1}'), _result(content)]))
     out = capsys.readouterr().out
     assert "* Wait (error: duration must not be negative)" in out
+
+
+async def test_pretty_prefix_shown_while_call_runs(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """An interactive terminal shows the prefix before the tool has a result."""
+    monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
+    while_running: list[str] = []
+
+    async def events() -> AsyncIterator[AgentEvent]:
+        yield _read_call('{"path": "a.txt"}')
+        while_running.append(capsys.readouterr().out)  # tool still running here
+        yield _result("L1-4\none\ntwo\n")
+
+    await render_events(events())
+    assert while_running == ["\n* Read 'a.txt' "]
+    assert capsys.readouterr().out == "L1-4 (done)\n\n"
+
+
+async def test_pretty_prefix_absent_when_not_a_tty(capsys: pytest.CaptureFixture[str]) -> None:
+    """Non-interactive output stays one whole-line write (pipes, logs, tests)."""
+    while_running: list[str] = []
+
+    async def events() -> AsyncIterator[AgentEvent]:
+        yield _read_call('{"path": "a.txt"}')
+        while_running.append(capsys.readouterr().out)
+        yield _result("L1-4\none\ntwo\n")
+
+    await render_events(events())
+    assert while_running == [""]
+    assert capsys.readouterr().out == "\n* Read 'a.txt' L1-4 (done)\n\n"
+
+
+async def test_pretty_verbose_keeps_full_result_output(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """Verbose mode stays a full log: no prefix line, whole result."""
+    monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
+    set_verbose_tool_results(True)
+    try:
+        await render_events(_aiter([_read_call('{"path": "a.txt"}'), _result("L1-4\none\n")]))
+    finally:
+        set_verbose_tool_results(False)
+    out = capsys.readouterr().out
+    assert "[tool ok]\nL1-4\none\n" in out
+    assert "* Read 'a.txt' " not in out
+
+
+async def test_pretty_parallel_batch_erases_open_prefix(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """A second call in the batch erases the open prefix; results print whole."""
+    monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
+    await render_events(
+        _aiter(
+            [
+                _pretty_call("read_file", '{"path": "a.txt"}'),
+                _pretty_call("read_file", '{"path": "b.txt"}'),
+                _result("L1-4\none\n"),
+                _result("L1-9\ntwo\n"),
+            ]
+        )
+    )
+    out = capsys.readouterr().out
+    assert "\x1b[2K\x1b[1A" in out  # the in-flight prefix was erased
+    assert "* Read 'a.txt' L1-4 (done)\n" in out
+    assert "* Read 'b.txt' L1-9 (done)\n" in out
+
+
+async def test_pretty_prefix_closed_before_error_line(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """Another event ends an open prefix line instead of appending to it."""
+    monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
+    await render_events(
+        _aiter(
+            [
+                _read_call('{"path": "a.txt"}'),
+                ErrorEvent(message="boom", source="tool"),
+                _result("error: tool 'read_file' failed: boom"),
+            ]
+        )
+    )
+    out = capsys.readouterr().out
+    assert "* Read 'a.txt' \n" in out
+    assert "[error] source=tool boom\n" in out
+    assert "* Read 'a.txt' (error)\n" in out

tests/test_cli/test_repl_commands.py

@@ -5,7 +5,7 @@ from typing import TYPE_CHECKING, Literal, overload
 import pytest
 import tomlkit
 
-from plyngent.agent import ChatAgent
+from plyngent.agent import ChatAgent, Notice
 from plyngent.cli.slash import handle_slash
 from plyngent.cli.state import ReplState
 from plyngent.config.models import DatabaseConfig, OpenAIProvider
@@ -764,3 +764,63 @@ async def test_btw_full_mode_clones_full_registry(
     assert captured["tools"] is True
     assert captured["session_state"] is not state.session_state
     assert captured["read_only"] is False
+
+
+async def test_btw_sends_notice_and_shows_it(
+    state: ReplState,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """The aside request carries a notice; it never enters the main transcript."""
+    captured: dict[str, object] = {}
+
+    async def fake_run_aside(_text: str, **kwargs: object) -> AsyncIterator[object]:
+        captured.update(kwargs)
+        if False:
+            yield None
+
+    monkeypatch.setattr(state.agent, "run_aside", fake_run_aside)
+    before = list(state.agent.messages)
+
+    assert await handle_slash(state, "/btw --tools no hello") is True
+
+    notice = captured["notice"]
+    assert isinstance(notice, Notice)
+    assert notice.kind == "aside"
+    assert "Tools are disabled" in notice.body
+    # Side turns are not persisted, so nothing is pushed into the main history.
+    assert state.agent.messages == before
+    out = capsys.readouterr().out
+    assert "[notice] aside:" in out
+    assert "btw: hello" in out
+
+
+async def test_btw_read_notice_states_read_only_scope(
+    state: ReplState,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    from plyngent.agent import ToolRegistry, ToolTag, tool
+
+    @tool(tags=ToolTag.LOCAL | ToolTag.READ_ONLY, register=False)
+    async def ro_tool() -> str:
+        return "ro"
+
+    state.tools_enabled = True
+    state.agent.tools = ToolRegistry([ro_tool], auto_bind_state=True)
+    captured: dict[str, object] = {}
+
+    async def fake_run_aside(_text: str, **kwargs: object) -> AsyncIterator[object]:
+        captured.update(kwargs)
+        if False:
+            yield None
+
+    monkeypatch.setattr(state.agent, "run_aside", fake_run_aside)
+
+    assert await handle_slash(state, "/btw hi") is True
+
+    notice = captured["notice"]
+    assert isinstance(notice, Notice)
+    assert "read-only" in notice.body
+    assert "not saved" in notice.body
+    assert "[notice] aside: read tools" in capsys.readouterr().out

tests/test_cli/test_resume_notice.py

@@ -0,0 +1,152 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Literal, overload
+
+import pytest
+import tomlkit
+
+from plyngent.agent import ChatAgent
+from plyngent.cli.app import _bind_session
+from plyngent.cli.state import ReplState
+from plyngent.config.models import DatabaseConfig, OpenAIProvider
+from plyngent.config.store import ConfigStore
+from plyngent.lmproto.openai_compatible.model import (
+    AssistantChatMessage,
+    ChatCompletionChunk,
+    ChatCompletionResponse,
+    ChatCompletionsParam,
+    DeveloperChatMessage,
+    UserChatMessage,
+)
+from plyngent.memory import MemoryStore
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+    from pathlib import Path
+
+
+class DummyClient:
+    """Client stub: these tests never run a model turn."""
+
+    @overload
+    async def chat_completions(
+        self, param: ChatCompletionsParam, *, stream: Literal[False] = False
+    ) -> ChatCompletionResponse: ...
+
+    @overload
+    async def chat_completions(
+        self, param: ChatCompletionsParam, *, stream: Literal[True]
+    ) -> AsyncIterator[ChatCompletionChunk]: ...
+
+    async def chat_completions(
+        self, param: ChatCompletionsParam, *, stream: bool = False
+    ) -> ChatCompletionResponse | AsyncIterator[ChatCompletionChunk]:
+        del param, stream
+        msg = "no model call expected"
+        raise AssertionError(msg)
+
+
+@pytest.fixture
+async def state(tmp_path: Path) -> AsyncIterator[ReplState]:
+    memory = await MemoryStore.open(DatabaseConfig())
+    provider = OpenAIProvider(access_key_or_token="sk-test")
+    config = ConfigStore(path=tmp_path / "plyngent.toml", document=tomlkit.document())
+    config.providers = {"local": provider}
+    st = ReplState(
+        config=config,
+        memory=memory,
+        workspace=tmp_path,
+        provider_name="local",
+        provider=provider,
+        model="gpt-test",
+        tools_enabled=False,
+    )
+    st.client = DummyClient()
+    st.agent = ChatAgent(st.client, model=st.model, memory=st.memory, session_id=None)
+    yield st
+    await memory.close()
+
+
+async def _seed_finished_turn(state: ReplState) -> int:
+    """Start a session and give it one finished exchange; return its id."""
+    await state.new_session("t")
+    sid = state.session_id
+    assert sid is not None
+    _ = await state.memory.append_message(sid, UserChatMessage(content="hello"))
+    _ = await state.memory.append_message(sid, AssistantChatMessage(content="hi there"))
+    return sid
+
+
+async def test_startup_resume_notifies_model_and_user(
+    state: ReplState,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    sid = await _seed_finished_turn(state)
+
+    await _bind_session(state, session_id=sid, new_session=False, oneshot=False, quiet=False)
+
+    rows = await state.memory.list_messages(sid)
+    tail = rows[-1]
+    assert isinstance(tail, DeveloperChatMessage)
+    assert tail.content.startswith("[plyngent notice: resume]")
+    assert "resumed in a new plyngent process" in tail.content
+    assert "grants were restored" in tail.content
+    # The live agent holds the same notice after loading history.
+    assert state.agent.messages[-1].content == tail.content
+    captured = capsys.readouterr()
+    assert "[notice] resume:" in captured.err
+    assert "resumed session" in captured.err
+
+
+async def test_quiet_startup_resume_still_notifies_model(
+    state: ReplState,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    state.quiet = True
+    sid = await _seed_finished_turn(state)
+
+    await _bind_session(state, session_id=sid, new_session=False, oneshot=False, quiet=True)
+
+    rows = await state.memory.list_messages(sid)
+    assert isinstance(rows[-1], DeveloperChatMessage)
+    assert "[notice]" not in capsys.readouterr().err
+
+
+async def test_mid_process_resume_has_no_restart_notice(
+    state: ReplState,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """``/resume`` in a live process loses nothing, so nothing is announced."""
+    sid = await _seed_finished_turn(state)
+    before = len(await state.memory.list_messages(sid))
+
+    await state.resume_session(sid)
+
+    assert len(await state.memory.list_messages(sid)) == before
+    assert "[notice]" not in capsys.readouterr().err
+
+
+async def test_fresh_session_gets_no_restart_notice(state: ReplState) -> None:
+    mode = await state.resume_latest_or_new(restarted=True)
+
+    assert mode == "new"
+    sid = state.session_id
+    assert sid is not None
+    assert await state.memory.list_messages(sid) == []
+
+
+async def test_latest_resume_stamps_previous_activity(
+    state: ReplState,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    sid = await _seed_finished_turn(state)
+
+    mode = await state.resume_latest_or_new(restarted=True)
+
+    assert mode == "resume"
+    rows = await state.memory.list_messages(sid)
+    tail = rows[-1]
+    assert isinstance(tail, DeveloperChatMessage)
+    # touch_session must not overwrite the stamp with the resume time.
+    assert "previous run's last activity: unknown" not in tail.content
+    assert "[notice] resume:" in capsys.readouterr().err

tests/test_cli/test_retry.py

@@ -4,7 +4,7 @@ from typing import TYPE_CHECKING, Literal, overload
 
 import pytest
 
-from plyngent.agent import ChatAgent
+from plyngent.agent import ChatAgent, ToolRegistry, tool
 from plyngent.cli.retry import (
     DEFAULT_MAX_AUTO_RETRIES,
     DEFAULT_RETRY_DELAYS_SECONDS,
@@ -17,6 +17,8 @@ from plyngent.cli.retry import (
 from plyngent.config.models import DatabaseConfig
 from plyngent.lmproto.openai_compatible.model import (
     AssistantChatMessage,
+    AssistantFunctionTool,
+    AssistantFunctionToolCall,
     ChatCompletionChoice,
     ChatCompletionChunk,
     ChatCompletionResponse,
@@ -365,3 +367,120 @@ async def test_manual_retry_after_exhausted(monkeypatch: pytest.MonkeyPatch) ->
     assert loaded[0].content == "hold-me"
     assert sum(1 for m in loaded if isinstance(m, UserChatMessage)) == 1
     await store.close()
+
+
+class StepClient:
+    """Scripted chat completions: each step is a response or a raised exception."""
+
+    steps: list[ChatCompletionResponse | Exception]
+    calls: int
+
+    def __init__(self, steps: list[ChatCompletionResponse | Exception]) -> None:
+        self.steps = list(steps)
+        self.calls = 0
+
+    @overload
+    async def chat_completions(
+        self, param: ChatCompletionsParam, *, stream: Literal[False] = False
+    ) -> ChatCompletionResponse: ...
+
+    @overload
+    async def chat_completions(
+        self, param: ChatCompletionsParam, *, stream: Literal[True]
+    ) -> AsyncIterator[ChatCompletionChunk]: ...
+
+    async def chat_completions(
+        self, param: ChatCompletionsParam, *, stream: bool = False
+    ) -> ChatCompletionResponse | AsyncIterator[ChatCompletionChunk]:
+        del param
+        self.calls += 1
+        step = self.steps.pop(0) if self.steps else RuntimeError("no more scripted steps")
+        if isinstance(step, Exception):
+            raise step
+        if stream:
+            return _as_stream(step)
+        return step
+
+
+def _tool_call(name: str) -> AssistantChatMessage:
+    return AssistantChatMessage(
+        content="",
+        tool_calls=[
+            AssistantFunctionToolCall(
+                id="1",
+                function=AssistantFunctionTool(name=name, arguments="{}"),
+            )
+        ],
+    )
+
+
+def _capture_waits(monkeypatch: pytest.MonkeyPatch) -> list[float]:
+    """Record retry waits instead of sleeping; returns the collected delays."""
+    waits: list[float] = []
+
+    async def fake_sleep(seconds: float) -> bool:
+        waits.append(seconds)
+        return True
+
+    monkeypatch.setattr("plyngent.cli.retry.sleep_cancellable", fake_sleep)
+    return waits
+
+
+async def test_retry_budget_resets_after_a_completed_round(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """A round the model answered restarts the retry budget and its delays."""
+    waits = _capture_waits(monkeypatch)
+
+    @tool(register=False)
+    def ping() -> str:
+        return "pong"
+
+    client = StepClient(
+        [
+            RuntimeError("connection lost"),  # attempt 1: no round completed
+            _response(_tool_call("ping")),  # attempt 2, round 1: answered
+            RuntimeError("connection lost"),  # attempt 2, round 2: dropped again
+            _response(AssistantChatMessage(content="done")),  # attempt 3: completes
+        ]
+    )
+    agent = ChatAgent(client, model="m", tools=ToolRegistry([ping]), stream=False)
+
+    ok = await run_turn_with_retries(
+        agent,
+        starter=lambda: agent.run("hold-me"),
+        delays=(5.0, 10.0, 15.0),
+    )
+    out = capsys.readouterr().out
+    assert ok is True
+    assert waits == [5.0, 5.0]  # the delay schedule restarted from the first step
+    assert out.count("auto-retry 1/3") == 2
+    assert "auto-retry 2/3" not in out
+
+
+async def test_retry_budget_keeps_counting_without_progress(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """Consecutive failures with no completed round share one budget."""
+    waits = _capture_waits(monkeypatch)
+    client = StepClient(
+        [
+            RuntimeError("connection lost"),
+            RuntimeError("connection lost"),
+            _response(AssistantChatMessage(content="done")),
+        ]
+    )
+    agent = ChatAgent(client, model="m", stream=False)
+
+    ok = await run_turn_with_retries(
+        agent,
+        starter=lambda: agent.run("hold-me"),
+        delays=(5.0, 10.0, 15.0),
+    )
+    out = capsys.readouterr().out
+    assert ok is True
+    assert waits == [5.0, 10.0]
+    assert "auto-retry 1/3" in out
+    assert "auto-retry 2/3" in out