release: plyngent 0.1.3

Bump package version; refresh lock (ruff 0.15.22, platformdirs 4.10.1).
prek.toml already tracks latest ruff/basedpyright via additional_dependencies.

CLAUDE.md

@@ -78,15 +78,15 @@ 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; clearer deny messages.
+- **`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`, `edit_lineno` (1-based range), `copy_path` / `move_path` / `delete_path`, `new_temporary_workspace` (system temp allowlist; cleanup on chat exit; no session rebind).
+- **`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), `copy_path` / `move_path` / `delete_path`, `new_temporary_workspace` (system temp allowlist; cleanup on chat exit; no session rebind).
-- **`process`**: `run_command` (argv, no shell, timeout, optional stdin/env); PTY `open_pty` / `read_pty` / `write_pty` (literal) / `write_pty_keys` (escapes) / `close_pty` (POSIX: `pty`+`fork`; Windows: ConPTY via `pywinpty` env marker dep).
+- **`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).
-- 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`); session limit/idle TTL/output budget; close terminate→kill.
+- 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).
-- **`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; agent injects a developer review message if open todos and no todo tool use in the turn.
+- **`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`).
 - **`DEFAULT_TOOLS`**: file + process + vcs + chat + todo tool list for a `ToolRegistry`.
 
 ### Prompting (`prompting.py`)
@@ -97,11 +97,11 @@ Shared interactive I/O: `ask` / `choose` / `form` / `confirm` with pluggable bac
 
 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 if unset/`:memory:`); workspace-bound; resume latest for cwd/`--workspace` by default (`--new` / `--session`). One-shot: `-p/--prompt` and non-TTY stdin; exit codes 0/1/2/3; `--yes` (YOLO on), `--stream/--no-stream`, `--quiet`. Root `--log-level`.
+- **`plyngent chat`**: provider/model (flags or interactive; Tab via readline in `prompting`); sessions store `provider_name`/`model` and restore on resume; SQLite via `[database]` (file DB under user data when url unset; explicit `url = ":memory:"` kept + warn); workspace-bound; resume latest for cwd/`--workspace` by default (`--new` / `--session`). One-shot: `-p/--prompt` and non-TTY stdin; exit codes 0/1/2/3; `--yes` (YOLO on), `--stream/--no-stream`, `--quiet`. Root `--log-level`.
-- Slash: Click group in `cli/slash.py` + `awaitlet` for async work; Tab completer from registry + ParamType `shell_complete`. Multiline `"""` … `"""`; `/edit` via `$EDITOR`. `/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.
+- 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.
 - Explicit `/resume` or `--session` from another workspace prompts: **keep** / **update** / **abort**.
 - Failed/cancelled turns: user kept; **committed tool rounds kept** (side effects not re-run on `/retry`); only unfinished assistant rolled back; Ctrl+C cancels; TTY confirms off-loop; auto-retry 10s/20s/30s then `/retry`.
-- **`plyngent providers`**, **`config path|edit`**. No providers + `$EDITOR` → optional edit then reload.
+- **`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.
 
 ### Composition utility: `Forward` descriptor

README.md

@@ -86,7 +86,7 @@ Config: `prek.toml`. CI still runs the same checks in GitHub Actions.
 ```bash
 # 1) Create / open config
 plyngent config path
-plyngent config edit    # needs $EDITOR
+plyngent config edit    # $VISUAL/$EDITOR, else system open (xdg-open/open/startfile)
 
 # Minimal provider (OpenAI platform — Responses API; preset defaults to openai):
 # [providers.oai]
@@ -111,7 +111,7 @@ Default config path (platformdirs):
 
 ```bash
 plyngent config path
-plyngent config edit    # opens $EDITOR (e.g. codium --wait)
+plyngent config edit    # $VISUAL/$EDITOR (e.g. codium --wait), else system default
 ```
 
 Copy the example and fill in a real token:
@@ -140,7 +140,7 @@ max_context_tokens = 200000
 
 Supported provider presets today: `openai` (default if `preset` is omitted; default models `gpt-5.4` / `gpt-5.4-mini` / `gpt-5.4-nano` when `models` is omitted), `openai-compatible`, `deepseek` (OpenAI convention; default models `deepseek-v4-flash` and `deepseek-v4-pro` if `models` is omitted). Anthropic presets are modeled in config but not wired in the runtime client yet.
 
-If `[database]` is omitted (or SQLite `url` is empty/`":memory:"`), chat uses a durable file under the user data dir (e.g. `~/.local/share/plyngent/chat.db` on Linux).
+If `[database]` is omitted (or SQLite `url` is unset/empty), chat uses a durable file under the user data dir (e.g. `~/.local/share/plyngent/chat.db` on Linux). Set `url = ":memory:"` for a true in-memory SQLite (CLI warns; no file; useful for tests).
 
 ## Chat
 
@@ -186,7 +186,7 @@ Exit codes (one-shot):
 ### Input ergonomics
 
 - **Multiline**: start a message with `"""`, end a later line with `"""`.
-- **`/edit`**: compose a turn in `$EDITOR` (empty buffer cancels).
+- **`/edit`**: compose a turn in `$VISUAL`/`$EDITOR` (blocking only; empty cancels).
 - **Tab**: completes slash commands and some arguments (provider, model, on/off, export, `/help` targets).
 
 ### Slash commands
@@ -211,7 +211,7 @@ Type `/help` in the REPL for the live list. Common ones:
 | `/models` | List config + remote `GET /models` (always re-fetches) |
 | `/models --persist` | Merge remote catalog into TOML for this provider |
 | `/todos` | Todo/task stack: list, push, pop, done, clear |
-| `/config` | Edit `plyngent.toml` in `$EDITOR` and reload |
+| `/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.
@@ -235,6 +235,7 @@ Safety defaults:
   Prefer file tools over full-screen editors (`vim`/`nano`) for edits. `read_pty` sanitizes CSI/controls
   so tool results cannot reprogram the host TTY (no host terminal reset on exit).
   `write_pty` is literal text only; use `write_pty_keys` for `\xHH`, `ctrl+x`, `key=esc|enter|…`.
+  For passwords/sudo/ssh prompts use `ask_into_pty` (human types locally; answer never returns to the model).
 
 ## Usage / context (CLI)
 

doc/plyngent.example.toml

@@ -2,11 +2,13 @@
 # Copy to the path shown by: plyngent config path
 # Then set access_key_or_token and adjust models.
 
-# Optional. If omitted (or SQLite url is empty / ":memory:"), the CLI uses a
+# Optional. If omitted (or SQLite url unset/empty), the CLI uses a durable file under
-# durable file under the platform user data dir (e.g. ~/.local/share/plyngent/chat.db).
+# the platform user data dir (e.g. ~/.local/share/plyngent/chat.db).
+# url = ":memory:" → true in-memory SQLite (CLI warns; useful for tests).
 # [database]
 # implementation = "sqlite"
 # url = "/path/to/chat.db"
+# # url = ":memory:"
 
 [agent]
 system_prompt = "You are a careful coding assistant. Prefer small, verified edits."
@@ -16,6 +18,10 @@ confirm_destructive = true
 path_denylist = ["/secrets/", ".ssh/", ".gnupg/"]
 # Soft context budget in estimated tokens (~4 chars/token when API usage missing).
 max_context_tokens = 200000
+# How open-todo nags are injected into model context:
+# developer (default) | user | synthetic_tool | none
+# (legacy system is treated as developer)
+# todo_nag_strategy = "developer"
 
 # Optional: custom compact/summarisation prompts (empty = built-in defaults).
 # compact_system_prompt = ""

pdm.lock

