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

@eavanvalkenburg is already working on this.

Since Sep 8, 2026.

## 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`

| Layer | Turn 1 | Turn 2 (`Make it shorter.`) |
| --- | --- | --- |
| AG-UI reconstructed `messages` | 1 | >1 (prior transcript + new user) |
| Same cached Workflow instance | created | reused |
| Writer `HistoryProvider` | 0 | >0 |
| Writer final messages to model | ≈ input only | ≈ **input + history** (`stacked=True` in our probe) |

Concrete probe line from a local run:

```text
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)

```python
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.