core/process+cli: PTY hardening and config/log polish

Mark PTY master FD non-inheritable; run read_pty/close_pty via to_thread;
close_all PTY sessions on chat exit. Add --log-level; clearer invalid
TOML errors. Keep export/status free of provider secrets.
This commit is contained in:
2026-07-15 13:48:58 +08:00
parent 723fa85879
commit 0dc5c876ee
9 changed files with 136 additions and 36 deletions
+14 -3
View File
@@ -116,15 +116,26 @@ Basedpyright `recommended`. Ruff includes `ANN` (private return types `ANN202` i
- G2.5: Click slash registry (`cli/slash.py`) + `awaitlet` for sync Click / async memory; auto `/help`; completer from `slash.list_commands`
- G3: multiline `"""``"""` input (`cli/input_text.py`); `/edit` via `$EDITOR` (`edit_text_in_editor`)
- G4: `plyngent chat -p/--prompt` (+ non-TTY stdin); exit codes 0/1/2/3; `--yes` / non-interactive confirm deny; `--stream/--no-stream`, `--quiet`
- G5: PTY master FD non-inheritable; `read_pty`/`close_pty` via `to_thread`; `PtyManager.close_all()` on chat exit; `--log-level`; clearer invalid TOML errors; export/status stay secret-free
**Planned**
- **G5 — Hardening**: tool timeouts/defaults review, config error clarity, cancel edges, optional `--log-level`, no secrets in status/export
- **G6 — Docs**: real README, example TOML, CLAUDE/help accuracy
Milestone order: G5 → G6.
Milestone order: G6.
- **Phase H**: multi-tenant platform (`router/`, auth, sandboxed tools, web).
**PTY / process model (decision)**
Today `PtyManager` (`tools/process/pty_session.py`) is **in-process**: `pty.openpty` + `os.fork` → child `execvp`, parent holds master FD; tools are **sync** and run on the asyncio loop thread (blocking `select`/`read`/`waitpid`). Session registry is process-global. Safe enough for single-user CLI if the child path stays fork-then-exec only.
| Question | Answer |
|----------|--------|
| Separate PTY supervisor process so the main app is safer with threads/greenlets? | **Not for Phase G.** Defer to **Phase H** (sandbox / multi-tenant) or if we hit real FD-leak / freeze bugs. |
| Why not now? | Fork-then-exec is already the right shape; rewrite cost (IPC, lifecycle, tests) dwarfs single-user CLI risk. awaitlet greenlets are slash-only; PTY is not forked from a worker thread today. |
| What *does* belong in G5 | (1) mark master (and ideally app) FDs non-inheritable before/after fork; (2) avoid unbounded loop block — `to_thread` or `add_reader` for long `read_pty`/`close`; (3) CLI exit / `finally`: `PtyManager.close_all()`; (4) keep `os.fork` on the loop/main thread — do not move `open` to a thread pool without redesign. |
| Phase H | Optional PTY helper process (JSON/Unix socket), or subprocess+PTY with asyncio reaping; sandboxed tools, multi-session isolation. |
- **Phase H**: multi-tenant platform (`router/`, auth, sandboxed tools, web). Optional **out-of-process PTY host** if isolation is required.
## Commit messages
+39 -1
View File
@@ -9,6 +9,7 @@ import click
import msgspec
from platformdirs import user_data_path
from plyngent import config as config_mod
from plyngent.agent.loop import DEFAULT_MAX_ROUNDS
from plyngent.cli.editor import (
load_config_with_optional_edit,
@@ -34,7 +35,12 @@ _DEFAULT_DB_FILENAME = "chat.db"
def _load_config(config_path: Path | None) -> ConfigStore:
try:
return load_config_with_optional_edit(config_path)
except config_mod.ConfigFormatError as exc:
path = resolve_config_path(config_path)
msg = f"invalid config TOML ({path}): {exc}"
raise click.ClickException(msg) from exc
def _database_config(store: ConfigStore, *, quiet: bool = False) -> DatabaseConfig:
@@ -208,12 +214,44 @@ async def _run_chat(
return EXIT_OK
finally:
await memory.close()
from plyngent.tools.process.pty_session import PtyManager
PtyManager.close_all()
def _configure_logging(level: str) -> None:
import logging
name = level.upper()
numeric = getattr(logging, name, None)
if not isinstance(numeric, int):
msg = f"invalid --log-level {level!r}"
raise click.ClickException(msg)
logging.basicConfig(
level=numeric,
format="%(levelname)s %(name)s: %(message)s",
stream=sys.stderr,
force=True,
)
# Avoid accidental secret leakage via HTTP libraries at DEBUG.
if numeric <= logging.DEBUG:
logging.getLogger("niquests").setLevel(logging.INFO)
logging.getLogger("urllib3").setLevel(logging.INFO)
@click.group(context_settings={"help_option_names": ["-h", "--help"]})
@click.version_option(package_name="plyngent")
def main() -> None:
@click.option(
"--log-level",
default="WARNING",
show_default=True,
help="Logging level for stderr (DEBUG, INFO, WARNING, ERROR).",
)
@click.pass_context
def main(ctx: click.Context, log_level: str) -> None:
"""Plyngent — LLM chat and agent toolkit."""
_ = ctx
_configure_logging(log_level)
@main.command("chat")
+5 -1
View File
@@ -148,7 +148,11 @@ def prompt_edit_config(path: Path, *, reason: str | None = None) -> bool:
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 and EDITOR is set, offer to edit and reload.
Raises:
config_mod.ConfigFormatError: Invalid TOML (caller should surface path).
"""
path = resolve_config_path(config_path)
store = config_mod.load(path)
if store.providers:
+5 -1
View File
@@ -35,7 +35,11 @@ def session_export_payload(
updated_at: datetime | None,
messages: Sequence[AnyChatMessage],
) -> dict[str, object]:
"""Build a JSON-serializable dict for a session transcript."""
"""Build a JSON-serializable dict for a session transcript.
Never includes provider tokens or config secrets — only session metadata
and chat messages from the DB.
"""
return {
"session_id": sid,
"name": name,
+8 -3
View File
@@ -1,12 +1,17 @@
from __future__ import annotations
import asyncio
from plyngent.agent import tool
from .pty_session import PtyManager, format_close_result
@tool
def close_pty(session_id: int) -> str:
"""Close a PTY session (SIGTERM, then SIGKILL after a short grace period)."""
result = PtyManager.close(session_id)
async def close_pty(session_id: int) -> str:
"""Close a PTY session (SIGTERM, then SIGKILL after a short grace period).
Runs off the event loop so grace sleeps do not freeze the chat UI.
"""
result = await asyncio.to_thread(PtyManager.close, session_id)
return format_close_result(result)
+6 -1
View File
@@ -132,8 +132,11 @@ class PtyManager:
raise WorkspaceError(msg)
master_fd, slave_fd = pty.openpty()
# Parent must not pass master into the child (or future children).
with contextlib.suppress(OSError, AttributeError):
os.set_inheritable(master_fd, False) # noqa: FBT003 — stdlib API
pid = os.fork()
if pid == 0: # child
if pid == 0: # child — keep path fork-then-exec only (no Python after exec fail except _exit)
try:
os.close(master_fd)
_ = os.setsid()
@@ -150,6 +153,8 @@ class PtyManager:
os._exit(127)
os.close(slave_fd)
with contextlib.suppress(OSError, AttributeError):
os.set_inheritable(master_fd, False) # noqa: FBT003 — stdlib API
with cls._lock:
session_id = cls._next_id
cls._next_id += 1
+13 -4
View File
@@ -1,17 +1,22 @@
from __future__ import annotations
import asyncio
from plyngent.agent import tool
from plyngent.tools.workspace import WorkspaceError
from .pty_session import DEFAULT_PTY_READ_BYTES, PtyManager, format_read_result
# Cap per-call wait so a misbehaving tool arg cannot stall forever even off-loop.
_MAX_READ_TIMEOUT = 120.0
@tool
def read_pty(
async def read_pty(
session_id: int,
*,
max_bytes: int = DEFAULT_PTY_READ_BYTES,
timeout: float = 2.0,
timeout: float = 2.0, # noqa: ASYNC109 — PTY poll budget (seconds), not asyncio.timeout
until: str | None = None,
) -> str:
"""Read PTY output with status.
@@ -23,12 +28,16 @@ def read_pty(
With ``until``, polls until the substring appears, the process exits, the
deadline elapses, or the session output budget is exhausted.
Empty data with alive=true means nothing was ready (not necessarily EOF).
Blocking I/O runs in a worker thread so the asyncio loop stays responsive.
"""
wait = min(max(0.0, timeout), _MAX_READ_TIMEOUT)
try:
result = PtyManager.read(
result = await asyncio.to_thread(
PtyManager.read,
session_id,
max_bytes=max_bytes,
timeout=timeout,
timeout=wait,
until=until,
)
except WorkspaceError as exc:
+10
View File
@@ -21,3 +21,13 @@ def test_help() -> None:
result = runner.invoke(main, ["--help"])
assert result.exit_code == 0
assert "chat" in result.output
assert "--log-level" in result.output
def test_invalid_config_toml(tmp_path: Path) -> None:
bad = tmp_path / "bad.toml"
_ = bad.write_text("this is = not [ valid toml\n", encoding="utf-8")
runner = CliRunner()
result = runner.invoke(main, ["providers", "--config", str(bad)])
assert result.exit_code != 0
assert "invalid config" in result.output.lower() or "toml" in result.output.lower()
+35 -21
View File
@@ -1,6 +1,6 @@
from __future__ import annotations
import time
import asyncio
from pathlib import Path
from plyngent.tools.process import close_pty, open_pty, read_pty, run_command, write_pty
@@ -84,18 +84,18 @@ async def test_run_command_env(workspace: object) -> None:
assert "from-env" in out
def test_pty_open_read_close(workspace: object) -> None:
async def test_pty_open_read_close(workspace: object) -> None:
del workspace
try:
opened = call_sync(open_pty, ["sleep", "30"])
assert "session_id=" in opened
session_id = _session_id(opened)
data = call_sync(read_pty, session_id, timeout=0.05)
data = await call_async(read_pty, session_id, timeout=0.05)
assert "alive=" in data
assert "--- data ---" in data
closed = call_sync(close_pty, session_id)
closed = await call_async(close_pty, session_id)
assert _field(closed, "closed") == "true"
assert "error" in call_sync(read_pty, session_id)
assert "error" in await call_async(read_pty, session_id)
finally:
PtyManager.close_all()
set_command_denylist(None)
@@ -106,30 +106,30 @@ def test_pty_denied(workspace: object) -> None:
assert "denied" in call_sync(open_pty, ["sudo", "ls"])
def test_pty_echo_output(workspace: object) -> None:
async def test_pty_echo_output(workspace: object) -> None:
del workspace
try:
opened = call_sync(open_pty, ["/bin/echo", "hello-pty"])
session_id = _session_id(opened)
text = call_sync(read_pty, session_id, timeout=2.0, until="hello-pty")
text = await call_async(read_pty, session_id, timeout=2.0, until="hello-pty")
assert "hello-pty" in text
assert _field(text, "matched") == "true"
closed = call_sync(close_pty, session_id)
closed = await call_async(close_pty, session_id)
assert _field(closed, "closed") == "true"
finally:
PtyManager.close_all()
def test_write_pty(workspace: object) -> None:
async def test_write_pty(workspace: object) -> None:
del workspace
try:
opened = call_sync(open_pty, ["cat"])
session_id = _session_id(opened)
written = call_sync(write_pty, session_id, "pty-input\n")
assert "wrote=" in written
text = call_sync(read_pty, session_id, timeout=2.0, until="pty-input")
text = await call_async(read_pty, session_id, timeout=2.0, until="pty-input")
assert "pty-input" in text
_ = call_sync(close_pty, session_id)
_ = await call_async(close_pty, session_id)
finally:
PtyManager.close_all()
@@ -139,15 +139,15 @@ def test_write_pty_unknown_session(workspace: object) -> None:
assert "error" in call_sync(write_pty, 999_999, "x")
def test_pty_exec_failure_surfaces(workspace: object) -> None:
async def test_pty_exec_failure_surfaces(workspace: object) -> None:
del workspace
try:
opened = call_sync(open_pty, ["definitely-not-a-real-binary-xyz"])
session_id = _session_id(opened)
text = call_sync(read_pty, session_id, timeout=2.0)
text = await call_async(read_pty, session_id, timeout=2.0)
# marker and/or dead process with 127
assert "plyngent-pty-exec-failed" in text or _field(text, "alive") == "false"
closed = call_sync(close_pty, session_id)
closed = await call_async(close_pty, session_id)
# exit 127 is conventional for exec failure
exit_code = _field(closed, "exit_code")
assert exit_code in {"127", "-9", ""} or exit_code.startswith("-")
@@ -188,7 +188,7 @@ def test_pty_session_limit_continue(workspace: object) -> None:
PtyManager.set_limit_continue_hook(None)
def test_pty_output_budget(workspace: object) -> None:
async def test_pty_output_budget(workspace: object) -> None:
del workspace
previous = PtyManager.session_output_budget
try:
@@ -200,22 +200,22 @@ def test_pty_output_budget(workspace: object) -> None:
budget_hit = False
last = ""
for _ in range(20):
last = call_sync(read_pty, session_id, timeout=0.5, max_bytes=32)
last = await call_async(read_pty, session_id, timeout=0.5, max_bytes=32)
if _field(last, "budget_exhausted") == "true":
budget_hit = True
break
if _field(last, "alive") == "false":
break
time.sleep(0.05)
await asyncio.sleep(0.05)
assert budget_hit or "x" in last
_ = call_sync(close_pty, session_id)
_ = await call_async(close_pty, session_id)
finally:
PtyManager.close_all()
PtyManager.configure(session_output_budget=previous)
PtyManager.set_limit_continue_hook(None)
def test_pty_output_budget_is_per_session(workspace: object) -> None:
async def test_pty_output_budget_is_per_session(workspace: object) -> None:
del workspace
previous = PtyManager.session_output_budget
try:
@@ -230,14 +230,28 @@ def test_pty_output_budget_is_per_session(workspace: object) -> None:
before = session.output_budget
# Force budget exhaustion path by setting bytes_read high.
session.bytes_read = session.output_budget
_ = call_sync(read_pty, session_id, timeout=0.1)
_ = await call_async(read_pty, session_id, timeout=0.1)
session2 = PtyManager.get(session_id)
assert session2 is not None
assert session2.output_budget > before
# Raising is per-session; class default for new sessions stays put.
assert PtyManager.session_output_budget == class_budget
_ = call_sync(close_pty, session_id)
_ = await call_async(close_pty, session_id)
finally:
PtyManager.close_all()
PtyManager.configure(session_output_budget=previous)
PtyManager.set_limit_continue_hook(None)
def test_pty_master_not_inheritable(workspace: object) -> None:
del workspace
import os
try:
opened = call_sync(open_pty, ["sleep", "5"])
session_id = _session_id(opened)
session = PtyManager.get(session_id)
assert session is not None
assert os.get_inheritable(session.master_fd) is False
finally:
PtyManager.close_all()