diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 0e0b87b21960..a5a9c1ae9a68 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -726,6 +726,11 @@ def run_codex_app_server_turn( "completed": False, "partial": True, "interrupted": _user_interrupted, + "turn_exit_reason": ( + "interrupted_by_user" + if _user_interrupted + else "local_processing_error(codex_app_server)" + ), **( {"interrupt_message": _interrupt_message} if _interrupt_message @@ -860,6 +865,17 @@ def run_codex_app_server_turn( except Exception: logger.debug("background review spawn raised", exc_info=True) + if _user_interrupted: + turn_exit_reason = "interrupted_by_user" + elif turn.interrupted: + turn_exit_reason = "interrupted_during_api_call" + elif turn.error is not None: + turn_exit_reason = "local_processing_error(codex_app_server)" + elif not str(turn.final_text or "").strip(): + turn_exit_reason = "empty_response_exhausted" + else: + turn_exit_reason = "text_response(finish_reason=stop)" + return { "final_response": turn.final_text, "messages": messages, @@ -867,6 +883,7 @@ def run_codex_app_server_turn( "completed": not turn.interrupted and turn.error is None, "partial": turn.interrupted or turn.error is not None, "interrupted": _user_interrupted, + "turn_exit_reason": turn_exit_reason, **( {"interrupt_message": _interrupt_message} if _interrupt_message diff --git a/gateway/hooks.py b/gateway/hooks.py index 1ea7faa32a14..c7e70097d65e 100644 --- a/gateway/hooks.py +++ b/gateway/hooks.py @@ -30,6 +30,32 @@ ``agent:end`` adds: response -- agent response text (truncated to 500 chars) + turn_exit_reason -- why the agent loop ended (truncated to 200 chars); + finalizer text is collapsed to a single line, + explicit stop/reset controls and mid-turn user correction + text without an explicit finalizer reason normalize to + "interrupted_by_user", + gateway system aborts and unclassified interrupts remain + distinct, and other missing or malformed reasons + normalize to "unknown" + api_call_count -- non-negative API iterations consumed by the turn; zero can + also mean not reported because proxy mode reports zero and + absent or malformed counts clamp to zero + stale -- boolean; true when the run was superseded and its output was + discarded, false when this is the current delivered turn + +``turn_exit_reason`` is an open vocabulary: specific agent-finalizer reason +semantics pass through after the delivered string is normalized. Gateway-owned +classes include ``interrupted_by_user``, ``unknown``, and the ``gateway_*`` / +``gateway_proxy_*`` families. + +A superseded run still emits ``agent:end`` with ``stale == True`` and an empty +``response`` before its output is discarded. Non-proxy runs whose prior result +was normal or missing a reason use ``gateway_stale_generation``; an otherwise +complete proxy run uses ``gateway_proxy_stale_generation``. More specific +abnormal reasons remain intact, so delivery-side handlers must gate on +``stale`` rather than a reason string and must not post a follow-up when it is +true. Handlers posting a follow-up into the same Telegram forum-topic should include ``message_thread_id=int(thread_id)`` when ``chat_type == "forum"`` diff --git a/gateway/run.py b/gateway/run.py index b97b74b928a1..7346168582b6 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2653,27 +2653,123 @@ def _dequeue_pending_event(adapter, session_key: str) -> MessageEvent | None: _INTERRUPT_REASON_SSE_DISCONNECT = "SSE client disconnected" _INTERRUPT_REASON_GATEWAY_SHUTDOWN = "Gateway shutting down" _INTERRUPT_REASON_GATEWAY_RESTART = "Gateway restarting" +_GATEWAY_AGENT_POLL_INTERVAL_SECONDS = 5.0 + +def _normalize_interrupt_message(message: object) -> str: + """Normalize an interrupt marker without trusting its input type.""" + try: + if not message: + return "" + return " ".join(str(message).strip().split()).lower() + except Exception: + return "" + _CONTROL_INTERRUPT_MESSAGES = frozenset( - { - _INTERRUPT_REASON_STOP.lower(), - _INTERRUPT_REASON_RESET.lower(), - _INTERRUPT_REASON_TIMEOUT.lower(), - _INTERRUPT_REASON_SSE_DISCONNECT.lower(), - _INTERRUPT_REASON_GATEWAY_SHUTDOWN.lower(), - _INTERRUPT_REASON_GATEWAY_RESTART.lower(), - } + _normalize_interrupt_message(message) + for message in ( + _INTERRUPT_REASON_STOP, + _INTERRUPT_REASON_RESET, + _INTERRUPT_REASON_TIMEOUT, + _INTERRUPT_REASON_SSE_DISCONNECT, + _INTERRUPT_REASON_GATEWAY_SHUTDOWN, + _INTERRUPT_REASON_GATEWAY_RESTART, + ) ) def _is_control_interrupt_message(message: Optional[str]) -> bool: """Return True when an interrupt message is internal control flow.""" - if not message: - return False - normalized = " ".join(str(message).strip().split()).lower() + normalized = _normalize_interrupt_message(message) return normalized in _CONTROL_INTERRUPT_MESSAGES +def _normalize_turn_exit_reason(reason: object) -> str: + """Collapse untrusted finalizer reason text to a single bounded-safe line.""" + try: + if reason is None: + return "" + collapsed = " ".join(str(reason).strip().split()) + return "".join(char for char in collapsed if char.isprintable()) + except Exception: + return "" + + +_SYSTEM_INTERRUPT_EXIT_REASONS = { + _normalize_interrupt_message(_INTERRUPT_REASON_TIMEOUT): + "gateway_inactivity_timeout", + _normalize_interrupt_message(_INTERRUPT_REASON_SSE_DISCONNECT): + "gateway_sse_disconnect", + _normalize_interrupt_message(_INTERRUPT_REASON_GATEWAY_SHUTDOWN): + "gateway_shutdown", + _normalize_interrupt_message(_INTERRUPT_REASON_GATEWAY_RESTART): + "gateway_restart", +} +_USER_INTERRUPT_MESSAGES = { + _normalize_interrupt_message(_INTERRUPT_REASON_STOP), + _normalize_interrupt_message(_INTERRUPT_REASON_RESET), +} + + +def _is_generic_agent_interrupt_exit_reason(reason: str) -> bool: + """Return True for agent-finalizer interruption classes and variants.""" + return reason.casefold().startswith("interrupt") + + +def _gateway_turn_exit_reason(agent_result: Dict[str, Any]) -> str: + """Return a stable reason for the ``agent:end`` hook. + + Normal agent results carry the finalizer's explicit reason. Early + interrupts can bypass that finalizer, so classify their gateway control + marker instead of collapsing system aborts into user stops. A non-control + message is user-originated correction text; only an interrupted turn with + no marker remains unclassified. + """ + reason = _normalize_turn_exit_reason(agent_result.get("turn_exit_reason")) + if reason and not _is_generic_agent_interrupt_exit_reason(reason): + return reason + + try: + interrupted = bool(agent_result.get("interrupted")) + except Exception: + interrupted = False + if not interrupted: + return reason or "unknown" + + normalized_message = _normalize_interrupt_message( + agent_result.get("interrupt_message") + ) + system_reason = _SYSTEM_INTERRUPT_EXIT_REASONS.get(normalized_message) + if system_reason: + return system_reason + if normalized_message in _USER_INTERRUPT_MESSAGES: + return "interrupted_by_user" + if reason: + return reason + if normalized_message and not _is_control_interrupt_message( + normalized_message + ): + return "interrupted_by_user" + return "gateway_interrupt_unclassified" + + +def _gateway_agent_end_metadata( + agent_result: Dict[str, Any], + *, + stale: bool = False, +) -> Dict[str, Any]: + """Return bounded, fail-safe ``agent:end`` termination metadata.""" + try: + api_call_count = max(0, int(agent_result.get("api_calls") or 0)) + except Exception: + api_call_count = 0 + return { + "turn_exit_reason": _gateway_turn_exit_reason(agent_result)[:200], + "api_call_count": api_call_count, + "stale": stale, + } + + def _skill_slug_from_frontmatter(skill_md: Path) -> tuple[str | None, str | None]: """Derive the /command slug and declared frontmatter name from a SKILL.md. @@ -14454,6 +14550,22 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g _quick_key or "?", run_generation, ) + stale_agent_end_metadata = _gateway_agent_end_metadata( + agent_result, + stale=True, + ) + stale_reason = stale_agent_end_metadata["turn_exit_reason"] + if ( + stale_reason == "unknown" + or stale_reason.startswith("text_response") + ): + stale_agent_end_metadata["turn_exit_reason"] = ( + "gateway_stale_generation" + ) + elif stale_reason == "gateway_proxy_response_complete": + stale_agent_end_metadata["turn_exit_reason"] = ( + "gateway_proxy_stale_generation" + ) _stale_adapter = self._adapter_for_source(source) if getattr(type(_stale_adapter), "pop_post_delivery_callback", None) is not None: _stale_adapter.pop_post_delivery_callback( @@ -14462,6 +14574,14 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g ) elif _stale_adapter and hasattr(_stale_adapter, "_post_delivery_callbacks"): _stale_adapter._post_delivery_callbacks.pop(_quick_key, None) + await self.hooks.emit( + "agent:end", + { + **hook_ctx, + "response": "", + **stale_agent_end_metadata, + }, + ) return None response = agent_result.get("final_response") or "" @@ -14648,10 +14768,16 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g if _footer_line and response and not agent_result.get("already_sent") and not _intentional_silence: response = f"{response}\n\n{_footer_line}" + # Retry/error-backoff interrupts can bypass turn_finalizer. Preserve + # explicit agent reasons, distinguish gateway system aborts from + # user stops, and make every otherwise-missing class "unknown". + agent_end_metadata = _gateway_agent_end_metadata(agent_result) + # Emit agent:end hook await self.hooks.emit("agent:end", { **hook_ctx, "response": (response or "")[:500], + **agent_end_metadata, }) # Check for pending process watchers (check_interval on background processes) @@ -20309,6 +20435,8 @@ async def _run_agent_via_proxy( "messages": [], "api_calls": 0, "tools": [], + "partial": False, + "turn_exit_reason": "gateway_proxy_dependency_missing", } proxy_url = self._get_proxy_url() @@ -20318,6 +20446,8 @@ async def _run_agent_via_proxy( "messages": [], "api_calls": 0, "tools": [], + "partial": False, + "turn_exit_reason": "gateway_proxy_not_configured", } proxy_key = os.getenv("GATEWAY_PROXY_KEY", "").strip() @@ -20448,6 +20578,7 @@ def _pause_typing_before_finalize( # Make the HTTP request with SSE streaming ----------------------- full_response = "" + proxy_partial = False _start = time.time() try: @@ -20469,6 +20600,8 @@ def _pause_typing_before_finalize( "messages": [], "api_calls": 0, "tools": [], + "partial": False, + "turn_exit_reason": "gateway_proxy_http_error", } # Parse SSE stream @@ -20488,6 +20621,9 @@ def _pause_typing_before_finalize( "history_offset": len(history), "session_id": session_id, "response_previewed": False, + "partial": False, + "turn_exit_reason": + "gateway_proxy_stale_generation", } text = chunk.decode("utf-8", errors="replace") buffer += text @@ -20529,8 +20665,11 @@ def _pause_typing_before_finalize( "messages": [], "api_calls": 0, "tools": [], + "partial": False, + "turn_exit_reason": "gateway_proxy_connection_error", } # Partial response — return what we got + proxy_partial = True finally: # Finalize stream consumer if _stream_consumer: @@ -20556,6 +20695,8 @@ def _pause_typing_before_finalize( "history_offset": len(history), "session_id": session_id, "response_previewed": False, + "partial": False, + "turn_exit_reason": "gateway_proxy_stale_generation", } logger.info( "proxy response: url=%s session=%s time=%.1fs response=%d chars", @@ -20573,6 +20714,19 @@ def _pause_typing_before_finalize( "history_offset": len(history), "session_id": session_id, "response_previewed": _stream_consumer is not None and bool(full_response), + # Preserve the proxy result's pre-REL-01 downstream control flow. + # The exit reason carries partial-stream evidence to the hook + # without activating unrelated ``result["partial"]`` consumers. + "partial": False, + "turn_exit_reason": ( + "gateway_proxy_partial_response" + if proxy_partial + else ( + "gateway_proxy_response_complete" + if full_response + else "gateway_proxy_empty_response" + ) + ), } # ------------------------------------------------------------------ @@ -21927,6 +22081,8 @@ def run_sync(): "messages": [], "api_calls": 0, "tools": [], + "failed": True, + "turn_exit_reason": "gateway_agent_runtime_resolution_failed", } pr = self._provider_routing @@ -23066,6 +23222,7 @@ def _approval_notify_sync(approval_data: dict) -> None: "final_response": final_response, "messages": result.get("messages", []), "api_calls": result.get("api_calls", 0), + "turn_exit_reason": result.get("turn_exit_reason"), "failed": result.get("failed", False), # Sibling of the non-empty-response return below (#64686): # the classifier's failure_reason must survive the @@ -23193,6 +23350,10 @@ def _title_failure_cb(task: str, exc: BaseException) -> None: "last_reasoning": result.get("last_reasoning"), "messages": result_holder[0].get("messages", []) if result_holder[0] else [], "api_calls": result_holder[0].get("api_calls", 0) if result_holder[0] else 0, + "turn_exit_reason": ( + result_holder[0].get("turn_exit_reason") + if result_holder[0] else None + ), "failed": result_holder[0].get("failed", False) if result_holder[0] else False, "failure_reason": ( result_holder[0].get("failure_reason") if result_holder[0] else None @@ -23503,7 +23664,7 @@ def _stream_confirmed_final_delivery( ) _inactivity_timeout = False - _POLL_INTERVAL = 5.0 + _POLL_INTERVAL = _GATEWAY_AGENT_POLL_INTERVAL_SECONDS if _agent_timeout is None: # Unlimited — still poll periodically for backup interrupt @@ -23691,6 +23852,7 @@ def _stream_confirmed_final_delivery( "tools": tools_holder[0] or [], "history_offset": 0, "failed": True, + "turn_exit_reason": "gateway_inactivity_timeout", } # Track fallback model state: if the agent switched to a diff --git a/tests/agent/test_codex_app_server_persist.py b/tests/agent/test_codex_app_server_persist.py index 85d4a5757f7e..e450cd45f580 100644 --- a/tests/agent/test_codex_app_server_persist.py +++ b/tests/agent/test_codex_app_server_persist.py @@ -72,6 +72,7 @@ def test_codex_success_flushes_and_reports_persisted(): effective_task_id="task-1", ) assert result["completed"] is True + assert result["turn_exit_reason"] == "text_response(finish_reason=stop)" # With the agent as sole persister, the gateway must SKIP its DB write. assert result["agent_persisted"] is True @@ -100,10 +101,86 @@ def clear_interrupt(): assert result["interrupted"] is True assert result["interrupt_message"] == "new correction" + assert result["turn_exit_reason"] == "interrupted_by_user" agent.clear_interrupt.assert_called_once_with() assert agent._interrupt_requested is False +def test_codex_non_user_interrupt_is_actionable(): + agent = _make_agent(session_db=None) + turn = _make_turn() + turn.interrupted = True + turn.final_text = "" + agent._codex_session.run_turn.return_value = turn + agent._interrupt_requested = False + + result = run_codex_app_server_turn( + agent, + user_message="hello", + original_user_message="hello", + messages=[{"role": "user", "content": "hello"}], + effective_task_id="task-1", + ) + + assert result["interrupted"] is False + assert result["turn_exit_reason"] == "interrupted_during_api_call" + + +def test_codex_empty_success_is_actionable(): + agent = _make_agent(session_db=None) + turn = _make_turn() + turn.final_text = " " + turn.projected_messages = [] + agent._codex_session.run_turn.return_value = turn + + result = run_codex_app_server_turn( + agent, + user_message="hello", + original_user_message="hello", + messages=[{"role": "user", "content": "hello"}], + effective_task_id="task-1", + ) + + assert result["completed"] is True + assert result["turn_exit_reason"] == "empty_response_exhausted" + + +def test_codex_turn_error_is_actionable(): + agent = _make_agent(session_db=None) + turn = _make_turn() + turn.error = "synthetic failure" + turn.final_text = "" + agent._codex_session.run_turn.return_value = turn + + result = run_codex_app_server_turn( + agent, + user_message="hello", + original_user_message="hello", + messages=[{"role": "user", "content": "hello"}], + effective_task_id="task-1", + ) + + assert result["completed"] is False + assert result["turn_exit_reason"] == "local_processing_error(codex_app_server)" + + +def test_codex_runtime_exception_is_actionable(): + agent = _make_agent(session_db=None) + agent._codex_session.run_turn.side_effect = RuntimeError("synthetic crash") + agent._interrupt_requested = False + + result = run_codex_app_server_turn( + agent, + user_message="hello", + original_user_message="hello", + messages=[{"role": "user", "content": "hello"}], + effective_task_id="task-1", + ) + + assert result["completed"] is False + assert result["turn_exit_reason"] == "local_processing_error(codex_app_server)" + + def test_codex_turn_persists_each_message_exactly_once(): """The user turn (flushed at turn start) must not be duplicated; the projected assistant message must land once. Uses a real SessionDB and the diff --git a/tests/gateway/test_agent_end_hook_metadata.py b/tests/gateway/test_agent_end_hook_metadata.py new file mode 100644 index 000000000000..1090d67ab54b --- /dev/null +++ b/tests/gateway/test_agent_end_hook_metadata.py @@ -0,0 +1,815 @@ +"""Gateway agent:end hook termination metadata.""" + +import sys +import threading +import types +from datetime import datetime +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +import gateway.run as gateway_run +from gateway.config import GatewayConfig, Platform +from gateway.platforms.base import MessageEvent +from gateway.session import SessionEntry, SessionSource + + +def _source(): + return SessionSource( + platform=Platform.TELEGRAM, + chat_id="-1001", + chat_type="group", + user_id="12345", + ) + + +def _event(): + return MessageEvent( + text="run a task", + source=_source(), + message_id="msg-42", + ) + + +def _runner(monkeypatch, tmp_path): + runner = gateway_run.GatewayRunner(GatewayConfig()) + runner.adapters = {} + runner._running_agents = {} + runner._running_agents_ts = {} + runner._pending_messages = {} + runner._pending_approvals = {} + runner._is_user_authorized = lambda _source: True + runner._set_session_env = lambda _context: None + runner._handle_active_session_busy_message = AsyncMock(return_value=False) + runner._session_db = MagicMock() + runner._recover_telegram_topic_thread_id = lambda _source: None + runner._cache_session_source = lambda _key, _source: None + runner._is_session_run_current = lambda _key, _gen: True + runner._reply_anchor_for_event = lambda _event: None + runner._get_guild_id = lambda _event: None + runner._should_send_voice_reply = lambda *_a, **_kw: False + runner.hooks = MagicMock() + runner.hooks.emit = AsyncMock() + + runner.session_store = MagicMock() + runner.session_store.get_or_create_session.return_value = SessionEntry( + session_key="agent:main:telegram:group:-1001:12345", + session_id="sess-agent-end", + created_at=datetime.now(), + updated_at=datetime.now(), + platform=Platform.TELEGRAM, + chat_type="group", + ) + runner.session_store.load_transcript.return_value = [] + runner.session_store.append_to_transcript = MagicMock() + runner.session_store.update_session = MagicMock() + + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + monkeypatch.setattr( + gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "fake"} + ) + monkeypatch.setattr( + "agent.model_metadata.get_model_context_length", + lambda *_args, **_kwargs: 100_000, + ) + return runner + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("turn_exit_reason", "api_calls", "interrupted"), + [ + ("text_response(finish_reason=stop)", 1, False), + ("max_iterations_reached(90/90)", 90, False), + ("guardrail_halt", 3, False), + ("all_retries_exhausted_no_response", 3, False), + ("interrupted_by_user", 2, True), + ], +) +async def test_agent_end_hook_includes_termination_metadata( + monkeypatch, + tmp_path, + turn_exit_reason, + api_calls, + interrupted, +): + runner = _runner(monkeypatch, tmp_path) + runner._run_agent = AsyncMock( + return_value={ + "final_response": "done", + "messages": [ + {"role": "user", "content": "run a task"}, + {"role": "assistant", "content": "done"}, + ], + "tools": [], + "history_offset": 0, + "last_prompt_tokens": 0, + "api_calls": api_calls, + "turn_exit_reason": turn_exit_reason, + "failed": False, + "interrupted": interrupted, + "interrupt_message": ( + gateway_run._INTERRUPT_REASON_STOP if interrupted else None + ), + } + ) + + await runner._handle_message_with_agent( + _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 + ) + + agent_end_contexts = [ + call.args[1] + for call in runner.hooks.emit.await_args_list + if call.args[0] == "agent:end" + ] + assert len(agent_end_contexts) == 1 + assert agent_end_contexts[0]["turn_exit_reason"] == turn_exit_reason + assert agent_end_contexts[0]["api_call_count"] == api_calls + assert agent_end_contexts[0]["stale"] is False + assert isinstance(agent_end_contexts[0]["turn_exit_reason"], str) + assert isinstance(agent_end_contexts[0]["api_call_count"], int) + assert agent_end_contexts[0]["session_id"] == "sess-agent-end" + assert agent_end_contexts[0]["user_id"] == "12345" + + +@pytest.mark.asyncio +async def test_agent_end_hook_normalizes_early_user_interrupt(monkeypatch, tmp_path): + runner = _runner(monkeypatch, tmp_path) + runner._run_agent = AsyncMock( + return_value={ + "final_response": "Operation interrupted during retry.", + "messages": [], + "tools": [], + "api_calls": 2, + "interrupted": True, + "interrupt_message": gateway_run._INTERRUPT_REASON_STOP, + "completed": False, + } + ) + + await runner._handle_message_with_agent( + _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 + ) + + agent_end = [ + call.args[1] + for call in runner.hooks.emit.await_args_list + if call.args[0] == "agent:end" + ][0] + assert agent_end["turn_exit_reason"] == "interrupted_by_user" + assert agent_end["api_call_count"] == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("interrupt_message", "expected_reason"), + [ + (gateway_run._INTERRUPT_REASON_STOP, "interrupted_by_user"), + (gateway_run._INTERRUPT_REASON_RESET, "interrupted_by_user"), + ( + gateway_run._INTERRUPT_REASON_TIMEOUT, + "gateway_inactivity_timeout", + ), + ( + gateway_run._INTERRUPT_REASON_SSE_DISCONNECT, + "gateway_sse_disconnect", + ), + ( + gateway_run._INTERRUPT_REASON_GATEWAY_SHUTDOWN, + "gateway_shutdown", + ), + ( + gateway_run._INTERRUPT_REASON_GATEWAY_RESTART, + "gateway_restart", + ), + (None, "gateway_interrupt_unclassified"), + ("new correction", "interrupted_by_user"), + ], +) +async def test_agent_end_hook_classifies_early_interrupt_actor( + monkeypatch, + tmp_path, + interrupt_message, + expected_reason, +): + runner = _runner(monkeypatch, tmp_path) + runner._run_agent = AsyncMock( + return_value={ + "final_response": "Operation interrupted.", + "messages": [], + "tools": [], + "api_calls": 1, + "interrupted": True, + "interrupt_message": interrupt_message, + } + ) + + await runner._handle_message_with_agent( + _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 + ) + + agent_end = [ + call.args[1] + for call in runner.hooks.emit.await_args_list + if call.args[0] == "agent:end" + ][0] + assert agent_end["turn_exit_reason"] == expected_reason + + +def test_control_interrupt_markers_have_exhaustive_actor_classification(): + classified = ( + set(gateway_run._SYSTEM_INTERRUPT_EXIT_REASONS) + | gateway_run._USER_INTERRUPT_MESSAGES + ) + assert classified == gateway_run._CONTROL_INTERRUPT_MESSAGES + + +@pytest.mark.asyncio +async def test_agent_end_hook_preserves_explicit_reason_over_interrupt_fallback( + monkeypatch, + tmp_path, +): + runner = _runner(monkeypatch, tmp_path) + runner._run_agent = AsyncMock( + return_value={ + "final_response": "halted", + "messages": [], + "tools": [], + "api_calls": 1, + "interrupted": True, + "interrupt_message": gateway_run._INTERRUPT_REASON_TIMEOUT, + "turn_exit_reason": " guardrail_halt\x00\x1b[31m\n\twith context ", + } + ) + + await runner._handle_message_with_agent( + _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 + ) + + agent_end = [ + call.args[1] + for call in runner.hooks.emit.await_args_list + if call.args[0] == "agent:end" + ][0] + assert agent_end["turn_exit_reason"] == "guardrail_halt[31m with context" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("generic_reason", "system_marker", "expected_reason"), + [ + ( + "Interrupted_By_User(iteration=7)", + gateway_run._INTERRUPT_REASON_SSE_DISCONNECT, + "gateway_sse_disconnect", + ), + ( + "interrupted_during_api_call", + gateway_run._INTERRUPT_REASON_GATEWAY_SHUTDOWN, + "gateway_shutdown", + ), + ], +) +async def test_agent_end_hook_system_marker_overrides_generic_agent_interrupt( + monkeypatch, + tmp_path, + generic_reason, + system_marker, + expected_reason, +): + runner = _runner(monkeypatch, tmp_path) + runner._run_agent = AsyncMock( + return_value={ + "final_response": "interrupted", + "messages": [], + "tools": [], + "api_calls": 1, + "interrupted": True, + "interrupt_message": system_marker, + "turn_exit_reason": generic_reason, + } + ) + + await runner._handle_message_with_agent( + _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 + ) + + agent_end = [ + call.args[1] + for call in runner.hooks.emit.await_args_list + if call.args[0] == "agent:end" + ][0] + assert agent_end["turn_exit_reason"] == expected_reason + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("agent_result", "expected_reason"), + [ + ( + {"turn_exit_reason": "interrupted_by_user"}, + "interrupted_by_user", + ), + ( + { + "turn_exit_reason": "interrupted_during_api_call", + "interrupted": True, + "interrupt_message": "future gateway interrupt", + }, + "interrupted_during_api_call", + ), + ], +) +async def test_agent_end_hook_preserves_generic_reason_without_known_marker( + monkeypatch, + tmp_path, + agent_result, + expected_reason, +): + runner = _runner(monkeypatch, tmp_path) + runner._run_agent = AsyncMock( + return_value={ + "final_response": "interrupted", + "messages": [], + "tools": [], + "api_calls": 1, + **agent_result, + } + ) + + await runner._handle_message_with_agent( + _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 + ) + + agent_end = [ + call.args[1] + for call in runner.hooks.emit.await_args_list + if call.args[0] == "agent:end" + ][0] + assert agent_end["turn_exit_reason"] == expected_reason + + +class _UnstringableReason: + def __bool__(self): + raise TypeError("malformed reason") + + def __str__(self): + raise TypeError("malformed reason") + + +class _ExplodingCount: + def __bool__(self): + raise RuntimeError("malformed count") + + +@pytest.mark.asyncio +async def test_agent_end_hook_bounds_reason_and_fails_safe_on_bad_count( + monkeypatch, + tmp_path, +): + runner = _runner(monkeypatch, tmp_path) + runner._run_agent = AsyncMock( + return_value={ + "final_response": "failed", + "messages": [], + "tools": [], + "api_calls": object(), + "turn_exit_reason": "local_processing_error(" + ("x" * 500) + ")", + "failed": True, + } + ) + + await runner._handle_message_with_agent( + _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 + ) + + agent_end = [ + call.args[1] + for call in runner.hooks.emit.await_args_list + if call.args[0] == "agent:end" + ][0] + assert len(agent_end["turn_exit_reason"]) == 200 + assert agent_end["turn_exit_reason"].startswith("local_processing_error(") + assert agent_end["api_call_count"] == 0 + + +@pytest.mark.asyncio +async def test_agent_end_hook_still_emits_when_count_protocol_raises( + monkeypatch, + tmp_path, +): + runner = _runner(monkeypatch, tmp_path) + runner._run_agent = AsyncMock( + return_value={ + "final_response": "failed", + "messages": [], + "tools": [], + "api_calls": _ExplodingCount(), + "turn_exit_reason": "guardrail_halt", + "failed": True, + } + ) + + await runner._handle_message_with_agent( + _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 + ) + + agent_end = [ + call.args[1] + for call in runner.hooks.emit.await_args_list + if call.args[0] == "agent:end" + ][0] + assert agent_end["turn_exit_reason"] == "guardrail_halt" + assert agent_end["api_call_count"] == 0 + + +@pytest.mark.asyncio +async def test_agent_end_hook_fails_safe_on_malformed_reason(monkeypatch, tmp_path): + runner = _runner(monkeypatch, tmp_path) + runner._run_agent = AsyncMock( + return_value={ + "final_response": "failed", + "messages": [], + "tools": [], + "api_calls": 1, + "turn_exit_reason": _UnstringableReason(), + "failed": True, + } + ) + + await runner._handle_message_with_agent( + _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 + ) + + agent_end = [ + call.args[1] + for call in runner.hooks.emit.await_args_list + if call.args[0] == "agent:end" + ][0] + assert agent_end["turn_exit_reason"] == "unknown" + + +@pytest.mark.asyncio +async def test_agent_end_hook_clamps_negative_count(monkeypatch, tmp_path): + runner = _runner(monkeypatch, tmp_path) + runner._run_agent = AsyncMock( + return_value={ + "final_response": "done", + "messages": [], + "tools": [], + "api_calls": -3, + "turn_exit_reason": "text_response(finish_reason=stop)", + "failed": False, + } + ) + + await runner._handle_message_with_agent( + _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 + ) + + agent_end = [ + call.args[1] + for call in runner.hooks.emit.await_args_list + if call.args[0] == "agent:end" + ][0] + assert agent_end["api_call_count"] == 0 + + +@pytest.mark.asyncio +async def test_stale_proxy_result_emits_agent_end_before_discard( + monkeypatch, + tmp_path, +): + events = [] + + class _StaleAdapter: + def pop_post_delivery_callback(self, _key, *, generation): + assert generation == 1 + events.append("cleanup") + + async def send(self, _chat_id, _content, *, metadata): + return SimpleNamespace(success=True) + + runner = _runner(monkeypatch, tmp_path) + runner._is_session_run_current = lambda _key, _gen: False + stale_adapter = _StaleAdapter() + runner._adapter_for_source = lambda _source: stale_adapter + + async def emit_after_cleanup(event_type, _context): + if event_type == "agent:end": + events.append("emit") + + runner.hooks.emit.side_effect = emit_after_cleanup + runner._run_agent = AsyncMock( + return_value={ + "final_response": "", + "messages": [], + "tools": [], + "api_calls": 0, + "partial": False, + "turn_exit_reason": "gateway_proxy_stale_generation", + } + ) + + result = await runner._handle_message_with_agent( + _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 + ) + + assert result is None + agent_end = [ + call.args[1] + for call in runner.hooks.emit.await_args_list + if call.args[0] == "agent:end" + ] + assert len(agent_end) == 1 + assert agent_end[0]["turn_exit_reason"] == "gateway_proxy_stale_generation" + assert agent_end[0]["api_call_count"] == 0 + assert agent_end[0]["response"] == "" + assert agent_end[0]["stale"] is True + assert events == ["cleanup", "emit"] + + +@pytest.mark.asyncio +async def test_stale_non_proxy_result_pairs_agent_start_and_end( + monkeypatch, + tmp_path, +): + runner = _runner(monkeypatch, tmp_path) + runner._is_session_run_current = lambda _key, _gen: False + runner._run_agent = AsyncMock( + return_value={ + "final_response": "discarded", + "messages": [], + "tools": [], + "api_calls": 1, + "partial": False, + "turn_exit_reason": "text_response(finish_reason=stop)", + } + ) + + result = await runner._handle_message_with_agent( + _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 + ) + + assert result is None + lifecycle_events = [ + (call.args[0], call.args[1]) + for call in runner.hooks.emit.await_args_list + if call.args[0] in {"agent:start", "agent:end"} + ] + assert [event_type for event_type, _context in lifecycle_events] == [ + "agent:start", + "agent:end", + ] + assert ( + lifecycle_events[1][1]["turn_exit_reason"] + == "gateway_stale_generation" + ) + assert lifecycle_events[1][1]["response"] == "" + assert lifecycle_events[1][1]["stale"] is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("agent_result", "expected_reason"), + [ + ( + { + "final_response": "", + "messages": [], + "tools": [], + "api_calls": 0, + }, + "gateway_stale_generation", + ), + ( + { + "final_response": "proxy output", + "messages": [], + "tools": [], + "api_calls": 1, + "turn_exit_reason": "gateway_proxy_response_complete", + }, + "gateway_proxy_stale_generation", + ), + ( + { + "final_response": "", + "messages": [], + "tools": [], + "api_calls": 1, + "turn_exit_reason": "gateway_inactivity_timeout", + }, + "gateway_inactivity_timeout", + ), + ], +) +async def test_stale_agent_end_preserves_provenance_and_abnormal_reason( + monkeypatch, + tmp_path, + agent_result, + expected_reason, +): + runner = _runner(monkeypatch, tmp_path) + runner._is_session_run_current = lambda _key, _gen: False + runner._run_agent = AsyncMock(return_value=agent_result) + + await runner._handle_message_with_agent( + _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 + ) + + agent_end = [ + call.args[1] + for call in runner.hooks.emit.await_args_list + if call.args[0] == "agent:end" + ][0] + assert agent_end["turn_exit_reason"] == expected_reason + assert agent_end["response"] == "" + assert agent_end["stale"] is True + + +def _runtime_runner(): + runner = object.__new__(gateway_run.GatewayRunner) + runner.adapters = {} + runner._voice_mode = {} + runner._prefill_messages = [] + runner._ephemeral_system_prompt = "" + runner._reasoning_config = None + runner._provider_routing = {} + runner._fallback_model = None + runner._session_db = None + runner._running_agents = {} + runner._session_run_generation = {} + runner._agent_cache = {} + runner._agent_cache_lock = threading.Lock() + runner.hooks = SimpleNamespace(loaded_hooks=False) + runner.config = SimpleNamespace( + thread_sessions_per_user=False, + group_sessions_per_user=False, + stt_enabled=False, + streaming=None, + ) + return runner + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("final_response", "turn_exit_reason", "completed", "failed"), + [ + ("done", "text_response(finish_reason=stop)", True, False), + ("halted", "guardrail_halt", False, True), + (None, "all_retries_exhausted_no_response", False, True), + ], +) +async def test_run_agent_propagates_exit_reason_through_result_mapping( + monkeypatch, + tmp_path, + final_response, + turn_exit_reason, + completed, + failed, +): + class _TerminationMetadataAgent: + def __init__(self, **_kwargs): + self.tools = [] + self._interrupt_requested = False + + @property + def is_interrupted(self): + return self._interrupt_requested + + def run_conversation( + self, + _message, + conversation_history=None, + task_id=None, + **_kwargs, + ): + return { + "final_response": final_response, + "messages": [], + "api_calls": 3, + "completed": completed, + "failed": failed, + "error": "synthetic failure" if failed else None, + "turn_exit_reason": turn_exit_reason, + } + + fake_run_agent = types.ModuleType("run_agent") + fake_run_agent.AIAgent = _TerminationMetadataAgent + monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + monkeypatch.setattr( + gateway_run, + "_resolve_runtime_agent_kwargs", + lambda: {"api_key": "fake"}, + ) + + result = await _runtime_runner()._run_agent( + message="run a task", + context_prompt="", + history=[], + source=_source(), + session_id="sess-agent-end-runtime", + session_key="agent:main:telegram:group:-1001:12345", + ) + + assert result["turn_exit_reason"] == turn_exit_reason + assert result["api_calls"] == 3 + assert result["completed"] is completed + assert result["failed"] is failed + + +@pytest.mark.asyncio +async def test_run_agent_classifies_runtime_resolution_failure( + monkeypatch, + tmp_path, +): + runner = _runtime_runner() + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + runner._resolve_session_agent_runtime = MagicMock( + side_effect=RuntimeError("synthetic runtime resolution failure") + ) + + result = await runner._run_agent( + message="run a task", + context_prompt="", + history=[], + source=_source(), + session_id="sess-agent-end-auth-failure", + session_key="agent:main:telegram:group:-1001:12345", + ) + + assert result["failed"] is True + assert result["turn_exit_reason"] == "gateway_agent_runtime_resolution_failed" + + +@pytest.mark.asyncio +async def test_run_agent_classifies_inactivity_timeout(monkeypatch, tmp_path): + interrupted = threading.Event() + + class _InactiveAgent: + def __init__(self, **_kwargs): + self.tools = [] + self._interrupt_requested = False + + @property + def is_interrupted(self): + return self._interrupt_requested + + def run_conversation( + self, + _message, + conversation_history=None, + task_id=None, + **_kwargs, + ): + interrupted.wait(timeout=2) + return { + "final_response": None, + "messages": [], + "api_calls": 1, + "completed": False, + "failed": True, + } + + def get_activity_summary(self): + return { + "last_activity_desc": "synthetic stalled provider", + "seconds_since_activity": 60, + "current_tool": None, + "api_call_count": 1, + "max_iterations": 90, + } + + def interrupt(self, _reason): + self._interrupt_requested = True + interrupted.set() + + fake_run_agent = types.ModuleType("run_agent") + fake_run_agent.AIAgent = _InactiveAgent + monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + monkeypatch.setattr( + gateway_run, + "_resolve_runtime_agent_kwargs", + lambda: {"api_key": "fake"}, + ) + monkeypatch.setenv("HERMES_AGENT_TIMEOUT", "0.001") + monkeypatch.setenv("HERMES_AGENT_TIMEOUT_WARNING", "0") + monkeypatch.setattr(gateway_run, "_GATEWAY_AGENT_POLL_INTERVAL_SECONDS", 0.01) + + result = await _runtime_runner()._run_agent( + message="run a task", + context_prompt="", + history=[], + source=_source(), + session_id="sess-agent-end-inactivity", + session_key="agent:main:telegram:group:-1001:12345", + ) + + assert result["failed"] is True + assert result["turn_exit_reason"] == "gateway_inactivity_timeout" + assert interrupted.is_set() diff --git a/tests/gateway/test_proxy_mode.py b/tests/gateway/test_proxy_mode.py index be98f7eb9acb..cf44a1de36ca 100644 --- a/tests/gateway/test_proxy_mode.py +++ b/tests/gateway/test_proxy_mode.py @@ -1,5 +1,6 @@ """Tests for gateway proxy mode — forwarding messages to a remote API server.""" +import sys from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -39,10 +40,11 @@ def _make_source(platform=Platform.MATRIX): class _FakeSSEResponse: """Simulates an aiohttp response with SSE streaming.""" - def __init__(self, status=200, sse_chunks=None, error_text=""): + def __init__(self, status=200, sse_chunks=None, error_text="", stream_error=None): self.status = status self._sse_chunks = sse_chunks or [] self._error_text = error_text + self._stream_error = stream_error self.content = self async def text(self): @@ -53,6 +55,8 @@ async def iter_any(self): if isinstance(chunk, str): chunk = chunk.encode("utf-8") yield chunk + if self._stream_error: + raise self._stream_error async def __aenter__(self): return self @@ -227,6 +231,42 @@ async def test_run_agent_skips_proxy_when_not_configured(self, monkeypatch): class TestRunAgentViaProxy: """Test the actual proxy HTTP forwarding logic.""" + @pytest.mark.asyncio + async def test_reports_missing_aiohttp_dependency(self, monkeypatch): + monkeypatch.setenv("GATEWAY_PROXY_URL", "http://host:8642") + runner = _make_runner() + + with patch.dict(sys.modules, {"aiohttp": None}): + result = await runner._run_agent_via_proxy( + message="hi", + context_prompt="", + history=[], + source=_make_source(), + session_id="test", + ) + + assert "failed" not in result + assert result["partial"] is False + assert result["turn_exit_reason"] == "gateway_proxy_dependency_missing" + + @pytest.mark.asyncio + async def test_reports_missing_proxy_configuration(self, monkeypatch): + monkeypatch.delenv("GATEWAY_PROXY_URL", raising=False) + runner = _make_runner() + + with patch("gateway.run._load_gateway_config", return_value={}): + result = await runner._run_agent_via_proxy( + message="hi", + context_prompt="", + history=[], + source=_make_source(), + session_id="test", + ) + + assert "failed" not in result + assert result["partial"] is False + assert result["turn_exit_reason"] == "gateway_proxy_not_configured" + @pytest.mark.asyncio async def test_builds_correct_request(self, monkeypatch): monkeypatch.setenv("GATEWAY_PROXY_URL", "http://host:8642") @@ -303,6 +343,9 @@ async def test_handles_http_error(self, monkeypatch): assert "Proxy error (401)" in result["final_response"] assert result["api_calls"] == 0 + assert "failed" not in result + assert result["partial"] is False + assert result["turn_exit_reason"] == "gateway_proxy_http_error" @pytest.mark.asyncio async def test_handles_connection_error(self, monkeypatch): @@ -333,6 +376,9 @@ async def __aexit__(self, *args): ) assert "Proxy connection error" in result["final_response"] + assert "failed" not in result + assert result["partial"] is False + assert result["turn_exit_reason"] == "gateway_proxy_connection_error" @pytest.mark.asyncio async def test_rejects_proxy_sse_without_line_boundary_after_buffer_cap(self, monkeypatch): @@ -359,6 +405,66 @@ async def test_rejects_proxy_sse_without_line_boundary_after_buffer_cap(self, mo assert "Proxy connection error" in result["final_response"] assert "exceeded max buffer size" in result["final_response"] assert result["api_calls"] == 0 + assert "failed" not in result + assert result["partial"] is False + assert result["turn_exit_reason"] == "gateway_proxy_connection_error" + + @pytest.mark.asyncio + async def test_partial_stream_preserves_response_without_downstream_partial_flow( + self, monkeypatch + ): + monkeypatch.setenv("GATEWAY_PROXY_URL", "http://host:8642") + monkeypatch.delenv("GATEWAY_PROXY_KEY", raising=False) + runner = _make_runner() + source = _make_source() + resp = _FakeSSEResponse( + status=200, + sse_chunks=[ + b'data: {"choices":[{"delta":{"content":"partial"}}]}\n\n' + ], + stream_error=ConnectionError("stream interrupted"), + ) + session = _FakeSession(resp) + + with patch("gateway.run._load_gateway_config", return_value={}): + with _patch_aiohttp(session): + with patch("aiohttp.ClientTimeout"): + result = await runner._run_agent_via_proxy( + message="hi", + context_prompt="", + history=[], + source=source, + session_id="test", + ) + + assert result["final_response"] == "partial" + assert "failed" not in result + assert result["partial"] is False + assert result["turn_exit_reason"] == "gateway_proxy_partial_response" + + @pytest.mark.asyncio + async def test_empty_stream_has_distinct_exit_reason(self, monkeypatch): + monkeypatch.setenv("GATEWAY_PROXY_URL", "http://host:8642") + monkeypatch.delenv("GATEWAY_PROXY_KEY", raising=False) + runner = _make_runner() + source = _make_source() + session = _FakeSession(_FakeSSEResponse(status=200, sse_chunks=[])) + + with patch("gateway.run._load_gateway_config", return_value={}): + with _patch_aiohttp(session): + with patch("aiohttp.ClientTimeout"): + result = await runner._run_agent_via_proxy( + message="hi", + context_prompt="", + history=[], + source=source, + session_id="test", + ) + + assert result["final_response"] == "(No response from remote agent)" + assert "failed" not in result + assert result["partial"] is False + assert result["turn_exit_reason"] == "gateway_proxy_empty_response" @pytest.mark.asyncio async def test_skips_tool_messages_in_history(self, monkeypatch): @@ -428,6 +534,9 @@ async def test_result_shape_matches_run_agent(self, monkeypatch): assert "messages" in result assert "api_calls" in result assert "tools" in result + assert "failed" not in result + assert result["partial"] is False + assert result["turn_exit_reason"] == "gateway_proxy_response_complete" assert "history_offset" in result assert result["history_offset"] == 2 # len(history) assert "session_id" in result @@ -466,6 +575,50 @@ async def test_proxy_stale_generation_returns_empty_result(self, monkeypatch): assert result["final_response"] == "" assert result["messages"] == [] assert result["api_calls"] == 0 + assert result["partial"] is False + assert result["turn_exit_reason"] == "gateway_proxy_stale_generation" + + @pytest.mark.asyncio + async def test_proxy_post_stream_stale_generation_has_distinct_exit_reason( + self, monkeypatch + ): + monkeypatch.setenv("GATEWAY_PROXY_URL", "http://host:8642") + monkeypatch.delenv("GATEWAY_PROXY_KEY", raising=False) + runner = _make_runner() + source = _make_source() + + resp = _FakeSSEResponse( + status=200, + sse_chunks=[ + b'data: {"choices":[{"delta":{"content":"stale"}}]}\n\n' + b"data: [DONE]\n\n" + ], + ) + session = _FakeSession(resp) + + with patch.object( + runner, + "_is_session_run_current", + side_effect=[True, False], + ): + with patch("gateway.run._load_gateway_config", return_value={}): + with _patch_aiohttp(session): + with patch("aiohttp.ClientTimeout"): + result = await runner._run_agent_via_proxy( + message="hi", + context_prompt="", + history=[], + source=source, + session_id="sess-123", + session_key="test-key", + run_generation=1, + ) + + assert result["final_response"] == "" + assert result["messages"] == [] + assert result["api_calls"] == 0 + assert result["partial"] is False + assert result["turn_exit_reason"] == "gateway_proxy_stale_generation" @pytest.mark.asyncio async def test_no_auth_header_without_key(self, monkeypatch): diff --git a/website/docs/user-guide/features/hooks.md b/website/docs/user-guide/features/hooks.md index 52417bf5741b..2769393bb4f5 100644 --- a/website/docs/user-guide/features/hooks.md +++ b/website/docs/user-guide/features/hooks.md @@ -80,11 +80,49 @@ async def handle(event_type: str, context: dict): | `session:reset` | User ran `/new` or `/reset` | `platform`, `user_id`, `session_key` | | `agent:start` | Agent begins processing a message | `platform`, `user_id`, `session_id`, `message` | | `agent:step` | Each iteration of the tool-calling loop | `platform`, `user_id`, `session_id`, `iteration`, `tool_names` | -| `agent:end` | Agent finishes processing | `platform`, `user_id`, `session_id`, `message`, `response` | +| `agent:end` | Agent finishes processing | same keys as `agent:start`, plus `response` (truncated to 500 chars), `turn_exit_reason`, `api_call_count`, `stale` | | `reaction:added` | An emoji reaction was added to a message the bot can see (Slack adapter currently). Requires the `reactions:read` scope + the `reaction_added` bot event subscription; the bot must be a member of the channel. | `platform`, `reaction`, `user_id`, `item_user_id`, `item_type`, `channel_id`, `message_ts`, `team_id`, `event_ts`, `raw_event` | | `reaction:removed` | An emoji reaction was removed from a message the bot can see. Requires the `reaction_removed` bot event subscription. | same shape as `reaction:added` | | `command:*` | Any slash command executed | `platform`, `user_id`, `command`, `args` | +`agent:end.turn_exit_reason` preserves the agent finalizer's reason. It +classifies both explicit stop/reset controls and mid-turn user correction text +without an explicit finalizer reason as `interrupted_by_user`, and keeps gateway +timeout, disconnect, shutdown, and restart aborts distinct. Missing or malformed +reasons are `unknown`; consumers should treat that class as actionable until +reconciled. Finalizer reason text is collapsed to a single line and truncated +to 200 characters before hook delivery. + +`api_call_count` is a non-negative integer. A value of `0` can also mean not +reported: proxy mode reports `0`, and absent or malformed counts clamp to `0`. +`stale` is a boolean: `true` means a newer run superseded this one and its +output was discarded; `false` means this is the current delivered turn. + +The value is an open vocabulary. Specific agent-finalizer reason semantics are +preserved, while the delivered string is collapsed to one printable line and +truncated to 200 characters. +Gateway-owned classes are `interrupted_by_user`, `unknown`, +`gateway_interrupt_unclassified`, `gateway_inactivity_timeout`, +`gateway_sse_disconnect`, `gateway_shutdown`, `gateway_restart`, and +`gateway_agent_runtime_resolution_failed`. A superseded non-proxy run can use +`gateway_stale_generation`. Proxy mode +uses `gateway_proxy_dependency_missing`, `gateway_proxy_not_configured`, +`gateway_proxy_http_error`, `gateway_proxy_connection_error`, +`gateway_proxy_partial_response`, `gateway_proxy_response_complete`, +`gateway_proxy_empty_response`, and `gateway_proxy_stale_generation`. +Consumers should recognize future +`gateway_*` and `gateway_proxy_*` extensions without treating the vocabulary as +closed. + +A superseded run still fires `agent:end` with `stale: true` before its +discarded result returns. That event has an empty `response`. A non-proxy run +whose prior result was normal or missing a reason uses +`gateway_stale_generation`; an otherwise complete proxy run uses +`gateway_proxy_stale_generation`. More specific abnormal reasons remain intact. +Observation and cleanup handlers may record these events, but delivery-side +handlers must gate on `stale` and must not post a follow-up when it is `true`, +regardless of the reason string. + #### Wildcard Matching Handlers registered for `command:*` fire for any `command:` event (`command:model`, `command:reset`, etc.). Monitor all slash commands with a single subscription. diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/hooks.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/hooks.md index 85265ecfd213..ffd2cb6d7916 100644 --- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/hooks.md +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/hooks.md @@ -80,9 +80,39 @@ async def handle(event_type: str, context: dict): | `session:reset` | 用户执行 `/new` 或 `/reset` | `platform`、`user_id`、`session_key` | | `agent:start` | Agent 开始处理消息 | `platform`、`user_id`、`session_id`、`message` | | `agent:step` | 工具调用循环的每次迭代 | `platform`、`user_id`、`session_id`、`iteration`、`tool_names` | -| `agent:end` | Agent 完成处理 | `platform`、`user_id`、`session_id`、`message`、`response` | +| `agent:end` | Agent 完成处理 | `platform`、`user_id`、`session_id`、`message`、`response`、`turn_exit_reason`、`api_call_count`、`stale` | | `command:*` | 任意斜杠命令执行 | `platform`、`user_id`、`command`、`args` | +`agent:end.turn_exit_reason` 会保留 Agent finalizer 给出的原因。显式 stop/reset 控制以及 +没有显式 finalizer 原因的轮次中途用户修正文本，都会分类为 `interrupted_by_user`； +Gateway 超时、断开连接、关闭和重启中止则保持为不同分类。缺失或格式错误的原因会归一化为 +`unknown`；消费者在完成核对前应将此类视为可操作异常。Finalizer 原因文本在传递给 hook +前会折叠为单行并截断为 200 个字符。 + +`api_call_count` 是非负整数。值为 `0` 也可能表示未报告：Proxy 模式固定报告 `0`， +缺失或格式错误的计数也会被限制为 `0`。 +`stale` 是布尔值：`true` 表示该运行已被更新的运行取代，输出已被丢弃； +`false` 表示这是当前实际交付的运行。 + +该字段采用开放词汇。具体的 Agent finalizer 原因语义会保留，但传递的字符串会 +折叠为单个可打印行并截断为 200 个字符。Gateway 自有分类包括 +`interrupted_by_user`、`unknown`、`gateway_interrupt_unclassified`、 +`gateway_inactivity_timeout`、`gateway_sse_disconnect`、`gateway_shutdown`、 +`gateway_restart` 和 `gateway_agent_runtime_resolution_failed`。被取代的非 Proxy 运行可使用 +`gateway_stale_generation`。Proxy 模式使用 +`gateway_proxy_dependency_missing`、`gateway_proxy_not_configured`、 +`gateway_proxy_http_error`、`gateway_proxy_connection_error`、 +`gateway_proxy_partial_response`、`gateway_proxy_response_complete`、 +`gateway_proxy_empty_response` 和 `gateway_proxy_stale_generation`。消费者应兼容未来的 `gateway_*` 与 +`gateway_proxy_*` 扩展，不应把当前词汇表视为封闭枚举。 + +被后续运行取代的运行仍会在丢弃结果前以 `stale: true` 触发 `agent:end`，且该事件的 +`response` 为空。如果非 Proxy 运行此前的结果正常或缺失原因，则使用 +`gateway_stale_generation`；此前正常完成的 Proxy 运行使用 +`gateway_proxy_stale_generation`。更具体的异常原因会保持不变。观察与清理处理器可以记录 +这些事件，但负责发送消息的处理器必须依据 `stale` 进行拦截；当它为 `true` 时，无论原因 +字符串为何都不得发布后续消息。 + #### 通配符匹配 注册了 `command:*` 的处理器会在任何 `command:` 事件（`command:model`、`command:reset` 等）触发时执行。通过单个订阅即可监控所有斜杠命令。 @@ -1329,4 +1359,4 @@ Shell hooks 以**你的完整用户凭据**运行——与 cron 条目或 shell ### 顺序与优先级 -Python 插件 hook 和 shell hook 都流经同一个 `invoke_hook()` 分发器。Python 插件先注册（`discover_and_load()`），shell hook 后注册（`register_from_config()`），因此在平局情况下 Python `pre_tool_call` 的 block 决定优先。第一个有效的 block 生效——聚合器在任何回调产生带非空 message 的 `{"action": "block", "message": str}` 时立即返回。 \ No newline at end of file +Python 插件 hook 和 shell hook 都流经同一个 `invoke_hook()` 分发器。Python 插件先注册（`discover_and_load()`），shell hook 后注册（`register_from_config()`），因此在平局情况下 Python `pre_tool_call` 的 block 决定优先。第一个有效的 block 生效——聚合器在任何回调产生带非空 message 的 `{"action": "block", "message": str}` 时立即返回。