Python: add signal protocol for AgentLoopMiddleware autonomous loops

python/packages/core/AGENTS.md

@@ -309,6 +309,7 @@ The vector store API is experimental under the shared `VECTOR_STORES` feature ID
 - **`todos_remaining(*, looping_modes=None)`** / **`todos_remaining_message`** - Helper factories for todo-driven loops (the Python counterpart of .NET's `TodoCompletionLoopEvaluator`), designed for `create_harness_agent` but usable with any agent that registers a `TodoProvider` via `context_providers`. They resolve the `TodoProvider`/`AgentModeProvider` from the *running agent* (`agent.context_providers`, via `_resolve_context_provider`) rather than taking the provider as an argument, so they can be wired directly into `loop_should_continue`/`loop_next_message`. `todos_remaining` returns a `should_continue` predicate that loops while any todo is open; pass `looping_modes=[...]` to gate looping to specific operating modes (case-insensitive; honors the `AgentModeProvider`'s `source_id`/`available_modes`), `looping_modes=None` (default) applies in every mode, and an empty sequence raises `ValueError`. `todos_remaining_message` is a `next_message` callable that lists the still-open todo titles and tells the agent to finish them, returning `None` when the session/agent/provider is unavailable or nothing is open (in which case the middleware's default `None` handling applies: reuse the previous iteration's messages verbatim under the default `fresh_context=False`, or `DEFAULT_NEXT_MESSAGE` only when `fresh_context=True`).
 - **`background_tasks_running()`** / **`background_tasks_running_message`** - Helper factories for background-agent-driven loops, mirroring the `todos_remaining` pair. They resolve the `BackgroundAgentsProvider` from the *running agent* (`agent.context_providers`, via `_resolve_context_provider`) rather than taking the provider as an argument, so they can be wired directly into `create_harness_agent`'s `loop_should_continue`/`loop_next_message`. `background_tasks_running` returns a `should_continue` predicate that loops while the provider's persisted state shows any task with `status == RUNNING` (pair it with `max_iterations` so the loop is bounded even if a task's persisted status is never refreshed). `background_tasks_running_message` is a `next_message` callable that lists the still-running tasks (`#<id> (<agent_name>): <description>`) and tells the agent to wait for them to finish and retrieve their results, returning `None` when the session/agent/provider is unavailable or no task is running.
   - **Approval escape hatch** - `_has_pending_approval_request(result)` checks whether an iteration's response carries a pending tool-approval request (any content with `type == "function_approval_request"`). Both the streaming and non-streaming loops stop and return that response to the caller *before* evaluating `should_continue`/`max_iterations` or injecting `next_message`, so the loop is HITL-safe even when wrapped outermost around a `ToolApprovalMiddleware` (mirrors the C# `LoopAgent`'s `HasPendingApprovalRequests`).
+  - **Signal protocol** (`_harness/_loop.py`) - `AgentLoopMiddleware` stamps `loop_exit_reason` in `AgentResponse.additional_properties` for framework-owned stops (`iteration_cap_reached` when the cap fires, overwriting any pre-existing value; `completed` when the judge from `with_judge` decides the task is answered). A `should_continue` callback may also stamp `loop_exit_reason` directly before returning `False`. See `python/samples/02-agents/harness/signal_protocol.py` for a complete `TASK_COMPLETE:` / `NEED_INPUT:` signal-token parsing example.
   - **Harness integration** - `create_harness_agent` enables the loop when a `loop_should_continue` callable is passed; it prepends `AgentLoopMiddleware(loop_should_continue, max_iterations=loop_max_iterations, next_message=loop_next_message)` ahead of `ToolApprovalMiddleware` so the loop is the outermost middleware (each iteration is a full agent run including tool approval, and the escape hatch hands pending approvals back to the caller). `loop_next_message` and `loop_max_iterations` only take effect together with `loop_should_continue` (with no `loop_should_continue` there is no loop, so they are ignored); `loop_max_iterations` defaults to the loop's default cap (`None` → unbounded).
 
 ### Workflows (`_workflows/`)

python/packages/core/agent_framework/_harness/_loop.py

