20 Commits

Author SHA1 Message Date
NCBM bec14e76af 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.
2026-07-20 04:30:07 +08:00
NCBM 59017fd61b core/agent: synthetic_tool nag uses real todo_list render body
developer/user keep OPEN WORK prose; synthetic_tool injects forged
todo_list call with stack.render() only so the result matches a real
tool output.
2026-07-20 04:10:21 +08:00
NCBM 2e6cb3834a core/agent: drop system todo nag; emit events for synthetic_tool
Remove system strategy (legacy maps to developer). synthetic_tool still
forges the call but uses real stack text as the tool result, and now
yields ToolCall/Result events so the CLI flushes and does not glue nag
text to the next assistant stream.
2026-07-20 03:55:50 +08:00
NCBM f6d14ed549 core/agent: configurable todo nag strategies
Add [agent] todo_nag_strategy: developer (default), system, user,
synthetic_tool (forged todo_list pair), or none. Shared inject helper
used for turn-start and end-of-turn nags; CLI wires agent config.
2026-07-20 03:33:02 +08:00
NCBM a4ab11c9a2 core/agent: stronger open-work signals in todo stack prompts
Split tags: [TODO OPEN WORK] for pending/in_progress (undone work) vs
[TODO HYGIENE] for all-terminal leftover groups. End-of-turn review lists
open titles; tool docs stress open items as unfinished work.
2026-07-20 03:17:44 +08:00
NCBM 0d783b6d96 core/tools: claim PTY controlling TTY after setsid (sudo without -S)
After setsid(), ioctl(TIOCSCTTY) (or reopen slave) so the session leader
has a real controlling terminal. Live-tested: open_pty([sudo,echo,test])
shows a password prompt; ask_into_pty + read_pty yields test with exit 0.
2026-07-20 03:15:56 +08:00
NCBM d6616d0da3 core/tools: timed human allow for denylisted commands
Denylist hits ask the user (default deny, timeout) instead of only hard
rejecting when a policy hook is installed. Independent of YOLO soft-confirm;
session grants skip re-prompt for the same basename. CLI installs the hook
via install_cli_limit_hooks.
2026-07-20 03:02:34 +08:00
NCBM a677e06af4 core/tools: add ask_into_pty for human-local PTY input
Prompt the human (optional no-echo secret) and write only to the PTY.
Tool results never include the answer so secrets stay off external APIs.
Empty/Ctrl-C cancel without writing; submit defaults to trailing newline.
2026-07-20 02:47:19 +08:00
NCBM aadcd237db core/cli: soft-fail GET /models cancel/timeout on provider switch
Map CancelledError and TimeoutError from models() to RuntimeError so
merged_model_choices falls back to config/cache. Stub remote fetch in
/provider unit tests so they never hit the network.
2026-07-19 22:38:22 +08:00
NCBM 58ff646034 core/cli: config open via VISUAL/EDITOR or system default
Prefer $VISUAL then $EDITOR (blocking). Without either, fall back to
xdg-open / open / startfile (non-blocking) for config paths only.
/edit stays blocking-only so the temp buffer can be re-read.
2026-07-19 22:33:02 +08:00
NCBM d528f65c4c core/tools: edit_replace max_replaces and richer match messages
Default still replaces first match only. Report how many matches were
found, how many replaced, and remaining hits so agents can raise
max_replaces or narrow old_string.
2026-07-19 21:55:16 +08:00
NCBM 0ec0d542e8 core/cli: keep database url :memory:; default url None
Unset/empty url still maps to durable user-data chat.db.
Explicit :memory: is not rewritten; CLI warns when not quiet.
DatabaseConfig.url default is None; engine treats None as memory.
2026-07-19 21:38:21 +08:00
NCBM 182546d60e core/tools: narrow list[object] for batch command argv typing
basedpyright treated bare list elements as Unknown after isinstance(list);
cast to list[object] so prek basedpyright exits clean.
2026-07-19 20:39:16 +08:00
NCBM e1e4df0f3f core/agent: strengthen todo prompts when stack not empty
Inject turn-start [TODO REMINDER] for non-empty stacks; rewrite end-of-turn
[TODO OPEN] copy so open items read as unfinished work. needs_review now
always true while open items remain (even if todo_* ran this turn); terminal-only
stacks still need review if untouched. Update tool docs and tests.
2026-07-19 20:34:01 +08:00
NCBM a54fa534f3 test/tools: clean shell confirm format test 2026-07-19 13:05:43 +08:00
NCBM fa9f28687f test/tools: fix shell confirm format test syntax 2026-07-19 13:02:24 +08:00
NCBM 95f277eda7 core/tools: shell confirm argv uses $(command); full -c body indented 2026-07-19 13:01:06 +08:00
NCBM f13ea9f316 core/tools: soft-confirm file overwrite only, not partial edits 2026-07-19 03:05:01 +08:00
NCBM f51d11d46f core/tools/process: add run_command_batch with pipe_out and mix_stderr 2026-07-19 02:46:21 +08:00
NCBM 5b51de12e0 core/cli: skip blocking GET /models on chat ready
Only fetch remote catalog when interactive model pick needs it; otherwise
use config/--model/session and warm Tab cache in the background. Cap list
fetch at 5s so a dead API cannot hang startup.
2026-07-19 02:08:58 +08:00
41 changed files with 2281 additions and 352 deletions
+8 -8
View File
@@ -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.
- **`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).
- **`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).
- 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.
- **`workspace`**: path resolve under primary root **or** temporary allowlist roots; path substring denylist; command basename denylist (CLI: timed human allow override, default deny on timeout; **not** skipped by YOLO); clearer deny messages.
- **`file`**: `read_file` (`with_lineno`), `write_file`, `listdir`, `tree` (VCS + default noise dirs + optional `skip_dirs` / denylist walk), `glob_paths`, `grep_files` (regex, skip VCS/binary), `edit_replace` (`max_replaces`, reports remaining matches), `edit_lineno` (1-based range), `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); `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`); 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`.
- 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.
- **`plyngent chat`**: provider/model (flags or interactive; Tab via readline in `prompting`); sessions store `provider_name`/`model` and restore on resume; SQLite via `[database]` (file DB under user data when url unset; explicit `url = ":memory:"` kept + warn); workspace-bound; resume latest for cwd/`--workspace` by default (`--new` / `--session`). One-shot: `-p/--prompt` and non-TTY stdin; exit codes 0/1/2/3; `--yes` (YOLO on), `--stream/--no-stream`, `--quiet`. Root `--log-level`.
- Slash: Click group in `cli/slash.py` + `awaitlet` for async work; Tab completer from registry + ParamType `shell_complete`. Multiline `"""``"""`; `/edit` via `$VISUAL`/`$EDITOR` (blocking only). `/yolo on|off|once` for soft destructive confirms. `/model --persist` / `/models --persist` write model catalog entries into TOML. `/todos` for human show/push/pop/clear of the todo stack.
- 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
+6 -5
View File
@@ -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)
+8 -2
View File
@@ -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
# durable file under the platform user data dir (e.g. ~/.local/share/plyngent/chat.db).
# Optional. If omitted (or SQLite url unset/empty), the CLI uses a durable file under
# 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 = ""
Generated
+22 -22
View File
@@ -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.0.tar.gz", hash = "sha256:31e761a6a0ca04faf7353ea759bdba55652be214725111e5aac52dfa29d4bef7"},
{file = "platformdirs-4.10.1-py3-none-any.whl", hash = "sha256:0e4eff26be2d75293977f7cddc153fd9b8eaa7fb0c7b64ffe4076cb443117443"},
{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.21-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:0f212c5d7d54c01bbfe6dcab02b724a39300f3e34ed7acbe995ccb320a2c58bd"},
{file = "ruff-0.15.21-py3-none-macosx_11_0_arm64.whl", hash = "sha256:e6312e41bc96791299614995ea3a977c5857c3b5662b1ecef6755b02b87cb646"},
{file = "ruff-0.15.21-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:01d65b4831c6b2a4ba8ee6faa84049d44d982b7a706e622c4094c509e51673be"},
{file = "ruff-0.15.21-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:2c5a913a589120ce67933d5d05fd6ddbcc2481c6a054980ee767f7414c72b4fd"},
{file = "ruff-0.15.21-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:5ef04b681d02ad4dc9620f00f83ac5c22f652d0e9a9cfe431d219b16ad5ccc41"},
{file = "ruff-0.15.21-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:16d090c0740916594157e75b80d666eab8e78083b39b3b0e1d698f4670a17b86"},
{file = "ruff-0.15.21-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:3a10e74757dd65004d779b73e2f3c5210156d9980b41224d50d2ebcf1db51e67"},
{file = "ruff-0.15.21-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:bab0905d2f29e0d9fbc3c373ed23db0095edaa3f71f1f4f519ec15134d9e85c8"},
{file = "ruff-0.15.21-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:00eca240af5789fec6fe7df74c088cc1f9644ed83027113468efba7c92b94075"},
{file = "ruff-0.15.21-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:262ab31557a75141325e32d3357f3597645a7f084e732b6b054dde428ecd9341"},
{file = "ruff-0.15.21-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:659c4e7a4212f83306045ec7c5e5a356d16d9a6ef4ae0c7a4d872914fc655d9d"},
{file = "ruff-0.15.21-py3-none-musllinux_1_2_i686.whl", hash = "sha256:9e866eab611a5f959d36df2d10e446973a3610bc42b0c15b31dc27977d59c233"},
{file = "ruff-0.15.21-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:e89bc93c0d3803ba870b55c29671bad9dc6d94bb1eb181b056b52eb05b52854f"},
{file = "ruff-0.15.21-py3-none-win32.whl", hash = "sha256:01f8d5be84823c172b389e123174f781f9daf86d6c58719d603f941932195cdd"},
{file = "ruff-0.15.21-py3-none-win_amd64.whl", hash = "sha256:d4b8d9a2f0f12b816b50447f6eccb9f4bb01a6b82c86b50fb3b5354b458dc6d3"},
{file = "ruff-0.15.21-py3-none-win_arm64.whl", hash = "sha256:6e83115d4b9377c1cbc13abf0e051f069fab0ef815ea0504a8a008cee24dd0a8"},
{file = "ruff-0.15.21.tar.gz", hash = "sha256:d0cfc841c572283c36548f82664a54ce6565567f1b0d5b4cf2caac693d8b7500"},
{file = "ruff-0.15.22-py3-none-linux_armv6l.whl", hash = "sha256:44423e73493737f5e7c5b41d475483898ff37afcdae38bc3da5085e29af1c2d8"},
{file = "ruff-0.15.22-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:b82c6482946e9eda7ff2e091d25b8bad3f718684e1916d41bd56873cee05b697"},
{file = "ruff-0.15.22-py3-none-macosx_11_0_arm64.whl", hash = "sha256:11c1c715af53a09f714e011106bffc419751ec8232fcb5da42173284ea3fec6f"},
{file = "ruff-0.15.22-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:742a29cf29bddb7c8327895d6a10e0e6c5b38a96dd407af9b5d0857f809c0576"},
{file = "ruff-0.15.22-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:72af58b951b0ae395935ae79763dc349bc0eb706319d28f7a33ad2cfb3cfc178"},
{file = "ruff-0.15.22-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:62d425005c1835eb24e2ee4161cb90e8db263415f4a71c8c72c33abaa6c0c224"},
{file = "ruff-0.15.22-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:e8b9b3f8779a4f08c969defc3c8c35abffaa757e601ed5ae66d6d1db6519969a"},
{file = "ruff-0.15.22-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:1e0dd1b2e4d3d585f897a0d137cbf4eaf6223bef4e8ce34d6bb12556c5f9249e"},
{file = "ruff-0.15.22-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:365523eb91d9224e1bcb03b022fbf0facb8f9e23792a2c53d9d4b3924bdbdebb"},
{file = "ruff-0.15.22-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:fabfd168afdf29fee5be98b831efa9683c94d7c5a3b58b9ce5a2e38444589a74"},
{file = "ruff-0.15.22-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:225dbf095a87f1d9f90f5fd7924d2613ee452a75a4308c63a8f50f761787aa7c"},
{file = "ruff-0.15.22-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:1877d63b9d24ed278744f1523fd11b85540566d54641f97c566d7d9dc5ca5296"},
{file = "ruff-0.15.22-py3-none-musllinux_1_2_i686.whl", hash = "sha256:a1606c510bd7215680d32efab38965f7cdec3ef69f5170a3f4791404ffdd5262"},
{file = "ruff-0.15.22-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:630479b18625f5ffc373f77603a22a9f8ac0acd7ff0501178b5db28ec71e9c64"},
{file = "ruff-0.15.22-py3-none-win32.whl", hash = "sha256:e5ba0e4a13fd14abbed2a77b517a3911290c6c6c59ef67784328d1668fab76cf"},
{file = "ruff-0.15.22-py3-none-win_amd64.whl", hash = "sha256:9be63ba1eb936acd2d1342fb8337c356353706fce233b2a15a09a97037e6acde"},
{file = "ruff-0.15.22-py3-none-win_arm64.whl", hash = "sha256:e1168075b72158510839f250027659cdd78476f40507dd517892304c41318661"},
{file = "ruff-0.15.22.tar.gz", hash = "sha256:3f15175b1fb580126f58285a5dae6b2ea89000136d980c64499211f116b54809"},
]
[[package]]
+1 -1
View File
@@ -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" },
+22
View File
@@ -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
+16 -7
View File
@@ -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
no ``todo_*`` tool use this turn, injects a review user message and continues
once (so the model must check or update the stack).
When *todo_stack* is set and still needs review after a natural stop
(open items, or non-empty stack untouched this turn), injects a review nag
(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,8 +375,15 @@ 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.
messages.append(DeveloperChatMessage(content=todo_stack.review_prompt()))
injected, nag_events = inject_todo_nag_for_stack_with_events(
messages,
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(
+183
View File
@@ -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,
)
+79 -5
View File
@@ -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) "
f"(not empty). You did not call todo_* this turn.",
headline,
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)
+40 -8
View File
@@ -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.
if raw.get("url") in {None, "", ":memory:"} and raw.get("implementation", "sqlite") == "sqlite":
url = raw.get("url")
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,19 +254,25 @@ 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
if needs_remote_models_for_selection(
provider,
preferred_model=preferred_model,
interactive=interactive,
):
client = create_client(provider)
try:
if client_supports_models(client):
remote_ids = await fetch_remote_model_ids(client)
except RuntimeError, TypeError, OSError, ValueError:
except RuntimeError, TypeError, OSError, ValueError, TimeoutError:
remote_ids = None
choices = model_choices_for_provider(provider, remote_ids=remote_ids)
model_id = select_model(
@@ -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,9 +529,20 @@ 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)
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}")
+128 -39
View File
@@ -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."""
value = os.environ.get("EDITOR", "").strip()
return value or None
"""Return ``$VISUAL`` or ``$EDITOR``, or ``None`` if both unset/empty.
``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(
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)
def _run_blocking_editor(editor_cmd: str, path: Path) -> None:
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:
"""Edit ``initial`` in ``$EDITOR``; return text or ``None`` if empty/cancelled.
def _open_with_system_default(path: Path) -> None:
"""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:
"""If ``EDITOR`` is set, ask whether to edit ``path``. Returns True if opened."""
if get_editor() is None:
return False
def prompt_edit_config(path: Path, *, reason: str | None = None) -> OpenOutcome | None:
"""Ask whether to edit ``path``. Returns open outcome, or ``None`` if skipped.
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
open_in_editor(path)
return True
return None
return open_in_editor(path, allow_system_open=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
+88 -3
View File
@@ -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)
+42 -2
View File
@@ -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]:
"""Call ``client.models()``; raise if missing or the call fails."""
async def fetch_remote_model_ids(
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):
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,
*,
+1 -1
View File
@@ -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 []
+14 -7
View File
@@ -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
re-read; current provider/model are kept when still valid.
Same file as ``plyngent config edit``. After a **blocking** editor exits,
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:
+31 -3
View File
@@ -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)
+11 -2
View File
@@ -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 = ""
+1 -1
View File
@@ -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
+35
View File
@@ -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],
+8
View File
@@ -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]
+98 -35
View File
@@ -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] + ""
return f"{via}: {base} -c (review code before allow)\nargv:\n {display}\n-c code:\n {preview}"
body = _indent_block(code, prefix=" ")
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:
return f"{via}: shell/runtime {base!r} (review before allow)\nargv:\n {display}"
if base in _SHELL_BASENAMES and "-c" not in argv[1:]:
return f"{via}: shell/runtime {base!r} without -c (review before allow)\n argv: {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}"
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:
# Path policy will fail later; no soft-confirm without a resolved target.
return None
return f"edit (replace) in {path!r}"
def _edit_lineno_reason(args: Mapping[str, object]) -> str | None:
path = args.get("path")
if not isinstance(path, str) or not path:
if target.is_file():
return f"overwrite existing file {path!r} ({target})"
# New file or directory path: no total-overwrite risk for soft-confirm.
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
+42 -5
View File
@@ -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:
"""Replace the first occurrence of ``old_string`` with ``new_string`` in a file."""
def edit_replace(path: str, old_string: str, new_string: str, max_replaces: int = 1) -> str:
"""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)
+12 -1
View File
@@ -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,
]
@@ -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)
+143
View File
@@ -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,
)
+30
View File
@@ -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)
+23 -153
View File
@@ -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 (
WorkspaceError,
check_command_allowed,
get_workspace_root,
resolve_path,
from plyngent.tools.workspace import WorkspaceError
from .command_exec import (
DEFAULT_TIMEOUT_SECONDS,
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.
``cwd`` is relative to or under the workspace root. Optional ``stdin`` is
written to the process stdin. Optional ``env`` overlays process environment
variables (merged with the current environment). Output is truncated.
On timeout the process is killed and any partial stdout/stderr is still
returned with ``timed_out=true``.
"""
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(
"""Run ``command`` (argv, no shell) under the workspace; capture stdout/stderr."""
try:
result = await execute_argv(
command,
workdir=str(workdir),
cwd=cwd,
env=env,
stdin=stdin,
timeout_seconds=timeout_seconds,
stdin_data=stdin_data,
env=_merge_env(env),
mix_stderr=False,
)
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,
except WorkspaceError as exc:
return f"error: {exc}"
return format_command_result(
returncode=result.exit_code,
workdir_display=result.cwd_display,
command=result.command,
stdout=result.stdout,
stderr=result.stderr,
timed_out=result.timed_out,
)
@@ -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}"
+14 -3
View File
@@ -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():
+77 -2
View File
@@ -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:
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)
+196 -3
View File
@@ -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 not any(isinstance(m, UserChatMessage) 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 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)
+95 -3
View File
@@ -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:
+86 -12
View File
@@ -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"):
open_in_editor(tmp_path / "x.toml", editor=None)
monkeypatch.delenv("VISUAL", raising=False)
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:
del editor
def fake_open(
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()
+12
View File
@@ -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
+37
View File
@@ -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
+17 -1
View File
@@ -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"
+23 -1
View File
@@ -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):
+57 -5
View File
@@ -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 "edit" in (classify_danger("edit_replace", {"path": "x.txt"}) or "")
assert "lines" in (classify_danger("edit_lineno", {"path": "x.txt", "start_line": 1, "end_line": 2}) or "")
assert reason is not None and "overwrite" in reason
(tmp_path / "dst.txt").write_text("d", encoding="utf-8")
(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)
+26 -1
View File
@@ -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:
+167
View File
@@ -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:
+79
View File
@@ -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"):