diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index 258777338b..ab3a3a4b4f 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -18,7 +18,7 @@ It covers: - approved, rejected, mixed, and replayed approval rounds; - reasoning content and opaque reasoning signatures bound to function calls; - history persistence and service-side continuation; -- error, user-input, middleware-termination, and loop-limit paths; +- error, user-input, middleware-termination, middleware-failure, and loop-limit paths; - provider and transport serialization of function calls and results. The primary implementation is in `python/packages/core/agent_framework/_tools.py`. History replay behavior in @@ -315,7 +315,19 @@ that manually replay messages own the equivalent rule: do not resend an approval ### Function calls and results - Every actionable local `function_call` produces exactly one terminal `function_result`, unless execution pauses - for a new user-input request. + for a new user-input request or the run is aborted by `MiddlewareFailure`. +- An ordinary exception raised by function middleware or a tool body becomes one terminal error `function_result` + and the loop continues; `MiddlewareFailure` is the loop's only fail-closed escape: it is never converted into a + tool result, the in-flight parallel batch is cancelled, no further tool call starts, no further model turn is + consumed, and the exception propagates to the caller (for streaming runs, when the stream is consumed). On a + service-managed conversation the loop first settles the aborted batch — one error `function_result` per dangling + call, submitted with `tool_choice="none"` in a single extra request whose response is discarded — so the hosted + thread is not left ending in unresolved function calls that the service would reject on the session's next + request; without a service-managed conversation no extra request is made. Batch + cancellation is cooperative: an async sibling stops at its next suspension point, while a synchronous tool body + already executing in a worker thread cannot be interrupted and may complete its side effects — its result is + discarded either way and never reaches the transcript, the model, or history. Middleware must not catch + `MiddlewareFailure` — swallowing it converts a fail-closed abort back into a running, possibly unguarded loop. - Parallel calls retain model order in the returned transcript. - Reused `call_id` values are correlated by logical occurrence, not one global value per id. - A completed function call/result pair is inert on later turns. @@ -459,6 +471,9 @@ that manually replay messages own the equivalent rule: do not resend an approval | Consecutive error cap | Error threshold stops repeated failures, submits collected results, and makes only the required final no-tool model call. | `test_function_invocation_config_max_consecutive_errors`, `test_streaming_function_invocation_config_max_consecutive_errors`, `test_approval_resume_error_limit_forces_final_no_tool_response` | | Unknown call handling | Configured false returns an error result; configured true raises. | `test_function_invocation_config_terminate_on_unknown_calls_false`, `test_function_invocation_config_terminate_on_unknown_calls_true`, streaming equivalents | | Middleware termination | Normal non-approval loop stops without a second model call. | `test_terminate_loop_single_function_call`, `test_terminate_loop_multiple_function_calls_one_terminates`, `test_terminate_loop_streaming_single_function_call` | +| Middleware failure (fatal) | `MiddlewareFailure` from function middleware or a tool body propagates to the caller without becoming a tool result; the tool does not execute (pre-invocation) or its result never feeds another model call (post-invocation); the cause chain is preserved; ordinary exceptions still become tool-error results and the loop continues. | `packages/core/tests/core/test_middleware_with_agent.py::TestMiddlewareFailure::test_failure_before_tool_aborts_run`, `test_failure_after_tool_aborts_run_before_next_model_turn`, `test_failure_cause_chain_reaches_caller`, `test_failure_from_tool_escapes_without_middleware`, `test_failure_streaming_reaches_stream_consumer`, `test_ordinary_exception_still_becomes_tool_error` | +| Middleware failure batch cancellation | A fatal signal fails the whole parallel batch: in-flight sibling tool invocations are cancelled and awaited before the failure propagates. Cancellation is cooperative — an async sibling stops at its next suspension point; a synchronous tool body already executing in a worker thread cannot be interrupted and may complete its side effects, but its result is discarded and never reaches the transcript, the model, or history, and failure propagation is not delayed behind it. | `TestMiddlewareFailure::test_failure_cancels_concurrent_sibling_tool`, `test_failure_with_sync_sibling_discards_late_result` | +| Middleware failure on a service-managed conversation | The continuation state is already persisted when the batch fails, so before propagating, the loop settles the hosted thread: one error `function_result` per dangling call, sent with `tool_choice="none"` in one extra request whose response is discarded; a settlement failure never masks the abort. Without a service-managed conversation no extra request is made. | `TestMiddlewareFailure::test_failure_settles_dangling_calls_on_service_conversation`, `test_failure_settles_service_conversation_streaming`, `test_failure_without_service_conversation_makes_no_settlement_request` | | Maximum iterations | No orphan calls; a final no-tool response or deterministic fallback is returned. | `test_max_iterations_limit`, `test_max_iterations_no_orphaned_function_calls`, `test_max_iterations_makes_final_toolchoice_none_call`, `test_max_iterations_blank_final_fallback_synthesizes_message`, streaming equivalents | | Maximum function calls | Parallel overshoot is bounded after the batch; every executed result group counts even without a `function_result`; blank final responses get fallback content. | `test_max_function_calls_limits_parallel_invocations`, `test_max_function_calls_single_calls_per_iteration`, `test_user_input_request_multiple_contents_propagate`, `test_approval_resume_user_input_counts_toward_function_call_budget`, `test_max_function_calls_blank_final_fallback_synthesizes_message`, streaming equivalent | | Provider tool content after an active limit | Locally actionable calls and local approval requests returned despite `tool_choice="none"` are removed in both response modes. Provider-executed informational call/result pairs, hosted approval requests, and metadata-only streaming updates remain visible; fallback text never replaces retained transcript content. | `test_function_invocation_limit_drops_unexecutable_tool_content`, `test_streaming_function_invocation_limit_drops_unexecutable_tool_content`, `test_streaming_function_invocation_limit_preserves_metadata_after_tool_content_is_dropped`, `test_function_invocation_limit_preserves_provider_executed_tool_pair`, `test_streaming_function_invocation_limit_preserves_provider_executed_tool_pair`, `test_function_invocation_limit_appends_fallback_after_provider_executed_tool_pair`, `test_streaming_function_invocation_limit_appends_fallback_after_provider_executed_tool_pair`, `test_function_invocation_limit_preserves_hosted_approval_request`, `test_streaming_function_invocation_limit_preserves_hosted_approval_request` | @@ -543,6 +558,7 @@ Before accepting an update, reviewers must confirm: ## Related issues - #7241 — approval-resolution result streaming +- #7522 — first-class fatal signal (`MiddlewareFailure`) for function middleware - #7267 / #7271 and #7304 — replayed calls and reused ids - #7043 — provider-injected approval execution - #6828 — AG-UI `confirm_changes` snapshot correlation diff --git a/python/CODING_STANDARD.md b/python/CODING_STANDARD.md index 256d4a7790..9081e52b87 100644 --- a/python/CODING_STANDARD.md +++ b/python/CODING_STANDARD.md @@ -270,6 +270,7 @@ AgentFrameworkException # Base for all AF exceptions │ └── ToolExecutionException # Failure during tool execution │ ├── MiddlewareException # Middleware failures +│ ├── MiddlewareFailure # Control-flow: fatal fail-closed abort of the run │ └── MiddlewareTermination # Control-flow: early middleware termination │ └── SettingNotFoundError # Required setting not resolved from any source diff --git a/python/packages/core/agent_framework/__init__.py b/python/packages/core/agent_framework/__init__.py index d7ab916875..9a786e2c79 100644 --- a/python/packages/core/agent_framework/__init__.py +++ b/python/packages/core/agent_framework/__init__.py @@ -185,6 +185,7 @@ "FunctionMiddleware", "FunctionMiddlewareTypes", "MiddlewareBundle", + "MiddlewareFailure", "MiddlewareTermination", "MiddlewareType", "MiddlewareTypes", @@ -507,6 +508,7 @@ "MessageInjectionMiddleware", "MiddlewareBundle", "MiddlewareException", + "MiddlewareFailure", "MiddlewareTermination", "MiddlewareType", "MiddlewareTypes", diff --git a/python/packages/core/agent_framework/__init__.pyi b/python/packages/core/agent_framework/__init__.pyi index f72269c02d..727d4f007b 100644 --- a/python/packages/core/agent_framework/__init__.pyi +++ b/python/packages/core/agent_framework/__init__.pyi @@ -144,6 +144,7 @@ from ._middleware import ( FunctionMiddleware, FunctionMiddlewareTypes, MiddlewareBundle, + MiddlewareFailure, MiddlewareTermination, MiddlewareType, MiddlewareTypes, @@ -471,6 +472,7 @@ __all__ = [ "MessageInjectionMiddleware", "MiddlewareBundle", "MiddlewareException", + "MiddlewareFailure", "MiddlewareTermination", "MiddlewareType", "MiddlewareTypes", diff --git a/python/packages/core/agent_framework/_agent_hooks.py b/python/packages/core/agent_framework/_agent_hooks.py index 669cdb262a..fc14d4d41f 100644 --- a/python/packages/core/agent_framework/_agent_hooks.py +++ b/python/packages/core/agent_framework/_agent_hooks.py @@ -37,7 +37,13 @@ not executed (or its result is discarded) and a tool-error payload is surfaced to the model so the agent loop can continue, per the spec's block-propagation rules. A ``host_error:*`` deny at the tool seam additionally halts the run (the enforcement - layer itself failed, so continuing would be unreliable). + layer itself failed, so continuing would be unreliable): the run is aborted through + the function-invocation loop's fail-closed escape + (:class:`~agent_framework.MiddlewareFailure`) and the + :class:`agent_hooks.InterceptionBlocked` propagates to the caller, exactly like a + run-level deny. Other unexpected failures inside the enforcement layer (projection + bug, emitter fault) abort the run the same way and surface as + :class:`~agent_framework.MiddlewareFailure`. - Framework middleware short-circuits (``MiddlewareTermination``) are guarded: a result substituted by another middleware still passes ``output`` / ``post_model_call`` / ``post_tool_call`` before it egresses or enters the transcript. @@ -113,6 +119,7 @@ FunctionInvocationContext, FunctionMiddleware, MiddlewareBundle, + MiddlewareFailure, MiddlewareTermination, ) from ._serialization import make_json_safe @@ -230,7 +237,6 @@ class _RunState: builder: AgentContextBuilder session_scoped: bool config: _AgentHooksConfig - halted: BaseException | None = None _RUN_STATE: ContextVar[_RunState | None] = ContextVar("agent_framework_agent_hooks_run_state", default=None) @@ -894,27 +900,63 @@ def _is_approval_request(result: Any) -> bool: return isinstance(result, Content) and result.type == "function_approval_request" -def _halt_on_enforcement_failure( - state: _RunState, context: FunctionInvocationContext, exc: BaseException, point: str -) -> NoReturn: - """Route an unexpected failure inside the enforcement layer through the fail-closed halt path. +def _halt_on_enforcement_failure(exc: BaseException, point: str) -> NoReturn: + """Abort the run fail-closed on an unexpected failure inside the enforcement layer. The function-invocation loop converts arbitrary exceptions raised by function middleware into tool-error results and keeps running; for a failure of the enforcement layer itself (projection bug, emitter fault) that would fail open — the failure would vanish from the audit trail and the run would continue unguarded. - Instead the loop is stopped via ``MiddlewareTermination`` (its only loud escape) and - the agent middleware re-raises the failure to the caller at the run boundary. + :class:`MiddlewareFailure` is the loop's explicit fail-closed escape: it propagates + to the caller at the run boundary, with the underlying failure chained as its cause. """ - message = f"agent-hooks {point} enforcement failed: {type(exc).__name__}" - context.result = {"error": message} if isinstance(exc, MiddlewareException): - failure: BaseException = exc - else: - failure = MiddlewareException(message) - failure.__cause__ = exc - state.halted = failure - raise MiddlewareTermination(message) from exc + raise MiddlewareFailure(str(exc)) from exc + raise MiddlewareFailure(f"agent-hooks {point} enforcement failed: {type(exc).__name__}") from exc + + +class _ToolSeamBlockFailure(MiddlewareFailure): + """Private tag for this feature's own tool-seam ``host_error`` halts. + + Only failures raised by :meth:`_AgentHooksFunctionMiddleware._maybe_halt` carry + this type, which is what authorizes :func:`_reraise_tool_seam_block` to unwrap + the chained :class:`agent_hooks.InterceptionBlocked` at the run boundary. A plain + ``MiddlewareFailure`` raised by third-party middleware is never unwrapped — even + when its ``__cause__`` happens to be an ``InterceptionBlocked`` — so untrusted + code cannot launder a crafted interception record into this feature's deny + surface. + """ + + +def _reraise_tool_seam_block(failure: MiddlewareFailure) -> NoReturn: + """Surface a tool-seam ``host_error`` block as the block itself at the run boundary. + + The tool seam sits behind the function-invocation loop, so its fail-closed halts + travel as :class:`MiddlewareFailure` (the loop's loud escape) with the triggering + exception chained as the cause. When the failure is this feature's own tagged + :class:`_ToolSeamBlockFailure`, re-raise its :class:`agent_hooks.InterceptionBlocked` + cause so callers see the same deny surface — ``InterceptionBlocked`` carrying the + interception record — at every seam (run-, model-, and tool-level). Any other + failure (including a third-party ``MiddlewareFailure`` with a crafted + ``InterceptionBlocked`` cause) propagates exactly as raised. + """ + from agent_hooks import InterceptionBlocked + + block = failure.__cause__ + if isinstance(failure, _ToolSeamBlockFailure) and isinstance(block, InterceptionBlocked): + # Detach the transport wrapper's back-links (both were set to the block when + # _maybe_halt raised the wrapper from it) before re-raising: re-raising the + # block while they are intact would make the two exceptions each other's + # cause/context — a chain cycle that every consumer walking + # __cause__/__context__ would have to guard against. The bare raise below + # records the wrapper as the block's __context__ instead (truthful: the block + # is re-raised while the wrapper is being handled), keeping both exceptions + # visible in tracebacks, acyclically. + failure.__cause__ = None + if failure.__context__ is block: + failure.__context__ = None + raise block + raise failure # endregion @@ -1024,8 +1066,6 @@ async def process(self, context: AgentContext, call_next: Callable[[], Awaitable # A middleware short-circuited the run; any substituted result # still egresses to the caller, so it passes the output point. termination = exc - if state.halted is not None: - raise state.halted result = context.result if isinstance(result, AgentResponse): await self._emit_output(state, result) @@ -1045,6 +1085,14 @@ async def process(self, context: AgentContext, call_next: Callable[[], Awaitable shutdown_reason = "error" context.result = None raise + except MiddlewareFailure as failure: + # A fail-closed abort from behind the function-invocation loop: the + # gated persistence is dropped and, for a tool-seam host_error block, + # the block itself is surfaced (one deny surface at every seam). + gate.drop() + shutdown_reason = "error" + context.result = None + _reraise_tool_seam_block(failure) except asyncio.CancelledError: shutdown_reason = "cancelled" raise @@ -1092,10 +1140,6 @@ async def _process_streaming( termination = exc inner = context.result if inner is None and termination is not None: - if state.halted is not None: - # The enforcement layer itself failed: strand the deferred - # persistence (fail-closed) and surface the halt. - raise state.halted # Terminated without a result: nothing will egress, so the no-egress # termination is a permitted outcome. Release the persistence the # drained in-pipeline work deferred — history of model calls that @@ -1118,6 +1162,12 @@ async def _process_streaming( # A middleware substituted its own stream: it is guarded, and the # termination still short-circuits the rest of the pipeline. raise termination + except MiddlewareFailure as failure: + # A fail-closed abort during the pipeline descent (e.g. a retry middleware + # drained an attempt whose tool seam hit a host_error block): surface the + # block itself for tool-seam blocks (one deny surface at every seam). The + # deferred persistence is stranded un-flushed — fail-closed. + _reraise_tool_seam_block(failure) except asyncio.CancelledError: shutdown_reason = "cancelled" raise @@ -1150,6 +1200,14 @@ async def _consume() -> tuple[Sequence[AgentResponseUpdate], AgentResponse[Any]] except asyncio.CancelledError: await self._emit_shutdown(state, "cancelled") raise + except MiddlewareFailure as failure: + # A fail-closed abort from behind the function-invocation loop: close + # the trail and, for a tool-seam host_error block, surface the block + # itself (one deny surface at every seam). The deferred persistence + # is dropped — denied content never becomes durable. + gate_handle.drop() + await self._emit_shutdown(state, "error") + _reraise_tool_seam_block(failure) except BaseException: await self._emit_shutdown(state, "error") raise @@ -1163,8 +1221,6 @@ async def _gate( from agent_hooks import InterceptionBlocked try: - if state.halted is not None: - raise state.halted if not isinstance(final, AgentResponse): raise MiddlewareException( f"agent-hooks cannot guard a streamed run result of type {type(final).__name__}; " @@ -1322,21 +1378,20 @@ async def _gate(_updates: list[ChatResponseUpdate], final: ChatResponse[Any]) -> class _AgentHooksFunctionMiddleware(_AgentHooksMiddlewareBase, FunctionMiddleware): """Tool bracket: ``pre_tool_call`` and ``post_tool_call``.""" - def _block( - self, state: _RunState, context: FunctionInvocationContext, exc: InterceptionBlocked, point: str - ) -> None: + def _block(self, context: FunctionInvocationContext, exc: InterceptionBlocked, point: str) -> None: """Enforce a tool-seam deny: surface a tool error and, on host errors, halt the run.""" context.result = _blocked_tool_result(point, exc.result) - self._maybe_halt(state, exc, point) + self._maybe_halt(exc, point) - def _maybe_halt(self, state: _RunState, exc: InterceptionBlocked, point: str) -> None: + def _maybe_halt(self, exc: InterceptionBlocked, point: str) -> None: record: InterceptionRecord = exc.result if _is_host_error(record): # The enforcement layer itself failed (interceptor crash/timeout, invalid - # context): continuing the loop would run unguarded. Halt the run; the - # agent middleware re-raises the block to the caller. - state.halted = exc - raise MiddlewareTermination( + # context): continuing the loop would run unguarded. Abort the run through + # the loop's fail-closed escape; the agent middleware re-raises the block + # itself to the caller (the private tag is what authorizes the unwrap — + # see _reraise_tool_seam_block). + raise _ToolSeamBlockFailure( f"agent-hooks {point} failed closed: {record.verdict.reason}", ) from exc @@ -1345,21 +1400,15 @@ async def process(self, context: FunctionInvocationContext, call_next: Callable[ state = _RUN_STATE.get() if state is None: - # No run state means the bundle's agent middleware never ran. A plain - # exception raised here would be converted into a tool error by the - # function-invocation loop and the run would continue unguarded (fail - # open); MiddlewareTermination is the only loud escape: the tool is never - # dispatched and the loop stops. - message = _TRIO_REQUIRED_MESSAGE.format(seam="function") - context.result = {"error": message} - raise MiddlewareTermination(message) + # No run state means the bundle's agent middleware never ran (a partial + # install the public API makes impossible). The tool must never be + # dispatched: abort the run through the loop's fail-closed escape. + raise MiddlewareFailure(_TRIO_REQUIRED_MESSAGE.format(seam="function")) if not self._shares_config(state.config): # A different bundle owns the innermost run state (stacked bundles): - # binding to it would silently misroute emissions. Halt the run - # fail-closed (the loop swallows plain exceptions, so route through the - # halt path). + # binding to it would silently misroute emissions. Abort fail-closed. _halt_on_enforcement_failure( - state, context, MiddlewareException(_FOREIGN_TRIO_MESSAGE.format(seam="function")), "pre_tool_call" + MiddlewareException(_FOREIGN_TRIO_MESSAGE.format(seam="function")), "pre_tool_call" ) try: @@ -1378,23 +1427,20 @@ async def process(self, context: FunctionInvocationContext, call_next: Callable[ args = effective except InterceptionBlocked as exc: # §6.2: the tool is not dispatched and no post_tool_call is emitted. - self._block(state, context, exc, "pre_tool_call") + self._block(context, exc, "pre_tool_call") return - except (MiddlewareTermination, asyncio.CancelledError): + except (MiddlewareTermination, MiddlewareFailure, asyncio.CancelledError): raise except BaseException as exc: - _halt_on_enforcement_failure(state, context, exc, "pre_tool_call") + _halt_on_enforcement_failure(exc, "pre_tool_call") termination: MiddlewareTermination | None = None try: await call_next() except MiddlewareTermination as exc: - if state.halted is not None or _is_approval_request(context.result): - # Our own halt path, or framework approval control flow (the tool did - # not run; an approved replay re-enters through pre_tool_call). - raise # A middleware short-circuited with a substituted result; that result - # still enters the transcript, so it is bracketed below. + # still enters the transcript, so it is bracketed below (approval-request + # control flow is passed through un-emitted further down instead). termination = exc except asyncio.CancelledError: raise @@ -1410,19 +1456,21 @@ async def process(self, context: FunctionInvocationContext, call_next: Callable[ except InterceptionBlocked as blocked: # A policy deny over an already-errored call changes nothing (the # result is discarded either way); a host error still halts the run. - self._maybe_halt(state, blocked, "post_tool_call") - except asyncio.CancelledError: + self._maybe_halt(blocked, "post_tool_call") + except (MiddlewareTermination, MiddlewareFailure, asyncio.CancelledError): raise except BaseException as emit_exc: - _halt_on_enforcement_failure(state, context, emit_exc, "post_tool_call") + _halt_on_enforcement_failure(emit_exc, "post_tool_call") raise if _is_approval_request(context.result): - # Framework approval control flow on the normal return path (a middleware - # set an approval request and returned): the tool has not run, so there is - # no result to bracket — pass the control object through un-emitted, exactly - # like the termination branch above. The approved replay re-enters through - # pre_tool_call. + # Framework approval control flow (a middleware set an approval request, + # whether it returned or short-circuited): the tool has not run, so there + # is no result to bracket — pass the control object through un-emitted. + # The approved replay re-enters through pre_tool_call. A short-circuit is + # re-raised so the loop still stops and surfaces the request. + if termination is not None: + raise termination return try: @@ -1433,11 +1481,11 @@ async def process(self, context: FunctionInvocationContext, call_next: Callable[ context.result = _ToolResultCodec.write_back(context.result, value, outcome.target) except InterceptionBlocked as exc: # §6.1: the result must be discarded as if the call had errored. - self._block(state, context, exc, "post_tool_call") - except (MiddlewareTermination, asyncio.CancelledError): + self._block(context, exc, "post_tool_call") + except (MiddlewareTermination, MiddlewareFailure, asyncio.CancelledError): raise except BaseException as exc: - _halt_on_enforcement_failure(state, context, exc, "post_tool_call") + _halt_on_enforcement_failure(exc, "post_tool_call") if termination is not None: raise termination diff --git a/python/packages/core/agent_framework/_middleware.py b/python/packages/core/agent_framework/_middleware.py index 77b452214e..593404d14c 100644 --- a/python/packages/core/agent_framework/_middleware.py +++ b/python/packages/core/agent_framework/_middleware.py @@ -82,6 +82,61 @@ def __init__(self, message: str = "Middleware terminated execution.", *, result: self.result = result +class MiddlewareFailure(MiddlewareException): + """Fatal middleware signal that aborts the run instead of being absorbed. + + Ordinary exceptions raised by **function** middleware (or by the tool it wraps) are + converted into tool-error results by the function-invocation loop, which then keeps + running — appropriate for recoverable tool failures, but fail-open for enforcement + layers and guardrails. ``MiddlewareFailure`` is the loop's explicit fail-closed + escape: it is never converted into a tool result, the current batch of concurrent + tool calls is cancelled, no further tool call starts, and the exception propagates + to the caller of :meth:`Agent.run` (for streaming runs, it is raised when the + stream is consumed). Cancellation is cooperative: an async sibling stops at its + next suspension point, while a synchronous tool body already executing in a worker + thread cannot be interrupted and may still complete its side effects — its result + is discarded either way and never reaches the transcript, the model, or history. + On a service-managed conversation (a persisted conversation id), the loop first + settles the aborted batch by submitting one error ``function_result`` per dangling + call — one extra request whose response is discarded — so the hosted thread is not + left with unresolved tool calls that would make the session's next request fail. + + Agent and chat middleware do not need a dedicated signal — every exception they + raise already propagates to the caller — and ``MiddlewareFailure`` behaves the same + there, so one exception type gives uniform fail-loud semantics across all three + middleware categories. + + Contrast with :class:`MiddlewareTermination`, which stops the loop *gracefully* + (optionally substituting a result that still flows back to the caller): + ``MiddlewareFailure`` produces no result at all. + + Middleware must not catch ``MiddlewareFailure`` (let it propagate through + ``call_next()``): swallowing it converts a fail-closed abort back into a running — + and possibly unguarded — loop. + + Chain the underlying error so it reaches the caller intact: + + .. code-block:: python + + from agent_framework import FunctionMiddleware, FunctionInvocationContext, MiddlewareFailure + + + class EnforcementMiddleware(FunctionMiddleware): + async def process(self, context: FunctionInvocationContext, call_next): + try: + verdict = await self.check(context.arguments) + except Exception as exc: + # The enforcement layer itself failed: abort instead of running unguarded. + raise MiddlewareFailure("policy check failed") from exc + if verdict.deny: + raise MiddlewareFailure(f"denied: {verdict.reason}") + await call_next() + """ + + def __init__(self, message: str = "Middleware failed.") -> None: + super().__init__(message, log_level=None) + + class MiddlewareType(str, Enum): """Enum representing the type of middleware. @@ -544,6 +599,13 @@ class FunctionMiddleware(ABC): FunctionMiddleware is an abstract base class. You must subclass it and implement the ``process()`` method to create custom function middleware. + Note: + Exception semantics inside the function-invocation loop: an ordinary exception + raised from function middleware is converted into a tool-error result and the + loop keeps running; raise :class:`MiddlewareTermination` to stop the loop + gracefully (optionally substituting a result), or :class:`MiddlewareFailure` to + abort the run fail-closed and propagate the failure to the caller. + Examples: .. code-block:: python diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index e7760c9cbd..0de6e7e249 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -1458,6 +1458,10 @@ async def _auto_invoke_function( Raises: KeyError: If the requested function is not found in the tool map. MiddlewareTermination: If middleware requests loop termination. + MiddlewareFailure: If middleware (or the tool) aborts the run fail-closed. + Unlike ordinary exceptions, which are converted into tool-error results, + this explicit signal is re-raised so it propagates to the run's caller. + UserInputRequiredException: If the tool requires user input to proceed. """ from ._types import Content @@ -1532,7 +1536,7 @@ async def _auto_invoke_function( additional_properties=function_call_content.additional_properties, ) - from ._middleware import FunctionInvocationContext + from ._middleware import FunctionInvocationContext, MiddlewareFailure if middleware_pipeline is None or not middleware_pipeline.has_middlewares: # No middleware - execute directly @@ -1556,7 +1560,9 @@ async def _auto_invoke_function( result=function_result, additional_properties=function_call_content.additional_properties, ) - except UserInputRequiredException: + except (MiddlewareFailure, UserInputRequiredException): + # Explicit control-flow signals escape the loop; only ordinary exceptions + # are absorbed into tool-error results below. raise except Exception as exc: return _function_execution_error_result(function_call_content, tool.name, exc, config) @@ -1620,7 +1626,10 @@ async def final_function_handler(context_obj: Any) -> Any: additional_properties=function_call_content.additional_properties, ) raise - except UserInputRequiredException: + except (MiddlewareFailure, UserInputRequiredException): + # MiddlewareFailure is the loop's explicit fail-closed escape: middleware that + # must abort the run (enforcement layers, guardrails) raises it instead of + # relying on the tool-error conversion below, and it propagates to the caller. raise except Exception as exc: return _function_execution_error_result(function_call_content, tool.name, exc, config) @@ -1844,7 +1853,20 @@ async def _try_execute_function_call_groups( ) for function_call in function_calls ] - execution_results = await asyncio.gather(*execution_tasks) + try: + execution_results = await asyncio.gather(*execution_tasks) + except BaseException: + # A loud escape from one call (e.g. MiddlewareFailure aborting the run + # fail-closed) fails the whole batch: cancel in-flight siblings and wait for + # them so no new tool work starts after the loop is abandoned. Cancellation + # is cooperative — a synchronous tool body already running in a worker thread + # (asyncio.to_thread) cannot be interrupted and may complete its side effects, + # but its result is discarded with the batch and never reaches the transcript, + # the model, or history. + for task in execution_tasks: + task.cancel() + await asyncio.gather(*execution_tasks, return_exceptions=True) + raise should_terminate = any(terminate for _, terminate in execution_results) return [result_contents for result_contents, _ in execution_results], should_terminate @@ -2619,6 +2641,59 @@ def _prepare_messages_for_next_iteration(prepared_messages: list[Message], respo prepared_messages[:] = response.messages[-1:] +async def _settle_dangling_service_function_calls( + *, + super_get_response: Callable[..., Any], + response: ChatResponse[Any], + prepared_messages: list[Message], + options: dict[str, Any], + request_kwargs: dict[str, Any], + compaction_strategy: CompactionStrategy | None, + tokenizer: TokenizerProtocol | None, +) -> None: + """Resolve a failed batch's function calls on a service-managed conversation. + + When ``MiddlewareFailure`` aborts a tool batch, the local run raises before any + result exists — but on a service-managed conversation the continuation state + (``session.service_session_id``) was already persisted when the model turn + completed, so the hosted thread ends with unresolved ``function_call`` items. + OpenAI-style continuations reject the next request over such a thread (missing + tool output), which would leave the session permanently stuck after a routine + policy abort. Settle the thread by submitting one error ``function_result`` per + dangling call (with ``tool_choice="none"`` so no new calls are requested); the + settlement response is discarded locally — the run still fails with the original + ``MiddlewareFailure``. Costs one extra request, only on the failure path and only + when a service-managed conversation is in play. + """ + from ._types import Content, Message + + if response.conversation_id is None and not options.get("conversation_id"): + return + error_results = [ + Content.from_function_result( + call_id=function_call.call_id, + result="Error: Tool execution was aborted by middleware before a result was produced.", + exception="MiddlewareFailure", + additional_properties=function_call.additional_properties, + ) + for function_call in _extract_function_calls(response) + if function_call.call_id is not None + ] + if not error_results: + return + response.messages.append(Message(role="tool", contents=error_results)) + _prepare_messages_for_next_iteration(prepared_messages, response) + options["tool_choice"] = "none" + await super_get_response( + messages=prepared_messages, + stream=False, + options=options, + compaction_strategy=compaction_strategy, + tokenizer=tokenizer, + client_kwargs=request_kwargs, + ) + + @dataclass class _FunctionProcessingResult: """Control data produced while resolving or executing function calls.""" @@ -2897,6 +2972,7 @@ async def _get_response_with_function_invocation( max_errors: int, ) -> ChatResponse[Any]: """Run the non-streaming function invocation loop.""" + from ._middleware import MiddlewareFailure from ._types import ChatResponse, add_usage_details errors_in_a_row = 0 @@ -2960,14 +3036,37 @@ async def _get_response_with_function_invocation( options=options, ) - function_processing = await _process_model_function_calls( - response=response, - options=options, - function_call_messages=function_call_messages, - errors_in_a_row=errors_in_a_row, - max_errors=max_errors, - execute_function_calls=execute_function_calls, - ) + try: + function_processing = await _process_model_function_calls( + response=response, + options=options, + function_call_messages=function_call_messages, + errors_in_a_row=errors_in_a_row, + max_errors=max_errors, + execute_function_calls=execute_function_calls, + ) + except MiddlewareFailure: + # Fail-closed abort: before propagating, settle the batch's calls on a + # service-managed conversation so the persisted continuation state does + # not point at a thread with unresolved function calls (best-effort — + # a settlement failure never masks the abort). + try: + await _settle_dangling_service_function_calls( + super_get_response=super_get_response, + response=response, + prepared_messages=prepared_messages, + options=options, + request_kwargs=request_kwargs, + compaction_strategy=compaction_strategy, + tokenizer=tokenizer, + ) + except Exception: + logger.warning( + "Failed to settle dangling function calls on the service-managed conversation; " + "the next request over this conversation may be rejected by the service.", + exc_info=True, + ) + raise total_function_calls = _record_function_calls( budget_state, total_function_calls, @@ -3029,6 +3128,8 @@ async def _stream_response_with_function_invocation( max_errors: int, ) -> AsyncIterable[ChatResponseUpdate]: """Run the streaming function invocation loop.""" + from ._middleware import MiddlewareFailure + errors_in_a_row = 0 total_function_calls = int(budget_state.get("total_function_calls", 0) or 0) max_function_calls = self.function_invocation_configuration.get("max_function_calls") @@ -3110,14 +3211,35 @@ async def _stream_response_with_function_invocation( yield _function_invocation_limit_fallback_update() return - function_processing = await _process_model_function_calls( - response=response, - options=options, - function_call_messages=None, - errors_in_a_row=errors_in_a_row, - max_errors=max_errors, - execute_function_calls=execute_function_calls, - ) + try: + function_processing = await _process_model_function_calls( + response=response, + options=options, + function_call_messages=None, + errors_in_a_row=errors_in_a_row, + max_errors=max_errors, + execute_function_calls=execute_function_calls, + ) + except MiddlewareFailure: + # See the non-streaming loop: settle a service-managed conversation's + # dangling calls before propagating the fail-closed abort. + try: + await _settle_dangling_service_function_calls( + super_get_response=super_get_response, + response=response, + prepared_messages=prepared_messages, + options=options, + request_kwargs=request_kwargs, + compaction_strategy=compaction_strategy, + tokenizer=tokenizer, + ) + except Exception: + logger.warning( + "Failed to settle dangling function calls on the service-managed conversation; " + "the next request over this conversation may be rejected by the service.", + exc_info=True, + ) + raise errors_in_a_row = function_processing.errors_in_a_row total_function_calls = _record_function_calls( budget_state, diff --git a/python/packages/core/tests/core/test_agent_hooks.py b/python/packages/core/tests/core/test_agent_hooks.py index 347d7bae4f..3ada03a5c0 100644 --- a/python/packages/core/tests/core/test_agent_hooks.py +++ b/python/packages/core/tests/core/test_agent_hooks.py @@ -25,6 +25,7 @@ Message, MiddlewareBundle, MiddlewareException, + MiddlewareFailure, MiddlewareTermination, ResponseStream, create_agent_hooks_middleware, @@ -690,6 +691,184 @@ async def test_interceptor_crash_fails_closed_and_halts_run(chat_client_base: Mo assert points(records)[-1] == "agent_shutdown" +@requires_sdk +async def test_interceptor_crash_at_tool_seam_fails_closed_streaming(chat_client_base: MockBaseChatClient) -> None: + # The streaming twin of the halt above: the tool-seam host_error block travels the + # function-invocation loop as MiddlewareFailure and is surfaced to the stream + # consumer as the InterceptionBlocked itself — one deny surface at every seam. + records: list[InterceptionRecord] = [] + chat_client_base.streaming_responses = [ + [ + ChatResponseUpdate( + contents=[ + Content.from_function_call( + call_id="call_1", name="weather_tool", arguments='{"location": "Seattle"}' + ) + ], + role="assistant", + finish_reason="tool_calls", + ) + ], + ] + agent = Agent( + client=chat_client_base, + tools=[weather_tool], + middleware=[create_agent_hooks_middleware([CrashingGuard("post_tool_call")], record_sink=records.append)], + ) + + updates: list[AgentResponseUpdate] = [] + with pytest.raises(InterceptionBlocked) as exc_info: + async for update in agent.run("get the weather", stream=True): + updates.append(update) + + assert exc_info.value.result.verdict.reason == "host_error:interceptor_failed" + assert updates == [] # nothing egressed before the halt + assert points(records)[-1] == "agent_shutdown" + + +@requires_sdk +async def test_tool_seam_block_exception_chain_is_acyclic(chat_client_base: MockBaseChatClient) -> None: + # Re-raising the InterceptionBlocked with the transport wrapper's back-links + # intact would make the two exceptions each other's cause/context — a chain + # cycle every __cause__/__context__ walker would have to guard against. The + # unwrap must detach the wrapper first, keeping both exceptions visible in a + # finite traceback. + import traceback + + chat_client_base.run_responses = [tool_call_response(), final_response()] + agent = Agent( + client=chat_client_base, + tools=[weather_tool], + middleware=[create_agent_hooks_middleware([CrashingGuard("post_tool_call")])], + ) + + with pytest.raises(InterceptionBlocked) as exc_info: + await agent.run("get the weather") + + block = exc_info.value + seen: set[int] = set() + node: BaseException | None = block + while node is not None: + assert id(node) not in seen, "exception chain contains a cycle" + seen.add(id(node)) + # Follow the chain the way traceback rendering does. + node = node.__cause__ if (node.__cause__ is not None or node.__suppress_context__) else node.__context__ + + formatted = "".join(traceback.format_exception(type(block), block, block.__traceback__)) + assert "InterceptionBlocked" in formatted + # The loop-transport wrapper stays visible as context, acyclically. + assert "failed closed" in formatted + + +@requires_sdk +async def test_third_party_middleware_failure_is_bracketed_and_propagates( + chat_client_base: MockBaseChatClient, +) -> None: + # A MiddlewareFailure raised by another (inner) function middleware is not + # agent-hooks' own halt: the tool bracket still closes with is_error=True (only + # the exception type name crosses the boundary) and the failure itself propagates + # to the caller un-unwrapped. + class InnerEnforcement(FunctionMiddleware): + async def process(self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]]) -> None: + raise MiddlewareFailure("inner enforcement denied") + + guard = AllowGuard() + records: list[InterceptionRecord] = [] + chat_client_base.run_responses = [tool_call_response(), final_response()] + agent = Agent( + client=chat_client_base, + tools=[weather_tool], + middleware=[create_agent_hooks_middleware([guard], record_sink=records.append), InnerEnforcement()], + ) + + with pytest.raises(MiddlewareFailure, match="inner enforcement denied"): + await agent.run("get the weather") + + assert weather_tool_calls == [] + post_tool = guard.contexts_for("post_tool_call")[0] + assert post_tool["tool_result"]["is_error"] is True + assert post_tool["tool_result"]["value"] == "MiddlewareFailure" + assert points(records)[-1] == "agent_shutdown" + + +@requires_sdk +async def test_third_party_crafted_interception_cause_is_not_unwrapped(chat_client_base: MockBaseChatClient) -> None: + # Adversarial probe: only this feature's own tagged tool-seam halts authorize + # unwrapping the chained InterceptionBlocked at the run boundary. A third-party + # MiddlewareFailure whose __cause__ is a crafted InterceptionBlocked must surface + # AS the MiddlewareFailure — otherwise untrusted middleware could launder an + # attacker-shaped interception record into this feature's audit-bearing deny + # surface. + from agent_hooks import EnforcementMode, InterceptionPoint + + crafted = InterceptionBlocked( + InterceptionRecord( + interception_point=InterceptionPoint.PRE_TOOL_CALL, + mode=EnforcementMode.ENFORCE, + verdict=Verdict.deny(reason="forged_deny"), + input_identity=None, + enforced_identity=None, + ) + ) + + class Laundering(FunctionMiddleware): + async def process(self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]]) -> None: + raise MiddlewareFailure("third-party failure") from crafted + + chat_client_base.run_responses = [tool_call_response(), final_response()] + agent = Agent( + client=chat_client_base, + tools=[weather_tool], + middleware=[create_agent_hooks_middleware([AllowGuard()]), Laundering()], + ) + + with pytest.raises(MiddlewareFailure, match="third-party failure"): + await agent.run("get the weather") + + assert weather_tool_calls == [] + + +@requires_sdk +async def test_inner_termination_re_raises_through_after_bracketing(chat_client_base: MockBaseChatClient) -> None: + # Pins the trailing `raise termination` in the function middleware: an inner + # short-circuit is bracketed (post_tool_call over the substituted result) and then + # still propagates as a short-circuit — outer middleware post-call_next code is + # skipped and the loop stops without another model call. + outer_events: list[str] = [] + + class Outer(FunctionMiddleware): + async def process(self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]]) -> None: + outer_events.append("before") + await call_next() + outer_events.append("after") # must be skipped by the re-raised termination + + class InnerShortCircuit(FunctionMiddleware): + async def process(self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]]) -> None: + context.result = "substituted" + raise MiddlewareTermination("stop") + + guard = AllowGuard() + chat_client_base.run_responses = [tool_call_response(), final_response()] + agent = Agent( + client=chat_client_base, + tools=[weather_tool], + middleware=[Outer(), create_agent_hooks_middleware([guard]), InnerShortCircuit()], + ) + + response = await agent.run("get the weather") + + assert outer_events == ["before"] + assert weather_tool_calls == [] + assert chat_client_base.call_count == 1 + # The substituted result was bracketed before the short-circuit propagated. + post_tool = guard.contexts_for("post_tool_call")[0] + assert post_tool["tool_result"]["value"] == "substituted" + results = [ + content for message in response.messages for content in message.contents if content.type == "function_result" + ] + assert len(results) == 1 + + @requires_sdk async def test_interceptor_crash_at_input_fails_closed(chat_client_base: MockBaseChatClient) -> None: agent = Agent(client=chat_client_base, middleware=[create_agent_hooks_middleware([CrashingGuard("input")])]) @@ -1167,7 +1346,8 @@ async def test_enforcement_failure_at_function_seam_halts_run(chat_client_base: async def test_function_seam_without_run_state_blocks_tool(chat_client_base: MockBaseChatClient) -> None: # The bundle makes a partial install impossible through the public API; this # exercises the internal defense directly: the private function middleware invoked - # without an active agent-hooks run must never dispatch the tool. + # without an active agent-hooks run must never dispatch the tool. The run aborts + # loudly through the loop's fail-closed escape (MiddlewareFailure). bundle = create_agent_hooks_middleware([AllowGuard()]) chat_client_base.run_responses = [tool_call_response(), final_response()] agent = Agent( @@ -1176,13 +1356,12 @@ async def test_function_seam_without_run_state_blocks_tool(chat_client_base: Moc middleware=[_bundle_member(bundle, "FunctionMiddleware")], ) - response = await agent.run("get the weather") + with pytest.raises(MiddlewareFailure, match="without an active agent-hooks run"): + await agent.run("get the weather") - # The tool is never dispatched and the loop terminates instead of continuing. + # The tool is never dispatched and the loop stops instead of continuing. assert weather_tool_calls == [] assert chat_client_base.call_count == 1 - transcript = str([content.result for message in response.messages for content in message.contents]) - assert "without an active agent-hooks run" in transcript @requires_sdk @@ -1190,7 +1369,7 @@ async def test_bundle_passed_to_chat_client_call_raises_instead_of_dropping_the_ chat_client_base: MockBaseChatClient, ) -> None: # The chat-client middleware seam installs only chat and function middleware; the - # bundle's agent member carries the output gate and the halt re-raise, so silently + # bundle's agent member carries the output gate and the deny surface, so silently # dropping it would install partial enforcement. The seam must raise instead. bundle = create_agent_hooks_middleware([AllowGuard()]) with pytest.raises(MiddlewareException, match="cannot be partially installed"): @@ -2231,6 +2410,45 @@ async def process(self, context: FunctionInvocationContext, call_next: Callable[ assert len(approval_requests) == 1 +@requires_sdk +async def test_approval_request_on_termination_path_passes_through(chat_client_base: MockBaseChatClient) -> None: + class ApprovalGate(FunctionMiddleware): + """Framework pattern: request human approval and short-circuit the pipeline.""" + + async def process(self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]]) -> None: + context.result = Content.from_function_approval_request( + id=str(context.metadata.get("call_id")), + function_call=Content.from_function_call( + str(context.metadata.get("call_id")), context.function.name, arguments={} + ), + ) + raise MiddlewareTermination("needs approval") + + records: list[InterceptionRecord] = [] + chat_client_base.run_responses = [tool_call_response(), final_response()] + agent = Agent( + client=chat_client_base, + tools=[weather_tool], + middleware=[create_agent_hooks_middleware([AllowGuard()], record_sink=records.append), ApprovalGate()], + ) + + response = await agent.run("get the weather") + + # The tool never ran and the short-circuit still stopped the loop: the control + # object is passed through un-bracketed (no post_tool_call reporting a value for + # a tool that never executed) and surfaces to the caller. + assert weather_tool_calls == [] + assert "post_tool_call" not in points(records) + assert chat_client_base.call_count == 1 + approval_requests = [ + content + for message in response.messages + for content in message.contents + if content.type == "function_approval_request" + ] + assert len(approval_requests) == 1 + + # endregion # region Tool-call transforms at post_model_call diff --git a/python/packages/core/tests/core/test_middleware_with_agent.py b/python/packages/core/tests/core/test_middleware_with_agent.py index 708951a88f..dce572e9ce 100644 --- a/python/packages/core/tests/core/test_middleware_with_agent.py +++ b/python/packages/core/tests/core/test_middleware_with_agent.py @@ -1,5 +1,7 @@ # Copyright (c) Microsoft. All rights reserved. +import asyncio +import threading from collections.abc import Awaitable, Callable from typing import Any, cast @@ -21,6 +23,7 @@ FunctionTool, Message, MiddlewareException, + MiddlewareFailure, MiddlewareTermination, MiddlewareType, SupportsChatGetResponse, @@ -2398,3 +2401,435 @@ async def __call__(self, arg1: Any, arg2: Any) -> None: assert "UndeterminedCallableMiddleware" in str(exc_info.value) assert "Cannot determine middleware type" in str(exc_info.value) + + +# region MiddlewareFailure fail-closed escape + + +def _tool_call_response(name: str = "sample_tool_function", call_id: str = "call_1") -> ChatResponse: + return ChatResponse( + messages=[ + Message( + role="assistant", + contents=[Content.from_function_call(call_id=call_id, name=name, arguments='{"location": "Seattle"}')], + ) + ] + ) + + +class TestMiddlewareFailure: + """The explicit fail-closed escape from the function-invocation loop. + + Ordinary exceptions raised by function middleware are converted into tool-error + results and the loop keeps running (a public behavior, pinned below); only the + explicit ``MiddlewareFailure`` signal escapes the loop and propagates to the + ``run()`` caller. + """ + + async def test_failure_before_tool_aborts_run(self, chat_client_base: "MockBaseChatClient") -> None: + executed: list[str] = [] + + def tool_impl(location: str) -> str: + executed.append(location) + return "ran" + + tracked_tool = FunctionTool(func=tool_impl, name="sample_tool_function", approval_mode="never_require") + + class Enforcement(FunctionMiddleware): + async def process( + self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]] + ) -> None: + raise MiddlewareFailure("policy denied") + + chat_client_base.run_responses = [ + _tool_call_response(), + ChatResponse(messages=[Message(role="assistant", contents=["Final"])]), + ] + agent = Agent(client=chat_client_base, middleware=[Enforcement()], tools=[tracked_tool]) + + with pytest.raises(MiddlewareFailure, match="policy denied"): + await agent.run("get the weather") + + # Fail-closed: the tool never executed and no further model turn was consumed. + assert executed == [] + assert chat_client_base.call_count == 1 + + async def test_failure_after_tool_aborts_run_before_next_model_turn( + self, chat_client_base: "MockBaseChatClient" + ) -> None: + executed: list[str] = [] + + def tool_impl(location: str) -> str: + executed.append(location) + return "ran" + + tracked_tool = FunctionTool(func=tool_impl, name="sample_tool_function", approval_mode="never_require") + + class PostCheck(FunctionMiddleware): + async def process( + self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]] + ) -> None: + await call_next() + raise MiddlewareFailure("result rejected") + + chat_client_base.run_responses = [ + _tool_call_response(), + ChatResponse(messages=[Message(role="assistant", contents=["Final"])]), + ] + agent = Agent(client=chat_client_base, middleware=[PostCheck()], tools=[tracked_tool]) + + with pytest.raises(MiddlewareFailure, match="result rejected"): + await agent.run("get the weather") + + # The tool ran once, but its result never fed another model iteration. + assert executed == ["Seattle"] + assert chat_client_base.call_count == 1 + + async def test_failure_cause_chain_reaches_caller(self, chat_client_base: "MockBaseChatClient") -> None: + class Enforcement(FunctionMiddleware): + async def process( + self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]] + ) -> None: + try: + raise ValueError("enforcement backend down") + except ValueError as exc: + raise MiddlewareFailure("enforcement failed") from exc + + chat_client_base.run_responses = [_tool_call_response()] + agent = Agent(client=chat_client_base, middleware=[Enforcement()], tools=[sample_tool_function]) + + with pytest.raises(MiddlewareFailure, match="enforcement failed") as exc_info: + await agent.run("get the weather") + + assert isinstance(exc_info.value.__cause__, ValueError) + + async def test_ordinary_exception_still_becomes_tool_error(self, chat_client_base: "MockBaseChatClient") -> None: + """The loop's absorb-into-tool-error contract for ordinary exceptions is unchanged.""" + + class Broken(FunctionMiddleware): + async def process( + self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]] + ) -> None: + raise RuntimeError("middleware bug") + + chat_client_base.run_responses = [ + _tool_call_response(), + ChatResponse(messages=[Message(role="assistant", contents=["Final"])]), + ] + agent = Agent(client=chat_client_base, middleware=[Broken()], tools=[sample_tool_function]) + + response = await agent.run("get the weather") + + # The exception was converted into a tool-error result and the loop continued. + assert chat_client_base.call_count == 2 + error_results = [ + content + for message in response.messages + for content in message.contents + if content.type == "function_result" and content.exception is not None + ] + assert len(error_results) == 1 + + async def test_failure_from_tool_escapes_without_middleware(self, chat_client_base: "MockBaseChatClient") -> None: + """The direct (no-middleware) execution path honors the same explicit signal.""" + + def tool_impl(location: str) -> str: + raise MiddlewareFailure("tool aborted the run") + + failing_tool = FunctionTool(func=tool_impl, name="sample_tool_function", approval_mode="never_require") + chat_client_base.run_responses = [_tool_call_response()] + agent = Agent(client=chat_client_base, tools=[failing_tool]) + + with pytest.raises(MiddlewareFailure, match="tool aborted the run"): + await agent.run("get the weather") + + assert chat_client_base.call_count == 1 + + async def test_failure_cancels_concurrent_sibling_tool(self, chat_client_base: "MockBaseChatClient") -> None: + """A fatal signal fails the whole batch: in-flight siblings are cancelled.""" + sibling_started = asyncio.Event() + sibling_cancelled: list[bool] = [] + + async def slow_impl(location: str) -> str: + sibling_started.set() + try: + await asyncio.Event().wait() # blocks until cancelled + except asyncio.CancelledError: + sibling_cancelled.append(True) + raise + return "never" + + slow_tool = FunctionTool(func=slow_impl, name="slow_tool", approval_mode="never_require") + + class FailFast(FunctionMiddleware): + async def process( + self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]] + ) -> None: + if context.function.name == "sample_tool_function": + # Fail only after the sibling is genuinely in flight. + await sibling_started.wait() + raise MiddlewareFailure("abort the batch") + await call_next() + + batch_response = ChatResponse( + messages=[ + Message( + role="assistant", + contents=[ + Content.from_function_call( + call_id="call_1", name="sample_tool_function", arguments='{"location": "Seattle"}' + ), + Content.from_function_call(call_id="call_2", name="slow_tool", arguments='{"location": "x"}'), + ], + ) + ] + ) + chat_client_base.run_responses = [batch_response] + agent = Agent(client=chat_client_base, middleware=[FailFast()], tools=[sample_tool_function, slow_tool]) + + with pytest.raises(MiddlewareFailure, match="abort the batch"): + await agent.run("run both tools") + + assert sibling_cancelled == [True] + assert chat_client_base.call_count == 1 + + async def test_failure_with_sync_sibling_discards_late_result(self, chat_client_base: "MockBaseChatClient") -> None: + """Batch cancellation is cooperative: a synchronous sibling cannot be interrupted. + + A synchronous tool body runs in a worker thread (``asyncio.to_thread``); + cancelling its wrapping task cannot stop the thread, so the body may complete + its side effects after the failure has already reached the caller. Its result + is discarded either way — the loop stops at one model call — and failure + propagation is not delayed behind the still-running thread. + """ + sync_started = threading.Event() + sync_release = threading.Event() + sync_completed: list[str] = [] + + def sync_slow(location: str) -> str: + sync_started.set() + sync_release.wait(10) + sync_completed.append("side effect") + return "late sync result" + + slow_tool = FunctionTool(func=sync_slow, name="slow_tool", approval_mode="never_require") + + class FailFast(FunctionMiddleware): + async def process( + self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]] + ) -> None: + if context.function.name == "sample_tool_function": + # Fail only once the synchronous sibling body is genuinely running. + await asyncio.to_thread(sync_started.wait, 5) + raise MiddlewareFailure("abort the batch") + await call_next() + + batch_response = ChatResponse( + messages=[ + Message( + role="assistant", + contents=[ + Content.from_function_call( + call_id="call_1", name="sample_tool_function", arguments='{"location": "Seattle"}' + ), + Content.from_function_call(call_id="call_2", name="slow_tool", arguments='{"location": "x"}'), + ], + ) + ] + ) + chat_client_base.run_responses = [batch_response] + agent = Agent(client=chat_client_base, middleware=[FailFast()], tools=[sample_tool_function, slow_tool]) + + try: + with pytest.raises(MiddlewareFailure, match="abort the batch"): + await agent.run("run both tools") + # The failure reached the caller while the synchronous body was still running. + assert sync_completed == [] + finally: + sync_release.set() + + for _ in range(500): + if sync_completed: + break + await asyncio.sleep(0.01) + # The worker thread survived cancellation and completed its side effect + # (the documented cooperative-cancellation limitation) ... + assert sync_completed == ["side effect"] + # ... but its result went nowhere: the loop never made another model call. + assert chat_client_base.call_count == 1 + + async def test_failure_streaming_reaches_stream_consumer(self, chat_client_base: "MockBaseChatClient") -> None: + executed: list[str] = [] + + def tool_impl(location: str) -> str: + executed.append(location) + return "ran" + + tracked_tool = FunctionTool(func=tool_impl, name="sample_tool_function", approval_mode="never_require") + + class Enforcement(FunctionMiddleware): + async def process( + self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]] + ) -> None: + raise MiddlewareFailure("policy denied") + + chat_client_base.streaming_responses = [ + [ + ChatResponseUpdate( + contents=[ + Content.from_function_call( + call_id="call_1", name="sample_tool_function", arguments='{"location": "Seattle"}' + ) + ], + role="assistant", + ) + ] + ] + agent = Agent(client=chat_client_base, middleware=[Enforcement()], tools=[tracked_tool]) + + with pytest.raises(MiddlewareFailure, match="policy denied"): + async for _ in agent.run("get the weather", stream=True): + pass + + assert executed == [] + assert chat_client_base.call_count == 1 + + async def test_failure_from_agent_middleware_propagates(self, chat_client_base: "MockBaseChatClient") -> None: + """Agent (and chat) middleware exceptions already propagate; the explicit signal behaves the same.""" + + class Guard(AgentMiddleware): + async def process(self, context: AgentContext, call_next: Callable[[], Awaitable[None]]) -> None: + raise MiddlewareFailure("run denied") + + agent = Agent(client=chat_client_base, middleware=[Guard()]) + + with pytest.raises(MiddlewareFailure, match="run denied"): + await agent.run("hello") + + assert chat_client_base.call_count == 0 + + async def test_failure_settles_dangling_calls_on_service_conversation( + self, chat_client_base: "MockBaseChatClient" + ) -> None: + """A service-managed conversation is settled before the failure propagates. + + The continuation state (``session.service_session_id``) is persisted when the + model turn completes — before tool execution — so an aborted batch would leave + the hosted thread ending in unresolved function calls, and OpenAI-style + continuations reject the next request over such a thread. The loop submits one + error ``function_result`` per dangling call (``tool_choice="none"``) and + discards the settlement response; the run still fails. + """ + from agent_framework import AgentSession + + requests: list[dict[str, Any]] = [] + + class Recorder(ChatMiddleware): + async def process(self, context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None: + requests.append({ + "contents": [(m.role, [c for c in m.contents]) for m in context.messages], + "conversation_id": (context.options or {}).get("conversation_id"), + "tool_choice": (context.options or {}).get("tool_choice"), + }) + await call_next() + + class Enforcement(FunctionMiddleware): + async def process( + self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]] + ) -> None: + raise MiddlewareFailure("policy denied") + + tool_turn = _tool_call_response() + tool_turn.conversation_id = "conv_123" + chat_client_base.run_responses = [tool_turn] + session = AgentSession() + agent = Agent(client=chat_client_base, middleware=[Enforcement(), Recorder()], tools=[sample_tool_function]) + + with pytest.raises(MiddlewareFailure, match="policy denied"): + await agent.run("get the weather", session=session) + + # The continuation state was already durable when the batch failed ... + assert session.service_session_id == "conv_123" + # ... so the loop settled the thread: one extra request carrying an error + # function_result for the dangling call, with tool calling disabled. + assert len(requests) == 2 + settlement = requests[1] + assert settlement["conversation_id"] == "conv_123" + assert settlement["tool_choice"] == "none" + settlement_results = [ + content + for _, contents in settlement["contents"] + for content in contents + if content.type == "function_result" + ] + assert [result.call_id for result in settlement_results] == ["call_1"] + assert settlement_results[0].exception == "MiddlewareFailure" + + async def test_failure_settles_service_conversation_streaming(self, chat_client_base: "MockBaseChatClient") -> None: + """The streaming loop settles a service-managed conversation the same way.""" + requests: list[dict[str, Any]] = [] + + class Recorder(ChatMiddleware): + async def process(self, context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None: + requests.append({ + "tool_choice": (context.options or {}).get("tool_choice"), + "messages": list(context.messages), + }) + await call_next() + + class Enforcement(FunctionMiddleware): + async def process( + self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]] + ) -> None: + raise MiddlewareFailure("policy denied") + + chat_client_base.streaming_responses = [ + [ + ChatResponseUpdate( + contents=[ + Content.from_function_call( + call_id="call_1", name="sample_tool_function", arguments='{"location": "Seattle"}' + ) + ], + role="assistant", + conversation_id="conv_123", + ) + ] + ] + agent = Agent(client=chat_client_base, middleware=[Enforcement(), Recorder()], tools=[sample_tool_function]) + + with pytest.raises(MiddlewareFailure, match="policy denied"): + async for _ in agent.run("get the weather", stream=True): + pass + + assert len(requests) == 2 + assert requests[1]["tool_choice"] == "none" + settlement_results = [ + content + for message in requests[1]["messages"] + for content in message.contents + if content.type == "function_result" + ] + assert [result.call_id for result in settlement_results] == ["call_1"] + + async def test_failure_without_service_conversation_makes_no_settlement_request( + self, chat_client_base: "MockBaseChatClient" + ) -> None: + """No service-managed conversation, no settlement cost: the abort stays at one model call.""" + + class Enforcement(FunctionMiddleware): + async def process( + self, context: FunctionInvocationContext, call_next: Callable[[], Awaitable[None]] + ) -> None: + raise MiddlewareFailure("policy denied") + + chat_client_base.run_responses = [_tool_call_response()] + agent = Agent(client=chat_client_base, middleware=[Enforcement()], tools=[sample_tool_function]) + + with pytest.raises(MiddlewareFailure, match="policy denied"): + await agent.run("get the weather") + + assert chat_client_base.call_count == 1 + + +# endregion