Python: fix(python): return call-level compaction summaries in agent responses

docs/specs/004-python-function-calling-loop.md

@@ -519,6 +519,9 @@ that manually replay messages own the equivalent rule: do not resend an approval
   same turn.
 - A trusted terminal result consumes the corresponding approval authority in explicit stateless replay; a result in a
   server-registered pending occurrence cannot consume that authority before local execution.
+- Non-streaming runs that exclude tool groups through in-run compaction return the inserted summary messages in the
+  final response transcript, each positioned before the group it replaces, so history loaded with `skip_excluded`
+  keeps the summarized content; summaries of caller-owned input messages stay out of the returned transcript.
 
 ## Scenario-to-test matrix
 
@@ -540,6 +543,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | Declaration-only call | The call is surfaced as user input and is not executed; streaming arguments appear once while finalized request metadata remains available. | `test_declaration_only_tool`, `test_streaming_declaration_only_tool_preserves_metadata_without_duplicate_arguments` |
 | Function invocation disabled | The client bypasses the invocation loop without losing invocation kwargs. | `test_function_invocation_config_enabled_false`, `test_function_invocation_config_enabled_false_preserves_invocation_kwargs`, `test_streaming_function_invocation_config_enabled_false` |
 | Runtime tool changes | Added tools become available on the next iteration and retain approval behavior. | `test_add_tools_available_next_iteration`, `test_add_tools_with_approval_required_tool` |
+| In-run compaction summaries | Summaries inserted for tool groups excluded by in-run compaction are returned in the final non-streaming response transcript before the group they replace, so history loaded with `skip_excluded` keeps the summarized content; summaries of caller-owned input messages are not added. | `packages/core/tests/core/test_clients.py::test_function_loop_returns_compaction_summaries_in_final_response`, `test_function_loop_returns_compaction_summaries_when_iteration_budget_exhausted`, `test_function_loop_reconciles_nested_compaction_summaries`, `test_function_loop_returns_compacted_transcript_on_early_terminal_exit`, `packages/core/tests/core/test_agents.py::test_agent_run_returns_and_persists_compaction_summaries` |
 
 ### Approval pause and resume
 
@@ -742,3 +746,4 @@ Before accepting an update, reviewers must confirm:
 - #6963 / #7095 — opaque reasoning-signature replay
 - #6074 / #7233 — reasoning-paired tool-call replay
 - #6450 / #6794 — provider message and tool-result serialization
+- #8099 — in-run compaction summaries in the returned transcript

python/packages/core/agent_framework/_compaction.py

@@ -416,13 +416,13 @@ def _set_group_summarized_by_summary_id(message: Message, summary_id: str) -> No
 def _reconcile_compaction_summaries(  # pyright: ignore[reportUnusedFunction]
     source_messages: list[Message],
     working_messages: Sequence[Message],
-    previous_message_ids: set[int],
+    source_message_identities: set[int],
 ) -> None:
-    """Reconcile summaries of source messages without persisting unrelated rewrites."""
+    """Reconcile summaries supported by source-owned messages without persisting unrelated rewrites."""
     source_message_ids = {message.message_id for message in source_messages if message.message_id}
     candidates: list[tuple[Message, set[str]]] = []
     for message in working_messages:
-        if id(message) in previous_message_ids:
+        if id(message) in source_message_identities:
             continue
 
         annotation = _read_group_annotation_raw(message)

python/packages/core/agent_framework/_middleware.py

@@ -1546,7 +1546,7 @@ def get_response(
             function_invocation_kwargs=function_invocation_kwargs,
         )
         source_messages = messages if isinstance(messages, list) else None
-        baseline_message_ids = {id(message) for message in messages}
+        source_message_identities = {id(message) for message in messages}
         middleware_messages = cast("list[Message]", context.messages)
         downstream_messages: list[Message] | None = None
 
@@ -1575,7 +1575,7 @@ def _final_handler(
                     _reconcile_compaction_summaries(
                         source_messages,
                         downstream_messages if downstream_messages is not None else middleware_messages,
-                        baseline_message_ids,
+                        source_message_identities,
                     )
 
         if stream:

python/packages/core/agent_framework/_tools.py

@@ -3788,8 +3788,6 @@ async def _process_model_function_calls(
     tools = _extract_tools(options)
     function_calls = _extract_function_calls(response)
     if not (function_calls and tools):
-        if function_call_messages is not None:
-            _prepend_function_call_messages(response, function_call_messages)
         if approval_requests:
             _store_pending_approval_requests(invocation_session, approval_requests)
         return _FunctionProcessingResult(errors_in_a_row=errors_in_a_row, action="return")
@@ -3986,6 +3984,7 @@ async def _get_response_with_function_invocation(
         middleware_pipeline: FunctionMiddlewarePipeline | None = None,
     ) -> ChatResponse[Any]:
         """Run the non-streaming function invocation loop."""
+        from ._compaction import _reconcile_compaction_summaries  # pyright: ignore[reportPrivateUsage]
         from ._middleware import MiddlewareFailure
         from ._types import ChatResponse, add_usage_details
 
@@ -4072,6 +4071,16 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non
                     client_kwargs=request_kwargs,
                 ),
             )
