core/cli: one-shot chat with -p/stdin and exit codes

Add plyngent chat -p/--prompt (and non-TTY stdin), --stream/--no-stream,
--yes, --quiet. Non-interactive mode uses NonInteractiveBackend and
disables limit prompts. Exit codes: 0 ok, 1 config, 2 cancelled, 3 failed.

CLAUDE.md

@@ -115,14 +115,14 @@ Basedpyright `recommended`. Ruff includes `ANN` (private return types `ANN202` i
   - G2: `/rename`, `/delete` (confirm), `/export md|json`
   - 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`
 
   **Planned**
 
-  - **G4 — One-shot**: `plyngent chat -p/--prompt` (and stdin non-TTY); exit codes; non-interactive safety (`--yes` / deny destructive)
   - **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: G4 → G5 → G6.
+  Milestone order: G5 → G6.
 
 - **Phase H**: multi-tenant platform (`router/`, auth, sandboxed tools, web).
 

src/plyngent/cli/app.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import sys
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -14,12 +15,15 @@ from plyngent.cli.editor import (
     open_in_editor,
     resolve_config_path,
 )
+from plyngent.cli.exit_codes import EXIT_CANCELLED, EXIT_OK, EXIT_TURN_FAILED
 from plyngent.cli.limits import install_cli_limit_hooks
 from plyngent.cli.repl import run_repl
+from plyngent.cli.retry import run_user_text_with_retries
 from plyngent.cli.selection import select_model, select_provider
 from plyngent.cli.state import ReplState
 from plyngent.config.models import DatabaseConfig
 from plyngent.memory import MemoryStore
+from plyngent.prompting import NonInteractiveBackend, configure_prompting
 from plyngent.runtime import ProviderNotSupportedError, create_client
 from plyngent.tools import set_workspace_root
 
@@ -33,16 +37,103 @@ def _load_config(config_path: Path | None) -> ConfigStore:
     return load_config_with_optional_edit(config_path)
 
 
-def _database_config(store: ConfigStore) -> DatabaseConfig:
+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":
         db_path = user_data_path("plyngent", ensure_exists=True) / _DEFAULT_DB_FILENAME
         raw = {**raw, "implementation": "sqlite", "url": str(db_path)}
-        click.secho(f"using database: {db_path}", fg="bright_black")
+        if not quiet:
+            click.secho(f"using database: {db_path}", fg="bright_black", err=True)
     return msgspec.convert(raw, DatabaseConfig)
 
 
+def _read_prompt_text(prompt: str | None, *, stdin_isatty: bool) -> str | None:
+    """Resolve one-shot prompt from ``-p`` and/or non-TTY stdin."""
+    chunks: list[str] = []
+    if prompt is not None and prompt.strip():
+        chunks.append(prompt)
+    if not stdin_isatty:
+        data = sys.stdin.read()
+        if data.strip():
+            chunks.append(data.rstrip("\n"))
+    if not chunks:
+        return None
+    text = "\n".join(chunks).strip()
+    return text or None
+
+
+def _setup_workspace_and_hooks(
+    store: ConfigStore,
+    workspace: Path,
+    *,
+    interactive: bool,
+) -> None:
+    _ = set_workspace_root(workspace)
+    from plyngent.tools import set_path_denylist
+
+    set_path_denylist(store.agent_config.path_denylist or None)
+    if interactive:
+        install_cli_limit_hooks()
+    else:
+        from plyngent.tools.process.pty_session import PtyManager
+
+        PtyManager.set_limit_continue_hook(None)
+
+
+async def _bind_session(
+    state: ReplState,
+    *,
+    session_id: int | None,
+    new_session: bool,
+    oneshot: bool,
+    quiet: bool,
+) -> None:
+    if session_id is not None:
+        try:
+            await state.resume_session(session_id)
+        except ValueError as exc:
+            raise click.ClickException(str(exc)) from exc
+        if not quiet and not oneshot:
+            click.echo(
+                f"resumed session {session_id} ({len(state.agent.messages)} messages) workspace={state.workspace}",
+                err=True,
+            )
+        return
+    if new_session or oneshot:
+        await state.new_session()
+        if not quiet and not oneshot:
+            click.echo(
+                f"new session {state.session_id} (workspace={state.workspace})",
+                err=True,
+            )
+        return
+    mode = await state.resume_latest_or_new()
+    if quiet:
+        return
+    if mode == "resume":
+        click.echo(
+            f"resumed latest session {state.session_id} for this workspace "
+            f"({len(state.agent.messages)} messages); use --new for a fresh chat",
+            err=True,
+        )
+    else:
+        click.echo(
+            f"new session {state.session_id} (workspace={state.workspace})",
+            err=True,
+        )
+
+
+async def _run_oneshot(state: ReplState, prompt_text: str) -> int:
+    try:
+        ok = await run_user_text_with_retries(state.agent, prompt_text, delays=())
+    except asyncio.CancelledError:
+        return EXIT_CANCELLED
+    except KeyboardInterrupt:
+        return EXIT_CANCELLED
+    return EXIT_OK if ok else EXIT_TURN_FAILED
+
+
 async def _run_chat(
     *,
     config_path: Path | None,
@@ -53,25 +144,37 @@ async def _run_chat(
     session_id: int | None,
     max_rounds: int,
     new_session: bool,
-) -> None:
+    prompt_text: str | None,
+    stream: bool,
+    yes: bool,
+    quiet: bool,
+) -> int:
+    oneshot = prompt_text is not None
+    interactive = not oneshot and sys.stdin.isatty() and sys.stdout.isatty()
+
+    if oneshot:
+        configure_prompting(backend=NonInteractiveBackend())
+
     store = _load_config(config_path)
-    if store.bad_providers:
+    if store.bad_providers and not quiet:
         names = ", ".join(sorted(store.bad_providers.keys()))
-        click.secho(f"warning: ignored bad providers: {names}", fg="yellow")
+        click.secho(f"warning: ignored bad providers: {names}", fg="yellow", err=True)
 
     try:
-        pname, provider = select_provider(store.providers, preferred=provider_name)
+        pname, provider = select_provider(
-        model_id = select_model(provider, preferred=model)
+            store.providers,
-        _ = create_client(provider)  # fail early if unsupported
+            preferred=provider_name,
+            interactive=interactive,
+        )
+        model_id = select_model(provider, preferred=model, interactive=interactive)
+        _ = create_client(provider)
     except ProviderNotSupportedError as exc:
         raise click.ClickException(str(exc)) from exc
 
-    _ = set_workspace_root(workspace)
+    _setup_workspace_and_hooks(store, workspace, interactive=interactive)
-    from plyngent.tools import set_path_denylist
+    confirm_destructive: bool | None = False if yes else None
 
-    set_path_denylist(store.agent_config.path_denylist or None)
+    memory = await MemoryStore.open(_database_config(store, quiet=quiet or oneshot))
-    install_cli_limit_hooks()
-    memory = await MemoryStore.open(_database_config(store))
     try:
         state = ReplState(
             config=store,
@@ -82,29 +185,27 @@ async def _run_chat(
             model=model_id,
             tools_enabled=tools,
             max_rounds=max_rounds,
+            stream_enabled=stream,
+            interactive_limits=interactive,
+            confirm_destructive=confirm_destructive,
         )
-        click.secho(f"workspace: {state.workspace}", fg="bright_black")
+        if not quiet and not oneshot:
-        if session_id is not None:
+            click.secho(f"workspace: {state.workspace}", fg="bright_black", err=True)
-            try:
+
-                await state.resume_session(session_id)
+        await _bind_session(
-            except ValueError as exc:
+            state,
-                raise click.ClickException(str(exc)) from exc
+            session_id=session_id,
-            click.echo(
+            new_session=new_session,
-                f"resumed session {session_id} ({len(state.agent.messages)} messages) workspace={state.workspace}"
+            oneshot=oneshot,
+            quiet=quiet,
         )
-        elif new_session:
+
-            await state.new_session()
+        if oneshot:
-            click.echo(f"new session {state.session_id} (workspace={state.workspace})")
+            assert prompt_text is not None
-        else:
+            return await _run_oneshot(state, prompt_text)
-            mode = await state.resume_latest_or_new()
+
-            if mode == "resume":
-                click.echo(
-                    f"resumed latest session {state.session_id} for this workspace "
-                    f"({len(state.agent.messages)} messages); use --new for a fresh chat"
-                )
-            else:
-                click.echo(f"new session {state.session_id} (workspace={state.workspace})")
         await run_repl(state)
+        return EXIT_OK
     finally:
         await memory.close()
 
@@ -147,6 +248,31 @@ def main() -> None:
     show_default=True,
     help="Max tool-loop rounds per user turn.",
 )
+@click.option(
+    "-p",
+    "--prompt",
+    "prompt",
+    default=None,
+    help="One-shot user message (non-interactive). Also reads stdin when not a TTY.",
+)
+@click.option(
+    "--stream/--no-stream",
+    default=True,
+    show_default=True,
+    help="Stream model output (one-shot and REPL default).",
+)
+@click.option(
+    "--yes",
+    is_flag=True,
+    default=False,
+    help="Allow destructive tools without confirm (one-shot / non-interactive).",
+)
+@click.option(
+    "--quiet",
+    is_flag=True,
+    default=False,
+    help="Less status noise on stderr.",
+)
 def chat_cmd(
     config_path: Path | None,
     provider_name: str | None,
@@ -156,16 +282,30 @@ def chat_cmd(
     session_id: int | None,
     new_session: bool,  # noqa: FBT001
     max_rounds: int,
+    prompt: str | None,
+    stream: bool,  # noqa: FBT001
+    yes: bool,  # noqa: FBT001
+    quiet: bool,  # noqa: FBT001
 ) -> None:
-    """Interactive chat REPL with optional tools and session memory."""
+    """Interactive chat REPL, or one-shot with ``-p`` / stdin.
+
+    Exit codes (one-shot): 0 ok, 1 config/usage, 2 cancelled, 3 turn failed.
+    """
     if max_rounds < 1:
         msg = "--max-rounds must be >= 1"
         raise click.ClickException(msg)
     if session_id is not None and new_session:
         msg = "use either --session or --new, not both"
         raise click.ClickException(msg)
+
+    prompt_text = _read_prompt_text(prompt, stdin_isatty=sys.stdin.isatty())
+    if prompt is None and prompt_text is None and not sys.stdin.isatty():
+        # Non-TTY with empty stdin and no -p: still require an explicit prompt.
+        msg = "no prompt: pass -p/--prompt or pipe text on stdin"
+        raise click.ClickException(msg)
+
     root = workspace if workspace is not None else Path.cwd()
-    asyncio.run(
+    code = asyncio.run(
         _run_chat(
             config_path=config_path,
             provider_name=provider_name,
@@ -175,8 +315,14 @@ def chat_cmd(
             session_id=session_id,
             max_rounds=max_rounds,
             new_session=new_session,
+            prompt_text=prompt_text,
+            stream=stream,
+            yes=yes,
+            quiet=quiet,
         )
     )
+    if code != EXIT_OK:
+        raise SystemExit(code)
 
 
 @main.command("providers")

src/plyngent/cli/exit_codes.py

@@ -0,0 +1,8 @@
+"""Process exit codes for ``plyngent chat`` (one-shot and fatal paths)."""
+
+from __future__ import annotations
+
+EXIT_OK = 0
+EXIT_ERROR = 1  # config / usage / fatal
+EXIT_CANCELLED = 2
+EXIT_TURN_FAILED = 3  # API / retry exhausted / incomplete turn

src/plyngent/cli/retry.py

@@ -185,9 +185,14 @@ async def run_turn_with_retries(
             return True
 
 
-async def run_user_text_with_retries(agent: ChatAgent, text: str) -> bool:
+async def run_user_text_with_retries(
-    """Send a new user message with auto-retry."""
+    agent: ChatAgent,
-    return await run_turn_with_retries(agent, starter=lambda: agent.run(text))
+    text: str,
+    *,
+    delays: tuple[float, ...] = DEFAULT_RETRY_DELAYS_SECONDS,
+) -> bool:
+    """Send a new user message with auto-retry (empty ``delays`` = no auto-retry)."""
+    return await run_turn_with_retries(agent, starter=lambda: agent.run(text), delays=delays)
 
 
 async def retry_pending_with_retries(agent: ChatAgent) -> bool:

src/plyngent/cli/selection.py

@@ -14,6 +14,7 @@ def select_provider(
     providers: Mapping[str, Provider],
     *,
     preferred: str | None = None,
+    interactive: bool = True,
 ) -> tuple[str, Provider]:
     """Pick a provider by name or interactive prompt."""
     if not providers:
@@ -29,9 +30,13 @@ def select_provider(
 
     if len(names) == 1:
         name = names[0]
-        click.echo(f"Using provider: {name}")
+        click.echo(f"Using provider: {name}", err=True)
         return name, providers[name]
 
+    if not interactive:
+        msg = f"multiple providers; pass --provider ({', '.join(names)})"
+        raise click.ClickException(msg)
+
     click.echo("Available providers:")
     for index, name in enumerate(names, start=1):
         preset = type(providers[name]).__struct_config__.tag
@@ -44,6 +49,7 @@ def select_model(
     provider: Provider,
     *,
     preferred: str | None = None,
+    interactive: bool = True,
 ) -> str:
     """Pick a model id from provider.models or free-form prompt."""
     model_names = sorted(provider.models.keys())
@@ -55,9 +61,16 @@ def select_model(
 
     if len(model_names) == 1:
         model = model_names[0]
-        click.echo(f"Using model: {model}")
+        click.echo(f"Using model: {model}", err=True)
         return model
 
+    if not interactive:
+        if model_names:
+            msg = f"multiple models; pass --model ({', '.join(model_names)})"
+        else:
+            msg = "no models listed; pass --model"
+        raise click.ClickException(msg)
+
     if model_names:
         click.echo("Available models:")
         for index, name in enumerate(model_names, start=1):

src/plyngent/cli/state.py

@@ -31,6 +31,10 @@ class ReplState:
     max_rounds: int = DEFAULT_MAX_ROUNDS
     stream_enabled: bool = True
     verbose: bool = False
+    # One-shot / scripts: never prompt to raise tool-loop limits.
+    interactive_limits: bool = True
+    # When False, skip destructive-tool confirms (e.g. --yes).
+    confirm_destructive: bool | None = None
     # Set by /edit; REPL sends as the next user turn then clears.
     pending_user_text: str | None = None
     client: ChatClient = field(init=False)
@@ -56,14 +60,18 @@ class ReplState:
             raise RuntimeError(msg)
         return key
 
+    def _confirm_destructive(self) -> bool:
+        if self.confirm_destructive is not None:
+            return self.confirm_destructive
+        return self.config.agent_config.confirm_destructive
+
     def _tool_registry(self) -> ToolRegistry | None:
         if not self.tools_enabled:
             return None
         from plyngent.cli.limits import prompt_confirm_tool_async
         from plyngent.tools.danger import classify_danger
 
-        agent_cfg = self.config.agent_config
+        if self._confirm_destructive():
-        if agent_cfg.confirm_destructive:
             return ToolRegistry(
                 list(DEFAULT_TOOLS),
                 danger=classify_danger,
@@ -76,6 +84,7 @@ class ReplState:
 
         agent_cfg = self.config.agent_config
         system_prompt = agent_cfg.system_prompt or None
+        on_limit = prompt_continue_limit_async if self.interactive_limits else None
         return ChatAgent(
             self.client,
             model=self.model,
@@ -83,7 +92,7 @@ class ReplState:
             memory=self.memory,
             session_id=self.session_id,
             max_rounds=self.max_rounds,
-            on_limit=prompt_continue_limit_async,
+            on_limit=on_limit,
             stream=self.stream_enabled,
             system_prompt=system_prompt,
             max_tool_result_chars=agent_cfg.max_tool_result_chars,

tests/test_cli/test_oneshot.py

@@ -0,0 +1,159 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import TYPE_CHECKING, Literal, overload
+
+import pytest
+from click.testing import CliRunner
+
+from plyngent.cli.app import _read_prompt_text, main
+from plyngent.cli.exit_codes import EXIT_OK
+from plyngent.lmproto.openai_compatible.model import (
+    AssistantChatMessage,
+    ChatCompletionChoice,
+    ChatCompletionChunk,
+    ChatCompletionResponse,
+    ChatCompletionsParam,
+)
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+
+def test_read_prompt_flag_only() -> None:
+    assert _read_prompt_text("hello", stdin_isatty=True) == "hello"
+    assert _read_prompt_text("  ", stdin_isatty=True) is None
+    assert _read_prompt_text(None, stdin_isatty=True) is None
+
+
+def test_read_prompt_stdin(monkeypatch: pytest.MonkeyPatch) -> None:
+    class FakeStdin:
+        def read(self) -> str:
+            return "from stdin\n"
+
+    monkeypatch.setattr("plyngent.cli.app.sys.stdin", FakeStdin())
+    assert _read_prompt_text(None, stdin_isatty=False) == "from stdin"
+    assert _read_prompt_text("flag", stdin_isatty=False) == "flag\nfrom stdin"
+
+
+def test_chat_oneshot_requires_provider_flags(tmp_path: Path) -> None:
+    config = tmp_path / "plyngent.toml"
+    _ = config.write_text(
+        """
+[providers.a]
+preset = "openai-compatible"
+url = "https://example.com/v1"
+access_key_or_token = "sk"
+
+[providers.a.models]
+"m1" = {}
+
+[providers.b]
+preset = "openai-compatible"
+url = "https://example.com/v1"
+access_key_or_token = "sk"
+
+[providers.b.models]
+"m2" = {}
+""",
+        encoding="utf-8",
+    )
+    runner = CliRunner()
+    result = runner.invoke(
+        main,
+        ["chat", "--config", str(config), "-p", "hi", "--workspace", str(tmp_path)],
+    )
+    assert result.exit_code != 0
+    assert "provider" in result.output.lower() or "provider" in (result.stderr or "").lower()
+
+
+def test_chat_oneshot_success(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    config = tmp_path / "plyngent.toml"
+    _ = config.write_text(
+        """
+[providers.local]
+preset = "openai-compatible"
+url = "https://example.com/v1"
+access_key_or_token = "sk"
+
+[providers.local.models]
+"tiny" = {}
+""",
+        encoding="utf-8",
+    )
+
+    class DummyClient:
+        @overload
+        async def chat_completions(
+            self, param: ChatCompletionsParam, *, stream: Literal[False] = False
+        ) -> ChatCompletionResponse: ...
+
+        @overload
+        async def chat_completions(
+            self, param: ChatCompletionsParam, *, stream: Literal[True]
+        ) -> AsyncIterator[ChatCompletionChunk]: ...
+
+        async def chat_completions(
+            self, param: ChatCompletionsParam, *, stream: bool = False
+        ) -> ChatCompletionResponse | AsyncIterator[ChatCompletionChunk]:
+            del param
+            if stream:
+
+                async def empty() -> AsyncIterator[ChatCompletionChunk]:
+                    if False:
+                        yield  # type: ignore[misc]
+                    return
+
+                return empty()
+            return ChatCompletionResponse(
+                id="1",
+                object="chat.completion",
+                created=0,
+                model="tiny",
+                choices=[
+                    ChatCompletionChoice(
+                        index=0,
+                        message=AssistantChatMessage(content="pong"),
+                        logprobs={},
+                        finish_reason="stop",
+                    )
+                ],
+                system_fingerprint="",
+                usage={},
+            )
+
+    monkeypatch.setattr("plyngent.cli.app.create_client", lambda _p: DummyClient())
+    monkeypatch.setattr(
+        "plyngent.cli.state.create_client",
+        lambda _p: DummyClient(),
+    )
+
+    runner = CliRunner()
+    result = runner.invoke(
+        main,
+        [
+            "chat",
+            "--config",
+            str(config),
+            "--provider",
+            "local",
+            "--model",
+            "tiny",
+            "-p",
+            "ping",
+            "--no-stream",
+            "--workspace",
+            str(tmp_path),
+            "--quiet",
+        ],
+    )
+    assert result.exit_code == EXIT_OK
+    assert "pong" in result.output
+
+
+def test_chat_help_mentions_prompt() -> None:
+    runner = CliRunner()
+    result = runner.invoke(main, ["chat", "--help"])
+    assert result.exit_code == 0
+    assert "--prompt" in result.output or "-p" in result.output
+    assert "Exit codes" in result.output or "one-shot" in result.output.lower()