@@ -23,6 +23,12 @@
    response into ``JudgeVerdict``. The loop continues while the answer is "no". This is a convenience
    wrapper that builds an async ``should_continue`` predicate, so it is a special case of (1).
 
+When ``max_iterations`` fires the loop stamps ``additional_properties["loop_exit_reason"] =
+"iteration_cap_reached"`` on the response; the judge (via :meth:`AgentLoopMiddleware.with_judge`)
+stamps ``"completed"`` when it decides the request was answered. Callers read the exit reason
+directly from ``response.additional_properties["loop_exit_reason"]``. A custom ``should_continue``
+callback may also stamp that key itself before returning ``False``.
+
 In every case, the input for the next iteration is controlled by the ``next_message`` callable.
 """
 
@@ -131,6 +137,10 @@ class JudgeVerdict(BaseModel):
 DEFAULT_JUDGE_MAX_ITERATIONS = 5
 
 
+# Key used in AgentResponse.additional_properties to surface the loop exit reason.
+LOOP_EXIT_REASON_KEY: str = "loop_exit_reason"
+
+
 # A callable invoked between iterations. It always receives the loop keyword arguments
 # (``iteration``, ``last_result``, ``messages``, ``original_messages``, ``session``, ``agent``,
 # ``progress``, ``feedback``). Callers declare only the keywords they need plus ``**kwargs`` to
@@ -210,6 +220,8 @@ async def _judge(
                 answered = False if JUDGE_VERDICT_MORE in text else JUDGE_VERDICT_DONE in text
                 reasoning = response.text.strip()
         # Continue looping while the request is not yet answered, surfacing the reasoning as feedback.
+        if answered:
+            last_result.additional_properties[LOOP_EXIT_REASON_KEY] = "completed"
         return (not answered), (reasoning or None)
 
     def _next_message(*, feedback: str | None = None, **kwargs: Any) -> AgentRunInputs:
@@ -538,6 +550,7 @@ async def _process_non_streaming(
         snapshot: dict[str, Any] | None,
     ) -> None:
         iteration = 0
+        cap_fired = False
         work_iterations = 0
         progress: list[str] = []
         # Aggregated transcript across iterations: each iteration's response messages plus the
@@ -560,6 +573,9 @@ async def _process_non_streaming(
                         f"got {type(result).__name__}."
                     )
 
+                # Clear any loop_exit_reason stamped by an inner agent or nested loop so the outer
+                # loop's own exit-reason decision is not masked by a stale inner value.
+                result.additional_properties.pop(LOOP_EXIT_REASON_KEY, None)
                 final_result = result
                 aggregated.extend(result.messages)
                 if result.usage_details is not None:
@@ -584,6 +600,8 @@ async def _process_non_streaming(
                 # Decide whether to stop and capture any feedback from should_continue first, so the
                 # feedback is available to both the progress and next-message callables this iteration.
                 stop, feedback = await self._evaluate_stop(loop_kwargs, work_iterations)
+                if stop and self.max_iterations is not None and work_iterations >= self.max_iterations:
+                    cap_fired = True
                 loop_kwargs = self._build_loop_kwargs(
                     context=context,
                     iteration=iteration,
@@ -617,6 +635,9 @@ async def _process_non_streaming(
         finally:
             context.options.pop(_LOOP_ITERATION_TOKEN_KEY, None)
 
+        if cap_fired:
+            final_result.additional_properties.setdefault(LOOP_EXIT_REASON_KEY, "iteration_cap_reached")
+
         if not self.return_final_only:
             context.result = self._aggregate_response(final_result, aggregated, aggregated_usage)
         await self._fire_turn_scoped_after_providers(
@@ -638,6 +659,7 @@ def _process_streaming(
 
         async def _generator() -> Any:
             iteration = 0
+            cap_fired = False
             work_iterations = 0
             progress: list[str] = []
             stamped_options = dict(context.options) if context.options is not None else {}
@@ -658,6 +680,9 @@ async def _generator() -> Any:
                             yield update
 
                         holder["final"] = await inner.get_final_response()
+                        # Clear any loop_exit_reason stamped by an inner agent or nested loop so the outer
+                        # loop's own exit-reason decision is not masked by a stale inner value.
+                        holder["final"].additional_properties.pop(LOOP_EXIT_REASON_KEY, None)
                     except MiddlewareTermination:
                         # The pipeline's MiddlewareTermination suppression is no longer active once
                         # process() has returned (the stream is consumed lazily), so a termination
@@ -686,6 +711,8 @@ async def _generator() -> Any:
                     # Decide whether to stop and capture any feedback from should_continue first, so the
                     # feedback is available to both the progress and next-message callables this iteration.
                     stop, feedback = await self._evaluate_stop(loop_kwargs, work_iterations)
+                    if stop and self.max_iterations is not None and work_iterations >= self.max_iterations:
+                        cap_fired = True
                     loop_kwargs = self._build_loop_kwargs(
                         context=context,
                         iteration=iteration,
@@ -721,6 +748,8 @@ async def _generator() -> Any:
                         yield self._message_to_update(message)
             finally:
                 context.options.pop(_LOOP_ITERATION_TOKEN_KEY, None)
+                if cap_fired and holder["final"] is not None:
+                    holder["final"].additional_properties.setdefault(LOOP_EXIT_REASON_KEY, "iteration_cap_reached")
                 await self._fire_turn_scoped_after_providers(context, holder["final"], original_messages)
 
         def _finalize(updates: Sequence[AgentResponseUpdate]) -> AgentResponse:

python/packages/core/tests/core/test_harness_loop.py

@@ -40,6 +40,7 @@
     DEFAULT_JUDGE_MAX_ITERATIONS,
     DEFAULT_MAX_ITERATIONS,
     DEFAULT_NEXT_MESSAGE,
+    LOOP_EXIT_REASON_KEY,
     AgentLoopMiddleware,
 )
 
@@ -1697,3 +1698,151 @@ async def test_concurrent_same_agent_run_during_stream_pause_is_not_suppressed()
 
     # once for the loop boundary, once for the concurrent run's own boundary
     assert turn_scoped.after_calls == 2
+
+
+# --- Framework stop-reason tests ---
+
+
+async def test_loop_sets_iteration_cap_reason_in_additional_properties() -> None:
+    client = RecordingChatClient()
+    agent = Agent(client=client, middleware=[AgentLoopMiddleware(lambda **_: True, max_iterations=2)])
+
+    response = await agent.run("task")
+
+    assert client.call_count == 2
+    assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "iteration_cap_reached"
+
+
+async def test_loop_sets_iteration_cap_reason_streaming() -> None:
+    client = RecordingChatClient()
+    agent = Agent(client=client, middleware=[AgentLoopMiddleware(lambda **_: True, max_iterations=2)])
+
+    stream = agent.run("task", stream=True)
+    async for _ in stream:
+        pass
+    response = await stream.get_final_response()
+
+    assert client.call_count == 2
+    assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "iteration_cap_reached"
+
+
+class StaleExitReasonChatClient(RecordingChatClient):
+    """Stamps a stale ``loop_exit_reason`` on every response, emulating an inner provider or nested loop."""
+
+    def _inner_get_response(
+        self,
+        *,
+        messages: Sequence[Message],
+        stream: bool = False,
+        options: Mapping[str, Any],
+        **kwargs: Any,
+    ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]:
+        inner = super()._inner_get_response(messages=messages, stream=stream, options=options, **kwargs)
+        if isinstance(inner, ResponseStream):
+
+            async def _gen() -> AsyncIterable[ChatResponseUpdate]:
+                async for update in inner:
+                    update.additional_properties = {LOOP_EXIT_REASON_KEY: "stale_reason"}
+                    yield update
+
+            return ResponseStream(_gen(), finalizer=ChatResponse.from_updates)
+
+        async def _get() -> ChatResponse:
+            response = await inner
+            response.additional_properties = {LOOP_EXIT_REASON_KEY: "stale_reason"}
+            return response
+
+        return _get()
+
+
+async def test_loop_cap_overwrites_pre_existing_exit_reason() -> None:
+    client = StaleExitReasonChatClient()
+    agent = Agent(client=client, middleware=[AgentLoopMiddleware(lambda **_: True, max_iterations=2)])
+
+    response = await agent.run("task")
+
+    assert client.call_count == 2
+    assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "iteration_cap_reached"
+
+
+async def test_loop_cap_overwrites_pre_existing_exit_reason_streaming() -> None:
+    client = StaleExitReasonChatClient()
+    agent = Agent(client=client, middleware=[AgentLoopMiddleware(lambda **_: True, max_iterations=2)])
+
+    stream = agent.run("task", stream=True)
+    async for _ in stream:
+        pass
+    response = await stream.get_final_response()
+
+    assert client.call_count == 2
+    assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "iteration_cap_reached"
+
+
+async def test_judge_sets_completed_in_additional_properties() -> None:
+    """Judge marks the response as completed when the task is answered."""
+    agent_client = RecordingChatClient(texts=["The task is fully complete."])
+    judge_client = RecordingChatClient(texts=['{"answered": true, "reasoning": "done"}'], honor_response_format=True)
+    agent = Agent(client=agent_client, middleware=[AgentLoopMiddleware.with_judge(judge_client)])
+
+    response = await agent.run("do the task")
+
+    assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "completed"
+
+
+async def test_judge_sets_completed_in_additional_properties_streaming() -> None:
+    agent_client = RecordingChatClient(texts=["The task is fully complete."])
+    judge_client = RecordingChatClient(texts=['{"answered": true, "reasoning": "done"}'], honor_response_format=True)
+    agent = Agent(client=agent_client, middleware=[AgentLoopMiddleware.with_judge(judge_client)])
+
+    stream = agent.run("do the task", stream=True)
+    async for _ in stream:
+        pass
+    response = await stream.get_final_response()
+
+    assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "completed"
+
+
+async def test_judge_sets_completed_only_once_answered() -> None:
+    """Judge does not stamp completed until the task is actually answered."""
+    agent_client = RecordingChatClient(texts=["Still working...", "Now done."])
+    judge_client = RecordingChatClient(
+        texts=['{"answered": false, "reasoning": "needs more"}', '{"answered": true, "reasoning": "done"}'],
+        honor_response_format=True,
+    )
+    agent = Agent(client=agent_client, middleware=[AgentLoopMiddleware.with_judge(judge_client)])
+
+    response = await agent.run("do the task")
+
+    assert agent_client.call_count == 2
+    assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "completed"
+
+
+async def test_judge_does_not_set_completed_when_cap_fires() -> None:
+    agent_client = RecordingChatClient()
+    judge_client = RecordingChatClient(texts=['{"answered": false}'] * 2, honor_response_format=True)
+    agent = Agent(client=agent_client, middleware=[AgentLoopMiddleware.with_judge(judge_client, max_iterations=2)])
+
+    response = await agent.run("do the task")
+
+    assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "iteration_cap_reached"
+
+
+async def test_stale_inner_exit_reason_is_cleared_before_outer_evaluation() -> None:
+    """A stale inner loop_exit_reason must not leak into, or block, the outer loop's decision."""
+    agent_client = StaleExitReasonChatClient(texts=["The task is fully complete."])
+    judge_client = RecordingChatClient(texts=['{"answered": true, "reasoning": "done"}'], honor_response_format=True)
+    agent = Agent(client=agent_client, middleware=[AgentLoopMiddleware.with_judge(judge_client)])
+
+    response = await agent.run("do the task")
+
+    assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "completed"
+
+
+async def test_stale_inner_exit_reason_is_cleared_when_loop_stops_without_reason() -> None:
+    """If the outer loop stops without stamping a reason, the inner stale value must not survive."""
+    client = StaleExitReasonChatClient()
+    agent = Agent(client=client, middleware=[AgentLoopMiddleware(lambda **_: False, max_iterations=5)])
+
+    response = await agent.run("task")
+
+    assert LOOP_EXIT_REASON_KEY not in response.additional_properties