Python: Python / AG-UI: Is stacking Thread Snapshot history with per-agent HistoryProvider intended? #8075

Summary

When an AG-UI workflow is backed by a Thread Snapshot store, follow-up turns that send messages without resume cause AgentFrameworkWorkflow to reconstruct the full prior thread transcript and pass it into workflow.run(message=...).

At the same time, each workflow participant still has its own agent session / InMemoryHistoryProvider (especially with require_per_service_call_history_persistence=True on local Chat Completions clients).

On turn 2+, the model call therefore sees roughly:

AG-UI snapshot replay (outer) + participant HistoryProvider (inner)

We are looking for maintainer guidance on whether this composition is intentional, and how conversation history is meant to be managed in this setup. We are not proposing a particular fix—we do not know what the intended practice is.

Related (similar dual-layer history, different surface): #7756 (SequentialBuilder + workflow.as_agent() + outer AgentSession). This issue is about the AG-UI Workflow + Thread Snapshot path.

What we observed

Setup:

• AgentFrameworkWorkflow(workflow_factory=...) + AG-UI Thread Snapshot store
• SequentialBuilder(participants=[writer, reviewer]) (easy to hit: every chat turn is messages without resume)
• Local OpenAI-compatible client (store=False / no service-managed conversation)
• Participants with require_per_service_call_history_persistence=True

Concrete probe line from a local run:

agent=writer workflow_input=2 history_provider=2 final_to_model=4 predicted_sum=4 stacked=True

Notes:

• On a Handoff AG-UI demo, mid-case turns are usually resume, so this stacking is harder to notice; a post-complete messages-without-resume kickoff on the same instance shows the same pattern.
• On Sequential, turn 2 of normal chat is enough.

Snapshot growth (second concern)

Thread Snapshot history appears to grow without compaction / truncation. For long multi-turn threads, the reconstructed messages list alone can become very large before it is even combined with per-agent history. We would also appreciate guidance on how this is expected to be handled.

Minimal reproduction (backend)

from agent_framework import Agent, ChatContext, ChatMiddleware
from agent_framework.ag_ui import (
    AgentFrameworkWorkflow,
    InMemoryAGUIThreadSnapshotStore,
    add_agent_framework_fastapi_endpoint,
)
from agent_framework.openai import OpenAIChatCompletionClient
from agent_framework.orchestrations import SequentialBuilder
from fastapi import FastAPI


class ContextProbe(ChatMiddleware):
    def __init__(self, agent_id: str) -> None:
        self._agent_id = agent_id

    async def process(self, context: ChatContext, call_next):
        before = list(context.messages)
        history = []
        if context.session is not None:
            history = list(context.session.state.get("in_memory", {}).get("messages", []) or [])
        await call_next()
        final = list(context.messages)
        print(
            f"[probe] {self._agent_id}: input={len(before)} "
            f"history={len(history)} final={len(final)} "
            f"stacked={len(final) == len(before) + len(history) and len(history) > 0 and len(before) > 1}"
        )


client = OpenAIChatCompletionClient(
    model="...",
    api_key="...",
    base_url="...",  # local Chat Completions; no service-managed history
)

writer = Agent(
    id="writer",
    name="writer",
    instructions="You are a concise copywriter.",
    client=client,
    middleware=[ContextProbe("writer")],
    require_per_service_call_history_persistence=True,
)
reviewer = Agent(
    id="reviewer",
    name="reviewer",
    instructions="You are a short reviewer.",
    client=client,
    middleware=[ContextProbe("reviewer")],
    require_per_service_call_history_persistence=True,
)

snapshot_store = InMemoryAGUIThreadSnapshotStore()

def workflow_factory(_thread_id: str):
    return SequentialBuilder(participants=[writer, reviewer]).build()

app = FastAPI()
add_agent_framework_fastapi_endpoint(
    app=app,
    agent=AgentFrameworkWorkflow(
        workflow_factory=workflow_factory,
        snapshot_store=snapshot_store,
    ),
    path="/sequential_demo",
    snapshot_store=snapshot_store,
    snapshot_scope_resolver=lambda _request: "demo",
)

Then POST twice on the same threadId (no resume):

1. messages: [{ "role": "user", "content": "Write a tagline for a budget-friendly eBike." }]
2. messages: [{ "role": "user", "content": "Make it shorter." }]

Turn 2 should log writer stacked=True (or equivalent: final ≈ input + history).

Questions for maintainers

1. Is stacking AG-UI Thread Snapshot history with each participant’s HistoryProvider intended for AG-UI workflows?
2. For AG-UI Workflow + SnapshotStore + agents with HistoryProvider, what is the recommended way to manage conversation history?
3. Thread Snapshots appear to grow without compaction/truncation—what is the recommended way to think about long multi-turn threads in this model?

Environment

• microsoft/agent-framework Python packages (workspace checkout)
• agent-framework-ag-ui + agent-framework-orchestrations
• Python 3.12+
• OpenAI-compatible local Chat Completions endpoint (no service-side conversation store)

Happy to add more traces or adjust the repro if that helps.