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.

CLAUDE.md

@@ -80,8 +80,8 @@ 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` (`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) / `close_pty` (POSIX: `pty`+`fork`; Windows: ConPTY via `pywinpty` env marker dep).
+- **`process`**: `run_command` (argv, no shell, timeout, optional stdin/env); `run_command_batch` (serial steps, `pipe_out` on provider, `mix_stderr`, `stop_on_error`); PTY `open_pty` / `read_pty` / `write_pty` (literal) / `write_pty_keys` (escapes) / `ask_into_pty` (human→PTY only; `secret` no-echo; answer never in tool result) / `close_pty` (POSIX: `pty`+`fork`; Windows: ConPTY via `pywinpty` env marker dep).
-- PTY: backend in `pty_backend.py`; structured status (`alive`/`exit_code`/`data`); `read_pty(..., until=)` (**CSI sanitized** so host TTY is never reset); keys via `write_pty_keys` only (`\xHH`, `ctrl+x`, `key=esc`); session limit/idle TTL/output budget; close terminate→kill.
+- PTY: backend in `pty_backend.py`; structured status (`alive`/`exit_code`/`data`); `read_pty(..., until=)` (**CSI sanitized** so host TTY is never reset); keys via `write_pty_keys` only (`\xHH`, `ctrl+x`, `key=esc`); secrets via `ask_into_pty` (not `write_pty` data); session limit/idle TTL/output budget; close terminate→kill.
 - CLI limit hooks: interactive confirm to raise tool-loop rounds, PTY session cap, or PTY output budget.
 - Destructive confirms: `classify_danger` + `ToolRegistry(on_confirm=…)`; CLI default deny; config `confirm_destructive` / `path_denylist`. Soft confirm also for shells/REPLs and `python|bash -c` one-liners (argv + code preview); deny may include a user comment for the model. Session YOLO: `/yolo on|off|once` and `--yes` (skip soft confirms; hard denylists unchanged; `once` expires after the next user turn).
 - **`vcs`**: read-only VCS tools (`vcs_kind` / `vcs_status` / `vcs_diff` / `vcs_log` / `vcs_branch`) via `VcsBackend` protocol; **git** implemented; detectors are pluggable for other systems.

README.md

@@ -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)
 

src/plyngent/prompting.py

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

src/plyngent/tools/__init__.py

@@ -16,6 +16,7 @@ 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

src/plyngent/tools/process/__init__.py

@@ -1,3 +1,4 @@
+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
@@ -15,5 +16,6 @@ PROCESS_TOOLS = [
     read_pty,
     write_pty,
     write_pty_keys,
+    ask_into_pty,
     close_pty,
 ]

src/plyngent/tools/process/ask_into_pty.py

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

tests/test_prompting.py

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

tests/test_tools/test_process.py

@@ -5,6 +5,7 @@ import sys
 from pathlib import Path
 
 from plyngent.tools.process import (
+    ask_into_pty,
     close_pty,
     open_pty,
     read_pty,
@@ -248,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: