refactor(approval): one "no user can answer" helper for every CLI-callback prompt

`hermes chat -q`, cron and unattended platforms can all have the CLI panel
callback registered with nobody to answer it. The MCP/vault consent prompt
(#133229) and the vault unlock prompt each spelled out
single_query or cron or unattended_platform by hand; the protected-instruction
gate and the memory prompt (salvaged above) checked single-query only, so a
cron job or a webhook session could still wait the full approval timeout
there. tools.approval_context._no_user_can_answer() now answers it once and
every one of those call sites uses it.

The memory prompt's _interactive_approval_available() wrapper is folded into
that one check, and its import- and runtime-failure fallbacks (and their
parametrized test) go: _no_user_can_answer reads session env through
_session_env, which already swallows errors, and tools.approval_context
imports nothing from the memory/terminal side.
This commit is contained in:
kshitijk4poor
2026-10-05 19:21:52 +05:30
committed by kshitij
parent 895dcd289f
commit 81208783ca
6 changed files with 23 additions and 93 deletions

View File

@@ -161,11 +161,7 @@ def is_unlocked(backend: str) -> bool:
def can_prompt_here() -> bool:
"""False in contexts where no human can answer (cron, webhook, api_server, -q)."""
from tools.approval_context import (
_is_cron_approval_context,
_is_single_query_approval_context,
_is_unattended_platform_approval_context,
)
if _is_cron_approval_context() or _is_unattended_platform_approval_context() or _is_single_query_approval_context():
from tools.approval_context import _no_user_can_answer
if _no_user_can_answer():
return False
return get_unlock_prompt_callback() is not None

View File

@@ -318,50 +318,19 @@ def test_memory_inline_deny_blocks(hermes_home, approval_callback_cleanup):
assert wa.pending_count("memory") == 0 # denied, not staged
@pytest.mark.parametrize("failure, staged", [("import", False), ("runtime", True)])
def test_memory_probe_failure_preserves_approval_contract(
failure, staged, hermes_home, approval_callback_cleanup, monkeypatch
@pytest.mark.parametrize("context", ["single_query", "cron", "webhook"])
def test_headless_memory_write_stages_without_inline_prompt(
hermes_home, approval_callback_cleanup, monkeypatch, context
):
import sys
from tools import approval, write_approval as wa
from tools.memory_tool import memory_tool, MemoryStore
from tools.terminal_tool import set_approval_callback
_set_approval("memory", True)
if failure == "import":
monkeypatch.setitem(sys.modules, "tools.approval", None)
else:
def broken_probe():
raise RuntimeError("probe failed")
monkeypatch.setattr(approval, "_is_single_query_approval_context", broken_probe)
calls = []
def approve(*args, **kwargs):
calls.append(args)
return "once"
set_approval_callback(approve)
store = MemoryStore()
store.load_from_disk()
result = json.loads(memory_tool("add", "memory", "probe fact", store=store))
assert result["success"] is True
assert bool(result.get("staged")) is staged
assert bool(calls) is not staged
assert store.memory_entries == ([] if staged else ["probe fact"])
assert wa.pending_count("memory") == int(staged)
def test_single_query_memory_stages_without_inline_prompt(
hermes_home, approval_callback_cleanup, monkeypatch
):
"""Headless ``hermes chat -q`` runs have no human to answer a callback."""
"""`hermes chat -q`, cron and unattended platforms can register the CLI callback but have no human to answer
it: the write stages at once instead of waiting the approval timeout."""
from tools.memory_tool import memory_tool, MemoryStore
from tools.terminal_tool import set_approval_callback
from tools import write_approval as wa
_set_approval("memory", True)
monkeypatch.setenv("HERMES_SINGLE_QUERY_SESSION", "1")
monkeypatch.setenv(*{"single_query": ("HERMES_SINGLE_QUERY_SESSION", "1"), "cron": ("HERMES_CRON_SESSION", "1"),
"webhook": ("HERMES_SESSION_PLATFORM", "webhook")}[context])
calls = []

View File

@@ -155,6 +155,13 @@ def _is_single_query_approval_context() -> bool:
return is_truthy_value(_session_env("HERMES_SINGLE_QUERY_SESSION"))
def _no_user_can_answer() -> bool:
"""True in single-query (-q), cron and unattended-platform sessions. `hermes chat -q` still registers the
CLI panel callback, so a prompt that only checks for a callback would wait the full timeout for nobody."""
return (_is_single_query_approval_context() or _is_cron_approval_context()
or _is_unattended_platform_approval_context())
def _is_gateway_approval_context() -> bool:
"""True inside a gateway/API session that can answer an approval.

View File

@@ -332,8 +332,7 @@ def request_elicitation_consent(message: str, description: str, *,
# Nobody can answer a -q, cron or unattended-platform worker. A `hermes chat -q` turn still has the CLI's
# panel callback registered, which would wait the full approval timeout before failing closed.
if (_ctx._is_single_query_approval_context() or _ctx._is_cron_approval_context()
or _ctx._is_unattended_platform_approval_context()):
if _ctx._no_user_can_answer():
logger.info("Elicitation consent (%s) declined: no user can answer in this session", surface)
return "decline"

View File

@@ -325,7 +325,8 @@ def _request_protected_instruction_approval(reasons: list[str], task_id: str = "
# approvals timeout, then reported "timed out" for a prompt no human saw. Detect that context the same
# way ``_run_approval_gate`` does (#86878) and fail closed immediately with the honest reason. Single-query
# runs are NEVER auto-approved here, whatever ``approvals.single_query_mode`` says.
if _approval._is_single_query_approval_context():
from tools.approval_context import _no_user_can_answer
if _no_user_can_answer():
return blocked.format(why=_NO_HUMAN)
# Same observer payload as the gateway branch (#131876), fired like the
# dangerous-command CLI prompt in tools/approval.py.

View File

@@ -216,50 +216,6 @@ def evaluate_gate(subsystem: str, *, inline_summary: str = "", inline_detail: st
return GateDecision(blocked=True, message="Memory write denied by user. The change was not saved.")
def _interactive_approval_available() -> bool:
"""True when a foreground memory write can be approved inline.
Inline prompting requires both a live interactive surface and a per-thread
approval callback registered by the interactive CLI
(``tools.terminal_tool.set_approval_callback``). A callback alone is not
sufficient: headless ``hermes chat -q`` runs also register one, but have no
human available to answer it. Kanban workers inherit this behavior because
the dispatcher launches them through ``hermes chat -q``.
Every other surface stages instead:
* **Single-query runs (including Kanban workers)** — headless even though a
CLI callback is registered; inline prompting would wait until timeout.
* **Gateway/API sessions** — the dangerous-command ``/approve`` round-trip
lives in the pending-approval queue (``submit_pending`` +
``_await_gateway_decision``), which ``prompt_dangerous_approval`` never
reaches; trying to prompt from a gateway session would hit the
``input()`` fallback and silently deny. Staging gives the user a real
review affordance (``/memory pending``) instead.
* Scripts, cron, and background threads — no user present.
"""
try:
from tools.approval import _is_single_query_approval_context
except ImportError:
# Keep the pre-existing callback path available if the optional context
# probe cannot be imported during early startup.
pass
else:
try:
if _is_single_query_approval_context():
return False
except Exception:
# An unreadable headless-context probe must fail safe to staging.
return False
try:
from tools.terminal_tool import _get_approval_callback
return _get_approval_callback() is not None
except Exception:
return False
def _prompt_inline_memory_approval(summary: str, detail: str) -> Optional[bool]:
"""Prompt inline for a memory write: True approved, False denied, None → stage. Uses the per-thread
CLI approval callback (``tools.terminal_tool.set_approval_callback``) directly, not
@@ -267,9 +223,11 @@ def _prompt_inline_memory_approval(summary: str, detail: str) -> Optional[bool]:
prompt_toolkit; silent deny in gateway sessions) and turns callback errors into a deny, whereas
here a missing channel or failed prompt must stage instead.
See #15216.
``hermes chat -q`` (and so every kanban worker), cron and unattended platforms can register a callback too,
but nobody answers it: stage at once instead of waiting the approval timeout. See #15216.
"""
if not _interactive_approval_available():
from tools.approval_context import _no_user_can_answer
if _no_user_can_answer():
return None
try:
from tools.terminal_tool import _get_approval_callback