fix(gateway): expose agent exit metadata to hooks

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,13 +865,25 @@ 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,
         "api_calls": api_calls,
         "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

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"``

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