diff --git a/src/plyngent/cli/app.py b/src/plyngent/cli/app.py index bb11233..aa41982 100644 --- a/src/plyngent/cli/app.py +++ b/src/plyngent/cli/app.py @@ -35,6 +35,7 @@ from plyngent.runtime import McpManager, ProviderNotSupportedError, create_clien if TYPE_CHECKING: from collections.abc import Mapping + from plyngent.config.models import NotifyConfig from plyngent.config.store import ConfigStore _DEFAULT_DB_FILENAME = "chat.db" @@ -128,13 +129,22 @@ def _read_prompt_text(prompt: str | None, *, stdin_isatty: bool) -> str | None: return text or None -def _setup_hooks(*, interactive: bool, auto_continue: bool) -> None: +def _setup_hooks(*, interactive: bool, auto_continue: bool, notify: NotifyConfig) -> None: """Install limit hooks (workspace policy is set on ReplState). ``auto_continue`` keeps the PTY/limit hook installed for one-shot runs so ``--auto-continue`` / ``[agent] auto_continue_limits`` can raise limits - without a prompt; otherwise non-interactive runs keep today's deny. + without a prompt; otherwise non-interactive runs keep today's deny. The + prompt-wait notifier only makes sense for a human at a terminal, so a + non-interactive run clears it whatever ``[notify]`` says. """ + from plyngent.cli.notify import install_notifier + from plyngent.prompting import set_prompt_wait_hook + + if interactive: + _ = install_notifier(notify) + else: + set_prompt_wait_hook(None) if interactive or auto_continue: install_cli_limit_hooks() else: @@ -248,7 +258,7 @@ async def _run_chat( # noqa: C901, PLR0915 — chat orchestration _warn_recoverable_providers(store.recoverable_providers) auto_effective = auto_continue or store.agent_config.auto_continue_limits - _setup_hooks(interactive=interactive, auto_continue=auto_effective) + _setup_hooks(interactive=interactive, auto_continue=auto_effective, notify=store.notify_config) # --yes forces sticky YOLO; else derive from config.confirm_destructive. from plyngent.cli.state import YoloMode diff --git a/src/plyngent/cli/limits.py b/src/plyngent/cli/limits.py index b10bbcd..8a2479c 100644 --- a/src/plyngent/cli/limits.py +++ b/src/plyngent/cli/limits.py @@ -4,6 +4,7 @@ import contextlib import selectors import shutil import sys +import time from dataclasses import dataclass from typing import TYPE_CHECKING, Literal @@ -13,6 +14,7 @@ from plyngent.cli.interrupt import off_loop_prompt, pause_task_cancel_for_prompt from plyngent.prompting import ( ChoiceOption, NonInteractiveError, + PromptAnswerSource, ask, ask_async, choose, @@ -20,6 +22,7 @@ from plyngent.prompting import ( configure_prompting, confirm, get_prompt_backend, + prompt_wait, set_prompt_output_hook, ) from plyngent.tools.access import AccessDecision @@ -30,12 +33,37 @@ from plyngent.tools.workspace import AccessMode, parse_access_mode if TYPE_CHECKING: from collections.abc import Mapping, Sequence from pathlib import Path + type WorkspaceMismatchChoice = Literal["keep", "rebind", "abort"] +# Answers a notification offers (see `cli/notify.py`) mapped onto the keys the +# prompts below already accept, so one answer source fits every confirm. +_NOTIFY_ANSWERS = {"allow": "y", "deny": "n"} + _BOX_MIN_WIDTH = 40 _BOX_MAX_WIDTH = 100 _BOX_PAD = 2 # spaces inside left/right borders +# How long one poll slice waits before a pending notification answer is checked. +_ANSWER_POLL_SECONDS = 0.1 + + +def _answer_source(wait: PromptAnswerSource | None) -> PromptAnswerSource | None: + """The wait's answer source, but only when the prompt offered answers.""" + return wait if wait is not None and wait.supports_actions else None + + +def _read_answer( + timeout_seconds: float | None, + wait: PromptAnswerSource | None, +) -> str | None: + """One answer line, from stdin or the notification (``None`` on timeout).""" + raw = _read_yes_no_line_with_timeout(timeout_seconds, source=_answer_source(wait)) + if raw is None: + return None + return _NOTIFY_ANSWERS.get(raw.strip().lower(), raw) + + # Limit prompts accept ``yyy`` to stop asking for the rest of the turn. _CONTINUE_PROMPT = "Raise limit and continue? [Y/n/yyy] (yyy = stop asking this turn): " _CONTINUE_YES_TOKENS = frozenset({"", "y", "yes"}) @@ -174,8 +202,9 @@ def _decide_continue_limit(reason: str) -> bool: backend = get_prompt_backend() if not backend.is_interactive(): return False - _echo_continue_limit(reason) - token = backend.read_line(_CONTINUE_PROMPT, default="y").strip().lower() + with prompt_wait("limit", "limit reached — continue?", reason): + _echo_continue_limit(reason) + token = backend.read_line(_CONTINUE_PROMPT, default="y").strip().lower() if token == _CONTINUE_AUTO_TOKEN: _auto_continue.turn = True return True @@ -205,12 +234,33 @@ async def prompt_continue_limit_async(reason: str) -> bool: return False +def _confirm_with(wait: PromptAnswerSource | None, prompt: str, *, default: bool) -> bool: + """Yes/no, also taking an answer that arrived through a notification.""" + source = _answer_source(wait) + if source is None: + return confirm(prompt, default=default) + suffix = "[Y/n]" if default else "[y/N]" + with contextlib.suppress(OSError): + _ = sys.stderr.write(f"{prompt} {suffix}: ") + _ = sys.stderr.flush() + try: + raw = _read_answer(None, wait) + except (KeyboardInterrupt, EOFError) as exc: + msg = "prompt cancelled" + raise NonInteractiveError(msg) from exc + token = (raw or "").strip().lower() + if not token: + return default + return token in {"y", "yes"} + + def _prompt_confirm_tool_sync(name: str, args: Mapping[str, object], reason: str) -> bool | str: """True allow; False deny; non-empty str = deny with comment for the model.""" del args try: - _echo_tool_confirm(name, reason) - allowed = confirm("Allow this tool call?", default=False) + with prompt_wait("confirm", f"tool {name!r} needs confirmation", reason) as wait: + _echo_tool_confirm(name, reason) + allowed = _confirm_with(wait, "Allow this tool call?", default=False) except NonInteractiveError: return False if allowed: @@ -242,34 +292,64 @@ def _try_readline() -> str | None: 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). - """ +def _stdin_selector() -> selectors.BaseSelector | None: + """A selector watching process stdin, or ``None`` when it cannot be polled.""" 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 None + return selector + + +def _read_yes_no_line_with_timeout( + timeout_seconds: float | None, + *, + source: PromptAnswerSource | None = None, +) -> str | None: + """Read one line from stdin with *timeout_seconds*; ``None`` on timeout/EOF. + + Polls the process stdin FD in short slices, so a *source* (a notification + answer) can end the read too; stdin wins whenever it becomes readable. A + ``None`` timeout waits for a line or an answer, whichever comes first; with + neither stdin nor a *source* pollable the read gives up at once. + On Windows without selector support, falls back to blocking readline (no + true timeout, and no answers from a *source*). + """ + try: + if not sys.stdin.isatty(): + return None if source is None else source.take() except OSError, ValueError, AttributeError: return _try_readline() + selector = _stdin_selector() + if selector is None: + return _try_readline() + + deadline = None if timeout_seconds is None else time.monotonic() + timeout_seconds + result: str | None = None try: - events = selector.select(timeout_seconds) - if not events: - return None - return sys.stdin.readline() - except OSError, ValueError, EOFError: - return None + while True: + if deadline is None: + remaining = _ANSWER_POLL_SECONDS + else: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + if selector.select(min(remaining, _ANSWER_POLL_SECONDS)): + try: + result = sys.stdin.readline() + except OSError, EOFError: + result = None + break + answer = None if source is None else source.take() + if answer is not None: + result = answer + break finally: selector.close() + return result def prompt_policy_command_confirm( @@ -290,7 +370,10 @@ def prompt_policy_command_confirm( f"(timeout {timeout_seconds:g}s defaults to DENY; not skipped by YOLO)" ) try: - with pause_task_cancel_for_prompt(): + with ( + pause_task_cancel_for_prompt(), + prompt_wait("policy", f"allow command {basename!r}?", reason) as wait, + ): backend = get_prompt_backend() if not backend.is_interactive(): return False @@ -301,7 +384,7 @@ def prompt_policy_command_confirm( with contextlib.suppress(OSError): _ = sys.stderr.write(prompt) _ = sys.stderr.flush() - raw = _read_yes_no_line_with_timeout(timeout_seconds) + raw = _read_answer(timeout_seconds, wait) if raw is None: with contextlib.suppress(OSError, NonInteractiveError): backend.secho( @@ -334,7 +417,10 @@ def prompt_policy_fetch_confirm( f"(timeout {timeout_seconds:g}s defaults to DENY; not skipped by YOLO)" ) try: - with pause_task_cancel_for_prompt(): + with ( + pause_task_cancel_for_prompt(), + prompt_wait("policy", f"allow fetch_local {host!r}?", reason) as wait, + ): backend = get_prompt_backend() if not backend.is_interactive(): return False @@ -346,7 +432,7 @@ def prompt_policy_fetch_confirm( with contextlib.suppress(OSError): _ = sys.stderr.write(prompt) _ = sys.stderr.flush() - raw = _read_yes_no_line_with_timeout(timeout_seconds) + raw = _read_answer(timeout_seconds, wait) if raw is None: with contextlib.suppress(OSError, NonInteractiveError): backend.secho( @@ -380,7 +466,10 @@ def prompt_skill_write_confirm( f"(timeout {timeout_seconds:g}s defaults to DENY; not skipped by YOLO)", ] try: - with pause_task_cancel_for_prompt(): + with ( + pause_task_cancel_for_prompt(), + prompt_wait("skills", f"allow {action} in {skill_dir.name!r}?", detail) as wait, + ): backend = get_prompt_backend() if not backend.is_interactive(): return SkillWriteDecision(allow=False) @@ -393,7 +482,7 @@ def prompt_skill_write_confirm( with contextlib.suppress(OSError): _ = sys.stderr.write(prompt) _ = sys.stderr.flush() - raw = _read_yes_no_line_with_timeout(timeout_seconds) + raw = _read_answer(timeout_seconds, wait) if raw is None: with contextlib.suppress(OSError, NonInteractiveError): backend.secho(f"[skills] timed out after {timeout_seconds:g}s — denied", fg="red", err=True) @@ -428,7 +517,10 @@ def prompt_directory_access_confirm( lines.append(f"why: {reason}") lines.append(f"(timeout {timeout_seconds:g}s defaults to DENY)") try: - with pause_task_cancel_for_prompt(): + with ( + pause_task_cancel_for_prompt(), + prompt_wait("access", f"allow {mode.name.lower()} access to {path}?", reason) as wait, + ): backend = get_prompt_backend() if not backend.is_interactive(): return None @@ -444,7 +536,7 @@ def prompt_directory_access_confirm( with contextlib.suppress(OSError): _ = sys.stderr.write(prompt) _ = sys.stderr.flush() - raw = _read_yes_no_line_with_timeout(timeout_seconds) + raw = _read_answer(timeout_seconds, wait) if raw is None: with contextlib.suppress(OSError, NonInteractiveError): backend.secho( @@ -470,8 +562,9 @@ async def prompt_confirm_tool_async(name: str, args: Mapping[str, object], reaso try: # Box is printed inside the worker thread via confirm's backend. def _run() -> bool: - _echo_tool_confirm(name, reason) - return confirm("Allow this tool call?", default=False) + with prompt_wait("confirm", f"tool {name!r} needs confirmation", reason) as wait: + _echo_tool_confirm(name, reason) + return _confirm_with(wait, "Allow this tool call?", default=False) from plyngent.prompting import run_prompt_async @@ -512,21 +605,30 @@ def _prompt_confirm_batch_sync(requests: Sequence[ToolConfirmRequest]) -> BatchC if not backend.is_interactive(): # Non-interactive: deny, like the default of the single prompt. return BatchConfirmDecision(allowed=False) - backend.echo() - backend.secho(format_batch_confirm_box(requests), fg="yellow") - backend.secho("m = review them one call at a time", fg="bright_black") - backend.echo() - try: - token = ( - backend.read_line( - f"Allow these {len(requests)} tool calls? [y/N/m]:", - default="n", - ) - .strip() - .lower() - ) - except NonInteractiveError: - return BatchConfirmDecision(allowed=False) + names = ", ".join(request.name for request in requests) + with prompt_wait("batch", f"{len(requests)} tool calls need confirmation", names) as wait: + backend.echo() + backend.secho(format_batch_confirm_box(requests), fg="yellow") + backend.secho("m = review them one call at a time", fg="bright_black") + backend.echo() + source = _answer_source(wait) + try: + if source is None: + token = ( + backend.read_line( + f"Allow these {len(requests)} tool calls? [y/N/m]:", + default="n", + ) + .strip() + .lower() + ) + else: + with contextlib.suppress(OSError): + _ = sys.stderr.write(f"Allow these {len(requests)} tool calls? [y/N]: ") + _ = sys.stderr.flush() + token = (_read_answer(None, wait) or "n").strip().lower() + except NonInteractiveError: + return BatchConfirmDecision(allowed=False) if token in _BATCH_MANUAL_TOKENS: return BatchConfirmDecision(allowed=False, each=True) if token in _BATCH_YES_TOKENS: diff --git a/src/plyngent/prompting.py b/src/plyngent/prompting.py index 69de584..a88e5e2 100644 --- a/src/plyngent/prompting.py +++ b/src/plyngent/prompting.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio import contextlib +import contextvars import select import sys import threading @@ -154,6 +155,9 @@ type PromptWaitHook = Callable[[PromptWait], AbstractContextManager[PromptAnswer # switched to another window; without a hook, prompting behaves as before. _wait_hook: PromptWaitHook | None = None +# Depth of the prompt_wait blocks currently entered (see prompt_wait). +_wait_depth: contextvars.ContextVar[int] = contextvars.ContextVar("prompt_wait_depth", default=0) + def set_prompt_wait_hook(hook: PromptWaitHook | None) -> None: """Install the hook run while a prompt blocks (None clears it).""" @@ -168,13 +172,22 @@ def get_prompt_wait_hook() -> PromptWaitHook | None: @contextlib.contextmanager def prompt_wait(kind: str, title: str, detail: str = "") -> Generator[PromptAnswerSource | None]: - """Announce a blocking prompt to the host; yields its answer source (or None).""" + """Announce a blocking prompt to the host; yields its answer source (or None). + + Only the outermost prompt of a nested read is announced: a confirm that asks + a follow-up question while it is already on screen must not notify — and + interrupt — a second time. + """ hook = _wait_hook - if hook is None: + if hook is None or _wait_depth.get() > 0: yield None return - with hook(PromptWait(kind, title, detail)) as source: - yield source + token = _wait_depth.set(_wait_depth.get() + 1) + try: + with hook(PromptWait(kind, title, detail)) as source: + yield source + finally: + _wait_depth.reset(token) def _readline_input(prompt: str, *, completions: Sequence[str] | None = None) -> str: diff --git a/tests/test_cli/test_limits.py b/tests/test_cli/test_limits.py index ce2840c..8aed77a 100644 --- a/tests/test_cli/test_limits.py +++ b/tests/test_cli/test_limits.py @@ -1,5 +1,6 @@ from __future__ import annotations +import contextlib from typing import TYPE_CHECKING import pytest @@ -15,12 +16,18 @@ from plyngent.cli.limits import ( reset_auto_continue_turn, set_auto_continue_default, ) -from plyngent.prompting import NonInteractiveBackend, get_prompt_backend, get_prompt_output_hook, temporary_backend +from plyngent.prompting import ( + NonInteractiveBackend, + get_prompt_backend, + get_prompt_output_hook, + get_prompt_wait_hook, + temporary_backend, +) from plyngent.tools.process.pty_session import PtyManager from tests.test_prompting import ScriptedBackend if TYPE_CHECKING: - from collections.abc import Iterator + from collections.abc import Generator, Iterator @pytest.fixture @@ -84,10 +91,11 @@ def test_prompt_continue_limit_noninteractive_denies() -> None: def test_setup_hooks_installs_pty_hook_for_auto_continue(clean_auto_continue: None) -> None: del clean_auto_continue from plyngent.cli.app import _setup_hooks + from plyngent.config.models import NotifyConfig - _setup_hooks(interactive=False, auto_continue=True) + _setup_hooks(interactive=False, auto_continue=True, notify=NotifyConfig()) assert PtyManager._limit_continue is not None - _setup_hooks(interactive=False, auto_continue=False) + _setup_hooks(interactive=False, auto_continue=False, notify=NotifyConfig()) assert PtyManager._limit_continue is None @@ -157,15 +165,134 @@ def test_directory_access_prompt_yes_and_level(monkeypatch: pytest.MonkeyPatch) target = Path("/tmp") with temporary_backend(ScriptedBackend([])): - monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout: "y\n") + monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout, **_kwargs: "y\n") assert limits.prompt_directory_access_confirm(target, AccessMode.READ, "why", 30.0) == AccessDecision( AccessMode.READ ) - monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout: "exec\n") + monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout, **_kwargs: "exec\n") assert limits.prompt_directory_access_confirm(target, AccessMode.READ, "why", 30.0) == AccessDecision( AccessMode.EXEC ) - monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout: "n\n") + monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout, **_kwargs: "n\n") assert limits.prompt_directory_access_confirm(target, AccessMode.READ, "why", 30.0) is None - monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout: None) + monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout, **_kwargs: None) assert limits.prompt_directory_access_confirm(target, AccessMode.READ, "why", 30.0) is None + + +class _AnswerSource: + """A host surface offering answers to a waiting prompt.""" + + supports_actions = True + + def __init__(self, *answers: str) -> None: + self._answers = list(answers) + + def take(self) -> str | None: + return self._answers.pop(0) if self._answers else None + + +@contextlib.contextmanager +def _answered(*answers: str) -> Generator[_AnswerSource]: + """Install a wait hook whose prompt can be answered by the notification.""" + from plyngent.prompting import set_prompt_wait_hook + + source = _AnswerSource(*answers) + previous = get_prompt_wait_hook() + set_prompt_wait_hook(lambda _wait: contextlib.nullcontext(source)) + try: + yield source + finally: + set_prompt_wait_hook(previous) + + +def test_directory_access_takes_a_notification_answer() -> None: + """``allow`` from the notification approves the requested level.""" + from pathlib import Path + + from plyngent.cli import limits + from plyngent.tools.access import AccessDecision + from plyngent.tools.workspace import AccessMode + + with temporary_backend(ScriptedBackend([])), _answered("allow"): + assert limits.prompt_directory_access_confirm(Path("/tmp"), AccessMode.READ, "why", 30.0) == AccessDecision( + AccessMode.READ + ) + + +def test_directory_access_denied_by_the_notification() -> None: + from pathlib import Path + + from plyngent.cli import limits + from plyngent.tools.workspace import AccessMode + + with temporary_backend(ScriptedBackend([])), _answered("deny"): + assert limits.prompt_directory_access_confirm(Path("/tmp"), AccessMode.READ, "why", 30.0) is None + + +def test_tool_confirm_takes_a_notification_answer() -> None: + with temporary_backend(ScriptedBackend([])), _answered("allow"): + assert prompt_confirm_tool("run_argv", {}, "runs a command") is True + + +def test_tool_confirm_denied_by_the_notification() -> None: + with temporary_backend(ScriptedBackend([""])), _answered("deny"): + assert prompt_confirm_tool("run_argv", {}, "runs a command") is False + + +def test_batch_confirm_takes_a_notification_answer() -> None: + from plyngent.agent import ToolConfirmRequest + from plyngent.cli.limits import _prompt_confirm_batch_sync + + requests = [ + ToolConfirmRequest(name="run_argv", args={}, reason="one"), + ToolConfirmRequest(name="run_argv", args={}, reason="two"), + ] + with temporary_backend(ScriptedBackend([])), _answered("allow"): + assert _prompt_confirm_batch_sync(requests).allowed is True + + +def test_batch_confirm_denied_by_the_notification() -> None: + from plyngent.agent import ToolConfirmRequest + from plyngent.cli.limits import _prompt_confirm_batch_sync + + requests = [ + ToolConfirmRequest(name="run_argv", args={}, reason="one"), + ToolConfirmRequest(name="run_argv", args={}, reason="two"), + ] + with temporary_backend(ScriptedBackend([""])), _answered("deny"): + assert _prompt_confirm_batch_sync(requests).allowed is False + + +def test_every_confirm_announces_itself() -> None: + """Each blocking confirm reports its kind, so a host can pick a surface.""" + from pathlib import Path + + from plyngent.agent import ToolConfirmRequest + from plyngent.cli import limits + from plyngent.tools.workspace import AccessMode + + kinds: list[str] = [] + + @contextlib.contextmanager + def hook(wait: object) -> Generator[None]: + kinds.append(wait.kind) # type: ignore[attr-defined] + yield None + + from plyngent.prompting import set_prompt_wait_hook + + previous = get_prompt_wait_hook() + set_prompt_wait_hook(hook) + try: + with temporary_backend(ScriptedBackend(["y", "y", "y"], confirms=[True])): + _ = prompt_confirm_tool("run_argv", {}, "reason") + _ = limits.prompt_directory_access_confirm(Path("/tmp"), AccessMode.READ, "why", 30.0) + _ = limits._prompt_confirm_batch_sync( + [ + ToolConfirmRequest(name="run_argv", args={}, reason="one"), + ToolConfirmRequest(name="run_argv", args={}, reason="two"), + ] + ) + _ = prompt_continue_limit("limit") + finally: + set_prompt_wait_hook(previous) + assert kinds == ["confirm", "access", "batch", "limit"] diff --git a/tests/test_cli/test_skills_cmd.py b/tests/test_cli/test_skills_cmd.py index 6e5becd..2e5113c 100644 --- a/tests/test_cli/test_skills_cmd.py +++ b/tests/test_cli/test_skills_cmd.py @@ -313,13 +313,13 @@ def test_prompt_skill_write_confirm_answers(monkeypatch: pytest.MonkeyPatch) -> target = Path("/skills/acme") with temporary_backend(ScriptedBackend([])): - monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout: "o\n") + monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout, **_kwargs: "o\n") once = limits.prompt_skill_write_confirm(target, "create", "create skill", 30.0) - monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout: "s\n") + monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout, **_kwargs: "s\n") session = limits.prompt_skill_write_confirm(target, "create", "create skill", 30.0) - monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout: "n\n") + monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout, **_kwargs: "n\n") denied = limits.prompt_skill_write_confirm(target, "create", "create skill", 30.0) - monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout: None) + monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout, **_kwargs: None) timed_out = limits.prompt_skill_write_confirm(target, "create", "create skill", 30.0) assert (once.allow, once.session) == (True, False) assert (session.allow, session.session) == (True, True)