@@ -326,13 +326,13 @@ files = [
 
 [[package]]
 name = "platformdirs"
-version = "4.10.0"
+version = "4.10.1"
 requires_python = ">=3.10"
 summary = "A small Python package for determining appropriate platform-specific dirs, e.g. a `user data dir`."
 groups = ["default"]
 files = [
-    {file = "platformdirs-4.10.0-py3-none-any.whl", hash = "sha256:fb516cdb12eb0d857d0cd85a7c57cea4d060bee4578d6cf5a14dfdf8cbf8784a"},
+    {file = "platformdirs-4.10.1-py3-none-any.whl", hash = "sha256:0e4eff26be2d75293977f7cddc153fd9b8eaa7fb0c7b64ffe4076cb443117443"},
-    {file = "platformdirs-4.10.0.tar.gz", hash = "sha256:31e761a6a0ca04faf7353ea759bdba55652be214725111e5aac52dfa29d4bef7"},
+    {file = "platformdirs-4.10.1.tar.gz", hash = "sha256:ceab4084426fe6319ce18e86deada8ab1b7487c7aee7040c55e277c9ae793695"},
 ]
 
 [[package]]
@@ -470,29 +470,29 @@ files = [
 
 [[package]]
 name = "ruff"
-version = "0.15.21"
+version = "0.15.22"
 requires_python = ">=3.7"
 summary = "An extremely fast Python linter and code formatter, written in Rust."
 groups = ["dev"]
 files = [
-    {file = "ruff-0.15.21-py3-none-linux_armv6l.whl", hash = "sha256:63ea0e965e5d73c90e95b2434beeafc70820536717f561b32ab6e777cb9bdf5d"},
+    {file = "ruff-0.15.22-py3-none-linux_armv6l.whl", hash = "sha256:44423e73493737f5e7c5b41d475483898ff37afcdae38bc3da5085e29af1c2d8"},
-    {file = "ruff-0.15.21-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:0f212c5d7d54c01bbfe6dcab02b724a39300f3e34ed7acbe995ccb320a2c58bd"},
+    {file = "ruff-0.15.22-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:b82c6482946e9eda7ff2e091d25b8bad3f718684e1916d41bd56873cee05b697"},
-    {file = "ruff-0.15.21-py3-none-macosx_11_0_arm64.whl", hash = "sha256:e6312e41bc96791299614995ea3a977c5857c3b5662b1ecef6755b02b87cb646"},
+    {file = "ruff-0.15.22-py3-none-macosx_11_0_arm64.whl", hash = "sha256:11c1c715af53a09f714e011106bffc419751ec8232fcb5da42173284ea3fec6f"},
-    {file = "ruff-0.15.21-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:01d65b4831c6b2a4ba8ee6faa84049d44d982b7a706e622c4094c509e51673be"},
+    {file = "ruff-0.15.22-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:742a29cf29bddb7c8327895d6a10e0e6c5b38a96dd407af9b5d0857f809c0576"},
-    {file = "ruff-0.15.21-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:2c5a913a589120ce67933d5d05fd6ddbcc2481c6a054980ee767f7414c72b4fd"},
+    {file = "ruff-0.15.22-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:72af58b951b0ae395935ae79763dc349bc0eb706319d28f7a33ad2cfb3cfc178"},
-    {file = "ruff-0.15.21-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:5ef04b681d02ad4dc9620f00f83ac5c22f652d0e9a9cfe431d219b16ad5ccc41"},
+    {file = "ruff-0.15.22-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:62d425005c1835eb24e2ee4161cb90e8db263415f4a71c8c72c33abaa6c0c224"},
-    {file = "ruff-0.15.21-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:16d090c0740916594157e75b80d666eab8e78083b39b3b0e1d698f4670a17b86"},
+    {file = "ruff-0.15.22-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:e8b9b3f8779a4f08c969defc3c8c35abffaa757e601ed5ae66d6d1db6519969a"},
-    {file = "ruff-0.15.21-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:3a10e74757dd65004d779b73e2f3c5210156d9980b41224d50d2ebcf1db51e67"},
+    {file = "ruff-0.15.22-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:1e0dd1b2e4d3d585f897a0d137cbf4eaf6223bef4e8ce34d6bb12556c5f9249e"},
-    {file = "ruff-0.15.21-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:bab0905d2f29e0d9fbc3c373ed23db0095edaa3f71f1f4f519ec15134d9e85c8"},
+    {file = "ruff-0.15.22-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:365523eb91d9224e1bcb03b022fbf0facb8f9e23792a2c53d9d4b3924bdbdebb"},
-    {file = "ruff-0.15.21-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:00eca240af5789fec6fe7df74c088cc1f9644ed83027113468efba7c92b94075"},
+    {file = "ruff-0.15.22-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:fabfd168afdf29fee5be98b831efa9683c94d7c5a3b58b9ce5a2e38444589a74"},
-    {file = "ruff-0.15.21-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:262ab31557a75141325e32d3357f3597645a7f084e732b6b054dde428ecd9341"},
+    {file = "ruff-0.15.22-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:225dbf095a87f1d9f90f5fd7924d2613ee452a75a4308c63a8f50f761787aa7c"},
-    {file = "ruff-0.15.21-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:659c4e7a4212f83306045ec7c5e5a356d16d9a6ef4ae0c7a4d872914fc655d9d"},
+    {file = "ruff-0.15.22-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:1877d63b9d24ed278744f1523fd11b85540566d54641f97c566d7d9dc5ca5296"},
-    {file = "ruff-0.15.21-py3-none-musllinux_1_2_i686.whl", hash = "sha256:9e866eab611a5f959d36df2d10e446973a3610bc42b0c15b31dc27977d59c233"},
+    {file = "ruff-0.15.22-py3-none-musllinux_1_2_i686.whl", hash = "sha256:a1606c510bd7215680d32efab38965f7cdec3ef69f5170a3f4791404ffdd5262"},
-    {file = "ruff-0.15.21-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:e89bc93c0d3803ba870b55c29671bad9dc6d94bb1eb181b056b52eb05b52854f"},
+    {file = "ruff-0.15.22-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:630479b18625f5ffc373f77603a22a9f8ac0acd7ff0501178b5db28ec71e9c64"},
-    {file = "ruff-0.15.21-py3-none-win32.whl", hash = "sha256:01f8d5be84823c172b389e123174f781f9daf86d6c58719d603f941932195cdd"},
+    {file = "ruff-0.15.22-py3-none-win32.whl", hash = "sha256:e5ba0e4a13fd14abbed2a77b517a3911290c6c6c59ef67784328d1668fab76cf"},
-    {file = "ruff-0.15.21-py3-none-win_amd64.whl", hash = "sha256:d4b8d9a2f0f12b816b50447f6eccb9f4bb01a6b82c86b50fb3b5354b458dc6d3"},
+    {file = "ruff-0.15.22-py3-none-win_amd64.whl", hash = "sha256:9be63ba1eb936acd2d1342fb8337c356353706fce233b2a15a09a97037e6acde"},
-    {file = "ruff-0.15.21-py3-none-win_arm64.whl", hash = "sha256:6e83115d4b9377c1cbc13abf0e051f069fab0ef815ea0504a8a008cee24dd0a8"},
+    {file = "ruff-0.15.22-py3-none-win_arm64.whl", hash = "sha256:e1168075b72158510839f250027659cdd78476f40507dd517892304c41318661"},
-    {file = "ruff-0.15.21.tar.gz", hash = "sha256:d0cfc841c572283c36548f82664a54ce6565567f1b0d5b4cf2caac693d8b7500"},
+    {file = "ruff-0.15.22.tar.gz", hash = "sha256:3f15175b1fb580126f58285a5dae6b2ea89000136d980c64499211f116b54809"},
 ]
 
 [[package]]

pyproject.toml

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

src/plyngent/agent/chat.py

@@ -19,6 +19,11 @@ from .budget import (
 )
 from .events import UsageEvent
 from .loop import DEFAULT_MAX_ROUNDS, run_chat_loop
+from .todo_nag import (
+    DEFAULT_TODO_NAG_STRATEGY,
+    inject_todo_nag_for_stack_with_events,
+    parse_todo_nag_strategy,
+)
 from .usage import TokenUsage
 
 if TYPE_CHECKING:
@@ -29,6 +34,7 @@ if TYPE_CHECKING:
 
     from .client import ChatClient
     from .events import AgentEvent
+    from .todo_nag import TodoNagStrategy
     from .todo_stack import TodoStack
     from .tools import ToolRegistry
 
@@ -120,6 +126,7 @@ class ChatAgent:
     parallel_tools: bool
     max_context_tokens: int
     todo_stack: TodoStack | None
+    todo_nag_strategy: TodoNagStrategy
     messages: list[AnyChatMessage]
     session_usage: TokenUsage
     last_turn_usage: TokenUsage
@@ -146,6 +153,7 @@ class ChatAgent:
         parallel_tools: bool = True,
         max_context_tokens: int = DEFAULT_CONTEXT_MAX_TOKENS,
         todo_stack: TodoStack | None = None,
+        todo_nag_strategy: str | TodoNagStrategy = DEFAULT_TODO_NAG_STRATEGY,
     ) -> None:
         self.client = client
         self.model = model
@@ -161,6 +169,7 @@ class ChatAgent:
         self.parallel_tools = parallel_tools
         self.max_context_tokens = max_context_tokens
         self.todo_stack = todo_stack
+        self.todo_nag_strategy = parse_todo_nag_strategy(str(todo_nag_strategy))
         self.messages = list(messages) if messages is not None else []
         self.session_usage = TokenUsage()
         self.last_turn_usage = TokenUsage()
@@ -301,6 +310,18 @@ class ChatAgent:
         turn_rounds = 0
         last_request = TokenUsage()
         try:
+            # Turn-start nag before first completion; yield events so CLI flushes
+            # (synthetic_tool → ToolCall/Result chrome, not glued to text).
+            if self.todo_stack is not None and not self.todo_stack.is_empty():
+                _injected, nag_events = inject_todo_nag_for_stack_with_events(
+                    self.messages,
+                    self.todo_stack,
+                    kind="turn_start",
+                    strategy=self.todo_nag_strategy,
+                )
+                for nag_event in nag_events:
+                    yield nag_event
+
             async for event in run_chat_loop(
                 self.client,
                 self.messages,
@@ -314,6 +335,7 @@ class ChatAgent:
                 parallel_tools=self.parallel_tools,
                 max_context_tokens=self.max_context_tokens,
                 todo_stack=self.todo_stack,
+                todo_nag_strategy=self.todo_nag_strategy,
             ):
                 if isinstance(event, UsageEvent):
                     turn_rounds += 1

src/plyngent/agent/loop.py

@@ -13,7 +13,6 @@ from plyngent.lmproto.openai_compatible.model import (
     AssistantChatMessage,
     AssistantFunctionToolCall,
     ChatCompletionsParam,
-    DeveloperChatMessage,
     StreamOptions,
     StreamToolCallDelta,
     ToolChatMessage,
@@ -38,6 +37,7 @@ from .events import (
     ToolResultEvent,
     UsageEvent,
 )
+from .todo_nag import DEFAULT_TODO_NAG_STRATEGY, inject_todo_nag_for_stack_with_events
 from .usage import resolve_round_usage, token_usage_from_api
 
 if TYPE_CHECKING:
@@ -46,6 +46,7 @@ if TYPE_CHECKING:
     from plyngent.lmproto.openai_compatible.model import AnyChatMessage, AnyToolItem
 
     from .client import ChatClient
+    from .todo_nag import TodoNagStrategy
     from .todo_stack import TodoStack
     from .tools import ToolRegistry
 
@@ -307,7 +308,7 @@ def _last_assistant(messages: list[AnyChatMessage], pre_len: int) -> AssistantCh
     return last
 
 
-async def run_chat_loop(
+async def run_chat_loop(  # noqa: C901 — multi-phase tool loop
     client: ChatClient,
     messages: list[AnyChatMessage],
     *,
@@ -321,6 +322,7 @@ async def run_chat_loop(
     parallel_tools: bool = True,
     max_context_tokens: int = DEFAULT_CONTEXT_MAX_TOKENS,
     todo_stack: TodoStack | None = None,
+    todo_nag_strategy: TodoNagStrategy = DEFAULT_TODO_NAG_STRATEGY,
 ) -> AsyncIterator[AgentEvent]:
     """Multi-round chat/tool loop; mutates ``messages`` in place and yields events.
 
@@ -329,9 +331,9 @@ async def run_chat_loop(
     Multiple tool calls in one round run in parallel when ``parallel_tools``.
     Request payloads may shrink older tool results when over ``max_context_tokens``.
 
-    When *todo_stack* is set and still has open items after a natural stop with
+    When *todo_stack* is set and still needs review after a natural stop
-    no ``todo_*`` tool use this turn, injects a review user message and continues
+    (open items, or non-empty stack untouched this turn), injects a review nag
-    once (so the model must check or update the stack).
+    (channel from *todo_nag_strategy*) once so the model reconciles unfinished work.
     """
     tool_items: Sequence[AnyToolItem] | None = None
     if tools is not None and len(tools) > 0:
@@ -373,9 +375,16 @@ async def run_chat_loop(
             if tool_calls is UNSET or not tool_calls or tools is None:
                 if todo_stack is not None and todo_stack.needs_review() and not todo_review_injected:
                     todo_review_injected = True
-                    # Non-user control identity so retry/history don't treat this as human input.
+                    injected, nag_events = inject_todo_nag_for_stack_with_events(
-                    messages.append(DeveloperChatMessage(content=todo_stack.review_prompt()))
+                        messages,
-                    continue
+                        todo_stack,
+                        kind="end_of_turn",
+                        strategy=todo_nag_strategy,
+                    )
+                    for nag_event in nag_events:
+                        yield nag_event
+                    if injected:
+                        continue
                 return
             async for event in _execute_tool_calls(
                 tools,

src/plyngent/agent/todo_nag.py

@@ -0,0 +1,183 @@
+"""Inject todo stack nags into the message list (configurable channel)."""
+
+from __future__ import annotations
+
+import uuid
+from typing import TYPE_CHECKING, Literal, cast
+
+from msgspec import UNSET
+
+from plyngent.lmproto.openai_compatible.model import (
+    AnyAssistantToolCall,
+    AssistantChatMessage,
+    AssistantFunctionTool,
+    AssistantFunctionToolCall,
+    DeveloperChatMessage,
+    ToolChatMessage,
+    UserChatMessage,
+)
+
+if TYPE_CHECKING:
+    from plyngent.lmproto.openai_compatible.model import AnyChatMessage
+
+    from .events import AgentEvent
+    from .todo_stack import TodoStack
+
+type TodoNagStrategy = Literal["developer", "user", "synthetic_tool", "none"]
+type TodoNagKind = Literal["turn_start", "end_of_turn"]
+
+TODO_NAG_STRATEGIES: frozenset[str] = frozenset({"developer", "user", "synthetic_tool", "none"})
+DEFAULT_TODO_NAG_STRATEGY: TodoNagStrategy = "developer"
+_SYNTHETIC_TOOL_NAME = "todo_list"
+
+
+def parse_todo_nag_strategy(raw: str | None) -> TodoNagStrategy:
+    """Normalize config/CLI text to a strategy; unknown → developer.
+
+    Legacy ``system`` is accepted as ``developer`` (mid-turn system was folded
+    into Responses ``instructions`` and was not a useful distinct channel).
+    """
+    token = (raw or DEFAULT_TODO_NAG_STRATEGY).strip().lower().replace("-", "_")
+    if token == "system":
+        return "developer"
+    if token not in TODO_NAG_STRATEGIES:
+        return DEFAULT_TODO_NAG_STRATEGY
+    return cast("TodoNagStrategy", token)
+
+
+def nag_body(stack: TodoStack, kind: TodoNagKind) -> str:
+    """Prose OPEN WORK / HYGIENE prompt (developer / user strategies)."""
+    if kind == "turn_start":
+        return stack.turn_reminder_prompt()
+    return stack.review_prompt()
+
+
+def synthetic_todo_list_result(stack: TodoStack) -> str:
+    """Body for synthetic_tool: same text as a real ``todo_list`` tool result.
+
+    No OPEN WORK lecture — just the stack dump the model would get from
+    ``todo_list``. The call remains forged; the payload is authentic render.
+    """
+    return stack.render()
+
+
+def _append_synthetic_todo_list(messages: list[AnyChatMessage], body: str) -> str:
+    """Append forged todo_list call + result. Returns the synthetic tool_call id."""
+    call_id = f"todo-nag-{uuid.uuid4().hex[:12]}"
+    messages.append(
+        AssistantChatMessage(
+            content=UNSET,
+            tool_calls=[
+                AssistantFunctionToolCall(
+                    id=call_id,
+                    function=AssistantFunctionTool(
+                        name=_SYNTHETIC_TOOL_NAME,
+                        arguments="{}",
+                    ),
+                )
+            ],
+        )
+    )
+    messages.append(ToolChatMessage(tool_call_id=call_id, content=body))
+    return call_id
+
+
+def inject_todo_nag(
+    messages: list[AnyChatMessage],
+    body: str,
+    *,
+    strategy: TodoNagStrategy = DEFAULT_TODO_NAG_STRATEGY,
+) -> bool:
+    """Append a todo nag using *strategy*. Returns True if anything was appended.
+
+    Strategies:
+    - ``developer`` (default): control-plane message (safe for retry/history).
+    - ``user``: looks like a human turn (can confuse retry — use with care).
+    - ``synthetic_tool``: forged ``todo_list`` call + result pair (no handler run).
+    - ``none``: no injection.
+
+    Prefer :func:`inject_todo_nag_with_events` when the CLI needs ToolCall/Result
+    events so the display buffer flushes and tool chrome is not glued to text.
+    """
+    return inject_todo_nag_with_events(messages, body, strategy=strategy)[0]
+
+
+def inject_todo_nag_with_events(
+    messages: list[AnyChatMessage],
+    body: str,
+    *,
+    strategy: TodoNagStrategy = DEFAULT_TODO_NAG_STRATEGY,
+) -> tuple[bool, list[AgentEvent]]:
+    """Like :func:`inject_todo_nag`, also return display events for synthetic_tool.
+
+    For ``synthetic_tool``, emits :class:`ToolCallEvent` then
+    :class:`ToolResultEvent` so streaming UIs flush the assistant buffer and
+    show tool chrome (result is real stack text; call was not model-authored).
+    """
+    from .events import ToolCallEvent, ToolResultEvent
+
+    if strategy == "none" or not body.strip():
+        return False, []
+
+    if strategy == "developer":
+        messages.append(DeveloperChatMessage(content=body))
+        return True, []
+    if strategy == "user":
+        messages.append(UserChatMessage(content=body))
+        return True, []
+    # strategy == "synthetic_tool" (only remaining inject path)
+    call_id = _append_synthetic_todo_list(messages, body)
+    assistant = messages[-2]
+    tool_msg = messages[-1]
+    events: list[AgentEvent] = []
+    if isinstance(assistant, AssistantChatMessage):
+        tool_calls = assistant.tool_calls
+        if tool_calls is not UNSET and tool_calls:
+            for call in tool_calls:
+                if isinstance(call, AssistantFunctionToolCall) and call.id == call_id:
+                    events.append(ToolCallEvent(tool_call=cast("AnyAssistantToolCall", call)))
+                    break
+    if isinstance(tool_msg, ToolChatMessage):
+        events.append(ToolResultEvent(message=tool_msg))
+    return True, events
+
+
+def body_for_strategy(
+    stack: TodoStack,
+    kind: TodoNagKind,
+    strategy: TodoNagStrategy,
+) -> str:
+    """Choose inject payload: prose nag vs real todo_list render."""
+    if strategy == "synthetic_tool":
+        return synthetic_todo_list_result(stack)
+    return nag_body(stack, kind)
+
+
+def inject_todo_nag_for_stack(
+    messages: list[AnyChatMessage],
+    stack: TodoStack,
+    *,
+    kind: TodoNagKind,
+    strategy: TodoNagStrategy = DEFAULT_TODO_NAG_STRATEGY,
+) -> bool:
+    """Build body from *stack* and inject with *strategy*."""
+    return inject_todo_nag(
+        messages,
+        body_for_strategy(stack, kind, strategy),
+        strategy=strategy,
+    )
+
+
+def inject_todo_nag_for_stack_with_events(
+    messages: list[AnyChatMessage],
+    stack: TodoStack,
+    *,
+    kind: TodoNagKind,
+    strategy: TodoNagStrategy = DEFAULT_TODO_NAG_STRATEGY,
+) -> tuple[bool, list[AgentEvent]]:
+    """Build body from *stack*; inject and return CLI-facing events."""
+    return inject_todo_nag_with_events(
+        messages,
+        body_for_strategy(stack, kind, strategy),
+        strategy=strategy,
+    )

src/plyngent/agent/todo_stack.py

@@ -8,6 +8,7 @@ from msgspec import Struct, field
 type TodoStatus = Literal["pending", "in_progress", "done", "cancelled"]
 
 _OPEN: frozenset[str] = frozenset({"pending", "in_progress"})
+_REVIEW_OPEN_TITLE_LIMIT = 8
 
 
 class TodoItem(Struct, omit_defaults=True):
@@ -88,7 +89,17 @@ class TodoStack:
         return not self._data.groups
 
     def needs_review(self) -> bool:
-        return bool(self.open_items()) and not self._touched_this_turn
+        """True when the stack still signals unfinished or unreconciled work.
+
+        Open (pending/in_progress) items always need attention. A non-empty stack
+        with only terminal items still needs a pop/clear if the agent ignored
+        todos this turn.
+        """
+        if self.is_empty():
+            return False
+        if self.open_items():
+            return True
+        return not self._touched_this_turn
 
     def to_data(self) -> TodoStackData:
         return self._data
@@ -173,17 +184,80 @@ class TodoStack:
                 lines.append(f"  {mark} {item.id}: {item.title}{note}")
         return "\n".join(lines)
 
+    def turn_reminder_prompt(self) -> str:
+        """Mid-context nudge when a turn starts with a non-empty stack.
+
+        A non-empty stack usually means unfinished work (open items) or unfinished
+        stack hygiene (terminal items still grouped). Prefer finishing real work
+        over pure bookkeeping when both exist.
+        """
+        n_open = len(self.open_items())
+        n_groups = self.depth
+        if n_open:
+            headline = (
+                f"[TODO OPEN WORK] Stack not empty: {n_open} open item(s) across "
+                f"{n_groups} group(s). This is unfinished work from earlier in the "
+                "session — not optional decoration."
+            )
+            action = (
+                "Treat open items as active commitments: continue them, mark "
+                "done/cancelled when finished, or push a child breakdown for the "
+                "current TOP item. Only todo_clear if the user abandoned the plan. "
+                "Do not ignore the stack while answering unrelated chatter."
+            )
+        else:
+            headline = (
+                f"[TODO HYGIENE] Stack not empty: {n_groups} group(s) with no open "
+                "items (all done/cancelled). Work may be finished; the stack is not."
+            )
+            action = (
+                "Pop finished TOP groups (or todo_clear when the whole plan is done). "
+                "Leaving only-terminal groups wastes context; clean up when you can."
+            )
+        lines = [
+            headline,
+            action,
+            "Tools: todo_list | todo_push(titles=[...]) | todo_update | todo_pop | todo_clear",
+            "Rules: TOP = current level; push = one sibling group; pop = whole TOP group.",
+            "Stack:",
+            self.render(),
+        ]
+        return "\n".join(lines)
+
     def review_prompt(self) -> str:
-        """Compact, model-facing nag: open work still exists this turn."""
+        """End-of-turn nag: non-empty stack still signals unfinished work."""
         open_items = self.open_items()
         n_open = len(open_items)
         n_groups = self.depth
+        if n_open:
+            open_titles = "; ".join(f"{item.id}:{item.title}" for item in open_items[:_REVIEW_OPEN_TITLE_LIMIT])
+            if n_open > _REVIEW_OPEN_TITLE_LIMIT:
+                open_titles += f"; …(+{n_open - _REVIEW_OPEN_TITLE_LIMIT} more)"
+            headline = (
+                f"[TODO OPEN WORK] You stopped with {n_open} open item(s) still on "
+                f"the stack ({n_groups} group(s)). That usually means undone work."
+            )
+            action = (
+                "Do not end the turn while open tasks remain unaddressed. Next: "
+                "resume an open item, mark done/cancelled, pop a finished TOP group, "
+                "or push a child breakdown. Clear only if the user no longer wants "
+                f"the plan. Open: {open_titles}"
+            )
+        else:
+            headline = (
+                f"[TODO HYGIENE] Stack still has {n_groups} group(s) but every item "
+                "is done/cancelled. Real work looks finished; stack cleanup does not."
+            )
+            action = (
+                "Pop finished TOP groups (or todo_clear when the whole plan is done). "
+                "A non-empty all-terminal stack is unfinished task hygiene, not new work."
+            )
         lines = [
-            f"[TODO OPEN] {n_open} open item(s) across {n_groups} group(s) "
+            headline,
-            f"(not empty). You did not call todo_* this turn.",
+            action,
             "Tools: todo_list | todo_push(titles=[...]) | todo_update | todo_pop | todo_clear",
             "Rules: TOP group = current level; push=one sibling group; pop=remove whole TOP group.",
-            "Act on open work before ending (finish/pop TOP, or push child tasks). Stack:",
+            "Stack:",
             self.render(),
         ]
         return "\n".join(lines)

src/plyngent/cli/app.py

@@ -82,12 +82,21 @@ def _warn_recoverable_providers(recoverable: Mapping[str, object]) -> None:
 
 def _database_config(store: ConfigStore, *, quiet: bool = False) -> DatabaseConfig:
     raw = dict(store.database)
-    # Prefer a durable file DB so sessions survive CLI restarts.
+    url = raw.get("url")
-    if raw.get("url") in {None, "", ":memory:"} and raw.get("implementation", "sqlite") == "sqlite":
+    impl = raw.get("implementation", "sqlite")
+    # Unset/empty → durable user-data chat.db so interactive sessions persist.
+    # Explicit ``:memory:`` is kept as-is (ephemeral; warn so it is not accidental).
+    if impl == "sqlite" and url in {None, ""}:
         db_path = user_data_path("plyngent", ensure_exists=True) / _DEFAULT_DB_FILENAME
         raw = {**raw, "implementation": "sqlite", "url": str(db_path)}
         if not quiet:
             click.secho(f"using database: {db_path}", fg="bright_black", err=True)
+    elif not quiet and impl == "sqlite" and url == ":memory:":
+        click.secho(
+            "warning: database url is :memory: — sessions are not persisted to disk",
+            fg="yellow",
+            err=True,
+        )
     return msgspec.convert(raw, DatabaseConfig)
 
 
@@ -245,20 +254,26 @@ async def _run_chat(  # noqa: C901, PLR0912, PLR0915 — chat orchestration
                 preferred_model=preferred_model,
                 interactive=interactive,
             )
-            # Build client early so we can always try GET /models for remote-first lists.
+            # Avoid blocking ready on GET /models unless interactive pick needs it.
             from plyngent.cli.models_source import (
                 client_supports_models,
                 fetch_remote_model_ids,
                 model_choices_for_provider,
+                needs_remote_models_for_selection,
             )
 
-            client = create_client(provider)
             remote_ids: list[str] | None = None
-            try:
+            if needs_remote_models_for_selection(
-                if client_supports_models(client):
+                provider,
-                    remote_ids = await fetch_remote_model_ids(client)
+                preferred_model=preferred_model,
-            except RuntimeError, TypeError, OSError, ValueError:
+                interactive=interactive,
-                remote_ids = None
+            ):
+                client = create_client(provider)
+                try:
+                    if client_supports_models(client):
+                        remote_ids = await fetch_remote_model_ids(client)
+                except RuntimeError, TypeError, OSError, ValueError, TimeoutError:
+                    remote_ids = None
             choices = model_choices_for_provider(provider, remote_ids=remote_ids)
             model_id = select_model(
                 provider,
@@ -282,9 +297,11 @@ async def _run_chat(  # noqa: C901, PLR0912, PLR0915 — chat orchestration
             interactive_limits=interactive,
             yolo=yolo,
         )
-        # Seed remote model cache from startup fetch so Tab/complete stays warm.
+        # Seed cache if we already fetched; else warm in background for Tab.
         if remote_ids is not None:
             state.seed_remote_models(remote_ids)
+        elif not oneshot and interactive:
+            state.schedule_remote_models_warm()
         if not quiet and not oneshot:
             click.secho(f"workspace: {state.workspace}", fg="bright_black", err=True)
 
@@ -310,6 +327,10 @@ async def _run_chat(  # noqa: C901, PLR0912, PLR0915 — chat orchestration
         from plyngent.tools.temp_workspace import cleanup_temporary_workspaces
 
         PtyManager.close_all()
+        from plyngent.tools.workspace import clear_policy_allowed_commands, set_policy_confirm_hook
+
+        set_policy_confirm_hook(None)
+        clear_policy_allowed_commands()
         _ = cleanup_temporary_workspaces()
 
 
@@ -508,10 +529,21 @@ def config_path_cmd(config_path: Path | None) -> None:
     help="Path to plyngent.toml (default: platform config dir).",
 )
 def config_edit_cmd(config_path: Path | None) -> None:
-    """Open the config file in $EDITOR (supports e.g. ``codium --wait``)."""
+    """Open the config file in $VISUAL/$EDITOR, or system default if unset.
+
+    Blocking editors (e.g. ``codium --wait``) wait for exit. Without VISUAL/EDITOR,
+    falls back to xdg-open / open / startfile (non-blocking).
+    """
     path = resolve_config_path(config_path)
-    open_in_editor(path)
+    outcome = open_in_editor(path, allow_system_open=True)
-    click.echo(f"edited {path}")
+    if outcome == "system":
+        click.secho(
+            f"opened {path} with system default (not waiting for the app to exit)",
+            fg="yellow",
+            err=True,
+        )
+    else:
+        click.echo(f"edited {path}")
 
 
 if __name__ == "__main__":

src/plyngent/cli/editor.py

@@ -4,8 +4,9 @@ import contextlib
 import os
 import shlex
 import subprocess
+import sys
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Literal
 
 import click
 
@@ -14,14 +15,18 @@ from plyngent import config as config_mod
 if TYPE_CHECKING:
     from plyngent.config.store import ConfigStore
 
+type OpenOutcome = Literal["waited", "system"]
+
 _MINIMAL_CONFIG = """\
 # plyngent configuration
 # edit providers below
 
-# Optional: omit [database] to use ~/.local/share/plyngent/chat.db (Linux).
+# Optional: omit [database] (or leave url unset) → ~/.local/share/plyngent/chat.db.
+# url = ":memory:" keeps a true in-memory SQLite (CLI warns; nothing on disk).
 # [database]
 # implementation = "sqlite"
 # url = "/path/to/chat.db"
+# # url = ":memory:"
 
 # [agent]
 # system_prompt = "You are a careful coding assistant."
@@ -52,9 +57,16 @@ _MINIMAL_CONFIG = """\
 
 
 def get_editor() -> str | None:
-    """Return the ``EDITOR`` environment value, or ``None`` if unset/empty."""
+    """Return ``$VISUAL`` or ``$EDITOR``, or ``None`` if both unset/empty.
-    value = os.environ.get("EDITOR", "").strip()
+
-    return value or None
+    ``VISUAL`` is preferred when set (common Unix convention for full-screen
+    editors); otherwise ``EDITOR``.
+    """
+    for key in ("VISUAL", "EDITOR"):
+        value = os.environ.get(key, "").strip()
+        if value:
+            return value
+    return None
 
 
 def resolve_config_path(config_path: Path | None) -> Path:
@@ -71,31 +83,14 @@ def ensure_config_file(path: Path) -> None:
         _ = path.write_text(_MINIMAL_CONFIG, encoding="utf-8")
 
 
-def open_in_editor(
+def _run_blocking_editor(editor_cmd: str, path: Path) -> None:
-    path: Path,
-    *,
-    editor: str | None = None,
-    ensure_exists: bool = True,
-) -> None:
-    """Open ``path`` with ``EDITOR`` (supports values like ``codium --wait``).
-
-    When ``ensure_exists`` is true (default), create a minimal config template
-    if the file is missing (used for ``plyngent config edit``).
-    """
-    editor_cmd = editor if editor is not None else get_editor()
-    if editor_cmd is None:
-        msg = "EDITOR is not set"
-        raise click.ClickException(msg)
-
-    if ensure_exists:
-        ensure_config_file(path)
     try:
         argv = [*shlex.split(editor_cmd, posix=os.name != "nt"), str(path)]
     except ValueError as exc:
-        msg = f"invalid EDITOR value {editor_cmd!r}: {exc}"
+        msg = f"invalid editor value {editor_cmd!r}: {exc}"
         raise click.ClickException(msg) from exc
     if not argv:
-        msg = "EDITOR is empty after parsing"
+        msg = "editor command is empty after parsing"
         raise click.ClickException(msg)
 
     try:
@@ -112,15 +107,97 @@ def open_in_editor(
         raise click.ClickException(msg)
 
 
-def edit_text_in_editor(initial: str = "", *, suffix: str = ".md") -> str | None:
+def _open_with_system_default(path: Path) -> None:
-    """Edit ``initial`` in ``$EDITOR``; return text or ``None`` if empty/cancelled.
+    """Open *path* with the OS file association (non-blocking).
 
-    Uses a temporary file. Does not create a config template.
+    Linux: ``xdg-open``; macOS: ``open``; Windows: ``os.startfile``.
+    Does not wait for the application to exit.
+    """
+    resolved = str(path.resolve())
+    if sys.platform == "win32":
+        try:
+            os.startfile(resolved)  # type: ignore[attr-defined]
+        except OSError as exc:
+            msg = f"failed to open with system default: {exc}"
+            raise click.ClickException(msg) from exc
+        return
+
+    if sys.platform == "darwin":
+        argv = ["open", resolved]
+    else:
+        # Linux and other POSIX: Free Desktop opener
+        argv = ["xdg-open", resolved]
+
+    try:
+        completed = subprocess.run(
+            argv,
+            check=False,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+    except FileNotFoundError as exc:
+        cmd = argv[0]
+        msg = (
+            f"{cmd} not found and neither VISUAL nor EDITOR is set; "
+            "set VISUAL or EDITOR to a blocking editor (e.g. nano, vim, codium --wait)"
+        )
+        raise click.ClickException(msg) from exc
+    except OSError as exc:
+        msg = f"failed to open with system default: {exc}"
+        raise click.ClickException(msg) from exc
+
+    if completed.returncode != 0:
+        msg = f"system open failed (exit {completed.returncode}); set VISUAL or EDITOR to a blocking editor"
+        raise click.ClickException(msg)
+
+
+def open_in_editor(
+    path: Path,
+    *,
+    editor: str | None = None,
+    ensure_exists: bool = True,
+    allow_system_open: bool = True,
+) -> OpenOutcome:
+    """Open ``path`` for editing.
+
+    Prefer a blocking editor (``editor`` arg, else ``$VISUAL`` / ``$EDITOR``).
+    When none is set and ``allow_system_open`` is true, fall back to the OS
+    default association (``xdg-open`` / ``open`` / ``os.startfile``) — this
+    does **not** wait for the app to exit.
+
+    Returns:
+        ``"waited"`` if a blocking editor ran to completion;
+        ``"system"`` if the OS opener was used (non-blocking).
+
+    When ``ensure_exists`` is true (default), create a minimal config template
+    if the file is missing (used for ``plyngent config edit``).
+    """
+    if ensure_exists:
+        ensure_config_file(path)
+
+    editor_cmd = editor if editor is not None else get_editor()
+    if editor_cmd is not None:
+        _run_blocking_editor(editor_cmd, path)
+        return "waited"
+
+    if not allow_system_open:
+        msg = "neither VISUAL nor EDITOR is set"
+        raise click.ClickException(msg)
+
+    _open_with_system_default(path)
+    return "system"
+
+
+def edit_text_in_editor(initial: str = "", *, suffix: str = ".md") -> str | None:
+    """Edit ``initial`` in a blocking editor; return text or ``None`` if empty.
+
+    Uses a temporary file. Requires ``$VISUAL`` or ``$EDITOR`` (no system-open
+    fallback — we must wait for the process and re-read the buffer).
     """
     import tempfile
 
     if get_editor() is None:
-        msg = "EDITOR is not set; cannot /edit"
+        msg = "neither VISUAL nor EDITOR is set; cannot /edit"
         raise click.ClickException(msg)
 
     with tempfile.NamedTemporaryFile(
@@ -134,7 +211,7 @@ def edit_text_in_editor(initial: str = "", *, suffix: str = ".md") -> str | None
         _ = handle.write(initial)
 
     try:
-        open_in_editor(path, ensure_exists=False)
+        _ = open_in_editor(path, ensure_exists=False, allow_system_open=False)
         text = path.read_text(encoding="utf-8")
     finally:
         with contextlib.suppress(OSError):
@@ -146,19 +223,24 @@ def edit_text_in_editor(initial: str = "", *, suffix: str = ".md") -> str | None
     return cleaned
 
 
-def prompt_edit_config(path: Path, *, reason: str | None = None) -> bool:
+def prompt_edit_config(path: Path, *, reason: str | None = None) -> OpenOutcome | None:
-    """If ``EDITOR`` is set, ask whether to edit ``path``. Returns True if opened."""
+    """Ask whether to edit ``path``. Returns open outcome, or ``None`` if skipped.
-    if get_editor() is None:
+
-        return False
+    Offers when a blocking editor is set **or** system open can be attempted.
+    """
+    has_editor = get_editor() is not None
+    # System open is always attempted as fallback when no editor; we still
+    # prompt so the user can decline on headless hosts.
     message = f"{reason} Edit config file {path}?" if reason else f"Edit config file {path}?"
+    if not has_editor:
+        message = f"{message} (no VISUAL/EDITOR; will try system default open — non-blocking)"
     if not click.confirm(message, default=False):
-        return False
+        return None
-    open_in_editor(path)
+    return open_in_editor(path, allow_system_open=True)
-    return True
 
 
 def load_config_with_optional_edit(config_path: Path | None) -> ConfigStore:
-    """Load config; if there are no providers and EDITOR is set, offer to edit and reload.
+    """Load config; if there are no providers, offer to edit and reload when waited.
 
     Raises:
         config_mod.ConfigFormatError: Invalid TOML (caller should surface path).
@@ -170,6 +252,13 @@ def load_config_with_optional_edit(config_path: Path | None) -> ConfigStore:
     reason = "No providers configured."
     if not path.exists():
         reason = f"Config file not found ({path})."
-    if prompt_edit_config(path, reason=reason):
+    outcome = prompt_edit_config(path, reason=reason)
+    if outcome == "waited":
         store = config_mod.load(path)
+    elif outcome == "system":
+        click.secho(
+            f"opened {path} with system default (not waiting). Save the file, then re-run plyngent to load providers.",
+            fg="yellow",
+            err=True,
+        )
     return store

src/plyngent/cli/limits.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+import contextlib
+import selectors
 import shutil
+import sys
 from typing import TYPE_CHECKING, Literal
 
 from plyngent.cli.interrupt import pause_task_cancel_for_prompt
@@ -19,8 +22,7 @@ from plyngent.prompting import (
 from plyngent.tools.process.pty_session import PtyManager
 
 if TYPE_CHECKING:
-    from collections.abc import Mapping
+    from collections.abc import Mapping, Sequence
-
 type WorkspaceMismatchChoice = Literal["keep", "rebind", "abort"]
 
 _BOX_MIN_WIDTH = 40
@@ -150,6 +152,86 @@ def prompt_confirm_tool(name: str, args: Mapping[str, object], reason: str) -> b
         return _prompt_confirm_tool_sync(name, args, reason)
 
 
+def _try_readline() -> str | None:
+    try:
+        return sys.stdin.readline()
+    except OSError, EOFError:
+        return None
+
+
+def _read_yes_no_line_with_timeout(timeout_seconds: float) -> str | None:
+    """Read one line from stdin with *timeout_seconds*; ``None`` on timeout/EOF.
+
+    Uses :mod:`selectors` on the process stdin FD. On Windows without selector
+    support, falls back to blocking readline (no true timeout).
+    """
+    try:
+        if not sys.stdin.isatty():
+            return None
+        fd = sys.stdin.fileno()
+    except OSError, ValueError, AttributeError:
+        return _try_readline()
+
+    try:
+        selector = selectors.DefaultSelector()
+        _ = selector.register(fd, selectors.EVENT_READ)
+    except OSError, ValueError, AttributeError:
+        return _try_readline()
+
+    try:
+        events = selector.select(timeout_seconds)
+        if not events:
+            return None
+        return sys.stdin.readline()
+    except OSError, ValueError, EOFError:
+        return None
+    finally:
+        selector.close()
+
+
+def prompt_policy_command_confirm(
+    basename: str,
+    argv: Sequence[str],
+    timeout_seconds: float,
+) -> bool:
+    """Ask whether to allow a denylisted command for this session (timed).
+
+    Independent of YOLO: always prompts when interactive. Default is **deny**.
+    Timeout, cancel, or non-interactive → False.
+    """
+    argv_preview = " ".join(str(part) for part in argv)
+    reason = (
+        f"policy denylist blocks basename {basename!r}\n"
+        f"argv: {argv_preview}\n"
+        f"Allow this basename for the rest of this process?\n"
+        f"(timeout {timeout_seconds:g}s defaults to DENY; not skipped by YOLO)"
+    )
+    try:
+        with pause_task_cancel_for_prompt():
+            backend = get_prompt_backend()
+            if not backend.is_interactive():
+                return False
+            backend.echo()
+            backend.secho(format_tool_confirm_box("policy", reason), fg="yellow")
+            backend.echo()
+            prompt = f"[policy] allow {basename!r}? [y/N] (timeout {timeout_seconds:g}s): "
+            with contextlib.suppress(OSError):
+                _ = sys.stderr.write(prompt)
+                _ = sys.stderr.flush()
+            raw = _read_yes_no_line_with_timeout(timeout_seconds)
+            if raw is None:
+                with contextlib.suppress(OSError, NonInteractiveError):
+                    backend.secho(
+                        f"[policy] timed out after {timeout_seconds:g}s — denied",
+                        fg="red",
+                        err=True,
+                    )
+                return False
+            return raw.strip().lower() in {"y", "yes"}
+    except NonInteractiveError, KeyboardInterrupt, EOFError:
+        return False
+
+
 async def prompt_confirm_tool_async(name: str, args: Mapping[str, object], reason: str) -> bool | str:
     """Async confirm: True allow, False deny, str = deny with user comment."""
     del args
@@ -268,6 +350,9 @@ async def prompt_workspace_mismatch_async(
 
 
 def install_cli_limit_hooks() -> None:
-    """Register interactive continue hooks and prompt cancel-pause for the CLI."""
+    """Register interactive continue hooks, policy confirm, and prompt cancel-pause."""
     configure_prompting(pause_factory=pause_task_cancel_for_prompt)
     PtyManager.set_limit_continue_hook(prompt_continue_limit)
+    from plyngent.tools.workspace import set_policy_confirm_hook
+
+    set_policy_confirm_hook(prompt_policy_command_confirm)

src/plyngent/cli/models_source.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import inspect
 from typing import TYPE_CHECKING, Literal, Protocol, cast, runtime_checkable
 
@@ -10,6 +11,8 @@ if TYPE_CHECKING:
 
 # Cache remote catalog this long (seconds) unless /models --refresh.
 DEFAULT_MODELS_CACHE_TTL = 300.0
+# Bound startup/interactive GET /models so a dead API cannot hang the CLI.
+DEFAULT_MODELS_FETCH_TIMEOUT = 5.0
 
 type ModelListPrefer = Literal["remote", "union", "config"]
 
@@ -58,21 +61,58 @@ def client_supports_models(client: object) -> bool:
     return isinstance(client, SupportsModels) or callable(getattr(client, "models", None))
 
 
-async def fetch_remote_model_ids(client: object) -> list[str]:
+async def fetch_remote_model_ids(
-    """Call ``client.models()``; raise if missing or the call fails."""
+    client: object,
+    *,
+    timeout_seconds: float = DEFAULT_MODELS_FETCH_TIMEOUT,
+) -> list[str]:
+    """Call ``client.models()``; raise if missing or the call fails.
+
+    *timeout_seconds* bounds the await; use ``0`` or less to wait indefinitely.
+    """
     method = getattr(client, "models", None)
     if not callable(method):
         msg = "client does not support listing models"
         raise TypeError(msg)
     result = method()
     if inspect.isawaitable(result):
-        result = await result
+        try:
+            if timeout_seconds > 0:
+                async with asyncio.timeout(timeout_seconds):
+                    result = await result
+            else:
+                result = await result
+        except TimeoutError as exc:
+            msg = f"models() timed out after {timeout_seconds}s"
+            raise RuntimeError(msg) from exc
+        except asyncio.CancelledError as exc:
+            # SSL/HTTP stacks may surface cancel instead of TimeoutError; treat as soft fail.
+            msg = "models() was cancelled or timed out"
+            raise RuntimeError(msg) from exc
     if not isinstance(result, list):
         msg = f"models() returned unexpected type {type(result)!r}"
         raise TypeError(msg)
     return [str(item) for item in cast("list[object]", result) if item]
 
 
+def needs_remote_models_for_selection(
+    provider: Provider,
+    *,
+    preferred_model: str | None,
+    interactive: bool,
+) -> bool:
+    """True when interactive model pick needs a remote catalog for a better list.
+
+    Skip network when the model is already known (``--model`` / session) or when
+    config has exactly one model (auto-selected).
+    """
+    if preferred_model is not None and preferred_model.strip():
+        return False
+    if not interactive:
+        return False
+    return len(config_model_ids(provider)) != 1
+
+
 def model_choices_for_provider(
     provider: Provider,
     *,

src/plyngent/cli/provider_recovery.py

@@ -22,7 +22,7 @@ async def discover_model_ids(provider: Provider) -> list[str]:
         return []
     try:
         return await fetch_remote_model_ids(client)
-    except RuntimeError, TypeError, OSError, ValueError:
+    except RuntimeError, TypeError, OSError, ValueError, TimeoutError:
         return []
 
 

src/plyngent/cli/slash.py

@@ -87,7 +87,7 @@ HELP_FOOTER = (
     "continue a prior chat after restart.\n"
     "\n"
     'Multiline: start a message with """ then end a later line with """.\n'
-    "Long prompts: /edit opens $EDITOR.\n"
+    "Long prompts: /edit opens $VISUAL/$EDITOR (blocking).\n"
 )
 
 
@@ -453,10 +453,10 @@ def clear_cmd(state: ReplState) -> None:
 @slash.command("edit")
 @click.pass_obj
 def edit_cmd(state: ReplState) -> None:
-    """Compose a user message in ``$EDITOR``, then send it.
+    """Compose a user message in ``$VISUAL``/``$EDITOR``, then send it.
 
     Opens a temporary buffer; save and quit the editor to submit.
-    Empty buffer cancels. Requires ``EDITOR`` (e.g. ``codium --wait``).
+    Empty buffer cancels. Requires a blocking editor (no system-open fallback).
     """
     from plyngent.cli.editor import edit_text_in_editor
 
@@ -475,20 +475,27 @@ def edit_cmd(state: ReplState) -> None:
 @slash.command("config")
 @click.pass_obj
 def config_cmd(state: ReplState) -> None:
-    """Open plyngent.toml in ``$EDITOR``, then reload providers/agent settings.
+    """Open plyngent.toml in ``$VISUAL``/``$EDITOR`` (or system default), then reload.
 
-    Same file as ``plyngent config edit``. After the editor exits, config is
+    Same file as ``plyngent config edit``. After a **blocking** editor exits,
-    re-read; current provider/model are kept when still valid.
+    config is re-read. System-open fallback does not wait — reload is skipped
+    with a hint to re-run ``/config`` after saving.
     """
     from plyngent import config as config_mod
     from plyngent.cli.editor import open_in_editor
 
     path = state.config.path
     try:
-        open_in_editor(path)
+        outcome = open_in_editor(path, allow_system_open=True)
     except click.ClickException as exc:
         click.echo(f"error: {exc}")
         return
+    if outcome == "system":
+        click.secho(
+            f"opened {path} with system default (not waiting). Save the file, then run /config again to reload.",
+            fg="yellow",
+        )
+        return
     try:
         state.reload_config_from_disk()
     except (config_mod.ConfigFormatError, ValueError, OSError) as exc:

src/plyngent/cli/state.py

@@ -187,6 +187,7 @@ class ReplState:
             parallel_tools=agent_cfg.parallel_tools,
             max_context_tokens=agent_cfg.max_context_tokens,
             todo_stack=self.todo_stack,
+            todo_nag_strategy=agent_cfg.todo_nag_strategy,
         )
 
     def rebuild_client(self) -> None:
@@ -224,6 +225,29 @@ class ReplState:
         self._remote_models_key = self._models_cache_key()
         self._remote_models_error = None
 
+    def schedule_remote_models_warm(self) -> None:
+        """Fire-and-forget ``GET /models`` so Tab complete warms without blocking ready."""
+        import asyncio
+
+        async def _warm() -> None:
+            try:
+                _ = await self.ensure_remote_models(refresh=False)
+            except RuntimeError, TypeError, OSError, ValueError, TimeoutError:
+                return
+
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            return
+        task = loop.create_task(_warm(), name="plyngent-warm-remote-models")
+        # Keep a strong ref until done (same pattern as todo persist tasks).
+        self._todo_persist_tasks.add(task)
+
+        def _done(t: object) -> None:
+            _ = self._todo_persist_tasks.discard(t)
+
+        task.add_done_callback(_done)
+
     def remember_session_ids(self, ids: Sequence[int]) -> None:
         """Cache session ids for Tab completion (``/resume`` / ``/delete``)."""
         self._session_id_cache = [int(i) for i in ids]
@@ -272,7 +296,7 @@ class ReplState:
             raise TypeError(msg)
         try:
             ids = await fetch_remote_model_ids(self.client)
-        except (RuntimeError, TypeError, OSError, ValueError) as exc:
+        except (RuntimeError, TypeError, OSError, ValueError, TimeoutError) as exc:
             self._remote_models_error = str(exc)
             raise
         self._remote_models = list(ids)
@@ -282,11 +306,15 @@ class ReplState:
         return list(ids)
 
     async def merged_model_choices(self, *, refresh: bool = False) -> list[str]:
-        """Config plus remote catalog; remote fetch best-effort when refresh/missing."""
+        """Config plus remote catalog; remote fetch best-effort when refresh/missing.
+
+        Network / timeout / cancel failures fall back to cache or config-only ids
+        so ``/provider`` switches never hang or fail solely on GET /models.
+        """
         remote: list[str] | None
         try:
             remote = await self.ensure_remote_models(refresh=refresh)
-        except RuntimeError, TypeError, OSError, ValueError:
+        except RuntimeError, TypeError, OSError, ValueError, TimeoutError:
             remote = self.cached_remote_models()
         return model_choices_for_provider(self.provider, remote_ids=remote)
 

src/plyngent/config/models.py

@@ -4,10 +4,15 @@ from msgspec import Struct, field
 
 
 class DatabaseConfig(Struct, omit_defaults=True):
-    """Database connection configuration."""
+    """Database connection configuration.
+
+    ``url`` is ``None`` when unset. The CLI fills a durable user-data
+    ``chat.db`` only for unset/empty url. Explicit ``":memory:"`` is a true
+    in-memory SQLite (no rewrite).
+    """
 
     implementation: str = "sqlite"
-    url: str = ":memory:"
+    url: str | None = None
     username: str | None = None
     password: str | None = None
 
@@ -22,6 +27,10 @@ class AgentConfig(Struct, omit_defaults=True):
     path_denylist: list[str] = field(default_factory=list)
     max_context_tokens: int = 200_000
 
+    # How to inject todo stack nags into model context (see agent/todo_nag.py).
+    # developer | user | synthetic_tool | none  (legacy "system" → developer)
+    todo_nag_strategy: str = "developer"
+
     # Compact / summarisation prompts (empty = use built-in defaults).
     compact_system_prompt: str = ""
     compact_user_prefix: str = ""

src/plyngent/memory/database/engine.py

@@ -20,7 +20,7 @@ def build_async_url(config: DatabaseConfig) -> str:
         raise UnsupportedDatabaseError(msg)
 
     url = config.url
-    if url in {":memory:", ""}:
+    if url is None or url in {":memory:", ""}:
         return "sqlite+aiosqlite:///:memory:"
     if url.startswith("sqlite+aiosqlite://"):
         return url

src/plyngent/prompting.py

@@ -55,6 +55,10 @@ class PromptBackend(Protocol):
         completions: Sequence[str] | None = None,
     ) -> str: ...
 
+    def read_secret_line(self, prompt: str) -> str:
+        """Read a line without echo (passwords). Cancel raises NonInteractiveError."""
+        ...
+
     def confirm(self, prompt: str, *, default: bool = False) -> bool: ...
 
     def echo(self, message: str = "", *, err: bool = False) -> None: ...
@@ -117,6 +121,16 @@ class ClickPromptBackend:
             return default
         return raw
 
+    def read_secret_line(self, prompt: str) -> str:
+        import getpass
+
+        try:
+            display = f"{prompt}: " if not prompt.endswith(": ") else prompt
+            return getpass.getpass(display)
+        except (KeyboardInterrupt, EOFError) as exc:
+            msg = "prompt cancelled"
+            raise NonInteractiveError(msg) from exc
+
     def confirm(self, prompt: str, *, default: bool = False) -> bool:
         try:
             return bool(click.confirm(prompt, default=default))
@@ -150,6 +164,10 @@ class NonInteractiveBackend:
         msg = f"non-interactive: cannot prompt for {prompt!r}"
         raise NonInteractiveError(msg)
 
+    def read_secret_line(self, prompt: str) -> str:
+        msg = f"non-interactive: cannot prompt for secret {prompt!r}"
+        raise NonInteractiveError(msg)
+
     def confirm(self, prompt: str, *, default: bool = False) -> bool:
         del prompt
         return default
@@ -255,6 +273,19 @@ def ask(
     return backend.read_line("Answer", default=default, completions=completions).strip()
 
 
+def ask_secret(prompt: str) -> str:
+    """Prompt for a secret line (no echo). Never use for values that return to the model.
+
+    Cancel (Ctrl+C / EOF) raises :class:`NonInteractiveError`.
+    """
+    backend = get_prompt_backend()
+    if not backend.is_interactive():
+        msg = f"non-interactive: cannot prompt for secret {prompt!r}"
+        raise NonInteractiveError(msg)
+    backend.secho(prompt, fg="yellow")
+    return backend.read_secret_line("Secret")
+
+
 def choose(
     prompt: str,
     options: Sequence[str] | Sequence[ChoiceOption],
@@ -363,6 +394,10 @@ async def ask_async(prompt: str, *, default: str | None = None) -> str:
     return await run_prompt_async(ask, prompt, default=default)
 
 
+async def ask_secret_async(prompt: str) -> str:
+    return await run_prompt_async(ask_secret, prompt)
+
+
 async def choose_async(
     prompt: str,
     options: Sequence[str] | Sequence[ChoiceOption],

src/plyngent/tools/__init__.py

@@ -16,10 +16,12 @@ from .file import read_file as read_file
 from .file import tree as tree
 from .file import write_file as write_file
 from .process import PROCESS_TOOLS as PROCESS_TOOLS
+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_command_batch as run_command_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
@@ -41,20 +43,26 @@ from .vcs import vcs_status as vcs_status
 from .workspace import (
     DEFAULT_COMMAND_DENYLIST as DEFAULT_COMMAND_DENYLIST,
 )
+from .workspace import DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS as DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS
 from .workspace import WorkspaceError as WorkspaceError
 from .workspace import add_workspace_allowlist as add_workspace_allowlist
 from .workspace import check_command_allowed as check_command_allowed
+from .workspace import clear_policy_allowed_commands as clear_policy_allowed_commands
 from .workspace import clear_workspace_allowlist as clear_workspace_allowlist
 from .workspace import clear_workspace_root as clear_workspace_root
 from .workspace import get_command_denylist as get_command_denylist
 from .workspace import get_path_denylist as get_path_denylist
+from .workspace import get_policy_confirm_timeout as get_policy_confirm_timeout
 from .workspace import get_workspace_root as get_workspace_root
+from .workspace import grant_policy_command as grant_policy_command
 from .workspace import list_workspace_allowlist as list_workspace_allowlist
 from .workspace import pop_owned_temporary_workspaces as pop_owned_temporary_workspaces
 from .workspace import remove_workspace_allowlist as remove_workspace_allowlist
 from .workspace import resolve_path as resolve_path
 from .workspace import set_command_denylist as set_command_denylist
 from .workspace import set_path_denylist as set_path_denylist
+from .workspace import set_policy_confirm_hook as set_policy_confirm_hook
+from .workspace import set_policy_confirm_timeout as set_policy_confirm_timeout
 from .workspace import set_workspace_root as set_workspace_root
 
 DEFAULT_TOOLS = [*FILE_TOOLS, *PROCESS_TOOLS, *VCS_TOOLS, *CHAT_TOOLS, *TODO_TOOLS]

src/plyngent/tools/danger.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import json
+import shlex
 from typing import TYPE_CHECKING, cast
 
 from plyngent.tools.workspace import WorkspaceError, resolve_path
@@ -7,8 +9,6 @@ from plyngent.tools.workspace import WorkspaceError, resolve_path
 if TYPE_CHECKING:
     from collections.abc import Mapping, Sequence
 
-_DISPLAY_ARGV_MAX = 400
-_CODE_PREVIEW_MAX = 240
 
 _SHELL_BASENAMES: frozenset[str] = frozenset(
     {
@@ -80,65 +80,92 @@ def _find_dash_c_code(argv: Sequence[str]) -> str | None:
     return None
 
 
+def _indent_block(text: str, *, prefix: str = "  ") -> str:
+    """Indent every line of *text* (for multi-line -c bodies in confirm prompts)."""
+    if not text:
+        return prefix
+    return "\n".join(prefix + line if line else prefix.rstrip() for line in text.splitlines())
+
+
+def _format_argv_for_confirm(argv: Sequence[str], *, code: str | None) -> str:
+    """One-line argv summary; replace -c payload with ``$(command)`` when present."""
+    if code is None:
+        return shlex.join(list(argv))
+    parts: list[str] = []
+    skip_next = False
+    for part in argv:
+        if skip_next:
+            skip_next = False
+            continue
+        if part == "-c":
+            parts.append("-c")
+            parts.append("$(command)")
+            skip_next = True
+            continue
+        parts.append(part)
+    return shlex.join(parts)
+
+
 def _shell_or_dash_c_reason(argv: Sequence[str], *, via: str) -> str | None:
-    """Confirm interactive shells/REPLs and ``-c`` one-liners so the user can inspect argv.
+    """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 (tool name), not repeated on every line.
+    label such as ``run_command`` 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.
     """
     if not argv:
         return None
     base = _basename(argv[0])
-    display = " ".join(argv)
-    if len(display) > _DISPLAY_ARGV_MAX:
-        display = display[:_DISPLAY_ARGV_MAX] + "…"
-
     code = _find_dash_c_code(argv)
+    display = _format_argv_for_confirm(argv, code=code)
+
     if code is not None:
-        preview = code if len(code) <= _CODE_PREVIEW_MAX else code[:_CODE_PREVIEW_MAX] + "…"
+        body = _indent_block(code, prefix="  ")
-        return f"{via}: {base} -c (review code before allow)\nargv:\n  {display}\n-c code:\n  {preview}"
+        return f"{via}: {base} -c (review code before allow)\n  argv: {display}\n  command:\n{body}"
 
     if base in _SHELL_BASENAMES and len(argv) == 1:
-        return f"{via}: interactive {base!r} (review before allow)\nargv:\n  {display}"
+        return f"{via}: interactive {base!r} (review before allow)\n  argv: {display}"
 
-    # e.g. python -i, bash --login without -c still needs a glance.
+    if base in _SHELL_BASENAMES and "-c" not in argv[1:]:
-    if base in _SHELL_BASENAMES:
+        return f"{via}: shell/runtime {base!r} without -c (review before allow)\n  argv: {display}"
-        return f"{via}: shell/runtime {base!r} (review before allow)\nargv:\n  {display}"
 
     return None
 
 
 def _write_file_reason(args: Mapping[str, object]) -> str | None:
+    """Confirm only when write_file would replace an existing file."""
     path = args.get("path")
     if not isinstance(path, str) or not path:
         return None
     try:
         target = resolve_path(path)
     except WorkspaceError:
-        return f"write file {path!r}"
+        # Path policy will fail later; no soft-confirm without a resolved target.
-    return f"write file {path!r} ({target})"
-
-
-def _edit_replace_reason(args: Mapping[str, object]) -> str | None:
-    path = args.get("path")
-    if not isinstance(path, str) or not path:
         return None
-    return f"edit (replace) in {path!r}"
+    if target.is_file():
-
+        return f"overwrite existing file {path!r} ({target})"
-
+    # New file or directory path: no total-overwrite risk for soft-confirm.
-def _edit_lineno_reason(args: Mapping[str, object]) -> str | None:
+    return None
-    path = args.get("path")
-    if not isinstance(path, str) or not path:
-        return None
-    start = args.get("start_line")
-    end = args.get("end_line")
-    return f"edit lines {start}-{end} in {path!r}"
 
 
 def _copy_path_reason(args: Mapping[str, object]) -> str | None:
+    """Confirm copy only when it would replace an existing destination path."""
     src = args.get("src")
     dst = args.get("dst")
-    return f"copy {src!r} → {dst!r}"
+    overwrite = bool(args.get("overwrite", False))
+    if not overwrite:
+        return None
+    if not isinstance(dst, str) or not dst:
+        return f"copy {src!r} → {dst!r} (overwrite)"
+    try:
+        target = resolve_path(dst)
+    except WorkspaceError:
+        return None
+    if target.exists():
+        return f"copy {src!r} → overwrite existing {dst!r} ({target})"
+    return None
 
 
 def _move_path_reason(args: Mapping[str, object]) -> str | None:
@@ -161,6 +188,44 @@ def _run_command_reason(args: Mapping[str, object]) -> str | None:
     return _shell_or_dash_c_reason(argv, via="run_command")
 
 
+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")
+    if not isinstance(command, list) or not command:
+        return None
+    argv: list[str] = []
+    for part in cast("list[object]", command):
+        if not isinstance(part, str):
+            return None
+        argv.append(part)
+    return argv or None
+
+
+def _run_command_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")
+    if isinstance(raw, str):
+        try:
+            loaded: object = json.loads(raw)
+        except json.JSONDecodeError:
+            return None
+        raw = loaded
+    if not isinstance(raw, list):
+        return None
+    risky = [
+        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
+    ]
+    if not risky:
+        return None
+    header = f"run_command_batch: {len(risky)} risky step(s) (review before allow)"
+    return header + "\n" + "\n".join(risky)
+
+
 def _open_pty_reason(args: Mapping[str, object]) -> str | None:
     argv = _as_argv(args)
     if argv is None:
@@ -183,12 +248,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 == "edit_replace":
-        return _edit_replace_reason(args)
-    if name == "edit_lineno":
-        return _edit_lineno_reason(args)
     if name == "run_command":
         return _run_command_reason(args)
+    if name == "run_command_batch":
+        return _run_command_batch_reason(args)
     if name == "open_pty":
         return _open_pty_reason(args)
     return None

src/plyngent/tools/file/edit_replace.py

@@ -4,17 +4,54 @@ from plyngent.agent import tool
 from plyngent.tools.workspace import resolve_path
 
 
+def _count_non_overlapping(text: str, needle: str) -> int:
+    """Count left-to-right non-overlapping matches (same as ``str.replace``)."""
+    if not needle:
+        return 0
+    n = 0
+    start = 0
+    while True:
+        index = text.find(needle, start)
+        if index < 0:
+            return n
+        n += 1
+        start = index + len(needle)
+
+
+def _success_message(path: str, *, replaced: int, found: int) -> str:
+    remaining = found - replaced
+    unit = "occurrence" if replaced == 1 else "occurrences"
+    if remaining == 0:
+        if found == 1:
+            return f"replaced 1 occurrence in {path}"
+        return f"replaced {replaced} {unit} in {path} (all {found} matches)"
+    return (
+        f"replaced {replaced} {unit} in {path} "
+        f"({replaced} of {found} matches; {remaining} remain — "
+        f"raise max_replaces or use a more specific old_string)"
+    )
+
+
 @tool
-def edit_replace(path: str, old_string: str, new_string: str) -> str:
+def edit_replace(path: str, old_string: str, new_string: str, max_replaces: int = 1) -> str:
-    """Replace the first occurrence of ``old_string`` with ``new_string`` in a file."""
+    """Replace occurrences of ``old_string`` with ``new_string`` in a file.
+
+    Replaces left-to-right, non-overlapping. Default ``max_replaces=1`` (first match
+    only). Raise ``max_replaces`` to change multiple identical hits; use a more
+    specific ``old_string`` when you need a particular occurrence.
+    """
     if not old_string:
         return "error: old_string must not be empty"
+    if max_replaces < 1:
+        return "error: max_replaces must be >= 1"
     target = resolve_path(path)
     if not target.is_file():
         return f"error: not a file: {path}"
     text = target.read_text(encoding="utf-8", errors="replace")
-    if old_string not in text:
+    found = _count_non_overlapping(text, old_string)
+    if found == 0:
         return "error: old_string not found in file"
-    updated = text.replace(old_string, new_string, 1)
+    n = min(max_replaces, found)
+    updated = text.replace(old_string, new_string, n)
     _ = target.write_text(updated, encoding="utf-8")
-    return f"replaced first occurrence in {path}"
+    return _success_message(path, replaced=n, found=found)

src/plyngent/tools/process/__init__.py

@@ -1,10 +1,21 @@
+from .ask_into_pty import ask_into_pty as ask_into_pty
 from .close_pty import close_pty as close_pty
 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_command_batch import run_command_batch as run_command_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, open_pty, read_pty, write_pty, write_pty_keys, close_pty]
+PROCESS_TOOLS = [
+    run_command,
+    run_command_batch,
+    open_pty,
+    read_pty,
+    write_pty,
+    write_pty_keys,
+    ask_into_pty,
+    close_pty,
+]

src/plyngent/tools/process/ask_into_pty.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+from typing import Literal
+
+from plyngent.agent import tool
+from plyngent.prompting import NonInteractiveError, ask_async, ask_secret_async
+from plyngent.tools.workspace import WorkspaceError
+
+from .pty_session import PtyManager
+from .write_pty import write_pty_payload
+
+type _PromptResult = tuple[Literal["ok"], str] | tuple[Literal["err"], str]
+
+
+def _status_without_payload(
+    status: str,
+    *,
+    secret: bool,
+    submit: bool,
+) -> str:
+    lines = [line for line in status.splitlines() if not line.startswith("wrote=")]
+    lines.append("wrote=true")
+    lines.append(f"secret={'true' if secret else 'false'}")
+    lines.append(f"submit={'true' if submit else 'false'}")
+    lines.append("source=human")
+    return "\n".join(lines)
+
+
+async def _prompt_answer(label: str, *, secret: bool) -> _PromptResult:
+    try:
+        answer = await ask_secret_async(label) if secret else await ask_async(label)
+    except NonInteractiveError as exc:
+        return ("err", f"error: {exc}")
+    except KeyboardInterrupt, EOFError:
+        return ("err", "error: prompt cancelled")
+    if answer == "":
+        return ("err", "error: empty input cancelled (nothing written to PTY)")
+    return ("ok", answer)
+
+
+@tool
+async def ask_into_pty(
+    session_id: int,
+    message: str,
+    *,
+    secret: bool = False,
+    submit: bool = True,
+) -> str:
+    """Prompt the **human** and write their answer into a PTY (local only).
+
+    Use for interactive prompts (e.g. sudo/ssh password). The answer is written
+    to the PTY master on this machine and is **never** included in the tool
+    result (so it is not sent to external model APIs).
+
+    ``message`` is shown to the human (what to enter), not the secret itself.
+    ``secret=true`` uses no-echo input. ``submit=true`` (default) appends a
+    newline after the answer. Empty input and Ctrl+C cancel without writing.
+    """
+    label = message.strip() or ("Secret" if secret else "Input")
+    try:
+        # Validate session before blocking the human.
+        _ = PtyManager.refresh(session_id)
+    except WorkspaceError as exc:
+        return f"error: {exc}"
+
+    kind, value = await _prompt_answer(label, secret=secret)
+    if kind == "err":
+        return value
+
+    payload = value + ("\n" if submit else "")
+    try:
+        status = write_pty_payload(session_id, payload)
+    except WorkspaceError as exc:
+        return f"error: {exc}"
+    except OSError as exc:
+        return f"error: failed to write PTY: {exc}"
+
+    return _status_without_payload(status, secret=secret, submit=submit)

src/plyngent/tools/process/command_exec.py

@@ -0,0 +1,143 @@
+from __future__ import annotations
+
+import asyncio
+import os
+import shlex
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+from plyngent.tools.workspace import (
+    WorkspaceError,
+    check_command_allowed,
+    get_workspace_root,
+    resolve_path,
+)
+
+DEFAULT_TIMEOUT_SECONDS = 30
+DEFAULT_MAX_OUTPUT_CHARS = 32_000
+DEFAULT_MAX_BATCH_STEPS = 20
+
+
+def truncate_output(text: str, label: str, max_chars: int = DEFAULT_MAX_OUTPUT_CHARS) -> str:
+    if len(text) <= max_chars:
+        return text
+    return text[:max_chars] + f"\n...[{label} truncated]"
+
+
+def format_command_result(
+    *,
+    returncode: int | None,
+    workdir_display: str,
+    command: list[str],
+    stdout: str,
+    stderr: str,
+    timed_out: bool = False,
+) -> str:
+    parts = [
+        f"exit_code={returncode if returncode is not None else ''}",
+        f"timed_out={'true' if timed_out else 'false'}",
+        f"cwd={workdir_display}",
+        f"cmd={shlex.join(command)}",
+        "--- stdout ---",
+        truncate_output(stdout, "stdout"),
+        "--- stderr ---",
+        truncate_output(stderr, "stderr"),
+    ]
+    return "\n".join(parts)
+
+
+def resolve_workdir(cwd: str) -> tuple["Path", str]:
+    """Return (absolute workdir, display path relative to workspace when possible)."""
+    workdir = resolve_path(cwd)
+    if not workdir.is_dir():
+        msg = f"not a directory: {cwd}"
+        raise WorkspaceError(msg)
+    try:
+        workdir_display = str(workdir.relative_to(get_workspace_root()))
+    except ValueError:
+        workdir_display = str(workdir)
+    return workdir, workdir_display
+
+
+@dataclass
+class CommandStepResult:
+    command: list[str]
+    cwd_display: str
+    exit_code: int | None
+    timed_out: bool
+    stdout: str
+    stderr: str
+    mix_stderr: bool
+    captured: str  # piped to the next step when provider sets pipe_out
+
+
+async def execute_argv(
+    command: list[str],
+    *,
+    cwd: str = ".",
+    env: dict[str, str] | None = None,
+    stdin: str | None = None,
+    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."""
+    if not command:
+        msg = "command argv must not be empty"
+        raise WorkspaceError(msg)
+    check_command_allowed(command)
+    workdir, workdir_display = resolve_workdir(cwd)
+
+    stdin_data = None if stdin is None else stdin.encode()
+    run_env: dict[str, str] | None = None
+    if env is not None:
+        run_env = {str(k): str(v) for k, v in {**os.environ, **env}.items()}
+
+    try:
+        stderr_arg = asyncio.subprocess.STDOUT if mix_stderr else asyncio.subprocess.PIPE
+        proc = await asyncio.create_subprocess_exec(
+            *command,
+            cwd=str(workdir),
+            env=run_env,
+            stdin=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=stderr_arg,
+        )
+    except FileNotFoundError:
+        msg = f"executable not found: {command[0]!r}"
+        raise WorkspaceError(msg) from None
+    except OSError as exc:
+        msg = f"failed to start command: {exc}"
+        raise WorkspaceError(msg) from exc
+
+    timed_out = False
+    try:
+        stdout_b, stderr_b = await asyncio.wait_for(
+            proc.communicate(input=stdin_data),
+            timeout=timeout_seconds,
+        )
+    except TimeoutError:
+        timed_out = True
+        proc.kill()
+        stdout_b, stderr_b = await proc.communicate()
+
+    stdout = (stdout_b or b"").decode(errors="replace")
+    if mix_stderr:
+        stderr = ""
+        captured = stdout
+    else:
+        stderr = (stderr_b or b"").decode(errors="replace")
+        captured = stdout
+
+    return CommandStepResult(
+        command=list(command),
+        cwd_display=workdir_display,
+        exit_code=proc.returncode,
+        timed_out=timed_out,
+        stdout=stdout,
+        stderr=stderr,
+        mix_stderr=mix_stderr,
+        captured=captured,
+    )

src/plyngent/tools/process/pty_backend.py

@@ -258,6 +258,34 @@ else:
             raise OSError(msg)
         return _spawn_posix(command, cwd=cwd)
 
+    def _claim_controlling_tty(slave_fd: int) -> None:
+        """Make *slave_fd* the controlling terminal (session leader must call this).
+
+        Best-effort: required for programs like ``sudo`` that open ``/dev/tty``
+        and refuse a non-controlling PTY. Failures are ignored so plain tools
+        still run if the ioctl is unavailable.
+        """
+        import fcntl
+        import termios
+
+        # Linux: arg 0 = become controlling tty of this session.
+        tio_csctty = getattr(termios, "TIOCSCTTY", None)
+        if tio_csctty is not None:
+            with contextlib.suppress(OSError):
+                _ = fcntl.ioctl(slave_fd, tio_csctty, 0)
+                return
+        # Fallback: reopen slave device path (some BSDs / older kernels).
+        with contextlib.suppress(OSError):
+            name = os.ttyname(slave_fd)
+            reopened = os.open(name, os.O_RDWR)
+            try:
+                if reopened != slave_fd:
+                    _ = os.dup2(reopened, slave_fd)
+            finally:
+                if reopened != slave_fd:
+                    with contextlib.suppress(OSError):
+                        os.close(reopened)
+
     def _spawn_posix(command: list[str], *, cwd: Path) -> PosixPtyHandle:
         import pty
 
@@ -269,6 +297,8 @@ else:
             try:
                 os.close(master_fd)
                 _ = os.setsid()
+                # Claim slave as controlling TTY before wiring stdio (sudo / PAM).
+                _claim_controlling_tty(slave_fd)
                 _ = os.dup2(slave_fd, 0)
                 _ = os.dup2(slave_fd, 1)
                 _ = os.dup2(slave_fd, _STDERR_FD)

src/plyngent/tools/process/run_command.py

@@ -1,133 +1,14 @@
 from __future__ import annotations
 
-import asyncio
-import os
-import shlex
-from typing import TYPE_CHECKING
-
 from plyngent.agent import tool
-from plyngent.tools.workspace import (
+from plyngent.tools.workspace import WorkspaceError
-    WorkspaceError,
+
-    check_command_allowed,
+from .command_exec import (
-    get_workspace_root,
+    DEFAULT_TIMEOUT_SECONDS,
-    resolve_path,
+    execute_argv,
+    format_command_result,
 )
 
-if TYPE_CHECKING:
-    from pathlib import Path
-
-DEFAULT_TIMEOUT_SECONDS = 30
-DEFAULT_MAX_OUTPUT_CHARS = 32_000
-
-
-def _truncate(text: str, label: str) -> str:
-    if len(text) <= DEFAULT_MAX_OUTPUT_CHARS:
-        return text
-    return text[:DEFAULT_MAX_OUTPUT_CHARS] + f"\n...[{label} truncated]"
-
-
-def _format_result(
-    *,
-    returncode: int | None,
-    workdir_display: str,
-    command: list[str],
-    stdout: str,
-    stderr: str,
-    timed_out: bool = False,
-) -> str:
-    parts = [
-        f"exit_code={'' if returncode is None else returncode}",
-        f"timed_out={'true' if timed_out else 'false'}",
-        f"cwd={workdir_display}",
-        f"cmd={shlex.join(command)}",
-    ]
-    if stdout:
-        parts.append("--- stdout ---\n" + stdout.rstrip("\n"))
-    if stderr:
-        parts.append("--- stderr ---\n" + stderr.rstrip("\n"))
-    return "\n".join(parts)
-
-
-def _validate_env(env: object) -> str | None:
-    """Runtime guard for tool-JSON args (may not match static typing at the boundary)."""
-    if env is None:
-        return None
-    if type(env) is not dict:
-        return "error: env must be an object of string keys and values"
-    # Tool schema should already enforce dict[str, str]; keep a light check.
-    return None
-
-
-def _merge_env(overrides: dict[str, str] | None) -> dict[str, str] | None:
-    if overrides is None:
-        return None
-    return {**os.environ, **overrides}
-
-
-async def _run_exec(
-    command: list[str],
-    *,
-    workdir: str,
-    timeout_seconds: float,
-    stdin_data: bytes | None,
-    env: dict[str, str] | None,
-) -> tuple[int | None, str, str, bool] | str:
-    """Return ``(returncode, stdout, stderr, timed_out)`` or an error string."""
-    stdin = asyncio.subprocess.PIPE if stdin_data is not None else None
-    try:
-        proc = await asyncio.create_subprocess_exec(
-            *command,
-            cwd=workdir,
-            env=env,
-            stdin=stdin,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-    except FileNotFoundError:
-        return f"error: executable not found: {command[0]}"
-    except OSError as exc:
-        return f"error: failed to start command: {exc}"
-
-    timed_out = False
-    try:
-        stdout_b, stderr_b = await asyncio.wait_for(
-            proc.communicate(input=stdin_data),
-            timeout=timeout_seconds,
-        )
-    except TimeoutError:
-        timed_out = True
-        proc.kill()
-        stdout_b, stderr_b = await proc.communicate()
-
-    stdout = _truncate(stdout_b.decode(errors="replace"), "stdout")
-    stderr = _truncate(stderr_b.decode(errors="replace"), "stderr")
-    return proc.returncode, stdout, stderr, timed_out
-
-
-def _validate_run_args(
-    command: list[str],
-    *,
-    cwd: str,
-    timeout_seconds: float,
-    env: dict[str, str] | None,
-) -> Path | str:
-    """Return resolved workdir, or an error string."""
-    if not command:
-        return "error: command must not be empty"
-    try:
-        check_command_allowed(command)
-        workdir = resolve_path(cwd)
-    except WorkspaceError as exc:
-        return f"error: {exc}"
-    if not workdir.is_dir():
-        return f"error: cwd is not a directory: {cwd}"
-    if timeout_seconds <= 0:
-        return "error: timeout_seconds must be > 0"
-    env_error = _validate_env(env)
-    if env_error is not None:
-        return env_error
-    return workdir
-
 
 @tool
 async def run_command(
@@ -138,35 +19,24 @@ async def run_command(
     stdin: str | None = None,
     env: dict[str, str] | None = None,
 ) -> str:
-    """Run a command without a shell (argv list) under the workspace.
+    """Run ``command`` (argv, no shell) under the workspace; capture stdout/stderr."""
+    try:
+        result = await execute_argv(
+            command,
+            cwd=cwd,
+            env=env,
+            stdin=stdin,
+            timeout_seconds=timeout_seconds,
+            mix_stderr=False,
+        )
+    except WorkspaceError as exc:
+        return f"error: {exc}"
 
-    ``cwd`` is relative to or under the workspace root. Optional ``stdin`` is
+    return format_command_result(
-    written to the process stdin. Optional ``env`` overlays process environment
+        returncode=result.exit_code,
-    variables (merged with the current environment). Output is truncated.
+        workdir_display=result.cwd_display,
-
+        command=result.command,
-    On timeout the process is killed and any partial stdout/stderr is still
+        stdout=result.stdout,
-    returned with ``timed_out=true``.
+        stderr=result.stderr,
-    """
+        timed_out=result.timed_out,
-    workdir = _validate_run_args(command, cwd=cwd, timeout_seconds=timeout_seconds, env=env)
-    if isinstance(workdir, str):
-        return workdir
-
-    stdin_data = None if stdin is None else stdin.encode()
-    result = await _run_exec(
-        command,
-        workdir=str(workdir),
-        timeout_seconds=timeout_seconds,
-        stdin_data=stdin_data,
-        env=_merge_env(env),
-    )
-    if isinstance(result, str):
-        return result
-    returncode, stdout, stderr, timed_out = result
-    return _format_result(
-        returncode=returncode,
-        workdir_display=str(workdir.relative_to(get_workspace_root())),
-        command=command,
-        stdout=stdout,
-        stderr=stderr,
-        timed_out=timed_out,
     )

src/plyngent/tools/process/run_command_batch.py

@@ -0,0 +1,214 @@
+from __future__ import annotations
+
+import json
+import shlex
+from typing import Any, cast
+
+from plyngent.agent import tool
+from plyngent.tools.workspace import WorkspaceError
+
+from .command_exec import (
+    DEFAULT_MAX_BATCH_STEPS,
+    DEFAULT_MAX_OUTPUT_CHARS,
+    DEFAULT_TIMEOUT_SECONDS,
+    CommandStepResult,
+    execute_argv,
+    truncate_output,
+)
+
+BATCH_OUTPUT_SOFT_CAP_FACTOR = 4
+
+
+def _as_bool(value: object, *, default: bool = False) -> bool:
+    if value is None:
+        return default
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        return value.strip().lower() in {"1", "true", "yes", "on"}
+    return bool(value)
+
+
+def _as_float(value: object, *, default: float) -> float:
+    if value is None:
+        return default
+    if isinstance(value, (int, float)):
+        return float(value)
+    if isinstance(value, str) and value.strip():
+        return float(value)
+    return default
+
+
+def _as_str_dict(value: object) -> dict[str, str] | None:
+    if value is None:
+        return None
+    if not isinstance(value, dict):
+        msg = "env must be an object of string keys/values"
+        raise WorkspaceError(msg)
+    return {str(key): str(val) for key, val in cast("dict[object, object]", value).items()}
+
+
+def _parse_step(raw: object, index: int) -> dict[str, Any]:
+    if not isinstance(raw, dict):
+        msg = f"commands[{index}] must be an object"
+        raise WorkspaceError(msg)
+    data = cast("dict[str, object]", raw)
+    command = data.get("command")
+    if not isinstance(command, list) or not command:
+        msg = f"commands[{index}].command must be a non-empty argv list"
+        raise WorkspaceError(msg)
+    command_parts = cast("list[object]", command)
+    argv = [part for part in command_parts if isinstance(part, str)]
+    if len(argv) != len(command_parts):
+        msg = f"commands[{index}].command 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,
+        "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),
+        "pipe_out": _as_bool(data.get("pipe_out"), default=False),
+        "mix_stderr": _as_bool(data.get("mix_stderr"), default=False),
+        "stop_on_error": None if stop is None else _as_bool(stop, default=True),
+        "timeout_seconds": _as_float(data.get("timeout_seconds"), default=DEFAULT_TIMEOUT_SECONDS),
+    }
+
+
+def _normalize_commands(commands: list[dict[str, object]] | str) -> list[object]:
+    if isinstance(commands, str):
+        try:
+            loaded: object = json.loads(commands)
+        except json.JSONDecodeError as exc:
+            msg = f"commands 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"
+            raise WorkspaceError(msg)
+        return cast("list[object]", loaded)
+    return cast("list[object]", commands)
+
+
+def _format_step(index: int, result: CommandStepResult, *, pipe_out: bool) -> str:
+    return "\n".join(
+        [
+            f"--- step {index} ---",
+            f"exit_code={result.exit_code if result.exit_code is not None else ''}",
+            f"timed_out={'true' if result.timed_out else 'false'}",
+            f"cwd={result.cwd_display}",
+            f"cmd={shlex.join(result.command)}",
+            f"pipe_out={'true' if pipe_out else 'false'}",
+            f"mix_stderr={'true' if result.mix_stderr else 'false'}",
+            "--- stdout ---",
+            truncate_output(result.stdout, "stdout"),
+            "--- stderr ---",
+            truncate_output(result.stderr, "stderr"),
+        ]
+    )
+
+
+def _merge_env(base_env: dict[str, str] | None, step_env: dict[str, str] | None) -> dict[str, str] | None:
+    if base_env is None and step_env is None:
+        return None
+    return {**(base_env or {}), **(step_env or {})}
+
+
+async def _run_batch_steps(
+    steps: list[dict[str, Any]],
+    *,
+    cwd: str,
+    env: dict[str, str] | None,
+    stop_on_error: bool,
+) -> tuple[list[tuple[dict[str, Any], CommandStepResult]], bool]:
+    results: list[tuple[dict[str, Any], CommandStepResult]] = []
+    prev_capture: str | None = None
+    prev_piped = False
+    stopped_early = False
+    total_chars = 0
+
+    for index, step in enumerate(steps):
+        stdin_text = prev_capture if prev_piped else step["stdin"]
+        result = await execute_argv(
+            step["command"],
+            cwd=step["cwd"] if step["cwd"] is not None else cwd,
+            env=_merge_env(env, step["env"]),
+            stdin=stdin_text,
+            timeout_seconds=step["timeout_seconds"],
+            mix_stderr=step["mix_stderr"],
+        )
+        results.append((step, result))
+        prev_capture = result.captured
+        prev_piped = bool(step["pipe_out"])
+        total_chars += len(result.stdout) + len(result.stderr)
+
+        failed = bool(result.timed_out) or result.exit_code != 0
+        effective_stop = stop_on_error if step["stop_on_error"] is None else bool(step["stop_on_error"])
+        if failed and effective_stop:
+            return results, True
+        if total_chars > DEFAULT_MAX_OUTPUT_CHARS * BATCH_OUTPUT_SOFT_CAP_FACTOR:
+            return results, True
+        del index  # used for clarity in loop only
+
+    return results, stopped_early
+
+
+@tool
+async def run_command_batch(
+    commands: list[dict[str, object]] | str,
+    *,
+    cwd: str = ".",
+    env: dict[str, str] | None = None,
+    stop_on_error: bool = True,
+) -> str:
+    """Run a serial batch of argv commands (no shell).
+
+    Each element of ``commands`` is an object::
+
+        {
+          "command": ["git", "status"],   # required argv
+          "cwd": ".",                     # optional, else batch cwd
+          "env": {"FOO": "1"},            # optional overlay
+          "stdin": "...",                 # optional; unused if previous pipe_out
+          "pipe_out": false,              # if true, feed capture into *next* stdin
+          "mix_stderr": false,            # OS-level merge stderr into capture
+          "stop_on_error": null,          # override batch stop_on_error
+          "timeout_seconds": 30
+        }
+
+    ``pipe_out`` is set on the **provider** step. Last-step ``pipe_out`` is an error.
+    Default ``stop_on_error`` is true. Max 20 steps; total output budget 32k chars.
+    """
+    try:
+        raw_steps = _normalize_commands(commands)
+        if not raw_steps:
+            return "error: commands must not be empty"
+        if len(raw_steps) > DEFAULT_MAX_BATCH_STEPS:
+            return f"error: at most {DEFAULT_MAX_BATCH_STEPS} commands 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"
+
+        results, stopped_early = await _run_batch_steps(steps, cwd=cwd, env=env, stop_on_error=stop_on_error)
+        body = "\n".join(
+            _format_step(i, result, pipe_out=bool(step["pipe_out"])) for i, (step, result) in enumerate(results)
+        )
+        if len(body) > DEFAULT_MAX_OUTPUT_CHARS:
+            body = truncate_output(body, "batch")
+        last_exit = results[-1][1].exit_code if results else ""
+        return "\n".join(
+            [
+                f"steps={len(steps)} ran={len(results)} "
+                f"stop_on_error={'true' if stop_on_error else 'false'} "
+                f"stopped_early={'true' if stopped_early else 'false'}",
+                body,
+                "--- summary ---",
+                f"last_exit={last_exit if last_exit is not None else ''}",
+            ]
+        )
+    except WorkspaceError as exc:
+        return f"error: {exc}"
+    except (TypeError, ValueError) as exc:
+        return f"error: invalid batch arguments: {exc}"

src/plyngent/tools/todo.py

@@ -42,7 +42,9 @@ def _notify() -> None:
 def todo_list() -> str:
     """Show the LIFO stack of **task groups** (TOP group = current breakdown level).
 
-    Push creates one group of siblings; pop removes the whole top group.
+    Non-empty stack with open (pending/in_progress) items = unfinished work.
+    All-terminal but non-empty = hygiene only (pop/clear). Push creates one
+    group of siblings; pop removes the whole top group.
     Pattern: push [T1,T2] → push [T1.1,T1.2] → finish children → pop → push [T2.1]…
     """
     stack = _require_stack()
@@ -74,7 +76,11 @@ def todo_push(titles: list[str], notes: str = "") -> str:
 
 @tool(name="todo_pop")
 def todo_pop() -> str:
-    """Pop the entire **top group** (all siblings from that push)."""
+    """Pop the entire **top group** (all siblings from that push).
+
+    Prefer after TOP items are done/cancelled so the stack does not stay
+    non-empty with only finished work.
+    """
     stack = _require_stack()
     group = stack.pop()
     if group is None:
@@ -93,7 +99,12 @@ def todo_update(
     title: str = "",
     notes: str = "",
 ) -> str:
-    """Update a task by id inside any group. Pop the group when that level is done."""
+    """Update a task by id inside any group.
+
+    Open items are unfinished work: mark done/cancelled when the work is truly
+    finished (not just deferred). Pop the TOP group when that breakdown level
+    is complete so the stack does not linger as false open work.
+    """
     stack = _require_stack()
     status_arg: TodoStatus | None = None
     if status.strip():

src/plyngent/tools/workspace.py

@@ -1,6 +1,10 @@
 from __future__ import annotations
 
 from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from collections.abc import Callable, Sequence
 
 DEFAULT_COMMAND_DENYLIST: frozenset[str] = frozenset(
     {
@@ -29,6 +33,12 @@ DEFAULT_COMMAND_DENYLIST: frozenset[str] = frozenset(
 # Max concurrent temporary workspaces registered in one process.
 MAX_TEMPORARY_WORKSPACES = 16
 
+# Timed human override for command denylist (independent of YOLO soft-confirm).
+DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS = 30.0
+
+# Hook: (basename, argv, timeout_seconds) -> True allow for this session basename.
+type PolicyConfirmHook = Callable[[str, Sequence[str], float], bool]
+
 
 class WorkspaceError(ValueError):
     """Raised when a path or command violates workspace policy."""
@@ -42,10 +52,17 @@ class _WorkspaceState:
     allowlist: list[Path]
     # Paths created by new_temporary_workspace (subset of allowlist); cleaned on chat exit.
     temporary_owned: list[Path]
+    # Basenames the human allowed this process (denylist override, not permanent).
+    policy_allowed_commands: set[str]
+    policy_confirm_hook: PolicyConfirmHook | None
+    policy_confirm_timeout_seconds: float
 
     def __init__(self) -> None:
         self.allowlist = []
         self.temporary_owned = []
+        self.policy_allowed_commands = set()
+        self.policy_confirm_hook = None
+        self.policy_confirm_timeout_seconds = DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS
 
 
 _state = _WorkspaceState()
@@ -87,12 +104,47 @@ def get_path_denylist() -> tuple[str, ...]:
 def set_command_denylist(names: list[str] | tuple[str, ...] | frozenset[str] | None) -> None:
     """Set denied command basenames (None restores defaults)."""
     _state.command_denylist = DEFAULT_COMMAND_DENYLIST if names is None else frozenset(names)
+    # Session overrides only make sense against the active denylist.
+    _state.policy_allowed_commands &= _state.command_denylist
 
 
 def get_command_denylist() -> frozenset[str]:
     return _state.command_denylist
 
 
+def set_policy_confirm_hook(hook: PolicyConfirmHook | None) -> None:
+    """Register a timed human confirm for denylisted commands (CLI installs this)."""
+    _state.policy_confirm_hook = hook
+
+
+def get_policy_confirm_hook() -> PolicyConfirmHook | None:
+    return _state.policy_confirm_hook
+
+
+def set_policy_confirm_timeout(seconds: float) -> None:
+    """Timeout for policy confirm prompts (must be > 0)."""
+    if seconds <= 0:
+        msg = "policy confirm timeout must be > 0"
+        raise WorkspaceError(msg)
+    _state.policy_confirm_timeout_seconds = float(seconds)
+
+
+def get_policy_confirm_timeout() -> float:
+    return _state.policy_confirm_timeout_seconds
+
+
+def clear_policy_allowed_commands() -> None:
+    """Drop session-scoped denylist overrides (tests / chat exit)."""
+    _state.policy_allowed_commands.clear()
+
+
+def grant_policy_command(basename: str) -> None:
+    """Allow *basename* for the rest of this process despite the denylist."""
+    name = basename.strip()
+    if name:
+        _state.policy_allowed_commands.add(name)
+
+
 def add_workspace_allowlist(root: Path | str, *, owned: bool = False) -> Path:
     """Allow tool paths under *root* in addition to the primary workspace.
 
@@ -180,11 +232,34 @@ def resolve_path(path: str | Path) -> Path:
 
 
 def check_command_allowed(argv: list[str]) -> None:
-    """Raise if argv is empty or the executable basename is denylisted."""
+    """Raise if argv is empty or the executable basename is denylisted.
+
+    Denylisted basenames are not hard-rejected when a policy confirm hook is
+    installed: the human is asked (with a timeout; default deny). Session grants
+    skip re-prompting for the same basename. Independent of YOLO soft-confirm.
+    """
     if not argv:
         msg = "command argv must not be empty"
         raise WorkspaceError(msg)
     binary = Path(argv[0]).name
-    if binary in _state.command_denylist:
+    if binary not in _state.command_denylist:
-        msg = f"command denied by policy (basename {binary!r} is blocked)"
+        return
+    if binary in _state.policy_allowed_commands:
+        return
+    hook = _state.policy_confirm_hook
+    if hook is not None:
+        timeout = _state.policy_confirm_timeout_seconds
+        try:
+            allowed = bool(hook(binary, list(argv), timeout))
+        except Exception as exc:
+            msg = f"command denied by policy (basename {binary!r}; confirm failed: {exc})"
+            raise WorkspaceError(msg) from exc
+        if allowed:
+            _state.policy_allowed_commands.add(binary)
+            return
+        msg = (
+            f"command denied by policy (basename {binary!r} is blocked; user declined or timed out after {timeout:g}s)"
+        )
         raise WorkspaceError(msg)
+    msg = f"command denied by policy (basename {binary!r} is blocked)"
+    raise WorkspaceError(msg)

tests/test_agent/test_todo_stack.py

@@ -3,17 +3,21 @@ from __future__ import annotations
 from typing import TYPE_CHECKING, Literal, overload
 
 import pytest
+from msgspec import UNSET
 
 from plyngent.agent import ChatAgent, ToolRegistry
+from plyngent.agent.todo_nag import inject_todo_nag, parse_todo_nag_strategy
 from plyngent.agent.todo_stack import TodoStack, parse_push_titles
 from plyngent.config.models import DatabaseConfig
 from plyngent.lmproto.openai_compatible.model import (
+    AnyChatMessage,
     AssistantChatMessage,
     ChatCompletionChoice,
     ChatCompletionChunk,
     ChatCompletionResponse,
     ChatCompletionsParam,
     DeveloperChatMessage,
+    ToolChatMessage,
     UserChatMessage,
 )
 from plyngent.memory import MemoryStore
@@ -103,13 +107,119 @@ def test_single_title_still_one_group() -> None:
 def test_todo_stack_needs_review() -> None:
     stack = TodoStack()
     assert not stack.needs_review()
-    _ = stack.push("work")
+    item = stack.push("work")
     stack.begin_turn()
+    # Open items always need review, even if todo_* was used this turn.
+    assert stack.needs_review()
+    stack.mark_touched()
+    assert stack.needs_review()
+    stack.update(item.id, status="done")
+    stack.begin_turn()
+    # Terminal-only stack: review only when untouched this turn.
     assert stack.needs_review()
     stack.mark_touched()
     assert not stack.needs_review()
 
 
+def test_parse_todo_nag_strategy() -> None:
+    assert parse_todo_nag_strategy("developer") == "developer"
+    assert parse_todo_nag_strategy("SYNTHETIC-TOOL") == "synthetic_tool"
+    assert parse_todo_nag_strategy("nope") == "developer"
+    assert parse_todo_nag_strategy(None) == "developer"
+    # Legacy alias — mid-turn system was not a useful Responses channel.
+    assert parse_todo_nag_strategy("system") == "developer"
+
+
+def test_inject_todo_nag_strategies() -> None:
+    from plyngent.agent.events import ToolCallEvent, ToolResultEvent
+    from plyngent.agent.todo_nag import (
+        inject_todo_nag_for_stack_with_events,
+        inject_todo_nag_with_events,
+        synthetic_todo_list_result,
+    )
+
+    body = "[TODO OPEN WORK] test"
+    messages: list[AnyChatMessage] = []
+    assert inject_todo_nag(messages, body, strategy="none") is False
+    assert messages == []
+
+    messages = []
+    assert inject_todo_nag(messages, body, strategy="developer") is True
+    assert isinstance(messages[0], DeveloperChatMessage)
+    assert body in messages[0].content
+
+    messages = []
+    assert inject_todo_nag(messages, body, strategy="user") is True
+    assert isinstance(messages[0], UserChatMessage)
+
+    # synthetic_tool: real todo_list-shaped body (stack.render), not OPEN WORK prose
+    stack = TodoStack()
+    _ = stack.push("synthetic body item")
+    real = synthetic_todo_list_result(stack)
+    assert real == stack.render()
+    assert "[TODO OPEN WORK]" not in real
+    assert "synthetic body item" in real
+
+    messages = []
+    ok, events = inject_todo_nag_for_stack_with_events(
+        messages,
+        stack,
+        kind="end_of_turn",
+        strategy="synthetic_tool",
+    )
+    assert ok is True
+    assert len(messages) == 2
+    assert isinstance(messages[0], AssistantChatMessage)
+    tool_calls = messages[0].tool_calls
+    assert tool_calls is not UNSET and tool_calls
+    assert isinstance(messages[1], ToolChatMessage)
+    tool_content = messages[1].content
+    assert isinstance(tool_content, str)
+    assert tool_content == real
+    assert messages[1].tool_call_id.startswith("todo-nag-")
+    assert len(events) == 2
+    assert isinstance(events[0], ToolCallEvent)
+    assert isinstance(events[1], ToolResultEvent)
+    result_event = events[1]
+    assert isinstance(result_event, ToolResultEvent)
+    result_content = result_event.message.content
+    assert isinstance(result_content, str)
+    assert result_content == real
+
+    # Raw inject still accepts an explicit body (e.g. tests / custom)
+    messages = []
+    ok2, events2 = inject_todo_nag_with_events(messages, body, strategy="synthetic_tool")
+    assert ok2
+    assert len(events2) == 2
+    assert isinstance(events2[1], ToolResultEvent)
+    raw_content = events2[1].message.content
+    assert isinstance(raw_content, str)
+    assert body in raw_content
+
+
+def test_todo_prompts_signal_undone_work() -> None:
+    stack = TodoStack()
+    item = stack.push("open work")
+    reminder = stack.turn_reminder_prompt()
+    assert "[TODO OPEN WORK]" in reminder
+    assert "Stack not empty" in reminder
+    assert "unfinished work" in reminder.lower()
+    assert "open work" in reminder
+
+    review = stack.review_prompt()
+    assert "[TODO OPEN WORK]" in review
+    assert "undone work" in review.lower() or "open item" in review.lower()
+    assert "open work" in review
+    assert "t1:open work" in review or "open work" in review
+
+    stack.update(item.id, status="done")
+    terminal_review = stack.review_prompt()
+    assert "[TODO HYGIENE]" in terminal_review
+    assert "done/cancelled" in terminal_review
+    terminal_reminder = stack.turn_reminder_prompt()
+    assert "[TODO HYGIENE]" in terminal_reminder
+
+
 def test_legacy_flat_and_frames_migrate() -> None:
     flat = TodoStack.from_raw(
         {
@@ -232,7 +342,90 @@ async def test_loop_injects_todo_review_when_untouched() -> None:
         async for _event in agent.run("do stuff"):
             pass
         assert client.calls >= 2
-        assert any(isinstance(m, DeveloperChatMessage) and "[TODO OPEN]" in m.content for m in agent.messages)
+        assert any(isinstance(m, DeveloperChatMessage) and "[TODO OPEN WORK]" in m.content for m in agent.messages)
-        assert not any(isinstance(m, UserChatMessage) and "[TODO OPEN]" in m.content for m in agent.messages)
+        assert not any(isinstance(m, UserChatMessage) and "[TODO OPEN WORK]" in m.content for m in agent.messages)
+    finally:
+        set_todo_stack(None)
+
+
+@pytest.mark.asyncio
+async def test_loop_injects_todo_review_when_open_after_touch() -> None:
+    """Open items still trigger end-of-turn review even if todo_* ran this turn."""
+    stack = TodoStack()
+    _ = stack.push("still open")
+    stack.begin_turn()
+    stack.mark_touched()  # simulates todo_list earlier in the turn
+    assert stack.needs_review()
+
+    client = ScriptedClient()
+    agent = ChatAgent(
+        client,  # type: ignore[arg-type]
+        model="m",
+        tools=ToolRegistry(list(TODO_TOOLS)),
+        stream=False,
+        todo_stack=stack,
+    )
+    set_todo_stack(stack)
+    try:
+        async for _event in agent.run("do stuff"):
+            pass
+        assert client.calls >= 2
+        assert any(isinstance(m, DeveloperChatMessage) and "[TODO OPEN WORK]" in m.content for m in agent.messages)
+    finally:
+        set_todo_stack(None)
+
+
+@pytest.mark.asyncio
+async def test_loop_synthetic_tool_nag_strategy() -> None:
+    stack = TodoStack()
+    _ = stack.push("open work")
+    stack.begin_turn()
+    client = ScriptedClient()
+    agent = ChatAgent(
+        client,  # type: ignore[arg-type]
+        model="m",
+        tools=ToolRegistry(list(TODO_TOOLS)),
+        stream=False,
+        todo_stack=stack,
+        todo_nag_strategy="synthetic_tool",
+    )
+    set_todo_stack(stack)
+    try:
+        async for _event in agent.run("do stuff"):
+            pass
+        # Result body is real todo_list shape (render), not OPEN WORK prose.
+        assert any(
+            isinstance(m, ToolChatMessage) and "open work" in m.content and "[TODO OPEN WORK]" not in m.content
+            for m in agent.messages
+        )
+        assert not any(isinstance(m, DeveloperChatMessage) and "[TODO OPEN WORK]" in m.content for m in agent.messages)
+    finally:
+        set_todo_stack(None)
+
+
+@pytest.mark.asyncio
+async def test_loop_none_nag_strategy_skips_inject() -> None:
+    stack = TodoStack()
+    _ = stack.push("open work")
+    stack.begin_turn()
+    client = ScriptedClient()
+    agent = ChatAgent(
+        client,  # type: ignore[arg-type]
+        model="m",
+        tools=ToolRegistry(list(TODO_TOOLS)),
+        stream=False,
+        todo_stack=stack,
+        todo_nag_strategy="none",
+    )
+    set_todo_stack(stack)
+    try:
+        async for _event in agent.run("do stuff"):
+            pass
+        # One completion only — no end-of-turn continue from nag.
+        assert client.calls == 1
+        assert not any(
+            isinstance(m, (DeveloperChatMessage, ToolChatMessage)) and "[TODO OPEN WORK]" in getattr(m, "content", "")
+            for m in agent.messages
+        )
     finally:
         set_todo_stack(None)

tests/test_cli/test_app.py

@@ -1,10 +1,102 @@
-from __future__ import annotations
-
 from pathlib import Path
 
+import pytest
 from click.testing import CliRunner
 
-from plyngent.cli.app import main
+from plyngent.cli.app import _database_config, main
+from plyngent.config import load as load_config
+
+
+def test_database_config_memory_url_kept_with_warning(
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """Explicit ``url = ":memory:"`` is not rewritten; CLI warns (when not quiet)."""
+    path = tmp_path / "mem.toml"
+    _ = path.write_text(
+        """
+[database]
+implementation = "sqlite"
+url = ":memory:"
+""",
+        encoding="utf-8",
+    )
+    store = load_config(path)
+    cfg = _database_config(store, quiet=False)
+    assert cfg.url == ":memory:"
+    assert cfg.implementation == "sqlite"
+    err = capsys.readouterr().err
+    assert ":memory:" in err
+    assert "not persisted" in err.lower() or "warning" in err.lower()
+
+
+def test_database_config_memory_url_quiet_no_warn(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+    path = tmp_path / "mem.toml"
+    _ = path.write_text(
+        """
+[database]
+implementation = "sqlite"
+url = ":memory:"
+""",
+        encoding="utf-8",
+    )
+    store = load_config(path)
+    cfg = _database_config(store, quiet=True)
+    assert cfg.url == ":memory:"
+    assert capsys.readouterr().err == ""
+
+
+def test_database_config_unset_url_uses_user_data(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Unset/empty url → durable user-data chat.db."""
+    data_dir = tmp_path / "user-data"
+    _ = data_dir.mkdir()
+
+    def fake_user_data_path(*_args: object, **_kwargs: object) -> Path:
+        return data_dir
+
+    monkeypatch.setattr("plyngent.cli.app.user_data_path", fake_user_data_path)
+    path = tmp_path / "empty-url.toml"
+    _ = path.write_text(
+        """
+[database]
+implementation = "sqlite"
+""",
+        encoding="utf-8",
+    )
+    store = load_config(path)
+    cfg = _database_config(store, quiet=True)
+    assert cfg.url == str(data_dir / "chat.db")
+
+
+def test_database_config_omitted_section_uses_user_data(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Omitted [database] → durable user-data chat.db (url default None)."""
+    data_dir = tmp_path / "user-data2"
+    _ = data_dir.mkdir()
+
+    def fake_user_data_path(*_args: object, **_kwargs: object) -> Path:
+        return data_dir
+
+    monkeypatch.setattr("plyngent.cli.app.user_data_path", fake_user_data_path)
+    path = tmp_path / "no-db.toml"
+    _ = path.write_text(
+        """
+[providers.local]
+preset = "openai-compatible"
+url = "http://127.0.0.1:1/v1"
+access_key_or_token = "x"
+""",
+        encoding="utf-8",
+    )
+    store = load_config(path)
+    assert store.database.get("url") is None
+    cfg = _database_config(store, quiet=True)
+    assert cfg.url == str(data_dir / "chat.db")
 
 
 def test_providers_list() -> None:

tests/test_cli/test_editor.py

@@ -15,10 +15,13 @@ from plyngent.cli.editor import (
 )
 
 
-def test_get_editor(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_get_editor_prefers_visual(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.delenv("EDITOR", raising=False)
+    monkeypatch.delenv("VISUAL", raising=False)
     assert get_editor() is None
-    monkeypatch.setenv("EDITOR", "  codium --wait  ")
+    monkeypatch.setenv("EDITOR", "  nano  ")
+    assert get_editor() == "nano"
+    monkeypatch.setenv("VISUAL", "  codium --wait  ")
     assert get_editor() == "codium --wait"
 
 
@@ -49,7 +52,8 @@ def test_open_in_editor_splits_args(tmp_path: Path, monkeypatch: pytest.MonkeyPa
         return Result()
 
     monkeypatch.setattr("plyngent.cli.editor.subprocess.run", fake_run)
-    open_in_editor(path, editor="codium --wait")
+    outcome = open_in_editor(path, editor="codium --wait")
+    assert outcome == "waited"
     assert len(calls) == 1
     assert calls[0][0] == "codium"
     assert calls[0][1] == "--wait"
@@ -57,10 +61,33 @@ def test_open_in_editor_splits_args(tmp_path: Path, monkeypatch: pytest.MonkeyPa
     assert path.is_file()
 
 
-def test_open_in_editor_missing_editor(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+def test_open_in_editor_system_fallback(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
     monkeypatch.delenv("EDITOR", raising=False)
-    with pytest.raises(Exception, match="EDITOR"):
+    monkeypatch.delenv("VISUAL", raising=False)
-        open_in_editor(tmp_path / "x.toml", editor=None)
+    path = tmp_path / "x.toml"
+    opened: list[str] = []
+
+    def fake_system(p: Path) -> None:
+        opened.append(str(p))
+
+    monkeypatch.setattr("plyngent.cli.editor._open_with_system_default", fake_system)
+    outcome = open_in_editor(path, editor=None, allow_system_open=True)
+    assert outcome == "system"
+    assert opened == [str(path)]
+    assert path.is_file()
+
+
+def test_open_in_editor_no_fallback_requires_editor(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("EDITOR", raising=False)
+    monkeypatch.delenv("VISUAL", raising=False)
+    with pytest.raises(Exception, match=r"VISUAL|EDITOR"):
+        open_in_editor(tmp_path / "x.toml", editor=None, allow_system_open=False)
 
 
 def test_edit_text_in_editor(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -99,9 +126,13 @@ def test_edit_text_empty_cancels(monkeypatch: pytest.MonkeyPatch) -> None:
     assert edit_text_in_editor("") is None
 
 
-def test_prompt_edit_no_editor(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+def test_edit_text_no_system_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
+    from plyngent.cli.editor import edit_text_in_editor
+
     monkeypatch.delenv("EDITOR", raising=False)
-    assert prompt_edit_config(tmp_path / "p.toml") is False
+    monkeypatch.delenv("VISUAL", raising=False)
+    with pytest.raises(Exception, match=r"VISUAL|EDITOR"):
+        edit_text_in_editor("x")
 
 
 def test_prompt_edit_declined(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -111,7 +142,7 @@ def test_prompt_edit_declined(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -
         return False
 
     monkeypatch.setattr("click.confirm", _confirm)
-    assert prompt_edit_config(tmp_path / "p.toml", reason="empty") is False
+    assert prompt_edit_config(tmp_path / "p.toml", reason="empty") is None
 
 
 def test_prompt_edit_accepted(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -124,15 +155,40 @@ def test_prompt_edit_accepted(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -
     monkeypatch.setattr("click.confirm", _confirm)
     opened: list[Path] = []
 
-    def fake_open(p: Path, *, editor: str | None = None) -> None:
+    def fake_open(
-        del editor
+        p: Path,
+        *,
+        editor: str | None = None,
+        ensure_exists: bool = True,
+        allow_system_open: bool = True,
+    ) -> str:
+        del editor, ensure_exists, allow_system_open
         opened.append(p)
+        return "waited"
 
     monkeypatch.setattr("plyngent.cli.editor.open_in_editor", fake_open)
-    assert prompt_edit_config(path, reason="No providers.") is True
+    assert prompt_edit_config(path, reason="No providers.") == "waited"
     assert opened == [path]
 
 
+def test_prompt_edit_without_editor_can_system_open(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("EDITOR", raising=False)
+    monkeypatch.delenv("VISUAL", raising=False)
+
+    def _confirm(*_a: object, **_k: object) -> bool:
+        return True
+
+    monkeypatch.setattr("click.confirm", _confirm)
+    monkeypatch.setattr(
+        "plyngent.cli.editor.open_in_editor",
+        lambda *a, **k: "system",
+    )
+    assert prompt_edit_config(tmp_path / "p.toml") == "system"
+
+
 def test_config_path_command(tmp_path: Path) -> None:
     runner = CliRunner()
     result = runner.invoke(main, ["config", "path", "--config", str(tmp_path / "a.toml")])
@@ -159,3 +215,21 @@ def test_config_edit_command(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) ->
     result = runner.invoke(main, ["config", "edit", "--config", str(path)])
     assert result.exit_code == 0
     assert calls and calls[0][-1] == str(path)
+    assert "edited" in result.output
+
+
+def test_config_edit_system_fallback_message(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    path = tmp_path / "edit.toml"
+    monkeypatch.delenv("EDITOR", raising=False)
+    monkeypatch.delenv("VISUAL", raising=False)
+    monkeypatch.setattr(
+        "plyngent.cli.editor._open_with_system_default",
+        lambda p: None,
+    )
+    runner = CliRunner()
+    result = runner.invoke(main, ["config", "edit", "--config", str(path)])
+    assert result.exit_code == 0
+    assert "system default" in result.output.lower() or "system default" in (result.stderr or "").lower()

tests/test_cli/test_limits.py

@@ -41,8 +41,20 @@ def test_format_tool_confirm_box_multiline() -> None:
 
 
 def test_install_cli_limit_hooks() -> None:
+    from plyngent.tools.workspace import get_policy_confirm_hook, set_policy_confirm_hook
+
     install_cli_limit_hooks()
     assert callable(getattr(PtyManager, "_limit_continue", None))
+    assert get_policy_confirm_hook() is not None
     PtyManager.set_limit_continue_hook(None)
+    set_policy_confirm_hook(None)
     # Backend remains usable after install.
     assert get_prompt_backend().is_interactive() or True
+
+
+def test_policy_confirm_noninteractive_denies() -> None:
+    from plyngent.cli.limits import prompt_policy_command_confirm
+    from plyngent.prompting import NonInteractiveBackend
+
+    with temporary_backend(NonInteractiveBackend()):
+        assert prompt_policy_command_confirm("sudo", ["sudo", "id"], 1.0) is False

tests/test_cli/test_models_source.py

@@ -8,6 +8,7 @@ from plyngent.cli.models_source import (
     fetch_remote_model_ids,
     merge_model_choices,
     model_choices_for_provider,
+    needs_remote_models_for_selection,
 )
 from plyngent.config.models import ModelConfig, OpenAICompatibleProvider
 
@@ -55,3 +56,39 @@ async def test_fetch_remote_model_ids() -> None:
 
     with pytest.raises(TypeError, match="does not support"):
         _ = await fetch_remote_model_ids(object())
+
+
+@pytest.mark.asyncio
+async def test_fetch_remote_model_ids_timeout() -> None:
+    import asyncio
+
+    class Slow:
+        async def models(self) -> list[str]:
+            await asyncio.sleep(10)
+            return ["late"]
+
+    with pytest.raises(RuntimeError, match="timed out"):
+        _ = await fetch_remote_model_ids(Slow(), timeout_seconds=0.05)
+
+
+def test_needs_remote_models_for_selection() -> None:
+    multi = OpenAICompatibleProvider(
+        access_key_or_token="sk",
+        url="https://x/v1",
+        models={"a": ModelConfig(), "b": ModelConfig()},
+    )
+    single = OpenAICompatibleProvider(
+        access_key_or_token="sk",
+        url="https://x/v1",
+        models={"only": ModelConfig()},
+    )
+    empty = OpenAICompatibleProvider(
+        access_key_or_token="sk",
+        url="https://x/v1",
+        models={},
+    )
+    assert needs_remote_models_for_selection(multi, preferred_model="a", interactive=True) is False
+    assert needs_remote_models_for_selection(multi, preferred_model=None, interactive=False) is False
+    assert needs_remote_models_for_selection(single, preferred_model=None, interactive=True) is False
+    assert needs_remote_models_for_selection(multi, preferred_model=None, interactive=True) is True
+    assert needs_remote_models_for_selection(empty, preferred_model=None, interactive=True) is True

tests/test_cli/test_repl_commands.py

@@ -234,6 +234,13 @@ async def test_provider_switch_prompts_when_model_missing(
     state.model = "only-a"
     state.rebuild_client()
 
+    # No real GET /models — catalog is config-only for unit tests.
+    async def _no_remote(*, refresh: bool = False) -> list[str]:
+        del refresh
+        return []
+
+    monkeypatch.setattr(state, "ensure_remote_models", _no_remote)
+
     # When switching to b, only-a is missing → select_model is invoked interactively.
     monkeypatch.setattr(
         "plyngent.cli.slash.select_model",
@@ -250,7 +257,10 @@ async def test_provider_switch_prompts_when_model_missing(
     assert row.model == "only-b"
 
 
-async def test_provider_switch_keeps_shared_model(state: ReplState) -> None:
+async def test_provider_switch_keeps_shared_model(
+    state: ReplState,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
     from plyngent.config.models import ModelConfig, OpenAICompatibleProvider, OpenAIProvider
 
     state.config.providers = {
@@ -268,6 +278,12 @@ async def test_provider_switch_keeps_shared_model(state: ReplState) -> None:
     state.provider = state.config.providers["a"]
     state.model = "shared"
     state.rebuild_client()
+
+    async def _no_remote(*, refresh: bool = False) -> list[str]:
+        del refresh
+        return []
+
+    monkeypatch.setattr(state, "ensure_remote_models", _no_remote)
     assert await handle_slash(state, "/provider b") is True
     assert state.provider_name == "b"
     assert state.model == "shared"

tests/test_prompting.py

@@ -18,9 +18,16 @@ from plyngent.prompting import (
 class ScriptedBackend:
     """Deterministic backend for unit tests."""
 
-    def __init__(self, lines: list[str], *, confirms: list[bool] | None = None) -> None:
+    def __init__(
+        self,
+        lines: list[str],
+        *,
+        confirms: list[bool] | None = None,
+        secrets: list[str] | None = None,
+    ) -> None:
         self.lines = list(lines)
         self.confirms = list(confirms or [])
+        self.secrets = list(secrets or [])
         self.interactive = True
         self.echoes: list[str] = []
 
@@ -42,6 +49,13 @@ class ScriptedBackend:
         msg = "no scripted lines left"
         raise NonInteractiveError(msg)
 
+    def read_secret_line(self, prompt: str) -> str:
+        del prompt
+        if self.secrets:
+            return self.secrets.pop(0)
+        msg = "no scripted secrets left"
+        raise NonInteractiveError(msg)
+
     def confirm(self, prompt: str, *, default: bool = False) -> bool:
         del prompt
         if self.confirms:
@@ -63,6 +77,14 @@ def test_ask_free_text() -> None:
         assert ask("Name?") == "hello world"
 
 
+def test_ask_secret() -> None:
+    from plyngent.prompting import ask_secret
+
+    backend = ScriptedBackend([], secrets=["s3cret"])
+    with temporary_backend(backend):
+        assert ask_secret("Password?") == "s3cret"
+
+
 def test_ask_default_when_empty_uses_backend_default() -> None:
     backend = ScriptedBackend([])
     with temporary_backend(backend):

tests/test_tools/test_danger.py

@@ -12,18 +12,30 @@ def test_classify_delete_and_move() -> None:
 
 
 def test_classify_copy() -> None:
-    assert "copy" in (classify_danger("copy_path", {"src": "a", "dst": "b"}) or "")
+    # Without overwrite flag, copy is not soft-confirmed.
+    assert classify_danger("copy_path", {"src": "a", "dst": "b"}) is None
+    assert classify_danger("copy_path", {"src": "a", "dst": "b", "overwrite": False}) is None
 
 
-def test_classify_write_and_edits(tmp_path: Path) -> None:
+def test_classify_write_overwrite_only(tmp_path: Path) -> None:
     from plyngent.tools.workspace import clear_workspace_root, set_workspace_root
 
     set_workspace_root(tmp_path)
     try:
+        # New file: no soft-confirm
+        assert classify_danger("write_file", {"path": "new.txt"}) is None
+        # Partial edits: never soft-confirm
+        assert classify_danger("edit_replace", {"path": "x.txt"}) is None
+        assert classify_danger("edit_lineno", {"path": "x.txt", "start_line": 1, "end_line": 2}) is None
+        # Existing file write: confirm total overwrite
+        (tmp_path / "x.txt").write_text("old", encoding="utf-8")
         reason = classify_danger("write_file", {"path": "x.txt"})
-        assert reason is not None and "write file" in reason
+        assert reason is not None and "overwrite" in reason
-        assert "edit" in (classify_danger("edit_replace", {"path": "x.txt"}) or "")
+        (tmp_path / "dst.txt").write_text("d", encoding="utf-8")
-        assert "lines" in (classify_danger("edit_lineno", {"path": "x.txt", "start_line": 1, "end_line": 2}) or "")
+        (tmp_path / "src.txt").write_text("s", encoding="utf-8")
+        assert classify_danger("copy_path", {"src": "src.txt", "dst": "dst.txt", "overwrite": False}) is None
+        creason = classify_danger("copy_path", {"src": "src.txt", "dst": "dst.txt", "overwrite": True})
+        assert creason is not None and "overwrite" in creason
     finally:
         clear_workspace_root()
 
@@ -36,6 +48,28 @@ def test_classify_safe_tools() -> None:
     assert classify_danger("run_command", {"command": ["ls", "-la"]}) is None
 
 
+def test_classify_run_command_batch_risky() -> None:
+    reason = classify_danger(
+        "run_command_batch",
+        {
+            "commands": [
+                {"command": ["echo", "ok"]},
+                {"command": ["bash", "-c", "echo risky"]},
+            ]
+        },
+    )
+    assert reason is not None
+    assert "run_command_batch" in reason
+    assert "risky" in reason or "bash" in reason
+    assert (
+        classify_danger(
+            "run_command_batch",
+            {"commands": [{"command": ["echo", "ok"]}]},
+        )
+        is None
+    )
+
+
 def test_classify_shell_and_dash_c() -> None:
     r = classify_danger("run_command", {"command": ["bash", "-c", "rm -rf /"]})
     assert r is not None
@@ -79,3 +113,21 @@ async def test_confirm_deny_with_comment() -> None:
     assert "denied" in out
     assert "user comment:" in out
     assert "too destructive" in out
+
+def test_shell_confirm_formats_command_placeholder() -> None:
+    script = chr(10).join(["line1", "line2", "line3"])
+    reason = classify_danger(
+        "run_command",
+        {"command": ["bash", "-c", script]},
+    )
+    assert reason is not None
+    assert "$(command)" in reason
+    assert "line1" in reason and "line2" in reason and "line3" in reason
+    argv_line = next(ln for ln in reason.splitlines() if "argv:" in ln)
+    assert "line1" not in argv_line
+    lines = reason.splitlines()
+    idx = lines.index("  command:")
+    body = lines[idx + 1 :]
+    assert body
+    assert all(ln.startswith("  ") for ln in body)
+

tests/test_tools/test_file.py

@@ -19,8 +19,33 @@ def test_write_read_listdir_edit(workspace: object) -> None:
 def test_edit_replace_first_only(workspace: object) -> None:
     del workspace
     _ = call_sync(write_file, "t.txt", "aa aa")
-    _ = call_sync(edit_replace, "t.txt", "aa", "bb")
+    result = call_sync(edit_replace, "t.txt", "aa", "bb")
     assert call_sync(read_file, "t.txt") == "bb aa"
+    assert "1 of 2" in result or "1 of 2 matches" in result
+    assert "remain" in result
+
+
+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 "2 of 3" in result
+    assert "1 remain" in result
+
+
+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 "all 2 matches" in result
+
+
+def test_edit_replace_max_replaces_invalid(workspace: object) -> None:
+    del workspace
+    _ = call_sync(write_file, "t.txt", "aa")
+    assert "max_replaces" in call_sync(edit_replace, "t.txt", "aa", "bb", max_replaces=0)
 
 
 def test_edit_missing_old_string(workspace: object) -> None:

tests/test_tools/test_process.py

@@ -5,10 +5,12 @@ import sys
 from pathlib import Path
 
 from plyngent.tools.process import (
+    ask_into_pty,
     close_pty,
     open_pty,
     read_pty,
     run_command,
+    run_command_batch,
     write_pty,
     write_pty_keys,
 )
@@ -46,6 +48,85 @@ async def test_run_command_echo(workspace: object) -> None:
     assert "hi" in out
 
 
+async def test_run_command_batch_serial(workspace: object) -> None:
+    del workspace
+    out = await call_async(
+        run_command_batch,
+        [
+            {"command": _py("print('a')")},
+            {"command": _py("print('b')")},
+        ],
+    )
+    assert "steps=2" in out
+    assert "ran=2" in out
+    assert "stopped_early=false" in out
+    assert "--- step 0 ---" in out
+    assert "--- step 1 ---" in out
+    assert "a" in out and "b" in out
+
+
+async def test_run_command_batch_stop_on_error(workspace: object) -> None:
+    del workspace
+    out = await call_async(
+        run_command_batch,
+        [
+            {"command": _py("import sys; sys.exit(2)")},
+            {"command": _py("print('should-not-run')")},
+        ],
+        stop_on_error=True,
+    )
+    assert "ran=1" in out
+    assert "stopped_early=true" in out
+    assert "should-not-run" not in out
+
+
+async def test_run_command_batch_pipe_out(workspace: object) -> None:
+    del workspace
+    out = await call_async(
+        run_command_batch,
+        [
+            {
+                "command": _py("print('piped-payload', end='')"),
+                "pipe_out": True,
+            },
+            {
+                "command": _py("import sys; print(sys.stdin.read())"),
+            },
+        ],
+    )
+    assert "ran=2" in out
+    assert "piped-payload" in out
+    assert "pipe_out=true" in out
+
+
+async def test_run_command_batch_last_pipe_out_error(workspace: object) -> None:
+    del workspace
+    out = await call_async(
+        run_command_batch,
+        [{"command": _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:
+    del workspace
+    out = await call_async(
+        run_command_batch,
+        [
+            {
+                "command": _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()))")},
+        ],
+    )
+    assert "ran=2" in out
+    # Merged capture should include both (order depends on OS scheduling).
+    assert "out" in out and "err" in out
+
+
 async def test_run_command_denied(workspace: object) -> None:
     del workspace
     out = await call_async(run_command, ["rm", "-rf", "."])
@@ -168,6 +249,92 @@ def test_write_pty_unknown_session(workspace: object) -> None:
     assert "error" in call_sync(write_pty, 999_999, "x")
 
 
+async def test_ask_into_pty_writes_without_leaking_secret(workspace: object) -> None:
+    """Human answer goes to PTY only; tool result must not contain the secret."""
+    del workspace
+    from plyngent.prompting import temporary_backend
+    from tests.test_prompting import ScriptedBackend
+
+    secret = "super-secret-password-xyz"
+    backend = ScriptedBackend([], secrets=[secret])
+    try:
+        opened = call_sync(
+            open_pty,
+            _py(
+                "import sys\n"
+                "while True:\n"
+                "    line = sys.stdin.readline()\n"
+                "    if not line:\n"
+                "        break\n"
+                "    sys.stdout.write(line)\n"
+                "    sys.stdout.flush()\n"
+            ),
+        )
+        session_id = _session_id(opened)
+        with temporary_backend(backend):
+            result = await call_async(
+                ask_into_pty,
+                session_id,
+                "Enter password",
+                secret=True,
+            )
+        assert "error:" not in result
+        assert "wrote=true" in result
+        assert "secret=true" in result
+        assert "source=human" in result
+        assert secret not in result
+        # Length of secret must not appear as wrote=N either.
+        assert f"wrote={len(secret)}" not in result
+        assert f"wrote={len(secret) + 1}" not in result
+
+        echoed = await call_async(read_pty, session_id, timeout=2.0, until=secret)
+        assert secret in echoed
+        closed = await call_async(close_pty, session_id)
+        assert "session_id=" in closed
+    finally:
+        PtyManager.close_all()
+
+
+async def test_ask_into_pty_empty_cancels(workspace: object) -> None:
+    del workspace
+    from plyngent.prompting import temporary_backend
+    from tests.test_prompting import ScriptedBackend
+
+    backend = ScriptedBackend([], secrets=[""])
+    try:
+        opened = call_sync(open_pty, _py("import time; time.sleep(30)"))
+        session_id = _session_id(opened)
+        with temporary_backend(backend):
+            result = await call_async(ask_into_pty, session_id, "Pass?", secret=True)
+        assert "error:" in result
+        assert "empty" in result.lower() or "cancel" in result.lower()
+    finally:
+        PtyManager.close_all()
+
+
+async def test_ask_into_pty_nonsecret_line(workspace: object) -> None:
+    del workspace
+    from plyngent.prompting import temporary_backend
+    from tests.test_prompting import ScriptedBackend
+
+    backend = ScriptedBackend(["yes"])
+    try:
+        opened = call_sync(
+            open_pty,
+            _py("import sys\nline = sys.stdin.readline()\nsys.stdout.write(line)\nsys.stdout.flush()\n"),
+        )
+        session_id = _session_id(opened)
+        with temporary_backend(backend):
+            result = await call_async(ask_into_pty, session_id, "Continue?", secret=False)
+        assert "wrote=true" in result
+        assert "secret=false" in result
+        assert "yes" not in result  # answer not in tool result even for non-secret
+        text = await call_async(read_pty, session_id, timeout=2.0, until="yes")
+        assert "yes" in text
+    finally:
+        PtyManager.close_all()
+
+
 async def test_pty_exec_failure_surfaces(workspace: object) -> None:
     del workspace
     try:

tests/test_tools/test_workspace.py

@@ -43,12 +43,91 @@ def test_path_denylist(workspace: object) -> None:
 
 def test_command_denylist(workspace: object) -> None:
     del workspace
+    from plyngent.tools.workspace import (
+        clear_policy_allowed_commands,
+        set_policy_confirm_hook,
+    )
+
+    set_policy_confirm_hook(None)
+    clear_policy_allowed_commands()
     with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
         check_command_allowed(["rm", "-rf", "/"])
     check_command_allowed(["echo", "ok"])
     set_command_denylist(None)
 
 
+def test_command_denylist_policy_confirm_allow(workspace: object) -> None:
+    del workspace
+    from plyngent.tools.workspace import (
+        clear_policy_allowed_commands,
+        set_policy_confirm_hook,
+    )
+
+    calls: list[tuple[str, float]] = []
+
+    def hook(basename: str, argv: object, timeout: float) -> bool:
+        del argv
+        calls.append((basename, timeout))
+        return basename == "sudo"
+
+    set_policy_confirm_hook(hook)
+    clear_policy_allowed_commands()
+    try:
+        check_command_allowed(["sudo", "echo", "test"])
+        assert calls == [("sudo", 30.0)]
+        # Session grant: second call does not re-prompt.
+        check_command_allowed(["sudo", "id"])
+        assert len(calls) == 1
+    finally:
+        set_policy_confirm_hook(None)
+        clear_policy_allowed_commands()
+
+
+def test_command_denylist_policy_confirm_deny(workspace: object) -> None:
+    del workspace
+    from plyngent.tools.workspace import (
+        clear_policy_allowed_commands,
+        set_policy_confirm_hook,
+    )
+
+    set_policy_confirm_hook(lambda *_a: False)
+    clear_policy_allowed_commands()
+    try:
+        with pytest.raises(WorkspaceError, match="declined or timed out"):
+            check_command_allowed(["sudo", "echo", "x"])
+    finally:
+        set_policy_confirm_hook(None)
+        clear_policy_allowed_commands()
+
+
+def test_command_denylist_policy_confirm_timeout_value(workspace: object) -> None:
+    del workspace
+    from plyngent.tools.workspace import (
+        clear_policy_allowed_commands,
+        set_policy_confirm_hook,
+        set_policy_confirm_timeout,
+    )
+
+    seen: list[float] = []
+
+    def hook(basename: str, argv: object, timeout: float) -> bool:
+        del basename, argv
+        seen.append(timeout)
+        return False
+
+    set_policy_confirm_timeout(5.0)
+    set_policy_confirm_hook(hook)
+    clear_policy_allowed_commands()
+    try:
+        with pytest.raises(WorkspaceError):
+            check_command_allowed(["rm", "x"])
+        assert seen == [5.0]
+    finally:
+        set_policy_confirm_timeout(30.0)
+        set_policy_confirm_hook(None)
+        clear_policy_allowed_commands()
+
+
 def test_root_required() -> None:
     clear_workspace_root()
     with pytest.raises(WorkspaceError, match="not set"):