Python: [Bug]: once-per-run history drops earlier tool-loop turns when function middleware terminates

Description

With a local HistoryProvider and the default once-per-run persistence (require_per_service_call_history_persistence unset/false), MiddlewareTermination from function middleware ends agent.run() with a complete AgentResponse (prior tool turns plus the terminating call and result). After the session is serialized, the history provider only keeps the last model call plus that terminating tool result. Intermediate assistant/tool turns from the same run are gone.

That means a later agent.run(..., session=session) cannot continue the same tool loop: the model no longer sees the work it already did in the terminated run.

Expected: once-per-run local history is the full conversation of that agent.run(), including when the function-calling loop stops early via MiddlewareTermination.

Actual on agent-framework-core==1.18.0:

• Default once-per-run: intermediate tool-loop turns are dropped; the terminating tool result is stored.
• require_per_service_call_history_persistence=True: intermediate turns are stored; the terminating tool result is not (documented in #4992 / the Foundry sample, to match service-managed storage). There is no local-history mode that keeps both.

This is separate from #4609, which was about aligning the terminal result with store=True and was closed as designed. The gap here is the default path dropping earlier turns in the same run.

Workaround we used: set require_per_service_call_history_persistence=True and, after termination, manually save_messages for role="tool" messages from the response.

Code Sample

import asyncio
from collections.abc import Awaitable, Callable, Sequence
from typing import Any

from agent_framework import (
    BaseChatClient,
    ChatMiddlewareLayer,
    ChatResponse,
    Content,
    FunctionInvocationContext,
    FunctionInvocationLayer,
    FunctionMiddleware,
    InMemoryHistoryProvider,
    Message,
    MiddlewareTermination,
)


class StopAfterRunTests(FunctionMiddleware):
    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        await call_next()
        if context.function.name == "run_tests":
            raise MiddlewareTermination(result=context.result)


class StubClient(FunctionInvocationLayer, ChatMiddlewareLayer, BaseChatClient):
    STORES_BY_DEFAULT = False

    def __init__(self) -> None:
        super().__init__()
        self.n = 0

    async def _inner_get_response(
        self, *, messages: Sequence[Message], **kwargs: Any
    ) -> ChatResponse:
        self.n += 1
        name = "read_file" if self.n == 1 else "run_tests"
        return ChatResponse(
            messages=[
                Message(
                    role="assistant",
                    contents=[
                        Content.from_function_call(
                            call_id=str(self.n),
                            name=name,
                            arguments={},
                        )
                    ],
                )
            ]
        )


async def read_file() -> str:
    return "file contents"


async def run_tests() -> dict[str, str]:
    return {"status": "failed"}


def summary(messages: list[Message]) -> list[tuple[str, list[tuple[str, str | None]]]]:
    return [
        (str(m.role), [(c.type, getattr(c, "call_id", None)) for c in m.contents])
        for m in messages
    ]


async def main() -> None:
    client = StubClient()
    history = InMemoryHistoryProvider("history")
    agent = client.as_agent(
        context_providers=[history],
        middleware=[StopAfterRunTests()],
        # default: once-per-run persistence
    )
    session = agent.create_session()
    await agent.run("goal", session=session, tools=[read_file, run_tests])
    stored = await history.get_messages(
        session.session_id,
        state=session.state[history.source_id],
    )
    print(summary(stored))
    # Observed:
    # [('user', [('text', None)]),
    #  ('assistant', [('function_call', '2')]),
    #  ('tool', [('function_result', '2')])]
    # Missing: assistant function_call '1' and tool result '1' (read_file).


asyncio.run(main())

Error Messages / Stack Traces

No exception. The next agent.run on the restored session starts without the earlier tool-loop turns.

Package Versions

agent-framework-core: 1.18.0 (also present on 1.14.0)

Python Version

Python 3.13

Additional Context

Related but not the same:

• #4609 / #4992: terminal tool result omitted under per-service-call persistence to match service storage. That part looks intentional.
• #5354: how MiddlewareTermination appears on agent.run, not history.

Ask: for local HistoryProviders (store=False / STORES_BY_DEFAULT = False), persist the full terminated run (intermediates and the terminal result). Service-managed conversations can keep omitting a result they cannot write remotely.