+            # Compaction inserts summaries only into prepared_messages while its exclusion
+            # flags land on Message objects shared with the transcript. Reconcile summaries
+            # supported entirely by transcript-owned messages before returning or persisting
+            # the response (issue #8099). This is unconditional because compaction may come
+            # from the inner client's default strategy, which this layer does not see here.
+            _reconcile_compaction_summaries(
+                function_call_messages,
+                prepared_messages,
+                {id(message) for message in function_call_messages},
+            )
             if options.get("tool_choice") == "none" and budget_state.get("truncated"):
                 _ensure_function_invocation_limit_fallback_response(response)
             aggregated_usage = add_usage_details(aggregated_usage, response.usage_details)
@@ -4082,6 +4091,7 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non
                 options=options,
             )
 
+            terminal_prefix_length = len(function_call_messages)
             try:
                 function_processing = await _process_model_function_calls(
                     response=response,
@@ -4115,6 +4125,7 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non
             )
             if function_processing.action == "return":
                 response.usage_details = aggregated_usage
+                _prepend_function_call_messages(response, function_call_messages[:terminal_prefix_length])
                 _clear_budget_state_from_session(invocation_session)
                 return _clear_internal_conversation_id(response)
             _apply_batch_limit_decision(
@@ -4150,6 +4161,13 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non
                 client_kwargs=request_kwargs,
             ),
         )
+        # See the Phase 2 reconciliation: the final no-tools call can compact further
+        # groups, so those summaries must also reach the returned transcript.
+        _reconcile_compaction_summaries(
+            function_call_messages,
+            prepared_messages,
+            {id(message) for message in function_call_messages},
+        )
         _ensure_function_invocation_limit_fallback_response(response)
         aggregated_usage = add_usage_details(aggregated_usage, response.usage_details)
         self._update_function_invocation_continuation_state(

python/packages/core/tests/core/test_agents.py

@@ -14,6 +14,7 @@
 from pytest import raises
 
 from agent_framework import (
+    EXCLUDED_KEY,
     GROUP_ANNOTATION_KEY,
     GROUP_TOKEN_COUNT_KEY,
     Agent,
@@ -37,6 +38,7 @@
     SlidingWindowStrategy,
     SupportsAgentRun,
     SupportsChatGetResponse,
+    ToolResultCompactionStrategy,
     TruncationStrategy,
     chat_middleware,
     enqueue_messages,
@@ -2436,6 +2438,86 @@ async def capturing_inner(
     assert captured_token_counts == [[23]]
 
 
+async def test_agent_run_returns_and_persists_compaction_summaries(
+    chat_client_base: Any,
+) -> None:
+    # End-to-end regression test for #8099: with call-level compaction and a history
+    # provider loading with skip_excluded=True, the run must return — and persist — the
+    # summaries replacing excluded tool groups; otherwise the next turn silently loses
+    # the summarized tool results with nothing replacing them.
+    from agent_framework._sessions import InMemoryHistoryProvider
+
+    chat_client_base.function_invocation_configuration["enabled"] = True  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
+    chat_client_base.function_invocation_configuration["max_iterations"] = 3  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
+
+    @tool(name="lookup_weather", approval_mode="never_require")
+    def lookup_weather(location: str) -> str:
+        return f"Weather in {location}: sunny"
+
+    chat_client_base.run_responses = [  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
+        ChatResponse(
+            messages=Message(
+                role="assistant",
+                contents=[
+                    Content.from_function_call(
+                        call_id="call_1",
+                        name="lookup_weather",
+                        arguments='{"location": "London"}',
+                    )
+                ],
+            ),
+            response_id="resp_call_1",
+        ),
+        ChatResponse(
+            messages=Message(
+                role="assistant",
+                contents=[
+                    Content.from_function_call(
+                        call_id="call_2",
+                        name="lookup_weather",
+                        arguments='{"location": "Paris"}',
+                    )
+                ],
+            ),
+            response_id="resp_call_2",
+        ),
+        ChatResponse(messages=Message(role="assistant", contents=["done"]), response_id="resp_done"),
+    ]
+
+    provider = InMemoryHistoryProvider(skip_excluded=True)
+    agent = Agent(
+        client=chat_client_base,
+        tools=[lookup_weather],
+        compaction_strategy=ToolResultCompactionStrategy(keep_last_tool_call_groups=1),
+        context_providers=[provider],
+    )
+    session = agent.create_session()
+
+    result = await agent.run("What is the weather in London?", session=session)
+
+    def _is_tool_result_summary(message: Message) -> bool:
+        return message.role == "assistant" and (message.text or "").startswith("[Tool results:")
+
+    returned_summaries = [message for message in result.messages if _is_tool_result_summary(message)]
+    assert len(returned_summaries) == 1, [message.text for message in result.messages]
+    assert "London" in (returned_summaries[0].text or "")
+
+    stored_messages = cast(list[Message], session.state[InMemoryHistoryProvider.DEFAULT_SOURCE_ID]["messages"])
+    stored_summaries = [message for message in stored_messages if _is_tool_result_summary(message)]
+    assert len(stored_summaries) == 1, [message.text for message in stored_messages]
+
+    # A follow-up turn loading with skip_excluded=True drops the excluded groups while
+    # the summary that replaces them survives.
+    loaded_messages = await provider.get_messages(
+        session_id="turn-2", state=cast("dict[str, Any]", session.state[InMemoryHistoryProvider.DEFAULT_SOURCE_ID])
+    )
+    loaded_summaries = [message for message in loaded_messages if _is_tool_result_summary(message)]
+    assert len(loaded_summaries) == 1
+    assert "London" in (loaded_summaries[0].text or "")
+    assert not any(message.additional_properties.get(EXCLUDED_KEY, False) for message in loaded_messages)
+    assert "done" in [message.text for message in loaded_messages]
+
+
 # region Test _merge_options