diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 21ca4af6bb4d..be3e75520cac 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -1691,6 +1691,70 @@ def cache_media_bytes( return CachedMedia(to_agent_visible_cache_path(path), out_mime, "document", display or fallback_name) +def _normalize_post_delivery_callbacks(entry: Any) -> list[dict]: + """Return post-delivery callback records for legacy and current storage.""" + if entry is None: + return [] + if isinstance(entry, list): + return [item for item in entry if isinstance(item, dict)] + if isinstance(entry, tuple) and len(entry) == 2: + generation, callback = entry + return [{"generation": generation, "callback": callback, "owner_token": None}] + return [{"generation": None, "callback": entry, "owner_token": None}] + + +def _compose_post_delivery_callbacks(entries: list[dict]) -> Callable | None: + """Compose callback records into the historical single-callable API. + + Multiple selected records must be owner-isolated: one stuck callback cannot + prevent a later generation/owner callback from entering its own ``finally`` + cleanup path (for example the goal-run finalizer). Run selected callbacks + concurrently and cancel only the unfinished records after the bounded drain. + """ + callbacks = [entry.get("callback") for entry in entries if callable(entry.get("callback"))] + if not callbacks: + return None + if len(callbacks) == 1: + return callbacks[0] + + async def _run_callback(callback: Callable) -> None: + try: + result = callback() + if inspect.isawaitable(result): + await result + except asyncio.CancelledError: + raise + except Exception: + logger.debug("Post-delivery callback failed", exc_info=True) + + def _consume_task_exception(task: asyncio.Task) -> None: + try: + task.exception() + except (asyncio.CancelledError, Exception): + pass + + async def _chained() -> None: + tasks = [asyncio.create_task(_run_callback(callback)) for callback in callbacks] + for task in tasks: + task.add_done_callback(_consume_task_exception) + try: + done, pending = await asyncio.wait( + tasks, + timeout=_POST_DELIVERY_CALLBACK_TIMEOUT_SECONDS, + ) + for task in done: + _consume_task_exception(task) + for task in pending: + task.cancel() + except asyncio.CancelledError: + for task in tasks: + if not task.done(): + task.cancel() + raise + + return _chained + + class MessageType(Enum): """Types of incoming messages.""" TEXT = "text" @@ -3922,91 +3986,85 @@ def register_post_delivery_callback( callback: Callable, *, generation: int | None = None, - ) -> None: + owner_token: object | None = None, + ) -> object | None: """Register a deferred callback to fire after the main response. ``generation`` lets callers tie the callback to a specific gateway run generation so stale runs cannot clear callbacks owned by a fresher run. - If a callback for the same ``session_key`` (and generation, when set) - is already registered, the new callback is chained — both fire, in - registration order, with per-callback exception isolation. This lets - independent features (background-review release + temporary-bubble - cleanup) coexist without clobbering each other. Stale-generation - callers never overwrite a fresher generation's slot. + ``owner_token`` lets a caller later remove only its own callback while + preserving unrelated same-session/same-generation callbacks. Without + an owner token, ``pop_post_delivery_callback(..., generation=N)`` keeps + the historical behavior of popping every callback for that generation. """ if not session_key or not callable(callback): - return - - existing = self._post_delivery_callbacks.get(session_key) - if existing is not None: - if isinstance(existing, tuple) and len(existing) == 2: - existing_gen, existing_cb = existing - else: - existing_gen, existing_cb = None, existing - # Stale-generation registrations never overwrite a fresher slot. - if ( - existing_gen is not None - and generation is not None - and int(generation) < int(existing_gen) - ): - return - # Same-or-newer generation: chain with the existing callback so - # both fire in registration order. - if callable(existing_cb) and ( - existing_gen is None - or generation is None - or int(existing_gen) == int(generation) - ): - _prev = existing_cb - _new = callback - - async def _chained() -> None: - # Both _prev and _new may be sync or async. The chained - # wrapper itself must be async because the outer invoker - # (``_handle_message`` etc.) awaits awaitable callbacks; a - # sync wrapper here would call ``_prev()`` / ``_new()`` and - # silently drop any returned coroutine, breaking chained - # async post-delivery hooks (e.g. ``/goal`` continuations). - for _cb in (_prev, _new): - try: - _result = _cb() - if inspect.isawaitable(_result): - await _result - except Exception: - logger.debug( - "Post-delivery callback failed", exc_info=True - ) - - callback = _chained + return None - if generation is None: - self._post_delivery_callbacks[session_key] = callback - else: - self._post_delivery_callbacks[session_key] = (int(generation), callback) + normalized_generation = int(generation) if generation is not None else None + entries = _normalize_post_delivery_callbacks( + self._post_delivery_callbacks.get(session_key) + ) + generation_entries = [ + entry for entry in entries if entry.get("generation") is not None + ] + if normalized_generation is not None and generation_entries: + max_generation = max(int(entry["generation"]) for entry in generation_entries) + if normalized_generation < max_generation: + return None + if normalized_generation > max_generation: + entries = [ + entry + for entry in entries + if entry.get("generation") is None + or int(entry["generation"]) == normalized_generation + ] + + entries.append( + { + "generation": normalized_generation, + "callback": callback, + "owner_token": owner_token, + } + ) + self._post_delivery_callbacks[session_key] = entries + return owner_token def pop_post_delivery_callback( self, session_key: str, *, generation: int | None = None, + owner_token: object | None = None, ) -> Callable | None: - """Pop a deferred callback, optionally requiring generation ownership.""" + """Pop deferred callback(s), optionally requiring generation/owner ownership.""" if not session_key: return None - entry = self._post_delivery_callbacks.get(session_key) - if entry is None: + entries = _normalize_post_delivery_callbacks( + self._post_delivery_callbacks.get(session_key) + ) + if not entries: return None - if isinstance(entry, tuple) and len(entry) == 2: - entry_generation, callback = entry - if generation is not None and int(entry_generation) != int(generation): - return None - self._post_delivery_callbacks.pop(session_key, None) - return callback if callable(callback) else None - if generation is not None: + normalized_generation = int(generation) if generation is not None else None + + def _matches(entry: dict) -> bool: + entry_generation = entry.get("generation") + if normalized_generation is not None: + if entry_generation is None or int(entry_generation) != normalized_generation: + return False + if owner_token is not None: + return entry.get("owner_token") is owner_token + return True + + selected = [entry for entry in entries if _matches(entry)] + if not selected: return None - self._post_delivery_callbacks.pop(session_key, None) - return entry if callable(entry) else None + remaining = [entry for entry in entries if not _matches(entry)] + if remaining: + self._post_delivery_callbacks[session_key] = remaining + else: + self._post_delivery_callbacks.pop(session_key, None) + return _compose_post_delivery_callbacks(selected) # ── Processing lifecycle hooks ────────────────────────────────────────── # Subclasses override these to react to message processing events diff --git a/gateway/run.py b/gateway/run.py index ef49bf67c47f..cd5f2350e50c 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -4085,6 +4085,16 @@ def _request_clean_exit(self, reason: str) -> None: def _running_agent_count(self) -> int: return len(self._running_agents) + def _active_goal_run_count(self) -> int: + """Count generation-owned goal workers still evaluating or gating.""" + lock, controls = self._goal_run_control_state() + with lock: + return sum( + not control["done"].is_set() + for session_controls in controls.values() + for control in session_controls + ) + def _active_cron_job_count(self) -> int: """Count of cron jobs currently executing, from the cron scheduler's own in-flight tracking (``cron.scheduler._running_job_ids``). @@ -5658,31 +5668,33 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session async def _drain_active_agents(self, timeout: float) -> tuple[Dict[str, Any], bool]: snapshot = self._snapshot_running_agents() last_active_count = self._running_agent_count() + last_goal_count = self._active_goal_run_count() last_cron_count = self._active_cron_job_count() last_status_at = 0.0 def _maybe_update_status(force: bool = False) -> None: - nonlocal last_active_count, last_cron_count, last_status_at + nonlocal last_active_count, last_goal_count, last_cron_count, last_status_at now = asyncio.get_running_loop().time() active_count = self._running_agent_count() + goal_count = self._active_goal_run_count() cron_count = self._active_cron_job_count() if ( force or active_count != last_active_count + or goal_count != last_goal_count or cron_count != last_cron_count or (now - last_status_at) >= 1.0 ): self._update_runtime_status("draining") last_active_count = active_count + last_goal_count = goal_count last_cron_count = cron_count last_status_at = now - # Cron jobs run on the scheduler's own thread pool, outside - # ``self._running_agents`` — fold their in-flight count into the - # same wait/timeout this method already applies to chat sessions, - # or a cron job's tool work gets killed with zero warning the - # instant it's the only active thing running (#60432). - if not self._running_agents and last_cron_count == 0: + # Cron jobs and /goal continuations run outside ``self._running_agents``; + # fold their in-flight counts into the same wait/timeout this method + # already applies to chat sessions. + if not self._running_agents and last_goal_count == 0 and last_cron_count == 0: _maybe_update_status(force=True) return snapshot, False @@ -5692,16 +5704,25 @@ def _maybe_update_status(force: bool = False) -> None: deadline = asyncio.get_running_loop().time() + timeout while ( - (self._running_agents or self._active_cron_job_count()) + ( + self._running_agents + or self._active_goal_run_count() + or self._active_cron_job_count() + ) and asyncio.get_running_loop().time() < deadline ): _maybe_update_status() await asyncio.sleep(0.1) - timed_out = bool(self._running_agents) or bool(self._active_cron_job_count()) + timed_out = ( + bool(self._running_agents) + or bool(self._active_goal_run_count()) + or bool(self._active_cron_job_count()) + ) _maybe_update_status(force=True) return snapshot, timed_out def _interrupt_running_agents(self, reason: str) -> None: + self._cancel_all_goal_runs() for session_key, agent in list(self._running_agents.items()): if agent is _AGENT_PENDING_SENTINEL: continue @@ -10303,6 +10324,7 @@ async def _do_undo(): session_entry=session_entry, source=source, final_response=_final_text, + run_generation=_run_generation, ) except Exception as _goal_exc: logger.debug("goal continuation hook failed: %s", _goal_exc) @@ -10317,14 +10339,13 @@ async def _do_undo(): # Putting it in finally guarantees the revert on success, exception, # and interrupt alike. self._restore_moa_one_shot(event, _quick_key) - # Unconditional release covers every exit path. _release_running_agent_state - # is idempotent (pop-on-absent is harmless) and, called without a - # run_generation guard, always clears the slot regardless of which - # generation it holds. This evicts the zombie left when session_reset - # bumps the generation (N -> N+1) mid-flight: gen-N's guarded release - # inside _run_agent returns False, and the old sentinel-only check here - # missed the leftover real agent — locking the session out forever (#28686). - self._release_running_agent_state(_quick_key) + # A stopped goal worker may unwind after /new has already reused the + # same session key. Release only this turn's generation so stale + # cleanup cannot clear the replacement run's slot. Interrupt/reset + # paths perform their own unconditional release before reuse. + self._release_running_agent_state( + _quick_key, run_generation=_run_generation + ) def _restore_moa_one_shot(self, event: "MessageEvent", quick_key: str) -> None: """Revert a ``/moa `` one-shot model override after its turn. @@ -12742,48 +12763,119 @@ async def _send_goal_status_notice(self, source: Any, message: str) -> None: getattr(result, "error", "unknown error"), ) - async def _defer_goal_status_notice_after_delivery(self, source: Any, message: str) -> None: + async def _defer_goal_status_notice_after_delivery( + self, + source: Any, + message: str, + *, + session_key: str = "", + run_generation: int | None = None, + control: dict | None = None, + ) -> bool: """Send a /goal status line after the main response is delivered. The gateway message handler returns the agent response to the platform - adapter, which sends it after this method's caller has returned. For a + adapter, which sends it after this method's caller has returned. For a natural Discord/Telegram reading order, goal status belongs after that - send. Platform adapters provide a one-shot post-delivery callback for + send. Platform adapters provide a one-shot post-delivery callback for exactly this boundary; when unavailable, fall back to direct awaited delivery rather than silently dropping the notice. + + When ``control`` is supplied, a deferred callback owns goal-run shutdown + accounting until the callback fires or cancellation removes that exact + owner token. """ adapter = self._adapter_for_source(source) if not adapter: logger.debug("goal continuation: no adapter for %s", getattr(source, "platform", None)) - return + return False + + deferred_owner = False + effective_generation = run_generation + if effective_generation is None and control is not None: + effective_generation = int(control["generation"]) async def _deliver() -> None: try: + if control is not None and ( + control["cancel"].is_set() + or not self._is_session_run_current( + session_key, + effective_generation or 0, + ) + ): + return await self._send_goal_status_notice(source, message) except Exception as exc: logger.warning("goal continuation: status send failed: %s", exc, exc_info=True) + finally: + if control is not None and deferred_owner: + self._finish_goal_run_control(session_key, control) - try: - session_key = self._session_key_for_source(source) - except Exception: - session_key = None + if not session_key: + try: + session_key = self._session_key_for_source(source) + except Exception: + session_key = "" if session_key and hasattr(adapter, "register_post_delivery_callback"): try: - generation = None - active = getattr(adapter, "_active_sessions", {}).get(session_key) - if active is not None: - generation = getattr(active, "_hermes_run_generation", None) - adapter.register_post_delivery_callback( - session_key, - _deliver, - generation=generation, - ) - return + generation = effective_generation + if generation is None: + active = getattr(adapter, "_active_sessions", {}).get(session_key) + if active is not None: + generation = getattr(active, "_hermes_run_generation", None) + + lock, controls = self._goal_run_control_state() + with lock: + if control is not None and ( + control not in controls.get(session_key, ()) + or control["cancel"].is_set() + ): + return False + owner_token = object() if control is not None else None + try: + adapter.register_post_delivery_callback( + session_key, + _deliver, + generation=generation, + owner_token=owner_token, + ) + except TypeError: + owner_token = None + adapter.register_post_delivery_callback( + session_key, + _deliver, + generation=generation, + ) + + callbacks = getattr(adapter, "_post_delivery_callbacks", None) + entry = callbacks.get(session_key) if isinstance(callbacks, dict) else None + accepted = callbacks is None or entry is not None + if accepted and generation is not None and isinstance(entry, tuple): + accepted = int(entry[0]) == int(generation) + elif accepted and generation is not None and isinstance(entry, list): + accepted = any( + isinstance(item, dict) + and item.get("callback") is _deliver + and item.get("generation") is not None + and int(item.get("generation")) == int(generation) + and item.get("owner_token") is owner_token + for item in entry + ) + if not accepted: + return False + if control is not None: + control["post_delivery_adapter"] = adapter + control["post_delivery_generation"] = generation + control["post_delivery_owner_token"] = owner_token + deferred_owner = True + return True except Exception as exc: logger.debug("goal continuation: post-delivery callback registration failed: %s", exc) await _deliver() + return False async def _post_turn_goal_continuation( self, @@ -12791,78 +12883,207 @@ async def _post_turn_goal_continuation( session_entry: Any, source: Any, final_response: str, + run_generation: int | None = None, ) -> None: """Run the goal judge after a gateway turn and, if still active, enqueue a continuation prompt for the same session. - Called from ``_handle_message_with_agent`` at turn boundary, AFTER - the response has been delivered. Safe when no goal is set. - - We use the adapter's pending-message / FIFO machinery so any real - user message that arrives simultaneously is handled by the same - queue and takes priority naturally. + Goal evaluation and gate execution are blocking-capable, so they run in + the gateway executor under session context. The async side owns a + generation-scoped control record until every status/continuation side + effect is complete or cancellation removes the exact owner. """ - try: - from hermes_cli.goals import GoalManager - except Exception as exc: - logger.debug("goal continuation: goals module unavailable: %s", exc) - return - sid = getattr(session_entry, "session_id", None) or "" if not sid: return max_turns = self._goal_max_turns_from_config() - mgr = GoalManager(session_id=sid, default_max_turns=max_turns) - if not mgr.is_active(): - return + session_key = getattr(session_entry, "session_key", None) or "" + if not session_key: + try: + session_key = self._session_key_for_source(source) + except Exception: + session_key = "" + if run_generation is None: + run_generation = int( + (self.__dict__.get("_session_run_generation") or {}).get(session_key, 0) + ) + control = self._begin_goal_run_control(session_key, run_generation) + cancelled_result = object() + + def _goal_run_is_current() -> bool: + return bool( + not control["cancel"].is_set() + and self._is_session_run_current(session_key, run_generation) + ) + from tools.approval import ( + register_gateway_notify, + reset_deferred_command_session_authorization_required, + reset_current_session_key, + set_current_session_key, + set_deferred_command_session_authorization_required, + unregister_gateway_notify, + ) + + adapter = self._adapter_for_source(source) if source is not None else None try: - from hermes_cli.goals import gather_background_processes as _gather_bg - _bg_procs = _gather_bg() + metadata = self._thread_metadata_for_source(source) if source is not None else None except Exception: - _bg_procs = None + metadata = None + loop = asyncio.get_running_loop() - decision = mgr.evaluate_after_turn( - final_response or "", - user_initiated=True, - background_processes=_bg_procs, - ) - msg = decision.get("message") or "" - - # Defer the status line until after the adapter has delivered the - # agent's visible final response. The judge runs after the response is - # produced but before BasePlatformAdapter sends it, so sending here - # would show "✓ Goal achieved" before the answer itself. Registering - # an awaited post-delivery callback preserves delivery reliability - # without reversing the user-visible ordering. - if msg and source is not None: - await self._defer_goal_status_notice_after_delivery(source, msg) - - if not decision.get("should_continue"): - return + def _approval_notify_for_goal(approval_data: dict) -> None: + if adapter is None: + return + try: + if hasattr(adapter, "pause_typing_for_chat"): + adapter.pause_typing_for_chat(source.chat_id) + cmd = _redact_approval_command(approval_data.get("command", "")) + desc = approval_data.get("description", "dangerous command") + command_prefix = getattr(adapter, "typed_command_prefix", "/") + cmd_preview = cmd[:200] + "..." if len(cmd) > 200 else cmd + choices = [ + f"Reply `{command_prefix}approve` to execute this one operation" + ] + if approval_data.get("allow_session", True): + choices.append( + f"`{command_prefix}approve session` to approve this pattern for the session" + ) + if approval_data.get("allow_permanent", True): + choices.append( + f"`{command_prefix}approve always` to approve permanently" + ) + choices.append(f"`{command_prefix}deny` to cancel") + msg = ( + "⚠️ **Dangerous command requires approval:**\n" + f"```\n{cmd_preview}\n```\n" + f"Reason: {desc}\n\n" + + ", ".join(choices[:-1]) + + f", or {choices[-1]}." + ) + fut = safe_schedule_threadsafe( + adapter.send(source.chat_id, msg, metadata=metadata), + loop, + logger=logger, + log_message="Goal approval text-send scheduling error", + ) + if fut is not None: + fut.result(timeout=15) + except Exception as exc: + logger.error("Failed to send goal approval request: %s", exc) - prompt = decision.get("continuation_prompt") or "" - if not prompt or source is None: - return + def _evaluate_goal_sync_inner(): + from hermes_cli.goals import GoalManager, gather_background_processes + + mgr = GoalManager(session_id=sid, default_max_turns=max_turns) + if not mgr.is_active(): + return None + try: + background_processes = gather_background_processes() + except Exception: + background_processes = None + return mgr.evaluate_after_turn( + final_response or "", + user_initiated=True, + background_processes=background_processes, + ) + + def _evaluate_goal_sync(): + self._set_goal_run_thread(session_key, control) + try: + if control["cancel"].is_set(): + return cancelled_result + decision = _evaluate_goal_sync_inner() + if control["cancel"].is_set(): + return cancelled_result + return decision + finally: + self._finish_goal_run_executor(session_key, control) - # Enqueue via the adapter's FIFO so a user message already in - # flight preempts the continuation naturally. + session_context = SessionContext( + source=source, + connected_platforms=[], + home_channels={}, + session_key=session_key, + session_id=sid, + created_at=getattr(session_entry, "created_at", None), + updated_at=getattr(session_entry, "updated_at", None), + ) + session_tokens = self._set_session_env(session_context) + approval_token = set_current_session_key(session_key) + deferred_authorization_token = ( + set_deferred_command_session_authorization_required(True) + ) + notify_registered = bool(adapter is not None and session_key) + notify_owner_token = None + if notify_registered: + notify_owner_token = register_gateway_notify( + session_key, _approval_notify_for_goal + ) + self._set_goal_run_notify_token( + session_key, control, notify_owner_token + ) try: - adapter = self._adapter_for_source(source) - _quick_key = self._session_key_for_source(source) - if adapter and _quick_key: - cont_event = MessageEvent( - text=prompt, - message_type=MessageType.TEXT, - source=source, - message_id=None, - channel_prompt=None, + decision = await self._run_in_executor_with_context(_evaluate_goal_sync) + except BaseException: + control["async_abandoned"].set() + if control["executor_done"].is_set(): + self._finish_goal_run_control(session_key, control) + raise + finally: + if notify_registered: + unregister_gateway_notify(session_key, notify_owner_token) + reset_deferred_command_session_authorization_required( + deferred_authorization_token + ) + reset_current_session_key(approval_token) + self._clear_session_env(session_tokens) + + notice_deferred = False + try: + if ( + decision is cancelled_result + or decision is None + or not _goal_run_is_current() + ): + return + msg = decision.get("message") or "" + + if msg and source is not None and _goal_run_is_current(): + notice_deferred = await self._defer_goal_status_notice_after_delivery( + source, + msg, + session_key=session_key, + run_generation=run_generation, + control=control, ) - self._enqueue_fifo(_quick_key, cont_event, adapter) - except Exception as exc: - logger.debug("goal continuation: enqueue failed: %s", exc) + + if not _goal_run_is_current() or not decision.get("should_continue"): + return + + prompt = decision.get("continuation_prompt") or "" + if not prompt or source is None: + return + + try: + adapter = self._adapter_for_source(source) + _quick_key = self._session_key_for_source(source) + if adapter and _quick_key and _goal_run_is_current(): + cont_event = MessageEvent( + text=prompt, + message_type=MessageType.TEXT, + source=source, + message_id=None, + channel_prompt=None, + ) + self._enqueue_fifo(_quick_key, cont_event, adapter) + except Exception as exc: + logger.debug("goal continuation: enqueue failed: %s", exc) + finally: + if not notice_deferred: + self._finish_goal_run_control(session_key, control) @@ -16131,6 +16352,204 @@ def _is_session_run_current(self, session_key: str, generation: int) -> bool: generations = self.__dict__.get("_session_run_generation") or {} return int(generations.get(session_key, 0)) == int(generation) + def _goal_run_control_state(self): + """Return lazily-created generation-owned goal worker state.""" + lock = getattr(self, "_goal_run_controls_lock", None) + if lock is None: + lock = threading.RLock() + self._goal_run_controls_lock = lock + controls = getattr(self, "_goal_run_controls", None) + if controls is None: + controls = {} + self._goal_run_controls = controls + return lock, controls + + def _begin_goal_run_control(self, session_key: str, generation: int) -> dict: + """Install a cancellation fence for one goal-evaluation generation.""" + control = { + "generation": int(generation), + "cancel": threading.Event(), + "done": threading.Event(), + "executor_done": threading.Event(), + "async_abandoned": threading.Event(), + "thread_id": None, + "notify_token": None, + "post_delivery_adapter": None, + "post_delivery_generation": None, + "post_delivery_owner_token": None, + } + lock, controls = self._goal_run_control_state() + with lock: + active_generation = int( + (self.__dict__.get("_session_run_generation") or {}).get( + session_key, 0 + ) + ) + if session_key and active_generation and int(generation) < active_generation: + control["cancel"].set() + control["done"].set() + return control + prior_controls = [ + prior + for prior in controls.get(session_key, ()) + if ( + not prior["done"].is_set() + and int(prior.get("generation", 0)) <= int(generation) + ) + ] + for prior in prior_controls: + prior["cancel"].set() + controls.setdefault(session_key, []).append(control) + self._signal_goal_run_cancellation(session_key, prior_controls) + return control + + def _set_goal_run_thread(self, session_key: str, control: dict) -> None: + """Bind the worker thread and honor cancellation that raced startup.""" + thread_id = threading.get_ident() + lock, controls = self._goal_run_control_state() + with lock: + if control not in controls.get(session_key, ()): + return + control["thread_id"] = thread_id + cancelled = control["cancel"].is_set() + if cancelled: + from tools.interrupt import set_interrupt + + set_interrupt(True, thread_id=thread_id) + + def _finish_goal_run_executor(self, session_key: str, control: dict) -> None: + """Atomically release a control's reusable executor-thread ownership.""" + from tools.interrupt import clear_current_thread_interrupt + + lock, _controls = self._goal_run_control_state() + with lock: + clear_current_thread_interrupt() + control["thread_id"] = None + control["executor_done"].set() + async_abandoned = control["async_abandoned"].is_set() + if async_abandoned: + self._finish_goal_run_control(session_key, control) + + def _set_goal_run_notify_token( + self, session_key: str, control: dict, notify_token: object + ) -> None: + lock, controls = self._goal_run_control_state() + with lock: + if control in controls.get(session_key, ()): + control["notify_token"] = notify_token + + def _signal_goal_run_cancellation( + self, session_key: str, controls: list[dict] + ) -> None: + """Release approval waits and interrupt workers already marked cancelled.""" + notify_tokens = [] + retired_callbacks = [] + lock, active_controls = self._goal_run_control_state() + with lock: + for control in controls: + if ( + control not in active_controls.get(session_key, ()) + or control["done"].is_set() + ): + continue + notify_token = control.get("notify_token") + if notify_token is not None: + notify_tokens.append(notify_token) + + callback_adapter = control.get("post_delivery_adapter") + callback_generation = control.get("post_delivery_generation") + callback_owner_token = control.get("post_delivery_owner_token") + if callback_adapter is not None and hasattr( + callback_adapter, "pop_post_delivery_callback" + ): + try: + callback = callback_adapter.pop_post_delivery_callback( + session_key, + generation=callback_generation, + owner_token=callback_owner_token, + ) + except TypeError: + callback = callback_adapter.pop_post_delivery_callback( + session_key, + generation=callback_generation, + ) + if callback is not None: + control["post_delivery_adapter"] = None + control["post_delivery_generation"] = None + control["post_delivery_owner_token"] = None + retired_callbacks.append(control) + + thread_id = control.get("thread_id") + if thread_id is not None and not control["executor_done"].is_set(): + from tools.interrupt import set_interrupt + + set_interrupt(True, thread_id=thread_id) + + for notify_token in notify_tokens: + from tools.approval import unregister_gateway_notify + + unregister_gateway_notify(session_key, notify_token) + for control in retired_callbacks: + self._finish_goal_run_control(session_key, control) + + def _cancel_goal_run(self, session_key: str) -> bool: + """Cancel every unfinished goal generation for one session.""" + lock, controls = self._goal_run_control_state() + with lock: + session_controls = [ + control + for control in controls.get(session_key, ()) + if not control["done"].is_set() + ] + if not session_controls: + return False + for control in session_controls: + control["cancel"].set() + + self._signal_goal_run_cancellation(session_key, session_controls) + return True + + def _cancel_all_goal_runs(self) -> None: + lock, controls = self._goal_run_control_state() + with lock: + active_by_session = { + session_key: [ + control + for control in session_controls + if not control["done"].is_set() + ] + for session_key, session_controls in controls.items() + } + active_by_session = { + session_key: session_controls + for session_key, session_controls in active_by_session.items() + if session_controls + } + for session_controls in active_by_session.values(): + for control in session_controls: + control["cancel"].set() + for session_key, session_controls in active_by_session.items(): + self._signal_goal_run_cancellation(session_key, session_controls) + + def _finish_goal_run_control(self, session_key: str, control: dict) -> None: + """Retire only the control owned by the finishing worker generation.""" + control["done"].set() + lock, controls = self._goal_run_control_state() + with lock: + session_controls = controls.get(session_key, []) + if control in session_controls: + session_controls.remove(control) + if not session_controls: + controls.pop(session_key, None) + + def _has_active_goal_run(self, session_key: str) -> bool: + lock, controls = self._goal_run_control_state() + with lock: + return any( + not control["done"].is_set() + for control in controls.get(session_key, ()) + ) + def _bind_adapter_run_generation( self, adapter: Any, @@ -16159,6 +16578,10 @@ async def _interrupt_and_clear_session( """Interrupt the current run and clear queued session state consistently.""" if not session_key: return + # Goal evaluation runs after the main AIAgent has released its slot, + # but can still be blocked in approval or a gate subprocess. Cancel + # that exact generation before releasing routing state. + self._cancel_goal_run(session_key) running_agent = self._running_agents.get(session_key) if running_agent and running_agent is not _AGENT_PENDING_SENTINEL: running_agent.interrupt(interrupt_reason) @@ -16953,27 +17376,44 @@ async def _run_agent( multiplexing is off this is a transparent pass-through — zero behavior change for single-profile gateways. """ - if not getattr(getattr(self, "config", None), "multiplex_profiles", False): - return await self._run_agent_inner( - message, context_prompt, history, source, session_id, - session_key=session_key, run_generation=run_generation, - _interrupt_depth=_interrupt_depth, event_message_id=event_message_id, - channel_prompt=channel_prompt, moa_config=moa_config, - persist_user_message=persist_user_message, - persist_user_timestamp=persist_user_timestamp, + deferred_goal_auth_token = None + if self._is_goal_continuation_event(message): + from tools.approval import ( + reset_deferred_command_session_authorization_required, + set_deferred_command_session_authorization_required, ) - profile_home = self._resolve_profile_home_for_source(source) - with _profile_runtime_scope(profile_home): - return await self._run_agent_inner( - message, context_prompt, history, source, session_id, - session_key=session_key, run_generation=run_generation, - _interrupt_depth=_interrupt_depth, event_message_id=event_message_id, - channel_prompt=channel_prompt, moa_config=moa_config, - persist_user_message=persist_user_message, - persist_user_timestamp=persist_user_timestamp, + deferred_goal_auth_token = ( + set_deferred_command_session_authorization_required(True) ) + try: + if not getattr(getattr(self, "config", None), "multiplex_profiles", False): + return await self._run_agent_inner( + message, context_prompt, history, source, session_id, + session_key=session_key, run_generation=run_generation, + _interrupt_depth=_interrupt_depth, event_message_id=event_message_id, + channel_prompt=channel_prompt, moa_config=moa_config, + persist_user_message=persist_user_message, + persist_user_timestamp=persist_user_timestamp, + ) + + profile_home = self._resolve_profile_home_for_source(source) + with _profile_runtime_scope(profile_home): + return await self._run_agent_inner( + message, context_prompt, history, source, session_id, + session_key=session_key, run_generation=run_generation, + _interrupt_depth=_interrupt_depth, event_message_id=event_message_id, + channel_prompt=channel_prompt, moa_config=moa_config, + persist_user_message=persist_user_message, + persist_user_timestamp=persist_user_timestamp, + ) + finally: + if deferred_goal_auth_token is not None: + reset_deferred_command_session_authorization_required( + deferred_goal_auth_token + ) + def _resolve_profile_home_for_source(self, source: SessionSource) -> "Path": """Resolve which profile's HERMES_HOME should serve this inbound source. @@ -18858,7 +19298,9 @@ def _approval_notify_sync(approval_data: dict) -> None: _approval_session_key = session_key or "" _approval_session_token = set_current_session_key(_approval_session_key) - register_gateway_notify(_approval_session_key, _approval_notify_sync) + _approval_notify_token = register_gateway_notify( + _approval_session_key, _approval_notify_sync + ) try: # If _prepare_inbound_message_text buffered image paths for native # attachment, wrap the user turn as an OpenAI-style multimodal @@ -18909,7 +19351,9 @@ def _approval_notify_sync(approval_data: dict) -> None: _conversation_kwargs["persist_user_timestamp"] = _persist_user_timestamp_override result = agent.run_conversation(_api_run_message, **_conversation_kwargs) finally: - unregister_gateway_notify(_approval_session_key) + unregister_gateway_notify( + _approval_session_key, _approval_notify_token + ) # Cancel any pending clarify entries so blocked agent # threads don't hang past the end of the run (interrupt, # completion, gateway shutdown). Idempotent. diff --git a/tests/acp/test_approval_isolation.py b/tests/acp/test_approval_isolation.py index 30d783f42e19..274e31d52611 100644 --- a/tests/acp/test_approval_isolation.py +++ b/tests/acp/test_approval_isolation.py @@ -15,6 +15,33 @@ import threading +import pytest + + +@pytest.fixture(autouse=True) +def _isolate_approval_state(monkeypatch): + """Keep approval-isolation regressions independent of user config. + + The real developer profile may carry broad permanent/session allowlist + entries. These tests assert routing behavior, so ambient approvals must not + short-circuit the dangerous-command path before the callback/gateway owner + logic under test runs. + """ + from tools import approval + + permanent = set(approval._permanent_approved) + session = {key: set(value) for key, value in approval._session_approved.items()} + approval._permanent_approved.clear() + approval._session_approved.clear() + monkeypatch.setattr(approval, "_get_approval_mode", lambda: "manual") + try: + yield + finally: + approval._permanent_approved.clear() + approval._permanent_approved.update(permanent) + approval._session_approved.clear() + approval._session_approved.update(session) + class TestThreadLocalApprovalCallback: @@ -192,6 +219,238 @@ def _run_in_fresh_context(session_id: str, pw: str) -> str: assert runs[1] == ("acp-session-B", "", "bravo-secret") +class TestGatewayNotifierOwnership: + def test_old_cleanup_preserves_replacement_notifier_and_queue(self): + from tools import approval + + session_key = "replacement-notifier-session" + old_cb = lambda _data: None + replacement_cb = lambda _data: None + old_token = approval.register_gateway_notify(session_key, old_cb) + replacement_token = approval.register_gateway_notify(session_key, replacement_cb) + old_entry = approval._ApprovalEntry({}, owner_token=old_token) + replacement_entry = approval._ApprovalEntry({}, owner_token=replacement_token) + approval._gateway_queues[session_key] = [old_entry, replacement_entry] + + try: + approval.unregister_gateway_notify(session_key, old_token) + + assert approval._gateway_notify_cbs[session_key] is replacement_cb + assert approval._gateway_queues[session_key] == [replacement_entry] + assert old_entry.event.is_set() is True + assert replacement_entry.event.is_set() is False + finally: + approval.unregister_gateway_notify(session_key, replacement_token) + + def test_owner_change_between_enqueue_and_notify_drops_stale_prompt( + self, monkeypatch + ): + from tools import approval + + session_key = "stale-notify-before-user-prompt-session" + old_notified = [] + replacement_notified = [] + replacement = {} + + def old_cb(data): + old_notified.append(data) + + def replacement_cb(data): + replacement_notified.append(data) + + old_token = approval.register_gateway_notify(session_key, old_cb) + original_fire_hook = approval._fire_approval_hook + + def _replace_owner_on_pre_hook(event, **kwargs): + result = original_fire_hook(event, **kwargs) + if event == "pre_approval_request": + approval.unregister_gateway_notify(session_key, old_token) + replacement_token = approval.register_gateway_notify( + session_key, replacement_cb + ) + replacement_entry = approval._ApprovalEntry( + {"command": "replacement command", "pattern_key": "replacement"}, + owner_token=replacement_token, + ) + approval._gateway_queues.setdefault(session_key, []).append( + replacement_entry + ) + replacement["token"] = replacement_token + replacement["entry"] = replacement_entry + return result + + monkeypatch.setattr(approval, "_fire_approval_hook", _replace_owner_on_pre_hook) + + try: + result = approval._await_gateway_decision( + session_key, + old_cb, + { + "command": "OLD dangerous command", + "pattern_key": "old-danger", + "pattern_keys": ["old-danger"], + "description": "old approval prompt", + }, + owner_token=old_token, + ) + + assert result["notify_failed"] is True + assert result["stale_owner"] is True + assert old_notified == [] + assert replacement_notified == [] + assert approval._gateway_queues[session_key] == [replacement["entry"]] + + assert approval.resolve_gateway_approval(session_key, "once") == 1 + assert replacement["entry"].result == "once" + finally: + approval.clear_session(session_key) + if "token" in replacement: + approval.unregister_gateway_notify(session_key, replacement["token"]) + + def test_mcp_elicitation_cleanup_cannot_steal_replacement_approval( + self, monkeypatch + ): + from tools import approval + + session_key = "replacement-mcp-elicitation-session" + old_notified = threading.Event() + replacement_notified = threading.Event() + old_done = threading.Event() + replacement_done = threading.Event() + results = {} + + monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1") + monkeypatch.setattr(approval, "_get_approval_timeout", lambda: 1.0) + + def _request(name, done): + token = approval.set_current_session_key(session_key) + try: + results[name] = approval.request_elicitation_consent( + f"{name} request", + f"{name} description", + ) + finally: + approval.reset_current_session_key(token) + done.set() + + old_token = approval.register_gateway_notify( + session_key, lambda _data: old_notified.set() + ) + old_thread = threading.Thread( + target=_request, args=("old", old_done), daemon=True + ) + replacement_thread = None + replacement_token = None + try: + old_thread.start() + assert old_notified.wait(timeout=0.5) + + replacement_token = approval.register_gateway_notify( + session_key, lambda _data: replacement_notified.set() + ) + approval.unregister_gateway_notify(session_key, old_token) + + replacement_thread = threading.Thread( + target=_request, + args=("replacement", replacement_done), + daemon=True, + ) + replacement_thread.start() + assert replacement_notified.wait(timeout=0.5) + + assert approval.resolve_gateway_approval(session_key, "once") == 1 + assert replacement_done.wait(timeout=0.5) + assert old_done.wait(timeout=0.5) + assert results == {"old": "decline", "replacement": "accept"} + finally: + approval.clear_session(session_key) + approval.unregister_gateway_notify(session_key, replacement_token) + old_thread.join(timeout=1.0) + if replacement_thread is not None: + replacement_thread.join(timeout=1.0) + + def test_mcp_elicitation_rejects_owner_unregistered_before_enqueue( + self, monkeypatch + ): + from tools import approval + + session_key = "stale-before-enqueue-mcp-session" + old_snapshot_taken = threading.Event() + release_old_enqueue = threading.Event() + replacement_notified = threading.Event() + old_done = threading.Event() + replacement_done = threading.Event() + results = {} + + monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1") + monkeypatch.setattr(approval, "_get_approval_timeout", lambda: 1.0) + original_await = approval._await_gateway_decision + old_token = approval.register_gateway_notify(session_key, lambda _data: None) + + def _await_after_snapshot( + key, notify_cb, approval_data, *, surface="gateway", owner_token=None + ): + if owner_token is old_token: + old_snapshot_taken.set() + assert release_old_enqueue.wait(timeout=1.0) + return original_await( + key, + notify_cb, + approval_data, + surface=surface, + owner_token=owner_token, + ) + + monkeypatch.setattr(approval, "_await_gateway_decision", _await_after_snapshot) + + def _request(name, done): + token = approval.set_current_session_key(session_key) + try: + results[name] = approval.request_elicitation_consent( + f"{name} request", f"{name} description" + ) + finally: + approval.reset_current_session_key(token) + done.set() + + old_thread = threading.Thread( + target=_request, args=("old", old_done), daemon=True + ) + replacement_thread = None + replacement_token = None + try: + old_thread.start() + assert old_snapshot_taken.wait(timeout=0.5) + + approval.unregister_gateway_notify(session_key, old_token) + replacement_token = approval.register_gateway_notify( + session_key, lambda _data: replacement_notified.set() + ) + release_old_enqueue.set() + + assert old_done.wait(timeout=0.5) + assert results["old"] == "decline" + + replacement_thread = threading.Thread( + target=_request, + args=("replacement", replacement_done), + daemon=True, + ) + replacement_thread.start() + assert replacement_notified.wait(timeout=0.5) + assert approval.resolve_gateway_approval(session_key, "once") == 1 + assert replacement_done.wait(timeout=0.5) + assert results == {"old": "decline", "replacement": "accept"} + finally: + release_old_enqueue.set() + approval.clear_session(session_key) + approval.unregister_gateway_notify(session_key, replacement_token) + old_thread.join(timeout=1.0) + if replacement_thread is not None: + replacement_thread.join(timeout=1.0) + + + class TestAcpExecAskGate: """GHSA-96vc-wcxf-jjff: ACP's _run_agent must set HERMES_INTERACTIVE so that tools.approval.check_all_command_guards takes the CLI-interactive diff --git a/tests/gateway/test_approve_deny_commands.py b/tests/gateway/test_approve_deny_commands.py index 1fbfc6601c06..8a86ca4f1426 100644 --- a/tests/gateway/test_approve_deny_commands.py +++ b/tests/gateway/test_approve_deny_commands.py @@ -71,6 +71,7 @@ def _clear_approval_state(): from tools import approval as mod mod._gateway_queues.clear() mod._gateway_notify_cbs.clear() + mod._gateway_notify_tokens.clear() mod._session_approved.clear() mod._permanent_approved.clear() mod._pending.clear() @@ -468,19 +469,23 @@ def agent_thread(): os.environ.pop("HERMES_SESSION_KEY", None) reset_current_session_key(token) - t = threading.Thread(target=agent_thread) - t.start() + with patch( + "tools.tirith_security.check_command_security", + return_value={"action": "allow", "findings": [], "summary": ""}, + ): + t = threading.Thread(target=agent_thread) + t.start() - for _ in range(50): - if notified: - break - time.sleep(0.05) + for _ in range(200): + if notified: + break + time.sleep(0.05) - assert len(notified) == 1 - assert "rm -rf /important" in notified[0]["command"] + assert len(notified) == 1 + assert "rm -rf /important" in notified[0]["command"] - resolve_gateway_approval(session_key, "once") - t.join(timeout=5) + resolve_gateway_approval(session_key, "once") + t.join(timeout=5) assert result_holder[0] is not None assert result_holder[0]["approved"] is True @@ -746,6 +751,9 @@ def setup_method(self): os.environ.pop("HERMES_SESSION_KEY", None) def teardown_method(self): + from gateway.session_context import reset_session_vars + + reset_session_vars() os.environ.pop("HERMES_SESSION_KEY", None) def test_contextvar_wins_over_clobbered_environ(self): diff --git a/tests/gateway/test_goal_verdict_send.py b/tests/gateway/test_goal_verdict_send.py index 535dbe555427..313221b00457 100644 --- a/tests/gateway/test_goal_verdict_send.py +++ b/tests/gateway/test_goal_verdict_send.py @@ -10,6 +10,9 @@ from __future__ import annotations import asyncio +import concurrent.futures +import threading +import time from datetime import datetime from pathlib import Path from unittest.mock import MagicMock, patch @@ -219,3 +222,271 @@ def __init__(self): final_response="whatever", ) await asyncio.sleep(0.05) + +@pytest.mark.asyncio +async def test_replaced_goal_generation_remains_in_shutdown_drain_until_done( + hermes_home, +): + runner, _adapter, session_entry, _src = _make_runner_with_adapter() + session_key = session_entry.session_key + + predecessor = runner._begin_goal_run_control(session_key, 1) + replacement = runner._begin_goal_run_control(session_key, 2) + + assert predecessor["cancel"].is_set() is True + assert predecessor["done"].is_set() is False + runner._finish_goal_run_control(session_key, replacement) + + assert runner._active_goal_run_count() == 1 + active_agents, timed_out = await runner._drain_active_agents(0.01) + assert active_agents == {} + assert timed_out is True + + runner._finish_goal_run_control(session_key, predecessor) + assert runner._active_goal_run_count() == 0 + + +@pytest.mark.asyncio +async def test_goal_control_remains_owned_through_post_evaluation_notice( + hermes_home, monkeypatch +): + runner, adapter, session_entry, src = _make_runner_with_adapter() + + from hermes_cli.goals import GoalManager + + GoalManager(session_entry.session_id).set("continue after the status notice") + notice_started = asyncio.Event() + release_notice = asyncio.Event() + + async def _blocked_notice(_source, _message, **_kwargs): + notice_started.set() + await release_notice.wait() + + monkeypatch.setattr( + runner, "_defer_goal_status_notice_after_delivery", _blocked_notice + ) + task = None + try: + with patch( + "hermes_cli.goals.judge_goal", + return_value=("continue", "needs another turn", False, None), + ): + task = asyncio.create_task( + runner._post_turn_goal_continuation( + session_entry=session_entry, + source=src, + final_response="partial result", + ) + ) + await asyncio.wait_for(notice_started.wait(), timeout=1.0) + assert runner._active_goal_run_count() == 1 + + runner._cancel_all_goal_runs() + release_notice.set() + await asyncio.wait_for(task, timeout=1.0) + finally: + release_notice.set() + runner._cancel_all_goal_runs() + if task is not None and not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + assert runner._active_goal_run_count() == 0 + assert adapter._pending_messages == {} + + +@pytest.mark.asyncio +async def test_shutdown_removes_generation_owned_post_delivery_goal_notice( + hermes_home, +): + runner, adapter, session_entry, src = _make_runner_with_adapter() + + from gateway.platforms.base import BasePlatformAdapter + from hermes_cli.goals import GoalManager + + adapter._post_delivery_callbacks = {} + adapter.register_post_delivery_callback = ( + BasePlatformAdapter.register_post_delivery_callback.__get__(adapter) + ) + adapter.pop_post_delivery_callback = ( + BasePlatformAdapter.pop_post_delivery_callback.__get__(adapter) + ) + GoalManager(session_entry.session_id).set("finish without a stale notice") + + with patch( + "hermes_cli.goals.judge_goal", + return_value=("done", "complete", False, None), + ): + await runner._post_turn_goal_continuation( + session_entry=session_entry, + source=src, + final_response="finished", + ) + + session_key = session_entry.session_key + assert runner._active_goal_run_count() == 1 + assert session_key in adapter._post_delivery_callbacks + + runner._cancel_all_goal_runs() + + assert runner._active_goal_run_count() == 0 + assert adapter.pop_post_delivery_callback(session_key, generation=0) is None + assert adapter.sends == [] + + +@pytest.mark.asyncio +async def test_post_delivery_goal_notice_retires_control_only_after_callback( + hermes_home, +): + runner, adapter, session_entry, src = _make_runner_with_adapter() + + from gateway.platforms.base import BasePlatformAdapter + from hermes_cli.goals import GoalManager + + adapter._post_delivery_callbacks = {} + adapter.register_post_delivery_callback = ( + BasePlatformAdapter.register_post_delivery_callback.__get__(adapter) + ) + adapter.pop_post_delivery_callback = ( + BasePlatformAdapter.pop_post_delivery_callback.__get__(adapter) + ) + GoalManager(session_entry.session_id).set("deliver before retiring ownership") + + with patch( + "hermes_cli.goals.judge_goal", + return_value=("done", "complete", False, None), + ): + await runner._post_turn_goal_continuation( + session_entry=session_entry, + source=src, + final_response="finished", + ) + + callback = adapter.pop_post_delivery_callback( + session_entry.session_key, + generation=0, + ) + assert callable(callback) + assert runner._active_goal_run_count() == 1 + + await callback() + + assert runner._active_goal_run_count() == 0 + assert len(adapter.sends) == 1 + assert "Goal achieved" in adapter.sends[0]["content"] + + +@pytest.mark.asyncio +async def test_cancelled_goal_notice_preserves_foreign_post_delivery_callback( + hermes_home, +): + runner, adapter, session_entry, src = _make_runner_with_adapter() + + from gateway.platforms.base import BasePlatformAdapter + from hermes_cli.goals import GoalManager + + adapter._post_delivery_callbacks = {} + adapter.register_post_delivery_callback = ( + BasePlatformAdapter.register_post_delivery_callback.__get__(adapter) + ) + adapter.pop_post_delivery_callback = ( + BasePlatformAdapter.pop_post_delivery_callback.__get__(adapter) + ) + session_key = session_entry.session_key + foreign_deliveries = [] + + def _foreign_callback(): + foreign_deliveries.append("foreign-delivered") + + adapter.register_post_delivery_callback( + session_key, + _foreign_callback, + generation=0, + ) + GoalManager(session_entry.session_id).set("finish without cancelling unrelated callbacks") + + with patch( + "hermes_cli.goals.judge_goal", + return_value=("done", "complete", False, None), + ): + await runner._post_turn_goal_continuation( + session_entry=session_entry, + source=src, + final_response="finished", + ) + + assert runner._active_goal_run_count() == 1 + + runner._cancel_all_goal_runs() + callback = adapter.pop_post_delivery_callback(session_key, generation=0) + assert callable(callback) + result = callback() + if asyncio.iscoroutine(result): + await result + + assert foreign_deliveries == ["foreign-delivered"] + assert adapter.sends == [] + assert runner._active_goal_run_count() == 0 + + +@pytest.mark.asyncio +async def test_post_evaluation_cancellation_does_not_interrupt_reused_worker( + hermes_home, monkeypatch +): + runner, _adapter, session_entry, src = _make_runner_with_adapter() + + from hermes_cli.goals import GoalManager + from tools.interrupt import clear_current_thread_interrupt, is_interrupted + + GoalManager(session_entry.session_id).set("retain lifecycle without thread ownership") + runner._executor = concurrent.futures.ThreadPoolExecutor(max_workers=1) + notice_started = asyncio.Event() + release_notice = asyncio.Event() + goal_worker_threads = [] + + async def _blocked_notice(_source, _message, **_kwargs): + notice_started.set() + await release_notice.wait() + + def _observe_reused_worker(): + observed = (threading.get_ident(), is_interrupted()) + clear_current_thread_interrupt() + return observed + + def _judge_goal(*_args, **_kwargs): + goal_worker_threads.append(threading.get_ident()) + return ("continue", "needs another turn", False, None) + + monkeypatch.setattr( + runner, "_defer_goal_status_notice_after_delivery", _blocked_notice + ) + task = None + try: + with patch("hermes_cli.goals.judge_goal", side_effect=_judge_goal): + task = asyncio.create_task( + runner._post_turn_goal_continuation( + session_entry=session_entry, + source=src, + final_response="partial result", + ) + ) + await asyncio.wait_for(notice_started.wait(), timeout=1.0) + control = next(iter(runner._goal_run_controls[session_entry.session_key])) + assert goal_worker_threads + completed_thread_id = goal_worker_threads[0] + assert control["executor_done"].is_set() is True + assert control["thread_id"] is None + + runner._cancel_all_goal_runs() + reused_thread_id, observed_interrupt = ( + await runner._run_in_executor_with_context(_observe_reused_worker) + ) + assert reused_thread_id == completed_thread_id + assert observed_interrupt is False + finally: + release_notice.set() + runner._cancel_all_goal_runs() + if task is not None and not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + runner._executor.shutdown(wait=True, cancel_futures=True) diff --git a/tests/gateway/test_issue401_lifecycle_replay_authority.py b/tests/gateway/test_issue401_lifecycle_replay_authority.py new file mode 100644 index 000000000000..584444db726e --- /dev/null +++ b/tests/gateway/test_issue401_lifecycle_replay_authority.py @@ -0,0 +1,245 @@ +"""Issue #401/#475 lifecycle replay and authority invariants.""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, patch + +import asyncio +import pytest + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import BasePlatformAdapter, MessageEvent, SendResult, SessionSource + + +class _Adapter(BasePlatformAdapter): + async def connect(self, *, is_reconnect: bool = False) -> bool: + return True + + async def disconnect(self) -> None: + return None + + async def send(self, chat_id: str, content: str, reply_to=None, metadata=None): + return SendResult(success=True) + + async def get_chat_info(self, chat_id: str): + return {} + + +def _adapter() -> _Adapter: + return _Adapter(PlatformConfig(enabled=True, token="test-token"), Platform.TELEGRAM) + + +def test_stale_generation_replay_cannot_clear_current_post_delivery_owner(): + adapter = _adapter() + session_key = "telegram:chat:thread" + old_token = object() + current_token = object() + + def old_callback(): + return "old" + + def current_callback(): + return "current" + + assert ( + adapter.register_post_delivery_callback( + session_key, old_callback, generation=1, owner_token=old_token + ) + is old_token + ) + assert ( + adapter.register_post_delivery_callback( + session_key, current_callback, generation=2, owner_token=current_token + ) + is current_token + ) + + # Registering the newer generation retires the older callback. A replay from + # generation 1 must be a no-op and must not remove the current owner slot. + assert ( + adapter.pop_post_delivery_callback( + session_key, generation=1, owner_token=old_token + ) + is None + ) + assert adapter._post_delivery_callbacks[session_key] == [ + { + "generation": 2, + "callback": current_callback, + "owner_token": current_token, + } + ] + + # Same generation but wrong owner is also a no-op. + assert ( + adapter.pop_post_delivery_callback( + session_key, generation=2, owner_token=old_token + ) + is None + ) + assert adapter._post_delivery_callbacks[session_key][0]["callback"] is current_callback + + callback = adapter.pop_post_delivery_callback( + session_key, generation=2, owner_token=current_token + ) + assert callback is current_callback + assert callable(callback) + assert callback() == "current" + assert session_key not in adapter._post_delivery_callbacks + + +def test_duplicate_lower_generation_lifecycle_replay_is_noop(): + adapter = _adapter() + session_key = "telegram:chat:thread" + current_token = object() + + def current_callback(): + return "current" + + def stale_callback(): + return "stale" + + adapter.register_post_delivery_callback( + session_key, current_callback, generation=3, owner_token=current_token + ) + + assert ( + adapter.register_post_delivery_callback( + session_key, stale_callback, generation=2, owner_token=object() + ) + is None + ) + assert adapter._post_delivery_callbacks[session_key] == [ + { + "generation": 3, + "callback": current_callback, + "owner_token": current_token, + } + ] + + +def test_stale_lower_goal_generation_start_cannot_cancel_newer_active_control(): + from gateway.run import GatewayRunner + + runner = object.__new__(GatewayRunner) + session_key = "telegram:chat:thread" + runner._session_run_generation = {session_key: 3} + + current = runner._begin_goal_run_control(session_key, 3) + stale = runner._begin_goal_run_control(session_key, 2) + + assert stale["cancel"].is_set() is True + assert stale["done"].is_set() is True + assert current["cancel"].is_set() is False + assert current["done"].is_set() is False + assert runner._goal_run_controls[session_key] == [current] + assert runner._active_goal_run_count() == 1 + + runner._finish_goal_run_control(session_key, current) + assert runner._active_goal_run_count() == 0 + + +@pytest.mark.asyncio +async def test_hanging_post_delivery_peer_cannot_block_goal_owner_finalizer(monkeypatch): + """A stuck peer callback must not strand a generation-owned goal run.""" + from gateway import platforms as gateway_platforms + from gateway.run import GatewayRunner + + adapter = _adapter() + adapter.config.typing_indicator = False + adapter._send_with_retry = AsyncMock(return_value=SendResult(success=True)) + adapter._stop_typing_refresh = AsyncMock() + adapter._flush_text_debounce_now = AsyncMock() + adapter._run_processing_hook = AsyncMock() + async def _handler(_event): + return None + + adapter.set_message_handler(_handler) + monkeypatch.setattr( + gateway_platforms.base, + "_POST_DELIVERY_CALLBACK_TIMEOUT_SECONDS", + 0.05, + ) + + runner = object.__new__(GatewayRunner) + session_key = "telegram:chat:thread" + generation = 7 + runner._session_run_generation = {session_key: generation} + control = runner._begin_goal_run_control(session_key, generation) + + entered: list[str] = [] + sends: list[str] = [] + + async def blocker_callback(): + entered.append("blocker-enter") + await asyncio.Event().wait() + + async def goal_callback(): + entered.append("goal-enter") + try: + return None + finally: + runner._finish_goal_run_control(session_key, control) + + adapter.register_post_delivery_callback( + session_key, + blocker_callback, + generation=generation, + owner_token=object(), + ) + adapter.register_post_delivery_callback( + session_key, + goal_callback, + generation=generation, + owner_token=object(), + ) + + interrupt_event = asyncio.Event() + setattr(interrupt_event, "_hermes_run_generation", generation) + adapter._active_sessions[session_key] = interrupt_event + adapter._session_tasks[session_key] = asyncio.current_task() + + event = MessageEvent( + text="done", + source=SessionSource( + platform=Platform.TELEGRAM, + chat_id="chat", + thread_id="thread", + user_id="user", + ), + ) + + await adapter._process_message_background(event, session_key) + + assert "blocker-enter" in entered + assert "goal-enter" in entered + assert adapter._post_delivery_callbacks == {} + assert sends == [] + adapter._send_with_retry.assert_not_called() + assert runner._active_goal_run_count() == 0 + + +@pytest.mark.parametrize("max_turns", [7, "9"]) +def test_multiplex_runtime_env_reload_preserves_auth_authority_without_dotenv( + tmp_path, monkeypatch, max_turns +): + from gateway import run as gateway_run + + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + (hermes_home / "config.yaml").write_text( + f"agent:\n max_turns: {max_turns}\n", encoding="utf-8" + ) + monkeypatch.setenv("HERMES_MAX_ITERATIONS", "stale-dotenv-value") + monkeypatch.setattr(gateway_run, "_hermes_home", hermes_home) + + def _forbidden_dotenv_reload(*args, **kwargs): + raise AssertionError("multiplex gateway must not reload live .env secrets") + + with ( + patch("agent.secret_scope.is_multiplex_active", return_value=True), + patch.object(gateway_run, "load_hermes_dotenv", _forbidden_dotenv_reload), + ): + gateway_run._reload_runtime_env_preserving_config_authority() + + assert gateway_run.os.environ["HERMES_MAX_ITERATIONS"] == str(max_turns) diff --git a/tests/gateway/test_plaintext_approval_routing.py b/tests/gateway/test_plaintext_approval_routing.py index 9f07be64808a..7f8333bed9a9 100644 --- a/tests/gateway/test_plaintext_approval_routing.py +++ b/tests/gateway/test_plaintext_approval_routing.py @@ -46,6 +46,7 @@ def _clear_approval_state(): from tools import approval as mod mod._gateway_queues.clear() mod._gateway_notify_cbs.clear() + mod._gateway_notify_tokens.clear() mod._session_approved.clear() mod._permanent_approved.clear() mod._pending.clear() diff --git a/tests/gateway/test_session_boundary_security_state.py b/tests/gateway/test_session_boundary_security_state.py index b00ae1d96c99..f7d2bf6e40c3 100644 --- a/tests/gateway/test_session_boundary_security_state.py +++ b/tests/gateway/test_session_boundary_security_state.py @@ -24,6 +24,7 @@ def _clear_approval_state(): approval_mod._gateway_queues.clear() approval_mod._gateway_notify_cbs.clear() + approval_mod._gateway_notify_tokens.clear() approval_mod._session_approved.clear() approval_mod._session_yolo.clear() approval_mod._permanent_approved.clear() @@ -32,6 +33,7 @@ def _clear_approval_state(): yield approval_mod._gateway_queues.clear() approval_mod._gateway_notify_cbs.clear() + approval_mod._gateway_notify_tokens.clear() approval_mod._session_approved.clear() approval_mod._session_yolo.clear() approval_mod._permanent_approved.clear() diff --git a/tests/gateway/test_status_command.py b/tests/gateway/test_status_command.py index cadeb9ca7068..d03ad1f53c6b 100644 --- a/tests/gateway/test_status_command.py +++ b/tests/gateway/test_status_command.py @@ -734,4 +734,6 @@ async def fake_handler(event): # "newer" anyway. assert fired == [] assert session_key in adapter._post_delivery_callbacks - assert adapter._post_delivery_callbacks[session_key][0] == 2 + callback_entry = adapter._post_delivery_callbacks[session_key][0] + assert callback_entry["generation"] == 2 + assert callable(callback_entry["callback"]) diff --git a/tests/gateway/test_stream_consumer_fresh_final.py b/tests/gateway/test_stream_consumer_fresh_final.py index f8270cfd86dc..db40203bd6cb 100644 --- a/tests/gateway/test_stream_consumer_fresh_final.py +++ b/tests/gateway/test_stream_consumer_fresh_final.py @@ -11,6 +11,7 @@ from __future__ import annotations import asyncio +import time from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock @@ -53,7 +54,7 @@ async def test_disabled_by_default_still_edits_in_place(self): ) await consumer._send_or_edit("hello") # Pretend the preview has been visible for a long time. - consumer._message_created_ts = 0.0 # far in the past + consumer._message_created_ts = time.monotonic() - 61.0 # far in the past await consumer._send_or_edit("hello world", finalize=True) # Should edit, not send a fresh message. assert adapter.send.call_count == 1 # only the initial send @@ -89,7 +90,7 @@ async def test_long_lived_preview_sends_fresh_final(self): ) await consumer._send_or_edit("hello") # Force the preview to look stale (visible for > 60s). - consumer._message_created_ts = 0.0 # zero = ~uptime seconds old + consumer._message_created_ts = time.monotonic() - 61.0 await consumer._send_or_edit("hello world", finalize=True) # Fresh send happened; no edit of the old preview. assert adapter.send.call_count == 2 @@ -114,7 +115,7 @@ async def test_fresh_final_without_delete_support_is_best_effort(self): config=StreamConsumerConfig(fresh_final_after_seconds=60.0), ) await consumer._send_or_edit("hello") - consumer._message_created_ts = 0.0 + consumer._message_created_ts = time.monotonic() - 61.0 await consumer._send_or_edit("hello world", finalize=True) assert adapter.send.call_count == 2 adapter.edit_message.assert_not_called() @@ -135,7 +136,7 @@ async def test_fresh_final_fallback_to_edit_on_send_failure(self): config=StreamConsumerConfig(fresh_final_after_seconds=60.0), ) await consumer._send_or_edit("hello") - consumer._message_created_ts = 0.0 + consumer._message_created_ts = time.monotonic() - 61.0 ok = await consumer._send_or_edit("hello world", finalize=True) # Fresh send was attempted and failed → edit happened instead. assert adapter.send.call_count == 2 @@ -152,7 +153,7 @@ async def test_only_finalize_triggers_fresh_final(self): config=StreamConsumerConfig(fresh_final_after_seconds=60.0), ) await consumer._send_or_edit("hello") - consumer._message_created_ts = 0.0 # stale + consumer._message_created_ts = time.monotonic() - 61.0 # stale await consumer._send_or_edit("hello partial") # no finalize assert adapter.send.call_count == 1 adapter.edit_message.assert_called_once() @@ -454,7 +455,7 @@ async def test_cancel_with_fresh_final_enabled_delivers_and_flags_via_handler(se consumer.on_delta("Reply with **bold** and `code` markers.") task = asyncio.create_task(consumer.run()) await asyncio.sleep(0.05) - consumer._message_created_ts = 0.0 # force the preview stale + consumer._message_created_ts = time.monotonic() - 61.0 # force the preview stale task.cancel() await asyncio.gather(task, return_exceptions=True) diff --git a/tests/tools/test_approval.py b/tests/tools/test_approval.py index e65a73333434..f5f829805095 100644 --- a/tests/tools/test_approval.py +++ b/tests/tools/test_approval.py @@ -365,6 +365,30 @@ def test_gateway_runner_binds_session_key_to_context_before_agent_run(self): assert "set_current_session_key" in called_names assert "reset_current_session_key" in called_names + def test_gateway_goal_continuation_enforces_deferred_session_authorization(self): + run_py = Path(__file__).resolve().parents[2] / "gateway" / "run.py" + module = ast.parse(run_py.read_text(encoding="utf-8")) + + functions = {} + for node in ast.walk(module): + if isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef)) and node.name in { + "_post_turn_goal_continuation", + "_run_agent", + }: + functions[node.name] = node + + assert "_post_turn_goal_continuation" in functions + assert "_run_agent" in functions + + for function_name in ("_post_turn_goal_continuation", "_run_agent"): + called_names = set() + for node in ast.walk(functions[function_name]): + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): + called_names.add(node.func.id) + + assert "set_deferred_command_session_authorization_required" in called_names + assert "reset_deferred_command_session_authorization_required" in called_names + @@ -2146,6 +2170,7 @@ def setup_method(self): from tools import approval as mod mod._gateway_queues.clear() mod._gateway_notify_cbs.clear() + mod._gateway_notify_tokens.clear() mod._session_approved.clear() mod._permanent_approved.clear() mod._pending.clear() @@ -2169,6 +2194,7 @@ def teardown_method(self): from tools import approval as mod mod._gateway_queues.clear() mod._gateway_notify_cbs.clear() + mod._gateway_notify_tokens.clear() for k, v in self._saved_env.items(): if v is None: os.environ.pop(k, None) diff --git a/tests/tools/test_approval_heartbeat.py b/tests/tools/test_approval_heartbeat.py index d8531403ec87..48407cdf1cc4 100644 --- a/tests/tools/test_approval_heartbeat.py +++ b/tests/tools/test_approval_heartbeat.py @@ -19,6 +19,7 @@ def _clear_approval_state(): from tools import approval as mod mod._gateway_queues.clear() mod._gateway_notify_cbs.clear() + mod._gateway_notify_tokens.clear() mod._session_approved.clear() mod._permanent_approved.clear() mod._pending.clear() diff --git a/tests/tools/test_approval_interrupt.py b/tests/tools/test_approval_interrupt.py index b991afd80883..b7ed1275ce55 100644 --- a/tests/tools/test_approval_interrupt.py +++ b/tests/tools/test_approval_interrupt.py @@ -23,6 +23,7 @@ def _clear_approval_state(): from tools import approval as mod mod._gateway_queues.clear() mod._gateway_notify_cbs.clear() + mod._gateway_notify_tokens.clear() mod._session_approved.clear() mod._permanent_approved.clear() mod._pending.clear() diff --git a/tests/tools/test_command_guards.py b/tests/tools/test_command_guards.py index 9b8a93c30bf8..15cb7d38f2c8 100644 --- a/tests/tools/test_command_guards.py +++ b/tests/tools/test_command_guards.py @@ -7,12 +7,16 @@ import tools.approval as approval_module from tools.approval import ( + approve_permanent, approve_session, check_all_command_guards, check_dangerous_command, + detect_dangerous_command, is_approved, - set_current_session_key, reset_current_session_key, + reset_deferred_command_session_authorization_required, + set_current_session_key, + set_deferred_command_session_authorization_required, ) # Ensure the module is importable so we can patch it @@ -235,6 +239,334 @@ def test_dangerous_only_allows_permanent(self, mock_tirith): assert cb.call_args[1]["allow_permanent"] is True +# --------------------------------------------------------------------------- +# Deferred command gates require explicit session authorization +# --------------------------------------------------------------------------- + +class TestDeferredCommandAuthorization: + @staticmethod + def _dangerous_command_and_pattern(): + command = "rm -rf /tmp/test" + dangerous, pattern_key, _description = detect_dangerous_command(command) + assert dangerous is True + assert pattern_key + return command, pattern_key + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_once_choice_does_not_authorize_deferred_gate(self, mock_tirith): + os.environ["HERMES_INTERACTIVE"] = "1" + cb = MagicMock(return_value="once") + command, _pattern_key = self._dangerous_command_and_pattern() + + result = check_all_command_guards( + command, + "local", + approval_callback=cb, + require_explicit_authorization=True, + ) + + assert result["approved"] is False + assert result["outcome"] == "session_authorization_required" + assert result["user_consent"] is False + cb.assert_called_once() + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_session_choice_authorizes_deferred_gate(self, mock_tirith): + os.environ["HERMES_INTERACTIVE"] = "1" + cb = MagicMock(return_value="session") + command, _pattern_key = self._dangerous_command_and_pattern() + + result = check_all_command_guards( + command, + "local", + approval_callback=cb, + require_explicit_authorization=True, + ) + + assert result["approved"] is True + assert result["user_approved"] is True + cb.assert_called_once() + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_existing_pattern_session_approval_does_not_authorize_deferred_gate(self, mock_tirith): + session_key = "deferred-session-approval" + token = set_current_session_key(session_key) + try: + os.environ["HERMES_INTERACTIVE"] = "1" + cb = MagicMock(return_value="once") + command, pattern_key = self._dangerous_command_and_pattern() + approve_session(session_key, pattern_key) + + result = check_all_command_guards( + command, + "local", + approval_callback=cb, + require_explicit_authorization=True, + ) + finally: + reset_current_session_key(token) + + assert result["approved"] is False + assert result["outcome"] == "session_authorization_required" + cb.assert_called_once() + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_existing_exact_dangerous_command_session_approval_authorizes_deferred_gate(self, mock_tirith): + session_key = "deferred-exact-dangerous-command-session" + token = set_current_session_key(session_key) + try: + os.environ["HERMES_INTERACTIVE"] = "1" + cb = MagicMock(return_value="deny") + command, _pattern_key = self._dangerous_command_and_pattern() + approve_session( + session_key, + approval_module._deferred_command_session_key(command), + ) + + result = check_all_command_guards( + command, + "local", + approval_callback=cb, + require_explicit_authorization=True, + ) + finally: + reset_current_session_key(token) + + assert result["approved"] is True + cb.assert_not_called() + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_deferred_session_approval_is_exact_command_not_pattern_scope(self, mock_tirith): + session_key = "deferred-exact-command-not-pattern" + first_command = "rm -rf /tmp/one" + second_command = "rm -rf /tmp/two" + first_dangerous, first_pattern, _ = detect_dangerous_command(first_command) + second_dangerous, second_pattern, _ = detect_dangerous_command(second_command) + assert first_dangerous is True + assert second_dangerous is True + assert first_pattern == second_pattern + token = set_current_session_key(session_key) + try: + os.environ["HERMES_INTERACTIVE"] = "1" + cb1 = MagicMock(return_value="session") + first_result = check_all_command_guards( + first_command, + "local", + approval_callback=cb1, + require_explicit_authorization=True, + ) + cb2 = MagicMock(return_value="once") + second_result = check_all_command_guards( + second_command, + "local", + approval_callback=cb2, + require_explicit_authorization=True, + ) + finally: + reset_current_session_key(token) + + assert first_result["approved"] is True + cb1.assert_called_once() + assert second_result["approved"] is False + assert second_result["outcome"] == "session_authorization_required" + cb2.assert_called_once() + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_permanent_approval_does_not_authorize_deferred_gate(self, mock_tirith): + os.environ["HERMES_INTERACTIVE"] = "1" + cb = MagicMock(return_value="once") + command, pattern_key = self._dangerous_command_and_pattern() + approve_permanent(pattern_key) + + result = check_all_command_guards( + command, + "local", + approval_callback=cb, + require_explicit_authorization=True, + ) + + assert result["approved"] is False + assert result["outcome"] == "session_authorization_required" + cb.assert_called_once() + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_command_allowlist_glob_does_not_authorize_deferred_gate(self, mock_tirith): + os.environ["HERMES_INTERACTIVE"] = "1" + cb = MagicMock(return_value="once") + command, _pattern_key = self._dangerous_command_and_pattern() + approve_permanent("rm *") + + result = check_all_command_guards( + command, + "local", + approval_callback=cb, + require_explicit_authorization=True, + ) + + assert result["approved"] is False + assert result["outcome"] == "session_authorization_required" + cb.assert_called_once() + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_approval_mode_off_does_not_authorize_deferred_gate(self, mock_tirith, monkeypatch): + os.environ["HERMES_INTERACTIVE"] = "1" + cb = MagicMock(return_value="once") + command, _pattern_key = self._dangerous_command_and_pattern() + monkeypatch.setattr( + approval_module, + "_get_approval_config", + lambda: {"mode": "off"}, + ) + + result = check_all_command_guards( + command, + "local", + approval_callback=cb, + require_explicit_authorization=True, + ) + + assert result["approved"] is False + assert result["outcome"] == "session_authorization_required" + cb.assert_called_once() + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_smart_auto_approval_does_not_authorize_deferred_gate(self, mock_tirith, monkeypatch): + os.environ["HERMES_INTERACTIVE"] = "1" + cb = MagicMock(return_value="once") + command, _pattern_key = self._dangerous_command_and_pattern() + monkeypatch.setattr( + approval_module, + "_get_approval_config", + lambda: {"mode": "smart"}, + ) + smart = MagicMock(return_value="approve") + monkeypatch.setattr(approval_module, "_smart_approve", smart) + + result = check_all_command_guards( + command, + "local", + approval_callback=cb, + require_explicit_authorization=True, + ) + + assert result["approved"] is False + assert result["outcome"] == "session_authorization_required" + smart.assert_not_called() + cb.assert_called_once() + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_terminal_guard_context_requires_deferred_session_authorization(self, mock_tirith, monkeypatch): + import tools.terminal_tool as terminal_tool + + os.environ["HERMES_INTERACTIVE"] = "1" + cb = MagicMock(return_value="once") + command, _pattern_key = self._dangerous_command_and_pattern() + monkeypatch.setattr(terminal_tool, "_get_approval_callback", lambda: cb) + token = set_deferred_command_session_authorization_required(True) + try: + result = terminal_tool._check_all_guards(command, "local") + finally: + reset_deferred_command_session_authorization_required(token) + + assert result["approved"] is False + assert result["outcome"] == "session_authorization_required" + cb.assert_called_once() + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_scanner_allowed_command_requires_deferred_session_authorization(self, mock_tirith): + os.environ["HERMES_INTERACTIVE"] = "1" + cb = MagicMock(return_value="once") + + result = check_all_command_guards( + "echo hello", + "local", + approval_callback=cb, + require_explicit_authorization=True, + ) + + assert result["approved"] is False + assert result["outcome"] == "session_authorization_required" + assert "deferred goal terminal command" in result["description"] + cb.assert_called_once() + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_scanner_allowed_command_session_choice_authorizes_deferred_gate(self, mock_tirith): + os.environ["HERMES_INTERACTIVE"] = "1" + cb = MagicMock(return_value="session") + + result = check_all_command_guards( + "echo hello", + "local", + approval_callback=cb, + require_explicit_authorization=True, + ) + + assert result["approved"] is True + assert result["user_approved"] is True + cb.assert_called_once() + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_existing_exact_command_session_approval_authorizes_deferred_gate(self, mock_tirith): + session_key = "deferred-exact-command-session" + command = "echo hello" + token = set_current_session_key(session_key) + try: + os.environ["HERMES_INTERACTIVE"] = "1" + cb = MagicMock(return_value="deny") + approve_session( + session_key, + approval_module._deferred_command_session_key(command), + ) + + result = check_all_command_guards( + command, + "local", + approval_callback=cb, + require_explicit_authorization=True, + ) + finally: + reset_current_session_key(token) + + assert result["approved"] is True + cb.assert_not_called() + + @pytest.mark.parametrize("env_type", ["docker", "singularity", "modal", "daytona"]) + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_isolated_backend_skip_does_not_bypass_deferred_gate(self, mock_tirith, env_type): + os.environ["HERMES_INTERACTIVE"] = "1" + cb = MagicMock(return_value="once") + command, _pattern_key = self._dangerous_command_and_pattern() + + result = check_all_command_guards( + command, + env_type, + approval_callback=cb, + require_explicit_authorization=True, + ) + + assert result["approved"] is False + assert result["outcome"] == "session_authorization_required" + cb.assert_called_once() + + @pytest.mark.parametrize("env_type", ["docker", "singularity", "modal", "daytona"]) + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_isolated_backend_session_choice_authorizes_deferred_gate(self, mock_tirith, env_type): + os.environ["HERMES_INTERACTIVE"] = "1" + cb = MagicMock(return_value="session") + command, _pattern_key = self._dangerous_command_and_pattern() + + result = check_all_command_guards( + command, + env_type, + approval_callback=cb, + require_explicit_authorization=True, + ) + + assert result["approved"] is True + assert result["user_approved"] is True + cb.assert_called_once() + + # --------------------------------------------------------------------------- # Manual command_allowlist glob entries # --------------------------------------------------------------------------- diff --git a/tools/approval.py b/tools/approval.py index 836066eda982..14e6c6acd546 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -50,6 +50,12 @@ "approval_tool_call_id", default="", ) +_deferred_command_session_authorization_required: contextvars.ContextVar[bool] = ( + contextvars.ContextVar( + "deferred_command_session_authorization_required", + default=False, + ) +) # Interactive-CLI flag. Concurrent ACP sessions run on a shared # ThreadPoolExecutor (acp_adapter/server.py), so mutating the process-global @@ -152,6 +158,36 @@ def reset_current_observability_context( _approval_turn_id.reset(turn_token) +def set_deferred_command_session_authorization_required( + required: bool, +) -> contextvars.Token[bool]: + """Require session-scoped approval for deferred command execution. + + Goal continuations run after the original user turn. A one-operation or + permanent approval must not authorize that automatic deferred gate; the + user must choose the current session explicitly. + """ + return _deferred_command_session_authorization_required.set(bool(required)) + + +def reset_deferred_command_session_authorization_required( + token: contextvars.Token[bool], +) -> None: + """Restore deferred-command approval scope enforcement.""" + _deferred_command_session_authorization_required.reset(token) + + +def deferred_command_session_authorization_required() -> bool: + """Return whether this context requires session-scoped command approval.""" + return bool(_deferred_command_session_authorization_required.get()) + + +def _deferred_command_session_key(command: str) -> str: + """Session-only approval key for an exact deferred terminal command.""" + digest = hashlib.sha256(command.encode("utf-8", errors="surrogatepass")).hexdigest() + return f"deferred-terminal-command:{digest}" + + def get_current_session_key(default: str = "default") -> str: """Return the active session key, preferring context-local state. @@ -1444,9 +1480,9 @@ def detect_dangerous_command(command: str) -> tuple: class _ApprovalEntry: """One pending dangerous-command approval inside a gateway session.""" - __slots__ = ("event", "data", "result", "reason") + __slots__ = ("event", "data", "result", "reason", "owner_token") - def __init__(self, data: dict): + def __init__(self, data: dict, *, owner_token=None): self.event = threading.Event() self.data = data # command, description, pattern_keys, … self.result: Optional[str] = None # "once"|"session"|"always"|"deny" @@ -1454,13 +1490,18 @@ def __init__(self, data: dict): # (``/deny ``) so the agent can adapt instead of only # hearing "denied". Ported from qwibitai/nanoclaw#2832. self.reason: Optional[str] = None + # Opaque notifier registration that owns this wait. Replacement + # generations sharing a session key must not be signalled by stale + # cleanup from the prior generation. + self.owner_token = owner_token _gateway_queues: dict[str, list] = {} # session_key → [_ApprovalEntry, …] _gateway_notify_cbs: dict[str, object] = {} # session_key → callable(approval_data) +_gateway_notify_tokens: dict[str, object] = {} # session_key → opaque owner token -def register_gateway_notify(session_key: str, cb) -> None: +def register_gateway_notify(session_key: str, cb): """Register a per-session callback for sending approval requests to the user. The callback signature is ``cb(approval_data: dict) -> None`` where @@ -1468,19 +1509,38 @@ def register_gateway_notify(session_key: str, cb) -> None: ``pattern_keys``. The callback bridges sync→async (runs in the agent thread, must schedule the actual send on the event loop). """ + owner_token = object() with _lock: _gateway_notify_cbs[session_key] = cb + _gateway_notify_tokens[session_key] = owner_token + return owner_token -def unregister_gateway_notify(session_key: str) -> None: +def unregister_gateway_notify(session_key: str, owner_token=None) -> None: """Unregister the per-session gateway approval callback. - Signals ALL blocked threads for this session so they don't hang forever - (e.g. when the agent run finishes or is interrupted). + When ``owner_token`` is provided, remove only the callback and queue entries + created by that exact registration. This prevents an old run's ``finally`` + block from unregistering a replacement generation under the same session + key. Legacy callers that omit the token retain the historical unconditional + teardown behavior. """ with _lock: - _gateway_notify_cbs.pop(session_key, None) - entries = _gateway_queues.pop(session_key, []) + if owner_token is None: + _gateway_notify_cbs.pop(session_key, None) + _gateway_notify_tokens.pop(session_key, None) + entries = _gateway_queues.pop(session_key, []) + else: + if _gateway_notify_tokens.get(session_key) is owner_token: + _gateway_notify_cbs.pop(session_key, None) + _gateway_notify_tokens.pop(session_key, None) + queue = _gateway_queues.get(session_key, []) + entries = [entry for entry in queue if entry.owner_token is owner_token] + remaining = [entry for entry in queue if entry.owner_token is not owner_token] + if remaining: + _gateway_queues[session_key] = remaining + else: + _gateway_queues.pop(session_key, None) for entry in entries: entry.event.set() @@ -1598,6 +1658,14 @@ def is_approved(session_key: str, pattern_key: str) -> bool: return any(alias in session_approvals for alias in aliases) +def _is_session_approved(session_key: str, pattern_key: str) -> bool: + """Check only session-scoped approval, ignoring permanent allowlists.""" + aliases = _approval_key_aliases(pattern_key) + with _lock: + session_approvals = _session_approved.get(session_key, set()) + return any(alias in session_approvals for alias in aliases) + + def approve_permanent(pattern_key: str): """Add a pattern to the permanent allowlist.""" with _lock: @@ -2138,8 +2206,10 @@ def _run_approval_gate( # "approval_required" on this path — it gets a definitive # approved/BLOCKED outcome. notify_cb = None + notify_token = None with _lock: notify_cb = _gateway_notify_cbs.get(session_key) + notify_token = _gateway_notify_tokens.get(session_key) if notify_cb is not None: from agent.redact import redact_sensitive_text @@ -2151,7 +2221,11 @@ def _run_approval_gate( "allow_permanent": True, } decision = _await_gateway_decision( - session_key, notify_cb, approval_data, surface="gateway" + session_key, + notify_cb, + approval_data, + surface="gateway", + owner_token=notify_token, ) if decision.get("notify_failed"): return { @@ -2442,7 +2516,7 @@ def _format_tirith_description(tirith_result: dict) -> str: def _await_gateway_decision(session_key: str, notify_cb, approval_data: dict, - *, surface: str = "gateway") -> dict: + *, surface: str = "gateway", owner_token=None) -> dict: """Enqueue *approval_data*, notify the user, and block the calling agent thread until the request is resolved or the gateway approval timeout elapses — firing pre/post approval hooks and cleaning up the queue entry. @@ -2461,8 +2535,23 @@ def _await_gateway_decision(session_key: str, notify_cb, approval_data: dict, primary_key = approval_data.get("pattern_key", "") all_keys = approval_data.get("pattern_keys", [primary_key]) - entry = _ApprovalEntry(approval_data) + entry = _ApprovalEntry(approval_data, owner_token=owner_token) with _lock: + if owner_token is not None and ( + _gateway_notify_tokens.get(session_key) is not owner_token + or _gateway_notify_cbs.get(session_key) is not notify_cb + ): + logger.info( + "Gateway approval owner changed before queue insertion for %s; " + "failing the stale request closed", + session_key, + ) + return { + "resolved": False, + "choice": None, + "notify_failed": True, + "stale_owner": True, + } _gateway_queues.setdefault(session_key, []).append(entry) def _drop_entry() -> None: @@ -2485,6 +2574,28 @@ def _drop_entry() -> None: surface=surface, ) + if owner_token is not None: + with _lock: + queue = _gateway_queues.get(session_key, []) + stale_owner = ( + _gateway_notify_tokens.get(session_key) is not owner_token + or _gateway_notify_cbs.get(session_key) is not notify_cb + or entry not in queue + ) + if stale_owner: + logger.info( + "Gateway approval owner changed before notification for %s; " + "failing the stale request closed", + session_key, + ) + _drop_entry() + return { + "resolved": False, + "choice": None, + "notify_failed": True, + "stale_owner": True, + } + # Notify the user (bridges sync agent thread → async gateway) try: notify_cb(approval_data) @@ -2561,7 +2672,8 @@ def _drop_entry() -> None: def check_all_command_guards(command: str, env_type: str, approval_callback=None, - has_host_access: bool = False) -> dict: + has_host_access: bool = False, + require_explicit_authorization: bool | None = None) -> dict: """Run all pre-exec security checks and return a single approval decision. Gathers findings from tirith and dangerous-command detection, then @@ -2573,9 +2685,19 @@ def check_all_command_guards(command: str, env_type: str, such a session is no longer isolated, so it goes through the normal flow instead of the container fast-path. """ + if require_explicit_authorization is None: + require_explicit_authorization = ( + deferred_command_session_authorization_required() + ) + # Skip isolated container backends for both checks. Docker stops skipping - # once host paths are bind-mounted into the sandbox. - if _should_skip_container_guards(env_type, has_host_access=has_host_access): + # once host paths are bind-mounted into the sandbox. Deferred goal-command + # execution is different: the sandbox may reduce command risk, but it must + # not bypass the user's explicit session authorization for automatic work. + if ( + not require_explicit_authorization + and _should_skip_container_guards(env_type, has_host_access=has_host_access) + ): return {"approved": True, "message": None} # Hardline floor: unconditional block for catastrophic commands @@ -2610,10 +2732,20 @@ def check_all_command_guards(command: str, env_type: str, # --yolo or approvals.mode=off: bypass all approval prompts. # Gateway /yolo is session-scoped; CLI --yolo remains process-scoped. approval_mode = _get_approval_mode() - if _YOLO_MODE_FROZEN or is_current_session_yolo_enabled() or approval_mode == "off": + if ( + not require_explicit_authorization + and ( + _YOLO_MODE_FROZEN + or is_current_session_yolo_enabled() + or approval_mode == "off" + ) + ): return {"approved": True, "message": None} - if _command_matches_permanent_allowlist(command): + if ( + not require_explicit_authorization + and _command_matches_permanent_allowlist(command) + ): return {"approved": True, "message": None} is_cli = _is_interactive_cli() @@ -2622,7 +2754,7 @@ def check_all_command_guards(command: str, env_type: str, # Preserve the existing non-interactive behavior: outside CLI/gateway/ask # flows, we do not block on approvals and we skip external guard work. - if not is_cli and not is_gateway and not is_ask: + if not is_cli and not is_gateway and not is_ask and not require_explicit_authorization: # Cron sessions: respect cron_mode config if env_var_enabled("HERMES_CRON_SESSION"): if _get_cron_approval_mode() == "deny": @@ -2741,6 +2873,20 @@ def check_all_command_guards(command: str, env_type: str, warnings = [] # list of (pattern_key, description, is_tirith) session_key = get_current_session_key() + deferred_session_key = _deferred_command_session_key(command) + session_authorization_matched = False + + def _already_authorized(pattern_key_to_check: str) -> bool: + nonlocal session_authorization_matched + if require_explicit_authorization: + # Deferred autonomous goal execution requires an exact-command + # session grant. A broad dangerous-pattern or Tirith rule session + # approval must not authorize a different future terminal command. + approved = _is_session_approved(session_key, deferred_session_key) + if approved: + session_authorization_matched = True + return approved + return is_approved(session_key, pattern_key_to_check) # Tirith block/warn → approvable warning with rich findings. # Previously, tirith "block" was a hard block with no approval prompt. @@ -2748,16 +2894,37 @@ def check_all_command_guards(command: str, env_type: str, # inspect the explanation and approve if they understand the risk. if tirith_result["action"] in {"block", "warn"}: findings = tirith_result.get("findings") or [] - rule_id = findings[0].get("rule_id", "unknown") if findings else "unknown" - tirith_key = f"tirith:{rule_id}" - tirith_desc = _format_tirith_description(tirith_result) - if not is_approved(session_key, tirith_key): - warnings.append((tirith_key, tirith_desc, True)) + seen_tirith_keys = set() + for finding in findings or [{}]: + raw_rule_id = finding.get("rule_id") if isinstance(finding, dict) else None + rule_id = str(raw_rule_id).strip() if raw_rule_id is not None else "" + tirith_key = f"tirith:{rule_id or 'unknown'}" + if tirith_key in seen_tirith_keys: + continue + seen_tirith_keys.add(tirith_key) + if _already_authorized(tirith_key): + continue + finding_result = dict(tirith_result) + finding_result["findings"] = [finding] if findings else [] + warnings.append( + (tirith_key, _format_tirith_description(finding_result), True) + ) if is_dangerous: - if not is_approved(session_key, pattern_key): + if not _already_authorized(pattern_key): warnings.append((pattern_key, description, False)) + if require_explicit_authorization and not warnings and not session_authorization_matched: + if not _is_session_approved(session_key, deferred_session_key): + warnings.append( + ( + deferred_session_key, + "deferred goal terminal command execution requires " + "session-scoped approval for this exact command", + False, + ) + ) + # Nothing to warn about if not warnings: return {"approved": True, "message": None} @@ -2766,7 +2933,7 @@ def check_all_command_guards(command: str, env_type: str, # When approvals.mode=smart, ask the aux LLM before prompting the user. # Inspired by OpenAI Codex's Smart Approvals guardian subagent # (openai/codex#13860). - if approval_mode == "smart": + if approval_mode == "smart" and not require_explicit_authorization: combined_desc_for_llm = "; ".join(desc for _, desc, _ in warnings) verdict = _smart_approve(command, combined_desc_for_llm) if verdict == "approve": @@ -2802,8 +2969,10 @@ def check_all_command_guards(command: str, env_type: str, # gets the command output (approved) or a definitive "BLOCKED" message. if is_gateway or is_ask: notify_cb = None + notify_token = None with _lock: notify_cb = _gateway_notify_cbs.get(session_key) + notify_token = _gateway_notify_tokens.get(session_key) if notify_cb is not None: # --- Blocking gateway approval (queue-based) --- @@ -2825,10 +2994,15 @@ def check_all_command_guards(command: str, env_type: str, "description": redact_sensitive_text(combined_desc), # Mirror the CLI's allow_permanent gate: a tirith warning downgrades # "always" to session scope below, so the UI must not offer it. - "allow_permanent": not has_tirith, + "allow_permanent": not has_tirith and not require_explicit_authorization, + "allow_session": True, } decision = _await_gateway_decision( - session_key, notify_cb, approval_data, surface="gateway" + session_key, + notify_cb, + approval_data, + surface="gateway", + owner_token=notify_token, ) if decision.get("notify_failed"): return { @@ -2879,7 +3053,26 @@ def check_all_command_guards(command: str, env_type: str, "deny_reason": deny_reason, } + if require_explicit_authorization and choice != "session": + return { + "approved": False, + "message": ( + "BLOCKED: Deferred command execution requires an explicit " + "session-scoped approval. A one-operation or permanent " + "choice does not authorize an automatic goal gate." + ), + "pattern_key": primary_key, + "description": combined_desc, + "outcome": "session_authorization_required", + "user_consent": False, + } + # User approved — persist based on scope (same logic as CLI) + if require_explicit_authorization: + approve_session(session_key, deferred_session_key) + return {"approved": True, "message": None, + "user_approved": True, "description": combined_desc} + for key, _, is_tirith in warnings: if choice == "session" or (choice == "always" and is_tirith): approve_session(session_key, key) @@ -2930,7 +3123,7 @@ def check_all_command_guards(command: str, env_type: str, surface="cli", ) choice = prompt_dangerous_approval(command, combined_desc, - allow_permanent=not has_tirith, + allow_permanent=(not has_tirith and not require_explicit_authorization), approval_callback=approval_callback) _fire_approval_hook( "post_approval_response", @@ -2960,7 +3153,26 @@ def check_all_command_guards(command: str, env_type: str, "user_consent": False, } + if require_explicit_authorization and choice != "session": + return { + "approved": False, + "message": ( + "BLOCKED: Deferred command execution requires an explicit " + "session-scoped approval. A one-operation or permanent choice " + "does not authorize an automatic goal gate." + ), + "pattern_key": primary_key, + "description": combined_desc, + "outcome": "session_authorization_required", + "user_consent": False, + } + # Persist approval for each warning individually + if require_explicit_authorization: + approve_session(session_key, deferred_session_key) + return {"approved": True, "message": None, + "user_approved": True, "description": combined_desc} + for key, _, is_tirith in warnings: if choice == "session" or (choice == "always" and is_tirith): # tirith: session only (no permanent broad allowlisting) @@ -3093,8 +3305,10 @@ def check_execute_code_guard(code: str, env_type: str, # verdict == "escalate" → fall through to manual approval notify_cb = None + notify_token = None with _lock: notify_cb = _gateway_notify_cbs.get(session_key) + notify_token = _gateway_notify_tokens.get(session_key) if notify_cb is None: # No gateway callback registered (e.g. ask-mode without a notifier): @@ -3125,7 +3339,11 @@ def check_execute_code_guard(code: str, env_type: str, "description": display_description, } decision = _await_gateway_decision( - session_key, notify_cb, approval_data, surface="gateway" + session_key, + notify_cb, + approval_data, + surface="gateway", + owner_token=notify_token, ) if decision.get("notify_failed"): return { @@ -3210,6 +3428,7 @@ def request_elicitation_consent( if _is_gateway_approval_context(): with _lock: notify_cb = _gateway_notify_cbs.get(session_key) + notify_token = _gateway_notify_tokens.get(session_key) if notify_cb is None: logger.warning( "Elicitation requested in gateway session %s but no " @@ -3226,7 +3445,11 @@ def request_elicitation_consent( } try: decision = _await_gateway_decision( - session_key, notify_cb, approval_data, surface=surface, + session_key, + notify_cb, + approval_data, + surface=surface, + owner_token=notify_token, ) except Exception as exc: logger.error(